fix(chat): avoid reparsing idle history during recovery

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

@@ -1921,3 +1921,26 @@ Reservation/enqueue crashes adopt the same queued Turn. A failed, partial or
 unknown review stays pending and cannot become another review or business retry.
 Other malformed actions and completed Turns without a recovered native result
 remain explicit holds; this path is not a generic action or execution replayer.
+
+### File-backed idle recovery observations
+
+Repeated recovery checks must not reparse every historical Turn and source body
+for each old request. The File Chat provider retains only validated terminal-Turn
+identity/status and the source-control fields used by the existing stop/new and
+later-input checks. Reuse requires an unchanged device, inode, nanosecond mtime,
+ctime and size, checked before and after the original read. Changed or unreadable
+files use the original reader; nonterminal Turns are always read in full. Deleted
+entries are pruned, concurrent callers receive independent metadata, and a read
+racing a replacement cannot seed a reusable observation.
+Windows keeps the original fresh-read path because its Python stat ctime can
+represent creation time rather than a usable change stamp; see the
+[Python stat time fields](https://docs.python.org/3.12/library/os.html#os.stat_result.st_ctime_ns).
+The private-reply journal observer shares the same change-token primitive while
+retaining its own legacy-repair projection and original effect checks.
+
+These observations are neither recovery decisions nor grants. Exact requests,
+selected Turns, current bindings, native effects and typed admission remain fresh.
+Source qualification compares the same File history and complete recovery plans,
+including newly queued input, strict read failures, stop/new changes and replaced
+files with preserved mtime. Installed responsiveness, provider/model availability
+and actual original-audience replies remain separate acceptance obligations.

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

@@ -1260,3 +1260,20 @@ worker 使用独立展示阶段返回这些回执，保留原失败 Turn、失
 预约或入队后的崩溃只采用同一个队列 Turn；失败、部分提交或结果未知的复核继续
 记为欠办，不能变成又一次复核或业务重试。其他错误动作、没有已恢复原生结果的
 completed Turn 继续明确保留，本路径不提供通用动作或执行重放。
+
+### File 后端的空闲恢复观测
+
+逐条检查旧请求时，不应反复解析全部历史 Turn 和来源正文。File Chat provider
+只保留经过验证的终态 Turn 身份/状态，以及既有 stop/new 和后续输入检查所需的
+来源控制字段。复用要求 device、inode、纳秒 mtime、ctime 和文件大小均未改变，
+并在原读取前后核对。变化或不可读的文件仍走原 reader；非终态 Turn 每次读取
+完整正文。删除项清理、并发调用返回独立元数据，读取期间发生替换时不保存可复用观测。
+Windows 的 Python stat ctime 可能代表创建时间，不能作为可靠的变化标识，因此
+继续走原来的新鲜读取路径，见
+[Python stat 时间字段说明](https://docs.python.org/3.12/library/os.html#os.stat_result.st_ctime_ns)。
+私聊回复 journal 的观测复用同一个变化标识实现，仍保留自身的旧修复投影和原效果核验。
+
+这些观测不是恢复决定或授权。精确请求、选中的 Turn、当前 binding、原生效果及
+typed 准入仍保持新鲜读取。源码资格以同一份 File 历史和完整恢复计划做对照，
+覆盖新入队输入、严格读取失败、stop/new 变化及保留 mtime 的文件替换。实际安装后的
+响应速度、provider/模型可用性和原受众真正收到答复，仍须分别验收。

loopx/capabilities/native_chat/external_conversations.py

@@ -24,6 +24,7 @@
 )
 from ...file_lock import exclusive_file_lock, try_exclusive_file_lock
 from .codex_context import checked_session_home
+from .file_discovery import FileChatDiscovery
 
 # Request journals are named by their 24-hex request_ref; lock metadata and
 # other siblings in this directory are not requests.
@@ -82,7 +83,7 @@ def user_messages(self, session_id: str) -> list[dict[str, Any]]:
 
     def source_controls(self) -> list[dict[str, Any]]:
         if self.controls is None:
-            self.controls = self.core.pending()
+            self.controls = self.core.source_controls()
         return self.controls
 
 
@@ -92,6 +93,7 @@ def __init__(self, controller: Any) -> None:
         self.bindings = controller.project_contexts.conversation_bindings
         self.bindings.controller = controller
         self.root = controller.store.root / "external-requests"
+        self._source_control_discovery = FileChatDiscovery()
         self.actions: Any | None = None
 
     def _context_recovery_plan(self, row: dict[str, Any], *,
@@ -182,7 +184,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                     and item.get("source") == row["source"] and item.get("command") is None
                     and datetime.fromisoformat(item["created_at"]) > datetime.fromisoformat(row["created_at"])
                     for item in (discovery_reads.source_controls() if discovery_reads is not None
-                                 else self.pending())) if turn.get("error_code") == "antigravity_step_error" else None,
+                                 else self.source_controls())) if turn.get("error_code") == "antigravity_step_error" else None,
                 "native_input": anchors[0] if len(anchors) == 1 else None,
                 "response_literal_open": bool(open_literal),
                 "metadata_capacity": metadata_capacity,
