fix(chat): qualify unfinished work after a delivered result review

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

@@ -2063,8 +2063,18 @@ separate, idempotent phase. Reservation crashes adopt that exact queued review.
 If writable continuation was already admitted before the request journal was
 updated, the worker first recovers that same queued Turn. Dispatch still
 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.
+The retained business obligation stays pending after the review. Once a
+successful plain-text review is delivered, the original worker can independently
+qualify the still-unspent `work-1` continuation against the original typed
+partial. The current owner of the same native thread must prove settled effects
+again, including the intervening review and other later inputs. Core rechecks
+the original source and grant, queue, stop and steering, and exact original and
+review identities. The worker verifies the inventory again before execution and
+returns the result separately from both earlier replies. A review failure,
+review gate, loss of native ownership, superseding input or an uncertain effect
+leaves the original obligation pending. An admitted continuation stays visible
+while it is failed, gated, unfinished or awaiting verified result delivery.
+The review itself grants no execution authority and can run only once.
 
 ### File-backed idle recovery observations
 

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

@@ -1353,8 +1353,15 @@ completed Turn 继续明确保留，本路径不提供通用动作或执行重
 发送、委派或替换原未完声明。旧问题中的指代仅使用原提问之前的 Core 上下文。
 原部分答复 Turn 和投递保持不变，由同一个飞书 worker 使用独立、幂等阶段补回新
 答复；预约崩溃只采用同一个已入队复核。如果可写续办已入队、原请求日志尚未更新，
-worker 先认回同一个 Turn。执行前仍重查原来源、队列和原生效果。原业务义务独立于复核正文或失败继续保留
-为待办。复核不能触发可写重试或又一次复核。
+worker 先认回同一个 Turn。执行前仍重查原来源、队列和原生效果。
+
+复核后原业务义务继续保留。无 gate 的复核成功送达后，原 worker 可以依据原有结构化
+部分答复重新核验尚未使用的 `work-1` 续办资格。原线程的当前 owner 必须重新证明效果
+已结清，包括期间的复核与后续输入。Core 核对原来源、授权、队列、停止与 steering，
+以及原 Turn 与复核 Turn 的精确关联。原 worker 在执行前再次核验效果清单，续办结果
+独立于此前两次答复送达。复核失败、复核 gate、原生 owner 丢失、新输入替代或效果未知
+时继续保留原待办。已准入的续办在失败、有 gate、未完成或尚未核验结果送达时继续可见。
+复核本身不授予执行权限，每条请求最多复核一次。
 
 ### File 后端的空闲恢复观测
 

loopx/canary/module_metric_baseline.json

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

loopx/capabilities/native_chat/external_conversations.py

@@ -96,6 +96,21 @@ def __init__(self, controller: Any) -> None:
         self._source_control_discovery = FileChatDiscovery()
         self.actions: Any | None = None
 
