fix(chat): keep answer recovery scoped and resume unstarted claims

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

@@ -1835,6 +1835,15 @@ adopts the reserved review instead of granting another attempt. Missing native
 evidence or capacity remains pending. The failure notice describes this qualified
 recovery instead of giving every restart the same blanket no-replay message.
 
+This isolated answer review receives its complete original question and verified
+visible results at dispatch. It does not import or serially compact the unrelated
+conversation before it can answer. Core retains the full transcript, and ordinary
+conversation restoration continues to preserve that history.
+If restart happens after the queue claim but before native dispatch, the worker
+adopts that same unstarted Turn through ordinary preparation and execution
+fences. It does not wait forever for a queued Turn to become running by itself;
+unfinished managed acceptance and already-started work retain their own owners.
+
 ### Explicit unfinished work after a completed Turn
 
 A terminal model Turn is distinct from completed business work. A private

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

@@ -1198,6 +1198,12 @@ provider 当前正文等于已知前文或目标正文。正文改变、效果
 与失败通知不变；重启或入队回执不明时采用已预留复核，不产生新次数。缺少原生证据
 或额度仍保留待办。失败提示说明这些恢复条件，不再对所有重启一律提示不会自动重放。
 
+这次独立答案复核在派发时收到完整原问题及已核验的可见结果，不在答复前导入、串行
+压缩无关的整段对话。Core 完整聊天记录保留；普通对话的历史恢复逻辑继续保留全部历史。
+若重启发生在取得队列 claim 后、原生派发前，worker 沿正常准备与执行屏障采用同一
+未启动 Turn，不空等 queued 自己变成 running；尚未发布的受管接收和已启动工作仍由
+各自原 owner 处理。
+
 ### Turn 完成后的显式未完工作
 
 模型 Turn 终态与业务完成分开。私有项目 worker 核验已结束工作的实际结果后，可在

loopx/canary/module_metric_baseline.json

@@ -39,7 +39,7 @@
     "loopx/chat_runtime.py": {
       "any_count": 38,
       "dict_any_count": 0,
-      "lines": 2419
+      "lines": 2439
     },
     "loopx/chat_server.py": {
       "any_count": 25,

loopx/chat_runtime.py

@@ -949,7 +949,11 @@ def check_route_configuration():
                     else str(session["upstream_thread_id"])
                 ),
                 history=(
-                    history
+                    # This isolated review receives the complete original
+                    # 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
                     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}
@@ -1497,6 +1501,7 @@ def _drain_session_queue(
                     or session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED
                 ):
                     return
+                turn = None
                 active_turn_id = str(session.get("active_turn_id") or "")
                 if active_turn_id:
                     try:
