Preserve the steward’s Codex checkpoint across isolated answer review

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

@@ -1721,6 +1721,18 @@ hide cancellation or reopen a closed Session.
 All prefixes share the configured hard Turn time budget; it is not renewed for
 each compaction.
 
+A bound Codex steward can retain its working thread across an isolated answer
+review. Before entering review, Core verifies the current working adapter is
+idle and stores its native identity, provider and source-policy basis with the
+review Turn. On return with the same authority, the adapter resumes that native
+thread directly if there are no new visible review answers. Otherwise it uses a
+native fork to inherit the compaction checkpoint and injects only those answers.
+The recovered user input already belongs to the original request. It is not a
+new history message. The child is published only after the exact policy, source,
+queued Turn and imported suffix are verified. An unknown fork or injection
+leaves the original binding and history intact. Unverified legacy or changed
+authority retains the existing context migration path.
+
 Discovery uses the local enabled binding and the same typed candidate planner,
 without provider identity IO for every historical failure. This is a queue hint;
 admission still re-resolves current App/owner/grant and exact source/native proof

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

@@ -1117,6 +1117,13 @@ Session 的关闭先发布既有存储状态转换，再等待 adapter 清理；
 也不能让已关闭的会话重新打开。
 所有批次共用已配置的 Turn 硬时限，不为每次压缩续期。
 
+已绑定的 Codex 管家进入隔离答复复核前，Core 核验当前工作 adapter 空闲，
+将原生线程、provider 和来源权限依据与复核 Turn 一起保存。回到相同权限范围时，
+没有新增可见答复就直接恢复原线程；有新增答复则通过原生分支继承压缩检查点，
+只注入这些答复。恢复使用的用户输入仍属于原请求，不作为新增历史。核验当前权限、
+来源、排队 Turn 和衔接内容后才发布子线程绑定。分支或注入结果未知时保留旧绑定
+与历史。无可验证记录的旧会话或权限变化沿用既有迁移路径。
+
 发现阶段使用本地已启用绑定和同一 typed 候选规划器，不为每条历史失败做网络身份
 核验。这只是队列提示；受理仍须在原 fence 内重新解析当前 App、本人、grant，
 核验确切原消息与原生凭证后，才能预留或排队任何续办。

loopx/canary/module_metric_baseline.json

@@ -34,20 +34,20 @@
       "dict_any_count": 0
     },
     "loopx/chat_agent.py": {
-      "lines": 2031
+      "lines": 2043
     },
     "loopx/chat_runtime.py": {
       "any_count": 38,
       "dict_any_count": 0,
-      "lines": 2439
+      "lines": 2475
     },
     "loopx/chat_server.py": {
       "any_count": 25,
       "dict_any_count": 0,
       "lines": 2046
     },
     "loopx/chat_store.py": {
-      "lines": 2113
+      "lines": 2141
     },
     "loopx/codex_cli_probe.py": {
       "any_count": 37,

loopx/capabilities/native_chat/codex_adapter.py

@@ -34,6 +34,17 @@
 _READ_ONLY_ITEM_TYPES = frozenset({"reasoning", "webSearch", "imageView", "contextCompaction"})
 
 
+def _history_digest(value: Any) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()
+
+
+@dataclass(frozen=True)
+class NativeWorkingContinuation:
+    thread_id: str
+    provider: str
+    suffix: list[dict[str, str]]
+
+
 @dataclass
 class NativeHistoryRestoration:
     """Live guards for unpublished model preparation, not a new work queue."""
@@ -46,10 +57,11 @@ class NativeHistoryRestoration:
     manager: bool
     invalidated: bool = False
     checked: bool = False
+    working_continuation: NativeWorkingContinuation | None = None
 
-    def capture(self) -> list[dict[str, Any]]:
+    def capture(self, *, strict: bool = False) -> list[dict[str, Any]]:
         from ...chat_manager import steward_machine_defaults
-        stored = self.controller.store.messages(self.session["session_id"])
+        stored = self.controller.store.messages(self.session["session_id"], **({"strict": True} if strict else {}))
         self.binding_id = (self.session.get("steward_context") or self.session.get("project_context") or {}).get("binding_id")
         self.binding = (next((row for row in self.controller.project_contexts.conversation_bindings.read()["bindings"]
             if row["binding_id"] == self.binding_id), None) if self.binding_id else None)
@@ -61,6 +73,93 @@ def messages(self, stored: list[dict[str, Any]]) -> list[dict[str, str]]:
             for item in stored if item.get("role") in {"user", "agent"}
             and (self.accepted_turn_id is None or item.get("turn_id") != self.accepted_turn_id)]
 
