fix(chat): recover private Bot routing outages without replaying work

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -1352,8 +1352,8 @@ 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
-terminal Core failure is returned and retained; it cannot enqueue another
-continuation.
+terminal Core failure on this unclassified-host/context-migration path is
+returned and retained; it cannot enqueue another continuation on that path.
 
 The existing in-flight transport retry also recognizes the exact native
 `other` diagnostic `workspace routing discovery timed out`. This provider hint
@@ -1365,6 +1365,23 @@ budget; it never reopens a terminal historical request or resets a reservation
 after restart. Native error prose is not copied into the retry journal. Existing
 App and Lark retry progress and final readback expose the same request's outcome.
 
+After those in-flight retries end, the original private reply worker can also
+recover a proven routing outage for ordinary project assistants and stewards.
+The exact diagnostic above, or its native remote-compaction wrapper, must be
+independently read back from the latest failed native Turn. Its complete stable
+inventory may contain only one user input and native compaction bookkeeping;
+answers, tools, corrections, unknown items and active native work remain holds.
+Fresh source, App/owner and grant verification still precede each admission.
+Core reserves at most three new, correlated continuations per original request,
+with 60/120/240-second backoff from the preceding persisted failure. Restart
+neither resets this budget nor reopens the failed Turn. A lost acknowledgement
+after queue admission adopts the exact existing client identity, including an
+already completed Turn, instead of enqueueing it again. Recovery attempts retain
+their own presentation keys and earlier reservation/notice history. A delivered
+failure of a continuation remains unfinished work; a completed answer does not
+activate recovery. This does not establish provider availability or eventual
+business completion, and it creates no additional event consumer.
+
 Recovery discovery backoff is only a temporary hint. The existing reply worker
 retains its per-request fairness count while that canonical request still exists;
 an omitted hint does not retire work or let earlier unresolved requests repeatedly

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -861,7 +861,8 @@ Turn，确认完整条目清单仅包含原用户输入后，Core 才能按每
 该有界准入也覆盖当前格式中的 `host_gate` 失败：原生错误为 `other`，且稳定、完整
 的原生清单证明只有原用户输入，没有任何执行条目时，才允许一次续办。它恢复既有
 线程，不迁移历史，也不匹配错误消息文本。这证明尚未执行，不证明 provider 健康
-或最终成功；续办若进入 Core 失败终态，就回传并保留失败，不能再生成下一次续办。
+或最终成功；这一未分类宿主／上下文迁移路径的续办若进入 Core 失败终态，就回传并
+保留失败，不能沿这一条路径再生成下一次续办。
 
 原有同轮传输重试也识别原生 `other` 类别下的确切诊断
 `workspace routing discovery timed out`。这个 provider 提示本身不授予重试权：
@@ -871,6 +872,16 @@ Turn、输入、Session 和至多三次的既有重试及退避预算，不重
 也不因重启重置预留。重试 journal 不复制原生错误正文；App 与 Lark 沿既有进展和
 最终读回路径展示同一请求的实际结果。
 
+同轮重试结束后，原私聊回复 worker 也可为普通项目助手与管家恢复已证实的路由故障。
+必须从最新失败的原生 Turn 独立读回上述确切诊断，或原生远端压缩任务附加的同一诊断。
+稳定、完整的条目清单只允许一条用户输入与原生压缩记账；回答、工具、追加纠正、未知
+条目和活动原生工作仍阻止恢复。每次受理前继续重新核验原消息、App／本人和 grant。
+Core 对每条原请求至多预留三次新的关联续办，分别从前次持久化失败起退避 60／120／240
+秒；重启不重置预算，也不重新打开失败 Turn。排队成功后丢失确认时，采用确切 client
+身份对应的既有 Turn，包括已经完成的 Turn，不再次排队。每次恢复有独立的呈现 key，
+并保留之前的预留与失败通知历史。续办的失败通知已送达仍不等于工作完成；已完成答复
+不会激活恢复。这不证明 provider 可用或业务最终完成，也不增加事件消费者。
+
 恢复发现的退避只是临时提示。规范请求仍存在时，原回复 worker 保留该请求的公平
 计数；某轮没有提示不代表工作已退役，也不能让前面的未结请求反复占满全部空位。
 四 worker 和不增加执行积压的上限保持。

