fix(chat): recover original messages after account preparation fails

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

@@ -1731,6 +1731,17 @@ The original failure and its delivery remain recorded; a qualified private
 steward notice describes conditional recovery rather than promising a replay.
 Unknown failures after native dispatch remain holds.
 
+Account transport preparation failures can also qualify for an original private
+project or Codex steward request. Core must retain exactly the Session-queue and
+failed events, with null durable `started_at` and `first_event_at` checkpoints,
+no native Turn/result, response, deltas or attachments, and unchanged input/origin.
+This includes the older exact `resume_failed` diagnostic emitted before the
+queue's first durable start checkpoint. Fresh source/grant and account transport
+checks precede admission; they start no model request. Preparation continues the
+existing quota/routing/preparation counter up to three, without resetting its
+backoff or history. A later dispatched failure cannot open another business
+retry family. The original failure and its delivered notice remain unchanged.
+
 Source verification accepts both ordinary text and text-only rich posts. A post
 must retain its original provider rendering, an explicit empty attachment
 inventory and the same admitted text after intake whitespace normalization.

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

@@ -1124,6 +1124,14 @@ Codex 与 Antigravity 管家在本地以 `manager_authorization_unavailable` 失
 仅修复关联，不重复派发。原失败及其送达记录保留；符合范围的私聊管家通知说明
 有条件接续，不承诺盲目重放。原生派发后的未知失败继续保留。
 
+原私聊项目助手或 Codex 管家的账户传输准备失败也可进入这一原队列。Core 必须恰好
+保留 Session 排队和失败两个事件，持久化 `started_at`、`first_event_at` 检查点均为空，
+没有原生 Turn、结果、答复、增量或附件，输入和 origin 未改变；这也覆盖旧版在队列
+首次持久化启动检查点之前产生的确切 `resume_failed` 诊断。准入前重新核验原消息、
+grant 和账户传输，核验本身不发起模型请求。准备接续沿用既有 quota/routing/preparation
+计数，上限仍为三次，不重置退避或历史；后来已派发的失败不能再开启其他业务重试
+类型。原失败及已送达的失败通知保持不变。
+
 来源核验同时支持普通文字与纯文字富文本。富文本必须保留原 provider 渲染、明确为空
 的附件清单，以及按原受理空白归一后相同的模型输入；provider 当前消息类型与完整
 渲染也必须相符。证据缺失或被编辑、图片及附件失败继续停止；恢复不下载资源，

loopx/canary/module_metric_baseline.json

