fix(chat): resume verified remaining work after result review

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

@@ -2066,6 +2066,16 @@ requires the original source, queue and native effect checks.
 The retained business obligation stays pending independently of the review's
 prose or failure; a review never starts a writable retry or another review.
 
+A completed review does not consume the original one-shot `work-1` reservation.
+If the original writable owner later becomes available through ordinary work,
+the existing reply worker can reconsider the retained original declaration.
+It must obtain fresh settled native-effect proof and recheck the exact original
+source, grant, completed review and binding snapshot. A later equivalent source
+submission still holds the old work. Review output and cold-reader inventory
+cannot substitute for that proof. Admission and dispatch use the existing
+continuation checks, and a crash adopts the same reserved Turn. A failed or
+still-incomplete continuation remains pending and gets no additional attempt.
+
 ### File-backed idle recovery observations
 
 Repeated recovery checks must not reparse every historical Turn and source body

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

@@ -1356,6 +1356,13 @@ completed Turn 继续明确保留，本路径不提供通用动作或执行重
 worker 先认回同一个 Turn。执行前仍重查原来源、队列和原生效果。原业务义务独立于复核正文或失败继续保留
 为待办。复核不能触发可写重试或又一次复核。
 
+已完成的复核不消耗原请求仅一次的 `work-1` 预约。如果原可写运行进程通过普通工作
+恢复可用，现有回复 worker 可以重新检查保留的原未完声明。它必须取得新鲜的原生
+效果结清证明，并重新核验原消息、授权、已完成复核和 binding 快照。后续相同正文的
+独立来源请求仍会暂停旧工作。复核正文和冷读取清单不能替代该证明。准入及执行前
+复用已有续办检查，崩溃只认回同一预约 Turn。续办失败或仍不完整时继续保持待办，
+不能增加执行次数。
+
 ### File 后端的空闲恢复观测
 
 逐条检查旧请求时，不应反复解析全部历史 Turn 和来源正文。File Chat provider

loopx/capabilities/native_chat/external_conversations.py

@@ -96,6 +96,18 @@ def __init__(self, controller: Any) -> None:
         self._source_control_discovery = FileChatDiscovery()
         self.actions: Any | None = None
 
