From e7cbcc8432b696caf5648a679923dc4d4f370918 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Tue, 28 Jul 2026 10:55:02 -0400 Subject: [PATCH] feat: session allowlist on dialectic and representation via filters (#882) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: apply session scoping to all working-representation query paths session_name was only applied to the recent-documents query in RepresentationManager; the semantic and most-derived paths ignored it, so limit_to_session leaked cross-session conclusions into perspectives. - Thread a session allowlist (session_names) uniformly through all three query paths; pushed down to pgvector and external vector stores - Accept a list so the upcoming session-allowlist API reuses this path - Fail closed on an empty allowlist (downstream stores drop empty IN clauses, which would silently widen scope) Fixes DEV-1994 Co-Authored-By: Claude Fable 5 * feat: bare-list membership sugar in the filter DSL {"session_id": ["s1", "s2"]} is now shorthand for {"session_id": {"in": [...]}} on regular columns, generically (peer_id, etc.). JSONB metadata columns are excluded — a bare list there keeps JSONB containment semantics, unchanged. Previously a bare list on a regular column compiled to a type-mismatched equality that matched nothing, so this is strictly additive. Also translates the same shape in the turbopuffer/lancedb filter builders, and fixes lancedb dropping empty IN clauses (fail-open) — an empty membership list now emits an always-false condition. Groundwork for DEV-1995 (session allowlist via the existing filters DSL, no new API params) Co-Authored-By: Claude Fable 5 * feat: session allowlist on dialectic and representation via filters Adds a constrained 'filters' body to peer.chat and /representation — the same DSL search and conclusions already accept, supporting only the session_id key (a session id, a bare list, or {"in": [...]}). Unsupported keys and shapes are rejected with 422, never silently ignored. Composes with session_id (must be included in the allowlist when both are given). Capped at 1,000 sessions per request. Enforcement is uniform at every recall chokepoint, fail-closed: - dialectic prefetch + search_memory: conclusion recall restricted to the allowlist; dream docs (session_name IS NULL) excluded - message tools (search/grep/date-range/temporal/context/history): strict intersection of allowlist and observer session membership - get_reasoning_chain: unavailable under an allowlist (chains traverse provenance across sessions and cannot be scoped without leaking) - empty allowlist short-circuits to empty results everywhere Auth: workspace keys pass the allowlist as-given; peer-scoped JWTs must be a member of every allowlisted session (403 otherwise), mirroring the existing single-session check. Fixes DEV-1995 Co-Authored-By: Claude Fable 5 * fix: fail closed on empty session allowlist across all filter builders Empty session allowlists relied solely on the early-return guard in _get_working_representation_internal. The layers below it were inconsistent, so a future direct caller (the DEV-1995 allowlist API) would silently widen scope instead of failing closed: - _build_filter_conditions used a truthiness check; an empty list was treated like None and dropped the filter. Now uses `is not None`, matching the recent/most-derived SQL paths. - turbopuffer emitted a bare `In []` with undocumented (possibly fail-open) semantics. Now emits an explicit always-false predicate, mirroring lancedb's `1 = 0`. Also extract the duplicated JSONB column tuple in filter.py to a JSONB_COLUMNS constant. Tests exercise each fail-closed guarantee at the layer it lives, rather than masking it behind the early-return guard. * fix: address tests * fix(crud): fail closed when session_name is outside the allowlist search/grep/history helpers scoped to a single session_name ignored the session_names allowlist entirely — a caller could read a session the allowlist forbids. The API routes guarded this with a 422, but the dialectic tools call these CRUD functions directly and bypassed it. Enforce it at the boundary: return [] when session_name is set and not in the allowlist, across _semantic_search_messages (covers search_messages + search_messages_temporal), grep_messages, get_messages_by_date_range, get_recent_history, and get_observation_context. Also rename the public param allowed_sessions -> session_names for consistency with representation.py / chat.py / peers.py; the resolved intersection keeps its distinct name allowed_session_names. * fix: test * fix(scopes): tighten and consolidate session allowlist per review Addresses review feedback on the session allowlist (DEV-1995). Behavior changes: - Auth gate on peers.chat now uses active membership (left_at IS NULL) via get_peer_session_names(active_only=True), matching the adjacent is_peer_in_session check on options.session_id. Previously a peer that had left a session was denied when naming it directly but permitted when naming it in filters.session_id. - Scoped conclusion recall is restricted to level == "explicit" (ALLOWLIST_SAFE_LEVELS). Dream-derived conclusions are stamped with a single session_name but synthesized across all sessions, so that stamp can't be scoped on. Applied at all four recall paths. Unscoped recall is unchanged. Follow-up to give conclusions an authoritative source-session set is tracked in DEV-2201. - The allowlist gate checks `is not None` rather than truthiness, so filters={"session_id": []} reaches it instead of being skipped. Refactors: - New crud.message.resolve_session_scope replaces four near-identical copies of the allowlist-membership intersection. Returns (allowlist, deny) and never returns an empty list, so the None vs [] distinction that external stores fail open on lives in one tested place. Takes db=None and opens its own short-lived session only when distinction that external stores fail open on lives in one tested place. Takes db=None and opens its own short-lived session only when an observer lookup is needed, preserving external-lookup-first ordering on the vector-store path. - extract_session_allowlist takes must_include, collapsing the session_id-in-allowlist check duplicated across both peer routes. - DialecticAgent._select_tools dedupes the two toolset-selection blocks and drops get_reasoning_chain under an allowlist, rather than paying for the schema plus a wasted turn to return a refusal. - Rename session_names -> session_allowlist across crud, agent tools, dialectic and routes, to remove the one-character ambiguity with session_name. Internal only; the public filters.session_id surface is unchanged. Docs: - session_allowlist documented across all message and recall entry points, including the None / [] / populated contract. - session_name marked deprecated for scoping. Not removed and not aliased: it also pins the query to one session, bypasses observer scoping, and drives session-history injection into the dialectic prompt, so it has no drop-in replacement. - Note at the Document branch in utils/filter.py that the raw-key fallback is load-bearing for session scoping. Tests: 20 -> 39 in tests/test_session_allowlist.py, covering the peer-scoped JWT gate (member, non-member, left-session, workspace-key bypass, empty allowlist), the resolve_session_scope tri-state including the no-DB-checkout path, must_include, and the level narrowing. --------- Co-authored-by: Claude Fable 5 --- src/crud/message.py | 181 +++++- src/crud/representation.py | 72 ++- src/dialectic/chat.py | 6 + src/dialectic/core.py | 40 +- src/routers/peers.py | 35 +- src/routers/sessions.py | 8 +- src/schemas/api.py | 19 + src/utils/agent_tools.py | 125 +++- src/utils/filter.py | 84 +++ src/utils/representation.py | 29 + tests/crud/test_representation_manager.py | 43 +- tests/test_session_allowlist.py | 679 ++++++++++++++++++++++ tests/utils/test_agent_tools.py | 65 +++ 13 files changed, 1276 insertions(+), 110 deletions(-) create mode 100644 tests/test_session_allowlist.py diff --git a/src/crud/message.py b/src/crud/message.py index ddd6df38..684ddb4d 100644 --- a/src/crud/message.py +++ b/src/crud/message.py @@ -55,11 +55,28 @@ async def get_peer_session_names( db: AsyncSession, workspace_name: str, peer_name: str, + *, + active_only: bool = False, ) -> list[str]: - """Get all session names where a peer has any membership record. + """Get all session names where a peer has a membership record. - Any membership record (regardless of joined_at/left_at) grants visibility - to all messages in that session. + By default any membership record (regardless of joined_at/left_at) grants + visibility to all messages in that session — this is the loose definition + recall scoping uses. + + Pass ``active_only=True`` for the strict definition (``left_at IS NULL``), + matching :func:`src.crud.session.is_peer_in_session`. The auth layer must + use the strict one so that a single peer-scoped key gets the same answer + whether it names a session directly or via a filter allowlist. + + Args: + db: Database session + workspace_name: Name of the workspace + peer_name: Name of the peer + active_only: Restrict to sessions the peer has not left + + Returns: + Distinct session names the peer has a matching membership record in. """ stmt = ( select(models.session_peers_table.c.session_name) @@ -67,10 +84,80 @@ async def get_peer_session_names( .where(models.session_peers_table.c.peer_name == peer_name) .distinct() ) + if active_only: + stmt = stmt.where(models.session_peers_table.c.left_at.is_(None)) result = await db.execute(stmt) return [row[0] for row in result.all()] +async def resolve_session_scope( + db: AsyncSession | None, + workspace_name: str, + session_name: str | None, + session_allowlist: list[str] | None, + observer: str | None, + *, + operation_name: str = "resolve_session_scope", +) -> tuple[list[str] | None, bool]: + """Resolve the effective session scope for a message query. + + Returns ``(allowed_session_names, deny)``: + + - ``allowed_session_names is None`` — apply no allowlist filter. Either the + query is unrestricted, or ``session_name`` already pins it to one session. + - a populated list — restrict the query to exactly these sessions. + - ``deny=True`` — the caller must return an empty result *without* querying. + + The distinction between ``None`` and an empty list is load-bearing: the + external vector stores drop an empty ``IN`` clause rather than matching + nothing, so collapsing the two would fail open. This function therefore + never returns an empty list — it returns ``deny=True`` instead. + + Touches the database only when an observer lookup is actually required, so + callers on the external-vector-store path don't check out a connection + before their network call. + + Args: + db: Database session to reuse. Pass None to let this function open its + own short-lived read-only session if (and only if) it needs one. + workspace_name: Name of the workspace + session_name: A single pinned session, if the caller named one + session_allowlist: Optional session allowlist. ``None`` is unrestricted; + an empty list fails closed. + observer: When set, scope is limited to this peer's sessions and then + intersected with ``session_allowlist`` + operation_name: Label for the self-managed DB session, when one is opened + + Returns: + Tuple of (allowlist to filter on or None, whether to deny outright). + """ + if session_name: + # A specific session was requested. Fail closed when the allowlist + # forbids it — routes guard this too, but other CRUD callers (the + # dialectic tools) don't, so enforce it at the boundary. + if session_allowlist is not None and session_name not in session_allowlist: + return None, True + return None, False + + if observer is None: + if session_allowlist is None: + return None, False + allowed = list(session_allowlist) + return (allowed, False) if allowed else (None, True) + + if db is not None: + allowed = await get_peer_session_names(db, workspace_name, observer) + else: + async with tracked_db(f"{operation_name}.peer_scope", read_only=True) as own_db: + allowed = await get_peer_session_names(own_db, workspace_name, observer) + + if session_allowlist is not None: + scope = set(session_allowlist) + allowed = [s for s in allowed if s in scope] + + return (allowed, False) if allowed else (None, True) + + def _apply_token_limit( base_conditions: list[ColumnElement[Any]], token_limit: int ) -> Select[tuple[models.Message]]: @@ -689,21 +776,29 @@ async def _semantic_search_messages( after_date: datetime | None = None, before_date: datetime | None = None, observer: str | None = None, + session_allowlist: 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 + session_allowlist 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 allowed_session_names: - return [] + # db=None: the helper opens its own short-lived session only if it needs + # an observer lookup, so the external-store path below stays the first + # thing that happens when no observer scoping applies. + allowed_session_names, deny = await resolve_session_scope( + None, + workspace_name, + session_name, + session_allowlist, + observer, + operation_name=operation_name, + ) + if deny: + return [] if settings.VECTOR_STORE.TYPE != "pgvector" and settings.VECTOR_STORE.MIGRATED: message_ids = await _search_messages_external( @@ -758,6 +853,7 @@ async def search_messages( context_window: int = 2, embedding: list[float] | None = None, observer: str | None = None, + session_allowlist: list[str] | None = None, ) -> list[tuple[list[models.Message], list[models.Message]]]: """ Search for messages using semantic similarity and return conversation snippets. @@ -768,12 +864,19 @@ async def search_messages( Args: workspace_name: Name of the workspace session_name: Name of the session (optional) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. query: Search query text limit: Maximum number of matching messages to return context_window: Number of messages before/after each match to include embedding: Optional pre-computed embedding observer: When provided and session_name is None, scope results to sessions this peer belongs to + session_allowlist: Optional session allowlist. None is unrestricted; an + empty list fails closed (empty result); a populated list is + intersected with the observer's session scope when observer is set Returns: List of tuples: (matched_messages, context_messages) @@ -799,6 +902,7 @@ async def search_messages( context_window=context_window, operation_name="message.search_messages", observer=observer, + session_allowlist=session_allowlist, ) @@ -846,6 +950,7 @@ async def grep_messages( limit: int = 10, context_window: int = 2, observer: str | None = None, + session_allowlist: list[str] | None = None, ) -> list[tuple[list[models.Message], list[models.Message]]]: """ Search for messages containing specific text (case-insensitive substring match). @@ -856,25 +961,29 @@ async def grep_messages( Args: workspace_name: Name of the workspace session_name: Name of the session (optional - searches all sessions if None) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. text: Text to search for (case-insensitive) limit: Maximum number of matching messages to return context_window: Number of messages before/after each match to include observer: When provided and session_name is None, scope results to sessions this peer belongs to + session_allowlist: Optional session allowlist. None is unrestricted; an + empty list fails closed (empty result); a populated list is + intersected with the observer's session scope when observer is set Returns: List of tuples: (matched_messages, context_messages) Each snippet may contain multiple matches if they were close together. """ 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 allowed_session_names: - return [] + allowed_session_names, deny = await resolve_session_scope( + db, workspace_name, session_name, session_allowlist, observer + ) + if deny: + return [] snippets = await _grep_messages_internal( db, @@ -898,6 +1007,7 @@ async def get_messages_by_date_range( limit: int = 20, order: str = "desc", observer: str | None = None, + session_allowlist: list[str] | None = None, ) -> list[models.Message]: """ Get messages within a date range. @@ -906,24 +1016,28 @@ async def get_messages_by_date_range( db: Database session workspace_name: Name of the workspace session_name: Name of the session (optional - searches all sessions if None) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. after_date: Return messages after this datetime before_date: Return messages before this datetime limit: Maximum messages to return order: Sort order - 'asc' for oldest first, 'desc' for newest first observer: When provided and session_name is None, scope results to sessions this peer belongs to + session_allowlist: Optional session allowlist. None is unrestricted; an + empty list fails closed (empty result); a populated list is + intersected with the observer's session scope when observer is set Returns: List of messages within the 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 allowed_session_names: - return [] + allowed_session_names, deny = await resolve_session_scope( + db, workspace_name, session_name, session_allowlist, observer + ) + if deny: + return [] stmt = select(models.Message).where(models.Message.workspace_name == workspace_name) @@ -957,6 +1071,7 @@ async def search_messages_temporal( context_window: int = 2, embedding: list[float] | None = None, observer: str | None = None, + session_allowlist: list[str] | None = None, ) -> list[tuple[list[models.Message], list[models.Message]]]: """ Search for messages using semantic similarity with optional date filtering. @@ -967,6 +1082,10 @@ async def search_messages_temporal( Args: workspace_name: Name of the workspace session_name: Name of the session (optional) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. query: Search query text after_date: Only return messages after this datetime before_date: Only return messages before this datetime @@ -975,6 +1094,9 @@ async def search_messages_temporal( embedding: Optional pre-computed embedding for the query observer: When provided and session_name is None, scope results to sessions this peer belongs to + session_allowlist: Optional session allowlist. None is unrestricted; an + empty list fails closed (empty result); a populated list is + intersected with the observer's session scope when observer is set Returns: List of tuples: (matched_messages, context_messages) @@ -1001,4 +1123,5 @@ async def search_messages_temporal( context_window=context_window, operation_name="message.search_messages_temporal", observer=observer, + session_allowlist=session_allowlist, ) diff --git a/src/crud/representation.py b/src/crud/representation.py index 69c9edc9..d4b86ffb 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -19,9 +19,11 @@ from src.telemetry.events import EmbeddingCallPurpose from src.telemetry.logging import accumulate_metric from src.utils.formatting import format_datetime_utc from src.utils.representation import ( + ALLOWLIST_SAFE_LEVELS, DeductiveObservation, ExplicitObservation, Representation, + allowlist_safe_levels, ) from src.utils.types import embedding_call_purpose @@ -213,7 +215,7 @@ class RepresentationManager: self, *, db: AsyncSession | None = None, - session_names: list[str] | None = None, + session_allowlist: list[str] | None = None, include_semantic_query: str | None = None, embedding: list[float] | None = None, semantic_search_top_k: int | None = None, @@ -229,7 +231,7 @@ class RepresentationManager: Args: db: Optional database session. If provided, uses it directly; otherwise creates a new session via tracked_db. - session_names: Optional session allowlist to filter by. Applied + session_allowlist: 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. @@ -270,7 +272,7 @@ class RepresentationManager: if db is not None: return await self._get_working_representation_internal( db, - session_names=session_names, + session_allowlist=session_allowlist, include_semantic_query=include_semantic_query, embedding=embedding, semantic_search_top_k=semantic_search_top_k, @@ -284,7 +286,7 @@ class RepresentationManager: ) as new_db: return await self._get_working_representation_internal( new_db, - session_names=session_names, + session_allowlist=session_allowlist, include_semantic_query=include_semantic_query, embedding=embedding, semantic_search_top_k=semantic_search_top_k, @@ -299,7 +301,7 @@ class RepresentationManager: self, db: AsyncSession, *, - session_names: list[str] | None = None, + session_allowlist: list[str] | None = None, include_semantic_query: str | None = None, embedding: list[float] | None = None, semantic_search_top_k: int | None = None, @@ -311,7 +313,7 @@ class RepresentationManager: # 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: + if session_allowlist is not None and not session_allowlist: return Representation() total = max_observations @@ -354,7 +356,7 @@ class RepresentationManager: top_k=semantic_observations, max_distance=semantic_search_max_distance, embedding=embedding, - session_names=session_names, + session_allowlist=session_allowlist, ) representation.merge_representation( Representation.from_documents(semantic_docs) @@ -363,7 +365,7 @@ class RepresentationManager: # Get most derived observations if requested if include_most_derived: derived_docs = await self._query_documents_most_derived( - db, top_k=top_observations, session_names=session_names + db, top_k=top_observations, session_allowlist=session_allowlist ) representation.merge_representation( Representation.from_documents(derived_docs) @@ -371,7 +373,7 @@ class RepresentationManager: # Get recent observations recent_docs = await self._query_documents_recent( - db, top_k=recent_observations, session_names=session_names + db, top_k=recent_observations, session_allowlist=session_allowlist ) representation.merge_representation(Representation.from_documents(recent_docs)) @@ -386,7 +388,7 @@ class RepresentationManager: max_distance: float | None = None, level: str | None = None, embedding: list[float] | None = None, - session_names: list[str] | None = None, + session_allowlist: list[str] | None = None, ) -> list[models.Document]: """Query documents by semantic similarity.""" try: @@ -398,7 +400,7 @@ class RepresentationManager: top_k, max_distance, embedding=embedding, - session_names=session_names, + session_allowlist=session_allowlist, ) else: documents = await crud.query_documents( @@ -410,7 +412,9 @@ class RepresentationManager: max_distance=max_distance, top_k=top_k, embedding=embedding, - filters=self._build_filter_conditions(session_names=session_names) + filters=self._build_filter_conditions( + session_allowlist=session_allowlist + ) or None, ) db.expunge_all() @@ -421,7 +425,7 @@ class RepresentationManager: return [] async def _query_documents_recent( - self, db: AsyncSession, top_k: int, session_names: list[str] | None = None + self, db: AsyncSession, top_k: int, session_allowlist: list[str] | None = None ) -> list[models.Document]: """Query most recent documents.""" stmt = ( @@ -433,8 +437,13 @@ class RepresentationManager: models.Document.observed == self.observed, models.Document.deleted_at.is_(None), *( - [models.Document.session_name.in_(session_names)] - if session_names is not None + [ + models.Document.session_name.in_(session_allowlist), + # Only levels with a trustworthy session stamp are + # scopeable — see ALLOWLIST_SAFE_LEVELS. + models.Document.level.in_(ALLOWLIST_SAFE_LEVELS), + ] + if session_allowlist is not None else [] ), ) @@ -447,7 +456,7 @@ class RepresentationManager: return list(documents) async def _query_documents_most_derived( - self, db: AsyncSession, top_k: int, session_names: list[str] | None = None + self, db: AsyncSession, top_k: int, session_allowlist: list[str] | None = None ) -> list[models.Document]: """Query most derived documents.""" stmt = ( @@ -459,8 +468,13 @@ class RepresentationManager: models.Document.observed == self.observed, models.Document.deleted_at.is_(None), *( - [models.Document.session_name.in_(session_names)] - if session_names is not None + [ + models.Document.session_name.in_(session_allowlist), + # Only levels with a trustworthy session stamp are + # scopeable — see ALLOWLIST_SAFE_LEVELS. + models.Document.level.in_(ALLOWLIST_SAFE_LEVELS), + ] + if session_allowlist is not None else [] ), ) @@ -499,7 +513,7 @@ class RepresentationManager: count: int, max_distance: float | None = None, embedding: list[float] | None = None, - session_names: list[str] | None = None, + session_allowlist: list[str] | None = None, ) -> list[models.Document]: """Query documents for a specific level.""" documents = await crud.query_documents( @@ -510,7 +524,9 @@ class RepresentationManager: query=query, max_distance=max_distance, top_k=count, - filters=self._build_filter_conditions(level, session_names=session_names), + filters=self._build_filter_conditions( + level, session_allowlist=session_allowlist + ), embedding=embedding, ) @@ -523,13 +539,13 @@ class RepresentationManager: def _build_filter_conditions( self, level: str | None = None, - session_names: list[str] | None = None, + session_allowlist: 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 + Callers must not pass an empty session_allowlist list — empty allowlists fail closed before any query is issued (see _get_working_representation_internal). """ @@ -542,8 +558,12 @@ class RepresentationManager: # an empty `in` so downstream stores fail closed, matching # _query_documents_recent / _query_documents_most_derived. Truthiness # here would silently drop the filter and widen scope. - if session_names is not None: - filters["session_name"] = {"in": session_names} + if session_allowlist is not None: + filters["session_name"] = {"in": session_allowlist} + # Only levels with a trustworthy session stamp are scopeable. This + # overrides any narrower `level` above; an empty intersection emits + # `{"in": []}`, which matches nothing rather than everything. + filters["level"] = {"in": allowlist_safe_levels([level] if level else None)} return filters @@ -557,7 +577,7 @@ async def get_working_representation( db: AsyncSession | None = None, observer: str, observed: str, - session_names: list[str] | None = None, + session_allowlist: list[str] | None = None, include_semantic_query: str | None = None, embedding: list[float] | None = None, semantic_search_top_k: int | None = None, @@ -590,7 +610,7 @@ async def get_working_representation( ) return await manager.get_working_representation( db=db, - session_names=session_names, + session_allowlist=session_allowlist, include_semantic_query=include_semantic_query, embedding=embedding, semantic_search_top_k=semantic_search_top_k, diff --git a/src/dialectic/chat.py b/src/dialectic/chat.py index 40d991b4..d2f7e741 100644 --- a/src/dialectic/chat.py +++ b/src/dialectic/chat.py @@ -26,6 +26,7 @@ async def agentic_chat( observer: str, observed: str, reasoning_level: ReasoningLevel = "low", + session_allowlist: list[str] | None = None, response_model: type[BaseModel] | None = None, ) -> str: """ @@ -38,6 +39,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_allowlist: Optional session allowlist restricting all recall response_model: Optional Pydantic model the answer must conform to. When set, the returned string is JSON matching the model's schema. @@ -82,6 +84,7 @@ async def agentic_chat( observer_peer_card=observer_peer_card, observed_peer_card=observed_peer_card, reasoning_level=reasoning_level, + session_allowlist=session_allowlist, ) return await agent.answer(query, response_model=response_model) @@ -94,6 +97,7 @@ async def agentic_chat_stream( observer: str, observed: str, reasoning_level: ReasoningLevel = "low", + session_allowlist: list[str] | None = None, response_model: type[BaseModel] | None = None, ) -> AsyncIterator[str]: """ @@ -106,6 +110,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_allowlist: Optional session allowlist restricting all recall response_model: Optional Pydantic model the answer must conform to. When set, the streamed text accumulates to JSON matching the model's schema. @@ -151,6 +156,7 @@ async def agentic_chat_stream( observer_peer_card=observer_peer_card, observed_peer_card=observed_peer_card, reasoning_level=reasoning_level, + session_allowlist=session_allowlist, ) async for chunk in agent.answer_stream(query, response_model=response_model): diff --git a/src/dialectic/core.py b/src/dialectic/core.py index 757423e0..9580b94c 100644 --- a/src/dialectic/core.py +++ b/src/dialectic/core.py @@ -70,6 +70,7 @@ class DialecticAgent: metric_key: str | None = None, reasoning_level: ReasoningLevel = "low", session_id: str | None = None, + session_allowlist: list[str] | None = None, ): """ Initialize the dialectic agent. @@ -84,9 +85,13 @@ class DialecticAgent: 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_allowlist: 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_allowlist: list[str] | None = session_allowlist self.session_id: str | None = session_id self.observer: str = observer self.observed: str = observed @@ -108,6 +113,24 @@ class DialecticAgent: self._prefetched_conclusion_count: int = 0 self._run_id: str = generate_nanoid() # Always generate for event correlation + def _select_tools(self) -> list[dict[str, Any]]: + """Pick the toolset for this query. + + Minimal reasoning uses a reduced set to reduce cost. Under a session + allowlist `get_reasoning_chain` is dropped entirely rather than left in + to fail at call time: chains traverse provenance across sessions, so it + can't be scoped, and offering it costs both the schema in context and a + wasted turn when the model tries it. + """ + tools = ( + DIALECTIC_TOOLS_MINIMAL + if self.reasoning_level == "minimal" + else DIALECTIC_TOOLS + ) + if self.session_allowlist is not None: + tools = [t for t in tools if t.get("name") != "get_reasoning_chain"] + return tools + async def _initialize_session_history(self) -> None: """Fetch and inject session history into the system prompt if configured.""" if self._session_history_initialized: @@ -197,6 +220,7 @@ class DialecticAgent: limit=prefetch_limit, levels=["explicit"], embedding=query_embedding, + session_allowlist=self.session_allowlist, ) derived_repr = await search_memory( @@ -207,6 +231,7 @@ class DialecticAgent: limit=prefetch_limit, levels=["deductive", "inductive", "contradiction"], embedding=query_embedding, + session_allowlist=self.session_allowlist, ) if explicit_repr.is_empty() and derived_repr.is_empty(): @@ -296,6 +321,7 @@ class DialecticAgent: ] = await create_tool_executor( workspace_name=self.workspace_name, session_name=self.session_name, + session_allowlist=self.session_allowlist, observer=self.observer, observed=self.observed, history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT, @@ -438,12 +464,7 @@ class DialecticAgent: # Get level-specific settings level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level] - # Use minimal tools for minimal reasoning to reduce cost - tools = ( - DIALECTIC_TOOLS_MINIMAL - if self.reasoning_level == "minimal" - else DIALECTIC_TOOLS - ) + tools = self._select_tools() # Use level-specific max_output_tokens if set, otherwise global default max_tokens = ( level_settings.MAX_OUTPUT_TOKENS @@ -520,12 +541,7 @@ class DialecticAgent: # Get level-specific settings level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level] - # Use minimal tools for minimal reasoning to reduce cost - tools = ( - DIALECTIC_TOOLS_MINIMAL - if self.reasoning_level == "minimal" - else DIALECTIC_TOOLS - ) + tools = self._select_tools() # Use level-specific max_output_tokens if set, otherwise global default max_tokens = ( level_settings.MAX_OUTPUT_TOKENS diff --git a/src/routers/peers.py b/src/routers/peers.py index d6e35269..bccb2dd0 100644 --- a/src/routers/peers.py +++ b/src/routers/peers.py @@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession 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 @@ -27,6 +28,7 @@ from src.exceptions import ( 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 extract_session_allowlist from src.utils.schema_conversion import json_response_schema_to_pydantic from src.utils.search import search from src.utils.types import embedding_call_purpose @@ -199,6 +201,25 @@ async def chat( ): raise AuthenticationException("JWT not permissioned for this resource") + # Parse the session allowlist from filters (422 on unsupported keys/shapes, + # and on a session_id the allowlist doesn't cover). + session_allowlist = extract_session_allowlist( + options.filters, must_include=options.session_id + ) + # A peer-scoped key may only name sessions its peer belongs to — the + # allowlist reaches message recall the same way session_id does above. + # `active_only` matches the is_peer_in_session check above, so both gates + # answer the same question for a peer that has left a session. + if jwt_params.p is not None and session_allowlist is not None: + 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, active_only=True + ) + ) + if not set(session_allowlist) <= member_sessions: + raise AuthenticationException("JWT not permissioned for this resource") + # Convert the caller's JSON Schema so malformed schemas fail immediately with 422 response_model: type[BaseModel] | None = None if options.response_format is not None: @@ -244,6 +265,7 @@ async def chat( observer=peer_id, observed=options.target if options.target is not None else peer_id, reasoning_level=options.reasoning_level, + session_allowlist=session_allowlist, response_model=response_model, ) ), @@ -259,6 +281,7 @@ async def chat( # 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_allowlist=session_allowlist, response_model=response_model, ) @@ -294,6 +317,12 @@ async def get_representation( If a target is provided, we get the Representation of the target from the perspective of the Peer. If no target is provided, we get the omniscient Honcho Representation of the Peer. """ + # Parse the session allowlist from filters (422 on unsupported keys/shapes, + # and on a session_id the allowlist doesn't cover). + session_allowlist = extract_session_allowlist( + options.filters, must_include=options.session_id + ) + try: embedding: list[float] | None = None if options.search_query: @@ -312,9 +341,9 @@ async def get_representation( workspace_id, observer=peer_id, observed=options.target if options.target is not None else peer_id, - session_names=[options.session_id] + session_allowlist=[options.session_id] if options.session_id is not None - else None, + else session_allowlist, include_semantic_query=options.search_query, embedding=embedding, semantic_search_top_k=options.search_top_k, @@ -477,7 +506,7 @@ async def get_peer_context( workspace_id, observer=peer_id, observed=observed, - session_names=None, # Peer context is global, not session-scoped + session_allowlist=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 3cc34043..191748d4 100644 --- a/src/routers/sessions.py +++ b/src/routers/sessions.py @@ -43,7 +43,7 @@ async def _get_working_representation_task( *, observer: str, observed: str, - session_names: list[str] | None, + session_allowlist: list[str] | None, search_top_k: int | None, search_max_distance: float | None, include_most_derived: bool, @@ -59,7 +59,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_names: Optional session allowlist to filter by + session_allowlist: 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 @@ -74,7 +74,7 @@ async def _get_working_representation_task( db=db, observer=observer, observed=observed, - session_names=session_names, + session_allowlist=session_allowlist, include_semantic_query=last_message, semantic_search_top_k=search_top_k, semantic_search_max_distance=search_max_distance, @@ -765,7 +765,7 @@ async def get_session_context( search_query, observer=observer, observed=observed, - session_names=[session_id] if limit_to_session else None, + session_allowlist=[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/schemas/api.py b/src/schemas/api.py index 8c8c2cbe..78e5a125 100644 --- a/src/schemas/api.py +++ b/src/schemas/api.py @@ -177,6 +177,15 @@ 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." + ), + ) target: str | None = Field( None, description="Optional peer ID to get the representation for, from the perspective of this peer", @@ -562,6 +571,16 @@ 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." + ), + ) 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 c5a99207..9f214009 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 @@ -30,7 +30,11 @@ from src.utils.formatting import ( parse_datetime_iso, utc_now_iso, ) -from src.utils.representation import Representation +from src.utils.representation import ( + ALLOWLIST_SAFE_LEVELS, + Representation, + allowlist_safe_levels, +) from src.utils.types import ToolResult, embedding_call_purpose, get_current_iteration logger = logging.getLogger(__name__) @@ -1030,6 +1034,7 @@ async def get_recent_history( session_name: str | None, observed: str | None = None, token_limit: int = 8192, + session_allowlist: list[str] | None = None, ) -> list[models.Message]: """ Retrieve recent conversation history. @@ -1042,6 +1047,10 @@ async def get_recent_history( db: Database session workspace_name: Workspace identifier session_name: Session identifier (optional) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. observed: Peer name to filter by when no session specified (optional) token_limit: Maximum tokens to retrieve (default: 8192) @@ -1049,6 +1058,9 @@ async def get_recent_history( List of messages in chronological order """ if session_name: + # Fail closed: a specific session outside the allowlist is not readable. + if session_allowlist is not None and session_name not in session_allowlist: + return [] # Get messages from a specific session messages_stmt = await crud.get_messages( workspace_name=workspace_name, @@ -1061,7 +1073,11 @@ async def get_recent_history( # Return in chronological order return list(reversed(messages)) elif observed: + # Fail closed on an empty allowlist + if session_allowlist is not None and not session_allowlist: + 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) @@ -1069,6 +1085,8 @@ async def get_recent_history( .order_by(models.Message.created_at.desc()) .limit(50) # Limit to recent messages ) + if session_allowlist is not None: + stmt = stmt.where(models.Message.session_name.in_(session_allowlist)) result = await db.execute(stmt) messages = list(result.scalars().all()) # Return in chronological order @@ -1086,6 +1104,7 @@ async def search_memory( limit: int, levels: list[str] | None = None, embedding: list[float] | None = None, + session_allowlist: list[str] | None = None, ) -> Representation: """ Search for observations in memory using semantic similarity. @@ -1106,10 +1125,22 @@ 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_allowlist is not None and not session_allowlist: + return Representation() + + if session_allowlist is not None: + levels = allowlist_safe_levels(levels) + if not levels: + 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_allowlist is not None: + filters["session_name"] = {"in": session_allowlist} documents = await crud.query_documents( db=None, @@ -1118,7 +1149,7 @@ async def search_memory( observed=observed, query=query, top_k=limit, - filters=filters, + filters=filters or None, embedding=embedding, ) @@ -1131,6 +1162,7 @@ async def get_observation_context( session_name: str | None, message_ids: list[str], observer: str | None = None, + session_allowlist: list[str] | None = None, ) -> list[models.Message]: """ Retrieve messages for given message IDs along with surrounding context. @@ -1143,9 +1175,16 @@ async def get_observation_context( db: Database session workspace_name: Workspace identifier session_name: Session identifier (optional) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. message_ids: List of message IDs to retrieve observer: When provided and session_name is None, scope results to sessions this peer belongs to + session_allowlist: Optional session allowlist. None is unrestricted; an + empty list fails closed (empty result); a populated list is + intersected with the observer's session scope when observer is set Returns: List of messages in chronological order, including the requested messages and surrounding context @@ -1153,16 +1192,13 @@ async def get_observation_context( if not message_ids: return [] - # 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 + from src.crud.message import resolve_session_scope - allowed_session_names = await get_peer_session_names( - db, workspace_name, observer - ) - if not allowed_session_names: - return [] + allowed_session_names, deny = await resolve_session_scope( + db, workspace_name, session_name, session_allowlist, observer + ) + if deny: + return [] # Use a CTE to get seq_in_session values for target messages stmt = ( @@ -1221,9 +1257,16 @@ async def extract_preferences( Args: workspace_name: Workspace identifier session_name: Session identifier (optional) + Deprecated for *scoping*: prefer session_allowlist, which + intersects with observer membership. This parameter also pins + the query to one session and bypasses observer scoping, so it + is not a drop-in equivalent and is not removed. observed: The peer whose preferences to extract observer: When provided and session_name is None, scope results to sessions this peer belongs to + session_allowlist: Optional session allowlist. None is unrestricted; an + empty list fails closed (empty result); a populated list is + intersected with the observer's session scope when observer is set Returns: Dict with 'messages' list containing potentially relevant messages @@ -1303,6 +1346,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_allowlist: list[str] | None = None # Telemetry context fields run_id: str | None = None agent_type: str | None = None # "dialectic", "deriver", "dreamer" @@ -1724,6 +1771,7 @@ async def _handle_get_recent_history( session_name=ctx.session_name, observed=ctx.observed, token_limit=ctx.history_token_limit, + session_allowlist=ctx.session_allowlist, ) if not history: return "No conversation history available" @@ -1769,15 +1817,28 @@ 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), + # and only levels with a trustworthy session stamp are served. + documents: Sequence[models.Document] + if ctx.session_allowlist is not None and not ctx.session_allowlist: + 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_allowlist}, + "level": {"in": list(ALLOWLIST_SAFE_LEVELS)}, + } + if ctx.session_allowlist is not None + else None, + ) mem = Representation.from_documents(documents) total_count = mem.len() if total_count == 0: @@ -1799,6 +1860,7 @@ async def _handle_search_memory( context_window=0, embedding=query_embedding, observer=ctx.observer, + session_allowlist=ctx.session_allowlist, ) if snippets: message_output = _format_message_snippets( @@ -1840,6 +1902,7 @@ async def _handle_get_observation_context( session_name=ctx.session_name, message_ids=tool_input["message_ids"], observer=ctx.observer, + session_allowlist=ctx.session_allowlist, ) if not messages: return f"No messages found for IDs {tool_input['message_ids']}" @@ -1882,6 +1945,7 @@ async def _handle_search_messages( context_window=2, embedding=query_embedding, observer=ctx.observer, + session_allowlist=ctx.session_allowlist, ) search_meta: dict[str, Any] = { "top_k": limit, @@ -1918,6 +1982,7 @@ async def _handle_grep_messages( limit=limit, context_window=context_window, observer=ctx.observer, + session_allowlist=ctx.session_allowlist, ) if not snippets: return f"No messages found containing '{text}'" @@ -1982,6 +2047,7 @@ async def _handle_get_messages_by_date_range( limit=limit, order=order, observer=ctx.observer, + session_allowlist=ctx.session_allowlist, ) msg_count = len(messages) messages_text = ( @@ -2054,6 +2120,7 @@ async def _handle_search_messages_temporal( before_date=before_date, limit=limit, context_window=context_window, + session_allowlist=ctx.session_allowlist, embedding=query_embedding, observer=ctx.observer, ) @@ -2310,6 +2377,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_allowlist 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" @@ -2435,6 +2510,7 @@ async def create_tool_executor( run_id: str | None = None, agent_type: str | None = None, parent_category: str | None = None, + session_allowlist: list[str] | None = None, ) -> Callable[[str, dict[str, Any]], Any]: """ Create a unified tool executor function for all agent operations. @@ -2475,6 +2551,7 @@ async def create_tool_executor( history_token_limit=history_token_limit, db_lock=shared_lock, configuration=configuration, + session_allowlist=session_allowlist, 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 36e34f4d..8df3590c 100644 --- a/src/utils/filter.py +++ b/src/utils/filter.py @@ -61,6 +61,83 @@ ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_DOCUMENTS = { } +MAX_SESSION_ALLOWLIST_ENTRIES = 1000 + + +def extract_session_allowlist( + filters: dict[str, Any] | None, + must_include: str | None = 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. + + Args: + filters: The raw ``filters`` body, or None. + must_include: A session id that must appear in the parsed allowlist — + used by routes that also accept a top-level ``session_id``, so the + two can't contradict each other. Ignored when filters is None. + + Returns: + None when filters is None. An explicit empty list is preserved so + downstream consumers fail closed. + + Raises: + FilterError: On an unsupported key or shape, an over-cap list, or a + ``must_include`` session missing from the allowlist. + """ + 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) + + if must_include is not None and must_include not in seen: + raise FilterError("session_id must be included in filters.session_id") + + return allowlist + + def apply_filter( stmt: Select[tuple[T]], model_class: type[T], filters: dict[str, Any] | None = None ) -> Select[tuple[T]]: @@ -221,6 +298,13 @@ def _build_field_condition( if model_class.__name__ == "Message": column_name = ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_MESSAGES.get(key) elif model_class.__name__ == "Document": + # NOTE: unlike Message/Workspace, Document falls back to the raw key so + # internal callers can filter on internal column names. The session + # allowlist depends on this: recall passes {"session_name": {"in": ...}} + # (see search_memory in utils/agent_tools.py and RepresentationManager), + # and "session_name" is deliberately absent from the mapping below. + # Tightening this to a strict allowlist would break session scoping — + # fail-closed, since an unmapped key raises, but silently. column_name = ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_DOCUMENTS.get( key, key, # fallback to the key itself if not found in the mapping for internal use here diff --git a/src/utils/representation.py b/src/utils/representation.py index 674b25bb..01e4b70a 100644 --- a/src/utils/representation.py +++ b/src/utils/representation.py @@ -7,6 +7,35 @@ from pydantic import BaseModel, Field, field_validator from src import models from src.utils.formatting import parse_datetime_iso +# Conclusion levels whose `session_name` stamp is trustworthy enough to scope on. +# +# Explicit conclusions come from the deriver over a single session's message +# batch, so their stamp is authoritative. Deductive/inductive conclusions are +# produced by the dreamer, which reads across *all* sessions (its discovery +# tools default to session_only=False) but stamps its output with one session — +# whichever holds the most recent explicit conclusion, see +# dreamer/dream_scheduler.py. Serving those under a session allowlist would leak +# conclusions synthesized from sessions outside it. +# +# ponytail: whole-level exclusion rather than per-conclusion provenance. The +# reasoning trees already link each conclusion to its premises, so the real fix +# is an authoritative source-session set per conclusion; until that exists this +# fails closed. Tracked in DEV-2201. +ALLOWLIST_SAFE_LEVELS = ("explicit",) + + +def allowlist_safe_levels(levels: list[str] | None) -> list[str]: + """Narrow a level filter to those safe to serve under a session allowlist. + + Returns the intersection with :data:`ALLOWLIST_SAFE_LEVELS`; ``None`` means + "no level filter requested" and yields the full safe set. An empty result + means the caller asked only for levels we can't scope, and should receive + nothing rather than unscoped conclusions. + """ + if levels is None: + return list(ALLOWLIST_SAFE_LEVELS) + return [level for level in levels if level in ALLOWLIST_SAFE_LEVELS] + def _strip_microseconds_and_timezone(timestamp: datetime) -> datetime: """ diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py index b1c24c01..f551de76 100644 --- a/tests/crud/test_representation_manager.py +++ b/tests/crud/test_representation_manager.py @@ -320,7 +320,7 @@ class TestRepresentationManagerSessionScoping: 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] + db_session, top_k=10, session_allowlist=[session_a.name] ) contents = [doc.content for doc in results] @@ -337,7 +337,7 @@ class TestRepresentationManagerSessionScoping: 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] + db_session, top_k=10, session_allowlist=[session_a.name] ) contents = [doc.content for doc in results] @@ -361,12 +361,15 @@ class TestRepresentationManagerSessionScoping: query="anything", top_k=5, embedding=[0.1], - session_names=[session_a.name], + session_allowlist=[session_a.name], ) assert mock_query.await_args is not None assert mock_query.await_args.kwargs["filters"] == { - "session_name": {"in": [session_a.name]} + "session_name": {"in": [session_a.name]}, + # Scoped recall serves only levels with a trustworthy session + # stamp (ALLOWLIST_SAFE_LEVELS / DEV-2201). + "level": {"in": ["explicit"]}, } @pytest.mark.asyncio @@ -403,7 +406,7 @@ class TestRepresentationManagerSessionScoping: representation = await manager.get_working_representation( db=db_session, - session_names=[session_a.name], + session_allowlist=[session_a.name], include_most_derived=True, ) @@ -425,7 +428,7 @@ class TestRepresentationManagerSessionScoping: representation = await manager.get_working_representation( db=db_session, - session_names=[], + session_allowlist=[], include_most_derived=True, ) @@ -441,13 +444,29 @@ class TestRepresentationManagerSessionScoping: "workspace", observer="observer", observed="observed" ) - assert manager._build_filter_conditions(session_names=[]) == { # pyright: ignore[reportPrivateUsage] - "session_name": {"in": []} + # Scoping also narrows to levels whose session stamp is trustworthy + # (see ALLOWLIST_SAFE_LEVELS / DEV-2201). + assert manager._build_filter_conditions(session_allowlist=[]) == { # pyright: ignore[reportPrivateUsage] + "session_name": {"in": []}, + "level": {"in": ["explicit"]}, } - # None means unscoped — no session filter emitted. - assert manager._build_filter_conditions(session_names=None) == {} # pyright: ignore[reportPrivateUsage] - assert manager._build_filter_conditions(session_names=["s1"]) == { # pyright: ignore[reportPrivateUsage] - "session_name": {"in": ["s1"]} + # None means unscoped — no session filter and no level narrowing. + assert manager._build_filter_conditions(session_allowlist=None) == {} # pyright: ignore[reportPrivateUsage] + assert manager._build_filter_conditions(session_allowlist=["s1"]) == { # pyright: ignore[reportPrivateUsage] + "session_name": {"in": ["s1"]}, + "level": {"in": ["explicit"]}, + } + # A requested level outside the safe set yields an empty `in`, which + # matches nothing rather than falling back to unscoped recall. + assert manager._build_filter_conditions( # pyright: ignore[reportPrivateUsage] + level="inductive", session_allowlist=["s1"] + ) == { + "session_name": {"in": ["s1"]}, + "level": {"in": []}, + } + # ...while an unscoped level filter is left exactly as asked. + assert manager._build_filter_conditions(level="inductive") == { # pyright: ignore[reportPrivateUsage] + "level": "inductive" } diff --git a/tests/test_session_allowlist.py b/tests/test_session_allowlist.py new file mode 100644 index 00000000..aa6077cb --- /dev/null +++ b/tests/test_session_allowlist.py @@ -0,0 +1,679 @@ +""" +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.config import settings +from src.crud.message import resolve_session_scope +from src.exceptions import FilterError +from src.models import Peer, Workspace +from src.security import JWTParams, create_jwt +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}) + + def test_must_include_satisfied(self): + assert extract_session_allowlist( + {"session_id": ["s1", "s2"]}, must_include="s2" + ) == ["s1", "s2"] + + def test_must_include_missing_rejected(self): + with pytest.raises(FilterError, match="must be included"): + extract_session_allowlist({"session_id": ["s1"]}, must_include="s2") + + def test_must_include_ignored_without_filters(self): + assert extract_session_allowlist(None, must_include="s1") is None + + def test_must_include_none_is_no_constraint(self): + assert extract_session_allowlist({"session_id": ["s1"]}, must_include=None) == [ + "s1" + ] + + +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_allowlist=["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_allowlist_narrows_levels_to_allowlist_safe(self): + """Only levels with a trustworthy session stamp survive scoping. + + Dream-derived levels are stamped with one session but synthesized + across many (DEV-2201), so they can't be served under an allowlist. + """ + 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", "inductive"], + embedding=[0.1], + session_allowlist=["s1"], + ) + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"]["level"] == {"in": ["explicit"]} + + @pytest.mark.asyncio + async def test_allowlist_defaults_to_explicit_when_no_levels_requested(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, + embedding=[0.1], + session_allowlist=["s1"], + ) + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"]["level"] == {"in": ["explicit"]} + + @pytest.mark.asyncio + async def test_derived_only_request_under_allowlist_returns_empty(self): + """The dialectic's derived prefetch short-circuits instead of querying.""" + 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, + levels=["deductive", "inductive", "contradiction"], + embedding=[0.1], + session_allowlist=["s1"], + ) + mock_query.assert_not_awaited() + assert result.is_empty() + + @pytest.mark.asyncio + async def test_levels_untouched_without_allowlist(self): + """No allowlist means no level narrowing — unscoped recall is unchanged.""" + 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=["deductive", "inductive"], + embedding=[0.1], + ) + assert mock_query.await_args is not None + assert mock_query.await_args.kwargs["filters"] == { + "level": {"in": ["deductive", "inductive"]} + } + + @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_allowlist=[], + ) + 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, + session_allowlist=[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, + session_allowlist=[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, + session_allowlist=[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, + session_allowlist=[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, + session_allowlist=[], + ) + assert messages == [] + + +class TestPeerScopedJWTAllowlistGate: + """A peer-scoped key may only allowlist sessions its peer actively belongs to. + + The gate uses `active_only=True` so it agrees with the `is_peer_in_session` + check on `options.session_id` — a peer that has left a session is denied by + both, not just one. + """ + + @pytest.fixture(autouse=True) + def _enable_auth(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(settings.AUTH, "USE_AUTH", True) + monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret") + + def _chat_as( + self, + client: TestClient, + workspace: Workspace, + peer: Peer, + token: str, + body: dict[str, Any], + ): + return client.post( + f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat", + json={"query": "what do you know?", **body}, + headers={"Authorization": f"Bearer {token}"}, + ) + + async def _session_with( + self, client: TestClient, workspace: Workspace, peer: Peer + ) -> str: + 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 + return session_id + + @pytest.mark.asyncio + async def test_member_sessions_allowed( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + session_id = await self._session_with(client, workspace, peer) + token = create_jwt(JWTParams(w=workspace.name, p=peer.name)) + + with patch( + "src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok") + ) as mock_chat: + resp = self._chat_as( + client, + workspace, + peer, + token, + {"filters": {"session_id": [session_id]}}, + ) + assert resp.status_code == 200 + # The allowlist reaches the agent rather than being dropped at the gate. + assert mock_chat.await_args is not None + assert mock_chat.await_args.kwargs["session_allowlist"] == [session_id] + + @pytest.mark.asyncio + async def test_non_member_session_denied( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + workspace, peer = sample_data + session_id = await self._session_with(client, workspace, peer) + token = create_jwt(JWTParams(w=workspace.name, p=peer.name)) + + # One allowlisted session the peer belongs to, one it doesn't: + # membership must hold for *every* entry. + resp = self._chat_as( + client, + workspace, + peer, + token, + {"filters": {"session_id": [session_id, str(generate_nanoid())]}}, + ) + assert resp.status_code == 401 + + @pytest.mark.asyncio + async def test_left_session_denied( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + """The regression this gate's `active_only` flag exists to prevent. + + With the loose membership definition a peer that left a session still + passed here, while the adjacent `session_id` check rejected it. + """ + workspace, peer = sample_data + session_id = await self._session_with(client, workspace, peer) + token = create_jwt(JWTParams(w=workspace.name, p=peer.name)) + + await crud.remove_peers_from_session( + db_session, + workspace_name=workspace.name, + session_name=session_id, + peer_names={peer.name}, + ) + await db_session.commit() + + resp = self._chat_as( + client, workspace, peer, token, {"filters": {"session_id": [session_id]}} + ) + assert resp.status_code == 401 + + # ...and the single-session gate agrees, which is the whole point. + resp = self._chat_as(client, workspace, peer, token, {"session_id": session_id}) + assert resp.status_code == 401 + + @pytest.mark.asyncio + async def test_workspace_scoped_key_bypasses_gate( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + """Workspace keys are trusted callers — the allowlist passes as given.""" + workspace, peer = sample_data + token = create_jwt(JWTParams(w=workspace.name)) + foreign = str(generate_nanoid()) + + with patch("src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")): + resp = self._chat_as( + client, workspace, peer, token, {"filters": {"session_id": [foreign]}} + ) + assert resp.status_code == 200 + + @pytest.mark.asyncio + async def test_empty_allowlist_still_gated( + self, client: TestClient, sample_data: tuple[Workspace, Peer] + ): + """`filters={"session_id": []}` is a real allowlist, not an absent one. + + It must reach the gate (and pass trivially, since the empty set is a + subset of anything) rather than being skipped by a truthiness check. + """ + workspace, peer = sample_data + token = create_jwt(JWTParams(w=workspace.name, p=peer.name)) + + with patch( + "src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok") + ) as mock_chat: + resp = self._chat_as( + client, workspace, peer, token, {"filters": {"session_id": []}} + ) + assert resp.status_code == 200 + assert mock_chat.await_args is not None + assert mock_chat.await_args.kwargs["session_allowlist"] == [] + + +class TestResolveSessionScope: + """The tri-state contract the four message-crud call sites depend on.""" + + @pytest.mark.asyncio + async def test_unrestricted_when_no_observer_and_no_allowlist( + self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] + ): + workspace, _ = sample_data + assert await resolve_session_scope( + db_session, workspace.name, None, None, None + ) == (None, False) + + @pytest.mark.asyncio + async def test_pinned_session_inside_allowlist_passes_through( + self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] + ): + workspace, _ = sample_data + # None (not [s1]) — the query filters on session_name directly. + assert await resolve_session_scope( + db_session, workspace.name, "s1", ["s1", "s2"], None + ) == (None, False) + + @pytest.mark.asyncio + async def test_pinned_session_outside_allowlist_denies( + self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] + ): + workspace, _ = sample_data + assert await resolve_session_scope( + db_session, workspace.name, "s3", ["s1", "s2"], None + ) == (None, True) + + @pytest.mark.asyncio + async def test_empty_allowlist_denies_rather_than_returning_empty_list( + self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] + ): + """Never returns [] — downstream stores drop an empty IN clause.""" + workspace, _ = sample_data + allowed, deny = await resolve_session_scope( + db_session, workspace.name, None, [], None + ) + assert (allowed, deny) == (None, True) + + @pytest.mark.asyncio + async def test_no_db_touched_when_no_observer_lookup_needed(self): + """Callers pass db=None on the external-vector-store path. + + The helper must not open a session of its own unless it actually needs + an observer lookup, or the external semantic lookup stops being the + first thing that happens (see + tests/integration/test_message_embeddings.py). + """ + with patch("src.crud.message.tracked_db") as mock_tracked_db: + # No observer: pinned session, unrestricted, and plain allowlist. + assert await resolve_session_scope(None, "w", "s1", None, None) == ( + None, + False, + ) + assert await resolve_session_scope(None, "w", None, None, None) == ( + None, + False, + ) + assert await resolve_session_scope(None, "w", None, ["s1"], None) == ( + ["s1"], + False, + ) + mock_tracked_db.assert_not_called() + + @pytest.mark.asyncio + async def test_observer_scope_intersected_with_allowlist( + self, + client: TestClient, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + workspace, peer = sample_data + 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 + + allowed, deny = await resolve_session_scope( + db_session, workspace.name, None, [session_id], peer.name + ) + assert (allowed, deny) == ([session_id], False) + + # Allowlisting only a session the observer isn't in denies outright. + allowed, deny = await resolve_session_scope( + db_session, workspace.name, None, [str(generate_nanoid())], peer.name + ) + assert (allowed, deny) == (None, True) + + +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 diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index 9bfb3990..71b93eb0 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -39,6 +39,8 @@ from src.utils.agent_tools import ( create_observations, create_tool_executor, extract_preferences, + get_observation_context, + get_recent_history, ) # ============================================================================= @@ -801,6 +803,7 @@ class TestSearchMemory: context_window: int = 2, embedding: list[float] | None = None, observer: str | None = None, + **_kwargs: Any, ) -> list[tuple[list[models.Message], list[models.Message]]]: _ = (workspace_name, session_name, query, limit, context_window, observer) fallback_embeddings.append(embedding) @@ -909,6 +912,7 @@ class TestSearchMessagesTemporal: context_window: int = 2, embedding: list[float] | None = None, observer: str | None = None, + **_kwargs: Any, ) -> list[tuple[list[models.Message], list[models.Message]]]: _ = ( workspace_name, @@ -1502,6 +1506,7 @@ class TestExtractPreferences: context_window: int, embedding: list[float] | None, observer: str | None = None, + **_kwargs: Any, ) -> list[tuple[list[models.Message], list[models.Message]]]: _ = (limit, context_window, observer) embedding_args.append(embedding) @@ -1848,3 +1853,63 @@ class TestObserverPeerNameWiring: await _handle_get_messages_by_date_range(ctx, {"after_date": "2024-01-01"}) assert captured_kwargs["observer"] == ctx.observer + + +@pytest.mark.asyncio +class TestSessionAllowlistFailClosed: + """A specific session_name outside the session_allowlist allowlist must fail closed. + + Routes guard this too, but these CRUD/tool functions are reachable directly + from the dialectic loop, so the allowlist is enforced at the boundary. + """ + + async def test_get_recent_history_respects_allowlist( + self, db_session: AsyncSession, tool_test_data: Any + ): + workspace, _peer1, peer2, session, _messages, _ = tool_test_data + + # session IS in the allowlist -> history returned + allowed = await get_recent_history( + db_session, + workspace_name=workspace.name, + session_name=session.name, + observed=peer2.name, + session_allowlist=[session.name], + ) + assert allowed # non-empty + + # session is NOT in the allowlist -> fail closed + blocked = await get_recent_history( + db_session, + workspace_name=workspace.name, + session_name=session.name, + observed=peer2.name, + session_allowlist=["some-other-session"], + ) + assert blocked == [] + + async def test_get_observation_context_fails_closed( + self, db_session: AsyncSession, tool_test_data: Any + ): + workspace, peer1, _peer2, session, messages, _ = tool_test_data + blocked = await get_observation_context( + db_session, + workspace_name=workspace.name, + session_name=session.name, + message_ids=[messages[0].id], + observer=peer1.name, + session_allowlist=["some-other-session"], + ) + assert blocked == [] + + async def test_get_messages_by_date_range_fails_closed( + self, db_session: AsyncSession, tool_test_data: Any + ): + workspace, _peer1, _peer2, session, _messages, _ = tool_test_data + blocked = await crud.get_messages_by_date_range( + db_session, + workspace_name=workspace.name, + session_name=session.name, + session_allowlist=["some-other-session"], + ) + assert blocked == []