+    def working_affinity(self, *, work_dir: Path, codex_home: Path, runtime: Mapping[str, Any],
+                         account_epoch: tuple[str, str] | None) -> str:
+        """Bind continuation to the working host, source grant and model settings."""
+        from ..manager_runtime import manager_runtime_session_fields
+        fields = ("channel_id", "upstream_mode", "steward_context", "manager_context_version", "agent_id", "agent_model_id",
+                  "codex_chat_configuration", "requested_agent_model_id", "requested_agent_reasoning_effort")
+        return _history_digest([
+            {key: self.session.get(key) for key in fields},
+            str(work_dir.resolve()), str(codex_home.resolve()), self.binding, self.defaults, account_epoch,
+            manager_runtime_session_fields(runtime),
+        ])
+
+    def park_working_context(self, stored: list[dict[str, Any]], *, provider: str | None,
+                             affinity: str) -> dict[str, Any] | None:
+        """Keep the first working native identity across isolated reviews."""
+        if self.accepted_turn_id is None or self.binding is None:
+            return None
+        if self.session.get("answer_review") is True:
+            parked = self.session.get("parked_native_working")
+            if not isinstance(parked, dict) or parked.get("affinity_sha256") != affinity:
+                return None
+            reviews = parked.get("review_turn_ids") or []
+            if len(reviews) >= 32 or self.accepted_turn_id in reviews:
+                return None
+            return {**parked, "review_turn_ids": [*reviews, self.accepted_turn_id]}
+        if not provider:
+            return None
+        return {
+            "thread_id": self.session["upstream_thread_id"],
+            "provider": provider,
+            "affinity_sha256": affinity,
+            "prefix_count": len(stored),
+            "prefix_sha256": _history_digest(stored),
+            "review_turn_ids": [self.accepted_turn_id],
+        }
+
+    def _working_suffix(self, stored: list[dict[str, Any]], parked: Mapping[str, Any]) -> list[dict[str, str]] | None:
+        from ...chat_store import TERMINAL_TURN_STATES
+        count = parked["prefix_count"]
+        if count > len(stored) or _history_digest(stored[:count]) != parked["prefix_sha256"]:
+            return None
+        review_ids = set(parked["review_turn_ids"])
+        store, sid = self.controller.store, self.session["session_id"]
+        turns = {tid: store.load_turn(sid, tid) or {} for tid in review_ids}
+        if any(turn.get("status") not in TERMINAL_TURN_STATES for turn in turns.values()):
+            return None
+        suffix = []
+        seen: set[str] = set()
+        for index, message in enumerate(stored):
+            role, tid = message.get("role"), message.get("turn_id")
+            if role not in {"user", "agent"} or (self.accepted_turn_id is not None and tid == self.accepted_turn_id):
+                continue
+            if tid not in review_ids:
+                if index < count or role == "user" and tid and (store.load_turn(sid, tid) or {}).get("status") == "queued":
+                    continue
+                return None  # An unrelated visible result cannot be skipped.
+            turn = turns[tid]
+            if turn["status"] != "completed":
+                continue
+            # An isolated recovery's user message is a copy of the
+            # original request. Only its new, visible answer is a suffix.
+            if role == "user":
+                continue
+            text = message.get("text")
+            if not isinstance(text, str) or message.get("attachments") or tid in seen:
+                return None
+            if text != (turn.get("response") or {}).get("message"):
+                return None
+            seen.add(tid)
+            if text.strip():
+                suffix.append({"role": "assistant", "content": text})
+        if any(tid not in seen for tid, turn in turns.items()
+               if turn["status"] == "completed" and (turn.get("response") or {}).get("message")):
+            return None
+        return suffix
+
+    def restore_working_context(self, stored: list[dict[str, Any]], *, affinity: str) -> NativeWorkingContinuation | None:
+        parked = self.session.get("parked_native_working")
+        if (not isinstance(parked, dict) or self.binding is None or parked.get("affinity_sha256") != affinity
+                or parked.get("thread_id") == self.session.get("upstream_thread_id")):
+            return None
+        suffix = self._working_suffix(stored, parked)
+        if suffix is None:
+            return None
+        self.working_continuation = NativeWorkingContinuation(parked["thread_id"], parked["provider"], suffix)
+        return self.working_continuation
+
     def manager_migration(self, *, refresh: bool) -> bool:
         from ...chat_manager import MANAGER_CONTEXT_VERSION, is_manager_channel
         from ...chat_codex_goal import UPSTREAM_MODE
