From 2d0906d9a73ed03f77b4b940b6b76f7290fdfe1c Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 18:55:12 +0800
Subject: [PATCH 1/2] fix(chat): qualify unfinished work after a delivered
 result review

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    |  14 ++-
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  11 +-
 loopx/canary/module_metric_baseline.json      |   2 +-
 .../native_chat/external_conversations.py     |  79 +++++++++++--
 loopx/chat_runtime.py                         |   6 +
 .../collaboration/conversation_binding.ts     |  60 +++++++---
 tests/test_lark_private_conversations.py      | 106 ++++++++++++++++--
 7 files changed, 240 insertions(+), 38 deletions(-)

diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index 927c48b270..80bbe565cd 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/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
 
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index 47ea747e5e..e8d08cc2d9 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/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 后端的空闲恢复观测
 
diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json
index cb12b2d3d3..37089fbaa5 100644
--- a/loopx/canary/module_metric_baseline.json
+++ b/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,
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index 8076eef6bb..3b1152f9cc 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/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 ""),
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index bb4724f066..dcc9584dca 100644
--- a/loopx/chat_runtime.py
+++ b/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()
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index a3abecb673..eaf8104f54 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -573,6 +573,10 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && originalResult.turn_id !== row.turn_id && originalResponse != null
       && unfinishedProjectWork(originalResponse) !== null
       && isDeepStrictEqual(originalResponse.unfinished_work, reservation.unfinished_work);
+    const continuationOwed = reservation?.work_attempt === 1
+      && turn?.client_turn_id === reservation.client_turn_id && reservation.failed_turn_id !== row.turn_id
+      && (turn?.status !== "completed" || response === null || response.gate != null
+        || response.unfinished_work != null || row.delivery_verified !== true);
     const recovered = row.native_result_recovery == null ? null
       : requireJsonObject(row.native_result_recovery, "recovered native result");
     const native = recovered?.native_observation as JsonObject | undefined;
@@ -588,7 +592,8 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && extendsOriginalText(response, result);
     return {pending: !returned && row.command === null && row.status === "accepted"
       && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
-      && (expandedResult || metadataOwed || answerReviewOwed || retainedWork || unfinished || turn?.status === "failed"
+      && (expandedResult || metadataOwed || answerReviewOwed || retainedWork || continuationOwed || unfinished
+        || turn?.status === "failed"
         && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure", "context_window_exceeded",
           "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error",
           "manager_authorization_unavailable", "runtime_error", "server_restarted"].includes(String(turn.error_code)))
@@ -814,15 +819,36 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
         && isDeepStrictEqual(adopted.native_observation, proof))};
   }
   const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "reserved context recovery");
+  const recoveryHistory = Array.isArray(row.context_recovery_history)
+    ? row.context_recovery_history.map(value => requireJsonObject(value, "prior recovery")) : [];
   // One completed-work continuation is terminal for this request, including
   // when it later fails. It cannot fall through to routing/quota retries.
   if ((previous?.work_attempt != null && previous.failed_turn_id !== row.turn_id)
-      || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.some(value =>
-        requireJsonObject(value, "prior recovery").work_attempt != null))) return no;
-  // A failed review cannot fall through to routing/quota business retries.
+      || recoveryHistory.some(item => item.work_attempt != null)) return no;
+  if (!steward && recovery.reviewed_original_turn != null) {
+    const review = previous?.work_attempt === 1 ? recoveryHistory.at(-1) : previous;
+    const original = requireJsonObject(recovery.reviewed_original_turn, "reviewed original work");
+    const reviewProof = review?.native_observation as JsonObject | undefined;
+    if (review?.read_only_review !== true || review.prior_result_delivery_verified !== true
+        || review.failed_turn_id !== original.turn_id || !isDeepStrictEqual(review.original_turn, original)
+        || !isDeepStrictEqual(review.binding_snapshot, selected)
+        || reviewProof?.completed_result_review !== true || reviewProof.read_only_review !== true
+        || reviewProof.thread_id !== current.upstream_thread_id || reviewProof.turn_id !== original.upstream_turn_id
+        || !isDeepStrictEqual(review.unfinished_work, reviewProof.unfinished_work)
+        || !isDeepStrictEqual(review.unfinished_work, (original.response as JsonObject | null)?.unfinished_work)
+        || original.status !== "completed" || original.client_turn_id !== `external-${row.request_ref}`
+        || original.message !== row.message || original.origin !== row.origin
+        || failed.status !== "completed" || failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
+        || failed.client_turn_id !== review.client_turn_id
+        || failed.client_turn_id !== `external-${row.request_ref}-recovery-${review.version}-review-1`
+        || failed.message !== row.message || failed.origin !== row.origin
+        || !pureResultText(failed.response as JsonObject | null)) return no;
+    // The review is a delivery predecessor, not proof of writable effects.
+    return planUnfinishedWork(recovery, row, current, original, previous, version, failed);
+  }
+  // Other completed or failed reviews cannot fall through to business retries.
   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;