+    @staticmethod
+    def _completed_work_review(row: dict[str, Any]) -> dict[str, Any]:
+        review = row.get("context_recovery") or {}
+        if review.get("work_attempt") == 1:
+            review = (row.get("context_recovery_history") or [{}])[-1]
+        return review if review.get("read_only_review") is True and review.get("unfinished_work") else {}
+
+    @staticmethod
+    def _has_later_equivalent_input(row: dict[str, Any], original: dict[str, Any],
+                                    messages: Sequence[dict[str, Any]], *, internal_turns: set[str]) -> bool:
+        created = datetime.fromisoformat(original["created_at"])
+        return any(item.get("role") == "user" and item.get("turn_id") not in internal_turns
+            and item.get("text") == row.get("message") and item.get("origin") in {None, "lark", "web", "external"}
+            and datetime.fromisoformat(item["created_at"]) > created for item in messages)
+
     def _context_recovery_plan(self, row: dict[str, Any], *,
                                observation: dict[str, Any] | None = None,
                                source_verified: bool = False,
@@ -134,6 +149,12 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         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)
+        original = None
+        review = self._completed_work_review(row)
+        if review and review["client_turn_id"] == turn["client_turn_id"]:
+            original = store.load_turn(session["session_id"], review["failed_turn_id"])
+            if original is not None:
+                events += store.events_after(session["session_id"], original["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
@@ -153,13 +174,12 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                     {"event_id": item["event_id"], "kind": item["kind"],
                      **{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]}
-        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 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"])
-                             if discovery_reads is not None else store.messages(session["session_id"])))
+        if preparation is not None or unfinished or original is not None:
+            work_turn = original or turn
+            messages = (discovery_reads.user_messages(session["session_id"])
+                        if discovery_reads is not None else store.messages(session["session_id"]))
+            later_equivalent = self._has_later_equivalent_input(row, work_turn, messages,
+                internal_turns={turn["turn_id"], work_turn["turn_id"]})
         from ...chat_manager import steward_machine_defaults
         return effect_runtime_result("collaboration.conversation.request", {
             "request": {key: value for key, value in row.items() if key != "attachments"},
@@ -183,6 +203,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 "stop_observed": self._source_stop_observed(row, turn, events,
                     source_controls=discovery_reads.source_controls if discovery_reads is not None else None),
                 "preparation_observation": preparation,
+                "reviewed_original_turn": original,
                 "later_equivalent_input_present": later_equivalent,
                 "preparation_scope_available": preparation_scope_available,
                 "later_source_input_present": any(item.get("binding_id") == row["binding_id"]
@@ -345,7 +366,9 @@ def context_recovery_candidate(self, request_ref: str, *,
             turn = self.controller.store.load_turn(row["session_id"], row["turn_id"])
             if not turn or turn.get("status") not in {"failed", "completed"}:
                 return False
-            if turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work"):
+            review = self._completed_work_review(row)
+            if (turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work")
+                    and review.get("client_turn_id") != turn.get("client_turn_id")):
                 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"
@@ -431,11 +454,16 @@ def source_requests() -> list[dict[str, Any]]:
                 if result_review or plan.get("work_attempt"):
                     with self.controller.lock:
                         adapter = self.controller.adapters.get(row["session_id"])
-                    if not result_review and isinstance(adapter, CodexAppServerAdapter):
+                    original_work_id = plan.get("work_original_turn_id")
+                    work_turn = (self.controller.store.load_turn(row["session_id"], original_work_id)
+                                 if original_work_id else turn)
+                    if not result_review and work_turn is not None and isinstance(adapter, CodexAppServerAdapter):
                         # Only the original live owner can prove writable effects.
                         proof = adapter.completed_work_observation(
-                            thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"],
+                            thread_ids=plan["native_thread_ids"], turn_id=work_turn["upstream_turn_id"],
                             known_turn_ids=self._settled_native_turn_ids(session["session_id"]))
+                    if original_work_id and proof is None:
+                        return row  # The delivered review is retained; no second review or cold proof.
                     if proof is None and (adapter is None or isinstance(adapter, CodexAppServerAdapter)
                             and adapter.upstream_thread_id in plan["native_thread_ids"]
                             and adapter.answer_review_replaceable()):
@@ -548,6 +576,8 @@ 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("work_original_turn_id"):
+                            recovery["work_original_turn_id"] = plan["work_original_turn_id"]
                     for kind in ("routing", "quota", "preparation"):
                         if plan.get(f"{kind}_attempt") is not None:
                             recovery.update({f"{kind}_attempt": plan[f"{kind}_attempt"],
@@ -1159,6 +1189,33 @@ def _checked_completed_result_review(self, session: dict[str, Any], turn: dict[s
                 raise CodexChatAgentError("The original completed result changed; review held.", error_code="host_gate", gate=None)
         return review
 
+    def _check_reviewed_work(self, session: dict[str, Any], turn: dict[str, Any],
+                             continuation: dict[str, Any]) -> None:
+        from ...chat_agent import CodexChatAgentError
+
+        row = self.read_request(continuation["request_ref"])
+        review = self._completed_work_review(row)
+        store = self.controller.store
+        session_id = session["session_id"]
+        original = store.load_turn(session_id, continuation["work_original_turn_id"])
+        previous = store.load_turn(session_id, continuation["failed_turn_id"])
+        current = store.load_session(session_id) or {}
+        binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
+        if (original is None or original != review.get("original_turn")
+                or previous is None or previous["status"] != "completed"
+                or previous["client_turn_id"] != review.get("client_turn_id")
+                or binding != review.get("binding_snapshot") or current.get("active_turn_id") != turn["turn_id"]
+                or row["turn_id"] not in {turn["turn_id"], previous["turn_id"]}):
+            raise CodexChatAgentError("The original result, review, source or grant changed; continuation held.",
+                                      error_code="host_gate", gate=None)
+        events = [event for target in (original, previous, turn)
+                  for event in store.events_after(session_id, target["turn_id"], None)]
+        if (self._source_stop_observed(row, original, events) or any(item.get("kind") == "turn.steered" for item in events)
+                or self._has_later_equivalent_input(row, original, store.messages(session_id, strict=True),
+                    internal_turns={original["turn_id"], previous["turn_id"], turn["turn_id"]})):
+            raise CodexChatAgentError("The original source stopped or has newer input; continuation held.",
+                                      error_code="host_gate", gate=None)
+
     def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, Any],
                                    native: Any, message: str) -> str:
         """Recheck reserved unfinished work at dispatch; keep the stored input intact."""
@@ -1184,6 +1241,8 @@ def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, An
         continuation = self.recovery_context(session, turn)
         if not continuation or continuation.get("work_attempt") != 1:
             return message
+        if continuation.get("work_original_turn_id"):
+            self._check_reviewed_work(session, turn, continuation)
         proof = continuation.get("native_observation") or {}
         if (self.controller.store.queued_turns(session["session_id"])
                 or completed_work_readback(native, str(proof.get("turn_id") or ""),

loopx/chat_runtime.py

@@ -862,6 +862,12 @@ def check_route_configuration():
         with self.lock:
             current = self.adapters.get(session_id)
             manager_profile_changed = NativeHistoryRestoration.manager_profile_changed(session, manager_runtime)
+            if recovery.get("work_original_turn_id") and (
+                    not isinstance(current, CodexAppServerAdapter)
+                    or current.upstream_thread_id != session.get("upstream_thread_id")
+                    or not current.answer_review_replaceable()):
+                raise CodexChatAgentError("The reviewed work no longer has an idle native owner; continuation held.",
+                    error_code="host_gate", gate=None)
             if (
                 current is not None
                 and current.healthcheck()