@@ -91,7 +190,7 @@ def check(self) -> None:
             latest = controller.store.load_session(sid)
             if (controller.closed.is_set() or latest is None or latest.get("status") == "closed"
                     or any(latest.get(key) != original.get(key) for key in
-                        ("active_turn_id", "upstream_thread_id", "channel_id", "agent_id", "codex_home",
+                        ("active_turn_id", "upstream_thread_id", "parked_native_working", "channel_id", "agent_id", "codex_home",
                          "project_context", "steward_context", "requested_agent_model_id", "requested_agent_reasoning_effort"))
                     or self.accepted_turn_id is not None and (controller.store.load_turn(sid, self.accepted_turn_id) or {}).get("status") != "queued"):
                 self.invalidated = True
@@ -103,6 +202,16 @@ def check(self) -> None:
             if self.defaults is not None and steward_machine_defaults(controller) != self.defaults:
                 self.invalidated = True
                 raise CodexChatAgentError("Steward configuration changed during history restoration", error_code="manager_authorization_unavailable", gate=None)
+            if self.working_continuation is not None:
+                if any(latest.get(key) != original.get(key) for key in (
+                    "answer_review", "manager_context_version", "agent_model_id", "codex_chat_configuration",
+                )):
+                    self.invalidated = True
+                    raise CodexChatAgentError("Working route changed during native continuation", error_code="resume_failed", gate=None)
+                messages = controller.store.messages(sid, strict=True)
+                if self._working_suffix(messages, original["parked_native_working"]) != self.working_continuation.suffix:
+                    self.invalidated = True
+                    raise CodexChatAgentError("Review history changed during native continuation", error_code="resume_failed", gate=None)
             self.route_check()
 
     def finish(self) -> None:
@@ -651,7 +760,15 @@ def start(
         account_runtime_root: Path | None = None,
         account_consumer: str = "worker",
         history_check: Callable[[], None] | None = None,
+        working_continuation: NativeWorkingContinuation | None = None,
     ) -> "CodexAppServerAdapter":
+        if working_continuation is not None:
+            if resume_thread_id or history is not None or history_check is None:
+                raise ValueError("native working continuation requires one guarded history source")
+            if working_continuation.suffix:
+                history = working_continuation.suffix
+            else:
+                resume_thread_id = working_continuation.thread_id
         history_items = []
         # Migration restores the full visible conversation in native history.
         # The objective is repeated per Turn; putting the transcript there
@@ -665,9 +782,12 @@ def start(
                     "type": "message", "role": role,
                     "content": [{"type": "output_text" if role == "assistant" else "input_text", "text": text}],
                 })
-        retry_timeout = bool(resume_thread_id)
+        retry_timeout = bool(resume_thread_id) and working_continuation is None
         while True:
             try:
+                if working_continuation is not None:
+                    assert history_check is not None
+                    history_check()
                 session = CodexChatAgentSession.start(
                     codex_bin=codex_bin,
                     work_dir=work_dir,
@@ -692,6 +812,10 @@ def start(
                     dynamic_tools=dynamic_tools,
                     account_runtime_root=account_runtime_root,
                     account_consumer=account_consumer,
+                    **({
+                        "expected_native_provider": working_continuation.provider,
+                        **({"fork_thread_id": working_continuation.thread_id} if working_continuation.suffix else {}),
+                    } if working_continuation is not None else {}),
                 )
                 if not resume_thread_id:
                     try:
@@ -706,7 +830,7 @@ def start(
                         restore_deadline = time.monotonic() + hard_timeout_sec if hard_timeout_sec is not None else None
 
                         def check_restore() -> None:
-                            if history_check is not None and len(batches) > 1:
+                            if history_check is not None and (working_continuation is not None or len(batches) > 1):
                                 history_check()
                             if restore_deadline is not None and time.monotonic() >= restore_deadline:
                                 raise CodexChatTimeoutError("Native history restoration exceeded its Turn time budget.", error_code="response_timeout", gate=None)
@@ -797,7 +921,8 @@ def start_runtime_adapter(controller: ChatRuntimeController, *, work_dir: Path,
         project_coordination: bool, loopx_tools: bool, codex_home_override: Path | None,
         read_only_review: bool, metadata_review_context: dict[str, object] | None,
         answer_review_context: dict[str, object] | None,
-        history_check: Callable[[], None] | None = None) -> CodexAppServerAdapter:
+        history_check: Callable[[], None] | None = None,
+        working_continuation: NativeWorkingContinuation | None = None) -> CodexAppServerAdapter:
     """Build the native adapter from already-authorized runtime arguments."""
     from ...chat_manager import MANAGER_AGENT_GOAL_ID
     from ..manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL
@@ -815,6 +940,7 @@ def start_runtime_adapter(controller: ChatRuntimeController, *, work_dir: Path,
             "runtime_profile": "restricted"} if review_context else native_context),
         history=history,
         history_check=history_check,
+        working_continuation=working_continuation,
         resume_thread_id=resume_thread_id,
         startup_timeout_sec=controller.startup_timeout_sec,
         idle_timeout_sec=controller.idle_timeout_sec,