diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index 0f4fd1286e..ec4e0ab6c3 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -1145,6 +1145,13 @@ def recover(self, *, request_ref: str | None = None) -> None: # delivering their pending response; do not add an idle probe. if row["status"] == "command_completed" and not row.get("commission_adoption_pending"): continue + if (row["status"] == "accepted" and not row.get("agent_target") + and not row.get("commission_adoption_pending")): + turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) + if turn and turn.get("status") == "completed": + # A committed answer needs delivery, not workspace or + # queue preparation. The transport checks its source. + continue if row["status"] == "prepared": # Admission resolves current authority after both request # and route fences. A separate pre-fence network probe diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index d9e8507e84..d705a542c5 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -53,7 +53,7 @@ from .control_plane.chat_turn_acceptance import managed_turn_recovery_allowed from .file_lock import ( LOCK_HOLDER_DEAD, LOCK_HOLDER_LIVE, LOCK_HOLDER_RELEASED, - _lock_id, lock_holder_host_label, lock_holder_liveness, try_exclusive_file_lock, + _lock_id, lock_holder_host_label, lock_holder_liveness, process_is_alive, try_exclusive_file_lock, ) from .kiro_cli_goal_mode import ( KIRO_CLI_BIN, @@ -1581,6 +1581,10 @@ def recover_exited_turn(self, session_id: str, turn_id: str | None = None) -> No return session = self.store.load_session(session_id) turn = self.store.load_turn(session_id, turn_id) + if turn and turn.get("status") == "interrupting" and process_is_alive(current["pid"]): + # Stop deliberately retires the execution worker before its + # caller commits `interrupted`; a free fence is expected here. + return if managed_turn_recovery_allowed( session, turn, managed=bool(session and session.get("session_mode") == CHAT_SESSION_MODE_MANAGED), owner_dead=True, diff --git a/tests/test_chat_runtime_activation.py b/tests/test_chat_runtime_activation.py index e62a68b36a..fb8d33de14 100644 --- a/tests/test_chat_runtime_activation.py +++ b/tests/test_chat_runtime_activation.py @@ -9,6 +9,7 @@ import plistlib import shlex import subprocess +import sys import threading from unittest import mock @@ -117,6 +118,28 @@ def test_same_process_worker_exit_is_recovered_without_replay(tmp_path, released assert not runtime.adapters and not runtime.session_queue_threads +@pytest.mark.parametrize("released", [False, True]) +def test_recovery_preserves_stop_after_the_live_service_worker_retires(tmp_path, released): + store = ChatSessionStore(tmp_path) + sid, tid, path = lost_execution(store, released=released) + store.begin_turn_interrupt(sid, tid, expected_statuses={"running"}) + before = content_hashes(store.root) + runtime = ChatRuntimeController(store=store, codex_bin="must-not-execute") + runtime.recover_exited_turn(sid, tid) + assert content_hashes(store.root) == before + assert store.load_turn(sid, tid)["status"] == "interrupting" + assert store.load_session(sid)["active_turn_id"] == tid + # A later service crash still belongs to the existing dead-owner recovery. + holder = json.loads(lock_holder_path(path).read_text()) + process = subprocess.Popen([sys.executable, "-c", "pass"]) + process.wait(timeout=5) + holder["pid"] = process.pid + lock_holder_path(path).write_text(json.dumps(holder)) + runtime.recover_exited_turn(sid, tid) + assert store.load_turn(sid, tid)["error_code"] == ("runtime_error" if released else "server_restarted") + assert store.load_session(sid)["active_turn_id"] is None + + @pytest.mark.parametrize("fault", ["absent", "foreign", "pid", "identity", "operation", "queued", "attached"]) def test_unproven_worker_exit_never_releases_accepted_work(tmp_path, fault): store = ChatSessionStore(tmp_path) diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 04dd76655a..d64cb2bb50 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -1763,6 +1763,90 @@ def test_completed_command_recovery_leaves_provenance_to_delivery(ordinary, comm runtime.close() +@pytest.mark.parametrize("fault", [None, "changed-source", "provider-unavailable", "revoked"]) +def test_completed_answer_delivery_does_not_prepare_another_execution(ordinary, monkeypatch, fault): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + event = provider.event("notes-app", "completed-original", "Complete this original request.") + try: + transport.admit("notes-app", event) + native = next(row for row in transport.core.pending() if row["message"] == event["content"]) + sid, tid = native["session_id"], native["turn_id"] + completed = runtime.wait_for_turn(session_id=sid, turn_id=tid, timeout_sec=10) + assert completed["status"] == "completed" + worker = runtime.session_queue_threads.get(sid) + if worker is not None: + worker.join(5) + assert not worker.is_alive() + followup, _ = store.create_queued_turn(sid, client_turn_id="saved-next-input", message="Saved next input.") + queued_before = store.load_turn(sid, followup["turn_id"]) + path = transport.root / f"{native['request_ref']}.json" + native_path = transport.core.root / path.name + original = native_path.read_bytes() + messages = store.messages(sid) + # Preparing another execution may be slow or unavailable while a + # committed answer is ready. Delivery must not depend on either call. + monkeypatch.setattr(runtime.project_contexts, "session_context", + lambda *_: pytest.fail("workspace preparation before committed-answer delivery")) + monkeypatch.setattr(runtime, "resume_session_queue", + lambda **_: pytest.fail("queue preparation before committed-answer delivery")) + provider.calls.clear() + transport.core.recover(request_ref=native["request_ref"]) + assert provider.calls == [] + assert native_path.read_bytes() == original + assert store.messages(sid) == messages + if fault == "changed-source": + provider.messages[event["message_id"]]["sender"]["id"] = "ou_another_human" + elif fault == "provider-unavailable": + provider.bot_ready = False + elif fault == "revoked": + transport.bindings.disconnect(native["binding_id"], expected_revision=transport.bindings.read()["revision"]) + # Disconnect legitimately stops the queued follow-up. Compare delivery + # against that new authority state, retaining the original answer. + delivery_messages = store.messages(sid) + delivery_followup = store.load_turn(sid, followup["turn_id"]) + if fault == "revoked": + assert delivery_followup["status"] in {"failed", "interrupted"} + delivered = transport.reconcile_request(path) + if fault != "revoked": + assert any("auth" in call for call in provider.calls) # Fresh delivery provenance remains mandatory. + if fault: + assert delivered == 0 and provider.writes == [] + assert json.loads(path.read_text())["status"] != "delivered" + else: + assert delivered == 1 + assert provider.writes == [("notes-app", "Runtime response.")] + assert json.loads(native_path.read_text())["delivery_verified"] is True + assert store.messages(sid) == delivery_messages + assert [m for m in delivery_messages if m.get("turn_id") == tid] == [m for m in messages if m.get("turn_id") == tid] + assert store.load_turn(sid, followup["turn_id"]) == delivery_followup + if fault != "revoked": + assert delivery_followup == queued_before + packets = [json.loads(line) for line in ordinary[4].read_text().splitlines()] + assert len([row for row in packets if row["method"] == "turn/start"]) == 1 + finally: + runtime.close() + + +def test_unexecuted_request_still_uses_original_queue_recovery(ordinary, monkeypatch): # noqa: F811 + store, runtime, provider, transport = connect(ordinary) + resumes = [] + monkeypatch.setattr(runtime, "resume_session_queue", lambda **kw: resumes.append(kw)) + event = provider.event("notes-app", "not-completed", "A completed-looking title is still new work.") + try: + transport.admit("notes-app", event) + native = next(row for row in transport.core.pending() if row["message"] == event["content"]) + assert store.load_turn(native["session_id"], native["turn_id"])["status"] == "queued" + resumes.clear() + provider.calls.clear() + transport.core.recover(request_ref=native["request_ref"]) + assert len(resumes) == 1 and resumes[0]["session_id"] == native["session_id"] + assert any("auth" in call for call in provider.calls) + assert provider.writes == [] + assert store.load_turn(native["session_id"], native["turn_id"])["status"] == "queued" + finally: + runtime.close() + + def test_lark_transport_recovery_returns_on_original_admission_without_model_replay(ordinary, monkeypatch): # noqa: F811 from test_chat_turn_steering import install_transport_failure