@@ -39,7 +39,7 @@
     "loopx/chat_runtime.py": {
       "any_count": 38,
       "dict_any_count": 0,
-      "lines": 2417
+      "lines": 2419
     },
     "loopx/chat_server.py": {
       "any_count": 25,

loopx/capabilities/native_chat/external_conversations.py

@@ -141,14 +141,16 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         preparation = None
         later_equivalent = None
         image_failure = turn.get("error_code") == "runtime_error" and bool(row.get("attachments"))
-        if turn.get("error_code") == "manager_authorization_unavailable" or image_failure:
+        runtime_failure = turn.get("error_code") in {"model_auth_unavailable", "resume_failed"}
+        if turn.get("error_code") == "manager_authorization_unavailable" or image_failure or runtime_failure:
             # Collect canonical facts only. The typed planner owns the exact
             # pre-dispatch sequence, retry budget and authority decision.
-            preparation = {"kind": ("core_attachment_preparation_failure" if image_failure
+            preparation = {"kind": ("core_runtime_preparation_failure" if runtime_failure else
+                                    "core_attachment_preparation_failure" if image_failure
                                     else "core_context_preparation_failure"), "session_id": session["session_id"],
                 "turn_id": turn["turn_id"], "events": [
                     {"event_id": item["event_id"], "kind": item["kind"],
-                     **{key: item.get("payload", {})[key] for key in ("phase", "schema_version", "error_code")
+                     **{key: item.get("payload", {})[key] for key in ("phase", "schema_version", "error_code", "delivery_mode", "message")
                         if key in item.get("payload", {})}} for item in events]}
             created = datetime.fromisoformat(turn["created_at"])
             later_equivalent = any(item.get("role") == "user" and item.get("turn_id") != turn["turn_id"]
@@ -167,7 +169,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                     "failed": _preparation_image_identity(turn.get("attachments")),
                     "admitted": _preparation_image_identity((admitted or {}).get("attachments"))},
                     "native_result_present": bool(store.native_result(session["session_id"], turn["turn_id"]))}
-                   if image_failure else {}),
+                    if image_failure else {"native_result_present": bool(store.native_result(session["session_id"], turn["turn_id"]))}
+                    if runtime_failure else {}),
                 "observed_at": datetime.now(timezone.utc).isoformat(),
                 "admitted_turn": ({key: value for key, value in admitted.items() if key != "attachments"}
                                   if image_failure and admitted is not None else admitted),
@@ -449,7 +452,8 @@ def source_requests() -> list[dict[str, Any]]:
             facts = proof.pop("retained_result", None) if proof is not None and metadata_review else None
             selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
             source = source_observation(selected)
-            scope_available = (self._preparation_scope_available(session)
+            scope_available = (self._preparation_scope_available(session,
+                runtime_preparation=turn.get("error_code") in {"model_auth_unavailable", "resume_failed"})
                 if preparation_retry and source is not None and not plan.get("admitted_turn_id") else False)
             # Identity and original-message reads are fallible provider IO too.
             # Never hold up newer source input while they wait. If another
@@ -542,13 +546,26 @@ def source_requests() -> list[dict[str, Any]]:
                 work_dir=context["project"], objective=context["objective"])
             return row
 
-    def _preparation_scope_available(self, session: dict[str, Any]) -> bool:
+    def _preparation_scope_available(self, session: dict[str, Any], *, runtime_preparation: bool = False) -> bool:
         # A local queue hint does not perform this fallible authority read.
         # Fresh admission must establish the current manager scope before
         # spending a preparation attempt; dispatch checks the scope again.
         from ...chat_manager_context import manager_authorization_scope_id_for_registry
 
         try:
+            if runtime_preparation:
+                # Preparation never starts a model request. Check the same
+                # optional transport owner before reserving a bounded retry.
+                from .codex_auth import native_api_transport
+                native_api_transport(self.controller.coordination_runtime_root,
+                    checked_session_home(self.controller.codex_home, session),
+                    consumer="worker" if session.get("project_context") is not None else "steward",
+                    model=session.get("requested_agent_model_id") or session.get("agent_model_id"))
+            if session.get("project_context") is not None:
+                # Resolve the existing project grant again; a project assistant
+                # has no manager portfolio scope to borrow.
+                self.controller.project_contexts.session_context(session)
+                return True
             scope = (self.controller.manager_scope_resolver(session)
                 if self.controller.manager_scope_resolver is not None else self.bindings.steward_scope(session))
             return bool(isinstance(scope, list) and self.controller.registry_path is not None

loopx/chat_runtime.py

@@ -987,16 +987,18 @@ def check_route_configuration():
                 raise
         except Exception as exc:
             history_restore.finish()
-            configuration_error = isinstance(exc, CodexChatAgentError) and exc.error_code == "unsupported_configuration"
+            error_code = exc.error_code if isinstance(exc, CodexChatAgentError) else None
+            preserve_error = error_code == "model_auth_unavailable" or (
+                settings_changed and error_code == "unsupported_configuration")
             if not history_restore.invalidated:
                 with commit():
                     self.store.update_session(
                         session_id,
                         status="busy" if accepted_turn_id is not None else "ready" if settings_changed else "resume_failed",
                         active_turn_id=accepted_turn_id,
-                        last_error_code="unsupported_configuration" if settings_changed and configuration_error else "resume_failed",
+                        last_error_code=error_code if preserve_error else "resume_failed",
                     )
-            if settings_changed and configuration_error:
+            if preserve_error:
                 raise
             gate = exc.gate if isinstance(exc, CodexChatAgentError) else None
             raise CodexChatAgentError(

loopx/control_plane/collaboration/conversation_binding.ts

@@ -630,6 +630,8 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
       || saved.project_ref !== selected.project_ref || (!steward && saved.grant !== selected.grant)
       || current.channel_id !== `${steward ? "manager.external.native" : "project.external"}.${selected.binding_id}.${saved.source_ref}`) return no;
   const failed = requireJsonObject(recovery.failed_turn, "original failed Turn");
+  const runtimePreparation = planRuntimePreparationRecovery(recovery, row, current, failed, version);
+  if (runtimePreparation !== null) return runtimePreparation;
   const preparation = steward ? planPreparationRecovery(recovery, row, current, failed, version) : null;
   if (preparation !== null) return preparation;
   // Once preparation has dispatched, a different terminal category must not
@@ -1278,6 +1280,79 @@ function legacyImagePreparation(recovery: JsonObject, failed: JsonObject): boole
     && images != null && samePreparationImages(images.request, images.failed);
 }
 
+/** The queue's durable started_at checkpoint precedes every model dispatch.
+ * Exact queued/failed records with that checkpoint still null qualify a local
+ * adapter preparation failure, including the older misleading resume error.
+ * Unknown or started Turns never enter this path. Continuations share the
+ * existing three-attempt counter across quota/routing/preparation families.
+ */
+function planRuntimePreparationRecovery(recovery: JsonObject, row: JsonObject, current: JsonObject,
+  failed: JsonObject, version: number): JsonObject | null {
+  if (!["model_auth_unavailable", "resume_failed"].includes(String(failed.error_code))) return null;
+  const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
+  const expectedError = failed.error_code === "model_auth_unavailable"
+    ? "Native account API transport is unavailable." : "The previous Agent conversation could not be restored.";
+  if (current.agent_id !== "codex" || failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
+      || failed.status !== "failed" || failed.error !== expectedError || failed.started_at !== null
+      || failed.first_event_at !== null || failed.response !== null || failed.upstream_turn_id !== null
+      || failed.delta_count !== 0 || failed.message !== row.message || failed.origin !== row.origin
+      || recovery.attachments_present !== false || recovery.steering_observed !== false
+      || recovery.native_observation != null || recovery.native_result_present !== false) return no;
+  const observed = recovery.preparation_observation == null ? null
+    : requireJsonObject(recovery.preparation_observation, "runtime preparation readback");
+  if (observed?.kind !== "core_runtime_preparation_failure" || observed.session_id !== failed.session_id
+      || observed.turn_id !== failed.turn_id || !Array.isArray(observed.events) || observed.events.length !== 2) return no;
+  const events = observed.events.map(value => requireJsonObject(value, "runtime preparation event"));
+  if (!events.every(item => typeof item.event_id === "string" && item.event_id.length > 0)
+      || events[0].event_id === events[1].event_id || events[0].kind !== "turn.queued"
+      || events[0].delivery_mode !== "session_queue" || events[1].kind !== "turn.failed"
+      || events[1].error_code !== failed.error_code || events[1].message !== expectedError) return no;
+  const original = `external-${ref(row.request_ref, "original request")}`;
+  const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "runtime preparation reservation");
+  if (row.context_recovery_history != null && !Array.isArray(row.context_recovery_history)) return no;
+  const history = ((row.context_recovery_history ?? []) as unknown[])
+    .map(value => requireJsonObject(value, "runtime preparation lineage"));
+  const lineage = [...history, ...(previous ? [previous] : [])];
+  const counters = ["quota", "routing", "preparation"];
+  const count = (item: JsonObject): number => {
+    const kinds = counters.filter(kind => item[`${kind}_attempt`] != null);
+    if (kinds.length !== 1 || item.version !== version
+        || ["read_only_review", "metadata_review", "work_attempt"].some(key => item[key] != null)) return 0;
+    const attempt = item[`${kinds[0]}_attempt`];
+    return Number.isSafeInteger(attempt) && Number(attempt) >= 1 && Number(attempt) <= 3
+      && item.client_turn_id === `${original}-recovery-${version}-${kinds[0]}-${attempt}` ? Number(attempt) : 0;
+  };
+  if (lineage.some(item => !count(item))) return no;
+  const reserved = previous?.failed_turn_id === failed.turn_id;
+  if (reserved && previous?.preparation_attempt == null) return no;
+  const prior = reserved ? history : lineage;
+  if (prior.some((item, i) => i > 0 && count(item) <= count(prior[i - 1]))) return no;
+  const last = prior.at(-1);
+  if (failed.client_turn_id !== (last?.client_turn_id ?? original)) return no;
+  const attempt = last ? count(last) + 1 : 1;
+  if (attempt > 3 || reserved && previous?.preparation_attempt !== attempt) return no;
+  const client = `${original}-recovery-${version}-preparation-${attempt}`;
+  if (reserved && previous?.client_turn_id !== client) return no;
+  if (reserved && recovery.admitted_turn != null) {
+    const admitted = requireJsonObject(recovery.admitted_turn, "admitted runtime preparation Turn");
+    if (admitted.client_turn_id !== client || admitted.session_id !== current.session_id
+        || admitted.turn_id === failed.turn_id || admitted.message !== row.message || admitted.origin !== row.origin
+        || !Array.isArray(admitted.attachments ?? []) || (admitted.attachments as unknown[] | undefined)?.length) return no;
+    return {...no, recovery_candidate: true, recovery_eligible: recovery.source_verified === true,
+      client_turn_id: client, admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted runtime preparation Turn"),
+      preparation_attempt: attempt, session_id: current.session_id, native_thread_ids: []};
+  }
+  if (current.active_turn_id != null || current.status !== "ready" || recovery.stop_observed !== false
+      || recovery.queued_input_present !== false || recovery.later_equivalent_input_present !== false) return no;
+  const now = Date.parse(String(recovery.observed_at)), completed = Date.parse(String(failed.completed_at));
+  const notBefore = completed + 60_000 * 2 ** (attempt - 1);
+  if (!Number.isFinite(now) || !Number.isFinite(completed) || now < notBefore) return no;
+  return {...no, recovery_candidate: true, recovery_eligible: recovery.source_verified === true
+    && recovery.preparation_scope_available === true, preparation_attempt: attempt,
+    not_before: new Date(notBefore).toISOString(), preparation_observation: observed,
+    client_turn_id: client, failed_turn_id: failed.turn_id, session_id: current.session_id, native_thread_ids: []};
+}
+
 function planPreparationRecovery(recovery: JsonObject, row: JsonObject, current: JsonObject,
   failed: JsonObject, version: number): JsonObject | null {
   const images = legacyImagePreparation(recovery, failed);

loopx/extensions/lark/manager_context.py

@@ -60,6 +60,7 @@ def manager_failure_reply(
         "server_restarted": "管家服务在处理过程中重启",
         "runtime_unavailable": "本地运行环境初始化失败，请检查 LoopX 安装与运行配置，修复后重试",
         "resume_failed": "原 Agent 会话恢复失败，请检查执行器状态，修复后重试",
+        "model_auth_unavailable": "账户模型连接暂时不可用",
         "manager_channel_executor_rebind_required": (
             "管家的执行器已由本机设置更改，需要重新应用一次管家连接"
         ),
@@ -76,6 +77,9 @@ def manager_failure_reply(
     if code == "manager_authorization_unavailable" and preparation_recovery:
         return code, ("已收到你的消息，但本轮在准备授权上下文时失败。请求已保留；只有核验为模型调用前失败、"
                       "连接恢复且没有后续重交时，原队列才会限次接续。可用 /status 查看状态。")
+    if code in {"model_auth_unavailable", "resume_failed"} and preparation_recovery:
+        return code, ("已收到你的消息，但本轮准备执行器连接时失败。请求已保留；只有核验为模型调用前失败、"
+                      "连接恢复且没有后续重交时，原队列才会按剩余次数接续。可用 /status 查看状态。")
     # Old hosts can terminate with `other` after typed disconnect retries.
     # Describe that observed history without changing the terminal category,
     # inferring an authorization failure, or granting permission to replay.

loopx/extensions/lark/private_conversations.py

@@ -1893,7 +1893,7 @@ def inbox() -> Path:
                                 for event in reversed(events) if event.get("kind") == "agent.phase"
                                 and event.get("payload", {}).get("method") == "error"), None)
                         preparation_recovery = (record["source"].get("private_human_message") is True
-                            and selected["binding"].get("context_kind") == "steward" and not native.get("agent_target"))
+                            and selected["binding"].get("context_kind") in {"project", "steward"} and not native.get("agent_target"))
                         response = manager_failure_reply(turn, retry_info=retry_info,
                             preparation_recovery=preparation_recovery)[1] + " 原会话已保留。"
                     else: