Scopes Phase 2b: scope option on chat, representation, context, and search

src/crud/__init__.py

@@ -51,6 +51,7 @@
     get_scope_session_names,
     get_scopes,
     remove_session_from_scope,
+    resolve_scope_peers,
 )
 from .session import (
     SessionDeletionResult,
@@ -135,6 +136,7 @@
     "get_scope_session_names",
     "get_scopes",
     "remove_session_from_scope",
+    "resolve_scope_peers",
     # Session
     "SessionDeletionResult",
     "get_sessions",

src/crud/message.py

@@ -699,19 +699,29 @@ async def _semantic_search_messages(
     after_date: datetime | None = None,
     before_date: datetime | None = None,
     observer: str | None = None,
+    allowed_sessions: list[str] | None = None,
 ) -> list[tuple[list[models.Message], list[models.Message]]]:
     """Run semantic message search with optional temporal filters.
 
     When observer is provided and session_name is None, results are
-    scoped to sessions the observer has any membership record in.
+    scoped to sessions the observer has any membership record in. When
+    allowed_sessions is provided, that membership scope is further
+    intersected with the allowlist (fail-closed: empty result on empty
+    intersection).
     """
     # Pre-fetch peer session scope if needed (short-lived DB session)
     allowed_session_names: list[str] | None = None
-    if observer and not session_name:
-        async with tracked_db(f"{operation_name}.peer_scope", read_only=True) as db:
-            allowed_session_names = await get_peer_session_names(
-                db, workspace_name, observer
-            )
+    if not session_name and (observer or allowed_sessions is not None):
+        if observer:
+            async with tracked_db(f"{operation_name}.peer_scope", read_only=True) as db:
+                allowed_session_names = await get_peer_session_names(
+                    db, workspace_name, observer
+                )
+            if allowed_sessions is not None:
+                scope = set(allowed_sessions)
+                allowed_session_names = [s for s in allowed_session_names if s in scope]
+        else:
+            allowed_session_names = list(allowed_sessions or [])
         if not allowed_session_names:
             return []
 
@@ -768,6 +778,7 @@ async def search_messages(
     context_window: int = 2,
     embedding: list[float] | None = None,
     observer: str | None = None,
+    allowed_sessions: list[str] | None = None,
 ) -> list[tuple[list[models.Message], list[models.Message]]]:
     """
     Search for messages using semantic similarity and return conversation snippets.
@@ -809,6 +820,7 @@ async def search_messages(
         context_window=context_window,
         operation_name="message.search_messages",
         observer=observer,
+        allowed_sessions=allowed_sessions,
     )
 
 
@@ -856,6 +868,7 @@ async def grep_messages(
     limit: int = 10,
     context_window: int = 2,
     observer: str | None = None,
+    allowed_sessions: list[str] | None = None,
 ) -> list[tuple[list[models.Message], list[models.Message]]]:
     """
     Search for messages containing specific text (case-insensitive substring match).
@@ -879,10 +892,18 @@ async def grep_messages(
     async with tracked_db("message.grep_messages", read_only=True) as db:
         # Pre-fetch peer session scope if needed
         allowed_session_names = None
-        if observer and not session_name:
-            allowed_session_names = await get_peer_session_names(
-                db, workspace_name, observer
-            )
+        if not session_name and (observer or allowed_sessions is not None):
+            if observer:
+                allowed_session_names = await get_peer_session_names(
+                    db, workspace_name, observer
+                )
+                if allowed_sessions is not None:
+                    scope = set(allowed_sessions)
+                    allowed_session_names = [
+                        s for s in allowed_session_names if s in scope
+                    ]
+            else:
+                allowed_session_names = list(allowed_sessions or [])
             if not allowed_session_names:
                 return []
 
@@ -908,6 +929,7 @@ async def get_messages_by_date_range(
     limit: int = 20,
     order: str = "desc",
     observer: str | None = None,
+    allowed_sessions: list[str] | None = None,
 ) -> list[models.Message]:
     """
     Get messages within a date range.
@@ -928,10 +950,16 @@ async def get_messages_by_date_range(
     """
     # Pre-fetch peer session scope if needed
     allowed_session_names = None
-    if observer and not session_name:
-        allowed_session_names = await get_peer_session_names(
-            db, workspace_name, observer
-        )
+    if not session_name and (observer or allowed_sessions is not None):
+        if observer:
+            allowed_session_names = await get_peer_session_names(
+                db, workspace_name, observer
+            )
+            if allowed_sessions is not None:
+                scope = set(allowed_sessions)
+                allowed_session_names = [s for s in allowed_session_names if s in scope]
+        else:
+            allowed_session_names = list(allowed_sessions or [])
         if not allowed_session_names:
             return []
 
@@ -967,6 +995,7 @@ async def search_messages_temporal(
     context_window: int = 2,
     embedding: list[float] | None = None,
     observer: str | None = None,
+    allowed_sessions: list[str] | None = None,
 ) -> list[tuple[list[models.Message], list[models.Message]]]:
     """
     Search for messages using semantic similarity with optional date filtering.
@@ -1011,4 +1040,5 @@ async def search_messages_temporal(
         context_window=context_window,
         operation_name="message.search_messages_temporal",
         observer=observer,
+        allowed_sessions=allowed_sessions,
     )

src/crud/representation.py

@@ -212,7 +212,7 @@ async def get_working_representation(
         self,
         *,
         db: AsyncSession | None = None,
-        session_name: str | None = None,
+        session_names: list[str] | None = None,
         include_semantic_query: str | None = None,
         embedding: list[float] | None = None,
         semantic_search_top_k: int | None = None,
@@ -228,7 +228,10 @@ async def get_working_representation(
         Args:
             db: Optional database session. If provided, uses it directly;
                 otherwise creates a new session via tracked_db.
-            session_name: Optional session to filter by
+            session_names: Optional session allowlist to filter by. Applied
+                uniformly to every query path (semantic, most-derived, and
+                recent). None means no session restriction; an empty list
+                fail-closes to an empty representation.
             include_semantic_query: Query for semantic search
             embedding: Pre-computed embedding for the semantic query.
             semantic_search_top_k: Number of semantic results
@@ -266,7 +269,7 @@ async def get_working_representation(
         if db is not None:
             return await self._get_working_representation_internal(
                 db,
-                session_name=session_name,
+                session_names=session_names,
                 include_semantic_query=include_semantic_query,
                 embedding=embedding,
                 semantic_search_top_k=semantic_search_top_k,
@@ -280,7 +283,7 @@ async def get_working_representation(
         ) as new_db:
             return await self._get_working_representation_internal(
                 new_db,
-                session_name=session_name,
+                session_names=session_names,
                 include_semantic_query=include_semantic_query,
                 embedding=embedding,
                 semantic_search_top_k=semantic_search_top_k,
@@ -295,7 +298,7 @@ async def _get_working_representation_internal(
         self,
         db: AsyncSession,
         *,
-        session_name: str | None = None,
+        session_names: list[str] | None = None,
         include_semantic_query: str | None = None,
         embedding: list[float] | None = None,
         semantic_search_top_k: int | None = None,
@@ -304,6 +307,12 @@ async def _get_working_representation_internal(
         max_observations: int = settings.DERIVER.WORKING_REPRESENTATION_MAX_OBSERVATIONS,
     ) -> Representation:
         """Internal implementation of get_working_representation."""
+        # Fail closed on an empty allowlist. This must short-circuit before
+        # any query: downstream stores drop an `IN ()` clause with an empty
+        # list (lancedb), which would silently widen the scope instead.
+        if session_names is not None and not session_names:
+            return Representation()
+
         total = max_observations
 
         # Calculate how many observations to get from each source
@@ -344,6 +353,7 @@ async def _get_working_representation_internal(
                 top_k=semantic_observations,
                 max_distance=semantic_search_max_distance,
                 embedding=embedding,
+                session_names=session_names,
             )
             representation.merge_representation(
                 Representation.from_documents(semantic_docs)
@@ -352,15 +362,15 @@ async def _get_working_representation_internal(
         # Get most derived observations if requested
         if include_most_derived:
             derived_docs = await self._query_documents_most_derived(
-                db, top_k=top_observations
+                db, top_k=top_observations, session_names=session_names
             )
             representation.merge_representation(
                 Representation.from_documents(derived_docs)
             )
 
         # Get recent observations
         recent_docs = await self._query_documents_recent(
-            db, top_k=recent_observations, session_name=session_name
+            db, top_k=recent_observations, session_names=session_names
         )
 
         representation.merge_representation(Representation.from_documents(recent_docs))
@@ -375,6 +385,7 @@ async def _query_documents_semantic(
         max_distance: float | None = None,
         level: str | None = None,
         embedding: list[float] | None = None,
+        session_names: list[str] | None = None,
     ) -> list[models.Document]:
         """Query documents by semantic similarity."""
         try:
@@ -386,6 +397,7 @@ async def _query_documents_semantic(
                     top_k,
                     max_distance,
                     embedding=embedding,
+                    session_names=session_names,
                 )
             else:
                 documents = await crud.query_documents(
@@ -397,6 +409,8 @@ async def _query_documents_semantic(
                     max_distance=max_distance,
                     top_k=top_k,
                     embedding=embedding,
+                    filters=self._build_filter_conditions(session_names=session_names)
+                    or None,
                 )
                 db.expunge_all()
                 return list(documents)
@@ -406,7 +420,7 @@ async def _query_documents_semantic(
             return []
 
     async def _query_documents_recent(
-        self, db: AsyncSession, top_k: int, session_name: str | None = None
+        self, db: AsyncSession, top_k: int, session_names: list[str] | None = None
     ) -> list[models.Document]:
         """Query most recent documents."""
         stmt = (
@@ -418,8 +432,8 @@ async def _query_documents_recent(
                 models.Document.observed == self.observed,
                 models.Document.deleted_at.is_(None),
                 *(
-                    [models.Document.session_name == session_name]
-                    if session_name is not None
+                    [models.Document.session_name.in_(session_names)]
+                    if session_names is not None
                     else []
                 ),
             )
@@ -432,7 +446,7 @@ async def _query_documents_recent(
         return list(documents)
 
     async def _query_documents_most_derived(
-        self, db: AsyncSession, top_k: int
+        self, db: AsyncSession, top_k: int, session_names: list[str] | None = None
     ) -> list[models.Document]:
         """Query most derived documents."""
         stmt = (
@@ -443,6 +457,11 @@ async def _query_documents_most_derived(
                 models.Document.observer == self.observer,
                 models.Document.observed == self.observed,
                 models.Document.deleted_at.is_(None),
+                *(
+                    [models.Document.session_name.in_(session_names)]
+                    if session_names is not None
+                    else []
+                ),
             )
             .order_by(
                 models.Document.times_derived.desc(),
@@ -479,6 +498,7 @@ async def _query_documents_for_level(
         count: int,
         max_distance: float | None = None,
         embedding: list[float] | None = None,
+        session_names: list[str] | None = None,
     ) -> list[models.Document]:
         """Query documents for a specific level."""
         documents = await crud.query_documents(
@@ -489,7 +509,7 @@ async def _query_documents_for_level(
             query=query,
             max_distance=max_distance,
             top_k=count,
-            filters=self._build_filter_conditions(level),
+            filters=self._build_filter_conditions(level, session_names=session_names),
             embedding=embedding,
         )
 
@@ -502,17 +522,24 @@ async def _query_documents_for_level(
     def _build_filter_conditions(
         self,
         level: str | None = None,
+        session_names: list[str] | None = None,
     ) -> dict[str, Any]:
         """
         Build filter conditions for document queries.
 
         Returns a flat dict of key-value pairs for vector store filtering.
+        Callers must not pass an empty session_names list — empty allowlists
+        fail closed before any query is issued (see
+        _get_working_representation_internal).
         """
         filters: dict[str, Any] = {}
 
         if level:
             filters["level"] = level
 
+        if session_names:
+            filters["session_name"] = {"in": session_names}
+
         return filters
 
 
@@ -525,7 +552,7 @@ async def get_working_representation(
     db: AsyncSession | None = None,
     observer: str,
     observed: str,
-    session_name: str | None = None,
+    session_names: list[str] | None = None,
     include_semantic_query: str | None = None,
     embedding: list[float] | None = None,
     semantic_search_top_k: int | None = None,
@@ -558,7 +585,7 @@ async def get_working_representation(
     )
     return await manager.get_working_representation(
         db=db,
-        session_name=session_name,
+        session_names=session_names,
         include_semantic_query=include_semantic_query,
         embedding=embedding,
         semantic_search_top_k=semantic_search_top_k,