fix(gateway): a delete committed before its retirement runs answers not_found on resume

Resume re-authorizes after its off-loop history read, so a delete that
retired the session meanwhile answers not_found. A delete commits off
the loop, though, and its retirement runs only when the loop gets back
to it. In between, the live entry is still there and passes the
re-authorize, but the row is gone:
- session.resume read the missing row in _handle and raised TypeError,
  an internal error (-32603) for the viewer;
- the hosted room's history read met the deleted receipt and answered
  storage_unavailable.

_handle now raises not_found for a missing row, before the viewer
joins, and unless_deleted turns the history read's refusal into
not_found when the session row is gone. This also covers a delete that
starts cold, and nothing waits: holding the session's mutation lock
instead would make a resume wait behind a model switch or MCP discovery.

Refs #134889
Refs #140449

(cherry picked from commit fc14e2a30b60b8d667afde7d411cb2113441d744)

gateway/session_authority.py

@@ -248,6 +248,8 @@ class SessionAuthority:
 
     def _handle(self, ref):
         row = self.db.get_session(ref.session_id)
+        if row is None:  # deleted after ``authorize``: the commit lands before the retirement runs
+            raise RuntimeStoreError('not_found')
         pending = self._pending_rows(ref)
         state = 'unknown' if any(r['status'] == 'unknown' for r in pending) else (
             'running' if any(r['status'] == 'started' for r in pending) else 'idle')
@@ -258,6 +260,17 @@ class SessionAuthority:
         self.authorize(actor, ref, 'session:read')
         return self._handle(ref)
 
+    def unless_deleted(self, ref, read):
+        """``read()``; a refusal because *ref* was deleted after ``authorize`` passed is ``not_found``.
+        A delete commits off the loop, so its rows can be gone while the live entry still waits for its
+        retirement, and readers meet the missing receipt instead."""
+        try:
+            return read()
+        except RuntimeStoreError:
+            if self.db.get_session(ref.session_id) is None:
+                raise RuntimeStoreError('not_found') from None
+            raise
+
     async def attach(self, actor, ref):
         self.authorize(actor, ref, 'session:read')
         from gateway.session_local_recovery import local_history, prepare_history
@@ -267,6 +280,7 @@ class SessionAuthority:
         live = self.sessions[ref.session_id]
         from gateway.session_queued_controls import pending_rows
         with live.event_stream.lock:
+            handle = self._handle(ref)  # before the viewer joins: a delete committed meanwhile is not_found
             subscription = next((key for key, member in live.subscribers.items()
                                  if member == actor), None) or uuid.uuid4().hex
             live.subscribers[subscription] = actor
@@ -277,12 +291,11 @@ class SessionAuthority:
             if transport is not None:
                 live.event_stream.fanout.attach(transport)
                 live.event_stream.on_overflow = partial(self._retire_overflowed, ref.session_id)
-            handle = self._handle(ref)
             active_generation = handle.execution_generation if handle.execution_state == "running" else None
             prompts = live.controls.snapshot(ref.session_id, active_generation)
             epoch, sequence = live.event_stream.watermark()