+      || recoveryHistory.some(item => item.read_only_review === true)) return no;
   if (failed.error_code === "server_restarted") {
     return planResultReview(recovery, row, current, failed, previous, version);
   }
@@ -1129,8 +1155,9 @@ function unfinishedProjectWork(result: JsonObject): JsonObject | null {
 }
 
 function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: JsonObject,
-  completed: JsonObject, previous: JsonObject | null, version: number): JsonObject {
+  completed: JsonObject, previous: JsonObject | null, version: number, deliveredReview?: JsonObject): JsonObject {
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
+  const deliveryHead = deliveredReview ?? completed;
   const response = completed.response;
   if (!response || typeof response !== "object" || Array.isArray(response)) return no;
   const work = unfinishedProjectWork(response as JsonObject);
@@ -1138,7 +1165,7 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   const matchesWork = (value: unknown) => value !== null && typeof value === "object" && !Array.isArray(value)
     && Object.keys(value).sort().join(",") === "dependency,effects_settled,remaining"
     && ["remaining", "dependency", "effects_settled"].every(key => (value as JsonObject)[key] === work[key]);
-  if (completed.session_id !== current.session_id || completed.turn_id !== row.turn_id
+  if (completed.session_id !== current.session_id || deliveryHead.turn_id !== row.turn_id
       || recovery.steering_observed !== false || recovery.stop_observed !== false
       || recovery.attachments_present !== false
       || typeof completed.upstream_turn_id !== "string" || !completed.upstream_turn_id) return no;
@@ -1146,20 +1173,23 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   // An admitted continuation may already have appended the original input.
   // Recover its existing receipt before considering a fresh result review.
   const existingWorkTurn = previous?.work_attempt === 1 && recovery.admitted_turn != null;
-  if (previous?.read_only_review === true || (!existingWorkTurn
-      && (recovery.later_equivalent_input_present === true || observedResult?.completed_result_review === true))) {
+  if (!deliveredReview && (previous?.read_only_review === true || (!existingWorkTurn
+      && (recovery.later_equivalent_input_present === true || observedResult?.completed_result_review === true)))) {
     return planCompletedResultReview(recovery, row, current, completed, previous, version, work);
   }
   const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-work-1`;
+  const originalTurnId = deliveredReview ? completed.turn_id : undefined;
   if (previous?.work_attempt != null && (previous.work_attempt !== 1
-      || previous.failed_turn_id !== completed.turn_id || previous.version !== version
-      || previous.client_turn_id !== client || !matchesWork(previous.unfinished_work))) return no;
+      || previous.failed_turn_id !== deliveryHead.turn_id || previous.version !== version
+      || previous.client_turn_id !== client || previous.work_original_turn_id !== originalTurnId
+      || !matchesWork(previous.unfinished_work))) return no;
   const candidate = {...no, recovery_candidate: true, work_attempt: 1, unfinished_work: work,
-    client_turn_id: client, session_id: current.session_id, failed_turn_id: completed.turn_id};
+    client_turn_id: client, session_id: current.session_id, failed_turn_id: deliveryHead.turn_id,
+    ...(deliveredReview ? {work_original_turn_id: originalTurnId} : {})};
   if (previous?.work_attempt === 1 && recovery.admitted_turn != null) {
     const admitted = requireJsonObject(recovery.admitted_turn, "admitted unfinished work");
     if (admitted.client_turn_id !== client || admitted.session_id !== current.session_id
-        || admitted.turn_id === completed.turn_id || admitted.message !== row.message
+        || admitted.turn_id === deliveryHead.turn_id || admitted.message !== row.message
         || admitted.origin !== row.origin || !Array.isArray(admitted.attachments ?? [])
         || (admitted.attachments as unknown[] | undefined)?.length) return no;
     return {...candidate, recovery_eligible: recovery.source_verified === true,
@@ -1168,7 +1198,7 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   }
   if (current.active_turn_id != null || current.status !== "ready"
       || recovery.queued_input_present !== false || recovery.later_equivalent_input_present !== false) return no;
-  const now = Date.parse(String(recovery.observed_at)), ended = Date.parse(String(completed.completed_at));
+  const now = Date.parse(String(recovery.observed_at)), ended = Date.parse(String(deliveryHead.completed_at));
   if (!Number.isFinite(now) || !Number.isFinite(ended) || now < ended + 60_000) return no;
   const native = sessionIdentity(current.upstream_thread_id, "original native thread");
   const planned = {...candidate, native_thread_ids: [native], not_before: new Date(ended + 60_000).toISOString()};
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 312f0b183c..7df0a3b677 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -2027,7 +2027,10 @@ def counted_revision():
 @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native",
     "background", "later-user", "completed-later", "equivalent-later", "later-failed-effect",
     "settled-command", "later-settled-command", "unsettled-command",
-    "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input", "private-path", "still-incomplete", "cold-native", "dead-native", "foreign-native", "failed-effect"])
+    "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input", "private-path", "still-incomplete",
+    "cold-native", "dead-native", "foreign-native", "failed-effect", "reviewed-work", "reviewed-enqueue-crash",
+    "reviewed-exit-130", "reviewed-stop", "reviewed-revoked", "reviewed-no-owner", "reviewed-equivalent",
+    "reviewed-owner-lost-after-admission"])
 def test_original_worker_continues_explicit_unfinished_project_work_once_after_restart(ordinary, monkeypatch, fault):  # noqa: F811
     """Reply-worker restart retains native ownership; native loss stays fenced."""
     from datetime import datetime, timedelta, timezone
@@ -2135,7 +2138,13 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
         if fault == "unsettled-command":
             snapshot["items"][1]["item"]["exitCode"] = None
         inventory.write_text(json.dumps(snapshot))
-    if fault == "cold-native":
+    review_first = bool(fault and fault.startswith("reviewed-"))
+    if fault == "reviewed-exit-130":
+        snapshot = json.loads(inventory.read_text())
+        snapshot["items"][1]["item"].update(type="commandExecution", status="failed", exitCode=130,
+            startedAtMs=1000, completedAtMs=1001, aggregatedOutput="Interrupted.")
+        inventory.write_text(json.dumps(snapshot))
+    if fault == "cold-native" or review_first:
         runtime.adapters.pop(original["session_id"]).close_session()
     elif fault == "dead-native":
         runtime.adapters[original["session_id"]].close_session()
@@ -2198,7 +2207,8 @@ def crash(**kwargs):
     artifact_before_recovery = artifact.read_bytes()
     selected.reconcile()
     continued = selected.core.read_request(ref)
-    reviewed = fault in {"cold-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"}
+    reviewed = review_first or fault in {"cold-native", "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"}
+    continued_after_review = fault in {"reviewed-work", "reviewed-enqueue-crash"}
     if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "dead-native", "foreign-native"}:
         assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery")
         assert artifact.read_bytes() == artifact_before_recovery
@@ -2219,9 +2229,11 @@ def crash(**kwargs):
         returned = deliveries[presentation_phase(continued, "terminal")]
         assert returned["verified"] and "The remaining work is still pending." in returned["text"]
         writes = list(provider.writes)
-        assert not worker().core.context_recovery_candidate(ref)
-        worker().reconcile()
-        assert provider.writes == writes
+        if not review_first:
+            # Finish the existing review assertions before its continuation backoff elapses.
+            clock = datetime.fromisoformat(review["completed_at"]) + timedelta(seconds=59)
+            worker().reconcile()
+            assert provider.writes == writes
 
         recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()][len(before_recovery_calls):]
         resume, = [row["params"] for row in recovery_calls if row.get("method") == "thread/resume"]
@@ -2232,6 +2244,82 @@ def crash(**kwargs):
         assert permissions["filesystem"][":workspace_roots"] == {".": "read"}
         assert permissions["network"]["enabled"] is False
         assert sum(row.get("method") == "turn/start" for row in recovery_calls) == 1
+        if review_first:
+            review_bytes = store._turn_path(review["session_id"], review["turn_id"]).read_bytes()
+            if fault == "reviewed-stop":
+                transport.admit("notes-app", provider.event("notes-app", "stop-after-review", "/stop"))
+            elif fault == "reviewed-revoked":
+                transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+            elif fault == "reviewed-no-owner":
+                runtime.adapters.pop(review["session_id"]).close_session()
+            elif fault == "reviewed-equivalent":
+                repeated = provider.event("notes-app", "repeat-after-review", event["content"])
+                transport.admit("notes-app", repeated)
+                same = selected.core.read_request(identity_ref(binding["provider_ref"], repeated["message_id"]))
+                runtime.wait_for_turn(session_id=same["session_id"], turn_id=same["turn_id"], timeout_sec=10)
+                selected.reconcile_request(transport.root / (same["request_ref"] + ".json"))
+            clock = datetime.now(timezone.utc) + timedelta(seconds=61)
+            if fault == "reviewed-enqueue-crash":
+                enqueue = runtime.enqueue_turn
+                def crash_after_enqueue(**kwargs):
+                    enqueue(**kwargs)
+                    raise OSError("synthetic post-review admission crash")
+                monkeypatch.setattr(runtime, "enqueue_turn", crash_after_enqueue)
+                with pytest.raises(OSError, match="post-review admission crash"):
+                    worker().reconcile()
+                monkeypatch.setattr(runtime, "enqueue_turn", enqueue)
+                assert selected.core.read_request(ref)["turn_id"] == review["turn_id"]
+
+            artifact_before = artifact.read_bytes()
+            owner_lost = fault == "reviewed-owner-lost-after-admission"
+            if owner_lost:
+                resume = runtime.resume_session_queue
+                monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None)
+                worker().reconcile()
+                queued = selected.core.read_request(ref)
+                assert queued["turn_id"] != review["turn_id"]
+                runtime.adapters.pop(review["session_id"]).close_session()
+                monkeypatch.setattr(runtime, "resume_session_queue", resume)
+                resume(session_id=queued["session_id"], work_dir=workspace, objective="")
+            worker().reconcile()
+            final = selected.core.read_request(ref)
+            if owner_lost:
+                held = runtime.wait_for_turn(session_id=final["session_id"], turn_id=final["turn_id"], timeout_sec=10)
+                assert held["status"] == "failed" and held["error_code"] == "host_gate", held
+                selected.reconcile_request(path)
+                assert artifact.read_bytes() == artifact_before
+                assert worker().core.quota_backlog_pending(selected.core.read_request(ref))
+                assert not worker().core.context_recovery_candidate(ref)
+            elif continued_after_review:
+                assert final["session_id"] == original["session_id"] and final["turn_id"] != review["turn_id"]
+                assert final["context_recovery"]["work_original_turn_id"] == original["turn_id"]
+                assert final["context_recovery"]["failed_turn_id"] == review["turn_id"]
+                assert final["context_recovery_history"] == [recovery]
+                result = runtime.wait_for_turn(session_id=final["session_id"], turn_id=final["turn_id"], timeout_sec=10)
+                assert result["status"] == "completed", result
+                selected.reconcile_request(path)
+                result_deliveries = json.loads(path.read_text())["deliveries"]
+                assert result_deliveries["terminal"] == original_delivery
+                assert result_deliveries[presentation_phase(continued, "terminal")] == returned
+                reply = result_deliveries[presentation_phase(final, "terminal")]
+                assert reply["verified"] and "Verified remaining work completed." in reply["text"]
+                assert artifact.read_bytes() == b"first\nremaining\n"
+                assert not worker().core.quota_backlog_pending(selected.core.read_request(ref))
+                assert not worker().core.context_recovery_candidate(ref)
+                assert store.load_session(final["session_id"])["project_context"] == session_before["project_context"]
+                assert store.load_session(final["session_id"])["upstream_thread_id"] == session_before["upstream_thread_id"]
+            else:
+                assert final["turn_id"] == review["turn_id"]
+                assert final["context_recovery"] == recovery
+                assert artifact.read_bytes() == artifact_before
+            assert store._turn_path(review["session_id"], review["turn_id"]).read_bytes() == review_bytes
+            starts = [json.loads(line)["params"] for line in capture.read_text().splitlines() if json.loads(line).get("method") == "turn/start"]
+            assert len(starts) == (3 if continued_after_review or fault == "reviewed-equivalent" else 2)
+            if continued_after_review:
+                assert "Do only the remaining authorized work" in starts[-1]["input"][0]["text"]
+            writes = list(provider.writes)
+            worker().reconcile()
+            assert provider.writes == writes
     else:
         assert continued["turn_id"] != original["turn_id"]
         assert continued["session_id"] == original["session_id"]
@@ -2269,8 +2357,10 @@ def crash(**kwargs):
             assert "Verified remaining work completed." in bodies
     assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_turn_bytes
     after_recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()]
-    expected_new = {"initialize": int(reviewed) + int(fault == "cold-native"),
-        "thread/start": 0, "thread/resume": int(reviewed)}
+    # Stop also resumes the native thread to check for background work.
+    extra_resume = continued_after_review or fault in {"reviewed-equivalent", "reviewed-stop"}
+    expected_new = {"initialize": int(reviewed) + int(fault == "cold-native" or review_first) + int(extra_resume),
+        "thread/start": 0, "thread/resume": int(reviewed) + int(extra_resume)}
     for method, count in expected_new.items():
         assert sum(row.get("method") == method for row in after_recovery_calls[len(before_recovery_calls):]) == count
 

From 80d2b4647022ef866ffdf76fa2c779159f513f2c Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 8 Nov 2026 14:01:07 +0800
Subject: [PATCH 2/2] docs(chat): align result-review continuation guidance

The executable behavior is already on main through #8329 and #8331. Keep the stronger retained-result-review path and avoid merging the superseded duplicate state machine.

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/canary/module_metric_baseline.json      |   2 +-
 .../native_chat/external_conversations.py     |  84 ++-----------
 loopx/chat_runtime.py                         |   6 -
 .../collaboration/conversation_binding.ts     |  60 +++-------
 tests/test_lark_private_conversations.py      | 110 ++----------------
 5 files changed, 35 insertions(+), 227 deletions(-)

diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json
index 37089fbaa5..cb12b2d3d3 100644
--- a/loopx/canary/module_metric_baseline.json
+++ b/loopx/canary/module_metric_baseline.json
@@ -39,7 +39,7 @@
     "loopx/chat_runtime.py": {
       "any_count": 38,
       "dict_any_count": 0,
-      "lines": 2481
+      "lines": 2475
     },
     "loopx/chat_server.py": {
       "any_count": 25,
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index 55c82e780a..1748c9c3b2 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -96,21 +96,6 @@ 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 _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 {}
@@ -167,13 +152,6 @@ 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)
-        review_turn = store.load_turn(session["session_id"], row["turn_id"])
-        if review and review_turn and review["client_turn_id"] == review_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)
         unfinished = turn.get("status") == "completed" and bool((turn.get("response") or {}).get("unfinished_work"))
         live_work_owner = False
         if unfinished:
@@ -202,15 +180,14 @@ 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 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"]))
-            internal_turns = {turn["turn_id"], work_turn["turn_id"]}
-            if retained_review is not None:
-                internal_turns.add(retained_review["turn"]["turn_id"])
-            later_equivalent = self._has_later_equivalent_input(row, work_turn, messages,
-                internal_turns=internal_turns)
+        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"])
+                             if discovery_reads is not None else store.messages(session["session_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"},
@@ -234,7 +211,6 @@ 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"]
@@ -399,9 +375,7 @@ 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
-            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")):
+            if turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work"):
                 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"
@@ -489,16 +463,11 @@ 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"])
-                    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):
+                    if not result_review 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=work_turn["upstream_turn_id"],
+                            thread_ids=plan["native_thread_ids"], turn_id=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()):
@@ -613,8 +582,6 @@ 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"]
                         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"])
@@ -1229,33 +1196,6 @@ 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."""
@@ -1281,8 +1221,6 @@ 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 continuation.get("completed_review_turn_id"):
             row = self.read_request(continuation["request_ref"])
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index dcc9584dca..bb4724f066 100644
--- a/loopx/chat_runtime.py
+++ b/loopx/chat_runtime.py
@@ -862,12 +862,6 @@ 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()
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 91305e9ee2..0a58b28e70 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -575,10 +575,6 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && originalResult.turn_id !== row.turn_id && originalResponse != null
       && unfinishedProjectWork(originalResponse) !== null
       && isDeepStrictEqual(originalResponse.unfinished_work, reservation.unfinished_work);
-    const continuationOwed = reservation?.work_attempt === 1
-      && turn?.client_turn_id === reservation.client_turn_id && reservation.failed_turn_id !== row.turn_id
-      && (turn?.status !== "completed" || response === null || response.gate != null
-        || response.unfinished_work != null || row.delivery_verified !== true);
     const recovered = row.native_result_recovery == null ? null
       : requireJsonObject(row.native_result_recovery, "recovered native result");
     const native = recovered?.native_observation as JsonObject | undefined;
@@ -594,8 +590,7 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && extendsOriginalText(response, result);
     return {pending: !returned && row.command === null && row.status === "accepted"
       && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
-      && (expandedResult || metadataOwed || answerReviewOwed || retainedWork || continuationOwed || unfinished
-        || turn?.status === "failed"
+      && (expandedResult || metadataOwed || answerReviewOwed || retainedWork || unfinished || turn?.status === "failed"
         && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure", "context_window_exceeded",
           "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error",
           "manager_authorization_unavailable", "runtime_error", "server_restarted"].includes(String(turn.error_code)))
@@ -824,36 +819,15 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
         && isDeepStrictEqual(adopted.native_observation, proof))};
   }
   const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "reserved context recovery");
-  const recoveryHistory = Array.isArray(row.context_recovery_history)
-    ? row.context_recovery_history.map(value => requireJsonObject(value, "prior recovery")) : [];
   // One completed-work continuation is terminal for this request, including
   // when it later fails. It cannot fall through to routing/quota retries.
   if ((previous?.work_attempt != null && previous.failed_turn_id !== row.turn_id)
-      || recoveryHistory.some(item => item.work_attempt != null)) return no;
-  if (!steward && recovery.reviewed_original_turn != null) {
-    const review = previous?.work_attempt === 1 ? recoveryHistory.at(-1) : previous;
-    const original = requireJsonObject(recovery.reviewed_original_turn, "reviewed original work");
-    const reviewProof = review?.native_observation as JsonObject | undefined;
-    if (review?.read_only_review !== true || review.prior_result_delivery_verified !== true
-        || review.failed_turn_id !== original.turn_id || !isDeepStrictEqual(review.original_turn, original)
-        || !isDeepStrictEqual(review.binding_snapshot, selected)
-        || reviewProof?.completed_result_review !== true || reviewProof.read_only_review !== true
-        || reviewProof.thread_id !== current.upstream_thread_id || reviewProof.turn_id !== original.upstream_turn_id
-        || !isDeepStrictEqual(review.unfinished_work, reviewProof.unfinished_work)
-        || !isDeepStrictEqual(review.unfinished_work, (original.response as JsonObject | null)?.unfinished_work)
-        || original.status !== "completed" || original.client_turn_id !== `external-${row.request_ref}`
-        || original.message !== row.message || original.origin !== row.origin
-        || failed.status !== "completed" || failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
-        || failed.client_turn_id !== review.client_turn_id
-        || failed.client_turn_id !== `external-${row.request_ref}-recovery-${review.version}-review-1`
-        || failed.message !== row.message || failed.origin !== row.origin
-        || !pureResultText(failed.response as JsonObject | null)) return no;
-    // The review is a delivery predecessor, not proof of writable effects.
-    return planUnfinishedWork(recovery, row, current, original, previous, version, failed);
-  }
-  // Other completed or failed reviews cannot fall through to business retries.
+      || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.some(value =>
+        requireJsonObject(value, "prior recovery").work_attempt != null))) return no;
+  // A failed review cannot fall through to routing/quota business retries.
   if ((previous?.read_only_review === true && previous.failed_turn_id !== row.turn_id)
-      || recoveryHistory.some(item => item.read_only_review === true)) return no;
+      || (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);
   }
@@ -1200,9 +1174,8 @@ function planWorkAfterResultReview(recovery: JsonObject, row: JsonObject, curren
 }
 
 function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: JsonObject,
-  completed: JsonObject, previous: JsonObject | null, version: number, deliveredReview?: JsonObject): JsonObject {
+  completed: JsonObject, previous: JsonObject | null, version: number): JsonObject {
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
-  const deliveryHead = deliveredReview ?? completed;
   const response = completed.response;
   if (!response || typeof response !== "object" || Array.isArray(response)) return no;
   const work = unfinishedProjectWork(response as JsonObject);
@@ -1210,7 +1183,7 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   const matchesWork = (value: unknown) => value !== null && typeof value === "object" && !Array.isArray(value)
     && Object.keys(value).sort().join(",") === "dependency,effects_settled,remaining"
     && ["remaining", "dependency", "effects_settled"].every(key => (value as JsonObject)[key] === work[key]);
-  if (completed.session_id !== current.session_id || deliveryHead.turn_id !== row.turn_id
+  if (completed.session_id !== current.session_id || completed.turn_id !== row.turn_id
       || recovery.steering_observed !== false || recovery.stop_observed !== false
       || recovery.attachments_present !== false
       || typeof completed.upstream_turn_id !== "string" || !completed.upstream_turn_id) return no;
@@ -1218,23 +1191,20 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   // An admitted continuation may already have appended the original input.
   // Recover its existing receipt before considering a fresh result review.
   const existingWorkTurn = previous?.work_attempt === 1 && recovery.admitted_turn != null;
-  if (!deliveredReview && (previous?.read_only_review === true || (!existingWorkTurn
-      && (recovery.later_equivalent_input_present === true || observedResult?.completed_result_review === true)))) {
+  if (previous?.read_only_review === true || (!existingWorkTurn
+      && (recovery.later_equivalent_input_present === true || observedResult?.completed_result_review === true))) {
     return planCompletedResultReview(recovery, row, current, completed, previous, version, work);
   }
   const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-work-1`;
-  const originalTurnId = deliveredReview ? completed.turn_id : undefined;
   if (previous?.work_attempt != null && (previous.work_attempt !== 1
-      || previous.failed_turn_id !== deliveryHead.turn_id || previous.version !== version
-      || previous.client_turn_id !== client || previous.work_original_turn_id !== originalTurnId
-      || !matchesWork(previous.unfinished_work))) return no;
+      || previous.failed_turn_id !== completed.turn_id || previous.version !== version
+      || previous.client_turn_id !== client || !matchesWork(previous.unfinished_work))) return no;
   const candidate = {...no, recovery_candidate: true, work_attempt: 1, unfinished_work: work,
-    client_turn_id: client, session_id: current.session_id, failed_turn_id: deliveryHead.turn_id,
-    ...(deliveredReview ? {work_original_turn_id: originalTurnId} : {})};
+    client_turn_id: client, session_id: current.session_id, failed_turn_id: completed.turn_id};
   if (previous?.work_attempt === 1 && recovery.admitted_turn != null) {
     const admitted = requireJsonObject(recovery.admitted_turn, "admitted unfinished work");
     if (admitted.client_turn_id !== client || admitted.session_id !== current.session_id
-        || admitted.turn_id === deliveryHead.turn_id || admitted.message !== row.message
+        || admitted.turn_id === completed.turn_id || admitted.message !== row.message
         || admitted.origin !== row.origin || !Array.isArray(admitted.attachments ?? [])
         || (admitted.attachments as unknown[] | undefined)?.length) return no;
     return {...candidate, recovery_eligible: recovery.source_verified === true,
@@ -1243,7 +1213,7 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   }
   if (current.active_turn_id != null || current.status !== "ready"
       || recovery.queued_input_present !== false || recovery.later_equivalent_input_present !== false) return no;
-  const now = Date.parse(String(recovery.observed_at)), ended = Date.parse(String(deliveryHead.completed_at));
+  const now = Date.parse(String(recovery.observed_at)), ended = Date.parse(String(completed.completed_at));
   if (!Number.isFinite(now) || !Number.isFinite(ended) || now < ended + 60_000) return no;
   const native = sessionIdentity(current.upstream_thread_id, "original native thread");
   const planned = {...candidate, native_thread_ids: [native], not_before: new Date(ended + 60_000).toISOString()};
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 013fcb52a0..949cfff87f 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -2027,10 +2027,7 @@ def counted_revision():
 @pytest.mark.parametrize("fault", [None, "changed-source", "revoked", "stop", "unknown-native",
     "background", "later-user", "completed-later", "equivalent-later", "later-failed-effect",
     "settled-command", "later-settled-command", "unsettled-command",
-    "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input", "private-path", "still-incomplete",
-    "cold-native", "dead-native", "foreign-native", "failed-effect", "reviewed-work", "reviewed-enqueue-crash",
-    "reviewed-exit-130", "reviewed-stop", "reviewed-revoked", "reviewed-no-owner", "reviewed-equivalent",
-    "reviewed-owner-lost-after-admission",
+    "reservation-crash", "post-enqueue-crash", "queued-newer", "queued-items", "queued-input", "private-path", "still-incomplete", "cold-native", "dead-native", "foreign-native", "failed-effect",
     "review-live", "review-live-source", "review-live-stop", "review-live-revoked", "review-live-equivalent", "review-live-unknown",
     "review-live-reservation-crash", "review-live-enqueue-crash", "transient-observation"])
 def test_original_worker_continues_explicit_unfinished_project_work_once_after_restart(ordinary, monkeypatch, fault):  # noqa: F811
@@ -2140,13 +2137,7 @@ def test_original_worker_continues_explicit_unfinished_project_work_once_after_r
         if fault == "unsettled-command":
             snapshot["items"][1]["item"]["exitCode"] = None
         inventory.write_text(json.dumps(snapshot))
-    review_first = bool(fault and fault.startswith("reviewed-"))
-    if fault == "reviewed-exit-130":
-        snapshot = json.loads(inventory.read_text())
-        snapshot["items"][1]["item"].update(type="commandExecution", status="failed", exitCode=130,
-            startedAtMs=1000, completedAtMs=1001, aggregatedOutput="Interrupted.")
-        inventory.write_text(json.dumps(snapshot))
-    if fault == "cold-native" or review_first or str(fault).startswith("review-live"):
+    if fault == "cold-native" or str(fault).startswith("review-live"):
         runtime.adapters.pop(original["session_id"]).close_session()
     elif fault == "dead-native":
         runtime.adapters[original["session_id"]].close_session()
@@ -2238,8 +2229,7 @@ def unavailable_once(**kwargs):
         clock += timedelta(seconds=61)
     selected.reconcile()
     continued = selected.core.read_request(ref)
-    reviewed = review_first or fault == "cold-native" or str(fault).startswith("review-live")
-    continued_after_review = fault in {"reviewed-work", "reviewed-enqueue-crash"}
+    reviewed = fault == "cold-native" or str(fault).startswith("review-live")
     if fault in {"changed-source", "revoked", "stop", "unknown-native", "background", "later-user", "dead-native", "foreign-native",
                  "failed-effect", "equivalent-later", "later-failed-effect", "unsettled-command"}:
         assert continued["turn_id"] == original["turn_id"] and not continued.get("context_recovery")
@@ -2263,11 +2253,9 @@ def unavailable_once(**kwargs):
         returned = deliveries[presentation_phase(continued, "terminal")]
         assert returned["verified"] and "The remaining work is still pending." in returned["text"]
         writes = list(provider.writes)
-        if not review_first:
-            # Finish the existing review assertions before its continuation backoff elapses.
-            clock = datetime.fromisoformat(review["completed_at"]) + timedelta(seconds=59)
-            worker().reconcile()
-            assert provider.writes == writes
+        assert not worker().core.context_recovery_candidate(ref)
+        worker().reconcile()
+        assert provider.writes == writes
 
         recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()][len(before_recovery_calls):]
         resume, = [row["params"] for row in recovery_calls if row.get("method") == "thread/resume"]
