fix(chat): recover untouched requests after transport loss

loopx/capabilities/native_chat/codex_adapter.py

@@ -44,7 +44,14 @@ def terminal() -> dict[str, Any] | None:
                 or any(item.get("status") not in {"completed", "failed", "interrupted"} for item in turns)):
             return None
         selected = next((item for item in turns if item["id"] == turn_id), None)
-        if not selected or selected.get("status") != "failed":
+        if not selected:
+            return None
+        if selected.get("status") == "interrupted" and selected.get("error") is None and ids[0] == turn_id:
+            # Closing a disconnected app-server can persist an interruption,
+            # without a native error. This is not a failed-request diagnostic;
+            # only an exact user-only inventory below can prove no execution.
+            return {"interrupted_before_execution": True}
+        if selected.get("status") != "failed":
             return None
         if _routing_discovery_timeout(selected.get("error")):
             # Later work must not strand an untouched accepted request. Keep
@@ -87,6 +94,9 @@ def terminal() -> dict[str, Any] | None:
             users += item["type"] == "userMessage"
         if users != 1 or (not routing and len(items) != 1) or terminal() != proof:
             return None
+        if proof.get("interrupted_before_execution") and list(
+                session._control_items("thread/items/list", {"turnId": turn_id}, deadline)) != items:
+            return None
         proof.pop("latest_routing_failure", None)
         return {"thread_id": session.thread_id, "turn_id": turn_id,
                 "only_original_user_item": len(items) == 1,

loopx/control_plane/collaboration/conversation_binding.ts

@@ -1166,14 +1166,19 @@ function planRoutingRecovery(recovery: JsonObject, row: JsonObject, current: Jso
   if (previous !== null && previous.failed_turn_id === failed.turn_id && previous[attemptKey] === undefined) return null;
   const raw = preparation ? recovery.preparation_observation : recovery.native_observation;
   const observed = raw == null ? null : requireJsonObject(raw, "pre-execution failure readback");
-  if (observed && !quota && !preparation && observed.pre_execution_routing_timeout !== true) return null;
+  const interrupted = observed?.interrupted_before_execution === true
+    && ["host_gate", "upstream_transport_failure"].includes(String(failed.error_code));
+  if (observed && !quota && !preparation && observed.pre_execution_routing_timeout !== true && !interrupted) return null;
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
   const reserved = previous?.failed_turn_id === failed.turn_id;
   const priorAttempt = previous?.[attemptKey] ?? 0;
   if (!Number.isSafeInteger(priorAttempt) || Number(priorAttempt) < 0) return no;
   const attempt = Number(priorAttempt) + (reserved ? 0 : 1);
+  const disconnected = current.status === "stale" && current.last_error_code === "transport_disconnected"
+    && ["host_gate", "upstream_transport_failure"].includes(String(failed.error_code));
   if (attempt < 1 || attempt > 3 || typeof current.upstream_thread_id !== "string"
-      || !current.upstream_thread_id || current.active_turn_id != null || current.status !== "ready") return no;
+      || !current.upstream_thread_id || current.active_turn_id != null
+      || (current.status !== "ready" && !disconnected)) return no;
   const now = typeof recovery.observed_at === "string" ? Date.parse(recovery.observed_at) : NaN;
   const completed = typeof failed.completed_at === "string" ? Date.parse(failed.completed_at) : NaN;
   const notBefore = completed + 60_000 * 2 ** (attempt - 1);
@@ -1208,6 +1213,7 @@ function planRoutingRecovery(recovery: JsonObject, row: JsonObject, current: Jso
     && (quota ? observed.codex_error_info === "usageLimitExceeded" && observed.only_original_user_item === true
       && observed.quota_available === true && typeof observed.quota_observed_at === "number"
       && now / 1000 >= observed.quota_observed_at && now / 1000 - observed.quota_observed_at <= 120
+      : interrupted ? recovery.stop_observed === false && observed.only_original_user_item === true
       : observed.codex_error_info === "other" && observed.only_pre_execution_items === true)};
 }
 

