loopx-agent · Nov 6, 2026 · Nov 6, 2026 · Nov 6, 2026
@@ -1829,10 +1829,15 @@ steward, group, attached-host and attachment-bearing requests remain outside it.
 The existing original-source reply worker may reserve one continuation after
 60 seconds. Core revalidates the original owner/App/source, unchanged message,
 workspace grant and stop/correction state. Native readback must independently
-show that the exact latest Turn completed, its full item inventory is stable,
+show that the exact original Turn completed, its full item inventory is stable,
 every effect item completed, no background work remains, and the same final
-answer explicitly contains that report. A later native Turn, unknown item,
-failed/declined/live effect, edited source or revoked grant remains a hold.
+answer explicitly contains that report. Historical work may continue only when
+every later native Turn correlates to a settled canonical Core Turn and its full
+item inventory is independently stable with no unknown or unfinished effects.
+The proof pins a digest of these inventories; a later equivalent source submission,
+foreign native input, unknown item, failed/declined/live effect, edited source or
+revoked grant remains a hold. Native loss remains a hold: a cold client cannot
+certify another process's background effects by resuming the thread.
 
 The original Session queue and deterministic `work-1` reservation own execution;
 restart adopts an already committed Turn instead of dispatching it again. The

@@ -1195,10 +1195,13 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果
 及携带附件的请求不进入该路径。
 
 既有原消息回复 worker 在 60 秒后才可预留一次续办。Core 重新核对原本人/App/
-来源、未变化的消息、工作区 grant 和停止/纠正状态。独立原生读回必须证明精确最新
+来源、未变化的消息、工作区 grant 和停止/纠正状态。独立原生读回必须证明精确原始
 Turn 已完成、完整 item 库存稳定、所有效果 item 均已完成、没有后台工作，且同一
-最终回答确实包含该报告。后续原生 Turn、未知 item、失败/拒绝/活动效果、来源
-被编辑或 grant 被撤销均保持等待。
+最终回答确实包含该报告。历史工作只有在全部后续原生 Turn 均关联已结束的规范
+Core Turn，且各自完整 item 库存独立稳定、没有未知或未结束效果时才可续办。
+证明绑定这些库存的摘要；后续同内容来源提交、外来原生输入、未知 item、失败/
+拒绝/活动效果、来源被编辑或 grant 被撤销均保持等待。原生 owner 丢失仍保持等待：
+冷客户端不能通过恢复 thread 来证明另一个进程没有后台效果。
 
 执行复用原 Session 队列和确定性的 `work-1` 预留；重启采用已提交 Turn，不能再次
 派发。实际派发前再次读回原生库存，有新输入排队时让其优先。worker 先读原历史和

@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import hashlib
 import json
 import logging
 import os
@@ -269,8 +270,9 @@ def terminal() -> dict[str, Any] | None:
         return None
 
 
-def completed_work_readback(session: CodexChatAgentSession, turn_id: str) -> dict[str, Any] | None:
-    """Read a latest, stable completed inventory with an explicit unfinished report.
+def completed_work_readback(session: CodexChatAgentSession, turn_id: str, *,
+                            known_turn_ids: set[str] | None = None) -> dict[str, Any] | None:
+    """Read stable settled work, including history correlated by its Core owner.
 
     This proves settled native items, not the safety of repeating their effects.
     Core owns the original source, grant, bounded continuation and stop checks.
@@ -286,12 +288,16 @@ def turns():
 
         inventory = turns()
         ids = [item.get("id") for item in inventory]
