From 672f4c66374f10714e62252e1c207b427a3e5fb3 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Fri, 24 Jul 2026 11:41:34 -0400 Subject: [PATCH 1/5] fix: apply session scoping to all working-representation query paths (#881) 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 * 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. --------- Co-authored-by: Claude Fable 5 --- src/crud/representation.py | 59 ++++-- src/routers/peers.py | 6 +- src/routers/sessions.py | 8 +- src/utils/filter.py | 17 +- src/vector_store/lancedb.py | 18 +- src/vector_store/turbopuffer.py | 28 ++- tests/crud/test_representation_manager.py | 218 ++++++++++++++++++++++ tests/test_advanced_filters.py | 80 ++++++++ tests/vector_store/test_lancedb.py | 21 +++ tests/vector_store/test_turbopuffer.py | 24 +++ 10 files changed, 446 insertions(+), 33 deletions(-) diff --git a/src/crud/representation.py b/src/crud/representation.py index 3061c2d9..69c9edc9 100644 --- a/src/crud/representation.py +++ b/src/crud/representation.py @@ -213,7 +213,7 @@ class RepresentationManager: 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, @@ -229,7 +229,10 @@ class RepresentationManager: 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 @@ -267,7 +270,7 @@ class RepresentationManager: 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, @@ -281,7 +284,7 @@ class RepresentationManager: ) 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, @@ -296,7 +299,7 @@ class RepresentationManager: 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, @@ -305,6 +308,12 @@ class RepresentationManager: 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 @@ -345,6 +354,7 @@ class RepresentationManager: top_k=semantic_observations, max_distance=semantic_search_max_distance, embedding=embedding, + session_names=session_names, ) representation.merge_representation( Representation.from_documents(semantic_docs) @@ -353,7 +363,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 + db, top_k=top_observations, session_names=session_names ) representation.merge_representation( Representation.from_documents(derived_docs) @@ -361,7 +371,7 @@ class RepresentationManager: # 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)) @@ -376,6 +386,7 @@ class RepresentationManager: 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: @@ -387,6 +398,7 @@ class RepresentationManager: top_k, max_distance, embedding=embedding, + session_names=session_names, ) else: documents = await crud.query_documents( @@ -398,6 +410,8 @@ class RepresentationManager: 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) @@ -407,7 +421,7 @@ class RepresentationManager: 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 = ( @@ -419,8 +433,8 @@ class RepresentationManager: 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 [] ), ) @@ -433,7 +447,7 @@ class RepresentationManager: 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 = ( @@ -444,6 +458,11 @@ class RepresentationManager: 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(), @@ -480,6 +499,7 @@ class RepresentationManager: 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( @@ -490,7 +510,7 @@ class RepresentationManager: 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, ) @@ -503,17 +523,28 @@ class RepresentationManager: 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 + # `is not None` (not truthiness): an explicit empty allowlist must emit + # 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} + return filters @@ -526,7 +557,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, @@ -559,7 +590,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/routers/peers.py b/src/routers/peers.py index b61bf442..d6e35269 100644 --- a/src/routers/peers.py +++ b/src/routers/peers.py @@ -312,7 +312,9 @@ async def get_representation( workspace_id, observer=peer_id, 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 None, include_semantic_query=options.search_query, embedding=embedding, semantic_search_top_k=options.search_top_k, @@ -475,7 +477,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 89ce665c..3cc34043 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_name: str | None, + session_names: 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_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 @@ -74,7 +74,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, @@ -765,7 +765,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/utils/filter.py b/src/utils/filter.py index 1394a8c6..36e34f4d 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 @@ -28,6 +29,10 @@ COMPARISON_OPERATORS = { NUMERIC_OPERATORS = {"gte", "lte", "gt", "lt", "ne"} +# JSONB columns keep containment semantics: bare lists are not membership +# sugar, and dict values map to nested-metadata conditions rather than IN/Eq. +JSONB_COLUMNS = ("h_metadata", "configuration", "internal_metadata") + ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING = { "id": "name", "created_at": "created_at", @@ -238,6 +243,12 @@ 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 JSONB_COLUMNS: + 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 @@ -248,12 +259,12 @@ def _build_field_condition( else: # This is a regular value that happens to be a dict # For JSONB fields (metadata, configuration), check if it contains nested comparison operators - if column_name in ("h_metadata", "configuration", "internal_metadata"): + if column_name in JSONB_COLUMNS: return _build_nested_metadata_conditions(column, value) # pyright: ignore else: return column == value else: - if column_name in ("h_metadata", "configuration", "internal_metadata"): + if column_name in JSONB_COLUMNS: return column.contains(value) else: return column == value diff --git a/src/vector_store/lancedb.py b/src/vector_store/lancedb.py index 0b621971..1b4c4880 100644 --- a/src/vector_store/lancedb.py +++ b/src/vector_store/lancedb.py @@ -298,10 +298,15 @@ class LanceDBVectorStore(VectorStore): 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 @@ class LanceDBVectorStore(VectorStore): 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 ded0c691..e7825310 100644 --- a/src/vector_store/turbopuffer.py +++ b/src/vector_store/turbopuffer.py @@ -20,9 +20,7 @@ from . import VectorQueryResult, VectorRecord, VectorStore logger = logging.getLogger(__name__) -# Type aliases for Turbopuffer's filter formats -EqFilter = tuple[str, Literal["Eq"], Any] -InFilter = tuple[str, Literal["In"], Sequence[Any]] +# Type alias for Turbopuffer's AND filter format AndFilter = tuple[Literal["And"], Sequence[Filter]] DISTANCE_METRIC = "cosine_distance" @@ -245,13 +243,17 @@ class TurbopufferVectorStore(VectorStore): if not filters: return None - filter_list: list[EqFilter | InFilter] = [] + filter_list: list[Filter] = [] for key, value in filters.items(): # Check if value is a dict with "in" operator if isinstance(value, dict) and "in" in value: # Membership filter using "In" operator - in_values = cast(Sequence[Any], value["in"]) - filter_list.append((key, "In", in_values)) + in_values = list(cast(Sequence[Any], value["in"])) + filter_list.append(self._membership_filter(key, in_values)) + elif isinstance(value, list | tuple | set): + # Bare-list sugar: same membership semantics as {"in": [...]} + in_values = list(cast(Sequence[Any], value)) + filter_list.append(self._membership_filter(key, in_values)) else: # Simple equality filter using "Eq" operator filter_list.append((key, "Eq", cast(Any, value))) @@ -266,6 +268,20 @@ class TurbopufferVectorStore(VectorStore): and_filter: AndFilter = ("And", filter_list) return and_filter + @staticmethod + def _membership_filter(key: str, values: list[Any]) -> Filter: + """Build an "In" membership filter, failing closed on an empty list. + + Turbopuffer's empty-"In" semantics are undocumented, so an empty + allowlist emits an explicit contradiction (`Eq(x) AND NotEq(x)` is + false for every document) rather than risk a fail-open widening. + Mirrors lancedb's `1 = 0` guard. + """ + if not values: + never: AndFilter = ("And", [(key, "Eq", ""), (key, "NotEq", "")]) + return never + return (key, "In", values) + async def delete_many(self, namespace: str, ids: list[str]) -> None: """ Delete multiple vectors from Turbopuffer. diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py index f4be22dd..b1c24c01 100644 --- a/tests/crud/test_representation_manager.py +++ b/tests/crud/test_representation_manager.py @@ -233,6 +233,224 @@ class TestRepresentationManagerSoftDelete: 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 == [] + + def test_build_filter_conditions_empty_allowlist_fails_closed(self): + """The filter-builder layer itself must fail closed, independent of the + early-return guard in _get_working_representation_internal. An empty + allowlist emits an empty `in` (renders as always-false downstream), not + an omitted filter.""" + manager = RepresentationManager( + "workspace", observer="observer", observed="observed" + ) + + assert manager._build_filter_conditions(session_names=[]) == { # pyright: ignore[reportPrivateUsage] + "session_name": {"in": []} + } + # 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"]} + } + + class TestRepresentationManagerSave: @pytest.mark.asyncio async def test_save_representation_filters_blank_observations_before_embedding( diff --git a/tests/test_advanced_filters.py b/tests/test_advanced_filters.py index 768f29bc..3b187470 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/vector_store/test_lancedb.py b/tests/vector_store/test_lancedb.py index 0c502e4c..9b52708a 100644 --- a/tests/vector_store/test_lancedb.py +++ b/tests/vector_store/test_lancedb.py @@ -37,6 +37,27 @@ def store() -> LanceDBVectorStore: return LanceDBVectorStore() +def test_build_where_clause_membership(store: LanceDBVectorStore) -> None: + """Both the dict `in` form and the bare-list sugar produce an IN clause.""" + assert ( + store._build_where_clause({"session_name": {"in": ["s1", "s2"]}}) # pyright: ignore[reportPrivateUsage] + == "session_name IN ('s1', 's2')" + ) + assert ( + store._build_where_clause({"session_name": ["s1", "s2"]}) # pyright: ignore[reportPrivateUsage] + == "session_name IN ('s1', 's2')" + ) + + +def test_build_where_clause_empty_membership_fails_closed( + store: LanceDBVectorStore, +) -> None: + """An empty membership list must emit an always-false predicate, never an + omitted condition that would widen scope (fail-open).""" + assert store._build_where_clause({"session_name": {"in": []}}) == "1 = 0" # pyright: ignore[reportPrivateUsage] + assert store._build_where_clause({"session_name": []}) == "1 = 0" # pyright: ignore[reportPrivateUsage] + + @pytest.mark.asyncio async def test_query_returns_empty_when_table_missing( store: LanceDBVectorStore, diff --git a/tests/vector_store/test_turbopuffer.py b/tests/vector_store/test_turbopuffer.py index ca8c4ff3..cdae44d5 100644 --- a/tests/vector_store/test_turbopuffer.py +++ b/tests/vector_store/test_turbopuffer.py @@ -53,6 +53,30 @@ async def test_upsert_many_raises_vector_store_error_on_5xx( namespace_mock.write.assert_awaited_once() +def test_build_filters_membership(store: TurbopufferVectorStore) -> None: + """Both the dict `in` form and the bare-list sugar produce an In filter.""" + assert store._build_filters({"session_name": {"in": ["s1", "s2"]}}) == ( # pyright: ignore[reportPrivateUsage] + "session_name", + "In", + ["s1", "s2"], + ) + assert store._build_filters({"session_name": ["s1", "s2"]}) == ( # pyright: ignore[reportPrivateUsage] + "session_name", + "In", + ["s1", "s2"], + ) + + +def test_build_filters_empty_membership_fails_closed( + store: TurbopufferVectorStore, +) -> None: + """An empty membership list must produce an always-false filter, never an + omitted/empty In that could widen scope (fail-open).""" + never = ("And", [("session_name", "Eq", ""), ("session_name", "NotEq", "")]) + assert store._build_filters({"session_name": {"in": []}}) == never # pyright: ignore[reportPrivateUsage] + assert store._build_filters({"session_name": []}) == never # pyright: ignore[reportPrivateUsage] + + @pytest.mark.asyncio async def test_upsert_many_short_circuits_on_empty( store: TurbopufferVectorStore, From 24f7a2cbd24ad4638d7d6b5dfec51e792869455e Mon Sep 17 00:00:00 2001 From: papesy384 Date: Fri, 24 Jul 2026 12:36:55 -0400 Subject: [PATCH 2/5] fix(dev): skip lancedb on macOS Intel and guard optional import (#496) Add a PEP 508 marker so lancedb is not installed on darwin/x86_64, wrap the LanceDB vector store import in try/except for a clear config error, and regenerate uv.lock. Branch rebased onto upstream/main; prior src/utils/clients.py CI tweak is obsolete because LLM wiring moved under src/llm/. Co-authored-by: Cursor --- pyproject.toml | 2 +- src/vector_store/__init__.py | 11 ++++++++++- uv.lock | 4 ++-- 3 files changed, 13 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index d7905fc6..586f0c45 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "typing-extensions>=4.11.0", "json-repair>=0.49.0", "turbopuffer>=1.8.1", - "lancedb>=0.25.3", + "lancedb>=0.25.3; sys_platform != \"darwin\" or platform_machine != \"x86_64\"", "pyarrow>=19.0.0", "redis>=7.0.0,<8.0.0", "cashews[redis]==7.5.0", diff --git a/src/vector_store/__init__.py b/src/vector_store/__init__.py index f0fccb63..85fdd9ff 100644 --- a/src/vector_store/__init__.py +++ b/src/vector_store/__init__.py @@ -202,7 +202,16 @@ def _create_store_by_type(store_type: str) -> VectorStore: return TurbopufferVectorStore() elif store_type == "lancedb": - from src.vector_store.lancedb import LanceDBVectorStore + try: + from src.vector_store.lancedb import LanceDBVectorStore + except ImportError as exc: + raise RuntimeError( + "VECTOR_STORE.TYPE is set to 'lancedb', but the 'lancedb' package " + "is not installed (for example on macOS Intel, where it is omitted " + "from dependencies because PyPI has no wheel). " + "Use TYPE 'pgvector' or 'turbopuffer', or install lancedb manually. " + f"Original import error: {exc}" + ) from exc return LanceDBVectorStore() else: diff --git a/uv.lock b/uv.lock index c27d054c..ffaa6255 100644 --- a/uv.lock +++ b/uv.lock @@ -1171,7 +1171,7 @@ dependencies = [ { name = "greenlet" }, { name = "httpx" }, { name = "json-repair" }, - { name = "lancedb" }, + { name = "lancedb", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" }, { name = "langfuse" }, { name = "nanoid" }, { name = "openai" }, @@ -1225,7 +1225,7 @@ requires-dist = [ { name = "greenlet", specifier = ">=3.0.3" }, { name = "httpx", specifier = ">=0.27.0" }, { name = "json-repair", specifier = ">=0.49.0" }, - { name = "lancedb", specifier = ">=0.25.3" }, + { name = "lancedb", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'", specifier = ">=0.25.3" }, { name = "langfuse", specifier = ">=3.3.2" }, { name = "nanoid", specifier = ">=2.0.0" }, { name = "openai", specifier = ">=1.99.7" }, From d7b64116accfdd7eb621f87fd84813c1b4e910e6 Mon Sep 17 00:00:00 2001 From: Leonardo Baray Date: Fri, 24 Jul 2026 13:54:31 -0300 Subject: [PATCH 3/5] fix: redact Redis password from cache connection logs (#869) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: redact Redis password from cache connection logs The cache client logged the full Redis URL — including the password — at INFO and WARNING levels on every connection attempt and failure. This exposed the live Redis credential in stdout/container logs and any downstream log aggregation. Add _redact_cache_url() to mask the password component before logging. URLs without a password are returned unchanged. Closes #866 * fix: handle malformed URLs and IPv6 in _redact_cache_url Address review feedback from VVoruganti and CodeRabbit: - Wrap urlparse/urlunparse in try/except so malformed URLs (e.g. invalid port) don't raise ValueError inside except blocks, which would crash startup instead of degrading gracefully - Preserve IPv6 brackets (e.g. [::1]) in reconstructed URLs - Add Google-style Args/Returns docstring sections - Add unit tests for password masking, no-password URLs, IPv6, malformed inputs, and the invalid-port regression * test: use real secrets in redaction test fixtures Three fixtures were weakened by copy-paste mangling: literal '***' placeholders instead of real passwords (assertions trivially true), an unescaped '#' that truncated netloc parsing via the URL fragment, and a no-password case that actually contained userinfo. Restore inputs that genuinely exercise the masking paths. * fix: never leak password through malformed-URL fallback The catch-all fallback returned the original URL when parsing failed, so a Redis URL with a password and an invalid port (typo, out-of-range) was logged in clear text - the exact leak #866 exists to fix. Narrow the handling: .port access gets its own try/except (invalid port is omitted from the output; userinfo/hostname masking never raises), and the outer fallback now returns a generic placeholder instead of the raw input. Tightened the invalid-port test to assert the password is absent and added out-of-range-port and unparseable-URL cases. * fix: redact secrets in query params and scheme-less URLs _redact_cache_url only masked userinfo, but a credential can reach the URL through two other real configuration paths: redis-py accepts ?password= (all querystring options become client kwargs) and cashews accepts ?secret= (its HMAC signing key) - and honcho's own default CACHE.URL already uses a query param (?suppress=true), so this is the expected configuration style. Separately, a URL missing its scheme (':pass@host:6379/0') parses with an empty netloc, making the password invisible to .password and echoing it back verbatim. Mask sensitive query values in place on the raw query string (no decode/re-encode, so non-secret params are preserved byte-for-byte) and return the generic placeholder for @-carrying strings with no parseable authority. Verified with a 20k-case randomized fuzz run in addition to the unit tests: no functional credential reaches the output. --- src/cache/client.py | 103 +++++++++++++++++++++++++-- tests/test_cache_redaction.py | 126 ++++++++++++++++++++++++++++++++++ 2 files changed, 224 insertions(+), 5 deletions(-) create mode 100644 tests/test_cache_redaction.py diff --git a/src/cache/client.py b/src/cache/client.py index 29d8c187..1ad8e3f5 100644 --- a/src/cache/client.py +++ b/src/cache/client.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import logging from typing import Any, cast +from urllib.parse import urlparse, urlunparse import sentry_sdk from cashews import cache @@ -20,10 +21,99 @@ from src.config import settings logger = logging.getLogger(__name__) - _cache_lock = asyncio.Lock() +# Query parameters that carry secrets when configured via URL: +# redis-py accepts ``?password=`` (all querystring options become client +# kwargs) and cashews accepts ``?secret=`` (HMAC key for value signing). +_SENSITIVE_QUERY_PARAMS = frozenset({"password", "secret"}) + + +def _mask_sensitive_query(query: str) -> str: + """Mask values of secret-bearing query parameters. + + Operates on the raw query string (no decode/re-encode round trip) + so non-secret parameters are preserved byte-for-byte. + + Args: + query: The raw query string from a parsed URL. + + Returns: + The query string with sensitive values replaced by ``***``, or + the original string if no sensitive parameter is present. + """ + if not query: + return query + parts: list[str] = [] + changed = False + for part in query.split("&"): + name, sep, _value = part.partition("=") + if sep and name.lower() in _SENSITIVE_QUERY_PARAMS: + parts.append(f"{name}=***") + changed = True + else: + parts.append(part) + return "&".join(parts) if changed else query + + +def _redact_cache_url(url: str) -> str: + """Mask credentials in a Redis connection URL before logging. + + Given ``redis://:password@host:port/db`` returns + ``redis://:***@host:port/db``; secret-bearing query parameters + (``?password=``, ``?secret=``) are masked as well. A URL carrying + no credentials is returned unchanged. This function never raises + and never returns a credential: an invalid port is omitted from + the output, and a URL that cannot be parsed at all is replaced by + a generic placeholder rather than echoed back, so that logging + inside ``except`` blocks can neither crash startup nor leak the + secrets this helper exists to hide. + + Args: + url: The Redis connection URL to redact. + + Returns: + The URL with its credentials masked, the original URL if it + carries none, or ``""`` if parsing + fails entirely. + """ + try: + parsed = urlparse(url) + query = _mask_sensitive_query(parsed.query) + # .password only splits netloc and never raises, unlike .port + if parsed.password is None and query == parsed.query: + # A string with an "@" but no parsed authority (e.g. a URL + # missing its scheme, ":pass@host:6379/0") may still carry + # userinfo that urlparse could not see — never echo it. + if "@" in url and not parsed.netloc: + return "" + return url + netloc = parsed.netloc + if parsed.password is not None: + userinfo = parsed.username or "" + hostname = parsed.hostname or "" + # Preserve IPv6 brackets (urlparse strips them from .hostname) + if hostname and ":" in hostname and not hostname.startswith("["): + hostname = f"[{hostname}]" + netloc = f"{userinfo}:***@{hostname}" + try: + port = parsed.port + except ValueError: + # Invalid or out-of-range port: omit it rather than let + # the outer fallback echo the raw URL (and its password) + # back. + port = None + if port is not None: + netloc += f":{port}" + parsed = parsed._replace(netloc=netloc, query=query) + return urlunparse(parsed) + except (ValueError, TypeError): + # Unparseable URL: never return the raw input — it may contain + # the very password this helper exists to hide. + return "" + + def is_cache_enabled() -> bool: return settings.CACHE.ENABLED @@ -59,7 +149,7 @@ async def init_cache() -> None: except Exception as setup_err: logger.warning( "Cache setup failed for %s: %s. Falling back to in-memory cache", - settings.CACHE.URL, + _redact_cache_url(settings.CACHE.URL), setup_err, ) if settings.SENTRY.ENABLED: @@ -87,7 +177,10 @@ async def init_cache() -> None: with attempt: async with asyncio.timeout(2): await cache.ping() - logger.info("Connected to cache at %s", settings.CACHE.URL) + logger.info( + "Connected to cache at %s", + _redact_cache_url(settings.CACHE.URL), + ) except ( redis_exc.TimeoutError, redis_exc.ConnectionError, @@ -96,7 +189,7 @@ async def init_cache() -> None: ) as e: logger.warning( "Failed to connect to cache at %s: %s. Falling back to in-memory cache", - settings.CACHE.URL, + _redact_cache_url(settings.CACHE.URL), e, ) if settings.SENTRY.ENABLED: @@ -107,7 +200,7 @@ async def init_cache() -> None: except Exception as e: logger.warning( "Unexpected cache error at %s: %s. Falling back to in-memory cache", - settings.CACHE.URL, + _redact_cache_url(settings.CACHE.URL), e, ) if settings.SENTRY.ENABLED: diff --git a/tests/test_cache_redaction.py b/tests/test_cache_redaction.py new file mode 100644 index 00000000..0ff2bbc7 --- /dev/null +++ b/tests/test_cache_redaction.py @@ -0,0 +1,126 @@ +"""Unit tests for the cache client's _redact_cache_url helper.""" + +import pytest + +from src.cache.client import _redact_cache_url + + +class TestRedactCacheUrl: + """Tests for _redact_cache_url — a security-relevant logging helper + that must never raise and must never leak a password.""" + + # --- Password masking --- + + def test_password_only_userinfo(self): + assert ( + _redact_cache_url("redis://:secret@localhost:6379/0") + == "redis://:***@localhost:6379/0" + ) + + def test_user_and_password(self): + result = _redact_cache_url("redis://user:s3cret@10.0.0.1:6380/2") + assert "***" in result + assert "s3cret" not in result + assert "user" in result + + def test_rediss_protocol(self): + result = _redact_cache_url("rediss://:secret@redis.example.com:6380") + assert result.startswith("rediss://") + assert "***" in result + assert "secret" not in result + + def test_complex_password(self): + result = _redact_cache_url("redis://:p%40ssw0rd!%24@host:6379/0") + assert "***" in result + assert "p%40ssw0rd" not in result + + def test_password_never_leaked(self): + """The original password must never appear in the redacted output.""" + for url in [ + "redis://:hunter2@localhost:6379/0", + "redis://admin:hunter2@localhost:6379/0", + "rediss://:hunter2@[::1]:6380/1", + ]: + assert "hunter2" not in _redact_cache_url(url) + + # --- Secrets in query parameters --- + # redis-py accepts ?password= (querystring options become client + # kwargs) and cashews accepts ?secret= (HMAC signing key), so both + # are real configuration paths that must not reach the logs. + + @pytest.mark.parametrize("param", ["password", "secret", "PASSWORD"]) + def test_query_param_secret_masked(self, param: str): + result = _redact_cache_url(f"redis://host:6379/0?{param}=s3cret") + assert "s3cret" not in result + assert f"{param}=***" in result + + def test_query_param_masking_preserves_other_params(self): + result = _redact_cache_url("redis://host:6379/0?db=1&password=s3cret&ssl=true") + assert "s3cret" not in result + assert "db=1" in result + assert "ssl=true" in result + + def test_userinfo_and_query_secret_both_masked(self): + result = _redact_cache_url("redis://:hunter2@host:6379/0?secret=s3cret") + assert "hunter2" not in result + assert "s3cret" not in result + + def test_non_secret_query_params_unchanged(self): + url = "redis://localhost:6379/0?suppress=true" + assert _redact_cache_url(url) == url + + # --- No-password URLs (returned unchanged) --- + + def test_user_without_password_unchanged(self): + assert ( + _redact_cache_url("redis://user@localhost:6379/0") + == "redis://user@localhost:6379/0" + ) + + def test_in_memory_url_unchanged(self): + assert _redact_cache_url("mem://") == "mem://" + + # --- IPv6 --- + + def test_ipv6_brackets_preserved(self): + result = _redact_cache_url("rediss://:secret@[::1]:6380/1") + assert "[::1]" in result + assert "***" in result + assert "secret" not in result + + # --- Malformed URLs (must NOT raise) --- + + def test_invalid_port_redacts_password(self): + """Regression test for two review findings: accessing + ``parsed.port`` on a URL with a non-numeric port raises + ``ValueError`` (must not crash startup inside an except block), + and the fallback must never echo the raw URL back — the + password has to be masked even when the port is unparseable. + """ + result = _redact_cache_url("redis://:pass@host:notaport/0") + assert "pass" not in result + assert "***" in result + + def test_out_of_range_port_redacts_password(self): + result = _redact_cache_url("redis://:supersecret@host:99999/0") + assert "supersecret" not in result + assert "***" in result + + def test_unparseable_url_never_echoed(self): + # Unbalanced IPv6 bracket makes urlparse itself raise; the + # fallback must return a placeholder, not the raw input. + result = _redact_cache_url("redis://:secret@[::1:6379/0") + assert "secret" not in result + + def test_missing_scheme_never_echoed(self): + # Without "redis://" urlparse sees no netloc, so the userinfo + # (and its password) is invisible to .password — the string + # must not be echoed back. + result = _redact_cache_url(":hunter2@host:6379/0") + assert "hunter2" not in result + + def test_garbage_input_does_not_raise(self): + assert isinstance(_redact_cache_url("not a url at all"), str) + + def test_empty_string_does_not_raise(self): + assert isinstance(_redact_cache_url(""), str) From 3ee890fa6f55388abd23b7660fb726e14d83459d Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Fri, 24 Jul 2026 15:42:41 -0400 Subject: [PATCH 4/5] Vineeth/sentry filter consolidation (#934) * fix: centralize sentry before_send filter config * chore: comply with linter * fix: default sentry filters --- src/main.py | 29 +------- src/telemetry/sentry.py | 46 +++++++++++- tests/telemetry/test_sentry_before_send.py | 82 ++++++++++++++++++++++ 3 files changed, 126 insertions(+), 31 deletions(-) create mode 100644 tests/telemetry/test_sentry_before_send.py diff --git a/src/main.py b/src/main.py index b0611f59..3c8bf3e8 100644 --- a/src/main.py +++ b/src/main.py @@ -4,15 +4,12 @@ import time import uuid from collections.abc import Awaitable, Callable from contextlib import asynccontextmanager -from typing import TYPE_CHECKING import sentry_sdk from fastapi import FastAPI, Request, Response -from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from fastapi_pagination import add_pagination -from pydantic import ValidationError from sentry_sdk.integrations.fastapi import FastApiIntegration from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration from sentry_sdk.integrations.starlette import StarletteIntegration @@ -42,9 +39,6 @@ from src.telemetry import ( from src.telemetry.logging import get_route_template from src.telemetry.sentry import initialize_sentry -if TYPE_CHECKING: - from sentry_sdk._types import Event, Hint - def get_log_level() -> int: """ @@ -88,30 +82,10 @@ class MetricsAccessFilter(logging.Filter): logging.getLogger("uvicorn.access").addFilter(MetricsAccessFilter()) -def before_send(event: "Event", hint: "Hint | None") -> "Event | None": - """Filter out events raised from known non-actionable exceptions before Sentry sees them.""" - if not hint: - return event - - exc_info = hint.get("exc_info") - if not exc_info: - return event - - _, exc_value, _ = exc_info - if isinstance(exc_value, HonchoException): - return None - - # Filters out ValidationErrors and RequestValidationErrors (typically coming from Pydantic) - if isinstance(exc_value, ValidationError | RequestValidationError): - logger.info(f"Filtering out validation error from Sentry: {exc_value}") - return None - - return event - - # Sentry Setup SENTRY_ENABLED = settings.SENTRY.ENABLED if SENTRY_ENABLED: + # before_send defaults to sentry.default_before_send (shared with the deriver). initialize_sentry( integrations=[ StarletteIntegration( @@ -123,7 +97,6 @@ if SENTRY_ENABLED: # Explicit so DB-query spans are not reliant on auto-enabling. SqlalchemyIntegration(), ], - before_send=before_send, ) diff --git a/src/telemetry/sentry.py b/src/telemetry/sentry.py index d84837ed..7b17e156 100644 --- a/src/telemetry/sentry.py +++ b/src/telemetry/sentry.py @@ -9,19 +9,58 @@ from functools import wraps from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast import sentry_sdk +from fastapi.exceptions import RequestValidationError +from pydantic import ValidationError +from sqlalchemy.exc import OperationalError from src.config import settings +from src.exceptions import HonchoException P = ParamSpec("P") T = TypeVar("T") if TYPE_CHECKING: - from sentry_sdk._types import EventProcessor + from sentry_sdk._types import Event, EventProcessor, Hint from sentry_sdk.integrations import Integration logger = logging.getLogger(__name__) +def default_before_send(event: Event, hint: Hint | None) -> Event | None: + """Filter/regroup known non-actionable events before Sentry ingests them. + + Shared by every entrypoint (API + deriver) so filtering is process-agnostic. + """ + if not hint: + return event + + exc_info = hint.get("exc_info") + if not exc_info: + return event + + _, exc_value, _ = exc_info + if isinstance(exc_value, HonchoException): + return None + + # Filters out ValidationErrors and RequestValidationErrors (typically from Pydantic) + if isinstance(exc_value, ValidationError | RequestValidationError): + logger.info(f"Filtering out validation error from Sentry: {exc_value}") + return None + + # DB connection-pool checkout timeouts are a fleet-wide saturation symptom, not a + # per-transaction bug. Collapse every occurrence into one issue (Sentry would otherwise + # split by transaction/endpoint) and drop to warning so it stops tripping error alerts. + # Watch it via a rate/spike metric alert instead. Root cause tracked in DEV-1852. + if isinstance(exc_value, OperationalError) and "connection timeout expired" in str( + exc_value + ): + event["fingerprint"] = ["honcho-db-connection-timeout"] + event["level"] = "warning" + return event + + return event + + # Paths whose transactions carry no debugging value but are hit constantly # (health checks, Prometheus scrapes, OpenAPI schema, docs). Tracing them at the # same rate as real traffic drowns the signal and burns tracing/profiling quota. @@ -81,13 +120,14 @@ def traces_sampler(sampling_context: dict[str, Any]) -> float: def initialize_sentry( *, integrations: Sequence[Integration], - before_send: EventProcessor | None = None, + before_send: EventProcessor | None = default_before_send, ) -> None: """Initialize Sentry SDK with project settings. Args: integrations: Sentry SDK integrations to enable (e.g., Starlette, FastAPI). - before_send: Optional event filter callback to suppress specific exceptions. + before_send: Event filter override. Defaults to ``default_before_send`` so + every entrypoint gets the shared filters; pass ``None`` to opt out. """ sentry_sdk.init( dsn=settings.SENTRY.DSN, diff --git a/tests/telemetry/test_sentry_before_send.py b/tests/telemetry/test_sentry_before_send.py new file mode 100644 index 00000000..7e63b555 --- /dev/null +++ b/tests/telemetry/test_sentry_before_send.py @@ -0,0 +1,82 @@ +"""Tests for the shared Sentry before_send filter. + +default_before_send runs in every entrypoint (API + deriver). It drops known +non-actionable exceptions and collapses DB connection-pool checkout timeouts +into a single warning-level issue so they stop spawning a fresh error issue per +transaction (fleet-wide saturation symptom, tracked in DEV-1852). +""" + +from typing import TYPE_CHECKING, cast + +import pytest +import sentry_sdk +from fastapi.exceptions import RequestValidationError +from pydantic import ValidationError +from sqlalchemy.exc import OperationalError + +from src.exceptions import ResourceNotFoundException +from src.telemetry.sentry import default_before_send, initialize_sentry + +if TYPE_CHECKING: + from sentry_sdk._types import Event, Hint + + +def _hint(exc: BaseException) -> "Hint": + return cast("Hint", {"exc_info": (type(exc), exc, None)}) + + +def _event(**kwargs: object) -> "Event": + return cast("Event", cast(object, dict(kwargs))) + + +def test_connection_timeout_is_consolidated_and_downgraded() -> None: + exc = OperationalError("SELECT 1", {}, Exception("connection timeout expired")) + out = default_before_send({}, _hint(exc)) + assert out == { + "fingerprint": ["honcho-db-connection-timeout"], + "level": "warning", + } + + +def test_unrelated_operational_error_passes_through() -> None: + exc = OperationalError("SELECT 1", {}, Exception("some other db failure")) + event = _event(level="error") + assert default_before_send(event, _hint(exc)) == {"level": "error"} + + +def test_honcho_and_validation_errors_are_dropped() -> None: + assert default_before_send({}, _hint(ResourceNotFoundException("nope"))) is None + assert ( + default_before_send({}, _hint(ValidationError.from_exception_data("x", []))) + is None + ) + assert default_before_send({}, _hint(RequestValidationError([]))) is None + + +def test_events_without_exc_info_pass_through() -> None: + event = _event(release="1.0") + assert default_before_send(event, None) == {"release": "1.0"} + assert default_before_send(event, cast("Hint", {})) == {"release": "1.0"} + + +def _captured_before_send(monkeypatch: pytest.MonkeyPatch, **kwargs: object) -> object: + captured: dict[str, object] = {} + + def fake_init(**init_kwargs: object) -> None: + captured.update(init_kwargs) + + monkeypatch.setattr(sentry_sdk, "init", fake_init) + initialize_sentry(integrations=[], **kwargs) # pyright: ignore[reportArgumentType] + return captured["before_send"] + + +def test_initialize_sentry_defaults_to_shared_filter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + assert _captured_before_send(monkeypatch) is default_before_send + + +def test_initialize_sentry_explicit_none_bypasses_shared_filter( + monkeypatch: pytest.MonkeyPatch, +) -> None: + assert _captured_before_send(monkeypatch, before_send=None) is None 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 5/5] 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 == []