tests/test_chat_agent.py

@@ -813,6 +813,49 @@ def write(packet):
     assert set(calls) <= {"thread/turns/list", "thread/items/list"}
 
 
+@pytest.mark.parametrize("fault", [None, "older", "active", "tool", "answer", "reasoning", "compaction",
+    "unknown", "correction", "foreign", "duplicate", "changed-items", "changed-turn", "native-error"])
+def test_disconnected_latest_interruption_requires_stable_user_only_inventory(tmp_path, monkeypatch, fault):
+    session = chat_agent.CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None),
+        messages=queue.Queue(), thread_id="original", work_dir=tmp_path, response_timeout_sec=.2)
+    calls, counts = [], {"thread/turns/list": 0, "thread/items/list": 0}
+
+    def write(packet):
+        method, params = packet["method"], packet["params"]
+        calls.append(method)
+        assert params["threadId"] == "original"
+        if method == "thread/turns/list":
+            counts[method] += 1
+            rows = [{"id": "interrupted", "status": "interrupted"}]
+            if fault == "native-error":
+                rows[0]["error"] = {"codexErrorInfo": "cyberPolicy"}
+            if fault in {"older", "active"} or fault == "changed-turn" and counts[method] > 1:
+                rows.insert(0, {"id": "later", "status": "inProgress" if fault == "active" else "completed"})
+        elif method == "thread/items/list":
+            counts[method] += 1
+            rows = [{"turnId": "foreign" if fault == "foreign" else "interrupted",
+                "item": {"id": "user", "type": "userMessage"}}]
+            kind = {"tool": "commandExecution", "answer": "agentMessage", "unknown": "futureItem",
+                "correction": "userMessage", "duplicate": "userMessage", "reasoning": "reasoning",
+                "compaction": "contextCompaction"}.get(fault)
+            if fault == "changed-items" and counts[method] > 1:
+                kind = "commandExecution"
+            if kind:
+                rows.append({"turnId": "interrupted", "item": {
+                    "id": "user" if fault == "duplicate" else "extra", "type": kind}})
+        else:
+            pytest.fail("The original user-only inventory must not require an unsupported or unrelated native RPC")
+        session.messages.put({"id": packet["id"], "result": {"data": rows}})
+
+    monkeypatch.setattr(session, "_write", write)
+    proof = failed_request_readback(session, "interrupted")
+    assert proof == ({"thread_id": "original", "turn_id": "interrupted", "only_original_user_item": True,
+        "interrupted_before_execution": True} if fault is None else None)
+    assert set(calls) <= {"thread/turns/list", "thread/items/list"}
+    if fault is None:
+        assert counts == {"thread/turns/list": 2, "thread/items/list": 2}
+
+
 @pytest.mark.parametrize("compact", [False, True])
 @pytest.mark.parametrize("reverse", [False, True])
 @pytest.mark.parametrize("diagnostic", ["timed out", "failed"])

tests/test_chat_conversation_bindings.py

@@ -218,6 +218,86 @@ def _history_page_brief():
         "inputs": [], "acceptance": ["Return the original conclusion"], "return_requirement": "Return here"}
 
 
