fix(gateway): a resume reading history while a fresh launch claims the chat keeps it

attach took a fresh launch's claim back before prepare_history, which yields to a worker thread. A
resume could pass the take-back before any claim existed, pause while the start found the chat
unwatched and claimed it, then join; the claim stood and the settle write deleted the reopened chat.

The take-back now runs after attach's last other await and never yields itself, so the viewer joins
in the same loop step: the start (which claims only on the loop) either claimed first and the claim
is taken back, or sees the viewer and claims nothing. A delete in flight still holds the lock across
its write; the take-back waits for it off the loop and retries, and the re-check answers not_found.

gateway/session_authority.py

@@ -322,11 +322,13 @@ class SessionAuthority:
 
     async def attach(self, actor, ref):
         self.authorize(actor, ref, 'session:read')
-        from gateway.session_onboarding import keep_from_discard
-        await keep_from_discard(self, ref.session_id)  # a delete already past it answers not_found below
         from gateway.session_local_recovery import local_history, prepare_history
         prepared = await prepare_history(self, ref)  # the lineage read stays off the loop and the lock
-        # A delete or eviction can land during that await: re-check, so it answers not_found, never a KeyError.
+        from gateway.session_onboarding import keep_from_discard
+        # The last await before the viewer joins: a fresh launch's claim is taken back with no yield
+        # before the join, and a delete already past it answers not_found below.
+        await keep_from_discard(self, ref.session_id)
+        # A delete or eviction can land during those awaits: re-check, so it answers not_found, never a KeyError.
         self.authorize(actor, ref, 'session:read')
         live = self.sessions[ref.session_id]
         from gateway.session_queued_controls import pending_rows

gateway/session_onboarding.py

@@ -179,14 +179,21 @@ def _discards(authority):
 
 
 async def keep_from_discard(authority, session_id):
-    """``attach``, before the viewer joins: a chat a fresh launch was about to delete is kept. Off the loop,
-    since a delete in flight holds the lock across its write."""
+    """``attach``, after its last other await: a chat a fresh launch was about to delete is kept. The
+    take-back itself never yields, so the viewer joins in the same loop step and a later start sees it
+    watched; a start claims only on the loop. A delete in flight holds the lock across its write, so
+    that wait runs off the loop and the take-back is retried after it."""
     discards = getattr(authority, '_onboarding_discards', None)
-    if discards is not None and session_id in discards.ids:
-        def keep():
+    while discards is not None and session_id in discards.ids:
+        if discards.lock.acquire(blocking=False):
+            discards.ids.discard(session_id)
+            discards.lock.release()
+            return
+
+        def delete_finished():
             with discards.lock:
-                discards.ids.discard(session_id)
-        await asyncio.to_thread(keep)
+                pass
+        await asyncio.to_thread(delete_finished)
 
 
 def _settle_runs(authority, request, plan, stored, discards):

tests/gateway/test_onboarding_on_the_authority.py

@@ -366,3 +366,59 @@ async def test_a_window_that_reopens_the_old_chat_while_a_fresh_launch_starts_ke
     finally:
         await other.close()
         await desktop.connection.close()
+
+
+@pytest.mark.asyncio
+async def test_a_window_whose_resume_reads_history_while_a_fresh_launch_claims_the_chat_keeps_it(tmp_path, monkeypatch):
+    """Another window's resume was reading the old chat's history when the fresh launch's start found
+    the chat unwatched and claimed it; the resume finished before the delete ran. The chat stays with its
+    messages, and its run ends as replaced."""
+    import threading
+    from agent import onboarding_record
+    from gateway import session_local_recovery
+    from gateway.session_controls import AuthorityConnection
+    turns = []
+    desktop = await _desktop(tmp_path, monkeypatch, turns)
+    first = (await _request(desktop, 1, "onboarding.start", {"rerun": True}))["result"]["stored_session_id"]
+    await async_wait_until(lambda: turns, timeout=10)
+    await _settled(desktop, first)
+    desktop.db.append_message(first, "user", "Call me Sid.")
+    messages = desktop.db.get_messages(first)
+    await desktop.connection.close()
+    identity = {"user_id": "uid:1", "provider": "local", "profile_id": desktop.authority.profile_id,
+                "instance_id": "owner"}
+    desktop.connection = AuthorityConnection(desktop.authority, desktop.transport, identity)
+    other = AuthorityConnection(desktop.authority, _Transport(), identity)
+    reading, release, resume_done = threading.Event(), threading.Event(), threading.Event()
+    read_rows = session_local_recovery.display_rows
+
+    def history_read_held_open(db, target):
+        if not reading.is_set():
+            reading.set()
+            assert release.wait(timeout=20), "the history read was never released"
+        return read_rows(db, target)
+
+    record_new_run = onboarding_record.start
+
+    def new_run_then_the_resume_finishes(db, stored, facts):
+        record_new_run(db, stored, facts)
+        release.set()
+        assert resume_done.wait(timeout=20), "the resume never finished"
+
+    monkeypatch.setattr(session_local_recovery, "display_rows", history_read_held_open)
+    resume = asyncio.ensure_future(other.dispatch({
+        "jsonrpc": "2.0", "id": 9, "method": "session.resume", "params": {"session_id": first}}))
+    resume.add_done_callback(lambda _: resume_done.set())
+    try:
+        await async_wait_until(reading.is_set, timeout=10)
+        monkeypatch.setattr(onboarding_record, "start", new_run_then_the_resume_finishes)
+        second = (await _request(desktop, 2, "onboarding.start", {"fresh_launch": True}))["result"]
+        assert second["stored_session_id"] != first
+        assert "result" in (await resume), resume.result()
+        assert desktop.db.get_messages(first) == messages
+        assert (onboarding_record.load(desktop.db, first) or {}).get("ended_by") == "replaced"
+        assert first in desktop.authority.sessions
+    finally:
+        release.set()
+        await other.close()
+        await desktop.connection.close()