From 0eaedc3223925dde787e3b08ae5cdf731dfcf6ae Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 04:01:20 +0800 Subject: [PATCH 1/2] fix(chat): reuse terminal observations during queued input deduplication Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/chat_ingress.py | 51 ++++++++++++++++++----- loopx/chat_store.py | 10 ----- tests/test_chat_store_input_validation.py | 38 +++++++++++++++++ 3 files changed, 78 insertions(+), 21 deletions(-) diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 21858e7ed4..e1ad1af415 100644 --- a/loopx/chat_ingress.py +++ b/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" diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 884cdc7603..0f3b17623d 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -1443,16 +1443,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 diff --git a/tests/test_chat_store_input_validation.py b/tests/test_chat_store_input_validation.py index bdf468df5c..c7120a5b0b 100644 --- a/tests/test_chat_store_input_validation.py +++ b/tests/test_chat_store_input_validation.py @@ -52,6 +52,44 @@ 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" + + @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) From 33659aeaf5288a00a1c877b1b9eec9f9606426e9 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 05:06:15 +0800 Subject: [PATCH 2/2] fix(chat): reuse terminal observations in managed App admission Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/chat_store.py | 32 ++++++++++---------- tests/test_chat_store_input_validation.py | 36 +++++++++++++++++++++++ 2 files changed, 51 insertions(+), 17 deletions(-) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 0f3b17623d..826b8127bd 100644 --- a/loopx/chat_store.py +++ b/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"): diff --git a/tests/test_chat_store_input_validation.py b/tests/test_chat_store_input_validation.py index c7120a5b0b..cca9da64e2 100644 --- a/tests/test_chat_store_input_validation.py +++ b/tests/test_chat_store_input_validation.py @@ -90,6 +90,42 @@ def observe_read(candidate: Path) -> dict: 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)