From d1e7c7b01399e563aa2d6639016df4a1602c6599 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:43:12 +0800 Subject: [PATCH] fix(chat): recover untouched requests after transport loss --- .../capabilities/native_chat/codex_adapter.py | 12 ++- .../collaboration/conversation_binding.ts | 10 ++- tests/test_chat_agent.py | 43 ++++++++++ tests/test_chat_conversation_bindings.py | 80 +++++++++++++++++++ 4 files changed, 142 insertions(+), 3 deletions(-) diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py index af5276bfac..1bed7bd812 100644 --- a/loopx/capabilities/native_chat/codex_adapter.py +++ b/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, diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 9a0eafc153..a6c551c4ed 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/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)}; } diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index 638a5b649e..c387e8932e 100644 --- a/tests/test_chat_agent.py +++ b/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"]) diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index f6a848d037..9fb056cf52 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/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