From f905c49860a76aa61dfb4bc831ddbc7c6579e0cd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 12 Nov 2026 01:47:41 +0800 Subject: [PATCH] perf(chat): skip unrelated transcript reads in bound Session discovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/conversation_bindings.py | 19 ++++--- loopx/chat_store.py | 37 ++++++------- tests/test_lark_private_conversations.py | 52 +++++++++++++++++++ 3 files changed, 80 insertions(+), 28 deletions(-) diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index 0ce429598c..db9821d0ea 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/loopx/capabilities/native_chat/conversation_bindings.py @@ -424,14 +424,19 @@ def local_session_snapshot(self, binding_id: str, *, app_steward_entry: bool = F raise ValueError("conversation binding is no longer authorized") observation = dict(self.observe(row["transport_ref"])) store = self.controller.store + + def mentions_binding(session: dict[str, Any]) -> bool: + return any(isinstance(session.get(key), dict) and session[key].get("binding_id") == binding_id + for key in ("project_context", "steward_context")) + + # Keep aliases in the candidate set even when their own metadata is + # unrelated. Core must still see duplicate canonical Session identities. + hints = (session for path, session in store.iter_session_metadata() + if path.parent.name != session["session_id"] or mentions_binding(session)) sessions: list[dict[str, Any]] = [] - for summary in store.list_sessions(): - session = store.load_session(summary["session_id"]) - # Narrow the IO payload only. Core independently validates every - # matching context, audience and canonical Session identity. - if session and any(isinstance(session.get(key), dict) - and session[key].get("binding_id") == binding_id - for key in ("project_context", "steward_context")): + for hint in sorted(hints, key=lambda item: str(item.get("updated_at") or ""), reverse=True): + session = store.load_session(hint["session_id"]) + if session and mentions_binding(session): sessions.append(session) current = self.read() if next((item for item in current["bindings"] if item["binding_id"] == binding_id), None) != row: diff --git a/loopx/chat_store.py b/loopx/chat_store.py index f85ddfc332..5d719fb771 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -12,7 +12,7 @@ from pathlib import Path import re import threading -from typing import Any +from typing import Any, Iterator import uuid from weakref import WeakValueDictionary @@ -825,32 +825,27 @@ def session_candidates( candidates.append(payload) return candidates - def list_sessions( - self, - *, - goal_id: str | None = None, - agent_id: str | None = None, - channel_id: str | None = None, - ) -> list[dict[str, Any]]: - rows: list[dict[str, Any]] = [] + def list_sessions(self, *, goal_id: str | None = None, agent_id: str | None = None, + channel_id: str | None = None) -> list[dict[str, Any]]: + metadata = self.iter_session_metadata(goal_id=goal_id, agent_id=agent_id, channel_id=channel_id) + rows = (self.public_session(payload) for _, payload in metadata) + return sorted(rows, key=lambda item: str(item.get("updated_at") or ""), reverse=True) + + def iter_session_metadata(self, *, goal_id: str | None = None, agent_id: str | None = None, + channel_id: str | None = None) -> Iterator[tuple[Path, dict[str, Any]]]: + """Read discoverable Session records without reading transcript versions.""" for path in self.sessions_root.glob("*/session.json"): try: payload = _read_json(path) except UnicodeDecodeError: - # Discovery can skip damaged metadata, just as malformed JSON. - # Direct Session/Turn reads stay strict: unreadable queued work - # must never become false evidence that a conversation is idle. - continue - if payload.get("schema_version") != CHAT_SESSION_SCHEMA_VERSION: - continue - if goal_id and payload.get("goal_id") != goal_id: + # Direct Session reads remain strict. Discovery skips damaged metadata. continue - if agent_id and payload.get("agent_id") != agent_id: + if (payload.get("schema_version") != CHAT_SESSION_SCHEMA_VERSION + or (goal_id and payload.get("goal_id") != goal_id) + or (agent_id and payload.get("agent_id") != agent_id) + or (channel_id and _session_channel(payload) != channel_id)): continue - if channel_id and _session_channel(payload) != channel_id: - continue - rows.append(self.public_session(payload)) - return sorted(rows, key=lambda item: str(item.get("updated_at") or ""), reverse=True) + yield path, payload def append_message( self, diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 6771b93483..65bd275297 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -20,6 +20,58 @@ from loopx.extensions.lark.goal_topic_runtime import poll_lark_goal_topic_profile_once +def test_app_steward_discovery_only_projects_the_bound_session(tmp_path, monkeypatch): + from collections import Counter + from loopx import chat_store + from test_native_steward_private import finish, steward + + fixture = steward.__wrapped__(tmp_path) + store, runtime, provider, transport, binding, capture, _ = next(fixture) + try: + event = provider.event("steward-app", "discovery-source", "Read the fixture workspace") + assert transport.admit("steward-app", event)["status"] == "durably_accepted" + request = transport.core.read_request(identity_ref(binding["provider_ref"], event["message_id"])) + assert finish(runtime, request)["status"] == "completed" + original_id = request["session_id"] + other = store.create_session(goal_id="different-goal", agent_id="codex", adapter_kind="codex_app_server", + upstream_thread_id="other-thread", upstream_mode="chat") + bindings = transport.bindings + bindings.select_app_steward(binding["binding_id"], expected_revision=bindings.read()["revision"]) + before = capture.read_bytes(), bindings.path.read_bytes(), store.messages(original_id) + metadata, transcripts = Counter(), [] + read, revision = chat_store._read_json, store._event_revision + + def read_session(path): + if path.name == "session.json": + metadata[path.parent.name] += 1 + return read(path) + + def read_revision(path): + if path.name == "messages.jsonl": + transcripts.append(path.parent.name) + return revision(path) + + with monkeypatch.context() as observations: + observations.setattr(chat_store, "_read_json", read_session) + observations.setattr(store, "_event_revision", read_revision) + start = len(provider.calls) + assert bindings.app_steward_target() == {"binding_id": binding["binding_id"], "session_id": original_id} + assert sum("auth" in args and "--verify" in args for args in provider.calls[start:]) == 1 + assert metadata == {original_id: 2, other["session_id"]: 1} + assert transcripts == [original_id] + assert before == (capture.read_bytes(), bindings.path.read_bytes(), store.messages(original_id)) + + # The alias's own context is unrelated, but its declared ID still + # resolves to the same canonical bound record. Core must see both. + alias_path = store._session_path(other["session_id"]) + alias_path.write_text(json.dumps({**other, "session_id": original_id})) + with pytest.raises(ValueError, match="canonical bound Session identity is ambiguous"): + bindings.app_steward_target() + finally: + runtime.close() + fixture.close() + + @pytest.mark.parametrize("outcome", ["adopted", "closed", "interrupted"]) def test_private_steward_holds_original_request_during_executor_adoption(tmp_path, monkeypatch, outcome): from test_native_steward_private import finish, steward