-        if (not ids or ids[0] != turn_id or inventory[0].get("status") != "completed"
+        target = next((item for item in inventory if item.get("id") == turn_id), None)
+        if (target is None or target.get("status") != "completed"
                 or any(not isinstance(value, str) or not value for value in ids)
                 or len(set(ids)) != len(ids)
                 or any(item.get("status") not in {"completed", "failed", "interrupted"} for item in inventory)
                 or session._background_terminals(deadline)):
             return None
+        later = ids[:ids.index(turn_id)]
+        if any(value not in (known_turn_ids or set()) for value in later):
+            return None  # A foreign native input may cancel or supersede the old request.
         items = list(session._control_items("thread/items/list", {"turnId": turn_id}, deadline))
         seen: set[str] = set()
         users = 0
@@ -322,8 +328,35 @@ def turns():
         unfinished = parse_agent_response(finals[0], protected_paths=[session.work_dir]).get("unfinished_work")
         if unfinished is None:
             return None
+        later_items = {}
+        for value in later:
+            rows = list(session._control_items("thread/items/list", {"turnId": value}, deadline))
+            if any(not isinstance(row, dict) or not isinstance(row.get("item"), dict) for row in rows):
+                return None
+            item_ids = [row.get("item", {}).get("id") for row in rows]
+            if (not rows or any(row.get("turnId") != value for row in rows)
+                    or any(not isinstance(item_id, str) or not item_id for item_id in item_ids)
+                    or len(set(item_ids)) != len(item_ids)
+                    or sum(row["item"].get("type") == "userMessage" for row in rows) != 1):
+                return None
+            for row in rows:
+                item = row["item"]
+                if (item.get("type") not in {"userMessage", "agentMessage", "commandExecution", "fileChange",
+                        "mcpToolCall", "reasoning", "webSearch", "imageView", "contextCompaction"}
+                        or item.get("type") == "agentMessage" and item.get("phase") not in {"commentary", "final_answer"}
+                        or item.get("type") in {"commandExecution", "fileChange", "mcpToolCall"}
+                        and item.get("status") != "completed"):
+                    return None
+            later_items[value] = rows
+        if (any(list(session._control_items("thread/items/list", {"turnId": value}, deadline)) != rows
+                for value, rows in later_items.items())
+                or turns() != inventory or session._background_terminals(deadline)):
+            return None
+        digest = hashlib.sha256(json.dumps([inventory, items, later_items],
+            sort_keys=True, separators=(",", ":")).encode()).hexdigest()
         return {"thread_id": session.thread_id, "turn_id": turn_id,
-                "completed_work": True, "all_effects_settled": True, "unfinished_work": unfinished}
+                "completed_work": True, "all_effects_settled": True, "unfinished_work": unfinished,
+                "native_inventory_sha256": digest}
     except (CodexChatAgentError, OSError, TypeError, ValueError):
         return None
 
@@ -384,7 +417,8 @@ class CodexAppServerAdapter:
     session: CodexChatAgentSession
     goal_driver: CodexGoalDriver | None = None
 
-    def completed_work_observation(self, *, thread_ids: list[str], turn_id: str) -> dict[str, Any] | None:
+    def completed_work_observation(self, *, thread_ids: list[str], turn_id: str,
+                                   known_turn_ids: set[str] | None = None) -> dict[str, Any] | None:
         """Read live effects through the adapter that owns the original thread.
 
         A cold app-server can read persisted items, but cannot inventory another
@@ -393,7 +427,7 @@ def completed_work_observation(self, *, thread_ids: list[str], turn_id: str) ->
         """
         if self.upstream_thread_id not in thread_ids or not self.healthcheck():
             return None
