diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index d0b01e2413..7a06a69f98 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -6,7 +6,7 @@
 """
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from copy import deepcopy
 from pathlib import Path
 import json
@@ -27,6 +27,8 @@
 # Request journals are named by their 24-hex request_ref; lock metadata and
 # other siblings in this directory are not requests.
 _REQUEST_JOURNAL_STEM = re.compile(r"[a-f0-9]{24}")
+QUOTA_BACKLOG_BATCH_SIZE = 64
+_QUOTA_BACKLOG_BATCH_BYTES = 1024 * 1024
 
 
 def _commission_goal_identity(goal: dict[str, Any]) -> dict[str, str]:
@@ -186,8 +188,8 @@ def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, Any],
                 and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"])
                 for item in (source_controls or self.pending)())))
 
-    def quota_backlog_observer(self) -> Callable[[dict[str, Any]], bool]:
-        """Share lazy journal IO within one read-only observation, never admission."""
+    def quota_backlog_observer(self) -> Callable[[Sequence[dict[str, Any]]], list[bool]]:
+        """Batch the existing typed projection; share controls only within this read."""
         controls: list[dict[str, Any]] | None = None
 
         def read_controls() -> list[dict[str, Any]]:
@@ -196,29 +198,87 @@ def read_controls() -> list[dict[str, Any]]:
                 controls = self.pending()
             return controls
 
-        return lambda row: self.quota_backlog_pending(row, source_controls=read_controls)
+        def observe(rows: Sequence[dict[str, Any]]) -> list[bool]:
+            if len(rows) > QUOTA_BACKLOG_BATCH_SIZE:
+                raise ValueError("select at most 64 pending requests per observation")
+            params = [self._quota_pending_inputs(row) for row in rows]
+            flags = self._quota_pending_flags(params)
+            stopped = []
+            for index, (row, values, pending) in enumerate(zip(rows, params, flags, strict=True)):
+                if pending:
+                    events = self.controller.store.events_after(row["session_id"], row["turn_id"], None)
+                    if self._source_stop_observed(row, values["pending_observation"]["turn"], events,
+                                                 source_controls=read_controls):
+                        values["pending_observation"]["stop_observed"] = True
+                        stopped.append(index)
+            for index, pending in zip(stopped, self._quota_pending_flags([params[i] for i in stopped]), strict=True):
+                flags[index] = pending
+            return flags
+
+        return observe
 
-    def quota_backlog_pending(self, row: dict[str, Any], *,
-                              source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
-        """Retain quota and declared unfinished work without admitting recovery."""
+    @staticmethod
+    def _quota_pending_flags(params: Sequence[dict[str, Any]]) -> list[bool]:
         from ...control_plane.effect_runtime import effect_runtime_result
 
+        flags: list[bool] = []
+        batch: list[dict[str, Any]] = []
+        size = 0
+
+        def flush() -> None:
+            if not batch:
+                return
+            result = effect_runtime_result("collaboration.conversation.request", {"pending_observations": batch})
+            values = result.get("pending_flags") if isinstance(result, dict) else None
+            if (not isinstance(values, list) or len(values) != len(batch)
+                    or any(type(value) is not bool for value in values)):
+                raise ValueError("pending observation results are unavailable")
+            flags.extend(values)
+            batch.clear()
+
+        for values in params:
+            encoded = len(json.dumps(values, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
+            if batch and (len(batch) >= QUOTA_BACKLOG_BATCH_SIZE or size + encoded > _QUOTA_BACKLOG_BATCH_BYTES):
+                flush()
+                size = 0
+            if encoded > _QUOTA_BACKLOG_BATCH_BYTES:
+                # Preserve the original single-request transport and its byte
+                # refusal; batching must not introduce a new content limit.
+                result = effect_runtime_result("collaboration.conversation.request", values)
+                pending = result.get("pending") if isinstance(result, dict) else None
+                if type(pending) is not bool:
+                    raise ValueError("pending observation result is unavailable")
+                flags.append(pending)
+                continue
+            batch.append(values)
+            size += encoded
+        flush()
+        return flags
+
+    def _quota_pending_inputs(self, row: dict[str, Any]) -> dict[str, Any]:
         store = self.controller.store
-        session = store.load_session(row["session_id"])
-        turn = store.load_turn(row["session_id"], row["turn_id"])
-        observation = {"turn": turn, "stop_observed": False}
-        params = {
+        return {
             "request": {key: value for key, value in row.items() if key != "attachments"},
-            "current_session": session,
-            "pending_observation": observation,
+            "current_session": store.load_session(row["session_id"]),
+            "pending_observation": {"turn": store.load_turn(row["session_id"], row["turn_id"]),
+                                    "stop_observed": False},
         }
+
+    def quota_backlog_pending(self, row: dict[str, Any], *,
+                              source_controls: Callable[[], list[dict[str, Any]]] | None = None) -> bool:
+        """Retain quota and declared unfinished work without admitting recovery."""
+        from ...control_plane.effect_runtime import effect_runtime_result
+
+        params = self._quota_pending_inputs(row)
+        observation = params["pending_observation"]
+        turn = observation["turn"]
         # The typed owner first excludes work that has no retained obligation even
         # without a stop. Stop evidence can only remove a pending request;
         # never expose a positive result before reading those native facts.
         if effect_runtime_result("collaboration.conversation.request", params).get("pending") is not True:
             return False
         assert turn is not None  # Required by the typed pending projection.
-        events = store.events_after(row["session_id"], row["turn_id"], None)
+        events = self.controller.store.events_after(row["session_id"], row["turn_id"], None)
         if self._source_stop_observed(row, turn, events, source_controls=source_controls):
             observation["stop_observed"] = True
             return effect_runtime_result("collaboration.conversation.request", params).get("pending") is True
@@ -439,6 +499,9 @@ def _preparation_scope_available(self, session: dict[str, Any]) -> bool:
     def _native_result_text(recovered: dict[str, Any]) -> str:
         response = recovered["native_observation"]["response"]
         text = "已恢复本次请求的原答复：\n\n" + response["message"]
+        if recovered["native_observation"].get("latest_input_verified") is False:
+            text = ("已补回本次请求在原生会话中生成的历史答复。本次只补回正文，没有重新执行业务。"
+                    "文中提到其他消息“已处理”的说法仍需逐条核实，未答的问题继续保留。\n\n" + response["message"])
         if any(response.get(key) for key in ("context_handoff", "proposals", "protected_action", "goal_draft")):
             text += "\n\n操作状态：本次仅恢复原答复，结构化操作或委派尚未提交；未重新执行原请求。"
         if (response.get("gate") or {}).get("summary") and response["gate"]["summary"] not in text:
@@ -461,8 +524,8 @@ def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
         if len(anchors) > 1:
             return None
         # Retained facts can repair only a completed Turn's text. Failed-Turn
-        # recovery keeps its live/latest-input requirements; old capture does
-        # not authorize another invocation or change unknown-effect handling.
+        # recovery keeps its live native read and exact input requirements; old
+        # capture does not authorize another invocation or unknown-effect replay.
         retained = self.controller.store.native_result(session["session_id"], turn["turn_id"])
         if (turn.get("status") == "completed" and len(anchors) == 1 and retained
                 and retained.get("upstream_turn_id") == turn.get("upstream_turn_id")):
diff --git a/loopx/chat_manager.py b/loopx/chat_manager.py
index 8e64824679..cef5b31882 100644
--- a/loopx/chat_manager.py
+++ b/loopx/chat_manager.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import hashlib
+import time
 from pathlib import Path
 from typing import Any, Callable, Mapping
 
@@ -991,7 +992,7 @@ def manager_session_quota_route(controller: Any, session: Mapping[str, Any]) ->
         allocation = manager_executor_allocation(controller, machine_defaults=defaults, environ=environment,
             endpoint_capabilities=[{**row, "available": row.get("agent_id") == endpoint} for row in capabilities])
         return {"allocation": allocation, "blocked_primary": route.get("blocked_primary"),
-                "observed_at": capacity["observed_at"] if capacity else None}
+                "observed_at": capacity["observed_at"] if capacity else time.time()}
     except Exception as exc:
         import logging
         logging.getLogger(__name__).warning("Steward quota observation held (%s)", type(exc).__name__)
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index a6c551c4ed..7550388453 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -446,6 +446,20 @@ export function listLocalBoundSessions(params: JsonObject): JsonObject {
  * command; Core binds its exact target before IO. Replays retain that target.
  */
 export function planBoundConversationRequest(params: JsonObject): JsonObject {
+  if (params.pending_observations !== undefined) {
+    const observations = params.pending_observations;
+    if (Object.keys(params).length !== 1 || !Array.isArray(observations) || observations.length > 64) {
+      throw new EffectRuntimeRequestError("select at most 64 read-only pending observations");
+    }
+    return {pending_flags: observations.map(value => {
+      const item = requireJsonObject(value, "pending observation input");
+      if (Object.keys(item).some(key => !["request", "current_session", "pending_observation"].includes(key))
+          || item.pending_observation === undefined) {
+        throw new EffectRuntimeRequestError("pending batches cannot admit or recover work");
+      }
+      return planBoundConversationRequest(item).pending;
+    })};
+  }
   const row = requireJsonObject(params.request, "external request");
   const request = ref(row.request_ref, "external request identity");
   const command = row.command;
@@ -509,7 +523,7 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && (expandedResult || metadataOwed || unfinished || turn?.status === "failed"
         && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure",
           "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error",
-          "manager_authorization_unavailable"].includes(String(turn.error_code)))
+          "manager_authorization_unavailable", "runtime_error", "server_restarted"].includes(String(turn.error_code)))
       && !(current?.session_id === row.session_id && current?.status === "closed")
       && observation.stop_observed !== true};
   }
@@ -687,6 +701,16 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
     const created = Date.parse(String(proof.input_created_at));
     const start = Date.parse(String(failed.started_at)), end = Date.parse(String(failed.completed_at));
     const response = proof.response == null ? null : requireJsonObject(proof.response, "native result response");
+    // Returning an already generated historical answer is not replay. Bind its
+    // exact original input and closed native segment; later ordinary questions
+    // must not strand that answer. Legacy unanchored results stay latest-only.
+    const anchoredHistorical = anchor !== null && anchor.thread_id === current.upstream_thread_id
+      && Number.isSafeInteger(anchor.start_step_index) && Number(anchor.start_step_index) >= 0
+      && typeof anchor.input_sha256 === "string" && BARE_SHA256_PATTERN.test(anchor.input_sha256)
+      && proof.start_step_index === anchor.start_step_index && proof.input_sha256 === anchor.input_sha256
+      && proof.latest_input_verified === false && Number.isSafeInteger(proof.end_step_index)
+      && Number(proof.end_step_index) > Number(proof.start_step_index) + 1
+      && Number(proof.end_step_index) < Number(proof.total_steps);
     return {...candidate, recovery_eligible: recovery.source_verified === true
       && proof.thread_id === current.upstream_thread_id && proof.request_ref === row.request_ref
       && typeof proof.execution_id === "string" && proof.execution_id.length > 0
@@ -697,7 +721,7 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
             && Number(proof.end_step_index) > Number(proof.start_step_index) + 1
             && Number(proof.end_step_index) < Number(proof.total_steps))
           && response !== null && extendsOriginalText(originalResponse!, response)
-        : proof.latest_input_verified === true)
+        : proof.latest_input_verified === true || anchoredHistorical)
       && proof.normal_terminal_verified === true
       && Number.isFinite(created) && created >= start && created <= end
       && Number.isSafeInteger(proof.start_step_index) && Number(proof.start_step_index) >= 0
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index fa78a14413..219071662a 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -15,7 +15,7 @@
 from pathlib import Path
 from typing import Any
 
-from ...capabilities.native_chat.external_conversations import ChatExternalConversations
+from ...capabilities.native_chat.external_conversations import ChatExternalConversations, QUOTA_BACKLOG_BATCH_SIZE
 from ...chat_store import TERMINAL_TURN_STATES, _atomic_write_json, _read_json
 from ...control_plane.collaboration import conversation_reply_context
 from ...control_plane.effect_runtime import effect_runtime_request_scope
@@ -225,6 +225,17 @@ def profiles(self) -> dict[str, dict[str, str]]:
     def health(self) -> dict[str, dict[str, int]]:
         health: dict[str, dict[str, int]] = {}
         quota_pending = self.core.quota_backlog_observer()
+        completed: list[tuple[dict[str, Any], dict[str, Any]]] = []
+
+        def observe_completed() -> None:
+            flags = quota_pending([native for _, native in completed])
+            for (row, _), pending in zip(completed, flags, strict=True):
+                if pending:
+                    health[row["binding_id"]]["pending_count"] += 1
+                    health[row["binding_id"]]["recovery_count"] += int(any(
+                        phase.get("started") and not phase.get("verified") for phase in row["deliveries"].values()))
+            completed.clear()
+
         for path in [*self.root.glob("*.json"), *self.root.glob("web-projections/*.json")]:
             row = _read_json(path)
             entry = health.setdefault(row["binding_id"], {"pending_count": 0, "recovery_count": 0})
@@ -239,11 +250,14 @@ def health(self) -> dict[str, dict[str, int]]:
                 if (native.get("status") == "accepted" and native.get("command") is None
                         and all(native.get(key) == row.get(key) for key in
                             ("binding_id", "source", "session_id", "turn_id"))):
-                    pending = quota_pending(native)
+                    completed.append((row, native))
+                    if len(completed) >= QUOTA_BACKLOG_BATCH_SIZE:
+                        observe_completed()
             if pending:
                 entry["pending_count"] += 1
                 entry["recovery_count"] += int(any(phase.get("started") and not phase.get("verified")
                                                    for phase in row["deliveries"].values()))
+        observe_completed()
         return health
 
     def _binding(self, profile: str) -> dict[str, Any]:
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 1e11b306d2..dd1b982478 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -243,6 +243,20 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
   assert.equal(plan.operation, "adopt_native_result");
   assert.equal(plan.recovery_eligible, true);
   assert.equal(plan.client_turn_id, undefined);
+  const anchor = {thread_id: "native", start_step_index: proof.start_step_index, input_sha256: proof.input_sha256};
+  const historicalResult = {...packet, context_recovery: {...packet.context_recovery, native_input: anchor,
+    native_observation: {...proof, latest_input_verified: false, end_step_index: 6}}};
+  assert.equal(planBoundConversationRequest(historicalResult).recovery_eligible, true);
+  assert.equal(planBoundConversationRequest(historicalResult).client_turn_id, undefined);
+  for (const change of [{native_input: null}, {source_verified: false}, {stop_observed: true},
+    {native_input: {...anchor, input_sha256: "9".repeat(64)}},
+    {native_input: {...anchor, start_step_index: 3}},
+    {native_input: {...anchor, thread_id: "other"}},
+    ...[null, 5, 8, 9].map(end_step_index => ({native_observation: {
+      ...historicalResult.context_recovery.native_observation, end_step_index}}))]) {
+    assert.equal(planBoundConversationRequest({...historicalResult,
+      context_recovery: {...historicalResult.context_recovery, ...change}}).recovery_eligible, false);
+  }
   for (const field of ["input_sha256", "answer_sha256"]) {
     for (const invalid of [null, false, 1, [], {}, "", "1".repeat(63), "1".repeat(65),
       "A".repeat(64), `sha256:${"1".repeat(64)}`]) {
@@ -1131,6 +1145,33 @@ test("local input cannot impersonate a group participant or inherit its workspac
     project_context: selected.context, origin: "web"}).kind, "unavailable");
 });
 
+test("pending batches preserve individual decisions and cannot enter execution or recovery", () => {
+  const request = {request_ref: "f".repeat(24), command: null, status: "accepted",
+    session_id: "original", turn_id: "failed"};
+  const turn = {session_id: "original", turn_id: "failed", status: "failed", error_code: "usage_limit_exceeded"};
+  const packet = {request, current_session: {session_id: "original", status: "ready"},
+    pending_observation: {turn, stop_observed: false}};
+  const observations = [packet,
+    {...packet, current_session: null},
+    {...packet, current_session: {...packet.current_session, status: "closed"}},
+    {...packet, pending_observation: {turn, stop_observed: true}},
+    ...[null, {...turn, error_code: "policy"}, {...turn, turn_id: "foreign"},
+      {...turn, status: "completed", response: {message: "Partial", proposals: [], gate: null,
+        protected_action: null, unfinished_work: {remaining: "Retained work", dependency: "offline", effects_settled: true}}}]
+      .map(value => ({...packet, pending_observation: {turn: value, stop_observed: false}}))];
+  const before = structuredClone(observations);
+  assert.deepEqual(planBoundConversationRequest({pending_observations: observations}),
+    {pending_flags: observations.map(item => planBoundConversationRequest(item).pending)});
+  assert.deepEqual(observations, before);
+  assert.deepEqual(planBoundConversationRequest({pending_observations: []}), {pending_flags: []});
+  for (const value of [null, {}, Array(65).fill(packet), [packet, {...packet, request: {}}],
+    [packet, {request, current_session: packet.current_session}],
+    [packet, {...packet, context_recovery: {}}], [{pending_observations: []}]]) {
+    assert.throws(() => planBoundConversationRequest({pending_observations: value}));
+  }
+  assert.throws(() => planBoundConversationRequest({pending_observations: [packet], request}));
+});
+
 test("quota backlog projection retires explicit lifecycle stops, not unavailable recovery", () => {
   const original = {request_ref: "f".repeat(24), command: null, status: "accepted",
     session_id: "original", turn_id: "failed"};
@@ -1160,7 +1201,7 @@ test("quota backlog projection retires explicit lifecycle stops, not unavailable
 });
 
 for (const errorCode of ["response_timeout", "upstream_transport_failure",
-  "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error"]) {
+  "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error", "runtime_error", "server_restarted"]) {
 test(`delivered ${errorCode} notices retain the exact failed request, including other executors`, () => {
   const request = {request_ref: "f".repeat(24), command: null, status: "accepted",
     session_id: "original", turn_id: "timed-out", delivery_verified: true};
diff --git a/tests/test_chat_manager_model_adoption.py b/tests/test_chat_manager_model_adoption.py
index 00eaa67ec7..f4347c2052 100644
--- a/tests/test_chat_manager_model_adoption.py
+++ b/tests/test_chat_manager_model_adoption.py
@@ -82,6 +82,36 @@ def start(**kwargs):
     return root, store, runtime, session, starts
 
 
+def test_api_primary_route_has_fresh_observation_without_reading_argon_quota(conversation, monkeypatch):
+    import sys
+    import time
+    from types import ModuleType, SimpleNamespace
+    import loopx.chat_manager as manager
+    from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter
+
+    root, _, runtime, _, _ = conversation
+    configure(root, model="gemini-4-argon-high", effort="high", endpoint="antigravity", policy="flexible")
+    calls = []
+    module = ModuleType("loopx_codex_provider_routing.native")
+
+    def route(**inputs):
+        calls.append(inputs)
+        return {"endpoint": "codex", "blocked_primary": None}
+
+    module.NativeAccounts = lambda _: SimpleNamespace(steward_route=route, model_transport=lambda **_: None)
+    monkeypatch.setitem(sys.modules, module.__name__, module)
+    monkeypatch.setattr(manager, "operator_credential_resolution", lambda _: {"environ": {}})
+    monkeypatch.setattr(AntigravityAdapter, "metadata_review_capacity",
+        lambda **_: pytest.fail("API primary must not wait for an unrelated native quota read"))
+    before = time.time()
+    result = manager.manager_session_quota_route(runtime, {"agent_id": "antigravity", "upstream_thread_id": "original"})
+    after = time.time()
+    assert result["allocation"]["executor_endpoint"] == "codex"
+    assert before <= result["observed_at"] <= after
+    assert len(calls) == 1 and calls[0]["primary_capacity"] is None
+    runtime.close()
+
+
 def test_idle_conversation_adopts_model_preserving_context_and_authority(conversation):
     root, store, runtime, session, starts = conversation
     store.append_message(session["session_id"], role="user", text="Keep using public sources.")
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index a005792b49..ff26f237a8 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -331,7 +331,8 @@ def connect(fixture):
                                                              runner=provider, cli_bin="lark-cli")
 
 
-@pytest.mark.parametrize("case", ["normal", "invalid-metadata", "metadata-only", "unfinished", "source-changed", "closed", "lost-ack", "message-crash"])
+@pytest.mark.parametrize("case", ["normal", "invalid-metadata", "metadata-only", "unfinished", "source-changed", "closed", "lost-ack", "message-crash",
+    "historical", "historical-source-changed", "historical-stop"])
 def test_original_worker_recovers_native_result_without_replay(tmp_path, language_server, monkeypatch, case):  # noqa: F811
     from test_native_steward_private import steward
     from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter
@@ -342,6 +343,8 @@ def test_original_worker_recovers_native_result_without_replay(tmp_path, languag
     monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high")
     fixture = steward.__wrapped__(tmp_path)
     store, runtime, provider, transport, binding, _, workspace = next(fixture)
+    historical = case.startswith("historical")
+    fault = case.removeprefix("historical-") if case != "historical" else "normal"
     try:
         binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
             executor_endpoint_id="antigravity", context_kind="steward")
@@ -376,12 +379,21 @@ def lost_core_terminal(adapter, message, event_sink):
         old = _read_json(path)
         if case not in {"source-changed", "closed"}:
             assert transport.core.context_recovery_candidate(ref)
+        if historical:
+            monkeypatch.setattr(AntigravityAdapter, "start_turn", original)
+            later = provider.event("steward-app", "later-separate-question", "An unrelated later question")
+            transport.admit("steward-app", later)
+            later_row = transport.core.read_request(identity_ref(binding["provider_ref"], later["message_id"]))
+            assert runtime.wait_for_turn(session_id=later_row["session_id"], turn_id=later_row["turn_id"], timeout_sec=15)["status"] == "completed"
+            transport.reconcile_request(transport.root / f"{later_row['request_ref']}.json")
         failed_bytes = store._turn_path(row["session_id"], row["turn_id"]).read_bytes()
         writes = len(provider.writes)
         monkeypatch.setattr(runtime, "enqueue_turn", lambda **_: pytest.fail("Result recovery enqueued work"))
         monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: pytest.fail("Result recovery resumed work"))
-        if case == "source-changed":
+        if fault == "source-changed":
             provider.messages[event["message_id"]]["content"] = "Edited source"
+        elif fault == "stop":
+            store.append_event(row["session_id"], row["turn_id"], kind="turn.interrupt_requested", payload={})
         elif case == "closed":
             store.update_session(row["session_id"], status="closed")
         elif case == "lost-ack":
@@ -409,7 +421,7 @@ def crash(*args, **kwargs):
         restarted.reconcile()
         native = restarted.core.read_request(ref)
         recovered = native.get("native_result_recovery")
-        if case in {"source-changed", "closed"}:
+        if fault in {"source-changed", "closed", "stop"}:
             assert recovered is None and len(provider.writes) == writes
         elif case == "lost-ack":
             assert recovered["delivery_verified"] is False
@@ -427,6 +439,9 @@ def crash(*args, **kwargs):
             body = json.loads(provider.messages[receipt["attempt"]["message_ref"]]["body"]["content"])
             assert body["zh_cn"]["content"][0][0]["text"] == message["text"]
             assert "Argon answer 1" in message["text"]
+            if historical:
+                assert "历史答复" in message["text"] and "没有重新执行业务" in message["text"]
+                assert "Argon answer 2" not in message["text"]
             assert len(provider.writes) == writes + 1
             assert restarted.health()[row["binding_id"]]["pending_count"] == (case in {"invalid-metadata", "unfinished"})
             if case == "unfinished":
@@ -434,8 +449,8 @@ def crash(*args, **kwargs):
             assert len([item for item in store.messages(row["session_id"]) if item["message_id"] == recovered["message_id"]]) == 1
         assert _read_json(path)["deliveries"]["terminal"] == old["deliveries"]["terminal"]
         assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == failed_bytes
-        assert len(list(store._turn_path(row["session_id"], row["turn_id"]).parent.glob("*.json"))) == 1
-        assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+        assert len(list(store._turn_path(row["session_id"], row["turn_id"]).parent.glob("*.json"))) == (2 if historical else 1)
+        assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == (2 if historical else 1)
     finally:
         fixture.close()
 
diff --git a/tests/test_private_chat_backlog_projection.py b/tests/test_private_chat_backlog_projection.py
new file mode 100644
index 0000000000..0049f86284
--- /dev/null
+++ b/tests/test_private_chat_backlog_projection.py
@@ -0,0 +1,107 @@
+"""Read-only backlog projection uses bounded RPCs and retains native stop facts."""
+from types import SimpleNamespace
+from copy import deepcopy
+import json
+
+import pytest
+
+from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
+from loopx.chat_store import CHAT_TURN_SCHEMA_VERSION, ChatSessionStore, _atomic_write_json
+from loopx.control_plane import effect_runtime
+from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+
+
+def fixture_history(tmp_path, count=130):
+    store = ChatSessionStore(tmp_path)
+    session = store.create_session(goal_id="demo", agent_id="codex", adapter_kind="codex_app_server",
+        upstream_thread_id="native", session_id="a" * 32)
+    controller = SimpleNamespace(store=store,
+        project_contexts=SimpleNamespace(conversation_bindings=SimpleNamespace()))
+    core = ChatExternalConversations(controller)
+    transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
+    transport.core, transport.root = core, store.root / "lark-private-deliveries"
+    rows = []
+    for index in range(count):
+        ref, tid = f"{index:024x}", f"{index:032x}"
+        row = {"request_ref": ref, "binding_id": "b" * 24,
+            "source": {"source_ref": "c" * 24}, "session_id": session["session_id"], "turn_id": tid,
+            "command": None, "status": "accepted", "delivery_verified": True,
+            "message": "Original user request", "created_at": "2026-01-01T00:00:00+00:00"}
+        turn = {"schema_version": CHAT_TURN_SCHEMA_VERSION, "session_id": session["session_id"],
+            "turn_id": tid, "client_turn_id": f"external-{ref}", "status": "failed",
+            "error_code": "usage_limit_exceeded", "response": None}
+        _atomic_write_json(store._turn_path(session["session_id"], tid), turn)
+        _atomic_write_json(core.root / f"{ref}.json", row)
+        _atomic_write_json(transport.root / f"{ref}.json", dict(row, status="delivered", deliveries={}))
+        rows.append(row)
+    return store, core, transport, rows
+
+
+def durable_state(store):
+    return {str(p.relative_to(store.root)): p.read_bytes() for p in store.root.rglob("*")
+        if p.suffix in {".json", ".jsonl"}}
+
+
+def test_full_history_is_read_once_per_batch_and_control_changes_are_fresh(tmp_path, monkeypatch):
+    store, core, transport, rows = fixture_history(tmp_path)
+    before = durable_state(store)
+    call = effect_runtime.effect_runtime_result
+    calls = []
+
+    def observed(method, params, **kwargs):
+        if method == "collaboration.conversation.request":
+            calls.append(deepcopy(params))
+        return call(method, params, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", observed)
+    assert transport.health()["b" * 24] == {"pending_count": 130, "recovery_count": 0}
+    assert [len(item["pending_observations"]) for item in calls] == [64, 64, 2]
+    assert durable_state(store) == before
+    # A later stop affects the next read even though the earlier scan shared IO.
+    control = dict(rows[0], request_ref="f" * 24, command="stop", created_at="2026-01-02T00:00:00+00:00")
+    _atomic_write_json(core.root / ("f" * 24 + ".json"), control)
+    after_stop = durable_state(store)
+    assert transport.health()["b" * 24]["pending_count"] == 0
+    assert durable_state(store) == after_stop
+    assert all(not core.quota_backlog_pending(row) for row in rows)
+
+
+@pytest.mark.parametrize("size", [600_000, 1_100_000])
+def test_byte_split_and_original_large_single_request_keep_order(tmp_path, monkeypatch, size):
+    _, core, _, rows = fixture_history(tmp_path, count=3)
+    rows[0]["message"] = "x" * size
+    rows[1]["message"] = "x" * size
+    # The third unrelated policy refusal must not acquire a neighbour's flag.
+    store = core.controller.store
+    store.update_turn(rows[2]["session_id"], rows[2]["turn_id"], error_code="policy")
+    expected = [core.quota_backlog_pending(row) for row in rows]
+    call = effect_runtime.effect_runtime_result
+    params_seen = []
+
+    def observed(method, params, **kwargs):
+        if method == "collaboration.conversation.request":
+            params_seen.append(deepcopy(params))
+        return call(method, params, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", observed)
+    assert core.quota_backlog_observer()(rows) == expected == [True, True, False]
+    assert len(params_seen) >= 2
+    assert all(len(json.dumps(p).encode()) < 1_250_000 for p in params_seen)
+    assert any("request" in p for p in params_seen) is (size > 1_048_576)
+
+
+@pytest.mark.parametrize("result", [None, {}, {"pending_flags": []}, {"pending_flags": [1]}])
+def test_unavailable_or_misaligned_batch_result_is_never_empty_work(tmp_path, monkeypatch, result):
+    _, core, _, rows = fixture_history(tmp_path, count=1)
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", lambda *_args, **_kwargs: result)
+    with pytest.raises(ValueError, match="unavailable"):
+        core.quota_backlog_observer()(rows)
+
+
+@pytest.mark.parametrize("code", ["runtime_error", "server_restarted"])
+def test_worker_or_preparation_failure_notice_does_not_answer_the_original_question(tmp_path, code):
+    store, _, transport, rows = fixture_history(tmp_path, count=1)
+    store.update_turn(rows[0]["session_id"], rows[0]["turn_id"], error_code=code)
+    before = durable_state(store)
+    assert transport.health()["b" * 24] == {"pending_count": 1, "recovery_count": 0}
+    assert durable_state(store) == before  # Visibility grants no execution or replay.