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 e14e8b557d..927c48b270 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/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
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 c5c8bfbe8b..47ea747e5e 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
@@ -1117,6 +1117,13 @@ Session 的关闭先发布既有存储状态转换，再等待 adapter 清理；
 也不能让已关闭的会话重新打开。
 所有批次共用已配置的 Turn 硬时限，不为每次压缩续期。
 
+已绑定的 Codex 管家进入隔离答复复核前，Core 核验当前工作 adapter 空闲，
+将原生线程、provider 和来源权限依据与复核 Turn 一起保存。回到相同权限范围时，
+没有新增可见答复就直接恢复原线程；有新增答复则通过原生分支继承压缩检查点，
+只注入这些答复。恢复使用的用户输入仍属于原请求，不作为新增历史。核验当前权限、
+来源、排队 Turn 和衔接内容后才发布子线程绑定。分支或注入结果未知时保留旧绑定
+与历史。无可验证记录的旧会话或权限变化沿用既有迁移路径。
+
 发现阶段使用本地已启用绑定和同一 typed 候选规划器，不为每条历史失败做网络身份
 核验。这只是队列提示；受理仍须在原 fence 内重新解析当前 App、本人、grant，
 核验确切原消息与原生凭证后，才能预留或排队任何续办。
diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json
index a50ecdb3e2..cb12b2d3d3 100644
--- a/loopx/canary/module_metric_baseline.json
+++ b/loopx/canary/module_metric_baseline.json
@@ -34,12 +34,12 @@
       "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,
