diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index db1c0673b2..0e45ed4e14 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": 2066 }, "loopx/collaboration_mcp.py": { "lines": 2058 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..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( @@ -1324,38 +1332,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", @@ -1416,6 +1446,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, diff --git a/tests/test_chat_session_active_turn.py b/tests/test_chat_session_active_turn.py index 57f6293ef4..93381af6c2 100644 --- a/tests/test_chat_session_active_turn.py +++ b/tests/test_chat_session_active_turn.py @@ -1314,6 +1314,182 @@ 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 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 + ``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"]) + 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() + 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) + + 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") + 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, + ) + claim_thread.start() + assert candidate_read.wait(timeout=2) + 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] if contender == "claim" else [[]]) + 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_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, +) -> 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,