+    def _retained_result_review(self, row: dict[str, Any]) -> dict[str, Any] | None:
+        """Read correlation only; the typed owner still checks continuation eligibility."""
+        reservation = row.get("context_recovery") or {}
+        if reservation.get("work_attempt") == 1:
+            history = row.get("context_recovery_history") or []
+            reservation = history[-1] if history else {}
+        if (reservation.get("read_only_review") is not True or not reservation.get("unfinished_work")
+                or reservation.get("answer_review") or reservation.get("history_review")):
+            return None
+        reviewed = self.controller.store.turn_for_client(row["session_id"], reservation["client_turn_id"])
+        return {"reservation": reservation, "turn": reviewed} if reviewed else None
+
     def _context_recovery_plan(self, row: dict[str, Any], *,
                                observation: dict[str, Any] | None = None,
                                source_verified: bool = False,
@@ -114,6 +126,11 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         turn = store.load_turn(str(row.get("session_id") or ""), str(row.get("turn_id") or ""))
         if not session or not turn:
             return {}
+        retained_review = self._retained_result_review(row)
+        if retained_review is not None:
+            turn = store.load_turn(row["session_id"], retained_review["reservation"]["failed_turn_id"])
+            if turn is None:
+                return {}
         if selected_binding is None and discovery_only:
             # A queue hint must not serialize provider identity IO across all
             # historical failures. Reuse the same typed planner with the local
@@ -129,11 +146,20 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
             if (reservation.get("routing_attempt") or reservation.get("quota_attempt") or reservation.get("preparation_attempt")
                 or reservation.get("read_only_review") or reservation.get("work_attempt")
                 or reservation.get("metadata_review"))
-            and reservation.get("failed_turn_id") == row["turn_id"] else None)
+            and reservation.get("failed_turn_id") == turn["turn_id"]
+            and (retained_review is None or reservation.get("work_attempt") == 1) else None)
         events = store.events_after(session["session_id"], turn["turn_id"], None)
         anchors = [event["payload"]["native_input"] for event in events
                    if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")]
         session = self._original_native_session(session, anchors)
+        live_work_owner = False
+        if retained_review is not None:
+            from .codex_adapter import CodexAppServerAdapter
+            with self.controller.lock:
+                adapter = self.controller.adapters.get(row["session_id"])
+            live_work_owner = (isinstance(adapter, CodexAppServerAdapter) and not adapter.session.read_only_review
+                and adapter.upstream_thread_id == session["upstream_thread_id"])
+            events += store.events_after(session["session_id"], retained_review["turn"]["turn_id"], None)
         # This only selects read-only result probes. Adoption still requires
         # the exact original anchor, native final, current source and typed rule.
         from ...chat import _review_openings
@@ -156,6 +182,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         if preparation is not None or unfinished:
             created = datetime.fromisoformat(turn["created_at"])
             later_equivalent = any(item.get("role") == "user" and item.get("turn_id") != turn["turn_id"]
+                and (retained_review is None or item.get("turn_id") != retained_review["turn"]["turn_id"])
                 and item.get("text") == row.get("message") and item.get("origin") in {None, "lark", "web"}
                 and datetime.fromisoformat(item["created_at"]) > created
                 for item in (discovery_reads.user_messages(session["session_id"])
@@ -191,6 +218,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                     for item in (discovery_reads.source_controls() if discovery_reads is not None
                                  else self.source_controls())) if turn.get("error_code") == "antigravity_step_error" else None,
                 "native_input": anchors[0] if len(anchors) == 1 else None,
+                **({"retained_result_review": retained_review, "live_work_owner": live_work_owner}
+                   if retained_review is not None else {}),
                 "response_literal_open": bool(open_literal),
                 "metadata_capacity": metadata_capacity,
                 "executor_configuration": steward_machine_defaults(self.controller) if metadata_capacity else None,
@@ -349,7 +378,7 @@ def context_recovery_candidate(self, request_ref: str, *,
                 from ...chat import _review_openings
                 _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0)
                 invalid_metadata = ((turn.get("response") or {}).get("gate") or {}).get("kind") == "response_metadata_invalid"
-                if not open_literal and not invalid_metadata:
+                if not open_literal and not invalid_metadata and not self._retained_result_review(row):
                     return False  # A lexical probe hint, never permission to adopt or send.
             return self._context_recovery_plan(row, discovery_only=True,
                 discovery_reads=discovery_reads).get("recovery_candidate") is True
@@ -375,8 +404,10 @@ def recover_failed_request(self, request_ref: str, *,
             if not plan.get("recovery_candidate"):
                 return row
             session = self.controller.store.load_session(row["session_id"])
-            turn = self.controller.store.load_turn(row["session_id"], row["turn_id"])
-            assert session is not None and turn is not None
+            current_turn = self.controller.store.load_turn(row["session_id"], row["turn_id"])
+            turn = (self.controller.store.load_turn(row["session_id"], plan["failed_turn_id"])
+                    if plan.get("completed_review_turn_id") else current_turn)
+            assert session is not None and turn is not None and current_turn is not None
             if plan.get("recover_native_result"):
                 return self._recover_native_result(path, row, session, turn, source_observation)
             metadata_review = plan.get("metadata_review") is True
@@ -497,7 +528,9 @@ def source_requests() -> list[dict[str, Any]]:
                 # its execution and newer input may legitimately have advanced.
                 # The typed planner below must still prove that exact receipt.
                 if not plan.get("admitted_turn_id") and (self.controller.store.load_session(row["session_id"]) != session
-                        or self.controller.store.load_turn(row["session_id"], row["turn_id"]) != turn
+                        or self.controller.store.load_turn(row["session_id"], current_turn["turn_id"]) != current_turn
+                        or plan.get("completed_review_turn_id")
+                        and self.controller.store.load_turn(row["session_id"], turn["turn_id"]) != turn
                         or source_requests() != requests):
                     return row
                 plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
@@ -523,7 +556,7 @@ def source_requests() -> list[dict[str, Any]]:
                 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"],
+                    recovery = {"version": MANAGER_CONTEXT_VERSION, "failed_turn_id": turn["turn_id"],
                         "native_observation": proof, "source_observation": source,
                         "client_turn_id": plan["client_turn_id"],
                         "failure_notice_delivery_verified": row["delivery_verified"]}
@@ -548,6 +581,9 @@ def source_requests() -> list[dict[str, Any]]:
                         recovery.update(work_attempt=1, unfinished_work=plan["unfinished_work"],
                             prior_result_delivery_verified=row["delivery_verified"],
                             presentation_key=f"{MANAGER_CONTEXT_VERSION}.work.1", not_before=plan["not_before"])
+                        if plan.get("completed_review_turn_id"):
+                            recovery.update(completed_review_turn_id=plan["completed_review_turn_id"],
+                                original_turn=deepcopy(turn), binding_snapshot=selected["binding"])
                     for kind in ("routing", "quota", "preparation"):
                         if plan.get(f"{kind}_attempt") is not None:
                             recovery.update({f"{kind}_attempt": plan[f"{kind}_attempt"],
@@ -1185,6 +1221,18 @@ def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, An
         if not continuation or continuation.get("work_attempt") != 1:
             return message
         proof = continuation.get("native_observation") or {}
+        if continuation.get("completed_review_turn_id"):
+            row = self.read_request(continuation["request_ref"])
+            retained = self._retained_result_review(row)
+            original = self.controller.store.load_turn(session["session_id"], continuation["failed_turn_id"])
+            binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
+            if (not retained or retained["turn"]["turn_id"] != continuation["completed_review_turn_id"]
+                    or retained["turn"].get("status") != "completed" or original != continuation["original_turn"]
+                    or binding != continuation["binding_snapshot"]
+                    or self._source_stop_observed(row, original or {}, self.controller.store.events_after(
+                        session["session_id"], continuation["failed_turn_id"], None))):
+                raise CodexChatAgentError("The retained work source, review or grant changed; continuation stopped.",
+                    error_code="host_gate", gate=None)
         if (self.controller.store.queued_turns(session["session_id"])
                 or completed_work_readback(native, str(proof.get("turn_id") or ""),
                     known_turn_ids=self._settled_native_turn_ids(session["session_id"])) != proof):

loopx/control_plane/collaboration/conversation_binding.ts

@@ -567,7 +567,9 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
     // the reviewer says; neither a delivered review nor its failure is replay authority.
     const originalResult = reservation?.original_turn as JsonObject | undefined;
     const originalResponse = originalResult?.response as JsonObject | undefined;
-    const retainedWork = reservation?.read_only_review === true && reservation.answer_review !== true
+    const retainedWork = reservation != null && reservation.answer_review !== true
+      && (reservation.read_only_review === true || reservation.work_attempt === 1
+        && typeof reservation.completed_review_turn_id === "string" && !pureResultText(response))
       && turn?.client_turn_id === reservation.client_turn_id && originalResult?.status === "completed"
       && originalResult.session_id === row.session_id && originalResult.turn_id === reservation.failed_turn_id
       && originalResult.turn_id !== row.turn_id && originalResponse != null
@@ -710,6 +712,9 @@ 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");
+  if (recovery.retained_result_review != null) {
+    return planWorkAfterResultReview(recovery, row, current, failed, selected, version);
+  }
   const runtimePreparation = planRuntimePreparationRecovery(recovery, row, current, failed, version);
   if (runtimePreparation !== null) return runtimePreparation;
   const preparation = steward ? planPreparationRecovery(recovery, row, current, failed, version) : null;
@@ -1128,6 +1133,46 @@ function unfinishedProjectWork(result: JsonObject): JsonObject | null {
   return work;
 }
 
+function planWorkAfterResultReview(recovery: JsonObject, row: JsonObject, current: JsonObject,
+  original: JsonObject, selected: JsonObject, version: number): JsonObject {
+  const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
+  const retained = requireJsonObject(recovery.retained_result_review, "retained result review");
+  const reservation = requireJsonObject(retained.reservation, "original review reservation");
+  const reviewed = requireJsonObject(retained.turn, "completed result review");
+  const previous = requireJsonObject(row.context_recovery, "current recovery reservation");
+  const history = Array.isArray(row.context_recovery_history) ? row.context_recovery_history as JsonObject[] : [];
+  const continuing = previous.work_attempt === 1;
+  const expected = continuing ? history.at(-1) : previous;
+  const work = original.response != null && typeof original.response === "object" && !Array.isArray(original.response)
+    ? unfinishedProjectWork(original.response as JsonObject) : null;
+  // A delivered read-only report retains intent. Only a fresh observation by
+  // the original writable owner can admit the already bounded remaining work.
+  if (selected.context_kind !== "project" || current.agent_id !== "codex" || work === null
+      || !isDeepStrictEqual(expected, reservation) || reservation.read_only_review !== true
+      || reservation.answer_review === true || reservation.history_review === true || reservation.work_attempt != null
+      || reservation.version !== version || !isDeepStrictEqual(reservation.binding_snapshot, selected)
+      || reservation.client_turn_id !== `external-${row.request_ref}-recovery-${version}-review-1`
+      || !isDeepStrictEqual(reservation.original_turn, original) || !isDeepStrictEqual(reservation.unfinished_work, work)
+      || original.status !== "completed" || original.session_id !== current.session_id
+      || original.turn_id !== reservation.failed_turn_id || original.client_turn_id !== `external-${row.request_ref}`
+      || original.message !== row.message || original.origin !== row.origin
+      || reviewed.status !== "completed" || reviewed.session_id !== current.session_id
+      || reviewed.turn_id !== row.turn_id || reviewed.turn_id === original.turn_id
+      || reviewed.client_turn_id !== reservation.client_turn_id || reviewed.message !== row.message
+      || reviewed.origin !== row.origin || typeof reviewed.upstream_turn_id !== "string" || !reviewed.upstream_turn_id
+      || !pureResultText(reviewed.response as JsonObject | null)
+      || history.some(item => item.work_attempt != null)
+      || continuing && (previous.failed_turn_id !== original.turn_id
+        || previous.completed_review_turn_id !== reviewed.turn_id)) return no;
+  const admitted = continuing ? recovery.admitted_turn : null;
+  if (admitted == null && (recovery.live_work_owner !== true || recovery.later_equivalent_input_present !== false)) return no;
+  const proof = recovery.native_observation as JsonObject | null | undefined;
+  if (proof != null && (proof.completed_work !== true || proof.all_effects_settled !== true)) return no;
+  const plan = planUnfinishedWork({...recovery, admitted_turn: admitted}, {...row, turn_id: original.turn_id},
+    current, original, continuing ? previous : null, version);
+  return plan.work_attempt === 1 ? {...plan, completed_review_turn_id: reviewed.turn_id} : no;
+}
+
 function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: JsonObject,
   completed: JsonObject, previous: JsonObject | null, version: number): JsonObject {
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};