fix(chat): preserve completed delivery and stop ownership

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

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,

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)

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