diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 5cbdfca1c..cacc724bd 100644 --- a/src/crud/__init__.py +++ b/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", diff --git a/src/crud/message.py b/src/crud/message.py index e66b8f495..74133ef7b 100644 --- a/src/crud/message.py +++ b/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, ) diff --git a/src/crud/representation.py b/src/crud/representation.py index 558a5cce1..50fb3de3e 100644 --- a/src/crud/representation.py +++ b/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,7 +362,7 @@ 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) @@ -360,7 +370,7 @@ async def _get_working_representation_internal( # 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, diff --git a/src/crud/scope.py b/src/crud/scope.py index ac6cf2832..7480e0285 100644 --- a/src/crud/scope.py +++ b/src/crud/scope.py @@ -10,6 +10,7 @@ in a follow-up (DEV-1999). """ +from collections.abc import Sequence from logging import getLogger from sqlalchemy import Select, select @@ -18,7 +19,11 @@ from src import models, schemas from src.cache.client import safe_cache_delete -from src.exceptions import ConflictException, ResourceNotFoundException +from src.exceptions import ( + ConflictException, + ResourceNotFoundException, + ValidationException, +) from src.utils.scopes import ( SCOPE_KIND, is_scope_peer_configuration, @@ -186,6 +191,66 @@ async def get_scope( return peer +async def resolve_scope_peers( + db: AsyncSession, + workspace_name: str, + scope_names: Sequence[str], +) -> list[str]: + """ + Resolve unprefixed scope names to their backing scope-peer names. + + Used by the read routes that accept a ``scope`` option (chat, + representation, session context, workspace search) to turn user-facing + scope names into the observer peers that implement them. + + Args: + db: Database session + workspace_name: Name of the workspace + scope_names: Unprefixed scope names (duplicates are collapsed, + preserving first-seen order) + + Returns: + The backing scope-peer names, in first-requested order + + Raises: + ResourceNotFoundException: If any named scope does not exist + ValidationException: If a peer occupies a scope's reserved name + without the authoritative kind flag (a legacy collision) + """ + requested: list[str] = [] + seen: set[str] = set() + for name in scope_names: + if name not in seen: + seen.add(name) + requested.append(name) + + peer_names = [scope_peer_name(name) for name in requested] + if not peer_names: + return [] + + result = await db.execute( + select(models.Peer) + .where(models.Peer.workspace_name == workspace_name) + .where(models.Peer.name.in_(peer_names)) + ) + peers_by_name = {peer.name: peer for peer in result.scalars().all()} + + resolved: list[str] = [] + for name, peer_name in zip(requested, peer_names, strict=True): + peer = peers_by_name.get(peer_name) + if peer is None: + raise ResourceNotFoundException( + f"Scope {name} not found in workspace {workspace_name}" + ) + if not is_scope_peer_configuration(peer.configuration): + raise ValidationException( + f"'{name}' does not name a scope: a non-scope peer occupies " + + "its reserved name." + ) + resolved.append(peer_name) + return resolved + + async def get_scope_session_names( db: AsyncSession, workspace_name: str, diff --git a/src/dialectic/chat.py b/src/dialectic/chat.py index 9c1188033..324c54ab1 100644 --- a/src/dialectic/chat.py +++ b/src/dialectic/chat.py @@ -24,6 +24,7 @@ async def agentic_chat( observer: str, observed: str, reasoning_level: ReasoningLevel = "low", + session_names: list[str] | None = None, ) -> str: """ Answer a query about a peer using the agentic dialectic. @@ -35,6 +36,7 @@ async def agentic_chat( observer: The peer making the query observed: The peer being queried about reasoning_level: Level of reasoning to apply + session_names: Optional session allowlist restricting all recall Returns: The synthesized answer string @@ -77,6 +79,7 @@ async def agentic_chat( observer_peer_card=observer_peer_card, observed_peer_card=observed_peer_card, reasoning_level=reasoning_level, + session_names=session_names, ) return await agent.answer(query) @@ -89,6 +92,7 @@ async def agentic_chat_stream( observer: str, observed: str, reasoning_level: ReasoningLevel = "low", + session_names: list[str] | None = None, ) -> AsyncIterator[str]: """ Stream an answer to a query about a peer using the agentic dialectic. @@ -100,6 +104,7 @@ async def agentic_chat_stream( observer: The peer making the query observed: The peer being queried about reasoning_level: Level of reasoning to apply + session_names: Optional session allowlist restricting all recall Yields: Chunks of the response text as they are generated @@ -142,6 +147,7 @@ async def agentic_chat_stream( observer_peer_card=observer_peer_card, observed_peer_card=observed_peer_card, reasoning_level=reasoning_level, + session_names=session_names, ) async for chunk in agent.answer_stream(query): diff --git a/src/dialectic/core.py b/src/dialectic/core.py index 6f5b17abb..c727b95ab 100644 --- a/src/dialectic/core.py +++ b/src/dialectic/core.py @@ -69,6 +69,7 @@ def __init__( metric_key: str | None = None, reasoning_level: ReasoningLevel = "low", session_id: str | None = None, + session_names: list[str] | None = None, ): """ Initialize the dialectic agent. @@ -83,9 +84,13 @@ def __init__( metric_key: Optional key for logging metrics (if provided, agent won't log separately) reasoning_level: Level of reasoning to apply session_id: ID used for grouping traces (not session_name) + session_names: Optional session allowlist restricting all recall + (conclusions and messages) to these sessions; empty list + fails closed """ self.workspace_name: str = workspace_name self.session_name: str | None = session_name + self.session_names: list[str] | None = session_names self.session_id: str | None = session_id self.observer: str = observer self.observed: str = observed @@ -196,6 +201,7 @@ async def _prefetch_relevant_observations(self, query: str) -> str | None: limit=prefetch_limit, levels=["explicit"], embedding=query_embedding, + session_names=self.session_names, ) derived_repr = await search_memory( @@ -206,6 +212,7 @@ async def _prefetch_relevant_observations(self, query: str) -> str | None: limit=prefetch_limit, levels=["deductive", "inductive", "contradiction"], embedding=query_embedding, + session_names=self.session_names, ) if explicit_repr.is_empty() and derived_repr.is_empty(): @@ -295,6 +302,7 @@ async def _prepare_query( ] = await create_tool_executor( workspace_name=self.workspace_name, session_name=self.session_name, + session_names=self.session_names, observer=self.observer, observed=self.observed, history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT, diff --git a/src/routers/peers.py b/src/routers/peers.py index b4698bd5e..b1dd92bc8 100644 --- a/src/routers/peers.py +++ b/src/routers/peers.py @@ -5,6 +5,7 @@ from collections.abc import AsyncIterator from contextlib import suppress from time import perf_counter +from typing import Any from fastapi import APIRouter, Body, Depends, Path, Query, Response from fastapi.responses import StreamingResponse @@ -14,18 +15,21 @@ from src import crud, schemas from src.config import settings +from src.crud.message import get_peer_session_names from src.crud.session import is_peer_in_session from src.dependencies import db, read_db, tracked_db from src.dialectic.chat import agentic_chat, agentic_chat_stream from src.embedding_client import embedding_client from src.exceptions import ( AuthenticationException, + AuthorizationException, ResourceNotFoundException, ValidationException, ) from src.security import JWTParams, require_auth from src.telemetry import prometheus_metrics from src.telemetry.events import EmbeddingCallPurpose, GetContextEvent, emit +from src.utils.filter import MAX_SESSION_ALLOWLIST_ENTRIES, extract_session_allowlist from src.utils.scopes import is_scope_peer_name, validate_no_scope_peer_names from src.utils.search import search from src.utils.types import embedding_call_purpose @@ -38,6 +42,66 @@ ) +def _validate_scope_option( + *, + filters: dict[str, Any] | None, + session_id: str | None, + jwt_params: JWTParams, +) -> None: + """Enforce the v1 `scope` exclusions and auth rule (chat/representation). + + `scope` is mutually exclusive with `filters` and `session_id` (422), and a + scope's member sessions may exceed a peer's own membership, so scoped + reads require a workspace- or admin-level key (403 for peer-scoped keys). + """ + if filters is not None: + raise ValidationException("`scope` and `filters` are mutually exclusive") + if session_id: + raise ValidationException("`scope` and `session_id` are mutually exclusive") + if jwt_params.p is not None: + raise AuthorizationException("`scope` requires a workspace- or admin-level key") + + +async def _resolve_scope_option( + workspace_id: str, + scope: str | list[str], + *, + db_action: str, +) -> tuple[str | None, list[str] | None]: + """Map a validated `scope` option to (observer_override, session_allowlist). + + A single scope swaps the observer to the scope peer: conclusion recall is + then confined to the (scope, observed) collection and message recall to + the scope's session membership by existing observer semantics. A list of + scopes keeps the path peer as observer and returns the union of the + scopes' member sessions as an explicit allowlist (fail-closed when empty). + """ + async with tracked_db(db_action, read_only=True) as scope_db: + if isinstance(scope, str): + [scope_peer] = await crud.resolve_scope_peers( + scope_db, workspace_id, [scope] + ) + return scope_peer, None + + scope_peers = await crud.resolve_scope_peers(scope_db, workspace_id, scope) + union: list[str] = [] + seen: set[str] = set() + for scope_peer in scope_peers: + for session_name in await get_peer_session_names( + scope_db, workspace_id, scope_peer + ): + if session_name not in seen: + seen.add(session_name) + union.append(session_name) + + if len(union) > MAX_SESSION_ALLOWLIST_ENTRIES: + raise ValidationException( + "The scopes' combined membership exceeds the maximum of " + + f"{MAX_SESSION_ALLOWLIST_ENTRIES} sessions per request" + ) + return None, union + + @router.post( "/list", response_model=Page[schemas.Peer], @@ -206,6 +270,22 @@ async def chat( "Scope peers cannot be a chat target: no representation is formed of a scope." ) + # Scoped reads (DEV-1998): a single scope swaps the observer to the scope + # peer; a list of scopes becomes a session allowlist over their union. + observer = peer_id + scope_session_union: list[str] | None = None + if options.scope is not None: + _validate_scope_option( + filters=options.filters, + session_id=options.session_id, + jwt_params=jwt_params, + ) + observer_override, scope_session_union = await _resolve_scope_option( + workspace_id, options.scope, db_action="peers.chat.resolve_scope" + ) + if observer_override is not None: + observer = observer_override + # The session id arrives in the body, so require_auth can't gate on it. A # peer-scoped key may only scope a chat to a session its peer belongs to; # without this check it could read any session's messages (the dialectic @@ -218,6 +298,27 @@ async def chat( ): raise AuthenticationException("JWT not permissioned for this resource") + # Parse the session allowlist from filters (422 on unsupported keys/shapes). + session_allowlist = extract_session_allowlist(options.filters) + if ( + session_allowlist is not None + and options.session_id + and options.session_id not in session_allowlist + ): + raise ValidationException("session_id must be included in filters.session_id") + # A peer-scoped key may only name sessions its peer belongs to — the + # allowlist widens message recall the same way session_id does above. + if jwt_params.p is not None and session_allowlist: + async with tracked_db("peers.chat.session_scope_auth", read_only=True) as s_db: + member_sessions = set( + await get_peer_session_names(s_db, workspace_id, jwt_params.p) + ) + if not set(session_allowlist) <= member_sessions: + raise AuthenticationException("JWT not permissioned for this resource") + + if scope_session_union is not None: + session_allowlist = scope_session_union + # Get or create the peer to ensure it exists async with tracked_db("peers.chat.get_or_create_peer") as peer_db: peers_result = await crud.get_or_create_peers( @@ -252,9 +353,10 @@ async def format_sse_stream( workspace_name=workspace_id, session_name=options.session_id, query=options.query, - observer=peer_id, + observer=observer, observed=options.target if options.target is not None else peer_id, reasoning_level=options.reasoning_level, + session_names=session_allowlist, ) ), media_type="text/event-stream", @@ -264,11 +366,13 @@ async def format_sse_stream( workspace_name=workspace_id, session_name=options.session_id, query=options.query, - observer=peer_id, + # a single `scope` swaps the observer to the scope peer + observer=observer, # if target is given, that's the observed peer. otherwise, observer==observed # and it's answered from the omniscient Honcho perspective observed=options.target if options.target is not None else peer_id, reasoning_level=options.reasoning_level, + session_names=session_allowlist, ) # Prometheus metrics @@ -284,9 +388,6 @@ async def format_sse_stream( @router.post( "/{peer_id}/representation", response_model=schemas.RepresentationResponse, - dependencies=[ - Depends(require_auth(workspace_name="workspace_id", peer_name="peer_id")) - ], ) async def get_representation( workspace_id: str = Path(...), @@ -294,6 +395,9 @@ async def get_representation( options: schemas.PeerRepresentationGet = Body( ..., description="Options for getting the peer representation" ), + jwt_params: JWTParams = Depends( + require_auth(workspace_name="workspace_id", peer_name="peer_id") + ), ): """Get a curated subset of a Peer's Representation. A Representation is always a subset of the total knowledge about the Peer. The subset can be scoped and filtered in various ways. @@ -309,6 +413,33 @@ async def get_representation( "Scope peers cannot be a representation target: no representation is formed of a scope." ) + # Scoped reads (DEV-1998): a single scope swaps the observer to the scope + # peer; a list of scopes becomes a session allowlist over their union. + observer = peer_id + scope_session_union: list[str] | None = None + if options.scope is not None: + _validate_scope_option( + filters=options.filters, + session_id=options.session_id, + jwt_params=jwt_params, + ) + observer_override, scope_session_union = await _resolve_scope_option( + workspace_id, options.scope, db_action="peers.representation.resolve_scope" + ) + if observer_override is not None: + observer = observer_override + + # Parse the session allowlist from filters (422 on unsupported keys/shapes). + session_allowlist = extract_session_allowlist(options.filters) + if ( + session_allowlist is not None + and options.session_id + and options.session_id not in session_allowlist + ): + raise ValidationException("session_id must be included in filters.session_id") + if scope_session_union is not None: + session_allowlist = scope_session_union + try: embedding: list[float] | None = None if options.search_query: @@ -325,9 +456,12 @@ async def get_representation( # If no target specified, get global representation (omniscient Honcho perspective) representation = await crud.get_working_representation( workspace_id, - observer=peer_id, + # a single `scope` swaps the observer to the scope peer + observer=observer, observed=options.target if options.target is not None else peer_id, - session_name=options.session_id, + session_names=[options.session_id] + if options.session_id is not None + else session_allowlist, include_semantic_query=options.search_query, embedding=embedding, semantic_search_top_k=options.search_top_k, @@ -468,6 +602,14 @@ async def get_peer_context( This is useful for getting all the context needed about a peer without making multiple API calls. """ + # Scope peers may not appear on the generic peer-context surface: no + # representation is formed of a scope, and scoped reads go through the + # `scope` option on chat/representation/session-context instead. + validate_no_scope_peer_names( + [name for name in (peer_id, target) if name is not None], + action="Use the `scope` option on the read routes instead.", + ) + # If no target specified, get the peer's own context (self-observation) observed = target if target is not None else peer_id context_started = perf_counter() @@ -490,7 +632,7 @@ async def get_peer_context( workspace_id, observer=peer_id, observed=observed, - session_name=None, # Peer context is global, not session-scoped + session_names=None, # Peer context is global, not session-scoped include_semantic_query=search_query, embedding=embedding, semantic_search_top_k=search_top_k, diff --git a/src/routers/sessions.py b/src/routers/sessions.py index f7e57b0ef..069b712d9 100644 --- a/src/routers/sessions.py +++ b/src/routers/sessions.py @@ -17,6 +17,7 @@ from src.embedding_client import embedding_client from src.exceptions import ( AuthenticationException, + AuthorizationException, ResourceNotFoundException, ValidationException, ) @@ -53,7 +54,7 @@ async def _get_working_representation_task( *, observer: str, observed: str, - session_name: str | None, + session_names: list[str] | None, search_top_k: int | None, search_max_distance: float | None, include_most_derived: bool, @@ -69,7 +70,7 @@ async def _get_working_representation_task( last_message: Optional last message for semantic query observer: Name of the observer peer observed: Name of the observed peer - session_name: Optional session to filter by + session_names: Optional session allowlist to filter by search_top_k: Number of semantic-search-retrieved observations to include in the representation search_max_distance: Maximum distance to search for semantically relevant observations include_most_derived: Whether to include the most derived observations in the representation @@ -84,7 +85,7 @@ async def _get_working_representation_task( db=db, observer=observer, observed=observed, - session_name=session_name, + session_names=session_names, include_semantic_query=last_message, semantic_search_top_k=search_top_k, semantic_search_max_distance=search_max_distance, @@ -660,19 +661,17 @@ async def get_session_peers( @router.get( "/{session_id}/context", response_model=schemas.SessionContext, - dependencies=[ - Depends( - require_auth( - workspace_name="workspace_id", - session_name="session_id", - allow_member_read=True, - ) - ) - ], ) async def get_session_context( workspace_id: str = Path(...), session_id: str = Path(...), + jwt_params: JWTParams = Depends( + require_auth( + workspace_name="workspace_id", + session_name="session_id", + allow_member_read=True, + ) + ), db: AsyncSession = read_db, tokens: int | None = Query( None, @@ -697,6 +696,10 @@ async def get_session_context( None, description="A peer to get context for. If given, response will attempt to include representation and card from the perspective of that peer. Must be provided with `peer_target`.", ), + scope: str | None = Query( + None, + description="An (unprefixed) scope name to use as the perspective source: the representation and peer card of `peer_target` are read from the scope's observations instead of the global (or `peer_perspective`) view. Must be provided with `peer_target`; mutually exclusive with `peer_perspective`. Requires a workspace- or admin-level key.", + ), limit_to_session: bool = Query( default=False, description="Only used if `search_query` is provided. Whether to limit the representation to the session (as opposed to everything known about the target peer)", @@ -740,6 +743,29 @@ async def get_session_context( "peer_target must be provided if peer_perspective is provided" ) + # Scope peers may not appear on the generic perspective surface; the + # `scope` parameter is the supported path to a scoped perspective. + validate_no_scope_peer_names( + [name for name in (peer_target, peer_perspective) if name is not None], + action="Use the `scope` parameter instead.", + ) + + if scope is not None: + if peer_perspective: + raise ValidationException( + "`scope` and `peer_perspective` are mutually exclusive" + ) + if not peer_target: + raise ValidationException( + "peer_target must be provided if scope is provided" + ) + # A scope's perspective spans sessions beyond this one, so scoped + # reads require a workspace- or admin-level key. + if jwt_params.p is not None or jwt_params.s is not None: + raise AuthorizationException( + "`scope` requires a workspace- or admin-level key" + ) + if not peer_target: # No representation or card needed summary, messages = await _get_session_context_task( @@ -773,6 +799,12 @@ async def get_session_context( observer = peer_perspective or peer_target observed = peer_target + # A scope swaps the perspective source: the scope peer becomes the + # observer for both the working representation and the peer card, so the + # scoped collection and scoped card are read instead of the global ones. + if scope is not None: + [observer] = await crud.resolve_scope_peers(db, workspace_id, [scope]) + # Pre-compute embedding outside the DB session (best-effort) embedding: list[float] | None = None if search_query: @@ -793,7 +825,7 @@ async def get_session_context( search_query, observer=observer, observed=observed, - session_name=session_id if limit_to_session else None, + session_names=[session_id] if limit_to_session else None, search_top_k=search_top_k, search_max_distance=search_max_distance, include_most_derived=include_most_frequent, diff --git a/src/routers/workspaces.py b/src/routers/workspaces.py index 8b19fdfde..e3f95b696 100644 --- a/src/routers/workspaces.py +++ b/src/routers/workspaces.py @@ -7,11 +7,12 @@ from fastapi_pagination.ext.sqlalchemy import apaginate from sqlalchemy.ext.asyncio import AsyncSession -from src import crud, schemas +from src import crud, models, schemas from src.config import settings -from src.dependencies import db, read_db +from src.crud.message import get_peer_session_names +from src.dependencies import db, read_db, tracked_db from src.deriver.enqueue import enqueue_deletion, enqueue_dream -from src.exceptions import AuthenticationException +from src.exceptions import AuthenticationException, ValidationException from src.security import JWTParams, require_auth from src.utils.search import search @@ -141,16 +142,38 @@ async def delete_workspace( ) async def search_workspace( workspace_id: str = Path(...), - body: schemas.MessageSearchOptions = Body( + body: schemas.WorkspaceMessageSearchOptions = Body( ..., description="Message search parameters" ), ): """ Search messages in a Workspace using optional filters. Use `limit` to control the number of results returned. + + Pass `scope` to restrict the search to a scope's member sessions. A scope + with no member sessions returns no results (fail-closed). """ # take user-provided filter and add workspace_id to it filters = body.filters or {} + if body.scope is not None: + if "session_id" in filters: + raise ValidationException( + "`scope` and a 'session_id' filter are mutually exclusive" + ) + async with tracked_db( + "workspaces.search.resolve_scope", read_only=True + ) as scope_db: + [scope_peer] = await crud.resolve_scope_peers( + scope_db, workspace_id, [body.scope] + ) + scope_sessions = await get_peer_session_names( + scope_db, workspace_id, scope_peer + ) + if not scope_sessions: + # A scope with no member sessions matches nothing, not everything. + no_results: list[models.Message] = [] + return no_results + filters["session_id"] = {"in": scope_sessions} filters["workspace_id"] = workspace_id return await search(body.query, filters=filters, limit=body.limit) diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py index aa5dc01cb..3309f6be8 100644 --- a/src/schemas/__init__.py +++ b/src/schemas/__init__.py @@ -55,6 +55,7 @@ WorkspaceBase, WorkspaceCreate, WorkspaceGet, + WorkspaceMessageSearchOptions, WorkspaceUpdate, ) from src.schemas.configuration import ( @@ -155,6 +156,7 @@ "WorkspaceBase", "WorkspaceCreate", "WorkspaceGet", + "WorkspaceMessageSearchOptions", "WorkspaceUpdate", # internal "DocumentBase", diff --git a/src/schemas/api.py b/src/schemas/api.py index 3bfa65301..46563f3ff 100644 --- a/src/schemas/api.py +++ b/src/schemas/api.py @@ -211,6 +211,28 @@ class PeerRepresentationGet(BaseModel): session_id: str | None = Field( None, description="Optional session ID within which to scope the representation" ) + filters: dict[str, Any] | None = Field( + None, + description=( + "Optional filters to scope the representation. This endpoint " + "supports only the 'session_id' key: a session id, a list of " + 'session ids, or {"in": [...]}. When session_id is also set, it ' + "must be included in the allowlist." + ), + ) + scope: str | list[str] | None = Field( + None, + description=( + "Optional (unprefixed) scope name(s) to confine the representation. " + "A single scope reads the scope's own representation of the target " + "peer, formed only from the scope's member sessions. A list of " + "scopes restricts the representation to conclusions from the union " + "of the scopes' member sessions (explicit allowlist, fail-closed: " + "an empty union yields an empty representation). Mutually " + "exclusive with `filters` and `session_id`. Requires a workspace- " + "or admin-level key." + ), + ) target: str | None = Field( None, description="Optional peer ID to get the representation for, from the perspective of this peer", @@ -667,6 +689,20 @@ def sanitize_query(cls, v: str) -> str: return v.replace("\x00", "") +class WorkspaceMessageSearchOptions(MessageSearchOptions): + """Workspace-level message search options, extended with `scope`.""" + + scope: str | None = Field( + default=None, + description=( + "Optional (unprefixed) scope name restricting search to the " + "scope's member sessions. A scope with no member sessions returns " + "no results. Mutually exclusive with a 'session_id' key in " + "`filters`." + ), + ) + + # --------------------------------------------------------------------------- # Dialectic schemas # --------------------------------------------------------------------------- @@ -676,6 +712,29 @@ class DialecticOptions(BaseModel): session_id: str | None = Field( None, description="ID of the session to scope the representation to" ) + filters: dict[str, Any] | None = Field( + None, + description=( + "Optional filters to scope recall. This endpoint supports only the " + "'session_id' key: a session id, a list of session ids, or " + '{"in": [...]}. Recall (conclusions and messages) is restricted to ' + "the allowlist; unsupported keys are rejected. When session_id is " + "also set, it must be included in the allowlist." + ), + ) + scope: str | list[str] | None = Field( + None, + description=( + "Optional (unprefixed) scope name(s) to confine recall. A single " + "scope answers from the scope's own representation of the target " + "peer: conclusion recall is confined to what the scope observed " + "and message recall to the scope's member sessions. A list of " + "scopes restricts recall to the union of the scopes' member " + "sessions (explicit allowlist, fail-closed: an empty union " + "recalls nothing). Mutually exclusive with `filters` and " + "`session_id`. Requires a workspace- or admin-level key." + ), + ) target: str | None = Field( None, description="Optional peer to get the representation for, from the perspective of this peer", diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index d4df768b2..a3202d581 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -1,7 +1,7 @@ import asyncio import logging import weakref -from collections.abc import Callable +from collections.abc import Callable, Sequence from dataclasses import dataclass from datetime import datetime from typing import Any, cast @@ -1011,6 +1011,7 @@ async def get_recent_history( session_name: str | None, observed: str | None = None, token_limit: int = 8192, + allowed_sessions: list[str] | None = None, ) -> list[models.Message]: """ Retrieve recent conversation history. @@ -1042,7 +1043,11 @@ async def get_recent_history( # Return in chronological order return list(reversed(messages)) elif observed: + # Fail closed on an empty allowlist + if allowed_sessions is not None and not allowed_sessions: + return [] # Get recent messages from the observed peer across all sessions + # (restricted to the session allowlist when one is provided) stmt = ( select(models.Message) .where(models.Message.workspace_name == workspace_name) @@ -1050,6 +1055,8 @@ async def get_recent_history( .order_by(models.Message.created_at.desc()) .limit(50) # Limit to recent messages ) + if allowed_sessions is not None: + stmt = stmt.where(models.Message.session_name.in_(allowed_sessions)) result = await db.execute(stmt) messages = list(result.scalars().all()) # Return in chronological order @@ -1067,6 +1074,7 @@ async def search_memory( limit: int, levels: list[str] | None = None, embedding: list[float] | None = None, + session_names: list[str] | None = None, ) -> Representation: """ Search for observations in memory using semantic similarity. @@ -1087,10 +1095,17 @@ async def search_memory( Returns: Representation object containing relevant observations """ - # Build filter for levels if specified - filters: dict[str, Any] | None = None + # Fail closed on an empty allowlist — downstream stores drop empty IN + # clauses, which would silently widen scope. + if session_names is not None and not session_names: + return Representation() + + # Build filters for levels / session allowlist if specified + filters: dict[str, Any] = {} if levels: - filters = {"level": {"in": levels}} + filters["level"] = {"in": levels} + if session_names is not None: + filters["session_name"] = {"in": session_names} documents = await crud.query_documents( db=None, @@ -1099,7 +1114,7 @@ async def search_memory( observed=observed, query=query, top_k=limit, - filters=filters, + filters=filters or None, embedding=embedding, ) @@ -1112,6 +1127,7 @@ async def get_observation_context( session_name: str | None, message_ids: list[str], observer: str | None = None, + allowed_sessions: list[str] | None = None, ) -> list[models.Message]: """ Retrieve messages for given message IDs along with surrounding context. @@ -1136,12 +1152,18 @@ async def get_observation_context( # Pre-fetch peer session scope if needed allowed_session_names: list[str] | None = None - if observer and not session_name: - from src.crud.message import get_peer_session_names + if not session_name and (observer or allowed_sessions is not None): + if observer: + from src.crud.message import get_peer_session_names - allowed_session_names = await get_peer_session_names( - db, workspace_name, 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 [] @@ -1284,6 +1306,10 @@ class ToolContext: db_lock: asyncio.Lock # Optional resolved configuration for checking feature flags configuration: ResolvedConfiguration | None = None + # Optional session allowlist (dialectic filters). When set, message and + # conclusion recall is restricted to these sessions (intersected with + # observer membership); empty list fails closed. + session_names: list[str] | None = None # Telemetry context fields run_id: str | None = None agent_type: str | None = None # "dialectic", "deriver", "dreamer" @@ -1638,6 +1664,7 @@ async def _handle_get_recent_history( session_name=ctx.session_name, observed=ctx.observed, token_limit=ctx.history_token_limit, + allowed_sessions=ctx.session_names, ) if not history: return "No conversation history available" @@ -1683,15 +1710,24 @@ async def _handle_search_memory( "query_tokens": _estimate_tokens_safe(query), } - documents = await crud.query_documents( - db=None, - workspace_name=ctx.workspace_name, - observer=ctx.observer, - observed=ctx.observed, - query=query, - top_k=top_k, - embedding=query_embedding, - ) + # Restrict conclusion recall to the session allowlist when one is set. + # Empty allowlist fails closed (downstream stores drop empty IN clauses). + documents: Sequence[models.Document] + if ctx.session_names is not None and not ctx.session_names: + documents = [] + else: + documents = await crud.query_documents( + db=None, + workspace_name=ctx.workspace_name, + observer=ctx.observer, + observed=ctx.observed, + query=query, + top_k=top_k, + embedding=query_embedding, + filters={"session_name": {"in": ctx.session_names}} + if ctx.session_names is not None + else None, + ) mem = Representation.from_documents(documents) total_count = mem.len() if total_count == 0: @@ -1713,6 +1749,7 @@ async def _handle_search_memory( context_window=0, embedding=query_embedding, observer=ctx.observer, + allowed_sessions=ctx.session_names, ) if snippets: message_output = _format_message_snippets( @@ -1754,6 +1791,7 @@ async def _handle_get_observation_context( session_name=ctx.session_name, message_ids=tool_input["message_ids"], observer=ctx.observer, + allowed_sessions=ctx.session_names, ) if not messages: return f"No messages found for IDs {tool_input['message_ids']}" @@ -1796,6 +1834,7 @@ async def _handle_search_messages( context_window=2, embedding=query_embedding, observer=ctx.observer, + allowed_sessions=ctx.session_names, ) search_meta: dict[str, Any] = { "top_k": limit, @@ -1832,6 +1871,7 @@ async def _handle_grep_messages( limit=limit, context_window=context_window, observer=ctx.observer, + allowed_sessions=ctx.session_names, ) if not snippets: return f"No messages found containing '{text}'" @@ -1896,6 +1936,7 @@ async def _handle_get_messages_by_date_range( limit=limit, order=order, observer=ctx.observer, + allowed_sessions=ctx.session_names, ) msg_count = len(messages) messages_text = ( @@ -1968,6 +2009,7 @@ async def _handle_search_messages_temporal( before_date=before_date, limit=limit, context_window=context_window, + allowed_sessions=ctx.session_names, embedding=query_embedding, observer=ctx.observer, ) @@ -2224,6 +2266,14 @@ async def _handle_get_reasoning_chain( ctx: ToolContext, tool_input: dict[str, Any] ) -> str: """Handle get_reasoning_chain tool.""" + # Reasoning chains traverse provenance across sessions by design, so a + # session allowlist cannot be enforced on the traversal without exposing + # out-of-scope premises/conclusions. Fail closed rather than leak. + if ctx.session_names is not None: + return ( + "Reasoning-chain traversal is unavailable for session-scoped " + "queries. Use search_memory and message tools instead." + ) observation_id = tool_input.get("observation_id") if not observation_id: return "ERROR: 'observation_id' is required" @@ -2349,6 +2399,7 @@ async def create_tool_executor( run_id: str | None = None, agent_type: str | None = None, parent_category: str | None = None, + session_names: list[str] | None = None, ) -> Callable[[str, dict[str, Any]], Any]: """ Create a unified tool executor function for all agent operations. @@ -2389,6 +2440,7 @@ async def create_tool_executor( history_token_limit=history_token_limit, db_lock=shared_lock, configuration=configuration, + session_names=session_names, run_id=run_id, agent_type=agent_type, parent_category=parent_category, diff --git a/src/utils/filter.py b/src/utils/filter.py index 1394a8c6a..aed736f7a 100644 --- a/src/utils/filter.py +++ b/src/utils/filter.py @@ -1,7 +1,8 @@ import datetime -from collections.abc import Callable +from collections.abc import Callable, Sequence from logging import getLogger from typing import Any, TypeVar +from typing import cast as typing_cast from sqlalchemy import ColumnElement, Select, and_, case, cast, literal, not_, or_ from sqlalchemy.types import Numeric @@ -56,6 +57,67 @@ } +MAX_SESSION_ALLOWLIST_ENTRIES = 1000 + + +def extract_session_allowlist( + filters: dict[str, Any] | None, +) -> list[str] | None: + """Parse a recall-path ``filters`` body into a session allowlist. + + The dialectic and representation endpoints accept a constrained subset of + the filter DSL: only the ``session_id`` key, valued as a single id, a bare + list of ids, or ``{"in": [...]}``. Unsupported keys or shapes raise + FilterError (422) rather than being silently ignored — a dropped filter + on these endpoints would widen recall scope. + + Returns None when filters is None. An explicit empty list is preserved so + downstream consumers fail closed. + """ + if filters is None: + return None + + unsupported = set(filters) - {"session_id"} + if unsupported: + raise FilterError( + f"Unsupported filter key(s) for this endpoint: {sorted(unsupported)}. Only 'session_id' is supported." + ) + if "session_id" not in filters: + raise FilterError("filters must contain 'session_id'") + + value = filters["session_id"] + entries: list[Any] + if isinstance(value, str): + entries = [value] + elif isinstance(value, list): + entries = list(typing_cast(Sequence[Any], value)) + elif ( + isinstance(value, dict) + and set(typing_cast("dict[str, Any]", value)) == {"in"} + and isinstance(value["in"], list) + ): + entries = list(typing_cast(Sequence[Any], value["in"])) + else: + raise FilterError( + 'filters.session_id must be a session id, a list of session ids, or {"in": [...]}' + ) + + if len(entries) > MAX_SESSION_ALLOWLIST_ENTRIES: + raise FilterError( + f"filters.session_id supports at most {MAX_SESSION_ALLOWLIST_ENTRIES} sessions per request" + ) + + allowlist: list[str] = [] + seen: set[str] = set() + for entry in entries: + if not isinstance(entry, str) or not entry: + raise FilterError("filters.session_id entries must be non-empty strings") + if entry not in seen: + seen.add(entry) + allowlist.append(entry) + return allowlist + + def apply_filter( stmt: Select[tuple[T]], model_class: type[T], filters: dict[str, Any] | None = None ) -> Select[tuple[T]]: @@ -238,6 +300,16 @@ def _build_field_condition( if value == "*": return None + # Bare-list sugar on regular columns: {"session_id": ["a", "b"]} is + # shorthand for {"session_id": {"in": ["a", "b"]}}. JSONB columns are + # excluded — a bare list there keeps JSONB containment semantics. + if isinstance(value, list | tuple | set) and column_name not in ( + "h_metadata", + "configuration", + "internal_metadata", + ): + value = {"in": list(typing_cast(Sequence[Any], value))} + # Handle comparison operators vs regular values if isinstance(value, dict): # Check if this is a comparison operators dict by looking for known operators diff --git a/src/vector_store/lancedb.py b/src/vector_store/lancedb.py index 0b6219715..1b4c48806 100644 --- a/src/vector_store/lancedb.py +++ b/src/vector_store/lancedb.py @@ -298,10 +298,15 @@ def _build_where_clause(self, filters: dict[str, Any]) -> str | None: if not _VALID_IDENTIFIER_PATTERN.match(key): raise ValueError(f"Invalid filter key: {key!r}") - # Check if value is a dict with "in" operator - if isinstance(value, dict) and "in" in value: - # IN clause for list membership - in_values = cast(Sequence[Any], value["in"]) + # Membership: dict form {"in": [...]} or bare-list sugar + if (isinstance(value, dict) and "in" in value) or isinstance( + value, list | tuple | set + ): + in_values = ( + cast(Sequence[Any], value["in"]) + if isinstance(value, dict) + else list(cast(Sequence[Any], value)) + ) if in_values: escaped_values = [ f"'{str(v).replace(chr(39), chr(39) + chr(39))}'" @@ -310,6 +315,11 @@ def _build_where_clause(self, filters: dict[str, Any]) -> str | None: for v in in_values ] conditions.append(f"{key} IN ({', '.join(escaped_values)})") + else: + # An empty membership list matches nothing. Emitting no + # condition would silently widen the result set + # (fail-open); force an always-false condition instead. + conditions.append("1 = 0") # Handle string values with proper quoting elif isinstance(value, str): # Escape single quotes in the value diff --git a/src/vector_store/turbopuffer.py b/src/vector_store/turbopuffer.py index ded0c691b..e66a94dc4 100644 --- a/src/vector_store/turbopuffer.py +++ b/src/vector_store/turbopuffer.py @@ -252,6 +252,9 @@ def _build_filters(self, filters: dict[str, Any]) -> Filter | None: # Membership filter using "In" operator in_values = cast(Sequence[Any], value["in"]) filter_list.append((key, "In", in_values)) + elif isinstance(value, list | tuple | set): + # Bare-list sugar: same membership semantics as {"in": [...]} + filter_list.append((key, "In", list(cast(Sequence[Any], value)))) else: # Simple equality filter using "Eq" operator filter_list.append((key, "Eq", cast(Any, value))) diff --git a/tests/conftest.py b/tests/conftest.py index af35c9990..d68d89f2c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -833,6 +833,7 @@ async def mock_tracked_db_context(_: str | None = None, *, read_only: bool = Fal "src.deriver.consumer.tracked_db", "src.deriver.enqueue.tracked_db", "src.routers.peers.tracked_db", + "src.routers.workspaces.tracked_db", "src.crud.representation.tracked_db", "src.dreamer.orchestrator.tracked_db", "src.dreamer.dream_scheduler.tracked_db", diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py index 7744e763b..ca20079e0 100644 --- a/tests/crud/test_representation_manager.py +++ b/tests/crud/test_representation_manager.py @@ -232,6 +232,206 @@ async def test_query_documents_most_derived_ties_break_by_recency( assert contents[1:] == ["tie 2", "tie 1", "tie 0"] +class TestRepresentationManagerSessionScoping: + """Tests that the session allowlist is applied uniformly to every query path. + + Regression for DEV-1994: session_name used to be applied only to the + recent-documents query; the semantic and most-derived paths ignored it, + so limit_to_session leaked cross-session conclusions. + """ + + async def _setup( + self, + db_session: AsyncSession, + test_workspace: models.Workspace, + test_peer: models.Peer, + ) -> tuple[models.Session, models.Session, RepresentationManager]: + """Create two sessions and documents in each, plus a session-less doc.""" + test_peer2 = models.Peer( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(test_peer2) + await db_session.flush() + + session_a = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + session_b = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add_all([session_a, session_b]) + await db_session.flush() + + collection = models.Collection( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + db_session.add(collection) + await db_session.flush() + + db_session.add_all( + [ + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content="in-scope observation", + session_name=session_a.name, + times_derived=1, + ), + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content="out-of-scope observation", + session_name=session_b.name, + times_derived=100, + ), + # Dream-produced documents have no session_name; a session + # allowlist must exclude them (fail-closed). + models.Document( + workspace_name=test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + content="sessionless dream observation", + session_name=None, + times_derived=50, + ), + ] + ) + await db_session.flush() + + manager = RepresentationManager( + test_workspace.name, + observer=test_peer.name, + observed=test_peer2.name, + ) + return session_a, session_b, manager + + @pytest.mark.asyncio + async def test_recent_respects_session_allowlist( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + test_workspace, test_peer = sample_data + session_a, _, manager = await self._setup(db_session, test_workspace, test_peer) + + results = await manager._query_documents_recent( # pyright: ignore[reportPrivateUsage] + db_session, top_k=10, session_names=[session_a.name] + ) + + contents = [doc.content for doc in results] + assert contents == ["in-scope observation"] + + @pytest.mark.asyncio + async def test_most_derived_respects_session_allowlist( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """The out-of-scope doc has far higher times_derived; it must still be excluded.""" + test_workspace, test_peer = sample_data + session_a, _, manager = await self._setup(db_session, test_workspace, test_peer) + + results = await manager._query_documents_most_derived( # pyright: ignore[reportPrivateUsage] + db_session, top_k=10, session_names=[session_a.name] + ) + + contents = [doc.content for doc in results] + assert contents == ["in-scope observation"] + + @pytest.mark.asyncio + async def test_semantic_passes_session_allowlist_as_filters( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """The semantic path must push the allowlist down to query_documents.""" + test_workspace, test_peer = sample_data + session_a, _, manager = await self._setup(db_session, test_workspace, test_peer) + + with patch( + "src.crud.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + await manager._query_documents_semantic( # pyright: ignore[reportPrivateUsage] + db_session, + query="anything", + top_k=5, + embedding=[0.1], + session_names=[session_a.name], + ) + + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"] == { + "session_name": {"in": [session_a.name]} + } + + @pytest.mark.asyncio + async def test_semantic_passes_no_filters_when_unscoped( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + test_workspace, test_peer = sample_data + _, _, manager = await self._setup(db_session, test_workspace, test_peer) + + with patch( + "src.crud.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + await manager._query_documents_semantic( # pyright: ignore[reportPrivateUsage] + db_session, + query="anything", + top_k=5, + embedding=[0.1], + ) + + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"] is None + + @pytest.mark.asyncio + async def test_working_representation_scoped_end_to_end( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """All blended paths active: only in-scope content may appear.""" + test_workspace, test_peer = sample_data + session_a, _, manager = await self._setup(db_session, test_workspace, test_peer) + + representation = await manager.get_working_representation( + db=db_session, + session_names=[session_a.name], + include_most_derived=True, + ) + + contents = [obs.content for obs in representation.explicit] + assert "in-scope observation" in contents + assert "out-of-scope observation" not in contents + assert "sessionless dream observation" not in contents + + @pytest.mark.asyncio + async def test_empty_allowlist_fails_closed( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """An empty allowlist must return an empty representation, not fall + back to unscoped behavior (downstream stores drop empty IN clauses).""" + test_workspace, test_peer = sample_data + _, _, manager = await self._setup(db_session, test_workspace, test_peer) + + representation = await manager.get_working_representation( + db=db_session, + session_names=[], + include_most_derived=True, + ) + + assert representation.explicit == [] + assert representation.deductive == [] + + class TestRepresentationManagerSave: @pytest.mark.asyncio async def test_save_representation_filters_blank_observations_before_embedding( diff --git a/tests/routes/test_scope_reads.py b/tests/routes/test_scope_reads.py new file mode 100644 index 000000000..0f053ad50 --- /dev/null +++ b/tests/routes/test_scope_reads.py @@ -0,0 +1,701 @@ +"""Tests for the `scope` option on the read routes (DEV-1998). + +A single scope swaps the observer to the scope peer, so recall is confined to +the (scope, observed) collection and the scope's member sessions by existing +observer semantics. A list of scopes keeps the path peer as observer and +restricts recall to the union of the scopes' member sessions (the DEV-1995 +session-allowlist arm). +""" + +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from nanoid import generate as generate_nanoid +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models +from src.config import settings +from src.models import Peer, Workspace +from src.security import JWTParams, create_jwt +from src.utils.scopes import scope_peer_name + + +def _create_scope(client: TestClient, workspace_name: str, scope_name: str): + response = client.post( + f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name} + ) + assert response.status_code in [200, 201] + return response + + +def _create_session( + client: TestClient, + workspace_name: str, + session_name: str | None = None, + **extra: Any, +) -> str: + session_name = session_name or str(generate_nanoid()) + response = client.post( + f"/v3/workspaces/{workspace_name}/sessions", + json={"id": session_name, **extra}, + ) + assert response.status_code in [200, 201] + return session_name + + +def _add_sessions_to_scope( + client: TestClient, workspace_name: str, scope_name: str, session_names: list[str] +) -> None: + response = client.post( + f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions", + json={"session_ids": session_names}, + ) + assert response.status_code == 200 + + +async def _seed_documents( + db_session: AsyncSession, + workspace_name: str, + *, + observer: str, + observed: str, + contents: list[tuple[str, str | None]], +) -> None: + """Seed a collection plus documents for an (observer, observed) pair. + + ``contents`` is a list of (content, session_name) tuples. + """ + collection = models.Collection( + workspace_name=workspace_name, + observer=observer, + observed=observed, + ) + db_session.add(collection) + await db_session.flush() + db_session.add_all( + [ + models.Document( + workspace_name=workspace_name, + observer=observer, + observed=observed, + content=content, + session_name=session_name, + ) + for content, session_name in contents + ] + ) + await db_session.commit() + + +async def _seed_legacy_collision_peer( + db_session: AsyncSession, workspace_name: str, scope_name: str +) -> None: + """Create a plain peer squatting on a scope's reserved internal name.""" + db_session.add( + models.Peer( + workspace_name=workspace_name, + name=scope_peer_name(scope_name), + ) + ) + await db_session.commit() + + +class TestScopeReadValidation: + """4xx paths shared by chat and representation. + + Chat validation happens before any LLM work, so these are safe to exercise. + """ + + def _chat( + self, client: TestClient, workspace: Workspace, peer: Peer, body: dict[str, Any] + ): + return client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat", + json={"query": "what do you know?", **body}, + ) + + def _representation( + self, client: TestClient, workspace: Workspace, peer: Peer, body: dict[str, Any] + ): + return client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation", + json=body, + ) + + def test_unknown_scope_404( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + unknown = str(generate_nanoid()) + assert ( + self._chat(client, workspace, peer, {"scope": unknown}).status_code == 404 + ) + assert ( + self._representation( + client, workspace, peer, {"scope": unknown} + ).status_code + == 404 + ) + + def test_unknown_scope_in_list_404( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + resp = self._representation( + client, workspace, peer, {"scope": [scope_name, str(generate_nanoid())]} + ) + assert resp.status_code == 404 + + async def test_non_scope_peer_as_scope_422( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + """A peer squatting on the reserved name without the kind flag is not a scope.""" + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + await _seed_legacy_collision_peer(db_session, workspace.name, scope_name) + + assert ( + self._chat(client, workspace, peer, {"scope": scope_name}).status_code + == 422 + ) + assert ( + self._representation( + client, workspace, peer, {"scope": scope_name} + ).status_code + == 422 + ) + + def test_scope_plus_filters_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + body = {"scope": scope_name, "filters": {"session_id": ["s1"]}} + assert self._chat(client, workspace, peer, body).status_code == 422 + assert self._representation(client, workspace, peer, body).status_code == 422 + + def test_scope_plus_session_id_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + body = {"scope": scope_name, "session_id": "s1"} + assert self._chat(client, workspace, peer, body).status_code == 422 + assert self._representation(client, workspace, peer, body).status_code == 422 + + def test_peer_scoped_jwt_403( + self, + client: TestClient, + sample_data: tuple[Workspace, Peer], + monkeypatch: pytest.MonkeyPatch, + ): + """A scope's sessions may exceed the peer's own membership: workspace/admin only.""" + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + + monkeypatch.setattr(settings.AUTH, "USE_AUTH", True) + monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret") + client.headers["Authorization"] = ( + f"Bearer {create_jwt(JWTParams(w=workspace.name, p=peer.name))}" + ) + + assert ( + self._chat(client, workspace, peer, {"scope": scope_name}).status_code + == 403 + ) + assert ( + self._representation( + client, workspace, peer, {"scope": scope_name} + ).status_code + == 403 + ) + + # A workspace-level key is allowed through validation (404 here only + # if the scope were unknown; representation of an empty scope is 200). + client.headers["Authorization"] = ( + f"Bearer {create_jwt(JWTParams(w=workspace.name))}" + ) + assert ( + self._representation( + client, workspace, peer, {"scope": scope_name} + ).status_code + == 200 + ) + + def test_scope_union_cap_422( + self, + client: TestClient, + sample_data: tuple[Workspace, Peer], + monkeypatch: pytest.MonkeyPatch, + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_a = _create_session(client, workspace.name) + session_b = _create_session(client, workspace.name) + _add_sessions_to_scope( + client, workspace.name, scope_name, [session_a, session_b] + ) + + monkeypatch.setattr("src.routers.peers.MAX_SESSION_ALLOWLIST_ENTRIES", 1) + resp = self._representation(client, workspace, peer, {"scope": [scope_name]}) + assert resp.status_code == 422 + assert "maximum" in resp.json()["detail"] + + +class TestRepresentationWithScope: + async def test_single_scope_reads_scope_collection( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + """A single scope swaps the observer: only the (scope, peer) collection is read.""" + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_a = _create_session(client, workspace.name) + session_b = _create_session(client, workspace.name) + _add_sessions_to_scope(client, workspace.name, scope_name, [session_a]) + + # Conclusions the scope observed (session A) ... + await _seed_documents( + db_session, + workspace.name, + observer=scope_peer_name(scope_name), + observed=peer.name, + contents=[("scoped fact about hiking", session_a)], + ) + # ... and global self-observations from another session + await _seed_documents( + db_session, + workspace.name, + observer=peer.name, + observed=peer.name, + contents=[("global fact about cooking", session_b)], + ) + + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation", + json={"scope": scope_name}, + ) + assert resp.status_code == 200 + representation = resp.json()["representation"] + assert "scoped fact about hiking" in representation + assert "global fact about cooking" not in representation + + async def test_scope_list_unions_member_sessions( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + """A scope list keeps the global observer and applies the union allowlist.""" + workspace, peer = sample_data + scope_a = str(generate_nanoid()) + scope_b = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_a) + _create_scope(client, workspace.name, scope_b) + session_a = _create_session(client, workspace.name) + session_b = _create_session(client, workspace.name) + session_c = _create_session(client, workspace.name) + _add_sessions_to_scope(client, workspace.name, scope_a, [session_a]) + _add_sessions_to_scope(client, workspace.name, scope_b, [session_b]) + + # All conclusions live in the GLOBAL (peer, peer) collection: only the + # union session-allowlist can explain the filtering below (this is the + # dynamic DEV-1995 arm, not the observer swap). + await _seed_documents( + db_session, + workspace.name, + observer=peer.name, + observed=peer.name, + contents=[ + ("fact from session a", session_a), + ("fact from session b", session_b), + ("fact from session c", session_c), + ("sessionless dream fact", None), + ], + ) + + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation", + json={"scope": [scope_a, scope_b]}, + ) + assert resp.status_code == 200 + representation = resp.json()["representation"] + assert "fact from session a" in representation + assert "fact from session b" in representation + assert "fact from session c" not in representation + assert "sessionless dream fact" not in representation + + def test_empty_scope_list_fails_closed( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + """A scope with no member sessions yields an empty representation.""" + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation", + json={"scope": [scope_name]}, + ) + assert resp.status_code == 200 + assert "fact" not in resp.json()["representation"] + + +class TestChatWithScope: + """Verify what the chat route hands the dialectic, without real LLM work. + + ``agentic_chat`` is mocked in conftest (``mock_llm_call_functions``); the + scoped peer-card fetch happens inside it and is covered end-to-end by the + session-context test. Here we assert the route passes the right observer / + observed / session_names — the wiring that keys the card fetch. + """ + + def test_single_scope_swaps_observer( + self, + client: TestClient, + sample_data: tuple[Workspace, Peer], + mock_llm_call_functions: dict[str, Any], + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat", + json={"query": "what do you know?", "scope": scope_name}, + ) + assert resp.status_code == 200 + + kwargs = mock_llm_call_functions["agentic_chat"].await_args.kwargs + # The scope peer is the observer; the path peer stays the observed + assert kwargs["observer"] == scope_peer_name(scope_name) + assert kwargs["observed"] == peer.name + # Single-scope confinement rides on observer semantics, not an allowlist + assert kwargs["session_names"] is None + + def test_scope_list_passes_union_allowlist( + self, + client: TestClient, + sample_data: tuple[Workspace, Peer], + mock_llm_call_functions: dict[str, Any], + ): + workspace, peer = sample_data + scope_a = str(generate_nanoid()) + scope_b = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_a) + _create_scope(client, workspace.name, scope_b) + session_a = _create_session(client, workspace.name) + session_b = _create_session(client, workspace.name) + _add_sessions_to_scope(client, workspace.name, scope_a, [session_a]) + _add_sessions_to_scope(client, workspace.name, scope_b, [session_b]) + + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat", + json={"query": "what do you know?", "scope": [scope_a, scope_b]}, + ) + assert resp.status_code == 200 + + kwargs = mock_llm_call_functions["agentic_chat"].await_args.kwargs + # Union path: the path peer stays the observer, the allowlist is the union + assert kwargs["observer"] == peer.name + assert kwargs["observed"] == peer.name + assert set(kwargs["session_names"]) == {session_a, session_b} + + +class TestWorkspaceSearchWithScope: + def _seed_message( + self, client: TestClient, workspace_name: str, session_name: str, peer: Peer + ) -> None: + resp = client.post( + f"/v3/workspaces/{workspace_name}/sessions/{session_name}/messages", + json={ + "messages": [{"peer_id": peer.name, "content": "needle in haystack"}] + }, + ) + assert resp.status_code == 201 + + def test_search_restricted_to_scope_sessions( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_a = _create_session(client, workspace.name, peers={peer.name: {}}) + session_b = _create_session(client, workspace.name, peers={peer.name: {}}) + _add_sessions_to_scope(client, workspace.name, scope_name, [session_a]) + self._seed_message(client, workspace.name, session_a, peer) + self._seed_message(client, workspace.name, session_b, peer) + + # Unscoped: both sessions' messages match + resp = client.post( + f"/v3/workspaces/{workspace.name}/search", + json={"query": "needle"}, + ) + assert resp.status_code == 200 + assert {m["session_id"] for m in resp.json()} == {session_a, session_b} + + # Scoped: only the scope's member session + resp = client.post( + f"/v3/workspaces/{workspace.name}/search", + json={"query": "needle", "scope": scope_name}, + ) + assert resp.status_code == 200 + results = resp.json() + assert results + assert {m["session_id"] for m in results} == {session_a} + + def test_empty_scope_returns_no_results( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_a = _create_session(client, workspace.name, peers={peer.name: {}}) + self._seed_message(client, workspace.name, session_a, peer) + + resp = client.post( + f"/v3/workspaces/{workspace.name}/search", + json={"query": "needle", "scope": scope_name}, + ) + assert resp.status_code == 200 + assert resp.json() == [] + + def test_unknown_scope_404( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, _ = sample_data + resp = client.post( + f"/v3/workspaces/{workspace.name}/search", + json={"query": "needle", "scope": str(generate_nanoid())}, + ) + assert resp.status_code == 404 + + def test_scope_plus_session_id_filter_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, _ = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + resp = client.post( + f"/v3/workspaces/{workspace.name}/search", + json={ + "query": "needle", + "scope": scope_name, + "filters": {"session_id": "s1"}, + }, + ) + assert resp.status_code == 422 + + +class TestSessionContextWithScope: + async def test_scope_swaps_perspective_source( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + """`scope` reads the scope's collection and the scoped peer card.""" + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + _add_sessions_to_scope(client, workspace.name, scope_name, [session_name]) + + await _seed_documents( + db_session, + workspace.name, + observer=scope_peer_name(scope_name), + observed=peer.name, + contents=[("scoped fact about hiking", session_name)], + ) + await _seed_documents( + db_session, + workspace.name, + observer=peer.name, + observed=peer.name, + contents=[("global fact about cooking", session_name)], + ) + await crud.set_peer_card( + db_session, + workspace.name, + peer_card=["SCOPED CARD"], + observer=scope_peer_name(scope_name), + observed=peer.name, + ) + await crud.set_peer_card( + db_session, + workspace.name, + peer_card=["GLOBAL CARD"], + observer=peer.name, + observed=peer.name, + ) + await db_session.commit() + + # Without scope: the global (self) perspective + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={"peer_target": peer.name}, + ) + assert resp.status_code == 200 + data = resp.json() + assert "global fact about cooking" in data["peer_representation"] + assert data["peer_card"] == ["GLOBAL CARD"] + + # With scope: the scope's perspective + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={"peer_target": peer.name, "scope": scope_name}, + ) + assert resp.status_code == 200 + data = resp.json() + assert "scoped fact about hiking" in data["peer_representation"] + assert "global fact about cooking" not in data["peer_representation"] + assert data["peer_card"] == ["SCOPED CARD"] + + def test_scope_requires_peer_target( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={"scope": scope_name}, + ) + assert resp.status_code == 422 + + def test_scope_and_peer_perspective_mutually_exclusive( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={ + "peer_target": peer.name, + "peer_perspective": peer.name, + "scope": scope_name, + }, + ) + assert resp.status_code == 422 + + def test_unknown_scope_404( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={"peer_target": peer.name, "scope": str(generate_nanoid())}, + ) + assert resp.status_code == 404 + + def test_narrow_keys_rejected_403( + self, + client: TestClient, + sample_data: tuple[Workspace, Peer], + monkeypatch: pytest.MonkeyPatch, + ): + """Peer- and session-scoped keys may not widen reads through a scope.""" + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + _add_sessions_to_scope(client, workspace.name, scope_name, [session_name]) + + monkeypatch.setattr(settings.AUTH, "USE_AUTH", True) + monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret") + url = f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context" + params = {"peer_target": peer.name, "scope": scope_name} + + # Peer-scoped key (member read grants access to the route, not to scope) + client.headers["Authorization"] = ( + f"Bearer {create_jwt(JWTParams(w=workspace.name, p=peer.name))}" + ) + assert client.get(url, params=params).status_code == 403 + + # Session-scoped key + client.headers["Authorization"] = ( + f"Bearer {create_jwt(JWTParams(w=workspace.name, s=session_name))}" + ) + assert client.get(url, params=params).status_code == 403 + + # Workspace-scoped key is allowed + client.headers["Authorization"] = ( + f"Bearer {create_jwt(JWTParams(w=workspace.name))}" + ) + assert client.get(url, params=params).status_code == 200 + + +class TestScopePeerGuardrailClosure: + """Scope peers are rejected on the generic perspective/context surfaces.""" + + def test_session_context_rejects_scope_peer_target( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={"peer_target": scope_peer_name(scope_name)}, + ) + assert resp.status_code == 422 + + def test_session_context_rejects_scope_peer_perspective( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + session_name = _create_session(client, workspace.name, peers={peer.name: {}}) + + resp = client.get( + f"/v3/workspaces/{workspace.name}/sessions/{session_name}/context", + params={ + "peer_target": peer.name, + "peer_perspective": scope_peer_name(scope_name), + }, + ) + assert resp.status_code == 422 + + def test_peer_context_rejects_scope_peer( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + scope_name = str(generate_nanoid()) + _create_scope(client, workspace.name, scope_name) + + # As the path-level peer + resp = client.get( + f"/v3/workspaces/{workspace.name}/peers/{scope_peer_name(scope_name)}/context" + ) + assert resp.status_code == 422 + + # As the target + resp = client.get( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/context", + params={"target": scope_peer_name(scope_name)}, + ) + assert resp.status_code == 422 diff --git a/tests/test_advanced_filters.py b/tests/test_advanced_filters.py index 768f29bc3..3b1874709 100644 --- a/tests/test_advanced_filters.py +++ b/tests/test_advanced_filters.py @@ -235,6 +235,86 @@ async def test_comparison_operators_filters( ), f"Unexpected message '{message_config['content']}' found in results for {description}" +@pytest.mark.asyncio +async def test_bare_list_membership_sugar( + client: TestClient, + sample_data: tuple[Workspace, Peer], +): + """A bare list on a regular column is shorthand for {"in": [...]}. + + JSONB metadata columns are excluded from the sugar: a bare list there + keeps JSONB containment semantics. + """ + test_workspace, test_peer = sample_data + + # Second peer so peer_id membership has something to exclude + peer2_name = str(generate_nanoid()) + client.post( + f"/v3/workspaces/{test_workspace.name}/peers", + json={"id": peer2_name}, + ) + + session_id = str(generate_nanoid()) + session_response = client.post( + f"/v3/workspaces/{test_workspace.name}/sessions", + json={ + "id": session_id, + "peer_names": {test_peer.name: {}, peer2_name: {}}, + }, + ) + assert session_response.status_code == 201 + + message_configs = [ + { + "content": "From peer one", + "peer_id": test_peer.name, + "metadata": {"tags": ["important", "urgent"]}, + }, + { + "content": "From peer two", + "peer_id": peer2_name, + "metadata": {"tags": ["normal"]}, + }, + ] + messages_response = client.post( + f"/v3/workspaces/{test_workspace.name}/sessions/{session_id}/messages", + json={"messages": message_configs}, + ) + assert messages_response.status_code == 201 + + def list_contents(filter_config: dict[str, Any]) -> list[str]: + response = client.post( + f"/v3/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list", + json={"filters": filter_config}, + ) + assert response.status_code == 200 + return [item["content"] for item in response.json()["items"]] + + # Bare list == membership on a regular column + assert list_contents({"peer_id": [test_peer.name]}) == ["From peer one"] + + # Multiple values + assert sorted(list_contents({"peer_id": [test_peer.name, peer2_name]})) == [ + "From peer one", + "From peer two", + ] + + # Equivalent to the explicit {"in": [...]} form + assert list_contents({"peer_id": [test_peer.name]}) == list_contents( + {"peer_id": {"in": [test_peer.name]}} + ) + + # Empty list matches nothing (fail-closed), never everything + assert list_contents({"peer_id": []}) == [] + + # JSONB metadata keeps containment semantics for bare lists: + # matches arrays containing ALL listed elements, not membership. + assert list_contents({"metadata": {"tags": ["important", "urgent"]}}) == [ + "From peer one" + ] + assert list_contents({"metadata": {"tags": ["important", "missing"]}}) == [] + + @pytest.mark.asyncio async def test_wildcard_filters( client: TestClient, sample_data: tuple[Workspace, Peer] diff --git a/tests/test_session_allowlist.py b/tests/test_session_allowlist.py new file mode 100644 index 000000000..a835cdc0c --- /dev/null +++ b/tests/test_session_allowlist.py @@ -0,0 +1,335 @@ +""" +Tests for the session allowlist (DEV-1995). + +Covers the constrained `filters` surface on dialectic/representation +(extract_session_allowlist), fail-closed conclusion recall (search_memory), +and the strict allowlist ∩ membership intersection in message cruds. +""" + +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest +from fastapi.testclient import TestClient +from nanoid import generate as generate_nanoid +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models +from src.exceptions import FilterError +from src.models import Peer, Workspace +from src.utils.agent_tools import search_memory +from src.utils.filter import ( + MAX_SESSION_ALLOWLIST_ENTRIES, + extract_session_allowlist, +) + + +class TestExtractSessionAllowlist: + def test_none_passthrough(self): + assert extract_session_allowlist(None) is None + + def test_single_id(self): + assert extract_session_allowlist({"session_id": "s1"}) == ["s1"] + + def test_bare_list(self): + assert extract_session_allowlist({"session_id": ["s1", "s2"]}) == ["s1", "s2"] + + def test_in_operator(self): + assert extract_session_allowlist({"session_id": {"in": ["s1"]}}) == ["s1"] + + def test_dedupes_preserving_order(self): + assert extract_session_allowlist({"session_id": ["s2", "s1", "s2"]}) == [ + "s2", + "s1", + ] + + def test_empty_list_preserved_for_fail_closed(self): + assert extract_session_allowlist({"session_id": []}) == [] + + def test_unsupported_key_rejected(self): + with pytest.raises(FilterError, match="Unsupported filter key"): + extract_session_allowlist({"peer_id": ["a"], "session_id": ["s1"]}) + + def test_missing_session_id_rejected(self): + with pytest.raises(FilterError, match="must contain"): + extract_session_allowlist({}) + + def test_bad_shapes_rejected(self): + for bad in [123, {"gte": "x"}, {"in": "s1"}, [1, 2], [""], None]: + with pytest.raises(FilterError): + extract_session_allowlist({"session_id": bad}) + + def test_cap_enforced(self): + too_many = [f"s{i}" for i in range(MAX_SESSION_ALLOWLIST_ENTRIES + 1)] + with pytest.raises(FilterError, match="at most"): + extract_session_allowlist({"session_id": too_many}) + + +class TestSearchMemoryAllowlist: + @pytest.mark.asyncio + async def test_allowlist_pushed_down_as_filters(self): + with patch( + "src.crud.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + await search_memory( + workspace_name="w", + observer="o", + observed="o", + query="q", + limit=5, + levels=["explicit"], + embedding=[0.1], + session_names=["s1", "s2"], + ) + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"] == { + "level": {"in": ["explicit"]}, + "session_name": {"in": ["s1", "s2"]}, + } + + @pytest.mark.asyncio + async def test_empty_allowlist_fails_closed_without_querying(self): + with patch( + "src.crud.query_documents", new=AsyncMock(return_value=[]) + ) as mock_query: + result = await search_memory( + workspace_name="w", + observer="o", + observed="o", + query="q", + limit=5, + embedding=[0.1], + session_names=[], + ) + mock_query.assert_not_awaited() + assert result.is_empty() + + +class TestMessageCrudAllowlistIntersection: + """allowlist ∩ observer-membership, fail-closed on empty intersection.""" + + async def _setup_two_sessions( + self, + client: TestClient, + workspace: Workspace, + peer: Peer, + ) -> tuple[str, str]: + ids: list[str] = [] + for marker in ("alpha", "beta"): + session_id = str(generate_nanoid()) + resp = client.post( + f"/v3/workspaces/{workspace.name}/sessions", + json={"id": session_id, "peer_names": {peer.name: {}}}, + ) + assert resp.status_code == 201 + resp = client.post( + f"/v3/workspaces/{workspace.name}/sessions/{session_id}/messages", + json={ + "messages": [ + { + "content": f"needle in {marker}", + "peer_id": peer.name, + } + ] + }, + ) + assert resp.status_code == 201 + ids.append(session_id) + return ids[0], ids[1] + + @pytest.mark.asyncio + async def test_grep_messages_intersects_allowlist( + self, + client: TestClient, + sample_data: tuple[Workspace, Peer], + ): + workspace, peer = sample_data + session_a, session_b = await self._setup_two_sessions(client, workspace, peer) + + snippets = await crud.grep_messages( + workspace_name=workspace.name, + session_name=None, + text="needle", + observer=peer.name, + allowed_sessions=[session_a], + ) + contents = [m.content for matches, _ in snippets for m in matches] + assert contents == ["needle in alpha"] + + # A session the observer is NOT a member of contributes nothing, + # even when allowlisted (strict intersection). + foreign = str(generate_nanoid()) + snippets = await crud.grep_messages( + workspace_name=workspace.name, + session_name=None, + text="needle", + observer=peer.name, + allowed_sessions=[foreign], + ) + assert snippets == [] + + # Both sessions allowlisted -> both found + snippets = await crud.grep_messages( + workspace_name=workspace.name, + session_name=None, + text="needle", + observer=peer.name, + allowed_sessions=[session_a, session_b], + ) + assert len(snippets) == 2 + + @pytest.mark.asyncio + async def test_get_messages_by_date_range_intersects_allowlist( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + workspace, peer = sample_data + session_a, _session_b = await self._setup_two_sessions(client, workspace, peer) + + messages = await crud.get_messages_by_date_range( + db_session, + workspace_name=workspace.name, + session_name=None, + observer=peer.name, + allowed_sessions=[session_a], + ) + assert [m.content for m in messages] == ["needle in alpha"] + + # Empty allowlist fails closed + messages = await crud.get_messages_by_date_range( + db_session, + workspace_name=workspace.name, + session_name=None, + observer=peer.name, + allowed_sessions=[], + ) + assert messages == [] + + +class TestChatRouteFilterValidation: + """Filter validation happens before any LLM work — safe to exercise.""" + + def _chat( + self, + client: TestClient, + workspace: Workspace, + peer: Peer, + body: dict[str, Any], + ): + return client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat", + json={"query": "what do you know?", **body}, + ) + + def test_unsupported_filter_key_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + resp = self._chat(client, workspace, peer, {"filters": {"peer_id": ["x"]}}) + assert resp.status_code == 422 + + def test_bad_filter_shape_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + resp = self._chat(client, workspace, peer, {"filters": {"session_id": 42}}) + assert resp.status_code == 422 + + def test_session_id_not_in_allowlist_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + resp = self._chat( + client, + workspace, + peer, + {"session_id": "s-outside", "filters": {"session_id": ["s1", "s2"]}}, + ) + assert resp.status_code == 422 + + def test_allowlist_cap_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + too_many = [f"s{i}" for i in range(MAX_SESSION_ALLOWLIST_ENTRIES + 1)] + resp = self._chat( + client, workspace, peer, {"filters": {"session_id": too_many}} + ) + assert resp.status_code == 422 + + +class TestRepresentationRouteFilters: + @pytest.mark.asyncio + async def test_representation_scoped_by_filters( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + workspace, peer = sample_data + + session_a = models.Session( + name=str(generate_nanoid()), workspace_name=workspace.name + ) + session_b = models.Session( + name=str(generate_nanoid()), workspace_name=workspace.name + ) + db_session.add_all([session_a, session_b]) + await db_session.flush() + + collection = models.Collection( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + ) + db_session.add(collection) + await db_session.flush() + + db_session.add_all( + [ + models.Document( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + content="fact from session a", + session_name=session_a.name, + ), + models.Document( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + content="fact from session b", + session_name=session_b.name, + ), + models.Document( + workspace_name=workspace.name, + observer=peer.name, + observed=peer.name, + content="sessionless dream fact", + session_name=None, + ), + ] + ) + await db_session.commit() + + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation", + json={"filters": {"session_id": [session_a.name]}}, + ) + assert resp.status_code == 200 + representation = resp.json()["representation"] + assert "fact from session a" in representation + assert "fact from session b" not in representation + assert "sessionless dream fact" not in representation + + def test_session_id_not_in_allowlist_422( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + resp = client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation", + json={"session_id": "s-out", "filters": {"session_id": ["s-in"]}}, + ) + assert resp.status_code == 422