diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index 36a708992d..e6f78ebece 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -432,6 +432,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from gateway.config import Platform, PlatformConfig from gateway.platforms.helpers import fence_state_after +from gateway.platforms.base_post_delivery import run_post_delivery_callback from gateway.platforms.base_exec_approval import ( approval_timeout_seconds, ea_action_labels, ea_default_reason_text, ea_header_text, ea_reason_label_text, ea_smart_deny_line_text, format_approval_deadline_line) @@ -4430,13 +4431,9 @@ class BasePlatformAdapter(ABC): """Run the one-shot post-delivery callback (bounded, errors swallowed). The generation is read HERE — stamped on the interrupt event DURING the handler await; an earlier snapshot would let stale runs fire a fresher run's callbacks.""" - _post_cb = self.pop_post_delivery_callback( - session_key, generation=getattr(interrupt_event, "_hermes_run_generation", None)) - if callable(_post_cb): - with contextlib.suppress(asyncio.TimeoutError, Exception): - _post_result = _post_cb() - if inspect.isawaitable(_post_result): - await asyncio.wait_for(_post_result, timeout=_POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS) + await run_post_delivery_callback(self.pop_post_delivery_callback( + session_key, generation=getattr(interrupt_event, "_hermes_run_generation", None)), + _POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS) def _finish_session_task(self, session_key: str, interrupt_event: asyncio.Event) -> None: """End-of-task guard/ownership reconciliation. A late ``_pending_messages`` arrival must not diff --git a/gateway/platforms/base_post_delivery.py b/gateway/platforms/base_post_delivery.py new file mode 100644 index 0000000000..07045b4a97 --- /dev/null +++ b/gateway/platforms/base_post_delivery.py @@ -0,0 +1,19 @@ +"""Running a turn's one-shot post-delivery callback (``BasePlatformAdapter.register_post_delivery_callback``). + +Shared by the adapter task that delivered a messaging reply (``_process_message_background``) and the +session authority's drain, which settles every turn no adapter task brackets (``session_ingress``). +""" +import asyncio +import contextlib +import inspect +from typing import Any + + +async def run_post_delivery_callback(callback: Any, timeout: float) -> None: + """Run *callback* (a no-op when None): bounded, errors swallowed, so a stuck or failing callback + never holds up the session whose reply it follows.""" + if callable(callback): + with contextlib.suppress(asyncio.TimeoutError, Exception): + result = callback() + if inspect.isawaitable(result): + await asyncio.wait_for(result, timeout=timeout) diff --git a/gateway/session_authority.py b/gateway/session_authority.py index f09f3cde3f..7da7310508 100644 --- a/gateway/session_authority.py +++ b/gateway/session_authority.py @@ -1316,6 +1316,8 @@ class SessionAuthority: # replayed, so its deferred reset is dropped (the next turn re-detects the overflow), # never left to fire after a later settled turn with a stale entry. self.deferred_resets.pop(live.route, None) + from gateway.session_ingress import fire_post_delivery + await fire_post_delivery(self, admission_id) # cancelled mid-turn: as an adapter task's finally if isinstance(exc, ManagedExecutionUnknown): return raise @@ -1393,13 +1395,16 @@ class SessionAuthority: # Uncommitted (unknown or discarded): the user is not told an answer the FIFO lost. self.pending_deliveries.pop(admission_id, None) drop_producer_turn(self, admission_id) + from gateway.session_ingress import fire_post_delivery + await fire_post_delivery(self, admission_id) from gateway.session_ingress import waiter_replies waiter_replies(self).pop(admission_id, None) # Taken on every exit, so a turn that never settled leaves no reset for a later one. reset = self.deferred_resets.pop(live.route, None) await self._run_deferred_reset(live.route, reset) - from gateway.session_ingress import deliver_settled + from gateway.session_ingress import deliver_settled, fire_post_delivery await deliver_settled(self, admission_id) + await fire_post_delivery(self, admission_id) await self._resolve_settled_waiter(admission_id, response) @staticmethod diff --git a/gateway/session_detached.py b/gateway/session_detached.py index 17e926d46b..c428607e6a 100644 --- a/gateway/session_detached.py +++ b/gateway/session_detached.py @@ -84,6 +84,11 @@ def forget_detached(authority, key) -> None: if evict is not None: evict(key) drop_owner_hints(runner.session_store, key) + from gateway.config import Platform + from gateway.session_local_recovery import local_adapter_map + local = local_adapter_map(authority).get(Platform.LOCAL) + if local is not None: + local.sources.pop(key, None) # what admitted the detached copy's background results def drop_owner_hints(store, key) -> None: diff --git a/gateway/session_ingress.py b/gateway/session_ingress.py index 32e04a36b8..aa8984c653 100644 --- a/gateway/session_ingress.py +++ b/gateway/session_ingress.py @@ -190,6 +190,28 @@ async def deliver_resolved(authority, admission_id, *, owed): await deliver_settled(authority, admission_id) +def _take_post_delivery(authority, admission_id, source, route): + """Move the turn's post-delivery callback off its adapter's per-route slot onto this admission.""" + adapter = authority.runner._delivery_adapter_for(source) if source is not None else None + callback = adapter.pop_post_delivery_callback(route) if adapter is not None else None + if callback is not None: + _registry(authority, 'post_delivery')[admission_id] = callback + + +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.""" + callback = _registry(authority, 'post_delivery').pop(admission_id, None) + if callback is None: + return + from gateway.platforms.base import _POST_DELIVERY_CALLBACK_TIMEOUT_SECONDS + from gateway.platforms.base_post_delivery import run_post_delivery_callback + 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) @@ -339,6 +361,10 @@ async def execute_admission(authority, ref, row): 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, live.route) if not handed: await _release_turn_marker(authority, event) executing_admission.reset(token) diff --git a/gateway/session_local.py b/gateway/session_local.py index 065036f481..242530de48 100644 --- a/gateway/session_local.py +++ b/gateway/session_local.py @@ -52,6 +52,16 @@ class LocalSessionAdapter(BasePlatformAdapter): def register_source(self, source): self.sources[self._source_key(source)] = source, self._identity(source) + def forget(self, source, route): + """A deleted conversation's admission source, frozen policy and post-delivery slot (a local + route names one conversation; a detached copy keeps only its own key, its chat's stays).""" + key = self._source_key(source) + if self.sources.get(key, (None,))[0] is source: + del self.sources[key] + if key == source.chat_id and source.platform == Platform.LOCAL: + self.policies.pop(key, None) + self._post_delivery_callbacks.pop(route, None) + async def connect(self, *, is_reconnect=False): return True diff --git a/gateway/session_mutations.py b/gateway/session_mutations.py index f033f559ea..b2ae2ac4cf 100644 --- a/gateway/session_mutations.py +++ b/gateway/session_mutations.py @@ -425,16 +425,22 @@ def retire_live_sessions(authority, session_ids, *, require_fence=False): release_session_workspaces(authority, session_ids) from tools.async_delegation import allow_for_session from gateway.session_vault_prompt import release_unlocks + from gateway.config import Platform + from gateway.session_local_recovery import local_adapter_map + local = local_adapter_map(authority).get(Platform.LOCAL) for sid in session_ids: # A closed session's refusal of new delegations (session_viewer_close) goes with its live state. allow_for_session((authority.profile_id, sid)) # So do the password managers it unlocked: this path never reaches the viewer-close settle. release_unlocks(sid) + authority.launch_sources.pop(sid, None) candidate = authority.sessions.pop(sid, None) if candidate is not None: # a detached conversation's route is its own key evict = getattr(authority.runner, '_evict_cached_agent', None) if callable(evict): evict(candidate.route) + if candidate.source is not None and local is not None: + local.forget(candidate.source, candidate.route) if candidate.detached: # the ids its key owned in the store (session_detached.detached_entry) from gateway.session_detached import drop_owner_hints drop_owner_hints(authority.runner.session_store, candidate.route) diff --git a/tests/gateway/test_local_post_delivery.py b/tests/gateway/test_local_post_delivery.py new file mode 100644 index 0000000000..cb59c6be57 --- /dev/null +++ b/tests/gateway/test_local_post_delivery.py @@ -0,0 +1,97 @@ +"""A local session's turn runs its post-delivery callbacks once it settles, and a deleted local session +leaves nothing keyed by it on the local adapter. + +A messaging reply's adapter task (``_process_message_background``) runs the callbacks a turn registered +(background-review release, temp-bubble cleanup, a deferred /goal verdict) after its send. A Desktop/TUI/CLI +turn runs through the authority's drain with no such task: the callbacks must run there, or they never +fire (the /goal verdict never reaches the viewer) and the adapter's per-route slot keeps the turn's whole +context alive until restart. +""" +import pytest + +from tests.waiting import async_wait_until + + +async def _owner(tmp_path, monkeypatch): + from gateway.config import GatewayConfig + from gateway.run import GatewayRunner + from gateway.session_authority import initialize_session_authority + from hermes_state import SessionDB + + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + runner = GatewayRunner(GatewayConfig(sessions_dir=tmp_path / "sessions")) + db = SessionDB(tmp_path / "state.db") + runner.session_store._db = runner._session_db = db + authority = await initialize_session_authority(runner, profile_id="default", instance_id="owner", db=db) + return runner, authority, db + + +def _viewer(): + from gateway.session_contract import Principal + return Principal("test-owner", "default", frozenset( + {"session:create", "session:submit", "session:read", "session:control"}), "viewer") + + +@pytest.mark.asyncio +async def test_local_turn_runs_its_post_delivery_callbacks_after_settling(tmp_path, monkeypatch): + from gateway.session_contract import Submission + from gateway.session_local import create_local_session + from hermes_state_runtime import get_session_admission + + runner, authority, db = await _owner(tmp_path, monkeypatch) + fired = [] + + async def turn(event): + live = authority.sessions[ref.session_id] + adapter = runner._delivery_adapter_for(live.source) + admission = event.message_id + + def after_delivery(): # what TurnRunner / the /goal judge register for the turn + fired.append((event.text, get_session_admission(db, admission_id=admission)["status"])) + adapter.register_post_delivery_callback(live.route, after_delivery, generation=1) + if event.text == "fails": + raise RuntimeError("turn failed") + return "ok" + + monkeypatch.setattr(runner, "_handle_message", turn) + try: + ref = create_local_session(authority, _viewer(), dict(request_id="r", source="tui", model="fixture", + toolsets=[])) + live = authority.sessions[ref.session_id] + for text in ("answers", "fails"): + await authority.submit(_viewer(), Submission(text, ref, {"text": text}, "queue")) + await async_wait_until(lambda: text in [t for t, _ in fired], timeout=10, what=f"the {text} turn") + adapter = runner._delivery_adapter_for(live.source) + assert adapter._post_delivery_callbacks == {} + finally: + db.close() + # Once each, after the turn's outcome committed: the reply was out before they ran. + assert fired == [("answers", "terminal"), ("fails", "terminal")] + + +@pytest.mark.asyncio +async def test_deleted_local_session_leaves_nothing_on_the_local_adapter(tmp_path, monkeypatch): + from gateway.session_local import create_local_session + from gateway.session_mutations import mutate_session + from gateway.session_queued_controls import _interactive + + runner, authority, db = await _owner(tmp_path, monkeypatch) + try: + refs = [create_local_session(authority, _viewer(), dict(request_id=f"r{i}", source="gui", model="fixture", + toolsets=[])) for i in range(2)] + kept, deleted = (authority.sessions[ref.session_id] for ref in refs) + adapter = runner._delivery_adapter_for(kept.source) + for live in (kept, deleted): + adapter.register_post_delivery_callback(live.route, lambda: None, generation=1) + with live.event_stream.lock: + _interactive(authority, live.source.chat_id) + result = await mutate_session(authority, _viewer(), refs[1], dict( + session_id=refs[1].session_id, request_id="delete", expected_revision=0, + expected_generation=0, operation="delete", payload={})) + assert result.get("deleted_ids") == [refs[1].session_id] + assert set(adapter.sources) == set(adapter.policies) == set(authority.launch_sources) == { + refs[0].session_id} + assert set(adapter._post_delivery_callbacks) == {kept.route} + assert adapter.authorize_source(kept.source) + finally: + db.close()