@@ -47,7 +47,7 @@
       "lines": 2046
     },
     "loopx/chat_store.py": {
-      "lines": 2113
+      "lines": 2141
     },
     "loopx/codex_cli_probe.py": {
       "any_count": 37,
diff --git a/loopx/capabilities/native_chat/codex_adapter.py b/loopx/capabilities/native_chat/codex_adapter.py
index 72f33eff03..17f72058b3 100644
--- a/loopx/capabilities/native_chat/codex_adapter.py
+++ b/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,
diff --git a/loopx/chat_agent.py b/loopx/chat_agent.py
index bf734b39f6..8acfe64e8a 100644
--- a/loopx/chat_agent.py
+++ b/loopx/chat_agent.py
@@ -680,6 +680,7 @@ class CodexChatAgentSession:
     project_context: dict[str, str] | None = None
     model: str | None = None
     resolved_model: str | None = None
+    native_model_provider: str | None = None
     reasoning_effort: str | None = None
     resolved_reasoning_effort: str | None = None
     service_tier: str | None = None
@@ -733,6 +734,8 @@ def start(
         idle_timeout_sec: float = 180.0,
         hard_timeout_sec: float | None = None,
         resume_thread_id: str | None = None,
+        fork_thread_id: str | None = None,
+        expected_native_provider: str | None = None,
         execution_mode: bool = False,
         isolate_process_tree: bool = False,
         runtime_profile: str = "restricted",
@@ -764,12 +767,18 @@ def start(
         root = work_dir.resolve()
         if runtime_profile not in {"restricted", "trusted_owner"}:
             raise ValueError("unsupported Codex Chat runtime profile")
+        if fork_thread_id and (
+            resume_thread_id or execution_mode or read_only_review or project_context
+            or metadata_review_context or answer_review_context or not expected_native_provider
+        ):
+            raise ValueError("Codex history fork requires an authorized ordinary native thread")
+        existing_thread_id = resume_thread_id or fork_thread_id
         from .capabilities.native_chat import codex_context
         policy, selected_sandbox = codex_context.native_conversation_policy(
             root=root, metadata_review_context=metadata_review_context, answer_review_context=answer_review_context,
             answer_review_host_read=answer_review_host_read, read_only_review=read_only_review,
             project_context=project_context, goal_id=goal_id, execution_mode=execution_mode,
-            runtime_profile=runtime_profile, dynamic_tools=dynamic_tools, resume_thread_id=resume_thread_id,
+            runtime_profile=runtime_profile, dynamic_tools=dynamic_tools, resume_thread_id=existing_thread_id,
             sandbox=sandbox)
         # Core owns the optional narrower project filesystem policy. A model,
         # transport or project config cannot substitute a profile of its choice.
@@ -925,7 +934,7 @@ def start(
             # the account owner has switched back to a subscription (or vice
             # versa). Follow Codex's effective route without selecting accounts,
             # copying auth, restarting clients or replacing the original thread.
-            follow_native_provider = bool(resume_thread_id and not execution_mode
+            follow_native_provider = bool(existing_thread_id and not execution_mode
                 and (not permissions_profile or metadata_review_context is not None) and not provider_override)
             read_native_config = read_project_defaults or bool(permissions_profile) or owner_git_write or follow_native_provider or api_transport is not None
             if read_native_config:
@@ -989,12 +998,15 @@ def start(
                 if any(value is not None and (not isinstance(value, str) or not value.strip())
                        for value in (model, reasoning_effort)):
                     raise session._runtime_error("Codex returned invalid project model settings.")
+            if expected_native_provider and provider_override and provider_override != expected_native_provider:
+                raise session._runtime_error("Codex native history belongs to a different model provider.")
+            thread_method = "thread/fork" if fork_thread_id else "thread/resume" if resume_thread_id else "thread/start"
             thread_result = session._request(
-                "thread/resume" if resume_thread_id else "thread/start",
+                thread_method,
                 {
                     **(
-                        {"threadId": resume_thread_id, "excludeTurns": True}
-                        if resume_thread_id
+                        {"threadId": existing_thread_id, "excludeTurns": True}
+                        if existing_thread_id
                         else {}
                     ),
                     "cwd": str(root),
@@ -1023,13 +1035,17 @@ def start(
                     "approvalPolicy": "never",
                     **(
                         {"dynamicTools": dynamic_tools}
-                        if dynamic_tools and not resume_thread_id
+                        if dynamic_tools and not existing_thread_id
                         else {}
                     ),
                 },
                 request_id=2,
             )
             codex_context.verify_conversation_provider(session, provider_override, thread_result)
+            native_provider = thread_result.get("modelProvider")
+            session.native_model_provider = native_provider if isinstance(native_provider, str) and native_provider else None
+            if expected_native_provider is not None and session.native_model_provider != expected_native_provider:
+                raise session._runtime_error("Codex native history belongs to a different model provider.")
             if model and thread_result.get("model") not in {None, model}:
                 raise CodexChatAgentError(
                     "Codex did not apply the requested conversation model.",
@@ -1077,6 +1093,20 @@ def start(
                 raise session._runtime_error(
                     "Codex app-server resumed an unexpected thread."
                 )
+            if fork_thread_id:
+                native_thread = thread_result.get("thread")
+                sandbox_type = {
+                    "read-only": "readOnly",
+                    "workspace-write": "workspaceWrite",
+                    "danger-full-access": "dangerFullAccess",
+                }.get(selected_sandbox)
+                applied_sandbox = thread_result.get("sandbox")
+                if (session.thread_id == fork_thread_id or not isinstance(native_thread, dict)
+                        or native_thread.get("forkedFromId") != fork_thread_id
+                        or thread_result.get("cwd") != str(root) or thread_result.get("approvalPolicy") != "never"
+                        or sandbox_type is None or not isinstance(applied_sandbox, dict)
+                        or applied_sandbox.get("type") != sandbox_type):
+                    raise session._runtime_error("Codex did not verify the original thread and policy of its history fork.")
             if permissions_profile:
                 session._public_source_read_context = codex_context.public_source_read_context(host_config or {})
             if read_only_review and session._background_terminals(time.monotonic() + response_timeout_sec):
@@ -1093,7 +1123,7 @@ def start(
             return session
         except _LegacyModelCatalogSchemaError as exc:
             session.close()
-            if _compatibility_catalog_path is not None:
+            if _compatibility_catalog_path is not None or fork_thread_id:
                 raise _model_catalog_compatibility_error() from exc
             with _current_builtin_model_catalog(resolved, environment=runtime_env) as catalog_path:
                 return cls.start(
@@ -1105,6 +1135,7 @@ def start(
                     idle_timeout_sec=idle_timeout_sec,
                     hard_timeout_sec=hard_timeout_sec,
                     resume_thread_id=resume_thread_id,
+                    expected_native_provider=expected_native_provider,
                     execution_mode=execution_mode,
                     isolate_process_tree=isolate_process_tree,
                     runtime_profile=runtime_profile,
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index eb4b9d69a0..bb4724f066 100644
--- a/loopx/chat_runtime.py
+++ b/loopx/chat_runtime.py
@@ -24,7 +24,9 @@
 )
 from .chat.coordination import PROJECT_COORDINATION_GUIDANCE, PROJECT_CONTEXT_VERSION, apply_context_handoff
 from .capabilities.native_chat import codex_context, project_context as project_context_policy
-from .capabilities.native_chat.codex_adapter import CodexAppServerAdapter, NativeHistoryRestoration, start_attachment_turn
+from .capabilities.native_chat.codex_adapter import (
+    CodexAppServerAdapter, NativeHistoryRestoration, NativeWorkingContinuation, start_attachment_turn,
+)
 from .capabilities.native_chat.machine_defaults import codex_profile_home, load_codex_chat_defaults
 from .capabilities.native_chat.steward_managed_mode import prepare_managed_session_route, require_managed_agent_session
 from .capabilities.native_chat.turn_release import TurnReleaseWakeup
@@ -365,6 +367,7 @@ def _start_adapter(
         metadata_review_context: dict[str, Any] | None = None,
         answer_review_context: dict[str, Any] | None = None,
         history_check: Callable[[], None] | None = None,
+        working_continuation: NativeWorkingContinuation | None = None,
     ) -> ChatRuntimeAdapter:
         if project_context is not None:
             project_context_policy.validate_project_native_executor(agent_id, project_context)
@@ -422,7 +425,7 @@ def _start_adapter(
                 project_coordination=project_coordination, loopx_tools=loopx_tools,
                 codex_home_override=codex_home_override, read_only_review=read_only_review,
                 metadata_review_context=metadata_review_context, answer_review_context=answer_review_context,
-                history_check=history_check)
+                history_check=history_check, working_continuation=working_continuation)
         if agent_id == "claude-code":
             return ClaudeCodeAdapter.start(
                 claude_bin=self.claude_bin,
@@ -841,9 +844,11 @@ def check_route_configuration():
             check_route_configuration()
         selected_agent = model_allocation["executor_endpoint"] if quota_route else session["agent_id"]
         native_account_epoch = None
-        if (selected_agent == "codex" and active and active_id == accepted_turn_id
-                and active.get("status") == "queued" and not active.get("upstream_turn_id")
-                and active.get("origin") != "native_loopx"):
+        if (selected_agent == "codex" and (
+            active and active_id == accepted_turn_id and active.get("status") == "queued"
+            and not active.get("upstream_turn_id") and active.get("origin") != "native_loopx"
+            or not active and session.get("parked_native_working")
+        )):
             from .capabilities.native_chat.codex_auth import conversation_account_epoch
             native_account_epoch = conversation_account_epoch(self.coordination_runtime_root, selected_codex_home,
                 **({"consumer": "steward", "model": (model_allocation or {}).get("model") or session.get("agent_model_id")}
@@ -853,6 +858,7 @@ def check_route_configuration():
             and active.get("loopx_execution") is True
             and (active.get("loopx_request") or {}).get("work_kind") == "report"
             and isinstance((active.get("loopx_request") or {}).get("report_quota_recovery"), dict))
+        working_provider = None
         with self.lock:
             current = self.adapters.get(session_id)
             manager_profile_changed = NativeHistoryRestoration.manager_profile_changed(session, manager_runtime)
@@ -862,6 +868,7 @@ def check_route_configuration():
                 and not manager_profile_changed
                 and not recovery.get("history_review")
                 and not recovery.get("quota_attempt")
+                and not (answer_fallback and session.get("parked_native_working"))
                 and not native_quota_refresh
                 and (native_account_epoch is None or getattr(current, "_native_account_epoch", None) == native_account_epoch)
                 and (isinstance(current, CodexAppServerAdapter) and current.session.read_only_review) == bool(recovery.get("read_only_review"))
@@ -874,12 +881,17 @@ def check_route_configuration():
                 # transport cannot prove that its original Chat worker exited.
                 raise ValueError("the original Chat worker is still active; read its current Turn")
             elif current is not None:
+                may_park_working = bool(answer_fallback and isinstance(current, CodexAppServerAdapter)
+                    and not current.session.read_only_review
+                    and current.upstream_thread_id == session.get("upstream_thread_id"))
                 if (recovery.get("history_review") is True or recovery.get("read_only_review")
-                        and recovery.get("unfinished_work")) and current.healthcheck():
+                        and recovery.get("unfinished_work") or may_park_working) and current.healthcheck():
                     if (not isinstance(current, CodexAppServerAdapter)
                             or not current.answer_review_replaceable()):
                         raise CodexChatAgentError("The original adapter still owns background work; answer review held.",
                             error_code="host_gate", gate=None)
+                    if may_park_working and getattr(current, "_native_account_epoch", None) == native_account_epoch:
+                        working_provider = current.session.native_model_provider
                 current.close_session()
                 self.adapters.pop(session_id, None)
         if reusable is not None:
@@ -927,13 +939,33 @@ def check_route_configuration():
                             payload={"error_code": "server_restarted", "message": failed["error"]},
                         )
         native_recovery = None
+        parked_working = None
+        working_continuation = None
         history_restore = NativeHistoryRestoration(self, session, accepted_turn_id, commit, check_route_configuration, manager_runtime is not None)
         try:
-            stored_messages = history_restore.capture()
+            same_working_route = (manager_runtime is not None and session.get("steward_context") is not None
+                and selected_agent == session["agent_id"] == "codex"
+                and session.get("manager_context_version") == MANAGER_CONTEXT_VERSION
+                and session.get("upstream_mode") == "chat" and not manager_profile_changed
+                and not settings_changed and not quota_route and model_allocation is None and not quota_fallback)
+            stored_messages = history_restore.capture(strict=bool(same_working_route
+                and (working_provider or session.get("parked_native_working"))))
             history = history_restore.messages(stored_messages)
             legacy_manager_context = history_restore.manager_migration(refresh=(manager_profile_changed
                 or model_allocation is not None or quota_fallback or answer_fallback))
             legacy_provider_context = history_restore.provider_migration(stored_messages)
+            if (same_working_route and (working_provider or session.get("parked_native_working"))
+                    and not legacy_provider_context and not legacy_project_context):
+                affinity = history_restore.working_affinity(
+                    work_dir=work_dir, codex_home=selected_codex_home, runtime=manager_runtime,
+                    account_epoch=native_account_epoch)
+                if answer_fallback:
+                    parked_working = history_restore.park_working_context(
+                        stored_messages, provider=working_provider, affinity=affinity)
+                    if parked_working is not None:
+                        history_restore.check()
+                elif session.get("answer_review") is True and not recovery.get("read_only_review"):
+                    working_continuation = history_restore.restore_working_context(stored_messages, affinity=affinity)
             adapter = self._start_adapter(
                 agent_id=str(selected_agent),
                 work_dir=work_dir,
@@ -945,7 +977,7 @@ def check_route_configuration():
                 objective=objective,
                 resume_thread_id=(
                     None
-                    if legacy_provider_context or legacy_manager_context or legacy_project_context
+                    if working_continuation or legacy_provider_context or legacy_manager_context or legacy_project_context
                     else str(session["upstream_thread_id"])
                 ),
                 history=(
@@ -953,7 +985,7 @@ def check_route_configuration():
                     # request and verified results at dispatch. Importing the
                     # whole conversation here needlessly compacts unrelated
                     # history before an unanswered question can be reviewed.
-                    None if answer_fallback else history
+                    None if answer_fallback or working_continuation else history
                     if legacy_provider_context or legacy_manager_context or legacy_project_context
                     or session.get("agent_id")
                     in {"anthropic-api", "openai-api", MANAGED_TURN_HOST}
@@ -976,6 +1008,7 @@ def check_route_configuration():
                 metadata_review_context=session.get("steward_context") if metadata_fallback else None,
                 answer_review_context=session.get("steward_context") if answer_fallback else None,
                 history_check=history_restore.check,
+                **({"working_continuation": working_continuation} if working_continuation is not None else {}),
             )
             try:
                 # Recovery reads back the provider-restored configuration.
@@ -1037,6 +1070,9 @@ def check_route_configuration():
                     }
                     if answer_fallback or session.get("answer_review") is True:
                         changes["answer_review"] = answer_fallback
+                    if parked_working is not None or session.get("parked_native_working") is not None:
+                        changes.update(parked_native_working=parked_working,
+                            expected_upstream_thread_id=session["upstream_thread_id"])
                     if legacy_manager_context and adapter.upstream_thread_id != session.get("upstream_thread_id"):
                         # Preserve provenance for read-only checks of failed
                         # native requests after this history migration. Publish
diff --git a/loopx/chat_store.py b/loopx/chat_store.py
index 8ec4fd4bf3..884cdc7603 100644
--- a/loopx/chat_store.py
+++ b/loopx/chat_store.py
@@ -488,7 +488,7 @@ def update_session(self, session_id: str, *, expected_statuses: set[str] | None
                     "requested_agent_model_id", "requested_agent_reasoning_effort",
                     "goal_id",
                     "agent_id", "executor_endpoint_id", "adapter_kind", "manager_quota_block",
-                    "answer_review",
+                    "answer_review", "parked_native_working",
                 }
                 unknown = set(changes) - allowed
                 if unknown:
@@ -496,6 +496,35 @@ def update_session(self, session_id: str, *, expected_statuses: set[str] | None
                 if "answer_review" in changes and (type(changes["answer_review"]) is not bool
                         or changes["answer_review"] and payload.get("steward_context") is None):
                     raise ValueError("answer review requires a bound steward and a boolean adapter marker")
+                if "parked_native_working" in changes:
+                    parked = changes["parked_native_working"]
+                    prior = payload.get("parked_native_working")
+                    if (expected_upstream_thread_id is None or "answer_review" not in changes
+                            or payload.get("steward_context") is None or payload.get("agent_id") != "codex"):
+                        raise ValueError("working native history must be bound to a steward review transition")
+                    if parked is not None:
+                        if not isinstance(parked, dict) or set(parked) != {
+                            "thread_id", "provider", "affinity_sha256", "prefix_count", "prefix_sha256", "review_turn_ids",
+                        }:
+                            raise ValueError("invalid parked working history")
+                        review_ids = parked["review_turn_ids"]
+                        active = self.load_turn(session_id, payload["active_turn_id"]) if payload.get("active_turn_id") else None
+                        if (changes["answer_review"] is not True or not active or active.get("status") != "queued"
+                                or active.get("upstream_turn_id")
+                                or changes.get("upstream_thread_id") in {None, expected_upstream_thread_id}
+                                or not isinstance(review_ids, list) or not 1 <= len(review_ids) <= 32
+                                or any(not isinstance(tid, str) for tid in review_ids)
+                                or len(set(review_ids)) != len(review_ids) or review_ids[-1] != active["turn_id"]
+                                or not isinstance(parked["provider"], str) or not 1 <= len(parked["provider"]) <= 128
+                                or type(parked["prefix_count"]) is not int or parked["prefix_count"] < 0
+                                or any(not isinstance(parked[key], str) or not re.fullmatch(r"[0-9a-f]{64}", parked[key])
+                                       for key in ("affinity_sha256", "prefix_sha256"))):
+                            raise ValueError("invalid parked working history or review Turn")
+                        if payload.get("answer_review") is True:
+                            if (not isinstance(prior, dict) or {**parked, "review_turn_ids": review_ids[:-1]} != prior):
+                                raise ValueError("a later review cannot replace the original working history")
+                        elif parked["thread_id"] != expected_upstream_thread_id or review_ids != [active["turn_id"]]:
+                            raise ValueError("parked history must name the original working thread")
                 if "manager_quota_block" in changes and "agent_id" not in changes:
                     raise ValueError("quota observation belongs to an executor transition")
                 if any(key in changes for key in ("agent_id", "executor_endpoint_id", "adapter_kind")):
diff --git a/tests/test_steward_interrupted_answer.py b/tests/test_steward_interrupted_answer.py
index ce3e6ef480..fad5da8cd6 100644
--- a/tests/test_steward_interrupted_answer.py
+++ b/tests/test_steward_interrupted_answer.py
@@ -16,7 +16,9 @@
 from loopx.extensions.lark.conversation_identity import identity_ref
 
 
-@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "unknown-native", "no-capacity", "native-changed", "review-action", "post-enqueue-crash", "image-view"])
+@pytest.mark.parametrize("fault", [None, "stop", "source-changed", "unknown-native", "no-capacity", "native-changed",
+    "review-action", "post-enqueue-crash", "image-view", "checkpoint-live", "checkpoint-cold",
+    "checkpoint-injection-unknown", "checkpoint-empty", "checkpoint-settings-change"])
 def test_restarted_codex_steward_gets_one_original_read_only_answer(steward, monkeypatch, fault):
     from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
     from loopx.chat_agent import CodexChatAgentError
@@ -24,11 +26,36 @@ def test_restarted_codex_steward_gets_one_original_read_only_answer(steward, mon
 
     store, runtime, provider, transport, binding, capture, workspace = steward
     fake = Path(runtime.codex_bin)
+    checkpoint = bool(fault and fault.startswith("checkpoint-"))
+    original_thread = "working-thread" if checkpoint else "durable-thread"
     code = fake.read_text().replace('result = {"thread": {"id": "durable-thread"}}', '''result = {"thread": {"id": "durable-thread"}, "model": "gpt-6.1-sol"}
         permissions = request.get("params", {}).get("permissions")
         if permissions:
             result.update(activePermissionProfile={"id": permissions}, runtimeWorkspaceRoots=[request["params"]["cwd"]])''')
-    if fault == "review-action":
+    if checkpoint:
+        code = code.replace('"durable-thread"', 'thread_id')
+        code = code.replace('active_turn = None', 'active_turn = None\nthread_id = "working-thread"', 1)
+        begin = code.index('    elif method in {"thread/start", "thread/resume"}:')
+        end = code.index('    elif method == "thread/goal/get":', begin)
+        code = code[:begin] + '''    elif method in {"thread/start", "thread/resume", "thread/fork"}:
+        params = request["params"]
+        permissions = params.get("permissions")
+        thread_id = ("continued-thread" if method == "thread/fork" else params.get("threadId")
+            if method == "thread/resume" else "review-thread" if permissions else "working-thread")
+        result = {"thread": {"id": thread_id}, "model": "gpt-6.1-sol", "modelProvider": "openai",
+                  "cwd": params["cwd"], "approvalPolicy": "never"}
+        if method == "thread/fork":
+            result["thread"]["forkedFromId"] = params["threadId"]
+            result["sandbox"] = {"type": {"danger-full-access": "dangerFullAccess",
+                "workspace-write": "workspaceWrite", "read-only": "readOnly"}[params["sandbox"]]}
+        if permissions:
+            result.update(activePermissionProfile={"id": permissions}, runtimeWorkspaceRoots=[params["cwd"]])
+''' + code[end:]
+        if fault == "checkpoint-injection-unknown":
+            code = code.replace('    elif method == "turn/start":', '''    elif method == "thread/inject_items" and thread_id == "continued-thread":
+        sys.exit(2)
+    elif method == "turn/start":''')
+    if fault in {"review-action", "checkpoint-empty"}:
         start = code.index("        response = 'Visible runtime response.")
         end = code.index('        print(json.dumps({"method": "item/agentMessage/delta"', start)
         reply = '<loopx-review-json>' + json.dumps({"message": "Forbidden repeat",
@@ -67,6 +94,8 @@ def observed(kind, payload):
     def start_scoped_review(**kwargs):
         if kwargs.get("answer_review_context") is not None:
             assert kwargs["history"] is None, "a restricted answer must not import the whole conversation"
+        if fault == "checkpoint-settings-change" and kwargs.get("working_continuation") is not None:
+            store.update_session(original["session_id"], requested_agent_model_id="gpt-6-luna")
         return start_adapter(**kwargs)
     monkeypatch.setattr(runtime, "_start_adapter", start_scoped_review)
     class Ahead(datetime):
@@ -106,7 +135,7 @@ def items(method, params, _deadline):
             return interrupted_result_readback(SimpleNamespace(thread_id="durable-thread",
                 response_timeout_sec=1, _control_items=items), failed["upstream_turn_id"])
         changed = fault == "native-changed" and bool((transport.core.read_request(ref).get("context_recovery") or {}).get("answer_review"))
-        return {"thread_id": "durable-thread", "turn_id": failed["upstream_turn_id"],
+        return {"thread_id": original_thread, "turn_id": failed["upstream_turn_id"],
             "read_only_review": True, "no_final_answer": True, "effects_observed": True,
             "native_inventory_sha256": ("2" if changed else "1") * 64}
     monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(observe))
@@ -134,7 +163,7 @@ def crash(**kwargs):
         assert recovery["answer_review"] is True and recovery["read_only_review"] is True
         assert not any(recovery.get(key) for key in ["work_attempt", "quota_attempt", "routing_attempt"])
         done = runtime.wait_for_turn(session_id=current["session_id"], turn_id=current["turn_id"], timeout_sec=15)
-        assert done["status"] == ("failed" if fault in {"native-changed", "review-action"} else "completed"), done
+        assert done["status"] == ("failed" if fault in {"native-changed", "review-action", "checkpoint-empty"} else "completed"), done
         restarted.reconcile_request(path)
         delivered = json.loads(path.read_text())
         terminal = delivered["deliveries"]["terminal.recovery.22.review.1"]
@@ -160,6 +189,48 @@ def crash(**kwargs):
         writes = list(provider.writes)
         restarted.reconcile_request(path)
         assert provider.writes == writes
+        if checkpoint:
+            sid = original["session_id"]
+            review_session = store.load_session(sid)
+            parked = review_session["parked_native_working"]
+            assert parked["thread_id"] == "working-thread" and parked["review_turn_ids"] == [done["turn_id"]]
+            assert review_session["upstream_thread_id"] == "review-thread"
+            if fault == "checkpoint-cold":
+                runtime.adapters.pop(sid).close_session()
+                assert store.load_session(sid)["parked_native_working"] == parked
+            monkeypatch.setattr(CodexAppServerAdapter, "start_turn", start_turn)
+            later = provider.event("steward-app", "after-working-checkpoint", "Next independent question")
+            restarted.admit("steward-app", later)
+            next_row = restarted.core.read_request(identity_ref(binding["provider_ref"], later["message_id"]))
+            next_turn = runtime.wait_for_turn(session_id=sid, turn_id=next_row["turn_id"], timeout_sec=15)
+            packets = [json.loads(line) for line in capture.read_text().splitlines()][len(calls):]
+            methods = [packet["method"] for packet in packets]
+            assert "thread/start" not in methods and "thread/compact/start" not in methods
+            if fault == "checkpoint-settings-change":
+                assert next_turn["status"] == "failed"
+                assert not {"thread/fork", "thread/resume", "thread/inject_items", "turn/start"}.intersection(methods)
+                assert store.load_session(sid)["parked_native_working"] == parked
+                assert store.load_session(sid)["upstream_thread_id"] == "review-thread"
+            elif fault == "checkpoint-empty":
+                assert next_turn["status"] == "completed"
+                assert [p["params"]["threadId"] for p in packets if p["method"] == "thread/resume"] == ["working-thread"]
+                assert "thread/inject_items" not in methods
+            else:
+                assert [p["params"]["threadId"] for p in packets if p["method"] == "thread/fork"] == ["working-thread"]
+                injected = [p["params"] for p in packets if p["method"] == "thread/inject_items"]
+                assert len(injected) == 1 and injected[0]["threadId"] == "continued-thread"
+                suffix = [(item["role"], item["content"][0]["text"]) for item in injected[0]["items"]]
+                assert suffix == [("assistant", done["response"]["message"])]
+                assert "Next independent question" not in str(suffix)
+                if fault == "checkpoint-injection-unknown":
+                    assert next_turn["status"] == "failed" and "turn/start" not in methods
+                    assert store.load_session(sid)["parked_native_working"] == parked
+                    assert store.load_session(sid)["upstream_thread_id"] == "review-thread"
+                else:
+                    assert next_turn["status"] == "completed"
+                    assert store.load_session(sid)["upstream_thread_id"] == "continued-thread"
+            if fault not in {"checkpoint-injection-unknown", "checkpoint-settings-change"}:
+                assert store.load_session(sid).get("parked_native_working") is None
     assert store._turn_path(original["session_id"], original["turn_id"]).read_bytes() == failure_bytes
     assert artifact.read_text() == "Original effect, do not repeat"