@@ -2278,83 +2266,6 @@ def unavailable_once(**kwargs):
         assert permissions["filesystem"][":workspace_roots"] == {".": "read"}
         assert permissions["network"]["enabled"] is False
         assert sum(row.get("method") == "turn/start" for row in recovery_calls) == 1
-        if review_first:
-            review_bytes = store._turn_path(review["session_id"], review["turn_id"]).read_bytes()
-            if fault == "reviewed-stop":
-                transport.admit("notes-app", provider.event("notes-app", "stop-after-review", "/stop"))
-            elif fault == "reviewed-revoked":
-                transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
-            elif fault == "reviewed-no-owner":
-                runtime.adapters.pop(review["session_id"]).close_session()
-            elif fault == "reviewed-equivalent":
-                repeated = provider.event("notes-app", "repeat-after-review", event["content"])
-                transport.admit("notes-app", repeated)
-                same = selected.core.read_request(identity_ref(binding["provider_ref"], repeated["message_id"]))
-                runtime.wait_for_turn(session_id=same["session_id"], turn_id=same["turn_id"], timeout_sec=10)
-                selected.reconcile_request(transport.root / (same["request_ref"] + ".json"))
-            clock = datetime.now(timezone.utc) + timedelta(seconds=61)
-            if fault == "reviewed-enqueue-crash":
-                enqueue = runtime.enqueue_turn
-                def crash_after_enqueue(**kwargs):
-                    enqueue(**kwargs)
-                    raise OSError("synthetic post-review admission crash")
-                monkeypatch.setattr(runtime, "enqueue_turn", crash_after_enqueue)
-                with pytest.raises(OSError, match="post-review admission crash"):
-                    worker().reconcile()
-                monkeypatch.setattr(runtime, "enqueue_turn", enqueue)
-                assert selected.core.read_request(ref)["turn_id"] == review["turn_id"]
-
-            artifact_before = artifact.read_bytes()
-            owner_lost = fault == "reviewed-owner-lost-after-admission"
-            if owner_lost:
-                resume = runtime.resume_session_queue
-                monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None)
-                worker().reconcile()
-                queued = selected.core.read_request(ref)
-                assert queued["turn_id"] != review["turn_id"]
-                runtime.adapters.pop(review["session_id"]).close_session()
-                monkeypatch.setattr(runtime, "resume_session_queue", resume)
-                resume(session_id=queued["session_id"], work_dir=workspace, objective="")
-            worker().reconcile()
-            final = selected.core.read_request(ref)
-            if owner_lost:
-                held = runtime.wait_for_turn(session_id=final["session_id"], turn_id=final["turn_id"], timeout_sec=10)
-                assert held["status"] == "failed" and held["error_code"] == "host_gate", held
-                selected.reconcile_request(path)
-                assert artifact.read_bytes() == artifact_before
-                assert worker().core.quota_backlog_pending(selected.core.read_request(ref))
-                assert not worker().core.context_recovery_candidate(ref)
-            elif continued_after_review:
-                assert final["session_id"] == original["session_id"] and final["turn_id"] != review["turn_id"]
-                assert final["context_recovery"]["work_original_turn_id"] == original["turn_id"]
-                assert final["context_recovery"]["failed_turn_id"] == review["turn_id"]
-                assert final["context_recovery_history"] == [recovery]
-                result = runtime.wait_for_turn(session_id=final["session_id"], turn_id=final["turn_id"], timeout_sec=10)
-                assert result["status"] == "completed", result
-                selected.reconcile_request(path)
-                result_deliveries = json.loads(path.read_text())["deliveries"]
-                assert result_deliveries["terminal"] == original_delivery
-                assert result_deliveries[presentation_phase(continued, "terminal")] == returned
-                reply = result_deliveries[presentation_phase(final, "terminal")]
-                assert reply["verified"] and "Verified remaining work completed." in reply["text"]
-                assert artifact.read_bytes() == b"first\nremaining\n"
-                assert not worker().core.quota_backlog_pending(selected.core.read_request(ref))
-                assert not worker().core.context_recovery_candidate(ref)
-                assert store.load_session(final["session_id"])["project_context"] == session_before["project_context"]
-                assert store.load_session(final["session_id"])["upstream_thread_id"] == session_before["upstream_thread_id"]
-            else:
-                assert final["turn_id"] == review["turn_id"]
-                assert final["context_recovery"] == recovery
-                assert artifact.read_bytes() == artifact_before
-            assert store._turn_path(review["session_id"], review["turn_id"]).read_bytes() == review_bytes
-            starts = [json.loads(line)["params"] for line in capture.read_text().splitlines() if json.loads(line).get("method") == "turn/start"]
-            assert len(starts) == (3 if continued_after_review or fault == "reviewed-equivalent" else 2)
-            if continued_after_review:
-                assert "Do only the remaining authorized work" in starts[-1]["input"][0]["text"]
-            writes = list(provider.writes)
-            worker().reconcile()
-            assert provider.writes == writes
-
         if str(fault).startswith("review-live"):
             # An ordinary source request naturally restores the writable owner.
             # Merely opening a cold reader must never manufacture effects proof.
@@ -2442,14 +2353,9 @@ def crash_after_review(**kwargs):
             assert "Verified remaining work completed." in bodies
     assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == original_turn_bytes
     after_recovery_calls = [json.loads(line) for line in capture.read_text().splitlines()]
-    # Stop also resumes the native thread to check for background work.
-    extra_resume = continued_after_review or fault in {"reviewed-equivalent", "reviewed-stop"}
     restored_owner = int(str(fault).startswith("review-live"))
-    expected_new = {
-        "initialize": int(reviewed) + int(fault == "cold-native" or review_first) + int(extra_resume) + 2 * restored_owner,
-        "thread/start": 0,
-        "thread/resume": int(reviewed) + int(extra_resume) + restored_owner,
-    }
+    expected_new = {"initialize": int(reviewed) + int(fault == "cold-native") + 2 * restored_owner,
+        "thread/start": 0, "thread/resume": int(reviewed) + restored_owner}
     for method, count in expected_new.items():
         assert sum(row.get("method") == method for row in after_recovery_calls[len(before_recovery_calls):]) == count