From 7836beeace624f01ffc9cb78b6eb30e78d4d36ae Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Tue, 27 Oct 2026 17:12:01 +0800 Subject: [PATCH 1/7] fix(chat): keep a completed Turn's event history when a stop races it interrupt_turn reads the Turn, appends the durable interrupt-intent event, and only then compare-and-sets the status. When the Turn completed in between, the CAS refuses the change but the intent event remains in the completed Turn's ledger, because #7191 moved the intent ahead of the status write without keeping the terminal-state check that write used to provide. That breaks the terminal event history pinned by test_interrupt_does_not_overwrite_a_completed_turn and makes the completed Turn report stop_observed, which suppresses the unfinished-work continuation in the external-conversation planner. Make the status check, the intent event and the status write one store transition under the Turn write lock. The intent still lands before the status change, so a crash cannot turn a requested stop into an automatic review, and a Turn that already reached a terminal state can no longer gain a stop event after its terminal outcome. Fixes #7476 Signed-off-by: JasonBuildAI --- loopx/chat_runtime.py | 9 +++++---- loopx/chat_store.py | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+), 4 deletions(-) diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 0927144cc9..6baa8a245d 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -2101,13 +2101,14 @@ def interrupt_turn(self, *, session_id: str, turn_id: str) -> dict[str, Any]: return turn # Persist stop intent before changing status or contacting native # IO, so a service crash cannot turn an interrupted stop into an - # automatic review of the work the user just cancelled. - self.store.append_event(session_id, turn_id, kind="turn.interrupt_requested", payload={}) - interrupting = self.store.update_turn( + # automatic review of the work the user just cancelled. The store + # checks the status and writes the intent under the Turn lock, so a + # Turn that completed while this call was reading it keeps its + # terminal event history and is not relabelled as interrupted. + interrupting = self.store.begin_turn_interrupt( session_id, turn_id, expected_statuses={"queued", "starting", "running", "interrupting"}, - status="interrupting", ) if interrupting is None: current = self.store.load_turn(session_id, turn_id) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 53be87ae11..49bb5cb314 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -1416,6 +1416,39 @@ def update_turn( _atomic_write_json(path, payload, preserve_mode=True) return payload + def begin_turn_interrupt( + self, + session_id: str, + turn_id: str, + *, + expected_statuses: set[str], + ) -> dict[str, Any] | None: + """Record stop intent and enter `interrupting` as one Turn transition. + + Holding the Turn write lock across the status check, the durable intent + event and the status write closes the completion race: a Turn that + already reached a terminal state cannot gain a stop-intent event after + its terminal outcome, and a Turn this call does interrupt cannot lose + that intent to a later completion. The event still precedes the status + change, so a crash between them leaves the requested stop recorded + rather than silently turning it into an automatic review. + """ + path = self._turn_path(session_id, turn_id) + with exclusive_file_lock( + path, + agent_id="loopx-chat", + operation="begin_chat_turn_interrupt", + ): + payload = self.load_turn(session_id, turn_id) + if payload is None: + raise KeyError("chat turn was not found") + if payload.get("status") not in expected_statuses: + return None + self.append_event(session_id, turn_id, kind="turn.interrupt_requested", payload={}) + payload["status"] = "interrupting" + _atomic_write_json(path, payload, preserve_mode=True) + return payload + def append_completed_response_events( self, session_id: str, From 43f7ddb1faa25bb84a8c18d3188f85d838847269 Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Tue, 27 Oct 2026 17:27:55 +0800 Subject: [PATCH 2/7] chore(canary): record the reviewed chat module ceilings The stop-intent transition adds 33 lines to chat_store.py and one net line to chat_runtime.py. Refresh the reviewed module metric ceilings in the same diff so the maintainability ratchet settles the accepted growth. Signed-off-by: JasonBuildAI --- loopx/canary/module_metric_baseline.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index db1c0673b2..ebb9fcb1a6 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -36,7 +36,7 @@ "loopx/chat_runtime.py": { "any_count": 37, "dict_any_count": 0, - "lines": 2319 + "lines": 2320 }, "loopx/chat_server.py": { "any_count": 25, @@ -44,7 +44,7 @@ "lines": 2020 }, "loopx/chat_store.py": { - "lines": 2003 + "lines": 2036 }, "loopx/collaboration_mcp.py": { "lines": 2058 From 7ab114fdc3ce85d4d1522768b3df6bcca143ba10 Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Tue, 27 Oct 2026 20:41:35 +0800 Subject: [PATCH 3/7] fix(chat): serialize a queued claim with the Turn interrupt transition `claim_next_queued_turn` wrote the Turn from the snapshot the queue scan had read, under the Session lock only, while `begin_turn_interrupt` read and wrote the same Turn under the Turn write lock. The two writers therefore did not share a serialization boundary: an interrupt that read `queued` could resume after the claim had committed `running` with a `host_claim_id` and had published the Session active pointer, then write its older snapshot back as `interrupting` with no claim id. Take the Turn write lock for the claim too, and re-read the Turn inside it. An interrupt that already committed now wins the transition, so the claim returns `None` instead of starting a Turn the stop request owns, and a claim that commits first is never rewritten from a snapshot that predates it. The queue scan stays a candidate list; the authoritative status check moves under the lock that the interrupt also takes. Cover the interleaving deterministically: the interrupt is paused after its read, the claim is attempted while the interrupt still holds the lock, and the test asserts the persisted status and claim id, the Session active pointer and the event ledger agree. Signed-off-by: JasonBuildAI --- loopx/chat_store.py | 86 +++++++++++++++--------- tests/test_chat_session_active_turn.py | 92 ++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 32 deletions(-) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 49bb5cb314..6acc7dc19b 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -1324,38 +1324,60 @@ def claim_next_queued_turn( self._recover_queued_turn_input(session_id, turn) if host_claim_id: now_text = utc_now() - turn.update( - { - "status": "running", - "host_claim_id": _opaque_id( - host_claim_id, - field="host_claim_id", - ), - **( - { - "admitted_goal_instance_id": _opaque_id( - admitted_goal_instance_id, - field="admitted_goal_instance_id", - ) - } - if admitted_goal_instance_id is not None - else {} - ), - "started_at": turn.get("started_at") or now_text, - "last_activity_at": now_text, - } - ) - _atomic_write_json( - self._turn_path(session_id, turn_id), - turn, - preserve_mode=True, - ) - self.append_event( - session_id, - turn_id, - kind="turn.claimed_by_attached_host", - payload={"host_claim_id": host_claim_id}, - ) + turn_path = self._turn_path(session_id, turn_id) + # Share the Turn write lock that `begin_turn_interrupt` + # takes: the claim is the other writer of this Turn's + # status and authority, so the queue scan above is only + # a candidate. Re-reading under the lock means an + # interrupt that already committed wins the transition + # instead of being overwritten by this write, and a + # claim that commits first is never rewritten from a + # snapshot that predates it. + with exclusive_file_lock( + turn_path, + agent_id="loopx-chat", + operation="claim_queued_chat_turn", + ): + claimed = self.load_turn(session_id, turn_id) + if claimed is None or claimed.get("status") != "queued": + # Another writer owns this Turn now (an + # interrupt, or a claim from another surface). + # Starting a different queued Turn here would + # leave the queue head in an unclaimed state. + return None + claimed.update( + { + "status": "running", + "host_claim_id": _opaque_id( + host_claim_id, + field="host_claim_id", + ), + **( + { + "admitted_goal_instance_id": _opaque_id( + admitted_goal_instance_id, + field="admitted_goal_instance_id", + ) + } + if admitted_goal_instance_id is not None + else {} + ), + "started_at": claimed.get("started_at") or now_text, + "last_activity_at": now_text, + } + ) + _atomic_write_json( + turn_path, + claimed, + preserve_mode=True, + ) + self.append_event( + session_id, + turn_id, + kind="turn.claimed_by_attached_host", + payload={"host_claim_id": host_claim_id}, + ) + turn = claimed session.update( { "status": "busy", diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index 57f6293ef4..ee81b148cf 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -1314,6 +1314,98 @@ def pause_after_interrupt_read( assert (session_id, turn_id) not in runtime.cancelled_turns +def test_queued_claim_and_interrupt_share_one_turn_transition( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A queued claim and an interrupt serialize on the same Turn write lock. + + Before the claim took the Turn lock, the two writers could interleave: the + interrupt read ``queued``, the claim committed ``running`` with a + ``host_claim_id`` and published the Session active pointer, then the + interrupt wrote its older snapshot back as ``interrupting`` with no claim + id. The Session still named the Turn as active, so the host's successful + claim had silently lost its authority. + """ + store = ChatSessionStore(tmp_path) + session = store.create_session( + goal_id="goal-one", + agent_id="codex", + executor_endpoint_id="codex", + adapter_kind="codex_app_server", + upstream_thread_id="thread-one", + upstream_mode="chat", + ) + session_id = str(session["session_id"]) + queued, created = store.create_queued_turn( + session_id, + client_turn_id="claimed-queued-turn", + message="claim me around a stop request", + ) + assert created is True + turn_id = str(queued["turn_id"]) + + original_load_turn = store.load_turn + interrupt_read = threading.Event() + resume_interrupt = threading.Event() + + def pause_after_interrupt_read( + requested_session_id: str, + requested_turn_id: str, + ) -> dict[str, object] | None: + loaded = original_load_turn(requested_session_id, requested_turn_id) + if threading.current_thread().name == "interrupt-turn": + interrupt_read.set() + resume_interrupt.wait(timeout=2) + return loaded + + monkeypatch.setattr(store, "load_turn", pause_after_interrupt_read) + interrupted: list[dict[str, object] | None] = [] + interrupt_thread = threading.Thread( + target=lambda: interrupted.append( + store.begin_turn_interrupt( + session_id, + turn_id, + expected_statuses={"queued", "starting", "running", "interrupting"}, + ) + ), + name="interrupt-turn", + daemon=True, + ) + interrupt_thread.start() + assert interrupt_read.wait(timeout=2) + + claimed: list[dict[str, object] | None] = [] + claim_thread = threading.Thread( + target=lambda: claimed.append( + store.claim_next_queued_turn(session_id, host_claim_id="host-claim") + ), + name="claim-turn", + daemon=True, + ) + claim_thread.start() + resume_interrupt.set() + interrupt_thread.join(timeout=5) + claim_thread.join(timeout=5) + + assert not interrupt_thread.is_alive() + assert not claim_thread.is_alive() + turn = store.load_turn(session_id, turn_id) + assert turn is not None + assert turn["status"] == "interrupting" + assert turn.get("host_claim_id") is None + # The claim must not publish a Turn the interrupt already owns as active. + assert claimed == [None] + current = store.load_session(session_id) + assert current is not None + assert current["active_turn_id"] is None + assert [ + event["kind"] + for event in store.events_after(session_id, turn_id, None) + if event["kind"].startswith("turn.") + ] == ["turn.queued", "turn.interrupt_requested"] + + def test_interrupting_a_queued_turn_does_not_touch_the_active_adapter( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From 203590e5b75530c9f0df403566683927623df764 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 23:56:24 +0800 Subject: [PATCH 4/7] test: qualify claim-first and crash-safe interrupt transitions Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/canary/module_metric_baseline.json | 2 +- tests/test_chat_session_active_turn.py | 39 ++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index ebb9fcb1a6..54f3703f7b 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -44,7 +44,7 @@ "lines": 2020 }, "loopx/chat_store.py": { - "lines": 2036 + "lines": 2058 }, "loopx/collaboration_mcp.py": { "lines": 2058 diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index ee81b148cf..43cfe91217 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -1406,6 +1406,45 @@ def pause_after_interrupt_read( ] == ["turn.queued", "turn.interrupt_requested"] +@pytest.mark.parametrize("crash_after_intent", [False, True]) +def test_interrupt_preserves_a_successful_claim_and_durable_stop_intent( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, crash_after_intent: bool, +) -> None: + store = ChatSessionStore(tmp_path) + session = store.create_session( + goal_id="goal-one", agent_id="codex", executor_endpoint_id="codex", + adapter_kind="codex_app_server", upstream_thread_id="thread-one", upstream_mode="chat", + ) + sid = str(session["session_id"]) + turn, _ = store.create_queued_turn(sid, client_turn_id="claim-first", message="claim then stop") + tid = str(turn["turn_id"]) + claimed = store.claim_next_queued_turn(sid, host_claim_id="host-claim") + assert claimed is not None and claimed["host_claim_id"] == "host-claim" + original_write = chat_store._atomic_write_json + + def write(path, payload, **kwargs): + if path == store._turn_path(sid, tid) and payload.get("status") == "interrupting": + # The intent must already survive independently of the status write. + assert store.events_after(sid, tid, None)[-1]["kind"] == "turn.interrupt_requested" + if crash_after_intent: + raise OSError("synthetic crash after durable intent") + return original_write(path, payload, **kwargs) + + monkeypatch.setattr(chat_store, "_atomic_write_json", write) + if crash_after_intent: + with pytest.raises(OSError, match="synthetic crash after durable intent"): + store.begin_turn_interrupt(sid, tid, expected_statuses={"running"}) + else: + assert store.begin_turn_interrupt(sid, tid, expected_statuses={"running"}) is not None + persisted = store.load_turn(sid, tid) + assert persisted is not None and persisted["host_claim_id"] == "host-claim" + assert persisted["status"] == ("running" if crash_after_intent else "interrupting") + assert store.load_session(sid)["active_turn_id"] == tid + assert [event["kind"] for event in store.events_after(sid, tid, None)] == [ + "turn.queued", "turn.claimed_by_attached_host", "turn.interrupt_requested", + ] + + def test_interrupting_a_queued_turn_does_not_touch_the_active_adapter( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From 7fdf3cade6dd51900bcfa6f43cdb60d76bc73447 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:07:16 +0800 Subject: [PATCH 5/7] test: expose queued expiry racing an accepted stop Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_chat_session_active_turn.py | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index 43cfe91217..d71e50ef53 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -1314,11 +1314,13 @@ def pause_after_interrupt_read( assert (session_id, turn_id) not in runtime.cancelled_turns +@pytest.mark.parametrize("contender", ["claim", "expiry"]) def test_queued_claim_and_interrupt_share_one_turn_transition( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + contender: str, ) -> None: - """A queued claim and an interrupt serialize on the same Turn write lock. + """A queued writer and an interrupt serialize on the same Turn write lock. Before the claim took the Turn lock, the two writers could interleave: the interrupt read ``queued``, the claim committed ``running`` with a @@ -1344,6 +1346,8 @@ def test_queued_claim_and_interrupt_share_one_turn_transition( ) assert created is True turn_id = str(queued["turn_id"]) + if contender == "expiry": + store.update_turn(session_id, turn_id, expires_at="2000-01-01T00:00:00Z") original_load_turn = store.load_turn interrupt_read = threading.Event() @@ -1379,6 +1383,9 @@ def pause_after_interrupt_read( claim_thread = threading.Thread( target=lambda: claimed.append( store.claim_next_queued_turn(session_id, host_claim_id="host-claim") + if contender == "claim" else store._settle_expired_queued_turns( + session_id, now=chat_store.datetime.max.replace(tzinfo=chat_store.timezone.utc), + ) ), name="claim-turn", daemon=True, @@ -1395,7 +1402,7 @@ def pause_after_interrupt_read( assert turn["status"] == "interrupting" assert turn.get("host_claim_id") is None # The claim must not publish a Turn the interrupt already owns as active. - assert claimed == [None] + assert claimed == ([None] if contender == "claim" else [[]]) current = store.load_session(session_id) assert current is not None assert current["active_turn_id"] is None From 318ccc1b90676ca1a5722b073a5560eeb0c33bb5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:09:28 +0800 Subject: [PATCH 6/7] fix(chat): serialize queued expiry and acceptance settlement with stops Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/canary/module_metric_baseline.json | 2 +- loopx/chat_store.py | 78 +++++++++++++----------- tests/test_chat_session_active_turn.py | 13 +++- 3 files changed, 56 insertions(+), 37 deletions(-) diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index 54f3703f7b..0e45ed4e14 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -44,7 +44,7 @@ "lines": 2020 }, "loopx/chat_store.py": { - "lines": 2058 + "lines": 2066 }, "loopx/collaboration_mcp.py": { "lines": 2058 diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 6acc7dc19b..fababacc89 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -1168,17 +1168,22 @@ def accept_managed_turn( payload={}, ) if plan.writes.settle_turn: - settled = self.load_turn(session_token, plan.turn_id) - if settled is None: - raise ValueError( - "chat turn acceptance lost its prepared turn" - ) - settled.pop("_acceptance", None) - _atomic_write_json( + with exclusive_file_lock( self._turn_path(session_token, plan.turn_id), - settled, - preserve_mode=True, - ) + agent_id="loopx-chat", + operation="settle_managed_chat_turn_acceptance", + ): + settled = self.load_turn(session_token, plan.turn_id) + if settled is None: + raise ValueError( + "chat turn acceptance lost its prepared turn" + ) + settled.pop("_acceptance", None) + _atomic_write_json( + self._turn_path(session_token, plan.turn_id), + settled, + preserve_mode=True, + ) payload = settled else: current = self.load_turn(session_token, plan.turn_id) @@ -1229,32 +1234,35 @@ def _settle_expired_queued_turns( live_turns: list[dict[str, Any]] = [] for turn in self.queued_turns(session_id): - expires_at = _instant(turn.get("expires_at")) - if expires_at is None or expires_at > now: - live_turns.append(turn) - continue turn_id = str(turn["turn_id"]) - completed = utc_now() - turn.update( - { - "status": "timed_out", - "error_code": "session_queue_expired", - "error": "Queued Agent ingress expired before dispatch.", - "completed_at": completed, - "last_activity_at": completed, - } - ) - _atomic_write_json( - self._turn_path(session_id, turn_id), - turn, - preserve_mode=True, - ) - self.append_event( - session_id, - turn_id, - kind="turn.failed", - payload={"error_code": "session_queue_expired"}, - ) + path = self._turn_path(session_id, turn_id) + with exclusive_file_lock( + path, agent_id="loopx-chat", operation="expire_queued_chat_turn", + ): + # Queue enumeration is only a candidate; a stop or claim may + # already own the Turn by the time its write lock is acquired. + current = self.load_turn(session_id, turn_id) + if current is None or current.get("status") != "queued": + continue + expires_at = _instant(current.get("expires_at")) + if expires_at is None or expires_at > now: + live_turns.append(current) + continue + completed = utc_now() + current.update( + { + "status": "timed_out", + "error_code": "session_queue_expired", + "error": "Queued Agent ingress expired before dispatch.", + "completed_at": completed, + "last_activity_at": completed, + } + ) + _atomic_write_json(path, current, preserve_mode=True) + self.append_event( + session_id, turn_id, kind="turn.failed", + payload={"error_code": "session_queue_expired"}, + ) return live_turns def claim_next_queued_turn( diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index d71e50ef53..cbde1063e1 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -1379,7 +1379,17 @@ def pause_after_interrupt_read( interrupt_thread.start() assert interrupt_read.wait(timeout=2) - claimed: list[dict[str, object] | None] = [] + candidate_read = threading.Event() + original_queued_turns = store.queued_turns + + def observe_candidate(requested_session_id): + turns = original_queued_turns(requested_session_id) + if threading.current_thread().name == "claim-turn": + candidate_read.set() + return turns + + monkeypatch.setattr(store, "queued_turns", observe_candidate) + claimed: list[dict[str, object] | list | None] = [] claim_thread = threading.Thread( target=lambda: claimed.append( store.claim_next_queued_turn(session_id, host_claim_id="host-claim") @@ -1391,6 +1401,7 @@ def pause_after_interrupt_read( daemon=True, ) claim_thread.start() + assert candidate_read.wait(timeout=2) resume_interrupt.set() interrupt_thread.join(timeout=5) claim_thread.join(timeout=5) From b7bd894bf6772d492d486c641888e01b7608cff1 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 00:18:03 +0800 Subject: [PATCH 7/7] test: preserve expired queue outcomes when stop arrives later Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_chat_session_active_turn.py | 27 ++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index cbde1063e1..93381af6c2 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -1424,6 +1424,33 @@ def observe_candidate(requested_session_id): ] == ["turn.queued", "turn.interrupt_requested"] +def test_expired_queue_keeps_its_terminal_reason_when_stop_arrives_late(tmp_path: Path) -> None: + store = ChatSessionStore(tmp_path) + session = store.create_session( + goal_id="goal-one", agent_id="codex", executor_endpoint_id="codex", + adapter_kind="codex_app_server", upstream_thread_id="thread-one", upstream_mode="chat", + ) + sid = str(session["session_id"]) + queued, _ = store.create_queued_turn(sid, client_turn_id="expiry-first", message="expire before stop") + tid = str(queued["turn_id"]) + store.update_turn(sid, tid, expires_at="2000-01-01T00:00:00Z") + # Exercise the real public claimant, including its expiry prefix. + assert store.claim_next_queued_turn(sid, host_claim_id="late-host") is None + expired = store.load_turn(sid, tid) + assert expired["status"] == "timed_out" + assert expired["error_code"] == "session_queue_expired" + runtime = ChatRuntimeController(store=store, codex_bin="missing-codex") + try: + assert runtime.interrupt_turn(session_id=sid, turn_id=tid) == expired + assert store.load_turn(sid, tid) == expired + assert store.load_session(sid)["active_turn_id"] is None + assert [event["kind"] for event in store.events_after(sid, tid, None)] == [ + "turn.queued", "turn.failed", + ] + finally: + runtime.close() + + @pytest.mark.parametrize("crash_after_intent", [False, True]) def test_interrupt_preserves_a_successful_claim_and_durable_stop_intent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, crash_after_intent: bool,