-            snapshot = SubscriptionSnapshot(subscription, handle, epoch,
-                                        sequence, tuple(local_history(self, ref, prepared)),
+            history = self.unless_deleted(ref, lambda: tuple(local_history(self, ref, prepared)))
+            snapshot = SubscriptionSnapshot(subscription, handle, epoch, sequence, history,
                                         tuple(self._pending_receipt(r) for r in pending_rows(self, ref)), prompts)
         from gateway.session_viewer_close import cancel_persisted_close
         await asyncio.to_thread(cancel_persisted_close, self, ref.session_id)  # the viewer cancels it

gateway/session_hosted_rpc.py

@@ -203,7 +203,8 @@ class HostedRoomAuthorityRPC:
         self.authority.authorize(self.principal, self.ref, 'session:read')
         prepared = await prepare_history(self.authority, self.ref)
         rows = self._rows()
-        history = list(local_history(self.authority, self.ref, prepared))
+        history = self.authority.unless_deleted(
+            self.ref, lambda: list(local_history(self.authority, self.ref, prepared)))
         for row, task, generation in rows:
             if row['status'] == 'terminal':
                 receipt = self._terminal(row, task, generation)

tests/gateway/test_resume_display_history.py

@@ -5,6 +5,8 @@ turn stays gone, a merged prompt shows once, and the archived part of the reply
 Real store, authority, local session, compaction verbs and ``AuthorityConnection.resume``; the runner
 is a stand-in with no cached agent, and the resume ``info`` block is stubbed out.
 """
+import asyncio
+import threading
 from types import SimpleNamespace
 
 import pytest
@@ -227,6 +229,52 @@ async def test_a_delete_during_the_resume_prepare_answers_not_found(tmp_path, mo
     assert reply.get('error', {}).get('message') == 'not_found', reply
 
 
+@pytest.mark.asyncio
+async def test_a_committed_but_unretired_delete_during_the_prepare_answers_not_found(tmp_path, monkeypatch):
+    """The delete's commit lands off the loop before its retirement runs: the live entry is still there,
+    so the re-authorize passes, but the row is gone. The resume answers ``not_found``, not an internal error."""
+    from gateway import session_local_recovery, session_mutations
+    from gateway.session_controls import AuthorityConnection
+    authority, _actor, ref, physical = await _local_session(tmp_path, monkeypatch)
+    authority.runner.session_authority = authority  # one process owns both writes of a delete
+    authority.db.append_message(physical, 'user', 'hello')
+
+    def connection():
+        return AuthorityConnection(authority, object(), {
+            'provider': 'local', 'user_id': 'owner', 'profile_id': 'fixture', 'instance_id': authority.instance_id,
+            'capabilities': ['session:create', 'session:read', 'session:control']})
+    prepare, write, deletes = session_local_recovery.prepare_history, session_mutations.mutate_runtime_session, []
+    committed, release = threading.Event(), threading.Event()
+
+    def write_then_hold(*args, **kwargs):
+        result = write(*args, **kwargs)
+        if kwargs.get('operation') == 'delete':
+            committed.set()
+            assert release.wait(10), 'the delete was never released'
+        return result
+
+    async def prepare_then_delete(authority_, ref_):
+        prepared = await prepare(authority_, ref_)
+        row = authority.db.get_session(ref.session_id)
+        deletes.append(asyncio.ensure_future(connection().dispatch({'id': 9, 'method': 'session.mutate', 'params': {
+            'session_id': ref.session_id, 'request_id': 'delete-1', 'operation': 'delete', 'payload': {},
+            'expected_revision': row['runtime_revision'], 'expected_generation': row['runtime_generation']}})))
+        assert await asyncio.to_thread(committed.wait, 10)
+        assert authority.db.get_session(ref.session_id) is None and ref.session_id in authority.sessions
+        release.set()  # the commit has landed; the retirement runs once the loop gets to it
+        return prepared
+    monkeypatch.setattr(session_mutations, 'mutate_runtime_session', write_then_hold)
+    monkeypatch.setattr(session_local_recovery, 'prepare_history', prepare_then_delete)
+    try:
+        reply = await connection().dispatch({'id': 1, 'method': 'session.resume',
+                                             'params': {'session_id': ref.session_id}})
+    finally:
+        release.set()
+    assert 'result' in await deletes[0], deletes
+    assert reply.get('error', {}).get('message') == 'not_found', reply
+    assert ref.session_id not in authority.sessions
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('mode', ['in_place', 'rotated'])
 async def test_rollback_turn_rewind_never_targets_an_archived_turn(tmp_path, monkeypatch, mode):

tests/gateway/test_session_hosted_rpc.py

@@ -332,3 +332,51 @@ def test_room_member_turns_cannot_block_on_an_unanswerable_clarify(hosted_owner,
         enabled_toolsets=list(policy.toolsets), disabled_toolsets=disabled or None, quiet_mode=True)}
     assert 'terminal' in tools
     assert 'clarify' not in tools
+
+
+def test_history_read_during_a_room_session_delete_is_not_found(hosted_owner, monkeypatch):
+    """A delete whose commit lands while the history read is off the loop, before its retirement runs,
+    answers ``not_found``: the read waits for the delete's mutation lock before it uses the entry."""
+    from gateway import session_local_recovery, session_mutations
+    from gateway.session_contract import SessionRef
+    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
+    from gateway.session_mutations import mutate_session
+    from hermes_state_runtime import RuntimeStoreError
+    authority, loop, principal, _ = hosted_owner
+    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default',
+                                 principal=principal, authorize=lambda *args: True)
+    coords = dict(profile='default', source='bot_room')
+    sid = rpc.create(**coords, title='Group: room')['session_id']
+    ref = SessionRef(authority.profile_id, sid)
+    prepare, write, deletes = session_local_recovery.prepare_history, session_mutations.mutate_runtime_session, []
+    committed, release = threading.Event(), threading.Event()
+
+    def write_then_hold(*args, **kwargs):
+        result = write(*args, **kwargs)
+        if kwargs.get('operation') == 'delete':
+            committed.set()
+            assert release.wait(10), 'the delete was never released'
+        return result
+
+    async def prepare_then_delete(authority_, ref_):
+        prepared = await prepare(authority_, ref_)
+        handle = authority._handle(ref)
+        deletes.append(asyncio.ensure_future(mutate_session(authority, principal, ref, {
+            'session_id': sid, 'request_id': 'delete', 'operation': 'delete', 'payload': {},
+            'expected_revision': handle.revision, 'expected_generation': handle.execution_generation})))
+        assert await asyncio.to_thread(committed.wait, 10)
+        assert authority.db.get_session(sid) is None and sid in authority.sessions
+        release.set()  # the commit has landed; the retirement runs once the loop gets to it
+        return prepared
+    monkeypatch.setattr(session_mutations, 'mutate_runtime_session', write_then_hold)
+    monkeypatch.setattr(session_local_recovery, 'prepare_history', prepare_then_delete)
+    try:
+        with pytest.raises(RuntimeStoreError, match='not_found'):
+            rpc.history(**coords, session_id=sid)
+    finally:
+        release.set()
+
+    async def deleted():
+        return await deletes[0]
+    asyncio.run_coroutine_threadsafe(deleted(), loop).result(10)
+    assert sid not in authority.sessions