From e55d87b3b53f9a3d5a7d0062bd52cc4697c473f6 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 8 Nov 2026 14:27:43 +0800
Subject: [PATCH 1/3] fix(chat): preserve result evidence across host
 preparation failures

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    | 11 +++
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  8 ++
 .../native_chat/external_conversations.py     | 78 ++++++++-----------
 .../native_chat/result_review_evidence.py     | 55 +++++++++++++
 .../collaboration/conversation_binding.ts     | 39 ++++++++--
 .../conversation_binding.test.ts              | 33 ++++++++
 tests/test_lark_legacy_result_reviews.py      | 45 ++++++++++-
 7 files changed, 216 insertions(+), 53 deletions(-)
 create mode 100644 loopx/capabilities/native_chat/result_review_evidence.py

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 625c7cb495..9414720577 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -2065,6 +2065,17 @@ intent. The original worker independently checks current artifacts and returns
 one read-only answer under this source's own idempotent identity. A legacy review
 cannot seed another legacy review; changed evidence and exhausted attempts hold.
 
+Host event-compaction filesystem stamps are excluded from the two retained Turn
+evidence locations before crossing the JSON/JavaScript boundary; nanoseconds
+must not become rounded answer or authority evidence. All semantic fields stay
+pinned. An already failed legacy preparation can recover once only when the
+canonical trace is exactly queue then local `host_gate`, with no native input,
+delta or response, and fresh native inventory excludes hidden dispatch. This
+retains the failed Turn and reservation in history and uses a distinct, pinned
+preparation receipt/phase. It does not renew the single model invocation: an
+actual dispatch, unknown native work, changed source/evidence, Stop, or a second
+preparation failure remains held. Reservation/queue crashes adopt that receipt.
+
 A new result review now requires an explicit observation that no original
 writable adapter is attached. An inconclusive effect readback while that owner
 is still attached remains a hold; it must not replace the owner with a model
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 86fce7ff6f..da345722a3 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
@@ -1357,6 +1357,14 @@ binding 和受众下，更早的独立来源提出完全相同的问题，且其
 原 worker 独立检查当前产物，沿本条原消息自己的幂等身份补回一次只读答案。
 旧请求复核不能再作为其他旧请求的准入依据；证据变化或次数用尽继续保持等待。
 
+两处保留 Turn 证据在跨 JSON/JavaScript 边界前排除宿主事件压缩用的文件系统标记，
+避免纳秒数值舍入被当成答案或授权变化；全部语义字段仍固定。旧准备阶段已失败时，
+只有规范轨迹恰为入队后本地 `host_gate`、没有原生输入、delta 或答复，且新鲜原生
+清单排除隐藏派发，才允许一次准备恢复。失败 Turn 与预约继续留在历史中，恢复使用
+独立固定的准备回执及投递阶段，不重新授予一次模型调用。已经派发、原生工作未知、
+来源或证据变化、停止，以及第二次准备失败继续保持等待；预约和入队崩溃只认回
+同一个准备回执。
+
 新的结果复核现在要求明确观测到没有挂接原可写 adapter。原进程仍挂接时，效果
 读回暂时没有结论，应保持等待，不能为复述已投递的部分答复而用模型复核替换
 原进程。已有回复 worker 的探测等待间隔继续生效；后续取得新鲜效果结清证明后，
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index b90f1cb482..184bf3fa38 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -23,6 +23,7 @@
     unavailable_automation_cadence_readback,
 )
 from ...file_lock import exclusive_file_lock, try_exclusive_file_lock
+from . import result_review_evidence
 from .codex_context import checked_session_home
 from .file_discovery import FileChatDiscovery
 
@@ -116,43 +117,11 @@ def _retained_result_review(self, row: dict[str, Any]) -> dict[str, Any] | None:
                 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"])
+        if reviewed is None and reservation.get("review_preparation_attempt") == 1:
+            reviewed = self.controller.store.load_turn(row["session_id"], reservation["preparation_failed_turn_id"])
         return {"reservation": reservation, "turn": reviewed} if reviewed else None
 
