fix(chat): recover original steward replies after a Codex service restart

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

@@ -1816,6 +1816,25 @@ native inventory reads and crashes before admission, after enqueue and before
 transport synchronization. Synthetic model/provider journeys and read-only native
 observations do not certify live model completion or the wider conversation SLA.
 
+### A steward answer interrupted by service restart
+
+A Codex steward using the configured flexible allowance route can reserve one
+restricted answer review after `server_restarted`, even when tools already ran.
+The original reply worker must independently read a stable, complete interrupted
+native inventory with no final answer or live items. Native dynamic-tool calls
+are effect items too; a stable terminal inventory permits only this read-only
+review, never business replay. Its digest is rechecked before review execution.
+
+The same source, owner, grant, stop checks and `review-1` queue reservation apply.
+Fresh configured Codex capacity must verify before admission and again before
+dispatch. The new restricted thread retains only authorized manager reads; writes,
+network effects, handoffs and business continuation are disabled. It answers from
+existing evidence or gives a concrete partial answer. The original failed Turn
+and failure notice remain unchanged; restart or an uncertain enqueue response
+adopts the reserved review instead of granting another attempt. Missing native
+evidence or capacity remains pending. The failure notice describes this qualified
+recovery instead of giving every restart the same blanket no-replay message.
+
 ### Explicit unfinished work after a completed Turn
 
 A terminal model Turn is distinct from completed business work. A private

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

@@ -1185,6 +1185,19 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果
 传输同步前的中断。合成模型/provider 通路及只读原生观察不证明真实模型完成，
 也不代表更广泛的会话 SLA 已通过。
 
+### 管家答复被服务重启中断
+
+使用已配置灵活额度路由的 Codex 管家，遇到 `server_restarted` 且工具已经执行后，
+仍可预约一次受限的答案复核。原回复 worker 必须独立读回稳定、完整、没有最终答案
+和活动 item 的原生中断清单。原生 dynamic-tool 调用同样属于效果条目；稳定的终态
+清单只允许这次只读复核，不授予业务重放权。实际复核前再次核验清单摘要。
+
+复核沿用原来源、本人、grant、停止检查及同一 `review-1` 队列预留。受理及派发前
+都须重新核验已配置的 Codex 额度。新受限 thread 只保留原授权管家读取，禁用写入、
+网络效果、委派及业务续办；基于现有证据答复，或给出具体的部分答案。原失败 Turn
+与失败通知不变；重启或入队回执不明时采用已预留复核，不产生新次数。缺少原生证据
+或额度仍保留待办。失败提示说明这些恢复条件，不再对所有重启一律提示不会自动重放。
+
 ### Turn 完成后的显式未完工作
 
 模型 Turn 终态与业务完成分开。私有项目 worker 核验已结束工作的实际结果后，可在

loopx/capabilities/native_chat/codex_adapter.py

@@ -413,7 +413,7 @@ def turns() -> list[dict[str, Any]]:
             elif kind == "agentMessage":
                 if item.get("phase") != "commentary":
                     return None
-            elif kind in {"commandExecution", "fileChange", "mcpToolCall"}:
+            elif kind in {"commandExecution", "fileChange", "mcpToolCall", "dynamicToolCall"}:
                 if item.get("status") not in {"completed", "failed", "declined"}:
                     return None
                 effects += 1
@@ -423,7 +423,10 @@ def turns() -> list[dict[str, Any]]:
                 or list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline)) != items):
             return None
         return {"thread_id": session.thread_id, "turn_id": turn_id, "read_only_review": True,
-                "no_final_answer": True, "effects_observed": True}
+                "no_final_answer": True, "effects_observed": True,
+                "native_inventory_sha256": hashlib.sha256(json.dumps([
+                    next(row for row in inventory if row["id"] == turn_id), items],
+                    sort_keys=True, separators=(",", ":")).encode()).hexdigest()}
     except (CodexChatAgentError, OSError, TypeError, ValueError):
         return None
 

loopx/capabilities/native_chat/external_conversations.py