loopx/capabilities/native_chat/codex_adapter.py

@@ -15,10 +15,84 @@
 import time
 from typing import Any, Callable
 
-from ...chat_agent import CodexChatAgentSession, CodexChatTimeoutError
+from ...chat_agent import (
+    CodexChatAgentError, CodexChatAgentSession, CodexChatTimeoutError,
+    _retry_error_details, _routing_discovery_timeout,
+)
 from ...chat_codex_goal import CodexGoalDriver
 
 
+def failed_request_readback(session: CodexChatAgentSession, turn_id: str) -> dict[str, Any] | None:
+    """Observe a historical pre-execution failure without resuming it.
+
+    Unlike a same-worker retry, a migration may inspect an older failure.
+    Every native Turn must be terminal. Routing recovery additionally
+    requires the latest Turn and permits only user input plus native
+    compaction bookkeeping: no answer, tool, correction or unknown item.
+    The Core still owns any new continuation admission.
+    """
+    deadline = time.monotonic() + session.response_timeout_sec
+
+    def terminal() -> dict[str, Any] | None:
+        turns = list(session._control_items("thread/turns/list", {
+            "sortDirection": "desc", "itemsView": "summary",
+        }, deadline))
+        ids = [item.get("id") for item in turns]
+        if (not ids or any(not isinstance(value, str) or not value for value in ids)
+                or len(set(ids)) != len(ids)
+                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":
+            return None
+        if _routing_discovery_timeout(selected.get("error")):
+            # A later request may supersede this one. Compaction is native
+            # bookkeeping, but only the exact latest routing failure can
+            # use that exception to the historical user-only inventory.
+            if ids[0] != turn_id:
+                return None
+            return {"codex_error_info": "other", "pre_execution_routing_timeout": True}
+        info = _retry_error_details(selected.get("error")).get("retry", {}).get("codex_error_info")
+        # `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()))
+            status = detail.get("httpStatusCode") if isinstance(detail, dict) else False
+            if isinstance(detail, dict) and "httpStatusCode" in detail and (
+                status is None or type(status) is int and 500 <= status <= 599
+            ):
+                return {"codex_error_info": info}
+        return None
+
+    try:
+        proof = terminal()
+        if proof is None:
+            return None
+        items = list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline))
+        routing = proof.get("pre_execution_routing_timeout") is True
+        allowed = {"userMessage", "contextCompaction"} if routing else {"userMessage"}
+        ids: set[str] = set()
+        users = 0
+        for entry in items:
+            item = entry.get("item")
+            if (entry.get("turnId") != turn_id or not isinstance(item, dict)
+                    or item.get("type") not in allowed or not item.get("id")
+                    or not isinstance(item["id"], str) or item["id"] in ids):
+                return None
+            ids.add(item["id"])
+            users += item["type"] == "userMessage"
+        if users != 1 or (not routing and len(items) != 1) or terminal() != proof:
+            return None
+        return {"thread_id": session.thread_id, "turn_id": turn_id,
+                "only_original_user_item": len(items) == 1,
+                **({"only_pre_execution_items": True} if routing else {}), **proof}
+    except (CodexChatAgentError, OSError, TypeError, ValueError):
+        return None
+
+
 @dataclass
 class CodexAppServerAdapter:
     session: CodexChatAgentSession