@@ -212,7 +214,7 @@ def _source_stop_observed(self, row: dict[str, Any], turn: dict[str, Any],
                 item.get("binding_id") == row["binding_id"] and item.get("source") == row["source"]
                 and item.get("command") in {"stop", "new"}
                 and datetime.fromisoformat(item["created_at"]) >= datetime.fromisoformat(row["created_at"])
-                for item in (source_controls or self.pending)())))
+                for item in (source_controls or self.source_controls)())))
 
     def quota_backlog_observer(self) -> Callable[[Sequence[dict[str, Any]]], list[bool]]:
         """Batch the existing typed projection; share controls only within this read."""
@@ -221,7 +223,7 @@ def quota_backlog_observer(self) -> Callable[[Sequence[dict[str, Any]]], list[bo
         def read_controls() -> list[dict[str, Any]]:
             nonlocal controls
             if controls is None:
-                controls = self.pending()
+                controls = self.source_controls()
             return controls
 
         def observe(rows: Sequence[dict[str, Any]]) -> list[bool]:
@@ -1248,6 +1250,25 @@ def pending(self) -> list[dict[str, Any]]:
         return [_read_json(path) for path in sorted(self.root.glob("*.json"))
                 if _REQUEST_JOURNAL_STEM.fullmatch(path.stem)]
 
+    def source_controls(self) -> list[dict[str, Any]]:
+        """Observe exact source-control fields without reparsing unchanged bodies.
+
+        Recovery/stop rules remain with their original typed owner and callers.
+        A new, replaced or modified journal is read again; pending() and exact
+        request reads still expose the complete current canonical records.
+        """
+        paths = (path for path in sorted(self.root.glob("*.json"))
+                 if _REQUEST_JOURNAL_STEM.fullmatch(path.stem))
+
+        def project(_path, row):
+            fields = ("binding_id", "source", "command", "created_at")
+            # The tolerant reader can return {} after transient IO failure.
+            # An incomplete read must not seed an unchanged-file observation.
+            return {key: row[key] for key in fields} if all(key in row for key in fields) else None
+
+        return [row for _path, row in self._source_control_discovery.rows(self.root, paths,
+            read=_read_json, project=project)]
+
     def read_request(self, request_ref: str) -> dict[str, Any]:
         return _read_json(self.root / f"{request_ref}.json")
 

loopx/capabilities/native_chat/file_discovery.py

@@ -0,0 +1,69 @@
+"""Small, invalidatable File Chat journal observations, never execution state."""
+from __future__ import annotations
+
+from collections.abc import Callable, Iterable, Iterator
+from copy import deepcopy
+from pathlib import Path
+import sys
+from threading import Lock
+from typing import Any
+
+
+def file_change_fingerprint(path: Path) -> tuple[int, ...] | None:
+    """A local mutation token, or no reuse when the platform lacks one."""
+    # Windows stat ctime can be creation time, not a mutation stamp. Keep
+    # the original fresh reader there rather than trusting an mtime match.
+    if sys.platform == "win32":
+        return None
+    stat = path.stat()
+    return stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size
+
+
+class FileChatDiscovery:
+    """Keep only caller-selected metadata while the complete file is unchanged.
+
+    Native Chat uses this for terminal-Turn discovery and source-control fields.
+    Selected Turns/requests and every admission still use their original fresh
+    readers. No body, grant, queue decision or provider effect is retained here.
+    """
+
+    def __init__(self) -> None:
+        self._lock = Lock()
+        self._directories: dict[Path, dict[Path, tuple[tuple[int, ...], dict[str, Any]]]] = {}
+
+    def rows(self, directory: Path, paths: Iterable[Path], *,
+             read: Callable[[Path], dict[str, Any]],
+             project: Callable[[Path, dict[str, Any]], dict[str, Any] | None],
+             ) -> Iterator[tuple[Path, dict[str, Any]]]:
+        seen = set()
+        with self._lock:
+            entries = self._directories.setdefault(directory, {})
+        for path in paths:
+            seen.add(path)
+            try:
+                before = file_change_fingerprint(path)
+            except OSError:
+                before = None
+            with self._lock:
+                saved = entries.get(path)
+                if before is None or saved is None or saved[0] != before:
+                    entries.pop(path, None)
+                    saved = None
+            if saved is not None:
+                yield path, deepcopy(saved[1])
+                continue
+            # Preserve the owning reader's missing/unreadable/strict behavior.
+            value = read(path)
+            view = project(path, value)
+            if before is not None and view is not None:
+                try:
+                    unchanged = file_change_fingerprint(path) == before
+                except OSError:
+                    unchanged = False
+                if unchanged:
+                    with self._lock:
+                        entries[path] = before, deepcopy(view)
+            yield path, value if view is None else view
+        with self._lock:
+            for path in entries.keys() - seen:
+                del entries[path]

loopx/chat_ingress.py

@@ -2,13 +2,15 @@
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
+from functools import cached_property
 import os
 import json
 from pathlib import Path
 import uuid
 from typing import Any
 from .chat import require_matching_message_replay, require_matching_replay
 from .file_lock import exclusive_file_lock
+from .capabilities.native_chat.file_discovery import FileChatDiscovery
 
 
 SESSION_QUEUE_MAX_PENDING = 20
@@ -69,6 +71,12 @@ def chat_runtime_activation_state(runtime_root: Path) -> dict[str, int]:
 
 
 class ChatIngressStore:
+    @cached_property
+    def _terminal_turn_discovery(self) -> FileChatDiscovery:
+        # File queue discovery owns this per-store hint. Concurrent first use
+        # may build two empty observers; neither holds execution authority.
+        return FileChatDiscovery()
+
     def runtime_activation_lock(self):
         return chat_runtime_activation_lock(self.root.parent)
 
@@ -81,9 +89,8 @@ def queued_turns(self, session_id: str, *, strict: bool = False, include_reserve
         active_turn_id = str(session.get("active_turn_id") or "")
         turns_dir = self._session_dir(session_id) / "turns"
         rows: list[dict[str, Any]] = []
-        for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
-            if path.name.endswith(".events.json"):
-                continue
+
+        def read(path):
             if strict:
                 try:
                     payload = json.loads(path.read_text(encoding="utf-8"))
@@ -100,6 +107,18 @@ def queued_turns(self, session_id: str, *, strict: bool = False, include_reserve
                     raise OSError("Queued Turn state is invalid")
             else:
                 payload = _read_json(path)
+            return payload
+
+        def terminal(path, payload):
+            if (payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
+                    and payload.get("session_id") == session_id and payload.get("turn_id") == path.stem
+                    and isinstance(payload.get("status"), str)
+                    and payload.get("status") in TERMINAL_TURN_STATES):
+                return {key: payload[key] for key in ("schema_version", "session_id", "turn_id", "status")}
+            return None
+
+        paths = (path for path in turns_dir.glob("*.json") if not path.name.endswith(".events.json"))
+        for path, payload in self._terminal_turn_discovery.rows(turns_dir, paths, read=read, project=terminal):
             if (
                 payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
                 and payload.get("status") == "queued"

loopx/extensions/lark/private_delivery_discovery.py

@@ -6,6 +6,7 @@
 from typing import Any
 
 from ...chat_store import _read_json
+from ...capabilities.native_chat.file_discovery import file_change_fingerprint
 from .private_progress import legacy_recovery_repairs
 
 
@@ -22,15 +23,11 @@ def rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]:
         """
         seen = set()
 
-        def fingerprint(path: Path) -> tuple[int, ...]:
-            stat = path.stat()
-            return (stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size)
-
         for path in sorted(self.root.glob("*.json")):
             seen.add(path)
-            before = fingerprint(path)
+            before = file_change_fingerprint(path)
             saved = self._completed.get(path)
-            if saved is not None and saved[0] == before:
+            if before is not None and saved is not None and saved[0] == before:
                 yield path, saved[1], saved[2]
                 continue
             self._completed.pop(path, None)
@@ -42,7 +39,7 @@ def fingerprint(path: Path) -> tuple[int, ...]:
                         ("status", "request_ref", "binding_id", "session_id", "turn_id")}
                 view.update(agent_target=bool(row.get("agent_target")),
                             event={"chat_type": row.get("event", {}).get("chat_type")})
-                if fingerprint(path) == before:
+                if before is not None and file_change_fingerprint(path) == before:
                     self._completed[path] = (before, view, repairs)
                 row = view
             yield path, row, repairs