@@ -379,6 +379,7 @@ def recover_failed_request(self, request_ref: str, *,
                 return self._recover_native_result(path, row, session, turn, source_observation)
             metadata_review = plan.get("metadata_review") is True
             answer_review = plan.get("answer_review") is True
+            codex_answer = answer_review and session.get("agent_id") == "codex"
             quota_answer = answer_review and session.get("agent_id") == "codex" and turn.get("error_code") == "usage_limit_exceeded"
             preparation_retry = plan.get("preparation_attempt") is not None
             model_auth_home = (None if metadata_review or answer_review or preparation_retry
@@ -397,7 +398,7 @@ def source_requests() -> list[dict[str, Any]]:
             # checks and the typed eligibility decision remain below that fence.
             proof = None
             capacity = None
-            if not plan.get("admitted_turn_id") and (metadata_review or answer_review and not quota_answer or turn.get("error_code") == "antigravity_step_error"):
+            if not plan.get("admitted_turn_id") and (metadata_review or answer_review and not codex_answer or turn.get("error_code") == "antigravity_step_error"):
                 from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL
                 events = self.controller.store.events_after(row["session_id"], row["turn_id"], None)
                 anchors = [item["payload"]["native_input"] for item in events
@@ -452,6 +453,11 @@ def source_requests() -> list[dict[str, Any]]:
                                 capacity = {"alternate_executor": route}
                             else:
                                 proof.update(quota_available=True, quota_observed_at=route["observed_at"])
+            if codex_answer and not quota_answer and proof is not None:
+                from ...chat_manager import manager_session_quota_route
+                route = manager_session_quota_route(self.controller, session)
+                if route and route["allocation"]["executor_endpoint"] == "codex":
+                    capacity = {"alternate_executor": route}
             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)
@@ -801,6 +807,12 @@ def _checked_answer_review(self, session: dict[str, Any], turn: dict[str, Any],
                 if observed is not None:
                     observed = {key: observed.get(key) for key in ("thread_id", "turn_id", "only_original_user_item", "codex_error_info")}
                 expected = {key: proof.get(key) for key in ("thread_id", "turn_id", "only_original_user_item", "codex_error_info")}
+            elif proof.get("read_only_review") is True and proof.get("turn_id"):
+                from .codex_adapter import CodexAppServerAdapter
+                observed = CodexAppServerAdapter.failed_request_observation(codex_bin=self.controller.codex_bin,
+                    codex_home=Path(session["codex_home"]), thread_ids=[proof["thread_id"]],
+                    turn_id=proof["turn_id"], read_only_review=True)
+                expected = proof
             else:
                 observed = AntigravityAdapter.interrupted_answer_observation(cascade_id=proof["thread_id"],
                     request_ref=row["request_ref"], message=original["message"], started_at=original["started_at"],

loopx/control_plane/collaboration/conversation_binding.ts

@@ -743,15 +743,15 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
   if ((previous?.read_only_review === true && previous.failed_turn_id !== row.turn_id)
       || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.some(value =>
         requireJsonObject(value, "prior recovery").read_only_review === true))) return no;
+  if (failed.error_code === "server_restarted") {
+    return planResultReview(recovery, row, current, failed, previous, version);
+  }
   const exhaustedAnswer = steward && current.agent_id === "codex"
     ? planExhaustedStewardAnswer(recovery, row, current, failed, previous, version) : null;
   if (exhaustedAnswer !== null) return exhaustedAnswer;
   if (failed.status === "completed" && !steward) {
     return planUnfinishedWork(recovery, row, current, failed, previous, version);
   }
-  if (!steward && failed.error_code === "server_restarted") {
-    return planResultReview(recovery, row, current, failed, previous, version);
-  }
   if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
       || failed.status !== "failed" || failed.delta_count !== 0 || failed.response !== null
       || recovery.steering_observed !== false
@@ -1223,16 +1223,20 @@ function planInterruptedStewardAnswer(recovery: JsonObject, row: JsonObject, cur
   return {...capacity, recovery_eligible: capacity.recovery_eligible === true && capacity.answer_executor === "codex"};
 }
 
-/** Original project result reviews retain their workspace-only read policy. */
+/** Reuse the original queue for a restricted result review, never business replay. */
 function planResultReview(recovery: JsonObject, row: JsonObject, current: JsonObject,
   failed: JsonObject, previous: JsonObject | null, version: number): JsonObject {
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
+  const steward = current.steward_context != null;
+  const review = steward ? {answer_review: true, answer_executor: "codex"} : {};
   if (failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
       || failed.status !== "failed" || failed.response !== null
       || !Number.isSafeInteger(failed.delta_count) || Number(failed.delta_count) <= 0
       || recovery.steering_observed !== false || recovery.stop_observed !== false
       || recovery.attachments_present !== false
-      || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id) return no;
+      || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id
+      || steward && (failed.client_turn_id !== `external-${row.request_ref}`
+        || failed.message !== row.message || failed.origin !== row.origin)) return no;
   const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-review-1`;
   if (previous?.read_only_review === true && (previous.failed_turn_id !== failed.turn_id
       || previous.version !== version || previous.client_turn_id !== client)) return no;
@@ -1242,23 +1246,28 @@ function planResultReview(recovery: JsonObject, row: JsonObject, current: JsonOb
         || admitted.turn_id === failed.turn_id || admitted.message !== row.message
         || admitted.origin !== row.origin || (admitted.attachments != null
           && (!Array.isArray(admitted.attachments) || admitted.attachments.length))) return no;
-    return {...no, recovery_candidate: true, recovery_eligible: recovery.source_verified === true,
+    return {...no, ...review, recovery_candidate: true, recovery_eligible: recovery.source_verified === true,
       read_only_review: true, client_turn_id: client, session_id: current.session_id,
       admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted result review"), native_thread_ids: []};
   }
   if (current.active_turn_id != null || current.status !== "ready") return no;
   const now = Date.parse(String(recovery.observed_at)), completed = Date.parse(String(failed.completed_at));
   if (!Number.isFinite(now) || !Number.isFinite(completed) || now < completed + 60_000) return no;
   const native = sessionIdentity(current.upstream_thread_id, "original native thread");
-  const candidate = {...no, recovery_candidate: true, read_only_review: true,
+  const candidate = {...no, ...review, recovery_candidate: true, read_only_review: true,
     client_turn_id: client, session_id: current.session_id, failed_turn_id: failed.turn_id,
     native_thread_ids: [native]};
   if (recovery.native_observation == null) return candidate;
   const observed = requireJsonObject(recovery.native_observation, "interrupted result inventory");
-  return {...candidate, recovery_eligible: recovery.source_verified === true
+  const eligible = recovery.source_verified === true
     && observed.thread_id === native && observed.turn_id === failed.upstream_turn_id
     && observed.read_only_review === true && observed.no_final_answer === true
-    && observed.effects_observed === true};
+    && observed.effects_observed === true && (!steward
+      || typeof observed.native_inventory_sha256 === "string"
+        && BARE_SHA256_PATTERN.test(observed.native_inventory_sha256));
+  if (!eligible || !steward) return {...candidate, recovery_eligible: eligible};
+  const capacity = nativeRecoveryCapacity(recovery, row, current, candidate);
+  return {...capacity, recovery_eligible: capacity.recovery_eligible === true && capacity.answer_executor === "codex"};
 }
 
 /** A local manager-context failure precedes native dispatch. Only the Core's

loopx/extensions/lark/manager_context.py

@@ -80,6 +80,9 @@ def manager_failure_reply(
     if code in {"model_auth_unavailable", "resume_failed"} and preparation_recovery:
         return code, ("已收到你的消息，但本轮准备执行器连接时失败。请求已保留；只有核验为模型调用前失败、"
                       "连接恢复且没有后续重交时，原队列才会按剩余次数接续。可用 /status 查看状态。")
+    if code == "server_restarted":
+        return code, ("处理被管家服务重启中断，原请求和会话已保留。系统会先核对原执行记录，"
+                      "符合条件的由原队列恢复结果回复；执行效果不明的操作不会直接重做。")
     # 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.

tests/control_plane_ts/conversation_binding.test.ts

@@ -19,6 +19,65 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision
 const request = {current, expected_revision: 0, operation: "configure", binding: row, observation,
   available_projects: [project]};
 
+test("a restarted Codex steward retains one restricted answer instead of replaying business", () => {
+  const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: []};
+  const configured = planConversationBinding({...request, binding: selected}).state;
+  const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};
+  const context = resolveBoundConversation({current: configured, binding_id: row.binding_id,
+    observation, available_projects: [project], ...source}).context;
+  const ref = "f".repeat(24), message = "Explain the existing results";
+  const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "codex", upstream_mode: "chat",
+    status: "ready", active_turn_id: null, upstream_thread_id: "native", steward_context: context,
+    channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`};
+  const failed = {session_id: "original", turn_id: "failed", client_turn_id: `external-${ref}`,
+    message, origin: "lark", status: "failed", error_code: "server_restarted", response: null,
+    delta_count: 1, upstream_turn_id: "interrupted", completed_at: "2026-01-01T00:00:00Z"};
+  const proof = {thread_id: "native", turn_id: "interrupted", read_only_review: true,
+    no_final_answer: true, effects_observed: true, native_inventory_sha256: "1".repeat(64)};
+  const allocation = {schema_version: "manager_executor_allocation_v0", selection_policy: "flexible",
+    configured_endpoint: "antigravity", executor_endpoint: "codex", allocation_reason: "flexible_availability_fallback",
+    configuration_revision: "current-pool", available: true, eligible_endpoints: ["antigravity", "codex"]};
+  const configuration = {status: "ready", selection_policy: "flexible", executor_endpoint: "antigravity",
+    eligible_endpoints: ["antigravity", "codex"], configuration_revision: "current-pool"};
+  const packet = {request: {request_ref: ref, command: null, status: "accepted", delivery_verified: true,
+    message, origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "failed", source},
+    binding: selected, current_session: session, context_recovery: {version: 22, failed_turn: failed,
+      observed_at: "2026-01-01T00:10:00Z", source_verified: true, steering_observed: false,
+      stop_observed: false, attachments_present: false, queued_input_present: false, native_observation: proof,
+      metadata_capacity: {alternate_executor: {allocation, observed_at: Date.parse("2026-01-01T00:10:00Z") / 1000}},
+      executor_configuration: configuration}};
+  const plan = planBoundConversationRequest(packet);
+  assert.equal(plan.recovery_eligible, true);
+  assert.equal(plan.answer_review, true);
+  assert.equal(plan.read_only_review, true);
+  assert.equal(plan.answer_executor, "codex");
+  assert.equal(plan.work_attempt, undefined);
+  assert.equal(plan.quota_attempt, undefined);
+  assert.equal(plan.client_turn_id, `external-${ref}-recovery-22-review-1`);
+  for (const change of [{source_verified: false}, {stop_observed: true}, {steering_observed: true},
+    {attachments_present: true}, {observed_at: "2026-01-01T00:00:59Z"}, {metadata_capacity: null},
+    ...[{thread_id: "foreign"}, {turn_id: "other"}, {no_final_answer: false},
+      {effects_observed: false}, {read_only_review: false}, {native_inventory_sha256: undefined}]
+      .map(value => ({native_observation: {...proof, ...value}})),
+    ...[{delta_count: 0}, {client_turn_id: "foreign"}, {message: "changed"}, {origin: "web"},
+      {response: {}}, {error_code: "runtime_error"}].map(value => ({failed_turn: {...failed, ...value}}))]) {
+    assert.notEqual(planBoundConversationRequest({...packet,
+      context_recovery: {...packet.context_recovery, ...change}}).recovery_eligible, true);
+  }
+  assert.notEqual(planBoundConversationRequest({...packet,
+    current_session: {...session, status: "busy", active_turn_id: "later"}}).recovery_eligible, true);
+  const previous = {version: 22, failed_turn_id: "failed", client_turn_id: plan.client_turn_id,
+    answer_review: true, read_only_review: true};
+  const admitted = {session_id: "original", turn_id: "review", client_turn_id: plan.client_turn_id,
+    message, origin: "lark"};
+  const reserved = {...packet, request: {...packet.request, context_recovery: previous},
+    context_recovery: {...packet.context_recovery, admitted_turn: admitted}};
+  assert.equal(planBoundConversationRequest(reserved).admitted_turn_id, "review");
+  assert.equal(planBoundConversationRequest({...reserved, request: {...reserved.request, turn_id: "review"},
+    context_recovery: {...reserved.context_recovery, admitted_turn: null,
+      failed_turn: {...failed, turn_id: "review", client_turn_id: plan.client_turn_id}}}).recovery_candidate, false);
+});
+
 test("an original project account-preparation failure shares its existing recovery budget", () => {
   const configured = planConversationBinding(request).state;
   const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};