+@pytest.mark.parametrize("fault", [None, "stop", "edited-source", "revoked", "unknown-native", "other-stale"])
+def test_original_worker_recovers_disconnected_user_only_interruption(steward, monkeypatch, fault):  # noqa: F811
+    from datetime import datetime, timedelta, timezone
+    from pathlib import Path
+    from loopx.capabilities.native_chat import external_conversations
+
+    store, runtime, provider, transport, binding, capture, _ = steward
+    Path(runtime.codex_home).mkdir(parents=True, exist_ok=True)
+    fake = Path(runtime.codex_bin)
+    marker = fake.with_name("interrupted-native-input.json")
+    source = fake.read_text()
+    source = source.replace('turn_count = 0', f'turn_count = 1 if __import__("os").path.exists({str(marker)!r}) else 0')
+    begin = '        if "typed terminal failure" in json.dumps(request.get("params") or {}):'
+    assert begin in source
+    source = source.replace(begin,
+        begin[:-1] + f' and not __import__("os").path.exists({str(marker)!r}):\n'
+        f'            open({str(marker)!r}, "w").write(json.dumps({{"turn_id": active_turn}}))\n'
+        '            raise SystemExit(0)')
+    branch = '    elif method == "thread/turns/list":'
+    assert branch in source
+    source = source.replace(branch,
+        f'    elif method == "thread/turns/list" and __import__("os").path.exists({str(marker)!r}):\n'
+        f'        result = {{"data": [{{"id": json.load(open({str(marker)!r}))["turn_id"], "status": "interrupted"}}]}}\n'
+        '    elif method == "thread/items/list":\n'
+        '        result = {"data": [{"turnId": request["params"]["turnId"], "item": {"id": "user", "type": "userMessage"}}]}\n'
+        + branch)
+    fake.write_text(source)
+    event = provider.event("steward-app", "lost-server", "typed terminal failure")
+    transport.admit("steward-app", event)
+    ref = identity_ref(binding["provider_ref"], event["message_id"])
+    row = transport.core.read_request(ref)
+    failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)
+    assert failed["error_code"] == "host_gate" and failed["delta_count"] == 0
+    deadline = time.monotonic() + 2
+    while store.load_session(row["session_id"])["status"] != "stale":
+        assert time.monotonic() < deadline
+        time.sleep(.01)
+    assert store.load_session(row["session_id"])["last_error_code"] == "transport_disconnected"
+    failed_bytes = store._turn_path(row["session_id"], row["turn_id"]).read_bytes()
+    assert transport.reconcile() == 1  # Keep the original failure notice.
+    clock = datetime.now(timezone.utc) + timedelta(seconds=61)
+    class AfterBackoff(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return clock
+    monkeypatch.setattr(external_conversations, "datetime", AfterBackoff)
+    if fault == "stop":
+        transport.admit("steward-app", provider.event("steward-app", "later-stop", "/stop"))
+    elif fault == "edited-source":
+        provider.messages[event["message_id"]]["content"] = "Changed original source"
+    elif fault == "revoked":
+        transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+    elif fault == "unknown-native":
+        fake.write_text(fake.read_text().replace('"type": "userMessage"', '"type": "unknownItem"'))
+    elif fault == "other-stale":
+        store.update_session(row["session_id"], last_error_code="unverified")
+    before = [json.loads(line) for line in capture.read_text().splitlines()]
+    transport._context_recovery_probe_at.clear()
+    transport.reconcile()
+    current = transport.core.read_request(ref)
+    if fault is None:
+        assert current["turn_id"] != row["turn_id"] and current["session_id"] == row["session_id"]
+        completed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=current["turn_id"], timeout_sec=10)
+        assert completed["status"] == "completed"
+        assert completed["client_turn_id"].endswith("-routing-1")
+        assert current["context_recovery"]["native_observation"]["interrupted_before_execution"] is True
+        transport.reconcile()
+        assert transport.core.read_request(ref)["delivery_verified"] is True
+        after = capture.read_bytes()
+        transport.reconcile()
+        assert capture.read_bytes() == after  # No second model dispatch after verified return.
+    else:
+        assert current["turn_id"] == row["turn_id"] and not current.get("context_recovery")
+    after = [json.loads(line) for line in capture.read_text().splitlines()]
+    def count(calls):
+        return sum(p.get("method") == "turn/start" for p in calls)
+    assert count(after) == count(before) + (fault is None)
+    assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == failed_bytes
+
+
 @pytest.mark.parametrize("change", ["status", "new-input", "source-read-input", "unrelated-source", "revoked", "unknown-native", "unchanged"])
 def test_slow_failed_request_probe_does_not_hold_the_source_route(steward, monkeypatch, change):  # noqa: F811
     import threading