fix(gateway): run local turns' post-delivery callbacks; stop pinning each local session's last turn

A messaging reply's adapter task (_process_message_background) pops and runs the post-delivery
callbacks a turn registers on its delivery adapter: the background-review release, temp-bubble
cleanup and the deferred /goal and /loop verdict notices. Desktop/TUI/CLI turns run through the
session authority's drain (execute_admission) with no such task, so on LocalSessionAdapter nothing
ever ran or popped them:

- the "Goal achieved"/"Goal continuing" verdict the gateway judged never reached a local viewer;
- the per-route slot kept every local session's last TurnContext alive until restart, and with it
  that turn's AIAgent (after the agent cache evicted it), history and result messages.

execute_admission now moves a non-native turn's callback off the adapter slot onto its admission,
and the drain runs it once the turn settled and its reply went out (after deliver_settled), or on
the unsettled and cancelled exits. Native messaging turns keep running them in their adapter task.

Sibling per-session state the local adapter/authority never shrank on delete: retire_live_sessions
now drops the deleted session's LocalSessionAdapter source, frozen policy and post-delivery slot and
its launch_sources entry; a detached conversation that returns to its chat drops its detached
admission source.

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

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)

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

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:

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)

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
 

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)

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()