fix(gateway): take post-delivery by the turn's own key; drop closed-session retirement from this PR

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.

gateway/session_authority.py

@@ -459,9 +459,6 @@ class SessionAuthority:
         wake_bot_receipts(self, ref.session_id)
 
     def _schedule(self, ref):
-        if ref.session_id not in self.sessions:
-            from gateway.session_viewer_close import restore_retired
-            restore_retired(self, ref.session_id)
         live = self.sessions[ref.session_id]
         live.rescan = True
         if live.task is None or live.task.done():

gateway/session_ingress.py

@@ -291,6 +291,8 @@ async def execute_admission(authority, ref, row):
     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']:
@@ -381,7 +383,7 @@ async def execute_admission(authority, ref, row):
         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)
+            _take_post_delivery(authority, row['admission_id'], event.source, route)
         if not handed:
             await _release_turn_marker(authority, event)
         executing_admission.reset(token)

gateway/session_local.py

@@ -137,9 +137,11 @@ class LocalSessionAdapter(BasePlatformAdapter):
 def forget_live_session(authority, session_id, live):
     """What the local adapter and the authority keyed by a deleted live session."""
     authority.launch_sources.pop(session_id, None)
+    if live.source is None or live.source.platform != Platform.LOCAL:
+        return
     from gateway.session_local_recovery import local_adapter_map
     local = local_adapter_map(authority).get(Platform.LOCAL)
-    if local is not None and live.source is not None:
+    if local is not None:
         local.forget(live.source, live.route)
 
 

gateway/session_viewer_close.py

@@ -320,16 +320,6 @@ async def settle_viewer_close(authority, session_id):
         live.event_stream.release()
 
 
-def restore_retired(authority, session_id):
-    """Input admitted for a conversation ``_retire_closed`` let go (its frozen policy is still on the
-    local adapter) brings it back from its receipt before the FIFO runs it."""
-    from gateway.config import Platform
-    from gateway.session_local_recovery import local_adapter_map, restore_local_session
-    local = local_adapter_map(authority).get(Platform.LOCAL)
-    if local is not None and session_id in local.policies:
-        restore_local_session(authority, session_id)
-
-
 async def _resolve_abandoned_unknown(authority, session_id, live, close):
     """A turn the previous owner lost (hard kill) recovers ``unknown`` and pauses the FIFO until
     someone Discards it, but the last viewer closed the session: nobody is left to. The viewer's

tests/gateway/test_local_post_delivery.py

@@ -69,6 +69,34 @@ async def test_local_turn_runs_its_post_delivery_callbacks_after_settling(tmp_pa
     assert fired == [("answers", "terminal"), ("fails", "terminal")]
 
 
+@pytest.mark.asyncio
+async def test_post_delivery_is_taken_by_the_key_the_turn_registered_under(tmp_path, monkeypatch):
+    """A route rebind while the turn runs must not strand the slot under the key it registered with."""
+    from gateway.session_contract import Submission
+    from gateway.session_local import create_local_session
+
+    runner, authority, db = await _owner(tmp_path, monkeypatch)
+    fired = []
+
+    async def turn(event):
+        live = authority.sessions[ref.session_id]
+        runner._delivery_adapter_for(live.source).register_post_delivery_callback(
+            live.route, lambda: fired.append(event.text), generation=1)
+        live.route = live.route + ":rebound"
+        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]
+        await authority.submit(_viewer(), Submission("answers", ref, {"text": "answers"}, "queue"))
+        await async_wait_until(lambda: fired == ["answers"], timeout=10, what="the post-delivery callback")
+        assert runner._delivery_adapter_for(live.source)._post_delivery_callbacks == {}
+    finally:
+        db.close()
+
+
 @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