@@ -50,7 +124,7 @@ def failed_request_observation(*, codex_bin: str, codex_home: Path,
             for thread_id in dict.fromkeys(thread_ids):
                 native.thread_id = thread_id
                 native.response_timeout_sec = max(.01, deadline - time.monotonic())
-                proof = native.failed_request_readback(turn_id)
+                proof = failed_request_readback(native, turn_id)
                 if proof is not None:
                     return proof
                 if time.monotonic() >= deadline:

loopx/capabilities/native_chat/external_conversations.py

@@ -66,10 +66,15 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 return {}
         else:
             selected_binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
+        reservation = row.get("context_recovery") or {}
+        admitted = (store.turn_for_client(session["session_id"], reservation["client_turn_id"])
+            if reservation.get("routing_attempt") and reservation.get("failed_turn_id") == row["turn_id"] else None)
         return effect_runtime_result("collaboration.conversation.request", {
             "request": {key: value for key, value in row.items() if key != "attachments"},
             "current_session": session, "binding": selected_binding,
             "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn,
+                "observed_at": datetime.now(timezone.utc).isoformat(),
+                "admitted_turn": admitted,
                 "steering_observed": any(event.get("kind") == "turn.steered"
                     for event in store.events_after(session["session_id"], turn["turn_id"], None)),
                 "native_observation": observation, "source_verified": source_verified},
@@ -83,8 +88,9 @@ def context_recovery_candidate(self, request_ref: str) -> bool:
             if row.get("status") != "accepted":
                 return False
             recovery = row.get("context_recovery")
-            if recovery and row.get("turn_id") != recovery["failed_turn_id"]:
-                return not row.get("delivery_verified")
+            if (recovery and row.get("turn_id") != recovery["failed_turn_id"]
+                    and not row.get("delivery_verified")):
+                return True
             if not row.get("delivery_verified"):
                 return False
             turn = self.controller.store.load_turn(row["session_id"], row["turn_id"])
@@ -118,7 +124,7 @@ def recover_failed_request(self, request_ref: str, *,
                 assert session is not None and turn is not None
                 self.controller._check_codex_home(session)
                 context = self.controller.project_contexts.session_context(session)
-                proof = CodexAppServerAdapter.failed_request_observation(
+                proof = None if plan.get("admitted_turn_id") else CodexAppServerAdapter.failed_request_observation(
                     codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
                     thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
                 selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
@@ -127,27 +133,37 @@ def recover_failed_request(self, request_ref: str, *,
                 if not plan.get("recovery_eligible"):
                     return row
                 recovery = row.get("context_recovery")
-                if recovery is None:
+                if recovery is None or recovery.get("client_turn_id") != plan["client_turn_id"]:
+                    if recovery is not None:
+                        row.setdefault("context_recovery_history", []).append(deepcopy(recovery))
                     recovery = {"version": MANAGER_CONTEXT_VERSION, "failed_turn_id": row["turn_id"],
                         "native_observation": proof, "source_observation": source,
                         "client_turn_id": plan["client_turn_id"],
                         "failure_notice_delivery_verified": row["delivery_verified"]}
+                    if plan.get("routing_attempt") is not None:
+                        recovery.update(routing_attempt=plan["routing_attempt"],
+                            presentation_key=f"{MANAGER_CONTEXT_VERSION}.routing.{plan['routing_attempt']}",
+                            not_before=plan["not_before"])
                     row["context_recovery"] = recovery
                     _atomic_write_json(path, row)
                 # Freeze correlation before the existing queue can dispatch.
                 self._record_steward_ingress(row, selected, session, plan["client_turn_id"])
-                continued, _ = self.controller.enqueue_turn(session_id=row["session_id"],
-                    client_turn_id=plan["client_turn_id"], message=row["message"],
-                    attachments=row.get("attachments"), origin=row.get("origin", "lark"),
-                    work_dir=context["project"], objective=context["objective"])
+                if plan.get("admitted_turn_id"):
+                    continued = self.controller.store.load_turn(row["session_id"], plan["admitted_turn_id"])
+                    assert continued is not None
+                else:
+                    continued, _ = self.controller.enqueue_turn(session_id=row["session_id"],
+                        client_turn_id=plan["client_turn_id"], message=row["message"],
+                        attachments=row.get("attachments"), origin=row.get("origin", "lark"),
+                        work_dir=context["project"], objective=context["objective"])
                 row.update(turn_id=continued["turn_id"], delivery_verified=False)
                 _atomic_write_json(path, row)
             self.controller.resume_session_queue(session_id=row["session_id"],
                 work_dir=context["project"], objective=context["objective"])
             return row
 
     def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any] | None:
-        match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*", str(turn.get("client_turn_id") or ""))
+        match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-routing-[1-3])?", str(turn.get("client_turn_id") or ""))
         if match is None:
             return None
         row = self.read_request(match[1])