From ec5b9dab21c50bfbdd14dc74cf8848a60d01bf4d Mon Sep 17 00:00:00 2001 From: alt-glitch Date: Thu, 22 Oct 2026 15:31:26 +0530 Subject: [PATCH] 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 | 8 ++- gateway/session_onboarding.py | 19 +++++-- .../test_onboarding_on_the_authority.py | 56 +++++++++++++++++++ 3 files changed, 74 insertions(+), 9 deletions(-) diff --git a/gateway/session_authority.py b/gateway/session_authority.py index ac386f0e8a..3bc6c596c7 100644 --- a/gateway/session_authority.py +++ b/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 diff --git a/gateway/session_onboarding.py b/gateway/session_onboarding.py index 71c35d3aa4..b80c5fac3a 100644 --- a/gateway/session_onboarding.py +++ b/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): diff --git a/tests/gateway/test_onboarding_on_the_authority.py b/tests/gateway/test_onboarding_on_the_authority.py index 28bf349645..a265f3ac2a 100644 --- a/tests/gateway/test_onboarding_on_the_authority.py +++ b/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()