-    def _related_result_review(self, row: dict[str, Any], original: dict[str, Any], *,
-                               discovery_reads: _RecoveryDiscoveryReads | None = None) -> dict[str, Any] | None:
-        """Collect same-question evidence, never equate distinct source identities.
 
-        Core qualifies the complete correlation. Discovery shares only a local
-        history index; admission and dispatch read the actual records afresh.
-        """
-        store = self.controller.store
-        messages = (discovery_reads.matching_user_messages(row["session_id"], row["message"])
-            if discovery_reads is not None else [item for item in store.messages(row["session_id"], strict=True)
-                if item.get("role") == "user" and item.get("text") == row["message"]])
-        pinned = ((row.get("context_recovery") or {}).get("related_result_review") or {}).get("request", {}).get("request_ref")
-        for message in sorted(messages, key=lambda item: item["created_at"], reverse=True):
-            if (message.get("turn_id") == original["turn_id"]
-                    or datetime.fromisoformat(message["created_at"]) <= datetime.fromisoformat(original["completed_at"])):
-                continue
-            reviewed = store.load_turn(row["session_id"], message.get("turn_id", ""))
-            if not reviewed or reviewed.get("status") != "completed":
-                continue
-            reservation = self.recovery_context({"session_id": row["session_id"]}, reviewed)
-            if not reservation or not reservation.get("unfinished_work") or reservation.get("legacy_result_review"):
-                continue
-            if pinned is not None and reservation["request_ref"] != pinned:
-                continue
-            sibling = self.read_request(reservation["request_ref"])
-            if (sibling["request_ref"] == row["request_ref"] or sibling.get("message") != row["message"]
-                    or sibling.get("binding_id") != row["binding_id"] or sibling.get("source") != row["source"]
-                    or sibling.get("turn_id") != reviewed["turn_id"] or not sibling.get("delivery_verified")):
-                continue
-            events = store.events_after(row["session_id"], reviewed["turn_id"], None)
-            events += store.events_after(row["session_id"], reservation["failed_turn_id"], None)
-            return {"request": {key: sibling.get(key) for key in ("request_ref", "binding_id", "source", "session_id",
-                        "message", "origin", "command", "status", "turn_id", "delivery_verified", "context_recovery")},
-                "turn": reviewed, "steering_observed": any(event.get("kind") == "turn.steered" for event in events)}
-        return None
 
     def _context_recovery_plan(self, row: dict[str, Any], *,
                                observation: dict[str, Any] | None = None,
@@ -200,7 +169,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                    if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")]
         session = self._original_native_session(session, anchors)
         unfinished = turn.get("status") == "completed" and bool((turn.get("response") or {}).get("unfinished_work"))
-        related_review = (self._related_result_review(row, turn, discovery_reads=discovery_reads)
+        related_review = (result_review_evidence.collect_related_result_review(self, row, turn, discovery_reads=discovery_reads)
             if turn.get("status") == "completed" and not unfinished
                 and (not reservation or reservation.get("legacy_result_review")) else None)
         live_work_owner = False
@@ -239,8 +208,19 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 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
+        recovery_row = {key: value for key, value in row.items() if key != "attachments"}
+        review_preparation = None
+        if reservation.get("legacy_result_review"):
+            recovery_row = deepcopy(recovery_row)
+            recovery_row["context_recovery"]["related_result_review"] = result_review_evidence.related_review_basis(reservation["related_result_review"])
+            if retained_review is not None:
+                failed_review = (store.load_turn(session["session_id"], reservation["preparation_failed_turn_id"])
+                    if reservation.get("review_preparation_attempt") == 1 else retained_review["turn"])
+                if failed_review is not None:
+                    review_preparation = {"turn": failed_review,
+                        "events": store.events_after(session["session_id"], failed_review["turn_id"], None)}
         return effect_runtime_result("collaboration.conversation.request", {
-            "request": {key: value for key, value in row.items() if key != "attachments"},
+            "request": recovery_row,
             "current_session": session, "binding": selected_binding,
             "context_recovery": {"version": MANAGER_CONTEXT_VERSION,
                 "failed_turn": ({key: value for key, value in turn.items() if key != "attachments"}
@@ -272,6 +252,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 **({"retained_result_review": retained_review} if retained_review is not None else {}),
                 **({"live_work_owner": live_work_owner} if unfinished or related_review is not None else {}),
                 **({"related_result_review": related_review} if related_review is not None else {}),
+                **({"result_review_preparation": review_preparation} if review_preparation 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,
@@ -431,7 +412,7 @@ def context_recovery_candidate(self, request_ref: str, *,
                 _, 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 and not self._retained_result_review(row)
-                        and self._related_result_review(row, turn, discovery_reads=discovery_reads) is None):
+                        and result_review_evidence.collect_related_result_review(self, row, turn, discovery_reads=discovery_reads) is None):
                     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
@@ -459,7 +440,7 @@ def recover_failed_request(self, request_ref: str, *,
             session = self.controller.store.load_session(row["session_id"])
             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)
+                    if plan.get("completed_review_turn_id") or plan.get("original_result_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)
@@ -585,7 +566,7 @@ def source_requests() -> list[dict[str, Any]]:
                 # 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"], current_turn["turn_id"]) != current_turn
-                        or plan.get("completed_review_turn_id")
+                        or (plan.get("completed_review_turn_id") or plan.get("original_result_turn_id"))
                         and self.controller.store.load_turn(row["session_id"], turn["turn_id"]) != turn
                         or source_requests() != requests):
                     return row
@@ -627,6 +608,10 @@ def source_requests() -> list[dict[str, Any]]:
                             else:
                                 recovery.update(legacy_result_review=True,
                                     related_result_review=deepcopy(plan["related_result_review"]))
+                            if plan.get("review_preparation_attempt") == 1:
+                                recovery.update(review_preparation_attempt=1,
+                                    preparation_failed_turn_id=plan["preparation_failed_turn_id"],
+                                    presentation_key=f"{MANAGER_CONTEXT_VERSION}.review.1.preparation.1")
                     if plan.get("answer_review"):
                         recovery.update(answer_review=True, answer_executor="codex", original_turn=deepcopy(turn),
                             binding_snapshot=selected["binding"])
@@ -934,7 +919,7 @@ def admit_history_answer(self, *, binding_id: str, source: dict[str, Any], reque
                 return row
 
     def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any] | None:
-        match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-((?:routing|quota|preparation)-[1-3]|review-1|work-1|metadata-1))?", str(turn.get("client_turn_id") or ""))
+        match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-((?:routing|quota|preparation)-[1-3]|review-1(?:-preparation-1)?|work-1|metadata-1))?", str(turn.get("client_turn_id") or ""))
         if match is None:
             return None
         row = self.read_request(match[1])
@@ -942,8 +927,10 @@ def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dic
         preparation = match[2] is not None and match[2].startswith("preparation-")
         if (row.get("session_id") == session["session_id"] and recovery
                 and recovery["client_turn_id"] == turn["client_turn_id"]):
-            if (match[2] == "review-1") != (recovery.get("read_only_review") is True):
+            if bool(match[2] and match[2].startswith("review-1")) != (recovery.get("read_only_review") is True):
                 raise ValueError("reserved result review policy changed")
+            if (match[2] == "review-1-preparation-1") != (recovery.get("review_preparation_attempt") == 1):
+                raise ValueError("reserved result review preparation changed")
             if recovery.get("context_correction") and (recovery.get("history_review") is not True or recovery.get("context_version") != 1):
                 raise ValueError("historical correction requires its qualified context")
             if (match[2] == "work-1") != (recovery.get("work_attempt") == 1):
@@ -954,7 +941,7 @@ def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dic
                     preparation and str(recovery["preparation_attempt"]) != match[2][-1]):
                 raise ValueError("reserved preparation recovery correlation changed")
             return {"request_ref": row["request_ref"], **recovery}
-        if preparation or match[2] in {"review-1", "work-1", "metadata-1"}:
+        if preparation or match[2] in {"review-1", "review-1-preparation-1", "work-1", "metadata-1"}:
             raise ValueError("reserved result review correlation unavailable")
         return None
 
@@ -1240,16 +1227,19 @@ def _checked_completed_result_review(self, session: dict[str, Any], turn: dict[s
         binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
         events = store.events_after(session["session_id"], turn["turn_id"], None)
         events += store.events_after(session["session_id"], review["failed_turn_id"], None)
+        predecessor = (review.get("preparation_failed_turn_id")
+            if review.get("legacy_result_review") and review.get("review_preparation_attempt") == 1 else None)
         if (not session.get("project_context") or review.get("answer_review")
                 or original != review.get("original_turn") or binding != review.get("binding_snapshot")
                 or current.get("active_turn_id") != turn["turn_id"] or current.get("status") == "closed"
-                or row["turn_id"] not in {turn["turn_id"], review["failed_turn_id"]}
+                or row["turn_id"] not in {turn["turn_id"], review["failed_turn_id"], predecessor}
                 or (store.turn_for_client(session["session_id"], review["client_turn_id"]) or {}).get("turn_id") != turn["turn_id"]
                 or store.queued_turns(session["session_id"]) or self._source_stop_observed(row, original or {}, events)
                 or any(event.get("kind") == "turn.steered" for event in events)):
             raise CodexChatAgentError("The original result review source, grant or queue changed; review held.",
                                       error_code="host_gate", gate=None)
-        if review.get("legacy_result_review") and self._related_result_review(row, original) != review.get("related_result_review"):
+        if (review.get("legacy_result_review") and result_review_evidence.collect_related_result_review(self, row, original)
+                != result_review_evidence.related_review_basis(review["related_result_review"])):
             raise CodexChatAgentError("The related result evidence changed; review held.", error_code="host_gate", gate=None)
         if native is not None:
             known = self._settled_native_turn_ids(session["session_id"], include_interrupted=True)
diff --git a/loopx/capabilities/native_chat/result_review_evidence.py b/loopx/capabilities/native_chat/result_review_evidence.py
new file mode 100644
index 0000000000..7278829589
--- /dev/null
+++ b/loopx/capabilities/native_chat/result_review_evidence.py
@@ -0,0 +1,55 @@
+"""Canonical result-review facts; eligibility remains in the typed Core owner."""
+from copy import deepcopy
+from datetime import datetime
+from typing import Any
+
+
+def related_review_basis(value: dict[str, Any]) -> dict[str, Any]:
+    """Host event-cache revisions are not result evidence or JS-safe integers.
+
+    Preserve every semantic field, including response, source, state and
+    budget; exclude only ChatSessionStore's derived filesystem cache stamp
+    at its two known Turn locations, including persisted older reservations.
+    """
+    basis = deepcopy(value)
+    basis["turn"].pop("event_compaction_revision", None)
+    basis["request"]["context_recovery"]["original_turn"].pop("event_compaction_revision", None)
+    return basis
+
+
+def collect_related_result_review(core: Any, row: dict[str, Any], original: dict[str, Any], *,
+                           discovery_reads: Any | None = None) -> dict[str, Any] | None:
+    """Collect same-question evidence, never equate distinct source identities.
+
+    Core qualifies the complete correlation. Discovery shares only a local
+    history index; admission and dispatch read the actual records afresh.
+    """
+    store = core.controller.store
+    messages = (discovery_reads.matching_user_messages(row["session_id"], row["message"])
+        if discovery_reads is not None else [item for item in store.messages(row["session_id"], strict=True)
+            if item.get("role") == "user" and item.get("text") == row["message"]])
+    pinned = ((row.get("context_recovery") or {}).get("related_result_review") or {}).get("request", {}).get("request_ref")
+    for message in sorted(messages, key=lambda item: item["created_at"], reverse=True):
+        if (message.get("turn_id") == original["turn_id"]
+                or datetime.fromisoformat(message["created_at"]) <= datetime.fromisoformat(original["completed_at"])):
+            continue
+        reviewed = store.load_turn(row["session_id"], message.get("turn_id", ""))
+        if not reviewed or reviewed.get("status") != "completed":
+            continue
+        reservation = core.recovery_context({"session_id": row["session_id"]}, reviewed)
+        if not reservation or not reservation.get("unfinished_work") or reservation.get("legacy_result_review"):
+            continue
+        if pinned is not None and reservation["request_ref"] != pinned:
+            continue
+        sibling = core.read_request(reservation["request_ref"])
+        if (sibling["request_ref"] == row["request_ref"] or sibling.get("message") != row["message"]
+                or sibling.get("binding_id") != row["binding_id"] or sibling.get("source") != row["source"]
+                or sibling.get("turn_id") != reviewed["turn_id"] or not sibling.get("delivery_verified")):
+            continue
+        events = store.events_after(row["session_id"], reviewed["turn_id"], None)
+        events += store.events_after(row["session_id"], reservation["failed_turn_id"], None)
+        return related_review_basis({"request": {key: sibling.get(key) for key in ("request_ref", "binding_id", "source", "session_id",
+                    "message", "origin", "command", "status", "turn_id", "delivery_verified", "context_recovery")},
+            "turn": reviewed, "steering_observed": any(event.get("kind") == "turn.steered" for event in events)})
+    return None
+
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index cb5c73cbe1..966391ee4e 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -1238,13 +1238,20 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
 function planRelatedResultReview(recovery: JsonObject, row: JsonObject, current: JsonObject,
   completed: JsonObject, previous: JsonObject | null, selected: JsonObject, version: number): JsonObject {
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
+  const baseClient = `external-${ref(row.request_ref, "original request")}-recovery-${version}-review-1`;
+  const preparationReserved = previous?.review_preparation_attempt === 1;
+  const history = (row.context_recovery_history ?? []) as JsonObject[];
   if (selected.context_kind !== "project" || current.agent_id !== "codex"
       || !pureResultText(completed.response as JsonObject | null)
       || completed.session_id !== current.session_id || completed.status !== "completed"
       || recovery.attachments_present !== false || recovery.steering_observed !== false
       || recovery.stop_observed !== false || typeof completed.upstream_turn_id !== "string"
       || !completed.upstream_turn_id || previous !== null && previous.legacy_result_review !== true
-      || (Array.isArray(row.context_recovery_history) && row.context_recovery_history.length)) return no;
+      || previous?.review_preparation_attempt != null && !preparationReserved
+      || !Array.isArray(history) || history.length && !(preparationReserved && history.length === 1
+        && history[0].legacy_result_review === true && history[0].read_only_review === true
+        && history[0].review_preparation_attempt == null && history[0].client_turn_id === baseClient
+        && history[0].failed_turn_id === completed.turn_id)) return no;
   const related = recovery.related_result_review as JsonObject | null | undefined;
   const sibling = related?.request as JsonObject | undefined;
   const reviewed = related?.turn as JsonObject | undefined;
@@ -1271,10 +1278,28 @@ function planRelatedResultReview(recovery: JsonObject, row: JsonObject, current:
       || !Number.isFinite(Date.parse(String(completed.completed_at)))
       || Date.parse(String(original.completed_at)) > Date.parse(String(completed.created_at))
       || Date.parse(String(reviewed.completed_at)) <= Date.parse(String(completed.completed_at))) return no;
-  // Completed or failed reviews never acquire a second attempt from another
-  // sibling. A lost queue acknowledgement may adopt only its exact receipt.
-  if (previous !== null && row.turn_id !== completed.turn_id && row.delivery_verified === true) return no;
-  return planCompletedResultReview(recovery, row, current, completed, previous, version, null);
+  // A local preparation failure may leave the single model invocation unused.
+  // Require the exact queue/host-gate terminal record, no native input/delta or
+  // response, fresh closed native inventory, and retain this failed reservation.
+  // Never recover a review that actually reached the model or spend twice.
+  const preparation = recovery.result_review_preparation as JsonObject | undefined;
+  const failed = preparation?.turn as JsonObject | undefined;
+  const events = preparation?.events;
+  const preparationOnly = previous !== null && failed?.status === "failed"
+    && failed.session_id === current.session_id && failed.client_turn_id === baseClient
+    && row.turn_id === failed.turn_id && failed.message === row.message && failed.origin === row.origin
+    && failed.error_code === "host_gate" && failed.upstream_turn_id == null && failed.response == null
+    && failed.delta_count === 0 && Array.isArray(events) && events.length === 2
+    && (events[0] as JsonObject).kind === "turn.queued" && (events[1] as JsonObject).kind === "turn.failed"
+    && ((events[1] as JsonObject).payload as JsonObject | undefined)?.error_code === "host_gate"
+    && Number.isFinite(Date.parse(String(failed.completed_at)))
+    && Date.parse(String(recovery.observed_at)) >= Date.parse(String(failed.completed_at)) + 60_000;
+  if (previous !== null && row.turn_id !== completed.turn_id && row.delivery_verified === true && !preparationOnly) return no;
+  const observation = preparationOnly && !preparationReserved ? {...recovery, admitted_turn: null} : recovery;
+  const plan = planCompletedResultReview(observation, row, current, completed, previous, version, null);
+  return preparationOnly && plan.recovery_candidate === true ? {...plan,
+    client_turn_id: `${baseClient}-preparation-1`, review_preparation_attempt: 1,
+    preparation_failed_turn_id: failed!.turn_id, original_result_turn_id: completed.turn_id} : plan;
 }
 
 /** A result-only fallback never consumes or restores the writable retry budget. */
@@ -1285,7 +1310,9 @@ function planCompletedResultReview(recovery: JsonObject, row: JsonObject, curren
   // owner with a result reader. Keep that owner for the next fresh work proof.
   // Already reserved reviews retain their own adoption and execution checks.
   if (previous === null && recovery.live_work_owner !== false) return no;
-  const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-review-1`;
+  const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-review-1`
+    + (previous?.review_preparation_attempt === 1 ? "-preparation-1" : "");
+  if (work !== null && previous?.review_preparation_attempt != null) return no;
   if (current.steward_context != null || completed.client_turn_id !== `external-${row.request_ref}`
       || completed.message !== row.message || completed.origin !== row.origin
       || previous !== null && (previous.read_only_review !== true || previous.failed_turn_id !== completed.turn_id
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 3775b05577..7036c77d0f 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -1208,6 +1208,39 @@ test("a delivered same-question review permits one independent legacy result rea
   const held = {...interrupted, request: {...interrupted.request, turn_id: admitted.turn_id},
     context_recovery: {...interrupted.context_recovery, retained_result_review: {reservation: own, turn: admitted}}};
   assert.equal(planBoundConversationRequest(held).recovery_candidate, false, "A delivered review never obtains another attempt");
+  const failedPreparation = {...admitted, status: "failed", response: null, upstream_turn_id: null,
+    error_code: "host_gate", delta_count: 0, completed_at: "2026-01-01T00:08:00Z"};
+  const events = [{kind: "turn.queued", payload: {delivery_mode: "session_queue"}},
+    {kind: "turn.failed", payload: {error_code: "host_gate"}}];
+  const beforeModel = {...packet, request: {...packet.request, context_recovery: own, turn_id: failedPreparation.turn_id},
+    context_recovery: {...packet.context_recovery, retained_result_review: {reservation: own, turn: failedPreparation},
+      admitted_turn: failedPreparation, result_review_preparation: {turn: failedPreparation, events}}};
+  const prepared = planBoundConversationRequest(beforeModel);
+  assert.equal(prepared.recovery_eligible, true);
+  assert.equal(prepared.read_only_review, true);
+  assert.equal(prepared.review_preparation_attempt, 1);
+  assert.equal(prepared.original_result_turn_id, completed.turn_id);
+  assert.equal(prepared.client_turn_id, `${plan.client_turn_id}-preparation-1`);
+  assert.equal(prepared.work_attempt, undefined);
+  for (const change of [{upstream_turn_id: "possibly-dispatched"}, {delta_count: 1}, {response},
+    {error_code: "runtime_error"}, {client_turn_id: "another-review"}]) {
+    assert.equal(planBoundConversationRequest({...beforeModel, context_recovery: {...beforeModel.context_recovery,
+      result_review_preparation: {turn: {...failedPreparation, ...change}, events}}}).recovery_eligible, false);
+  }
+  for (const change of [{source_verified: false}, {stop_observed: true}, {native_observation: null},
+    {result_review_preparation: {turn: failedPreparation, events: [...events, {kind: "turn.started"}]}}]) {
+    assert.equal(planBoundConversationRequest({...beforeModel,
+      context_recovery: {...beforeModel.context_recovery, ...change}}).recovery_eligible, false);
+  }
+  const retriedReservation = {...own, client_turn_id: prepared.client_turn_id,
+    review_preparation_attempt: 1, preparation_failed_turn_id: failedPreparation.turn_id};
+  const admittedPreparation = {...admitted, turn_id: "prepared-review", client_turn_id: prepared.client_turn_id};
+  const lostAck = {...beforeModel, request: {...beforeModel.request, context_recovery: retriedReservation,
+    context_recovery_history: [own]}, current_session: {...session, status: "busy", active_turn_id: admittedPreparation.turn_id},
+    context_recovery: {...beforeModel.context_recovery, admitted_turn: admittedPreparation, native_observation: null}};
+  assert.equal(planBoundConversationRequest(lostAck).admitted_turn_id, admittedPreparation.turn_id);
+  assert.equal(planBoundConversationRequest({...lostAck, request: {...lostAck.request, turn_id: admittedPreparation.turn_id}})
+    .recovery_candidate, false, "A second preparation recovery is never granted");
 });
 
 test("the App steward is an explicit source preference, never a new grant or peer fallback", () => {
diff --git a/tests/test_lark_legacy_result_reviews.py b/tests/test_lark_legacy_result_reviews.py
index 538cb98074..ce2f1a7288 100644
--- a/tests/test_lark_legacy_result_reviews.py
+++ b/tests/test_lark_legacy_result_reviews.py
@@ -1,5 +1,6 @@
 """Independent original-source returns through the existing worker and native protocol."""
 from datetime import datetime, timedelta, timezone
+from copy import deepcopy
 import json
 from pathlib import Path
 
@@ -66,7 +67,8 @@ def historical_native(fake, workspace):
     return artifact
 
 
-@pytest.mark.parametrize("fault", [None, "source-changed", "stop", "revoked", "reservation-crash", "enqueue-crash"])
+@pytest.mark.parametrize("fault", [None, "source-changed", "stop", "revoked", "reservation-crash", "enqueue-crash",
+    "compacted-source", "precision-failure", "precision-reservation-crash", "precision-enqueue-crash"])
 def test_original_worker_returns_each_legacy_message_without_replaying_business(ordinary, monkeypatch, fault):  # noqa: F811
     store, runtime, contexts, _, capture, fake, workspace = ordinary
     contexts.workspace_grant = "workspace_write"
@@ -106,6 +108,17 @@ def now(cls, tz=None):
         assert runtime.wait_for_turn(session_id=reviewed["session_id"], turn_id=reviewed["turn_id"], timeout_sec=10)["status"] == "completed"
         transport.reconcile_request(earlier[3])
         assert transport.core.read_request(reviewed["request_ref"])["delivery_verified"] is True
+        if fault == "compacted-source" or str(fault).startswith("precision-"):
+            # Actual old event compaction leaves nanosecond filesystem stamps.
+            # They exceed JSON/JS integer precision but do not change an answer.
+            peer_path = transport.core.root / f"{reviewed['request_ref']}.json"
+            peer = json.loads(peer_path.read_text())
+            peer["context_recovery"]["original_turn"]["event_compaction_revision"] = [7, 123, 2**60 + 123]
+            peer_path.write_text(json.dumps(peer))
+            turn_path = store._turn_path(reviewed["session_id"], reviewed["turn_id"])
+            peer_turn = json.loads(turn_path.read_text())
+            peer_turn["event_compaction_revision"] = [8, 456, 2**60 + 789]
+            turn_path.write_text(json.dumps(peer_turn))
         if fault == "source-changed":
             provider.messages[legacy[0]["message_id"]]["content"] = "A changed instruction"
         elif fault == "stop":
@@ -115,10 +128,29 @@ def now(cls, tz=None):
         def worker():
             return LarkPrivateConversations(controller=runtime, runtime_root=store.root.parent,
                 runner=provider, cli_bin="lark-cli")
-        if fault in {"reservation-crash", "enqueue-crash"}:
+        failed_preparation = None
+        if str(fault).startswith("precision-"):
+            # Deliberately restore the released serialization defect at the
+            # real worker/Core boundary. No synthetic success receipt is added.
+            with monkeypatch.context() as old_release:
+                old_release.setattr(external_conversations.result_review_evidence,
+                    "related_review_basis", deepcopy)
+                old_worker = worker()
+                old_worker.reconcile()
+                failed_request = old_worker.core.read_request(legacy[1]["request_ref"])
+                failed_preparation = runtime.wait_for_turn(session_id=failed_request["session_id"],
+                    turn_id=failed_request["turn_id"], timeout_sec=10)
+                assert failed_preparation["status"] == "failed"
+                assert failed_preparation["error_code"] == "host_gate" and failed_preparation["upstream_turn_id"] is None
+                assert failed_preparation["delta_count"] == 0 and failed_preparation["response"] is None
+                assert [e["kind"] for e in store.events_after(failed_request["session_id"], failed_request["turn_id"], None)] == [
+                    "turn.queued", "turn.failed"]
+                old_worker.reconcile_request(legacy[3])
+            clock += timedelta(seconds=61)
+        if fault in {"reservation-crash", "enqueue-crash", "precision-reservation-crash", "precision-enqueue-crash"}:
             enqueue = runtime.enqueue_turn
             def crash(**kwargs):
-                if fault == "enqueue-crash":
+                if fault in {"enqueue-crash", "precision-enqueue-crash"}:
                     enqueue(**kwargs)
                 raise OSError("synthetic lost admission acknowledgement")
             monkeypatch.setattr(runtime, "enqueue_turn", crash)
@@ -136,6 +168,13 @@ def crash(**kwargs):
             recovery = returned["context_recovery"]
             assert recovery["legacy_result_review"] and recovery["read_only_review"]
             assert "unfinished_work" not in recovery and "work_attempt" not in recovery
+            if failed_preparation is not None:
+                assert recovery["review_preparation_attempt"] == 1
+                assert recovery["preparation_failed_turn_id"] == failed_preparation["turn_id"]
+                assert len(returned["context_recovery_history"]) == 1
+                old_failure = store.load_turn(returned["session_id"], failed_preparation["turn_id"])
+                assert old_failure["status"] == "failed" and old_failure["upstream_turn_id"] is None
+                assert old_failure["error"] == failed_preparation["error"]
             review = runtime.wait_for_turn(session_id=returned["session_id"], turn_id=returned["turn_id"], timeout_sec=10)
             assert review["status"] == "completed", review
             assert review["response"]["message"] == "Verified existing result: catalog-item:verified"

From 61630caba8ea46ed59249277faf4f6aefe3ddf1d Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 8 Nov 2026 14:36:28 +0800
Subject: [PATCH 2/3] style(chat): normalize result evidence module ending

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 loopx/capabilities/native_chat/result_review_evidence.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/loopx/capabilities/native_chat/result_review_evidence.py b/loopx/capabilities/native_chat/result_review_evidence.py
index 7278829589..c312983b26 100644
--- a/loopx/capabilities/native_chat/result_review_evidence.py
+++ b/loopx/capabilities/native_chat/result_review_evidence.py
@@ -52,4 +52,3 @@ def collect_related_result_review(core: Any, row: dict[str, Any], original: dict
                     "message", "origin", "command", "status", "turn_id", "delivery_verified", "context_recovery")},
             "turn": reviewed, "steering_observed": any(event.get("kind") == "turn.steered" for event in events)})
     return None
-

From ac3433301e9c310984bfcee231f11ebbe87b3e25 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sun, 8 Nov 2026 14:50:42 +0800
Subject: [PATCH 3/3] fix(chat): hold read-only preparation receipts on older
 runtimes

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../rfcs/app-conversation-and-async-inbox-v0.md    |  2 ++
 .../app-conversation-and-async-inbox-v0.zh-CN.md   |  2 ++
 .../native_chat/external_conversations.py          | 14 ++++++++------
 .../collaboration/conversation_binding.ts          |  8 +++++---
 .../control_plane_ts/conversation_binding.test.ts  |  2 +-
 tests/test_lark_legacy_result_reviews.py           |  2 ++
 6 files changed, 20 insertions(+), 10 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 9414720577..2ab2883f53 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -2075,6 +2075,8 @@ retains the failed Turn and reservation in history and uses a distinct, pinned
 preparation receipt/phase. It does not renew the single model invocation: an
 actual dispatch, unknown native work, changed source/evidence, Stop, or a second
 preparation failure remains held. Reservation/queue crashes adopt that receipt.
+Its existing `preparation-1` key carries the explicit read-only preparation tag;
+older readers reject that policy combination before model dispatch on rollback.
 
 A new result review now requires an explicit observation that no original
 writable adapter is attached. An inconclusive effect readback while that owner
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 da345722a3..ed6109d81c 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
@@ -1364,6 +1364,8 @@ binding 和受众下，更早的独立来源提出完全相同的问题，且其
 独立固定的准备回执及投递阶段，不重新授予一次模型调用。已经派发、原生工作未知、
 来源或证据变化、停止，以及第二次准备失败继续保持等待；预约和入队崩溃只认回
 同一个准备回执。
+它沿用 `preparation-1` 标识并携带显式只读准备标记，旧版本在回滚后会于模型
+派发前拒绝这一策略组合。
 
 新的结果复核现在要求明确观测到没有挂接原可写 adapter。原进程仍挂接时，效果
 读回暂时没有结论，应保持等待，不能为复述已投递的部分答复而用模型复核替换
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index 184bf3fa38..d09395faa2 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -919,7 +919,7 @@ def admit_history_answer(self, *, binding_id: str, source: dict[str, Any], reque
                 return row
 
     def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any] | None:
-        match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-((?:routing|quota|preparation)-[1-3]|review-1(?:-preparation-1)?|work-1|metadata-1))?", str(turn.get("client_turn_id") or ""))
+        match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-((?:routing|quota|preparation)-[1-3]|review-1|work-1|metadata-1))?", str(turn.get("client_turn_id") or ""))
         if match is None:
             return None
         row = self.read_request(match[1])
@@ -927,9 +927,11 @@ def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dic
         preparation = match[2] is not None and match[2].startswith("preparation-")
         if (row.get("session_id") == session["session_id"] and recovery
                 and recovery["client_turn_id"] == turn["client_turn_id"]):
-            if bool(match[2] and match[2].startswith("review-1")) != (recovery.get("read_only_review") is True):
+            review_preparation = (match[2] == "preparation-1" and recovery.get("legacy_result_review") is True
+                and recovery.get("read_only_review") is True and recovery.get("review_preparation_attempt") == 1)
+            if (match[2] == "review-1" or review_preparation) != (recovery.get("read_only_review") is True):
                 raise ValueError("reserved result review policy changed")
-            if (match[2] == "review-1-preparation-1") != (recovery.get("review_preparation_attempt") == 1):
+            if review_preparation != (recovery.get("review_preparation_attempt") == 1):
                 raise ValueError("reserved result review preparation changed")
             if recovery.get("context_correction") and (recovery.get("history_review") is not True or recovery.get("context_version") != 1):
                 raise ValueError("historical correction requires its qualified context")
@@ -937,11 +939,11 @@ def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dic
                 raise ValueError("reserved unfinished work correlation changed")
             if (match[2] == "metadata-1") != (recovery.get("metadata_review") is True):
                 raise ValueError("reserved metadata review correlation changed")
-            if preparation != (recovery.get("preparation_attempt") is not None) or (
-                    preparation and str(recovery["preparation_attempt"]) != match[2][-1]):
+            if (preparation and not review_preparation) != (recovery.get("preparation_attempt") is not None) or (
+                    preparation and not review_preparation and str(recovery["preparation_attempt"]) != match[2][-1]):
                 raise ValueError("reserved preparation recovery correlation changed")
             return {"request_ref": row["request_ref"], **recovery}
-        if preparation or match[2] in {"review-1", "review-1-preparation-1", "work-1", "metadata-1"}:
+        if preparation or match[2] in {"review-1", "work-1", "metadata-1"}:
             raise ValueError("reserved result review correlation unavailable")
         return None
 
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 966391ee4e..94b423b33a 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -1298,7 +1298,7 @@ function planRelatedResultReview(recovery: JsonObject, row: JsonObject, current:
   const observation = preparationOnly && !preparationReserved ? {...recovery, admitted_turn: null} : recovery;
   const plan = planCompletedResultReview(observation, row, current, completed, previous, version, null);
   return preparationOnly && plan.recovery_candidate === true ? {...plan,
-    client_turn_id: `${baseClient}-preparation-1`, review_preparation_attempt: 1,
+    client_turn_id: `external-${row.request_ref}-recovery-${version}-preparation-1`, review_preparation_attempt: 1,
     preparation_failed_turn_id: failed!.turn_id, original_result_turn_id: completed.turn_id} : plan;
 }
 
@@ -1310,8 +1310,10 @@ function planCompletedResultReview(recovery: JsonObject, row: JsonObject, curren
   // owner with a result reader. Keep that owner for the next fresh work proof.
   // Already reserved reviews retain their own adoption and execution checks.
   if (previous === null && recovery.live_work_owner !== false) return no;
-  const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-review-1`
-    + (previous?.review_preparation_attempt === 1 ? "-preparation-1" : "");
+  // Old readers recognize preparation-1 and reject its read-only policy, so a
+  // rollback holds this receipt instead of treating an unknown key as ordinary input.
+  const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-`
+    + (previous?.review_preparation_attempt === 1 ? "preparation-1" : "review-1");
   if (work !== null && previous?.review_preparation_attempt != null) return no;
   if (current.steward_context != null || completed.client_turn_id !== `external-${row.request_ref}`
       || completed.message !== row.message || completed.origin !== row.origin
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 7036c77d0f..a6e54e8296 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -1220,7 +1220,7 @@ test("a delivered same-question review permits one independent legacy result rea
   assert.equal(prepared.read_only_review, true);
   assert.equal(prepared.review_preparation_attempt, 1);
   assert.equal(prepared.original_result_turn_id, completed.turn_id);
-  assert.equal(prepared.client_turn_id, `${plan.client_turn_id}-preparation-1`);
+  assert.equal(prepared.client_turn_id, String(plan.client_turn_id).replace(/-review-1$/, "-preparation-1"));
   assert.equal(prepared.work_attempt, undefined);
   for (const change of [{upstream_turn_id: "possibly-dispatched"}, {delta_count: 1}, {response},
     {error_code: "runtime_error"}, {client_turn_id: "another-review"}]) {
diff --git a/tests/test_lark_legacy_result_reviews.py b/tests/test_lark_legacy_result_reviews.py
index ce2f1a7288..ac543fd77a 100644
--- a/tests/test_lark_legacy_result_reviews.py
+++ b/tests/test_lark_legacy_result_reviews.py
@@ -170,6 +170,8 @@ def crash(**kwargs):
             assert "unfinished_work" not in recovery and "work_attempt" not in recovery
             if failed_preparation is not None:
                 assert recovery["review_preparation_attempt"] == 1
+                assert recovery["client_turn_id"].endswith("-preparation-1")
+                assert "preparation_attempt" not in recovery
                 assert recovery["preparation_failed_turn_id"] == failed_preparation["turn_id"]
                 assert len(returned["context_recovery_history"]) == 1
                 old_failure = store.load_turn(returned["session_id"], failed_preparation["turn_id"])