Avoid reparsing completed Chat Turns during App and Lark admission

loopx/chat_ingress.py

@@ -1,6 +1,7 @@
 """Durable queued inputs and ingress receipts for the existing Chat store."""
 from __future__ import annotations
 
+from collections.abc import Callable, Iterator
 from datetime import datetime, timedelta, timezone
 from functools import cached_property
 import os
@@ -73,10 +74,47 @@ 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
+        # Queue discovery and client lookup share this hint. Concurrent first use
         # may build two empty observers; neither holds execution authority.
         return FileChatDiscovery()
 
+    def _turn_discovery_rows(
+        self, session_id: str, turns_dir: Path, read: Callable[[Path], dict[str, Any]],
+    ) -> Iterator[tuple[Path, dict[str, Any]]]:
+        from .chat_store import CHAT_TURN_SCHEMA_VERSION, TERMINAL_TURN_STATES
+
+        def terminal(path: Path, payload: dict[str, Any]) -> dict[str, Any] | None:
+            status = payload.get("status")
+            client_id = payload.get("client_turn_id")
+            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(status, str) and status in TERMINAL_TURN_STATES
+                    and (client_id is None or isinstance(client_id, str))):
+                return {
+                    "schema_version": CHAT_TURN_SCHEMA_VERSION, "session_id": session_id,
+                    "turn_id": path.stem, "status": status, "client_turn_id": client_id,
+                }
+            return None
+
+        paths = (path for path in turns_dir.glob("*.json") if not path.name.endswith(".events.json"))
+        return self._terminal_turn_discovery.rows(turns_dir, paths, read=read, project=terminal)
+
+    def turn_for_client(self, session_id: str, client_turn_id: str) -> dict[str, Any] | None:
+        from .chat_store import _read_json
+
+        turns_dir = self._session_dir(session_id) / "turns"
+        if not turns_dir.is_dir():
+            return None
+        for path, observation in self._turn_discovery_rows(session_id, turns_dir, _read_json):
+            if observation.get("client_turn_id") != client_turn_id:
+                continue
+            # Replay validation needs the complete current record, including
+            # the original request body and the latest persisted result.
+            payload = _read_json(path)
+            if payload.get("client_turn_id") == client_turn_id:
+                return payload
+        return None
+
     def runtime_activation_lock(self):
         return chat_runtime_activation_lock(self.root.parent)
 
@@ -109,16 +147,7 @@ def read(path):
                 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):