-        return completed_work_readback(self.session, turn_id)
+        return completed_work_readback(self.session, turn_id, known_turn_ids=known_turn_ids)
 
     @staticmethod
     def failed_request_observation(*, codex_bin: str, codex_home: Path,

@@ -142,6 +142,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         later_equivalent = None
         image_failure = turn.get("error_code") == "runtime_error" and bool(row.get("attachments"))
         runtime_failure = turn.get("error_code") in {"model_auth_unavailable", "resume_failed"}
+        unfinished = turn.get("status") == "completed" and bool((turn.get("response") or {}).get("unfinished_work"))
         if turn.get("error_code") == "manager_authorization_unavailable" or image_failure or runtime_failure:
             # Collect canonical facts only. The typed planner owns the exact
             # pre-dispatch sequence, retry budget and authority decision.
@@ -152,6 +153,7 @@ 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"}
@@ -430,7 +432,8 @@ def source_requests() -> list[dict[str, Any]]:
                         adapter = self.controller.adapters.get(row["session_id"])
                     if isinstance(adapter, CodexAppServerAdapter):
                         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=turn["upstream_turn_id"],
+                            known_turn_ids=self._settled_native_turn_ids(session["session_id"]))
                 else:
                     proof = CodexAppServerAdapter.failed_request_observation(
                         codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
@@ -889,6 +892,16 @@ def validate_metadata_review_response(self, session: dict[str, Any], turn: dict[
         # append each actual receipt/failure before the original worker replies.
         return {**response, "message": "原委派格式已复核，以下为实际提交回执；接收者结论仍待回传。"}
 
+    def _settled_native_turn_ids(self, session_id: str) -> set[str]:
+        """Correlate native history through canonical Core Turns, without adopting it."""
+        store = self.controller.store
+        ids = {item["turn_id"] for item in store.messages(session_id, strict=True)
+               if item.get("role") == "user" and item.get("turn_id")}
+        turns = [store.load_turn(session_id, value) for value in ids]
+        return {item["upstream_turn_id"] for item in turns if item
+                and item.get("status") in {"completed", "failed"}
+                and isinstance(item.get("upstream_turn_id"), str) and item["upstream_turn_id"]}
+
     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."""
@@ -900,7 +913,8 @@ def prepare_continuation_input(self, session: dict[str, Any], turn: dict[str, An
             return message
         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 "")) != proof):
+                or completed_work_readback(native, str(proof.get("turn_id") or ""),
+                    known_turn_ids=self._settled_native_turn_ids(session["session_id"])) != proof):
             raise CodexChatAgentError(
                 "The original unfinished-work inventory changed or newer input is waiting; continuation stopped.",
                 error_code="host_gate", gate=None,

@@ -1033,7 +1033,7 @@ function nativeRecoveryCapacity(recovery: JsonObject, row: JsonObject, current:
 
 /** A completed model Turn can explicitly report unfinished business. The
  * proposal alone grants nothing: only the original worker may reserve one
- * continuation after stable latest-native/source/grant/stop checks.
+ * continuation after stable correlated-native/source/grant/stop checks.
  */
 function unfinishedProjectWork(result: JsonObject): JsonObject | null {
   const raw = result.unfinished_work;
@@ -1078,7 +1078,8 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
       admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted unfinished work"), native_thread_ids: [],
       not_before: previous.not_before};
   }
-  if (current.active_turn_id != null || current.status !== "ready") return no;
+  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));
   if (!Number.isFinite(now) || !Number.isFinite(ended) || now < ended + 60_000) return no;
   const native = sessionIdentity(current.upstream_thread_id, "original native thread");
@@ -1088,6 +1089,7 @@ function planUnfinishedWork(recovery: JsonObject, row: JsonObject, current: Json
   return {...planned, recovery_eligible: recovery.source_verified === true
     && observed.thread_id === native && observed.turn_id === completed.upstream_turn_id
     && observed.completed_work === true && observed.all_effects_settled === true
+    && typeof observed.native_inventory_sha256 === "string" && BARE_SHA256_PATTERN.test(observed.native_inventory_sha256)
     && matchesWork(observed.unfinished_work)};
 }
 

@@ -799,7 +799,7 @@ for (const endpoint of ["codex", "antigravity", "kiro-cli"]) {
   });
 }
 
-test("unfinished completed work requires typed intent and latest settled native/source proof, once", () => {
+test("unfinished completed work requires typed intent and pinned settled native/source proof, once", () => {
   const configured = planConversationBinding(request).state;
   const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};
   const context = resolveBoundConversation({current: configured, binding_id: row.binding_id,
@@ -812,18 +812,22 @@ test("unfinished completed work requires typed intent and latest settled native/
     upstream_thread_id: "native", active_turn_id: null, project_context: context,
     channel_id: `project.external.${row.binding_id}.${source.source_ref}`};
   const native = {thread_id: "native", turn_id: "native-partial", completed_work: true,
+    native_inventory_sha256: "a".repeat(64),
     all_effects_settled: true, unfinished_work: {effects_settled: true, dependency: work.dependency, remaining: work.remaining}};
   const packet = {request: {request_ref: "f".repeat(24), command: null, status: "accepted", delivery_verified: true,
     message: "Original request", origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "partial", source},
     binding: row, current_session: session, context_recovery: {version: 21, failed_turn: completed,
       observed_at: "2026-01-01T00:10:00Z", steering_observed: false, stop_observed: false, attachments_present: false,
+      queued_input_present: false, later_equivalent_input_present: false,
       source_verified: true, native_observation: native}};
   const plan = planBoundConversationRequest(packet);
   assert.equal(plan.recovery_eligible, true);
   assert.equal(plan.work_attempt, 1);
   assert.ok(String(plan.client_turn_id).endsWith("-work-1"));
   for (const change of [{source_verified: false}, {steering_observed: true}, {stop_observed: true},
+    {queued_input_present: true}, {later_equivalent_input_present: true}, {later_equivalent_input_present: null},
     {attachments_present: true}, {observed_at: "2026-01-01T00:00:59Z"},
+    {native_observation: {...native, native_inventory_sha256: null}},
     {native_observation: {...native, all_effects_settled: false}},
     {native_observation: {...native, thread_id: "foreign"}},
     {native_observation: {...native, unfinished_work: {...work, remaining: "Another task"}}}]) {