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