+        for _, payload in self._turn_discovery_rows(session_id, turns_dir, read):
             if (
                 payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION
                 and payload.get("status") == "queued"

loopx/chat_store.py

@@ -1090,23 +1090,21 @@ def accept_managed_turn(
                 agent_id="loopx-chat",
                 operation="accept_managed_chat_turn",
             ):
-                turns = [
-                    payload
-                    for path in (
-                        self._session_dir(session_token) / "turns"
-                    ).glob("*.json")
-                    if (
-                        payload := _read_json(path)
-                    ).get("schema_version") == CHAT_TURN_SCHEMA_VERSION
-                ]
-                existing = next(
-                    (
-                        turn
-                        for turn in turns
-                        if turn.get("client_turn_id") == client_id
-                    ),
-                    None,
-                )
+                turns_dir = self._session_dir(session_token) / "turns"
+                turns = []
+                existing = None
+                for path, turn in self._turn_discovery_rows(session_token, turns_dir, _read_json):
+                    if turn.get("schema_version") != CHAT_TURN_SCHEMA_VERSION:
+                        continue
+                    # The planner needs live prepared Turns and the full
+                    # current request when its client ID matches.
+                    if existing is None and turn.get("client_turn_id") == client_id:
+                        turn = _read_json(path)
+                        if turn.get("schema_version") != CHAT_TURN_SCHEMA_VERSION:
+                            continue
+                        if turn.get("client_turn_id") == client_id:
+                            existing = turn
+                    turns.append(turn)
                 session = self.load_session(session_token)
                 active_turn = None
                 if session is not None and session.get("active_turn_id"):
@@ -1443,16 +1441,6 @@ def claim_next_queued_turn(
                     return turn
                 return None
 
-    def turn_for_client(self, session_id: str, client_turn_id: str) -> dict[str, Any] | None:
-        turns_dir = self._session_dir(session_id) / "turns"
-        for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
-            if path.name.endswith(".events.json"):
-                continue
-            payload = _read_json(path)
-            if payload.get("client_turn_id") == client_turn_id:
-                return payload
-        return None
-
     def load_turn(self, session_id: str, turn_id: str) -> dict[str, Any] | None:
         payload = _read_json(self._turn_path(session_id, turn_id))
         return payload if payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION else None

tests/test_chat_store_input_validation.py

@@ -52,6 +52,80 @@ def _dedup_store(tmp_path: Path) -> tuple[ChatSessionStore, str]:
     return store, session_id
 
 
+def test_queue_admission_reuses_terminal_observation_but_reads_replay(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from loopx.capabilities.native_chat.file_discovery import file_change_fingerprint
+
+    store, sid = _dedup_store(tmp_path)
+    request = {"client_turn_id": "original-input", "message": "Original request"}
+    turn, _ = store.create_queued_turn(sid, **request)
+    tid = turn["turn_id"]
+    store.update_turn(sid, tid, status="completed", response={"message": "Original answer"})
+    path = store._turn_path(sid, tid)
+    assert store.queued_turns(sid, strict=True) == []
+
+    read_json = chat_store._read_json
+    reads = []
+
+    def observe_read(candidate: Path) -> dict:
+        if candidate == path:
+            reads.append(candidate)
+        return read_json(candidate)
+
+    monkeypatch.setattr(chat_store, "_read_json", observe_read)
+    _, created = store.create_queued_turn(sid, client_turn_id="new-input", message="New request")
+    assert created
+    if file_change_fingerprint(path) is not None:
+        assert not reads
+
+    reads.clear()
+    replay, created = store.create_queued_turn(sid, **request)
+    assert not created
+    assert reads
+    assert replay["turn_id"] == tid
+    assert replay["response"]["message"] == "Original answer"
+
+    store.update_turn(sid, tid, response={"message": "Updated answer"})
+    assert store.turn_for_client(sid, request["client_turn_id"])["response"]["message"] == "Updated answer"
+
+
+def test_managed_admission_uses_terminal_observations_and_full_replay(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from loopx.capabilities.native_chat.file_discovery import file_change_fingerprint
+
+    store, sid = _dedup_store(tmp_path)
+    request = {"client_turn_id": "original-managed", "message": "Original App request"}
+    original, _ = store.create_turn(sid, **request)
+    tid = original["turn_id"]
+    store.update_turn(sid, tid, status="completed", response={"message": "Original App answer"})
+    assert store.release_active_turn(sid, tid)
+    path = store._turn_path(sid, tid)
+    assert store.queued_turns(sid, strict=True) == []
+
+    read_json = chat_store._read_json
+    reads = []
+
+    def observe_read(candidate: Path) -> dict:
+        if candidate == path:
+            reads.append(candidate)
+        return read_json(candidate)
+
+    monkeypatch.setattr(chat_store, "_read_json", observe_read)
+    _, created = store.create_turn(sid, client_turn_id="new-managed", message="New App request")
+    assert created
+    if file_change_fingerprint(path) is not None:
+        assert not reads
+
+    store.update_turn(sid, tid, response={"message": "Updated App answer"})
+    replay, created = store.create_turn(sid, **request)
+    assert not created
+    assert replay["turn_id"] == tid and replay["response"]["message"] == "Updated App answer"
+    with pytest.raises(ValueError, match="different request"):
+        store.create_turn(sid, **{**request, "message": "Conflicting App request"})
+
+
 @pytest.mark.parametrize("fault", [None, "changed-turn", "failed", "wrong-invocation", "immutable"])
 def test_completed_native_retention_is_fenced_and_private(tmp_path: Path, fault: str | None) -> None:
     store, sid = _dedup_store(tmp_path)