@@ -1506,30 +1511,40 @@ def _drain_session_queue(
                         # non-executing lifecycle repair, never the old input.
                         self.closed.wait(0.25)
                         continue
-                    with self.lock:
-                        attached = self.adapters.get(session_id)
-                    if attached is None or not attached.healthcheck():
-                        try:
-                            self._ensure_adapter(
-                                session,
-                                work_dir=work_dir,
-                                objective=objective,
-                            )
-                        except Exception as exc:
-                            self._fail_queue_preparation(session_id, active_turn_id, exc)
-                        else:
-                            continue
                     active = self.store.load_turn(session_id, active_turn_id)
-                    if active and active.get("status") not in TERMINAL_TURN_STATES:
-                        self.closed.wait(0.05)
-                        continue
-                    self.store.release_active_turn(session_id, active_turn_id)
+                    if (active and active.get("status") == "queued" and "_acceptance" not in active
+                            and active.get("started_at") is None and active.get("upstream_turn_id") is None
+                            and active.get("delta_count") == 0):
+                        # A crash can leave the queue claim before native
+                        # dispatch. Keep that exact claim and pass through the
+                        # normal preparation and execution fences below.
+                        turn = active
+                    else:
+                        with self.lock:
+                            attached = self.adapters.get(session_id)
+                        if attached is None or not attached.healthcheck():
+                            try:
+                                self._ensure_adapter(
+                                    session,
+                                    work_dir=work_dir,
+                                    objective=objective,
+                                )
+                            except Exception as exc:
+                                self._fail_queue_preparation(session_id, active_turn_id, exc)
+                            else:
+                                continue
+                        active = self.store.load_turn(session_id, active_turn_id)
+                        if active and active.get("status") not in TERMINAL_TURN_STATES:
+                            self.closed.wait(0.05)
+                            continue
+                        self.store.release_active_turn(session_id, active_turn_id)
                 refreshed = self.store.load_session(session_id)
                 if refreshed is None:
                     return
                 with self.lock:
                     self.session_queue_wakeups.discard(session_id)
-                turn = self.store.claim_next_queued_turn(session_id)
+                if turn is None:
+                    turn = self.store.claim_next_queued_turn(session_id)
                 if turn is None:
                     with self.lock:
                         if session_id in self.session_queue_wakeups:
@@ -1547,11 +1562,16 @@ def _drain_session_queue(
                     # An empty queue needs no provider preparation. Claim real
                     # work first, keeping the existing enqueue/retirement fence
                     # above and the Turn's execution/stop checks below.
+                    current = self.store.load_session(session_id)
+                    if current is None or current.get("active_turn_id") != turn_id:
+                        return
                     adapter = self._ensure_adapter(
-                        self.store.load_session(session_id) or refreshed,
+                        current,
                         work_dir=work_dir,
                         objective=objective,
                     )
+                    if (self.store.load_session(session_id) or {}).get("active_turn_id") != turn_id:
+                        return
                 except Exception as exc:
                     # The claimed request still owns a terminal outcome when
                     # workspace preparation or adapter restoration fails.

tests/test_chat_runtime_preparation_recovery.py

@@ -11,6 +11,62 @@
 from loopx.chat_store import _read_json
 
 
+@pytest.mark.parametrize("prior_dispatch", [None, "native-id", "started"])
+def test_restart_adopts_claimed_unstarted_queue_without_duplicate_dispatch(tmp_path, monkeypatch, prior_dispatch):
+    import queue
+    import threading
+    from types import SimpleNamespace
+    from loopx.chat_agent import CodexChatAgentSession
+    from loopx.chat_runtime import ChatRuntimeController, CodexAppServerAdapter
+    from loopx.chat_store import ChatSessionStore
+
+    store, sid, original_runtime = session_runtime(tmp_path)
+    original_runtime.close()
+    original, _ = store.create_queued_turn(sid, client_turn_id="claimed-before-crash",
+        message="The same original input", origin="web")
+    assert store.claim_next_queued_turn(sid)["turn_id"] == original["turn_id"]
+    assert store.load_turn(sid, original["turn_id"])["started_at"] is None
+    if prior_dispatch:
+        store.update_turn(sid, original["turn_id"], **(
+            {"upstream_turn_id": "unknown-native-dispatch"} if prior_dispatch == "native-id"
+            else {"started_at": "2026-01-01T00:00:00Z"}))
+    reopened = ChatSessionStore(tmp_path)
+    runtime = ChatRuntimeController(store=reopened, codex_bin="unused-test-codex")
+    native = CodexChatAgentSession(process=SimpleNamespace(poll=lambda: None),
+        messages=queue.Queue(), thread_id="public-thread", work_dir=tmp_path)
+    adapter = CodexAppServerAdapter(native)
+    sent = []
+    def write(packet):
+        assert packet["method"] == "turn/start"
+        sent.append(packet["params"])
+        native.messages.put({"id": packet["id"], "result": {"turn": {"id": "original-dispatch"}}})
+        native.messages.put({"method": "item/completed", "params": {"item": {
+            "type": "agentMessage", "phase": "final_answer", "text": "Recovered the original queue."}}})
+        native.messages.put({"method": "turn/completed", "params": {"turn": {"status": "completed"}}})
+    monkeypatch.setattr(native, "_write", write)
+    prepared = threading.Event()
+    def ensure(*_a, **_kw):
+        prepared.set()
+        return adapter
+    monkeypatch.setattr(runtime, "_ensure_adapter", ensure)
+    try:
+        runtime.resume_session_queue(session_id=sid, work_dir=tmp_path, objective="Original request")
+        if prior_dispatch:
+            assert prepared.wait(timeout=2)
+            current = reopened.load_turn(sid, original["turn_id"])
+            assert current["status"] == "queued" and not sent
+            assert current.get("upstream_turn_id") == ("unknown-native-dispatch" if prior_dispatch == "native-id" else None)
+            return
+        result = runtime.wait_for_turn(session_id=sid, turn_id=original["turn_id"], timeout_sec=3)
+        assert result["status"] == "completed"
+        assert len(sent) == 1 and sent[0]["input"][0]["text"].endswith(original["message"])
+        assert reopened.turn_for_client(sid, original["client_turn_id"])["turn_id"] == original["turn_id"]
+        assert len([e for e in reopened.events_after(sid, original["turn_id"], None)
+                    if e["kind"] == "turn.started"]) == 1
+    finally:
+        runtime.close()
+
+
 def test_account_preparation_keeps_its_error_before_native_dispatch(tmp_path, monkeypatch):
     store, sid, runtime = session_runtime(tmp_path)
 

tests/test_steward_interrupted_answer.py

@@ -58,6 +58,16 @@ def observed(kind, payload):
     transport.reconcile_request(path)
     failure_bytes = store._turn_path(original["session_id"], original["turn_id"]).read_bytes()
     notice = deepcopy(json.loads(path.read_text())["deliveries"])
+    history_marker = "Unrelated retained conversation. " * 5000
+    if fault is None:
+        store.append_message(original["session_id"], role="agent", text=history_marker,
+                             turn_id="unrelated-historical-turn")
+    start_adapter = runtime._start_adapter
+    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"
+        return start_adapter(**kwargs)
+    monkeypatch.setattr(runtime, "_start_adapter", start_scoped_review)
     class Ahead(datetime):
         @classmethod
         def now(cls, tz=None):
@@ -129,6 +139,11 @@ def crash(**kwargs):
         assert opened["config"]["permissions"][opened["permissions"]]["network"]["enabled"] is False
         assert [t["name"] for t in opened["dynamicTools"]] == ["loopx_manager_read"]
         assert not any(c["method"] == "thread/resume" for c in calls), "unknown old business must not be resumed"
+        assert not any(c["method"] in {"thread/inject_items", "thread/compact/start"} for c in calls)
+        if fault is None:
+            assert any(item["text"] == history_marker for item in store.messages(original["session_id"]))
+            review_inputs = [c["params"] for c in calls if c["method"] == "turn/start"]
+            assert original["message"] in json.dumps(review_inputs, ensure_ascii=False)
         writes = list(provider.writes)
         restarted.reconcile_request(path)
         assert provider.writes == writes