From 879df0b63d1a424417045df8f915c7abf544cc11 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 20 Oct 2026 02:57:16 +0800 Subject: [PATCH] fix(chat): continue proven pre-execution host failures once Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 9 +++- ...p-conversation-and-async-inbox-v0.zh-CN.md | 7 ++- loopx/chat_agent.py | 5 +- .../collaboration/conversation_binding.ts | 13 +++-- tests/test_chat_agent.py | 8 +-- tests/test_native_steward_private.py | 50 +++++++++++++++++++ 6 files changed, 81 insertions(+), 11 deletions(-) diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index 085aca0fce..1483869d36 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1318,7 +1318,7 @@ without provider identity IO for every historical failure. This is a queue hint; admission still re-resolves current App/owner/grant and exact source/native proof under the existing fences before reserving or enqueueing any continuation. -Core may admit one correlated continuation per request for this migration only +Core may admit one correlated continuation per request and context version after current App/owner/grant and unchanged source verification, an exact failed native Turn readback, and a complete inventory containing only its original user item. Policy/usage denials, corrections, tools, unknown effects, closed Sessions, @@ -1329,6 +1329,13 @@ original-source result. A restart resumes the deterministic reservation rather than dispatching it twice. Existing provider replies accompany recovery as data so completed work can be retained without repeating its effects. +The same bounded admission also covers a current-format `host_gate` failure +whose native error is `other`, only when the stable native inventory proves +there was exactly the original user input and no execution item. It resumes the +existing thread, without a history migration or error-message text matching. +This proves absence of execution, not provider health or eventual success. A +failed continuation is returned and retained; it cannot generate another retry. + Continuation admission, progress, cursor and final delivery use a version-scoped presentation journal. They never reopen the acknowledged original Inbox source or edit its failure notice as a new draft. The final replaces only the new draft. diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 0b9fe629f3..448273bbdc 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md @@ -837,13 +837,18 @@ attached 执行。 核验确切原消息与原生凭证后，才能预留或排队任何续办。 只有重新核验当前 App、本人、grant 和未改变的原消息，并独立读回确切失败的原生 -Turn，确认完整条目清单仅包含原用户输入后，Core 才能为本次格式迁移给每条请求 +Turn，确认完整条目清单仅包含原用户输入后，Core 才能按每条请求和上下文版本 受理一次关联续办。策略或额度拒绝、追加纠正、工具、未知效果、关闭的 Session、 群或 attached 路由及读回不可用均继续保留。原失败 Turn 与已核验的失败通知不变； 续办由原 Session 队列执行，最终原消息结果使用独立冻结的送达阶段。重启恢复同一个 确定性预留，不重复派发。已有 provider 回复作为数据随恢复提供，供保留已完成工作、 避免重复效果。 +该有界准入也覆盖当前格式中的 `host_gate` 失败：原生错误为 `other`，且稳定、完整 +的原生清单证明只有原用户输入，没有任何执行条目时，才允许一次续办。它恢复既有 +线程，不迁移历史，也不匹配错误消息文本。这证明尚未执行，不证明 provider 健康 +或最终成功；续办若再次失败，就回传并保留失败，不能再生成下一次重试。 + 续办的接收提示、进展、游标与最终回传按版本独立记账，不重新打开原 Inbox 已确认的 消息，也不把原失败通知改成新草稿；最终结果只替换本次新草稿。对于已经回传、但旧版 误用原失败通知的续办，原 worker 可凭确切已确认的编辑凭证恢复旧通知，并结束随后 diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py index 04404f8821..94bc5a0979 100644 --- a/loopx/chat_agent.py +++ b/loopx/chat_agent.py @@ -1485,7 +1485,10 @@ def terminal() -> dict[str, Any] | None: if not selected or selected.get("status") != "failed": return None info = _retry_error_details(selected.get("error")).get("retry", {}).get("codex_error_info") - if info in ("contextWindowExceeded", "serverOverloaded", "internalServerError"): + # `other` is evidence of a failed native request, not permission to + # retry it. The typed admission owner separately decides whether + # this proven user-only failure is eligible for one continuation. + if info in ("contextWindowExceeded", "serverOverloaded", "internalServerError", "other"): return {"codex_error_info": info} if isinstance(info, dict) and len(info) == 1: detail = next(iter(info.values())) diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index 7cf457094e..902e606fbc 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -503,15 +503,20 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO if (Number.isSafeInteger(current.manager_context_version) && Number(current.manager_context_version) < version) { oldThreads.push(sessionIdentity(current.upstream_thread_id, "original native thread")); } - if (!oldThreads.length) return no; + const currentHostFailure = current.manager_context_version === version && failed.error_code === "host_gate"; + const nativeThreads = currentHostFailure + ? [sessionIdentity(current.upstream_thread_id, "current native thread"), ...oldThreads] : oldThreads; + if (!nativeThreads.length) return no; const candidate = {...no, recovery_candidate: true, client_turn_id: client, - session_id: current.session_id, failed_turn_id: failed.turn_id, native_thread_ids: [...new Set(oldThreads)]}; + session_id: current.session_id, failed_turn_id: failed.turn_id, native_thread_ids: [...new Set(nativeThreads)]}; if (recovery.native_observation == null) return candidate; const observed = requireJsonObject(recovery.native_observation, "native failure readback"); - if (!oldThreads.includes(String(observed.thread_id)) || observed.turn_id !== failed.upstream_turn_id + if (!nativeThreads.includes(String(observed.thread_id)) || observed.turn_id !== failed.upstream_turn_id || observed.only_original_user_item !== true || recovery.source_verified !== true) return candidate; const info = observed.codex_error_info; - let transient = ["contextWindowExceeded", "serverOverloaded", "internalServerError"].includes(String(info)); + let transient = (oldThreads.includes(String(observed.thread_id)) + && ["contextWindowExceeded", "serverOverloaded", "internalServerError"].includes(String(info))) + || (currentHostFailure && observed.thread_id === current.upstream_thread_id && info === "other"); if (info && typeof info === "object" && !Array.isArray(info)) { const entries = Object.entries(info); if (entries.length === 1 && ["httpConnectionFailed", "responseStreamConnectionFailed", "responseStreamDisconnected", "responseTooManyFailedAttempts"].includes(entries[0][0])) { diff --git a/tests/test_chat_agent.py b/tests/test_chat_agent.py index 5e5a3d54b9..d92abfba7c 100644 --- a/tests/test_chat_agent.py +++ b/tests/test_chat_agent.py @@ -263,7 +263,7 @@ def write(packet): assert all(method in {"thread/turns/list", "thread/items/list"} for method, _ in calls) -@pytest.mark.parametrize("fault", [None, "older", "active", "tool", "answer", "correction", +@pytest.mark.parametrize("fault", [None, "older", "host-other", "active", "tool", "answer", "correction", "foreign", "unknown", "changed-error", "policy", "usage", "unauthorized", "http401"]) def test_historical_context_recovery_reads_only_proven_pre_execution_failure(tmp_path, monkeypatch, fault): session = chat_agent.CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None), @@ -277,7 +277,7 @@ def write(packet): assert params["threadId"] == "original-thread" if method == "thread/turns/list": reads += 1 - info = {"policy": "cyberPolicy", "usage": "usageLimitExceeded", + info = {"host-other": "other", "policy": "cyberPolicy", "usage": "usageLimitExceeded", "unauthorized": "unauthorized", "http401": {"responseStreamDisconnected": {"httpStatusCode": 401}}}.get(fault, "contextWindowExceeded") if fault == "changed-error" and reads > 1: info = "other" @@ -296,9 +296,9 @@ def write(packet): monkeypatch.setattr(session, "_write", write) proof = session.failed_request_readback("failed") - if fault in {None, "older"}: + if fault in {None, "older", "host-other"}: assert proof == {"thread_id": "original-thread", "turn_id": "failed", - "only_original_user_item": True, "codex_error_info": "contextWindowExceeded"} + "only_original_user_item": True, "codex_error_info": "other" if fault == "host-other" else "contextWindowExceeded"} else: assert proof is None assert set(calls) <= {"thread/turns/list", "thread/items/list"} diff --git a/tests/test_native_steward_private.py b/tests/test_native_steward_private.py index fcc04c2bb5..961b26e1f0 100644 --- a/tests/test_native_steward_private.py +++ b/tests/test_native_steward_private.py @@ -238,6 +238,56 @@ def crash_sync(*args, **kwargs): assert "Prior verified completion" in capture.read_text() +@pytest.mark.parametrize("fault", [None, "native-effect", "policy", "failure-again"]) +def test_current_context_host_failure_gets_one_proven_continuation(steward, monkeypatch, fault): + from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter + from loopx.chat_manager import MANAGER_CONTEXT_VERSION + + store, runtime, provider, transport, binding, capture, _ = steward + fake = Path(runtime.codex_bin) + source = fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', '"codexErrorInfo": "other"') + source = source.replace(' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '') + fake.write_text(source) + event = provider.event("steward-app", "host-startup", "typed terminal failure") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + failed = finish(runtime, row) + assert failed["status"] == "failed" and failed["error_code"] == "host_gate" and failed["delta_count"] == 0 + assert store.load_session(row["session_id"])["manager_context_version"] == MANAGER_CONTEXT_VERSION + failed_path = store._turn_path(row["session_id"], row["turn_id"]) + failed_bytes = failed_path.read_bytes() + transport.reconcile() + path = transport.root / (ref + ".json") + notice = json.loads(path.read_text())["deliveries"]["terminal"] + runtime.adapters.pop(row["session_id"]).close_session() + if fault != "failure-again": + fake.write_text(source.replace('if "typed terminal failure" in json.dumps(request.get("params") or {}):', 'if False:')) + proof = {"thread_id": "durable-thread", "turn_id": failed["upstream_turn_id"], + "only_original_user_item": fault != "native-effect", "codex_error_info": "cyberPolicy" if fault == "policy" else "other"} + monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **_: proof)) + transport.reconcile() + current = transport.core.read_request(ref) + if fault in {"native-effect", "policy"}: + assert current["turn_id"] == row["turn_id"] and not current.get("context_recovery") + expected_starts = 1 + else: + assert current["turn_id"] != row["turn_id"] and current["session_id"] == row["session_id"] + assert finish(runtime, current)["status"] == ("failed" if fault == "failure-again" else "completed") + transport.reconcile() + assert json.loads(path.read_text())["deliveries"][f"terminal.recovery.{MANAGER_CONTEXT_VERSION}"]["verified"] + expected_starts = 2 + # A persistent provider failure is surfaced once, never an infinite replay. + transport.reconcile() + transport.reconcile() + requests = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(r["method"] == "turn/start" for r in requests) == expected_starts + assert not any(r["method"] == "thread/inject_items" for r in requests) + assert failed_path.read_bytes() == failed_bytes + assert json.loads(path.read_text())["deliveries"]["terminal"] == notice + assert len(store.list_sessions()) == 1 + + def test_personal_steward_uses_registered_portfolio_and_configured_inbox_root(steward, tmp_path): from loopx.capabilities.manager_context.discovery import agent_page from loopx.capabilities.manager_context import configure_delivery_scope