Skip unrelated transcript reads when opening a bound conversation

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:

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,

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