mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
d88b2ea44a
The drain took the post-delivery slot by live.route read in the turn's finally, so a rebind mid-turn looked under the new key and stranded the callback the turn registered under its own. The key is now captured when the turn starts. Deleting a session forgets its local-adapter state only for LOCAL sources, so a runner without an adapter map (non-local test doubles) no longer raises on delete/prune. Retiring closed local sessions from authority.sessions (and the _schedule restore) moves to a follow-up PR.
569 lines
32 KiB
Python
569 lines
32 KiB
Python
"""Trusted messaging admission and the existing TurnRunner invocation boundary."""
|
|
import asyncio
|
|
import logging
|
|
from contextvars import ContextVar
|
|
from contextlib import asynccontextmanager, nullcontext
|
|
from dataclasses import replace
|
|
|
|
from gateway.platforms.event import MessageEvent, ProcessingOutcome
|
|
from gateway.session_envelope import restore_native
|
|
from gateway.session_producer_lifecycle import close_producer_turn, producer_turn
|
|
from hermes_cli.observability.shared_metrics_busy import (
|
|
busy_turn_started, note_busy_admitted, note_busy_parked, settle_busy_arrival)
|
|
from hermes_state_runtime import RuntimeStoreError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
admission_author = ContextVar('admission_author', default=None)
|
|
executing_admission = ContextVar('executing_admission', default=False)
|
|
# The live session while the executing turn is one its viewer submitted (``is_viewer_input``), else None.
|
|
viewer_turn: ContextVar = ContextVar('viewer_turn', default=None)
|
|
# Set by a busy-path caller that must be released once its input is durably queued.
|
|
busy_acceptance = ContextVar('busy_acceptance', default=None)
|
|
# Set beside it by a caller that must also know when the row is durably committed (a busy burst's
|
|
# commit task, which command barriers and /restart wait on): ``busy_acceptance`` may already resolve
|
|
# once the input's FIFO position is fixed behind an earlier input of its chat still capturing.
|
|
busy_commit = ContextVar('busy_commit', default=None)
|
|
# ``_validate_native`` clauses about the message's own content, safe to name to the sender's chat.
|
|
_CONTENT_CLAUSES = frozenset({
|
|
'text', 'auto_skill', 'channel_prompt', 'channel_context', 'reply_expected',
|
|
'media_urls', 'media_types', 'media_text_inlined', 'media_types_length', 'media_text_inlined_length',
|
|
'media_filenames', 'media_filenames_length',
|
|
})
|
|
|
|
# Nothing was committed (main's busy-queue cap); /retry answers it before rewriting (#139257).
|
|
QUEUE_FULL_REPLY = ('⏳ Too many messages are already waiting in this conversation; '
|
|
'send this again after the current reply.')
|
|
|
|
|
|
async def admit_message(authority, event):
|
|
from gateway.session_authority import messaging_admission
|
|
accepted, committed = busy_acceptance.get(), busy_commit.get()
|
|
busy_acceptance.set(None) # the drain task this schedules must not inherit the busy caller's future
|
|
busy_commit.set(None)
|
|
ingress = messaging_admission.set(True if accepted is None else accepted)
|
|
try:
|
|
receipt = await authority.admit_native(event)
|
|
if committed is not None and not committed.done():
|
|
committed.set_result(None) # the row is durable (any status): a barrier may go on
|
|
except RuntimeStoreError as exc:
|
|
if exc.reason == 'stale_generation' and exc.field == 'route':
|
|
return None # a later /stop or /new of the chat fenced it, as main cancels a queued row
|
|
if exc.reason == 'invalid_params' and exc.field:
|
|
# Refused before commit (_validate_native): a refusal notice instead of a generic turn error.
|
|
# The reply reaches the whole chat, so only content clauses are named; trust and provenance
|
|
# clauses (and any new one) stay generic, and gateway.log keeps the clause (``_refuse``).
|
|
named = f'{exc.reason}: {exc.field}' if exc.field in _CONTENT_CLAUSES else exc.reason
|
|
return (f'⚠️ Hermes refused this message ({named}), so it was not saved '
|
|
'or answered. Ask the operator to check gateway.log.')
|
|
if exc.reason != 'queue_full':
|
|
raise
|
|
# Nothing was committed (main's busy-queue cap): say so instead of a generic turn error.
|
|
return QUEUE_FULL_REPLY
|
|
finally:
|
|
messaging_admission.reset(ingress)
|
|
if receipt.status == 'terminal':
|
|
return None
|
|
# Only the delivery waiter is process-local; execution reads the committed snapshot.
|
|
authority.native_waiters.add(receipt.admission_id)
|
|
waiter = authority.waiters.setdefault(receipt.admission_id, asyncio.get_running_loop().create_future())
|
|
adopters = _marker_adopters(authority)
|
|
if getattr(event, '_turn_marker_handoff', False):
|
|
# The adapter lifecycle that sends this reply releases the turn's crash marker only once
|
|
# the reply is in the delivery ledger; the drain hands its marker to this event.
|
|
adopters[receipt.admission_id] = event
|
|
# A drain already running can claim, run and settle the row before this caller resumes; one
|
|
# that settled it already sent (or owes) the reply itself and resolves no waiter registered now.
|
|
from hermes_state_runtime import get_session_admission
|
|
row = get_session_admission(authority.db, admission_id=receipt.admission_id)
|
|
status = 'terminal' if row is None else row['status'] # None: deleted with its chat, nothing owed
|
|
note_busy_admitted(event, receipt.admission_id, status, None if row is None else row.get('outcome'))
|
|
if status in ('terminal', 'unknown'):
|
|
authority.native_waiters.discard(receipt.admission_id)
|
|
if authority.waiters.get(receipt.admission_id) is waiter:
|
|
del authority.waiters[receipt.admission_id]
|
|
if adopters.get(receipt.admission_id) is event:
|
|
del adopters[receipt.admission_id]
|
|
return None if status == 'terminal' else pause_notice(authority, receipt.ref, 'unknown_execution')
|
|
if accepted is not None and not accepted.done():
|
|
accepted.set_result(None)
|
|
try:
|
|
return await asyncio.shield(waiter)
|
|
except RuntimeStoreError as exc:
|
|
row = get_session_admission(authority.db, admission_id=receipt.admission_id)
|
|
if row is not None and row['status'] == 'terminal':
|
|
# The drain refused this head (``_refuse_head``): cancelled, not saved behind a pause.
|
|
return (f'⚠️ Hermes did not answer this message ({exc.reason}): the conversation it was '
|
|
'queued for moved, so it was dropped. Send it again to continue.')
|
|
if row is not None:
|
|
note_busy_parked(event) # still queued (saved): answered after the pause, not dropped
|
|
return pause_notice(authority, receipt.ref, exc.reason)
|
|
finally:
|
|
if adopters.get(receipt.admission_id) is event:
|
|
del adopters[receipt.admission_id]
|
|
|
|
|
|
def _marker_adopters(authority):
|
|
adopters = getattr(authority, 'marker_adopters', None)
|
|
if adopters is None:
|
|
adopters = authority.marker_adopters = {}
|
|
return adopters
|
|
|
|
|
|
_MARKER_FIELDS = ('_gateway_active_turn_session_key', '_gateway_active_turn_token')
|
|
|
|
|
|
async def _hand_over_turn_marker(authority, admission_id, event):
|
|
"""The executed turn's durable active-turn marker outlives the handler, as the in-process
|
|
turn's does on main, and is released only once its reply is in the delivery ledger (or nothing
|
|
is owed). A kill between the terminal commit and that ledger write then leaves a marked turn
|
|
whose persisted reply the next unclean boot ledgers and sends once, without new inference,
|
|
instead of a settled answer nobody owes. The waiting adapter lifecycle adopts it; a recovered
|
|
no-waiter reply keeps it until ``deliver_settled``; any other turn releases it here."""
|
|
adopter = _marker_adopters(authority).get(admission_id)
|
|
if adopter is not None:
|
|
_move_marker(event, adopter)
|
|
return
|
|
if admission_id in authority.pending_deliveries:
|
|
# Released by ``deliver_settled``, also when settlement failed and dropped the delivery.
|
|
_held_markers(authority)[admission_id] = event
|
|
return
|
|
await _release_turn_marker(authority, event)
|
|
|
|
|
|
def _move_marker(source, target):
|
|
for name in _MARKER_FIELDS:
|
|
if hasattr(source, name):
|
|
setattr(target, name, getattr(source, name))
|
|
delattr(source, name)
|
|
|
|
|
|
def _held_markers(authority):
|
|
held = getattr(authority, 'held_turn_markers', None)
|
|
if held is None:
|
|
held = authority.held_turn_markers = {}
|
|
return held
|
|
|
|
|
|
def _registry(authority, name):
|
|
found = getattr(authority, name, None)
|
|
if found is None:
|
|
found = {}
|
|
setattr(authority, name, found)
|
|
return found
|
|
|
|
|
|
def waiter_replies(authority):
|
|
"""admission -> the reply a live native waiter is about to send, kept only until the drain
|
|
settles the turn, so a turn fenced ``unknown`` can still owe it to the chat."""
|
|
return _registry(authority, 'waiter_replies')
|
|
|
|
|
|
def retain_unsettled_delivery(authority, admission_id):
|
|
"""Settlement could not commit and the turn is fenced ``unknown``, but this owner keeps its
|
|
captured answer for Discard (``resolve_unknown``). The reply stays owed to its destination,
|
|
and so does the turn's crash marker: the live waiter is released with a pause notice, so the
|
|
marker is taken back from it. Restart while unknown clears the marker (no answer is owed,
|
|
``release_unknown_turn_markers``); a Discard that commits the captured answer makes it owed,
|
|
sent once by ``deliver_resolved`` or, after a kill before that send, by the boot ledger."""
|
|
delivery = waiter_replies(authority).pop(admission_id, None)
|
|
if delivery is not None:
|
|
adopter = _marker_adopters(authority).get(admission_id)
|
|
if adopter is not None:
|
|
_move_marker(adopter, delivery[1])
|
|
_held_markers(authority)[admission_id] = delivery[1]
|
|
else:
|
|
delivery = authority.pending_deliveries.pop(admission_id, None)
|
|
if delivery is not None:
|
|
_registry(authority, 'unsettled_deliveries')[admission_id] = delivery
|
|
|
|
|
|
async def deliver_resolved(authority, admission_id, *, owed):
|
|
"""After a Discard committed: send the retained reply when its captured answer committed with
|
|
the resolution (``owed``), else release its marker and drop it. Popped first, so a repeated
|
|
resolution never sends twice; a turn the drain still owns is delivered by the drain."""
|
|
delivery = _registry(authority, 'unsettled_deliveries').pop(admission_id, None)
|
|
if delivery is None:
|
|
return
|
|
if owed:
|
|
authority.pending_deliveries[admission_id] = delivery
|
|
await deliver_settled(authority, admission_id)
|
|
|
|
|
|
def _take_post_delivery(authority, admission_id, source, route):
|
|
"""Move the turn's post-delivery slot off its adapter onto this admission. A native waiter (the
|
|
messaging adapter task that sends the reply) runs it after its own send instead. A runner or
|
|
adapter without the seam (a webhook sink, a bare test double) holds none."""
|
|
resolve = getattr(authority.runner, '_adapter_for_source', None)
|
|
native = admission_id in authority.native_waiters
|
|
slots = getattr(resolve(source), '_post_delivery_callbacks', None) if resolve and source and not native else None
|
|
entry = slots.pop(route, None) if slots is not None else None
|
|
if entry is not None:
|
|
_registry(authority, 'post_delivery')[admission_id] = (slots, route, entry)
|
|
|
|
|
|
def _hand_back_post_delivery(authority, admission_id):
|
|
"""A native waiter registered after the turn started: its adapter task sends the reply and then
|
|
runs the slot, which goes back unless a newer turn's already took it."""
|
|
taken = _registry(authority, 'post_delivery').pop(admission_id, None)
|
|
if taken is not None:
|
|
slots, route, entry = taken
|
|
slots.setdefault(route, entry)
|
|
|
|
|
|
async def fire_post_delivery(authority, admission_id):
|
|
"""Run a settled turn's post-delivery callbacks (background-review release, temp-bubble cleanup,
|
|
a deferred /goal or /loop verdict) after its reply was sent, as ``_process_message_background``
|
|
does after a messaging reply. Never run, they also pinned the whole turn (its agent, history
|
|
and result) in the adapter's slot until restart."""
|
|
if admission_id in authority.native_waiters:
|
|
_hand_back_post_delivery(authority, admission_id)
|
|
return
|
|
taken = _registry(authority, 'post_delivery').pop(admission_id, None)
|
|
if taken is None:
|
|
return
|
|
from gateway.platforms.base import _POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS, _split_post_delivery_entry
|
|
from gateway.platforms.base_post_delivery import run_post_delivery_callback
|
|
callback = _split_post_delivery_entry(taken[2])[1]
|
|
with await _execution_scope(authority, None): # off-turn: the owning profile's scope
|
|
await run_post_delivery_callback(callback, _POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS)
|
|
|
|
|
|
async def _release_turn_marker(authority, event):
|
|
if getattr(event, '_gateway_active_turn_token', None):
|
|
await authority.runner._clear_durable_active_turn(event)
|
|
|
|
|
|
def pause_notice(authority, ref, reason):
|
|
"""The committed input stays queued behind a paused FIFO (a turn lost across an owner
|
|
restart, or a head the preflight refused). Tell the platform user once per pause episode
|
|
through the ordinary reply path; later messages onto the same pause are admitted silently."""
|
|
live = authority.sessions[ref.session_id]
|
|
if live.pause_notified:
|
|
return None
|
|
live.pause_notified = True
|
|
if reason == 'runtime_draining':
|
|
# Transient: the row runs when the restarted owner drains it, so /reset would be wrong advice.
|
|
return '⏳ Hermes is restarting — your message is saved and will be answered when it is back.'
|
|
if reason == 'session_busy':
|
|
# Transient: queued behind work already running on this conversation (a worker execution).
|
|
return '⏳ Your message is saved and will be answered when the work running in this conversation finishes.'
|
|
if reason == 'unknown_execution':
|
|
# GW-10: the way out is in this chat. /discard acknowledges the lost turn (never replayed)
|
|
# through the same fenced resolve_unknown as Desktop/TUI, and the saved messages then run.
|
|
# Slack registers no native /discard (its slash cap): the umbrella `/hermes discard` reaches it.
|
|
slack = getattr(getattr(live.source, 'platform', None), 'value', None) == 'slack'
|
|
discard = '/hermes discard' if slack else '/discard'
|
|
return ('⏸️ This conversation is paused: a previous turn did not finish when Hermes restarted, '
|
|
f'and only you know whether to send it again. Your message is saved. Send {discard} to drop '
|
|
'the lost turn and continue (it is not re-run), or /reset to start a fresh conversation.')
|
|
cause = f'a queued message could not be re-authorized ({reason})'
|
|
return (f'⏸️ This conversation is paused: {cause}. Your message is saved but will not be answered here. '
|
|
'Send /reset to start a fresh conversation, or ask the operator to resume this one.')
|
|
|
|
|
|
def row_turn_author(policy, row):
|
|
"""Who wrote the admitted input, for memory attribution only: the producer's stamp when it
|
|
left one, else the forwarded peer bound into the session policy; never grants anything."""
|
|
payload = row['payload']
|
|
for key in ('local_automation_v1', 'api_turn_v1', 'cli_turn_v1'):
|
|
author = (payload.get(key) or {}).get('turn_author')
|
|
if author is not None:
|
|
return author
|
|
from gateway.session_a2a import forward_author
|
|
return forward_author(policy)
|
|
|
|
|
|
async def execute_admission(authority, ref, row):
|
|
busy_turn_started(row['admission_id'])
|
|
from gateway.session_completion_notes import note_if_unviewed
|
|
noted = await note_if_unviewed(authority, ref, row) # #134889 LC-7: no viewer, no turn
|
|
if noted is not None:
|
|
return noted
|
|
from gateway.session_policy import policy_for_source
|
|
policy = policy_for_source(authority.runner, authority.sessions[ref.session_id].source)
|
|
if policy is not None and policy.source == 'cron':
|
|
from gateway.session_cron import execute
|
|
return await execute(authority, ref, row, policy)
|
|
local_policy = policy
|
|
from gateway.session_managed_worker import managed_policy, admit_managed
|
|
policy = managed_policy(authority, ref)
|
|
if policy is not None:
|
|
return await admit_managed(authority, ref, row, policy)
|
|
live = authority.sessions[ref.session_id]
|
|
# The key the turn registers its post-delivery slot under; a rebind mid-turn must not move the take.
|
|
route = live.route
|
|
native = row['admission_id'] in authority.native_waiters
|
|
authority.native_waiters.discard(row['admission_id'])
|
|
if 'native_text_v1' in row['payload']:
|
|
event = restore_native(row['payload'], authority.runner)
|
|
else:
|
|
from gateway.session_ingress_media import restore_attachments
|
|
event = MessageEvent(text=row['payload']['text'], source=live.source,
|
|
message_id=row['admission_id'], **restore_attachments(row['payload']))
|
|
if 'local_automation_v1' in row['payload']:
|
|
from gateway.session_automation import restore_local_automation
|
|
event = restore_local_automation(authority, ref, row)
|
|
if event.internal and 'gateway_session_id' in event.metadata:
|
|
# Main's ownership proof (802fd5782f3): the session the pre-claim check just validated, so the
|
|
# turn's resolver never re-reads (and on a transient failure drops) an admitted completion.
|
|
event.metadata['gateway_proven_session_id'] = (
|
|
authority.physical_target(ref) if 'native_text_v1' in row['payload'] else event.metadata['gateway_session_id'])
|
|
provenance = row['payload'].get('native_text_v1', {}).get('provenance')
|
|
home = None
|
|
if provenance is not None:
|
|
from gateway.session_ingress_context import restore_provenance
|
|
home = restore_provenance(authority.runner, event.source, provenance)
|
|
from gateway.config import Platform
|
|
from gateway.session_api_turn import api_execution, prepare_api_execution
|
|
from gateway.session_results import execution_result
|
|
is_api = live.source.platform == Platform.API_SERVER
|
|
if not (event.internal or is_api or live.source.platform == Platform.LOCAL):
|
|
# The claim on a chat route (native or a viewer's submit): a /new before the turn resolves
|
|
# its session must not move it onto the fresh conversation (main's /new cancelled its task).
|
|
event._claimed_owner = (ref.session_id, row['generation'])
|
|
author = event.metadata.get('turn_author')
|
|
author_token = admission_author.set(author if author is not None else row_turn_author(local_policy, row))
|
|
from gateway.session_operator import is_viewer_input
|
|
viewer_token = viewer_turn.set(live if is_viewer_input(row) else None)
|
|
api_token = api_execution.set(None)
|
|
captured = {}
|
|
result_token = execution_result.set(captured)
|
|
token = executing_admission.set(True)
|
|
# This drain, not the handler's unwind, releases the turn's crash marker (see below).
|
|
event._turn_marker_handoff = True
|
|
handed = False
|
|
try:
|
|
async with _admission_scope(authority, home, row, event, captured):
|
|
if is_api:
|
|
# Committed media resolves under the owning profile's home, like admission did.
|
|
prepared = prepare_api_execution(authority, ref, row['payload'])
|
|
api_execution.set(prepared)
|
|
if isinstance(event.text, list):
|
|
# The durable transcript keeps the committed media references (the same
|
|
# ``[Image attached ...]`` hints native transports persist), not only the caption.
|
|
event.text = '\n'.join(part['text'] for part in prepared['content'] if part.get('type') == 'text')
|
|
event.allow_gateway_control = False
|
|
event.internal = True # trust comes from the private binding and preclaim, never client JSON
|
|
response = await authority.runner._handle_message(event)
|
|
result = captured.get('terminal_result', captured.get('result'))
|
|
if result is None:
|
|
# No TurnRunner result means the handler answered without executing the turn.
|
|
# A turn that never ran because it failed (agent initialization raised, history
|
|
# unreadable) recorded its own failure via ``record_unexecuted_failure`` on every
|
|
# surface; any other reply here is a deliberate notice. An API caller asked for
|
|
# work, so its receipt is a failure either way, never a completed apology; so did
|
|
# a finite (``chat -q`` / ``-z``) prompt, whose exit code is the only verdict a
|
|
# script reads, unless the prompt was a command and the reply is its answer.
|
|
result = {'final_response': response or '', 'messages': []}
|
|
if is_api or (row['payload'].get('finite') is True and not event.get_command()):
|
|
result = {'final_response': '', 'messages': [], 'failed': True, 'completed': False,
|
|
'error': response or 'The admitted turn failed.'}
|
|
# The drain commits this under the stream lock so no viewer reads `terminal`
|
|
# before the completion event exists in the replay ring.
|
|
authority.pending_results[row['admission_id']] = {
|
|
'result': result, 'usage': captured.get('usage', {}),
|
|
**({'session_usage': captured['session_usage']} if 'session_usage' in captured else {})}
|
|
if not is_api and response:
|
|
adapter = authority.runner._adapter_for_source(event.source)
|
|
if adapter is not None:
|
|
# A recovered turn with no live delivery waiter: the drain sends this only after
|
|
# it commits the terminal outcome (deliver_settled), so a failed settlement never
|
|
# leaves an externally answered admission started -> unknown. A live waiter
|
|
# sends its own; the record only outlives a settlement fenced unknown.
|
|
delivery = (adapter, event, live.route, response, home)
|
|
if native:
|
|
waiter_replies(authority)[row['admission_id']] = delivery
|
|
else:
|
|
authority.pending_deliveries[row['admission_id']] = delivery
|
|
handed = True
|
|
await _hand_over_turn_marker(authority, row['admission_id'], event)
|
|
return response
|
|
finally:
|
|
if not native:
|
|
# No adapter task awaits this turn to run its post-delivery callbacks: the drain runs them
|
|
# once the turn settled and its reply went out (``fire_post_delivery``).
|
|
_take_post_delivery(authority, row['admission_id'], event.source, route)
|
|
if not handed:
|
|
await _release_turn_marker(authority, event)
|
|
executing_admission.reset(token)
|
|
execution_result.reset(result_token)
|
|
api_execution.reset(api_token)
|
|
admission_author.reset(author_token)
|
|
viewer_turn.reset(viewer_token)
|
|
|
|
|
|
def _launch_body_if_launch_owner(authority):
|
|
"""A standalone owner's turn binds nothing, but the launch profile's is marked at entry: a second
|
|
profile switching this gateway to multiplexing mid-turn must not fail its later credential reads
|
|
closed. Any other owner (a hosted named profile) stays unmarked, never reading the launch env."""
|
|
from pathlib import Path
|
|
from agent.secret_scope import _is_process_home
|
|
owner = Path(str(authority.profile_id))
|
|
if not owner.is_absolute() or not _is_process_home(owner):
|
|
return nullcontext()
|
|
from tui_gateway.launch_profile_policy import launch_body
|
|
return launch_body()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _admission_scope(authority, home, row, event, captured):
|
|
"""The execution scope, inside the adapter lifecycle the executor owns for a producer-admitted
|
|
turn (``session_producer_lifecycle``; a no-op for every other admission)."""
|
|
async with producer_turn(authority, row, event, captured):
|
|
with await _execution_scope(authority, home):
|
|
yield
|
|
|
|
|
|
async def _execution_scope(authority, home):
|
|
"""Under multiplex, owner-side execution runs under the OWNING profile's home (agent build,
|
|
config, secrets, state.db), never the launch profile's ambient scope; native provenance
|
|
(``home``) refines it. A single-profile gateway keeps its ambient scope byte-for-byte.
|
|
Secret sources (external CLIs) hydrate off the loop, as main's per-message scope does (#99519)."""
|
|
multiplex = getattr(getattr(authority.runner, 'config', None), 'multiplex_profiles', False)
|
|
if home is None and not multiplex:
|
|
return _launch_body_if_launch_owner(authority)
|
|
from gateway.run import _load_profile_secret_scope, _profile_runtime_scope
|
|
|
|
def load():
|
|
from pathlib import Path
|
|
scope_home = home
|
|
if scope_home is None: # owner_scope's home; symbolic test ids keep the ambient scope
|
|
owner = Path(str(authority.profile_id))
|
|
scope_home = owner if owner.is_absolute() and owner.is_dir() else None
|
|
return scope_home, scope_home and _load_profile_secret_scope(scope_home)
|
|
|
|
scope_home, secrets = await asyncio.to_thread(load)
|
|
return nullcontext() if scope_home is None else _profile_runtime_scope(scope_home, secrets)
|
|
|
|
|
|
async def deliver_settled(authority, admission_id):
|
|
"""Send a recovered no-waiter turn's reply after its terminal outcome committed. A send
|
|
failure cannot rewrite that outcome; it is logged, and inference never runs again."""
|
|
held = _held_markers(authority).pop(admission_id, None)
|
|
delivery = authority.pending_deliveries.get(admission_id)
|
|
if delivery is not None and admission_id in authority.native_waiters:
|
|
# ``admit_message`` registered its delivery waiter after this turn started: the drain
|
|
# hands that waiter the reply and its adapter lifecycle sends it, so this send is not owed.
|
|
authority.native_waiters.discard(admission_id)
|
|
_hand_back_post_delivery(authority, admission_id)
|
|
authority.pending_deliveries.pop(admission_id)
|
|
adopter = _marker_adopters(authority).get(admission_id)
|
|
if held is not None and adopter is not None:
|
|
_move_marker(held, adopter)
|
|
held = None
|
|
delivery = None
|
|
if delivery is None:
|
|
if held is not None:
|
|
# Fenced unknown: no answer is owed, so no later boot may deliver one either.
|
|
await _release_turn_marker(authority, held)
|
|
# Settled with nothing (left) to send: its reply streamed, or it had none.
|
|
await close_producer_turn(authority, admission_id, ProcessingOutcome.SUCCESS)
|
|
return
|
|
adapter, event, session_key, response, home = delivery
|
|
delivered = False
|
|
try:
|
|
with await _execution_scope(authority, home):
|
|
delivered = await deliver_response(adapter, event, session_key, response)
|
|
except Exception:
|
|
logger.exception('Delivery of settled admission %s failed', admission_id)
|
|
finally:
|
|
# Held until the send returns: an empty map means every settled reply was handed over.
|
|
authority.pending_deliveries.pop(admission_id, None)
|
|
# The ledgered send already released the marker; an empty or failed send owes no more.
|
|
await _release_turn_marker(authority, event)
|
|
await close_producer_turn(
|
|
authority, admission_id, ProcessingOutcome.SUCCESS if delivered else ProcessingOutcome.FAILURE)
|
|
|
|
|
|
async def deliver_response(adapter, event, session_key, response):
|
|
"""Send a turn's reply; whether it was delivered (``_process_message_background``'s verdict:
|
|
nothing to send, or at least one send that succeeded)."""
|
|
from gateway.platforms.base import _thread_metadata_for_event, _mark_notify_metadata
|
|
text, ttl = adapter._unwrap_ephemeral(response)
|
|
if not text:
|
|
return True
|
|
extracted = await adapter._extract_response_content(text, event, session_key, is_ephemeral_response=ttl > 0)
|
|
metadata = _mark_notify_metadata(_thread_metadata_for_event(event))
|
|
results = []
|
|
if extracted.text_content:
|
|
await adapter._send_final_text(event, session_key, extracted.text_content,
|
|
metadata, ttl > 0, ttl, results.append)
|
|
await adapter._deliver_attachments(event, extracted, metadata, anything_sent=bool(results),
|
|
record_delivery=results.append)
|
|
return any(getattr(result, 'success', False) for result in results if result is not None)
|
|
|
|
|
|
async def dispatch_shared_busy(adapter, event, session_key, *, durable=False):
|
|
"""Serial receive loops (IRC, Signal) await this before reading their next frame, so return
|
|
once the follow-up is committed to the FIFO, or queued behind an earlier input of its chat still
|
|
capturing (its position is then fixed; acceptance still waits for the commit); a tracked
|
|
observer delivers its one reply. *durable*: return only once the row is committed (or the input
|
|
was refused, fenced or answered), never on the earlier position release: a busy burst's commit
|
|
task, which a command barrier (``/restart`` closing admission) waits on. Otherwise an input
|
|
released before its commit is held until it commits or its handler ends, and a serial reader's
|
|
next command waits for that too (``burst_gate``): a non-burst busy input (a second video, a voice
|
|
note, text outside queue mode) has no burst commit task to wait on."""
|
|
delivery_event = replace(event, source=replace(event.source))
|
|
accepted = asyncio.get_running_loop().create_future()
|
|
committed = asyncio.get_running_loop().create_future()
|
|
|
|
async def observe():
|
|
# Like the first message's lifecycle, this sender adopts the turn's crash marker and
|
|
# releases it only once the reply is ledgered (the marker moves onto ``event``).
|
|
event._turn_marker_handoff = getattr(adapter, 'gateway_runner', None) is not None
|
|
try:
|
|
response = await adapter._message_handler(event)
|
|
# The ledgered send releases the marker off the event it sends.
|
|
_move_marker(event, delivery_event)
|
|
delivery_event._turn_marker_handoff = event._turn_marker_handoff
|
|
await deliver_response(adapter, delivery_event, session_key, response)
|
|
except Exception as exc:
|
|
if not accepted.done():
|
|
raise # nothing was queued: the receive loop sees the failure as before
|
|
logger.error('[%s] Queued busy follow-up failed: %s', adapter.name, exc, exc_info=True)
|
|
await adapter._notify_turn_error(delivery_event, exc)
|
|
finally:
|
|
settle_busy_arrival(event)
|
|
# Clears the marker if no ledgered send did, then wakes the drain's bounded wait,
|
|
# which listens on the adopter (``event``).
|
|
await adapter._release_turn_marker(delivery_event)
|
|
await adapter._release_turn_marker(event)
|
|
|
|
token, commit_token = busy_acceptance.set(accepted), busy_commit.set(committed)
|
|
try:
|
|
task = asyncio.create_task(observe())
|
|
finally:
|
|
busy_acceptance.reset(token)
|
|
busy_commit.reset(commit_token)
|
|
adapter._background_tasks.add(task)
|
|
task.add_done_callback(adapter._background_tasks.discard)
|
|
try:
|
|
await asyncio.wait({accepted, task}, return_when=asyncio.FIRST_COMPLETED)
|
|
if durable and not task.done():
|
|
await asyncio.wait({committed, task}, return_when=asyncio.FIRST_COMPLETED)
|
|
except asyncio.CancelledError:
|
|
if not accepted.done():
|
|
task.cancel() # cancelled before commit: the input was never ACKed
|
|
raise
|
|
if task.done():
|
|
task.result() # refused, consumed or answered inline before any durable acceptance
|
|
elif not committed.done():
|
|
_hold_until_committed(adapter, committed, task)
|
|
|
|
|
|
def _hold_until_committed(adapter, committed, task):
|
|
"""Released before its commit: the reader reads on, but this input is not durable yet. A serial
|
|
reader's next command waits for its commit or its handler's end (``burst_gate``), so /restart never
|
|
closes admission, and /new never cancels the chat's rows, before it lands."""
|
|
settled = asyncio.get_running_loop().create_future()
|
|
|
|
def settle(_):
|
|
if not settled.done():
|
|
settled.set_result(None)
|
|
committed.add_done_callback(settle)
|
|
task.add_done_callback(settle)
|
|
held = adapter.__dict__.setdefault('_uncommitted_inputs', set())
|
|
held.add(settled)
|
|
settled.add_done_callback(held.discard)
|