fix(chat): continue proven pre-execution host failures once

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.

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 可凭确切已确认的编辑凭证恢复旧通知，并结束随后

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()))

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])) {

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"}

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