RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: session allowlist on dialectic and representation via filters (#882) 

* 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 <noreply@anthropic.com>

* feat: bare-list membership sugar in the filter DSL

{"session_id": ["s1", "s2"]} is now shorthand for
{"session_id": {"in": [...]}} on regular columns, generically
(peer_id, etc.). JSONB metadata columns are excluded — a bare list
there keeps JSONB containment semantics, unchanged.

Previously a bare list on a regular column compiled to a type-mismatched
equality that matched nothing, so this is strictly additive.

Also translates the same shape in the turbopuffer/lancedb filter
builders, and fixes lancedb dropping empty IN clauses (fail-open) —
an empty membership list now emits an always-false condition.

Groundwork for DEV-1995 (session allowlist via the existing filters
DSL, no new API params)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* feat: session allowlist on dialectic and representation via filters

Adds a constrained 'filters' body to peer.chat and /representation —
the same DSL search and conclusions already accept, supporting only the
session_id key (a session id, a bare list, or {"in": [...]}).
Unsupported keys and shapes are rejected with 422, never silently
ignored. Composes with session_id (must be included in the allowlist
when both are given). Capped at 1,000 sessions per request.

Enforcement is uniform at every recall chokepoint, fail-closed:
- dialectic prefetch + search_memory: conclusion recall restricted to
  the allowlist; dream docs (session_name IS NULL) excluded
- message tools (search/grep/date-range/temporal/context/history):
  strict intersection of allowlist and observer session membership
- get_reasoning_chain: unavailable under an allowlist (chains traverse
  provenance across sessions and cannot be scoped without leaking)
- empty allowlist short-circuits to empty results everywhere

Auth: workspace keys pass the allowlist as-given; peer-scoped JWTs must
be a member of every allowlisted session (403 otherwise), mirroring the
existing single-session check.

Fixes DEV-1995

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* fix: fail closed on empty session allowlist across all filter builders

Empty session allowlists relied solely on the early-return guard in
_get_working_representation_internal. The layers below it were
inconsistent, so a future direct caller (the DEV-1995 allowlist API)
would silently widen scope instead of failing closed:

- _build_filter_conditions used a truthiness check; an empty list was
  treated like None and dropped the filter. Now uses `is not None`,
  matching the recent/most-derived SQL paths.
- turbopuffer emitted a bare `In []` with undocumented (possibly
  fail-open) semantics. Now emits an explicit always-false predicate,
  mirroring lancedb's `1 = 0`.

Also extract the duplicated JSONB column tuple in filter.py to a
JSONB_COLUMNS constant.

Tests exercise each fail-closed guarantee at the layer it lives, rather
than masking it behind the early-return guard.

* fix: address tests

* fix(crud): fail closed when session_name is outside the allowlist

search/grep/history helpers scoped to a single session_name ignored the
session_names allowlist entirely — a caller could read a session the
allowlist forbids. The API routes guarded this with a 422, but the
dialectic tools call these CRUD functions directly and bypassed it.

Enforce it at the boundary: return [] when session_name is set and not in
the allowlist, across _semantic_search_messages (covers search_messages +
search_messages_temporal), grep_messages, get_messages_by_date_range,
get_recent_history, and get_observation_context.

Also rename the public param allowed_sessions -> session_names for
consistency with representation.py / chat.py / peers.py; the resolved
intersection keeps its distinct name allowed_session_names.

* fix: test

* fix(scopes): tighten and consolidate session allowlist per review

Addresses review feedback on the session allowlist (DEV-1995).

Behavior changes:

- Auth gate on peers.chat now uses active membership (left_at IS NULL)
  via get_peer_session_names(active_only=True), matching the adjacent
  is_peer_in_session check on options.session_id. Previously a peer that
  had left a session was denied when naming it directly but permitted
  when naming it in filters.session_id.
- Scoped conclusion recall is restricted to level == "explicit"
  (ALLOWLIST_SAFE_LEVELS). Dream-derived conclusions are stamped with a
  single session_name but synthesized across all sessions, so that stamp
  can't be scoped on. Applied at all four recall paths. Unscoped recall
  is unchanged. Follow-up to give conclusions an authoritative
  source-session set is tracked in DEV-2201.
- The allowlist gate checks `is not None` rather than truthiness, so
  filters={"session_id": []} reaches it instead of being skipped.

Refactors:

- New crud.message.resolve_session_scope replaces four near-identical
  copies of the allowlist-membership intersection. Returns
  (allowlist, deny) and never returns an empty list, so the None vs []
  distinction that external stores fail open on lives in one tested
  place. Takes db=None and opens its own short-lived session only when
  distinction that external stores fail open on lives in one tested
  place. Takes db=None and opens its own short-lived session only when
  an observer lookup is needed, preserving external-lookup-first
  ordering on the vector-store path.
- extract_session_allowlist takes must_include, collapsing the
  session_id-in-allowlist check duplicated across both peer routes.
- DialecticAgent._select_tools dedupes the two toolset-selection blocks
  and drops get_reasoning_chain under an allowlist, rather than paying
  for the schema plus a wasted turn to return a refusal.
- Rename session_names -> session_allowlist across crud, agent tools,
  dialectic and routes, to remove the one-character ambiguity with
  session_name. Internal only; the public filters.session_id surface is
  unchanged.

Docs:

- session_allowlist documented across all message and recall entry
  points, including the None / [] / populated contract.
- session_name marked deprecated for scoping. Not removed and not
  aliased: it also pins the query to one session, bypasses observer
  scoping, and drives session-history injection into the dialectic
  prompt, so it has no drop-in replacement.
- Note at the Document branch in utils/filter.py that the raw-key
  fallback is load-bearing for session scoping.

Tests: 20 -> 39 in tests/test_session_allowlist.py, covering the
peer-scoped JWT gate (member, non-member, left-session, workspace-key
bypass, empty allowlist), the resolve_session_scope tri-state including
the no-DB-checkout path, must_include, and the level narrowing.

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Vineeth Voruganti 2026-07-28 10:55:02 -04:00 committed by GitHub
parent 3ee890fa6f
commit e7cbcc8432
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 1276 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

181
src/crud/message.py
View File

@ -55,11 +55,28 @@ async def get_peer_session_names(
db: AsyncSession,
workspace_name: str,
peer_name: str,
*,
active_only: bool = False,
) -> list[str]:
"""Get all session names where a peer has any membership record.
"""Get all session names where a peer has a membership record.
Any membership record (regardless of joined_at/left_at) grants visibility
to all messages in that session.
By default any membership record (regardless of joined_at/left_at) grants
visibility to all messages in that session — this is the loose definition
recall scoping uses.
Pass ``active_only=True`` for the strict definition (``left_at IS NULL``),
matching :func:`src.crud.session.is_peer_in_session`. The auth layer must
use the strict one so that a single peer-scoped key gets the same answer
whether it names a session directly or via a filter allowlist.
Args:
db: Database session
workspace_name: Name of the workspace
peer_name: Name of the peer
active_only: Restrict to sessions the peer has not left
Returns:
Distinct session names the peer has a matching membership record in.
"""
stmt = (
select(models.session_peers_table.c.session_name)
@ -67,10 +84,80 @@ async def get_peer_session_names(
.where(models.session_peers_table.c.peer_name == peer_name)
.distinct()
)
if active_only:
stmt = stmt.where(models.session_peers_table.c.left_at.is_(None))
result = await db.execute(stmt)
return [row[0] for row in result.all()]
async def resolve_session_scope(
db: AsyncSession | None,
workspace_name: str,
session_name: str | None,
session_allowlist: list[str] | None,
observer: str | None,
*,
operation_name: str = "resolve_session_scope",
) -> tuple[list[str] | None, bool]:
"""Resolve the effective session scope for a message query.
Returns ``(allowed_session_names, deny)``:
- ``allowed_session_names is None`` — apply no allowlist filter. Either the
query is unrestricted, or ``session_name`` already pins it to one session.
- a populated list — restrict the query to exactly these sessions.
- ``deny=True`` — the caller must return an empty result *without* querying.
The distinction between ``None`` and an empty list is load-bearing: the
external vector stores drop an empty ``IN`` clause rather than matching
nothing, so collapsing the two would fail open. This function therefore
never returns an empty list — it returns ``deny=True`` instead.
Touches the database only when an observer lookup is actually required, so
callers on the external-vector-store path don't check out a connection
before their network call.
Args:
db: Database session to reuse. Pass None to let this function open its
own short-lived read-only session if (and only if) it needs one.
workspace_name: Name of the workspace
session_name: A single pinned session, if the caller named one
session_allowlist: Optional session allowlist. ``None`` is unrestricted;
an empty list fails closed.
observer: When set, scope is limited to this peer's sessions and then
intersected with ``session_allowlist``
operation_name: Label for the self-managed DB session, when one is opened
Returns:
Tuple of (allowlist to filter on or None, whether to deny outright).
"""
if session_name:
# A specific session was requested. Fail closed when the allowlist
# forbids it — routes guard this too, but other CRUD callers (the
# dialectic tools) don't, so enforce it at the boundary.
if session_allowlist is not None and session_name not in session_allowlist:
return None, True
return None, False
if observer is None:
if session_allowlist is None:
return None, False
allowed = list(session_allowlist)
return (allowed, False) if allowed else (None, True)
if db is not None:
allowed = await get_peer_session_names(db, workspace_name, observer)
else:
async with tracked_db(f"{operation_name}.peer_scope", read_only=True) as own_db:
allowed = await get_peer_session_names(own_db, workspace_name, observer)
if session_allowlist is not None:
scope = set(session_allowlist)
allowed = [s for s in allowed if s in scope]
return (allowed, False) if allowed else (None, True)
def _apply_token_limit(
base_conditions: list[ColumnElement[Any]], token_limit: int
) -> Select[tuple[models.Message]]:
@ -689,21 +776,29 @@ async def _semantic_search_messages(
after_date: datetime | None = None,
before_date: datetime | None = None,
observer: str | None = None,
session_allowlist: list[str] | None = None,
) -> list[tuple[list[models.Message], list[models.Message]]]:
"""Run semantic message search with optional temporal filters.
When observer is provided and session_name is None, results are
scoped to sessions the observer has any membership record in.
scoped to sessions the observer has any membership record in. When
session_allowlist is provided, that membership scope is further
intersected with the allowlist (fail-closed: empty result on empty
intersection).
"""
# Pre-fetch peer session scope if needed (short-lived DB session)
allowed_session_names: list[str] | None = None
if observer and not session_name:
async with tracked_db(f"{operation_name}.peer_scope", read_only=True) as db:
allowed_session_names = await get_peer_session_names(
db, workspace_name, observer
)
if not allowed_session_names:
return []
# db=None: the helper opens its own short-lived session only if it needs
# an observer lookup, so the external-store path below stays the first
# thing that happens when no observer scoping applies.
allowed_session_names, deny = await resolve_session_scope(
None,
workspace_name,
session_name,
session_allowlist,
observer,
operation_name=operation_name,
)
if deny:
return []
if settings.VECTOR_STORE.TYPE != "pgvector" and settings.VECTOR_STORE.MIGRATED:
message_ids = await _search_messages_external(
@ -758,6 +853,7 @@ async def search_messages(
context_window: int = 2,
embedding: list[float] | None = None,
observer: str | None = None,
session_allowlist: list[str] | None = None,
) -> list[tuple[list[models.Message], list[models.Message]]]:
"""
Search for messages using semantic similarity and return conversation snippets.
@ -768,12 +864,19 @@ async def search_messages(
Args:
workspace_name: Name of the workspace
session_name: Name of the session (optional)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
query: Search query text
limit: Maximum number of matching messages to return
context_window: Number of messages before/after each match to include
embedding: Optional pre-computed embedding
observer: When provided and session_name is None, scope results
to sessions this peer belongs to
session_allowlist: Optional session allowlist. None is unrestricted; an
empty list fails closed (empty result); a populated list is
intersected with the observer's session scope when observer is set
Returns:
List of tuples: (matched_messages, context_messages)
@ -799,6 +902,7 @@ async def search_messages(
context_window=context_window,
operation_name="message.search_messages",
observer=observer,
session_allowlist=session_allowlist,
)
@ -846,6 +950,7 @@ async def grep_messages(
limit: int = 10,
context_window: int = 2,
observer: str | None = None,
session_allowlist: list[str] | None = None,
) -> list[tuple[list[models.Message], list[models.Message]]]:
"""
Search for messages containing specific text (case-insensitive substring match).
@ -856,25 +961,29 @@ async def grep_messages(
Args:
workspace_name: Name of the workspace
session_name: Name of the session (optional - searches all sessions if None)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
text: Text to search for (case-insensitive)
limit: Maximum number of matching messages to return
context_window: Number of messages before/after each match to include
observer: When provided and session_name is None, scope results
to sessions this peer belongs to
session_allowlist: Optional session allowlist. None is unrestricted; an
empty list fails closed (empty result); a populated list is
intersected with the observer's session scope when observer is set
Returns:
List of tuples: (matched_messages, context_messages)
Each snippet may contain multiple matches if they were close together.
"""
async with tracked_db("message.grep_messages", read_only=True) as db:
# Pre-fetch peer session scope if needed
allowed_session_names = None
if observer and not session_name:
allowed_session_names = await get_peer_session_names(
db, workspace_name, observer
)
if not allowed_session_names:
return []
allowed_session_names, deny = await resolve_session_scope(
db, workspace_name, session_name, session_allowlist, observer
)
if deny:
return []
snippets = await _grep_messages_internal(
db,
@ -898,6 +1007,7 @@ async def get_messages_by_date_range(
limit: int = 20,
order: str = "desc",
observer: str | None = None,
session_allowlist: list[str] | None = None,
) -> list[models.Message]:
"""
Get messages within a date range.
@ -906,24 +1016,28 @@ async def get_messages_by_date_range(
db: Database session
workspace_name: Name of the workspace
session_name: Name of the session (optional - searches all sessions if None)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
after_date: Return messages after this datetime
before_date: Return messages before this datetime
limit: Maximum messages to return
order: Sort order - 'asc' for oldest first, 'desc' for newest first
observer: When provided and session_name is None, scope results
to sessions this peer belongs to
session_allowlist: Optional session allowlist. None is unrestricted; an
empty list fails closed (empty result); a populated list is
intersected with the observer's session scope when observer is set
Returns:
List of messages within the date range
"""
# Pre-fetch peer session scope if needed
allowed_session_names = None
if observer and not session_name:
allowed_session_names = await get_peer_session_names(
db, workspace_name, observer
)
if not allowed_session_names:
return []
allowed_session_names, deny = await resolve_session_scope(
db, workspace_name, session_name, session_allowlist, observer
)
if deny:
return []
stmt = select(models.Message).where(models.Message.workspace_name == workspace_name)
@ -957,6 +1071,7 @@ async def search_messages_temporal(
context_window: int = 2,
embedding: list[float] | None = None,
observer: str | None = None,
session_allowlist: list[str] | None = None,
) -> list[tuple[list[models.Message], list[models.Message]]]:
"""
Search for messages using semantic similarity with optional date filtering.
@ -967,6 +1082,10 @@ async def search_messages_temporal(
Args:
workspace_name: Name of the workspace
session_name: Name of the session (optional)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
query: Search query text
after_date: Only return messages after this datetime
before_date: Only return messages before this datetime
@ -975,6 +1094,9 @@ async def search_messages_temporal(
embedding: Optional pre-computed embedding for the query
observer: When provided and session_name is None, scope results
to sessions this peer belongs to
session_allowlist: Optional session allowlist. None is unrestricted; an
empty list fails closed (empty result); a populated list is
intersected with the observer's session scope when observer is set
Returns:
List of tuples: (matched_messages, context_messages)
@ -1001,4 +1123,5 @@ async def search_messages_temporal(
context_window=context_window,
operation_name="message.search_messages_temporal",
observer=observer,
session_allowlist=session_allowlist,
)

72
src/crud/representation.py
View File

@ -19,9 +19,11 @@ from src.telemetry.events import EmbeddingCallPurpose
from src.telemetry.logging import accumulate_metric
from src.utils.formatting import format_datetime_utc
from src.utils.representation import (
ALLOWLIST_SAFE_LEVELS,
DeductiveObservation,
ExplicitObservation,
Representation,
allowlist_safe_levels,
)
from src.utils.types import embedding_call_purpose
@ -213,7 +215,7 @@ class RepresentationManager:
self,
*,
db: AsyncSession | None = None,
session_names: list[str] | None = None,
session_allowlist: list[str] | None = None,
include_semantic_query: str | None = None,
embedding: list[float] | None = None,
semantic_search_top_k: int | None = None,
@ -229,7 +231,7 @@ class RepresentationManager:
Args:
db: Optional database session. If provided, uses it directly;
otherwise creates a new session via tracked_db.
session_names: Optional session allowlist to filter by. Applied
session_allowlist: Optional session allowlist to filter by. Applied
uniformly to every query path (semantic, most-derived, and
recent). None means no session restriction; an empty list
fail-closes to an empty representation.
@ -270,7 +272,7 @@ class RepresentationManager:
if db is not None:
return await self._get_working_representation_internal(
db,
session_names=session_names,
session_allowlist=session_allowlist,
include_semantic_query=include_semantic_query,
embedding=embedding,
semantic_search_top_k=semantic_search_top_k,
@ -284,7 +286,7 @@ class RepresentationManager:
) as new_db:
return await self._get_working_representation_internal(
new_db,
session_names=session_names,
session_allowlist=session_allowlist,
include_semantic_query=include_semantic_query,
embedding=embedding,
semantic_search_top_k=semantic_search_top_k,
@ -299,7 +301,7 @@ class RepresentationManager:
self,
db: AsyncSession,
*,
session_names: list[str] | None = None,
session_allowlist: list[str] | None = None,
include_semantic_query: str | None = None,
embedding: list[float] | None = None,
semantic_search_top_k: int | None = None,
@ -311,7 +313,7 @@ class RepresentationManager:
# Fail closed on an empty allowlist. This must short-circuit before
# any query: downstream stores drop an `IN ()` clause with an empty
# list (lancedb), which would silently widen the scope instead.
if session_names is not None and not session_names:
if session_allowlist is not None and not session_allowlist:
return Representation()
total = max_observations
@ -354,7 +356,7 @@ class RepresentationManager:
top_k=semantic_observations,
max_distance=semantic_search_max_distance,
embedding=embedding,
session_names=session_names,
session_allowlist=session_allowlist,
)
representation.merge_representation(
Representation.from_documents(semantic_docs)
@ -363,7 +365,7 @@ class RepresentationManager:
# Get most derived observations if requested
if include_most_derived:
derived_docs = await self._query_documents_most_derived(
db, top_k=top_observations, session_names=session_names
db, top_k=top_observations, session_allowlist=session_allowlist
)
representation.merge_representation(
Representation.from_documents(derived_docs)
@ -371,7 +373,7 @@ class RepresentationManager:
# Get recent observations
recent_docs = await self._query_documents_recent(
db, top_k=recent_observations, session_names=session_names
db, top_k=recent_observations, session_allowlist=session_allowlist
)
representation.merge_representation(Representation.from_documents(recent_docs))
@ -386,7 +388,7 @@ class RepresentationManager:
max_distance: float | None = None,
level: str | None = None,
embedding: list[float] | None = None,
session_names: list[str] | None = None,
session_allowlist: list[str] | None = None,
) -> list[models.Document]:
"""Query documents by semantic similarity."""
try:
@ -398,7 +400,7 @@ class RepresentationManager:
top_k,
max_distance,
embedding=embedding,
session_names=session_names,
session_allowlist=session_allowlist,
)
else:
documents = await crud.query_documents(
@ -410,7 +412,9 @@ class RepresentationManager:
max_distance=max_distance,
top_k=top_k,
embedding=embedding,
filters=self._build_filter_conditions(session_names=session_names)
filters=self._build_filter_conditions(
session_allowlist=session_allowlist
)
or None,
)
db.expunge_all()
@ -421,7 +425,7 @@ class RepresentationManager:
return []
async def _query_documents_recent(
self, db: AsyncSession, top_k: int, session_names: list[str] | None = None
self, db: AsyncSession, top_k: int, session_allowlist: list[str] | None = None
) -> list[models.Document]:
"""Query most recent documents."""
stmt = (
@ -433,8 +437,13 @@ class RepresentationManager:
models.Document.observed == self.observed,
models.Document.deleted_at.is_(None),
*(
[models.Document.session_name.in_(session_names)]
if session_names is not None
[
models.Document.session_name.in_(session_allowlist),
# Only levels with a trustworthy session stamp are
# scopeable — see ALLOWLIST_SAFE_LEVELS.
models.Document.level.in_(ALLOWLIST_SAFE_LEVELS),
]
if session_allowlist is not None
else []
),
)
@ -447,7 +456,7 @@ class RepresentationManager:
return list(documents)
async def _query_documents_most_derived(
self, db: AsyncSession, top_k: int, session_names: list[str] | None = None
self, db: AsyncSession, top_k: int, session_allowlist: list[str] | None = None
) -> list[models.Document]:
"""Query most derived documents."""
stmt = (
@ -459,8 +468,13 @@ class RepresentationManager:
models.Document.observed == self.observed,
models.Document.deleted_at.is_(None),
*(
[models.Document.session_name.in_(session_names)]
if session_names is not None
[
models.Document.session_name.in_(session_allowlist),
# Only levels with a trustworthy session stamp are
# scopeable — see ALLOWLIST_SAFE_LEVELS.
models.Document.level.in_(ALLOWLIST_SAFE_LEVELS),
]
if session_allowlist is not None
else []
),
)
@ -499,7 +513,7 @@ class RepresentationManager:
count: int,
max_distance: float | None = None,
embedding: list[float] | None = None,
session_names: list[str] | None = None,
session_allowlist: list[str] | None = None,
) -> list[models.Document]:
"""Query documents for a specific level."""
documents = await crud.query_documents(
@ -510,7 +524,9 @@ class RepresentationManager:
query=query,
max_distance=max_distance,
top_k=count,
filters=self._build_filter_conditions(level, session_names=session_names),
filters=self._build_filter_conditions(
level, session_allowlist=session_allowlist
),
embedding=embedding,
)
@ -523,13 +539,13 @@ class RepresentationManager:
def _build_filter_conditions(
self,
level: str | None = None,
session_names: list[str] | None = None,
session_allowlist: list[str] | None = None,
) -> dict[str, Any]:
"""
Build filter conditions for document queries.
Returns a flat dict of key-value pairs for vector store filtering.
Callers must not pass an empty session_names list — empty allowlists
Callers must not pass an empty session_allowlist list — empty allowlists
fail closed before any query is issued (see
_get_working_representation_internal).
"""
@ -542,8 +558,12 @@ class RepresentationManager:
# an empty `in` so downstream stores fail closed, matching
# _query_documents_recent / _query_documents_most_derived. Truthiness
# here would silently drop the filter and widen scope.
if session_names is not None:
filters["session_name"] = {"in": session_names}
if session_allowlist is not None:
filters["session_name"] = {"in": session_allowlist}
# Only levels with a trustworthy session stamp are scopeable. This
# overrides any narrower `level` above; an empty intersection emits
# `{"in": []}`, which matches nothing rather than everything.
filters["level"] = {"in": allowlist_safe_levels([level] if level else None)}
return filters
@ -557,7 +577,7 @@ async def get_working_representation(
db: AsyncSession | None = None,
observer: str,
observed: str,
session_names: list[str] | None = None,
session_allowlist: list[str] | None = None,
include_semantic_query: str | None = None,
embedding: list[float] | None = None,
semantic_search_top_k: int | None = None,
@ -590,7 +610,7 @@ async def get_working_representation(
)
return await manager.get_working_representation(
db=db,
session_names=session_names,
session_allowlist=session_allowlist,
include_semantic_query=include_semantic_query,
embedding=embedding,
semantic_search_top_k=semantic_search_top_k,

6
src/dialectic/chat.py
View File

@ -26,6 +26,7 @@ async def agentic_chat(
observer: str,
observed: str,
reasoning_level: ReasoningLevel = "low",
session_allowlist: list[str] | None = None,
response_model: type[BaseModel] | None = None,
) -> str:
"""
@ -38,6 +39,7 @@ async def agentic_chat(
observer: The peer making the query
observed: The peer being queried about
reasoning_level: Level of reasoning to apply
session_allowlist: Optional session allowlist restricting all recall
response_model: Optional Pydantic model the answer must conform to.
When set, the returned string is JSON matching the model's schema.
@ -82,6 +84,7 @@ async def agentic_chat(
observer_peer_card=observer_peer_card,
observed_peer_card=observed_peer_card,
reasoning_level=reasoning_level,
session_allowlist=session_allowlist,
)
return await agent.answer(query, response_model=response_model)
@ -94,6 +97,7 @@ async def agentic_chat_stream(
observer: str,
observed: str,
reasoning_level: ReasoningLevel = "low",
session_allowlist: list[str] | None = None,
response_model: type[BaseModel] | None = None,
) -> AsyncIterator[str]:
"""
@ -106,6 +110,7 @@ async def agentic_chat_stream(
observer: The peer making the query
observed: The peer being queried about
reasoning_level: Level of reasoning to apply
session_allowlist: Optional session allowlist restricting all recall
response_model: Optional Pydantic model the answer must conform to.
When set, the streamed text accumulates to JSON matching the
model's schema.
@ -151,6 +156,7 @@ async def agentic_chat_stream(
observer_peer_card=observer_peer_card,
observed_peer_card=observed_peer_card,
reasoning_level=reasoning_level,
session_allowlist=session_allowlist,
)
async for chunk in agent.answer_stream(query, response_model=response_model):

40
src/dialectic/core.py
View File

@ -70,6 +70,7 @@ class DialecticAgent:
metric_key: str | None = None,
reasoning_level: ReasoningLevel = "low",
session_id: str | None = None,
session_allowlist: list[str] | None = None,
):
"""
Initialize the dialectic agent.
@ -84,9 +85,13 @@ class DialecticAgent:
metric_key: Optional key for logging metrics (if provided, agent won't log separately)
reasoning_level: Level of reasoning to apply
session_id: ID used for grouping traces (not session_name)
session_allowlist: Optional session allowlist restricting all recall
(conclusions and messages) to these sessions; empty list
fails closed
"""
self.workspace_name: str = workspace_name
self.session_name: str | None = session_name
self.session_allowlist: list[str] | None = session_allowlist
self.session_id: str | None = session_id
self.observer: str = observer
self.observed: str = observed
@ -108,6 +113,24 @@ class DialecticAgent:
self._prefetched_conclusion_count: int = 0
self._run_id: str = generate_nanoid() # Always generate for event correlation
def _select_tools(self) -> list[dict[str, Any]]:
"""Pick the toolset for this query.
Minimal reasoning uses a reduced set to reduce cost. Under a session
allowlist `get_reasoning_chain` is dropped entirely rather than left in
to fail at call time: chains traverse provenance across sessions, so it
can't be scoped, and offering it costs both the schema in context and a
wasted turn when the model tries it.
"""
tools = (
DIALECTIC_TOOLS_MINIMAL
if self.reasoning_level == "minimal"
else DIALECTIC_TOOLS
)
if self.session_allowlist is not None:
tools = [t for t in tools if t.get("name") != "get_reasoning_chain"]
return tools
async def _initialize_session_history(self) -> None:
"""Fetch and inject session history into the system prompt if configured."""
if self._session_history_initialized:
@ -197,6 +220,7 @@ class DialecticAgent:
limit=prefetch_limit,
levels=["explicit"],
embedding=query_embedding,
session_allowlist=self.session_allowlist,
)
derived_repr = await search_memory(
@ -207,6 +231,7 @@ class DialecticAgent:
limit=prefetch_limit,
levels=["deductive", "inductive", "contradiction"],
embedding=query_embedding,
session_allowlist=self.session_allowlist,
)
if explicit_repr.is_empty() and derived_repr.is_empty():
@ -296,6 +321,7 @@ class DialecticAgent:
] = await create_tool_executor(
workspace_name=self.workspace_name,
session_name=self.session_name,
session_allowlist=self.session_allowlist,
observer=self.observer,
observed=self.observed,
history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT,
@ -438,12 +464,7 @@ class DialecticAgent:
# Get level-specific settings
level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level]
# Use minimal tools for minimal reasoning to reduce cost
tools = (
DIALECTIC_TOOLS_MINIMAL
if self.reasoning_level == "minimal"
else DIALECTIC_TOOLS
)
tools = self._select_tools()
# Use level-specific max_output_tokens if set, otherwise global default
max_tokens = (
level_settings.MAX_OUTPUT_TOKENS
@ -520,12 +541,7 @@ class DialecticAgent:
# Get level-specific settings
level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level]
# Use minimal tools for minimal reasoning to reduce cost
tools = (
DIALECTIC_TOOLS_MINIMAL
if self.reasoning_level == "minimal"
else DIALECTIC_TOOLS
)
tools = self._select_tools()
# Use level-specific max_output_tokens if set, otherwise global default
max_tokens = (
level_settings.MAX_OUTPUT_TOKENS

35
src/routers/peers.py
View File

@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src import crud, schemas
from src.config import settings
from src.crud.message import get_peer_session_names
from src.crud.session import is_peer_in_session
from src.dependencies import db, read_db, tracked_db
from src.dialectic.chat import agentic_chat, agentic_chat_stream
@ -27,6 +28,7 @@ from src.exceptions import (
from src.security import JWTParams, require_auth
from src.telemetry import prometheus_metrics
from src.telemetry.events import EmbeddingCallPurpose, GetContextEvent, emit
from src.utils.filter import extract_session_allowlist
from src.utils.schema_conversion import json_response_schema_to_pydantic
from src.utils.search import search
from src.utils.types import embedding_call_purpose
@ -199,6 +201,25 @@ async def chat(
):
raise AuthenticationException("JWT not permissioned for this resource")
# Parse the session allowlist from filters (422 on unsupported keys/shapes,
# and on a session_id the allowlist doesn't cover).
session_allowlist = extract_session_allowlist(
options.filters, must_include=options.session_id
)
# A peer-scoped key may only name sessions its peer belongs to — the
# allowlist reaches message recall the same way session_id does above.
# `active_only` matches the is_peer_in_session check above, so both gates
# answer the same question for a peer that has left a session.
if jwt_params.p is not None and session_allowlist is not None:
async with tracked_db("peers.chat.session_scope_auth", read_only=True) as s_db:
member_sessions = set(
await get_peer_session_names(
s_db, workspace_id, jwt_params.p, active_only=True
)
)
if not set(session_allowlist) <= member_sessions:
raise AuthenticationException("JWT not permissioned for this resource")
# Convert the caller's JSON Schema so malformed schemas fail immediately with 422
response_model: type[BaseModel] | None = None
if options.response_format is not None:
@ -244,6 +265,7 @@ async def chat(
observer=peer_id,
observed=options.target if options.target is not None else peer_id,
reasoning_level=options.reasoning_level,
session_allowlist=session_allowlist,
response_model=response_model,
)
),
@ -259,6 +281,7 @@ async def chat(
# and it's answered from the omniscient Honcho perspective
observed=options.target if options.target is not None else peer_id,
reasoning_level=options.reasoning_level,
session_allowlist=session_allowlist,
response_model=response_model,
)
@ -294,6 +317,12 @@ async def get_representation(
If a target is provided, we get the Representation of the target from the perspective of the Peer.
If no target is provided, we get the omniscient Honcho Representation of the Peer.
"""
# Parse the session allowlist from filters (422 on unsupported keys/shapes,
# and on a session_id the allowlist doesn't cover).
session_allowlist = extract_session_allowlist(
options.filters, must_include=options.session_id
)
try:
embedding: list[float] | None = None
if options.search_query:
@ -312,9 +341,9 @@ async def get_representation(
workspace_id,
observer=peer_id,
observed=options.target if options.target is not None else peer_id,
session_names=[options.session_id]
session_allowlist=[options.session_id]
if options.session_id is not None
else None,
else session_allowlist,
include_semantic_query=options.search_query,
embedding=embedding,
semantic_search_top_k=options.search_top_k,
@ -477,7 +506,7 @@ async def get_peer_context(
workspace_id,
observer=peer_id,
observed=observed,
session_names=None, # Peer context is global, not session-scoped
session_allowlist=None, # Peer context is global, not session-scoped
include_semantic_query=search_query,
embedding=embedding,
semantic_search_top_k=search_top_k,

8
src/routers/sessions.py
View File

@ -43,7 +43,7 @@ async def _get_working_representation_task(
*,
observer: str,
observed: str,
session_names: list[str] | None,
session_allowlist: list[str] | None,
search_top_k: int | None,
search_max_distance: float | None,
include_most_derived: bool,
@ -59,7 +59,7 @@ async def _get_working_representation_task(
last_message: Optional last message for semantic query
observer: Name of the observer peer
observed: Name of the observed peer
session_names: Optional session allowlist to filter by
session_allowlist: Optional session allowlist to filter by
search_top_k: Number of semantic-search-retrieved observations to include in the representation
search_max_distance: Maximum distance to search for semantically relevant observations
include_most_derived: Whether to include the most derived observations in the representation
@ -74,7 +74,7 @@ async def _get_working_representation_task(
db=db,
observer=observer,
observed=observed,
session_names=session_names,
session_allowlist=session_allowlist,
include_semantic_query=last_message,
semantic_search_top_k=search_top_k,
semantic_search_max_distance=search_max_distance,
@ -765,7 +765,7 @@ async def get_session_context(
search_query,
observer=observer,
observed=observed,
session_names=[session_id] if limit_to_session else None,
session_allowlist=[session_id] if limit_to_session else None,
search_top_k=search_top_k,
search_max_distance=search_max_distance,
include_most_derived=include_most_frequent,

19
src/schemas/api.py
View File

@ -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",

125
src/utils/agent_tools.py
View File

@ -1,7 +1,7 @@
import asyncio
import logging
import weakref
from collections.abc import Callable
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from datetime import datetime
from typing import Any, cast
@ -30,7 +30,11 @@ from src.utils.formatting import (
parse_datetime_iso,
utc_now_iso,
)
from src.utils.representation import Representation
from src.utils.representation import (
ALLOWLIST_SAFE_LEVELS,
Representation,
allowlist_safe_levels,
)
from src.utils.types import ToolResult, embedding_call_purpose, get_current_iteration
logger = logging.getLogger(__name__)
@ -1030,6 +1034,7 @@ async def get_recent_history(
session_name: str | None,
observed: str | None = None,
token_limit: int = 8192,
session_allowlist: list[str] | None = None,
) -> list[models.Message]:
"""
Retrieve recent conversation history.
@ -1042,6 +1047,10 @@ async def get_recent_history(
db: Database session
workspace_name: Workspace identifier
session_name: Session identifier (optional)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
observed: Peer name to filter by when no session specified (optional)
token_limit: Maximum tokens to retrieve (default: 8192)
@ -1049,6 +1058,9 @@ async def get_recent_history(
List of messages in chronological order
"""
if session_name:
# Fail closed: a specific session outside the allowlist is not readable.
if session_allowlist is not None and session_name not in session_allowlist:
return []
# Get messages from a specific session
messages_stmt = await crud.get_messages(
workspace_name=workspace_name,
@ -1061,7 +1073,11 @@ async def get_recent_history(
# Return in chronological order
return list(reversed(messages))
elif observed:
# Fail closed on an empty allowlist
if session_allowlist is not None and not session_allowlist:
return []
# Get recent messages from the observed peer across all sessions
# (restricted to the session allowlist when one is provided)
stmt = (
select(models.Message)
.where(models.Message.workspace_name == workspace_name)
@ -1069,6 +1085,8 @@ async def get_recent_history(
.order_by(models.Message.created_at.desc())
.limit(50) # Limit to recent messages
)
if session_allowlist is not None:
stmt = stmt.where(models.Message.session_name.in_(session_allowlist))
result = await db.execute(stmt)
messages = list(result.scalars().all())
# Return in chronological order
@ -1086,6 +1104,7 @@ async def search_memory(
limit: int,
levels: list[str] | None = None,
embedding: list[float] | None = None,
session_allowlist: list[str] | None = None,
) -> Representation:
"""
Search for observations in memory using semantic similarity.
@ -1106,10 +1125,22 @@ async def search_memory(
Returns:
Representation object containing relevant observations
"""
# Build filter for levels if specified
filters: dict[str, Any] | None = None
# Fail closed on an empty allowlist — downstream stores drop empty IN
# clauses, which would silently widen scope.
if session_allowlist is not None and not session_allowlist:
return Representation()
if session_allowlist is not None:
levels = allowlist_safe_levels(levels)
if not levels:
return Representation()
# Build filters for levels / session allowlist if specified
filters: dict[str, Any] = {}
if levels:
filters = {"level": {"in": levels}}
filters["level"] = {"in": levels}
if session_allowlist is not None:
filters["session_name"] = {"in": session_allowlist}
documents = await crud.query_documents(
db=None,
@ -1118,7 +1149,7 @@ async def search_memory(
observed=observed,
query=query,
top_k=limit,
filters=filters,
filters=filters or None,
embedding=embedding,
)
@ -1131,6 +1162,7 @@ async def get_observation_context(
session_name: str | None,
message_ids: list[str],
observer: str | None = None,
session_allowlist: list[str] | None = None,
) -> list[models.Message]:
"""
Retrieve messages for given message IDs along with surrounding context.
@ -1143,9 +1175,16 @@ async def get_observation_context(
db: Database session
workspace_name: Workspace identifier
session_name: Session identifier (optional)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
message_ids: List of message IDs to retrieve
observer: When provided and session_name is None, scope results
to sessions this peer belongs to
session_allowlist: Optional session allowlist. None is unrestricted; an
empty list fails closed (empty result); a populated list is
intersected with the observer's session scope when observer is set
Returns:
List of messages in chronological order, including the requested messages and surrounding context
@ -1153,16 +1192,13 @@ async def get_observation_context(
if not message_ids:
return []
# Pre-fetch peer session scope if needed
allowed_session_names: list[str] | None = None
if observer and not session_name:
from src.crud.message import get_peer_session_names
from src.crud.message import resolve_session_scope
allowed_session_names = await get_peer_session_names(
db, workspace_name, observer
)
if not allowed_session_names:
return []
allowed_session_names, deny = await resolve_session_scope(
db, workspace_name, session_name, session_allowlist, observer
)
if deny:
return []
# Use a CTE to get seq_in_session values for target messages
stmt = (
@ -1221,9 +1257,16 @@ async def extract_preferences(
Args:
workspace_name: Workspace identifier
session_name: Session identifier (optional)
Deprecated for *scoping*: prefer session_allowlist, which
intersects with observer membership. This parameter also pins
the query to one session and bypasses observer scoping, so it
is not a drop-in equivalent and is not removed.
observed: The peer whose preferences to extract
observer: When provided and session_name is None, scope results
to sessions this peer belongs to
session_allowlist: Optional session allowlist. None is unrestricted; an
empty list fails closed (empty result); a populated list is
intersected with the observer's session scope when observer is set
Returns:
Dict with 'messages' list containing potentially relevant messages
@ -1303,6 +1346,10 @@ class ToolContext:
db_lock: asyncio.Lock
# Optional resolved configuration for checking feature flags
configuration: ResolvedConfiguration | None = None
# Optional session allowlist (dialectic filters). When set, message and
# conclusion recall is restricted to these sessions (intersected with
# observer membership); empty list fails closed.
session_allowlist: list[str] | None = None
# Telemetry context fields
run_id: str | None = None
agent_type: str | None = None # "dialectic", "deriver", "dreamer"
@ -1724,6 +1771,7 @@ async def _handle_get_recent_history(
session_name=ctx.session_name,
observed=ctx.observed,
token_limit=ctx.history_token_limit,
session_allowlist=ctx.session_allowlist,
)
if not history:
return "No conversation history available"
@ -1769,15 +1817,28 @@ async def _handle_search_memory(
"query_tokens": _estimate_tokens_safe(query),
}
documents = await crud.query_documents(
db=None,
workspace_name=ctx.workspace_name,
observer=ctx.observer,
observed=ctx.observed,
query=query,
top_k=top_k,
embedding=query_embedding,
)
# Restrict conclusion recall to the session allowlist when one is set.
# Empty allowlist fails closed (downstream stores drop empty IN clauses),
# and only levels with a trustworthy session stamp are served.
documents: Sequence[models.Document]
if ctx.session_allowlist is not None and not ctx.session_allowlist:
documents = []
else:
documents = await crud.query_documents(
db=None,
workspace_name=ctx.workspace_name,
observer=ctx.observer,
observed=ctx.observed,
query=query,
top_k=top_k,
embedding=query_embedding,
filters={
"session_name": {"in": ctx.session_allowlist},
"level": {"in": list(ALLOWLIST_SAFE_LEVELS)},
}
if ctx.session_allowlist is not None
else None,
)
mem = Representation.from_documents(documents)
total_count = mem.len()
if total_count == 0:
@ -1799,6 +1860,7 @@ async def _handle_search_memory(
context_window=0,
embedding=query_embedding,
observer=ctx.observer,
session_allowlist=ctx.session_allowlist,
)
if snippets:
message_output = _format_message_snippets(
@ -1840,6 +1902,7 @@ async def _handle_get_observation_context(
session_name=ctx.session_name,
message_ids=tool_input["message_ids"],
observer=ctx.observer,
session_allowlist=ctx.session_allowlist,
)
if not messages:
return f"No messages found for IDs {tool_input['message_ids']}"
@ -1882,6 +1945,7 @@ async def _handle_search_messages(
context_window=2,
embedding=query_embedding,
observer=ctx.observer,
session_allowlist=ctx.session_allowlist,
)
search_meta: dict[str, Any] = {
"top_k": limit,
@ -1918,6 +1982,7 @@ async def _handle_grep_messages(
limit=limit,
context_window=context_window,
observer=ctx.observer,
session_allowlist=ctx.session_allowlist,
)
if not snippets:
return f"No messages found containing '{text}'"
@ -1982,6 +2047,7 @@ async def _handle_get_messages_by_date_range(
limit=limit,
order=order,
observer=ctx.observer,
session_allowlist=ctx.session_allowlist,
)
msg_count = len(messages)
messages_text = (
@ -2054,6 +2120,7 @@ async def _handle_search_messages_temporal(
before_date=before_date,
limit=limit,
context_window=context_window,
session_allowlist=ctx.session_allowlist,
embedding=query_embedding,
observer=ctx.observer,
)
@ -2310,6 +2377,14 @@ async def _handle_get_reasoning_chain(
ctx: ToolContext, tool_input: dict[str, Any]
) -> str:
"""Handle get_reasoning_chain tool."""
# Reasoning chains traverse provenance across sessions by design, so a
# session allowlist cannot be enforced on the traversal without exposing
# out-of-scope premises/conclusions. Fail closed rather than leak.
if ctx.session_allowlist is not None:
return (
"Reasoning-chain traversal is unavailable for session-scoped "
"queries. Use search_memory and message tools instead."
)
observation_id = tool_input.get("observation_id")
if not observation_id:
return "ERROR: 'observation_id' is required"
@ -2435,6 +2510,7 @@ async def create_tool_executor(
run_id: str | None = None,
agent_type: str | None = None,
parent_category: str | None = None,
session_allowlist: list[str] | None = None,
) -> Callable[[str, dict[str, Any]], Any]:
"""
Create a unified tool executor function for all agent operations.
@ -2475,6 +2551,7 @@ async def create_tool_executor(
history_token_limit=history_token_limit,
db_lock=shared_lock,
configuration=configuration,
session_allowlist=session_allowlist,
run_id=run_id,
agent_type=agent_type,
parent_category=parent_category,

84
src/utils/filter.py
View File

@ -61,6 +61,83 @@ ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_DOCUMENTS = {
}
MAX_SESSION_ALLOWLIST_ENTRIES = 1000
def extract_session_allowlist(
filters: dict[str, Any] | None,
must_include: str | None = None,
) -> list[str] | None:
"""Parse a recall-path ``filters`` body into a session allowlist.
The dialectic and representation endpoints accept a constrained subset of
the filter DSL: only the ``session_id`` key, valued as a single id, a bare
list of ids, or ``{"in": [...]}``. Unsupported keys or shapes raise
FilterError (422) rather than being silently ignored — a dropped filter
on these endpoints would widen recall scope.
Args:
filters: The raw ``filters`` body, or None.
must_include: A session id that must appear in the parsed allowlist —
used by routes that also accept a top-level ``session_id``, so the
two can't contradict each other. Ignored when filters is None.
Returns:
None when filters is None. An explicit empty list is preserved so
downstream consumers fail closed.
Raises:
FilterError: On an unsupported key or shape, an over-cap list, or a
``must_include`` session missing from the allowlist.
"""
if filters is None:
return None
unsupported = set(filters) - {"session_id"}
if unsupported:
raise FilterError(
f"Unsupported filter key(s) for this endpoint: {sorted(unsupported)}. Only 'session_id' is supported."
)
if "session_id" not in filters:
raise FilterError("filters must contain 'session_id'")
value = filters["session_id"]
entries: list[Any]
if isinstance(value, str):
entries = [value]
elif isinstance(value, list):
entries = list(typing_cast(Sequence[Any], value))
elif (
isinstance(value, dict)
and set(typing_cast("dict[str, Any]", value)) == {"in"}
and isinstance(value["in"], list)
):
entries = list(typing_cast(Sequence[Any], value["in"]))
else:
raise FilterError(
'filters.session_id must be a session id, a list of session ids, or {"in": [...]}'
)
if len(entries) > MAX_SESSION_ALLOWLIST_ENTRIES:
raise FilterError(
f"filters.session_id supports at most {MAX_SESSION_ALLOWLIST_ENTRIES} sessions per request"
)
allowlist: list[str] = []
seen: set[str] = set()
for entry in entries:
if not isinstance(entry, str) or not entry:
raise FilterError("filters.session_id entries must be non-empty strings")
if entry not in seen:
seen.add(entry)
allowlist.append(entry)
if must_include is not None and must_include not in seen:
raise FilterError("session_id must be included in filters.session_id")
return allowlist
def apply_filter(
stmt: Select[tuple[T]], model_class: type[T], filters: dict[str, Any] | None = None
) -> Select[tuple[T]]:
@ -221,6 +298,13 @@ def _build_field_condition(
if model_class.__name__ == "Message":
column_name = ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_MESSAGES.get(key)
elif model_class.__name__ == "Document":
# NOTE: unlike Message/Workspace, Document falls back to the raw key so
# internal callers can filter on internal column names. The session
# allowlist depends on this: recall passes {"session_name": {"in": ...}}
# (see search_memory in utils/agent_tools.py and RepresentationManager),
# and "session_name" is deliberately absent from the mapping below.
# Tightening this to a strict allowlist would break session scoping —
# fail-closed, since an unmapped key raises, but silently.
column_name = ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING_DOCUMENTS.get(
key,
key, # fallback to the key itself if not found in the mapping for internal use here

29
src/utils/representation.py
View File

@ -7,6 +7,35 @@ from pydantic import BaseModel, Field, field_validator
from src import models
from src.utils.formatting import parse_datetime_iso
# Conclusion levels whose `session_name` stamp is trustworthy enough to scope on.
#
# Explicit conclusions come from the deriver over a single session's message
# batch, so their stamp is authoritative. Deductive/inductive conclusions are
# produced by the dreamer, which reads across *all* sessions (its discovery
# tools default to session_only=False) but stamps its output with one session —
# whichever holds the most recent explicit conclusion, see
# dreamer/dream_scheduler.py. Serving those under a session allowlist would leak
# conclusions synthesized from sessions outside it.
#
# ponytail: whole-level exclusion rather than per-conclusion provenance. The
# reasoning trees already link each conclusion to its premises, so the real fix
# is an authoritative source-session set per conclusion; until that exists this
# fails closed. Tracked in DEV-2201.
ALLOWLIST_SAFE_LEVELS = ("explicit",)
def allowlist_safe_levels(levels: list[str] | None) -> list[str]:
"""Narrow a level filter to those safe to serve under a session allowlist.
Returns the intersection with :data:`ALLOWLIST_SAFE_LEVELS`; ``None`` means
"no level filter requested" and yields the full safe set. An empty result
means the caller asked only for levels we can't scope, and should receive
nothing rather than unscoped conclusions.
"""
if levels is None:
return list(ALLOWLIST_SAFE_LEVELS)
return [level for level in levels if level in ALLOWLIST_SAFE_LEVELS]
def _strip_microseconds_and_timezone(timestamp: datetime) -> datetime:
"""

43
tests/crud/test_representation_manager.py
View File

@ -320,7 +320,7 @@ class TestRepresentationManagerSessionScoping:
session_a, _, manager = await self._setup(db_session, test_workspace, test_peer)
results = await manager._query_documents_recent( # pyright: ignore[reportPrivateUsage]
db_session, top_k=10, session_names=[session_a.name]
db_session, top_k=10, session_allowlist=[session_a.name]
)
contents = [doc.content for doc in results]
@ -337,7 +337,7 @@ class TestRepresentationManagerSessionScoping:
session_a, _, manager = await self._setup(db_session, test_workspace, test_peer)
results = await manager._query_documents_most_derived( # pyright: ignore[reportPrivateUsage]
db_session, top_k=10, session_names=[session_a.name]
db_session, top_k=10, session_allowlist=[session_a.name]
)
contents = [doc.content for doc in results]
@ -361,12 +361,15 @@ class TestRepresentationManagerSessionScoping:
query="anything",
top_k=5,
embedding=[0.1],
session_names=[session_a.name],
session_allowlist=[session_a.name],
)
assert mock_query.await_args is not None
assert mock_query.await_args.kwargs["filters"] == {
"session_name": {"in": [session_a.name]}
"session_name": {"in": [session_a.name]},
# Scoped recall serves only levels with a trustworthy session
# stamp (ALLOWLIST_SAFE_LEVELS / DEV-2201).
"level": {"in": ["explicit"]},
}
@pytest.mark.asyncio
@ -403,7 +406,7 @@ class TestRepresentationManagerSessionScoping:
representation = await manager.get_working_representation(
db=db_session,
session_names=[session_a.name],
session_allowlist=[session_a.name],
include_most_derived=True,
)
@ -425,7 +428,7 @@ class TestRepresentationManagerSessionScoping:
representation = await manager.get_working_representation(
db=db_session,
session_names=[],
session_allowlist=[],
include_most_derived=True,
)
@ -441,13 +444,29 @@ class TestRepresentationManagerSessionScoping:
"workspace", observer="observer", observed="observed"
)
assert manager._build_filter_conditions(session_names=[]) == { # pyright: ignore[reportPrivateUsage]
"session_name": {"in": []}
# Scoping also narrows to levels whose session stamp is trustworthy
# (see ALLOWLIST_SAFE_LEVELS / DEV-2201).
assert manager._build_filter_conditions(session_allowlist=[]) == { # pyright: ignore[reportPrivateUsage]
"session_name": {"in": []},
"level": {"in": ["explicit"]},
}
# None means unscoped — no session filter emitted.
assert manager._build_filter_conditions(session_names=None) == {} # pyright: ignore[reportPrivateUsage]
assert manager._build_filter_conditions(session_names=["s1"]) == { # pyright: ignore[reportPrivateUsage]
"session_name": {"in": ["s1"]}
# None means unscoped — no session filter and no level narrowing.
assert manager._build_filter_conditions(session_allowlist=None) == {} # pyright: ignore[reportPrivateUsage]
assert manager._build_filter_conditions(session_allowlist=["s1"]) == { # pyright: ignore[reportPrivateUsage]
"session_name": {"in": ["s1"]},
"level": {"in": ["explicit"]},
}
# A requested level outside the safe set yields an empty `in`, which
# matches nothing rather than falling back to unscoped recall.
assert manager._build_filter_conditions( # pyright: ignore[reportPrivateUsage]
level="inductive", session_allowlist=["s1"]
) == {
"session_name": {"in": ["s1"]},
"level": {"in": []},
}
# ...while an unscoped level filter is left exactly as asked.
assert manager._build_filter_conditions(level="inductive") == { # pyright: ignore[reportPrivateUsage]
"level": "inductive"
}

679
tests/test_session_allowlist.py Normal file
View File

@ -0,0 +1,679 @@
"""
Tests for the session allowlist (DEV-1995).
Covers the constrained `filters` surface on dialectic/representation
(extract_session_allowlist), fail-closed conclusion recall (search_memory),
and the strict allowlist ∩ membership intersection in message cruds.
"""
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from fastapi.testclient import TestClient
from nanoid import generate as generate_nanoid
from sqlalchemy.ext.asyncio import AsyncSession
from src import crud, models
from src.config import settings
from src.crud.message import resolve_session_scope
from src.exceptions import FilterError
from src.models import Peer, Workspace
from src.security import JWTParams, create_jwt
from src.utils.agent_tools import search_memory
from src.utils.filter import (
MAX_SESSION_ALLOWLIST_ENTRIES,
extract_session_allowlist,
)
class TestExtractSessionAllowlist:
def test_none_passthrough(self):
assert extract_session_allowlist(None) is None
def test_single_id(self):
assert extract_session_allowlist({"session_id": "s1"}) == ["s1"]
def test_bare_list(self):
assert extract_session_allowlist({"session_id": ["s1", "s2"]}) == ["s1", "s2"]
def test_in_operator(self):
assert extract_session_allowlist({"session_id": {"in": ["s1"]}}) == ["s1"]
def test_dedupes_preserving_order(self):
assert extract_session_allowlist({"session_id": ["s2", "s1", "s2"]}) == [
"s2",
"s1",
]
def test_empty_list_preserved_for_fail_closed(self):
assert extract_session_allowlist({"session_id": []}) == []
def test_unsupported_key_rejected(self):
with pytest.raises(FilterError, match="Unsupported filter key"):
extract_session_allowlist({"peer_id": ["a"], "session_id": ["s1"]})
def test_missing_session_id_rejected(self):
with pytest.raises(FilterError, match="must contain"):
extract_session_allowlist({})
def test_bad_shapes_rejected(self):
for bad in [123, {"gte": "x"}, {"in": "s1"}, [1, 2], [""], None]:
with pytest.raises(FilterError):
extract_session_allowlist({"session_id": bad})
def test_cap_enforced(self):
too_many = [f"s{i}" for i in range(MAX_SESSION_ALLOWLIST_ENTRIES + 1)]
with pytest.raises(FilterError, match="at most"):
extract_session_allowlist({"session_id": too_many})
def test_must_include_satisfied(self):
assert extract_session_allowlist(
{"session_id": ["s1", "s2"]}, must_include="s2"
) == ["s1", "s2"]
def test_must_include_missing_rejected(self):
with pytest.raises(FilterError, match="must be included"):
extract_session_allowlist({"session_id": ["s1"]}, must_include="s2")
def test_must_include_ignored_without_filters(self):
assert extract_session_allowlist(None, must_include="s1") is None
def test_must_include_none_is_no_constraint(self):
assert extract_session_allowlist({"session_id": ["s1"]}, must_include=None) == [
"s1"
]
class TestSearchMemoryAllowlist:
@pytest.mark.asyncio
async def test_allowlist_pushed_down_as_filters(self):
with patch(
"src.crud.query_documents", new=AsyncMock(return_value=[])
) as mock_query:
await search_memory(
workspace_name="w",
observer="o",
observed="o",
query="q",
limit=5,
levels=["explicit"],
embedding=[0.1],
session_allowlist=["s1", "s2"],
)
assert mock_query.await_args is not None
assert mock_query.await_args.kwargs["filters"] == {
"level": {"in": ["explicit"]},
"session_name": {"in": ["s1", "s2"]},
}
@pytest.mark.asyncio
async def test_allowlist_narrows_levels_to_allowlist_safe(self):
"""Only levels with a trustworthy session stamp survive scoping.
Dream-derived levels are stamped with one session but synthesized
across many (DEV-2201), so they can't be served under an allowlist.
"""
with patch(
"src.crud.query_documents", new=AsyncMock(return_value=[])
) as mock_query:
await search_memory(
workspace_name="w",
observer="o",
observed="o",
query="q",
limit=5,
levels=["explicit", "inductive"],
embedding=[0.1],
session_allowlist=["s1"],
)
assert mock_query.await_args is not None
assert mock_query.await_args.kwargs["filters"]["level"] == {"in": ["explicit"]}
@pytest.mark.asyncio
async def test_allowlist_defaults_to_explicit_when_no_levels_requested(self):
with patch(
"src.crud.query_documents", new=AsyncMock(return_value=[])
) as mock_query:
await search_memory(
workspace_name="w",
observer="o",
observed="o",
query="q",
limit=5,
embedding=[0.1],
session_allowlist=["s1"],
)
assert mock_query.await_args is not None
assert mock_query.await_args.kwargs["filters"]["level"] == {"in": ["explicit"]}
@pytest.mark.asyncio
async def test_derived_only_request_under_allowlist_returns_empty(self):
"""The dialectic's derived prefetch short-circuits instead of querying."""
with patch(
"src.crud.query_documents", new=AsyncMock(return_value=[])
) as mock_query:
result = await search_memory(
workspace_name="w",
observer="o",
observed="o",
query="q",
limit=5,
levels=["deductive", "inductive", "contradiction"],
embedding=[0.1],
session_allowlist=["s1"],
)
mock_query.assert_not_awaited()
assert result.is_empty()
@pytest.mark.asyncio
async def test_levels_untouched_without_allowlist(self):
"""No allowlist means no level narrowing — unscoped recall is unchanged."""
with patch(
"src.crud.query_documents", new=AsyncMock(return_value=[])
) as mock_query:
await search_memory(
workspace_name="w",
observer="o",
observed="o",
query="q",
limit=5,
levels=["deductive", "inductive"],
embedding=[0.1],
)
assert mock_query.await_args is not None
assert mock_query.await_args.kwargs["filters"] == {
"level": {"in": ["deductive", "inductive"]}
}
@pytest.mark.asyncio
async def test_empty_allowlist_fails_closed_without_querying(self):
with patch(
"src.crud.query_documents", new=AsyncMock(return_value=[])
) as mock_query:
result = await search_memory(
workspace_name="w",
observer="o",
observed="o",
query="q",
limit=5,
embedding=[0.1],
session_allowlist=[],
)
mock_query.assert_not_awaited()
assert result.is_empty()
class TestMessageCrudAllowlistIntersection:
"""allowlist ∩ observer-membership, fail-closed on empty intersection."""
async def _setup_two_sessions(
self,
client: TestClient,
workspace: Workspace,
peer: Peer,
) -> tuple[str, str]:
ids: list[str] = []
for marker in ("alpha", "beta"):
session_id = str(generate_nanoid())
resp = client.post(
f"/v3/workspaces/{workspace.name}/sessions",
json={"id": session_id, "peer_names": {peer.name: {}}},
)
assert resp.status_code == 201
resp = client.post(
f"/v3/workspaces/{workspace.name}/sessions/{session_id}/messages",
json={
"messages": [
{
"content": f"needle in {marker}",
"peer_id": peer.name,
}
]
},
)
assert resp.status_code == 201
ids.append(session_id)
return ids[0], ids[1]
@pytest.mark.asyncio
async def test_grep_messages_intersects_allowlist(
self,
client: TestClient,
sample_data: tuple[Workspace, Peer],
):
workspace, peer = sample_data
session_a, session_b = await self._setup_two_sessions(client, workspace, peer)
snippets = await crud.grep_messages(
workspace_name=workspace.name,
session_name=None,
text="needle",
observer=peer.name,
session_allowlist=[session_a],
)
contents = [m.content for matches, _ in snippets for m in matches]
assert contents == ["needle in alpha"]
# A session the observer is NOT a member of contributes nothing,
# even when allowlisted (strict intersection).
foreign = str(generate_nanoid())
snippets = await crud.grep_messages(
workspace_name=workspace.name,
session_name=None,
text="needle",
observer=peer.name,
session_allowlist=[foreign],
)
assert snippets == []
# Both sessions allowlisted -> both found
snippets = await crud.grep_messages(
workspace_name=workspace.name,
session_name=None,
text="needle",
observer=peer.name,
session_allowlist=[session_a, session_b],
)
assert len(snippets) == 2
@pytest.mark.asyncio
async def test_get_messages_by_date_range_intersects_allowlist(
self,
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
workspace, peer = sample_data
session_a, _session_b = await self._setup_two_sessions(client, workspace, peer)
messages = await crud.get_messages_by_date_range(
db_session,
workspace_name=workspace.name,
session_name=None,
observer=peer.name,
session_allowlist=[session_a],
)
assert [m.content for m in messages] == ["needle in alpha"]
# Empty allowlist fails closed
messages = await crud.get_messages_by_date_range(
db_session,
workspace_name=workspace.name,
session_name=None,
observer=peer.name,
session_allowlist=[],
)
assert messages == []
class TestPeerScopedJWTAllowlistGate:
"""A peer-scoped key may only allowlist sessions its peer actively belongs to.
The gate uses `active_only=True` so it agrees with the `is_peer_in_session`
check on `options.session_id` — a peer that has left a session is denied by
both, not just one.
"""
@pytest.fixture(autouse=True)
def _enable_auth(self, monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(settings.AUTH, "USE_AUTH", True)
monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret")
def _chat_as(
self,
client: TestClient,
workspace: Workspace,
peer: Peer,
token: str,
body: dict[str, Any],
):
return client.post(
f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat",
json={"query": "what do you know?", **body},
headers={"Authorization": f"Bearer {token}"},
)
async def _session_with(
self, client: TestClient, workspace: Workspace, peer: Peer
) -> str:
session_id = str(generate_nanoid())
resp = client.post(
f"/v3/workspaces/{workspace.name}/sessions",
json={"id": session_id, "peer_names": {peer.name: {}}},
)
assert resp.status_code == 201
return session_id
@pytest.mark.asyncio
async def test_member_sessions_allowed(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
session_id = await self._session_with(client, workspace, peer)
token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
with patch(
"src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")
) as mock_chat:
resp = self._chat_as(
client,
workspace,
peer,
token,
{"filters": {"session_id": [session_id]}},
)
assert resp.status_code == 200
# The allowlist reaches the agent rather than being dropped at the gate.
assert mock_chat.await_args is not None
assert mock_chat.await_args.kwargs["session_allowlist"] == [session_id]
@pytest.mark.asyncio
async def test_non_member_session_denied(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
session_id = await self._session_with(client, workspace, peer)
token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
# One allowlisted session the peer belongs to, one it doesn't:
# membership must hold for *every* entry.
resp = self._chat_as(
client,
workspace,
peer,
token,
{"filters": {"session_id": [session_id, str(generate_nanoid())]}},
)
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_left_session_denied(
self,
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
"""The regression this gate's `active_only` flag exists to prevent.
With the loose membership definition a peer that left a session still
passed here, while the adjacent `session_id` check rejected it.
"""
workspace, peer = sample_data
session_id = await self._session_with(client, workspace, peer)
token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
await crud.remove_peers_from_session(
db_session,
workspace_name=workspace.name,
session_name=session_id,
peer_names={peer.name},
)
await db_session.commit()
resp = self._chat_as(
client, workspace, peer, token, {"filters": {"session_id": [session_id]}}
)
assert resp.status_code == 401
# ...and the single-session gate agrees, which is the whole point.
resp = self._chat_as(client, workspace, peer, token, {"session_id": session_id})
assert resp.status_code == 401
@pytest.mark.asyncio
async def test_workspace_scoped_key_bypasses_gate(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
"""Workspace keys are trusted callers — the allowlist passes as given."""
workspace, peer = sample_data
token = create_jwt(JWTParams(w=workspace.name))
foreign = str(generate_nanoid())
with patch("src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")):
resp = self._chat_as(
client, workspace, peer, token, {"filters": {"session_id": [foreign]}}
)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_empty_allowlist_still_gated(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
"""`filters={"session_id": []}` is a real allowlist, not an absent one.
It must reach the gate (and pass trivially, since the empty set is a
subset of anything) rather than being skipped by a truthiness check.
"""
workspace, peer = sample_data
token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
with patch(
"src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")
) as mock_chat:
resp = self._chat_as(
client, workspace, peer, token, {"filters": {"session_id": []}}
)
assert resp.status_code == 200
assert mock_chat.await_args is not None
assert mock_chat.await_args.kwargs["session_allowlist"] == []
class TestResolveSessionScope:
"""The tri-state contract the four message-crud call sites depend on."""
@pytest.mark.asyncio
async def test_unrestricted_when_no_observer_and_no_allowlist(
self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
):
workspace, _ = sample_data
assert await resolve_session_scope(
db_session, workspace.name, None, None, None
) == (None, False)
@pytest.mark.asyncio
async def test_pinned_session_inside_allowlist_passes_through(
self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
):
workspace, _ = sample_data
# None (not [s1]) — the query filters on session_name directly.
assert await resolve_session_scope(
db_session, workspace.name, "s1", ["s1", "s2"], None
) == (None, False)
@pytest.mark.asyncio
async def test_pinned_session_outside_allowlist_denies(
self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
):
workspace, _ = sample_data
assert await resolve_session_scope(
db_session, workspace.name, "s3", ["s1", "s2"], None
) == (None, True)
@pytest.mark.asyncio
async def test_empty_allowlist_denies_rather_than_returning_empty_list(
self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
):
"""Never returns [] — downstream stores drop an empty IN clause."""
workspace, _ = sample_data
allowed, deny = await resolve_session_scope(
db_session, workspace.name, None, [], None
)
assert (allowed, deny) == (None, True)
@pytest.mark.asyncio
async def test_no_db_touched_when_no_observer_lookup_needed(self):
"""Callers pass db=None on the external-vector-store path.
The helper must not open a session of its own unless it actually needs
an observer lookup, or the external semantic lookup stops being the
first thing that happens (see
tests/integration/test_message_embeddings.py).
"""
with patch("src.crud.message.tracked_db") as mock_tracked_db:
# No observer: pinned session, unrestricted, and plain allowlist.
assert await resolve_session_scope(None, "w", "s1", None, None) == (
None,
False,
)
assert await resolve_session_scope(None, "w", None, None, None) == (
None,
False,
)
assert await resolve_session_scope(None, "w", None, ["s1"], None) == (
["s1"],
False,
)
mock_tracked_db.assert_not_called()
@pytest.mark.asyncio
async def test_observer_scope_intersected_with_allowlist(
self,
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
workspace, peer = sample_data
session_id = str(generate_nanoid())
resp = client.post(
f"/v3/workspaces/{workspace.name}/sessions",
json={"id": session_id, "peer_names": {peer.name: {}}},
)
assert resp.status_code == 201
allowed, deny = await resolve_session_scope(
db_session, workspace.name, None, [session_id], peer.name
)
assert (allowed, deny) == ([session_id], False)
# Allowlisting only a session the observer isn't in denies outright.
allowed, deny = await resolve_session_scope(
db_session, workspace.name, None, [str(generate_nanoid())], peer.name
)
assert (allowed, deny) == (None, True)
class TestChatRouteFilterValidation:
"""Filter validation happens before any LLM work — safe to exercise."""
def _chat(
self,
client: TestClient,
workspace: Workspace,
peer: Peer,
body: dict[str, Any],
):
return client.post(
f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat",
json={"query": "what do you know?", **body},
)
def test_unsupported_filter_key_422(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
resp = self._chat(client, workspace, peer, {"filters": {"peer_id": ["x"]}})
assert resp.status_code == 422
def test_bad_filter_shape_422(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
resp = self._chat(client, workspace, peer, {"filters": {"session_id": 42}})
assert resp.status_code == 422
def test_session_id_not_in_allowlist_422(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
resp = self._chat(
client,
workspace,
peer,
{"session_id": "s-outside", "filters": {"session_id": ["s1", "s2"]}},
)
assert resp.status_code == 422
def test_allowlist_cap_422(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
too_many = [f"s{i}" for i in range(MAX_SESSION_ALLOWLIST_ENTRIES + 1)]
resp = self._chat(
client, workspace, peer, {"filters": {"session_id": too_many}}
)
assert resp.status_code == 422
class TestRepresentationRouteFilters:
@pytest.mark.asyncio
async def test_representation_scoped_by_filters(
self,
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[Workspace, Peer],
):
workspace, peer = sample_data
session_a = models.Session(
name=str(generate_nanoid()), workspace_name=workspace.name
)
session_b = models.Session(
name=str(generate_nanoid()), workspace_name=workspace.name
)
db_session.add_all([session_a, session_b])
await db_session.flush()
collection = models.Collection(
workspace_name=workspace.name,
observer=peer.name,
observed=peer.name,
)
db_session.add(collection)
await db_session.flush()
db_session.add_all(
[
models.Document(
workspace_name=workspace.name,
observer=peer.name,
observed=peer.name,
content="fact from session a",
session_name=session_a.name,
),
models.Document(
workspace_name=workspace.name,
observer=peer.name,
observed=peer.name,
content="fact from session b",
session_name=session_b.name,
),
models.Document(
workspace_name=workspace.name,
observer=peer.name,
observed=peer.name,
content="sessionless dream fact",
session_name=None,
),
]
)
await db_session.commit()
resp = client.post(
f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation",
json={"filters": {"session_id": [session_a.name]}},
)
assert resp.status_code == 200
representation = resp.json()["representation"]
assert "fact from session a" in representation
assert "fact from session b" not in representation
assert "sessionless dream fact" not in representation
def test_session_id_not_in_allowlist_422(
self, client: TestClient, sample_data: tuple[Workspace, Peer]
):
workspace, peer = sample_data
resp = client.post(
f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation",
json={"session_id": "s-out", "filters": {"session_id": ["s-in"]}},
)
assert resp.status_code == 422

65
tests/utils/test_agent_tools.py
View File

@ -39,6 +39,8 @@ from src.utils.agent_tools import (
create_observations,
create_tool_executor,
extract_preferences,
get_observation_context,
get_recent_history,
)
# =============================================================================
@ -801,6 +803,7 @@ class TestSearchMemory:
context_window: int = 2,
embedding: list[float] | None = None,
observer: str | None = None,
**_kwargs: Any,
) -> list[tuple[list[models.Message], list[models.Message]]]:
_ = (workspace_name, session_name, query, limit, context_window, observer)
fallback_embeddings.append(embedding)
@ -909,6 +912,7 @@ class TestSearchMessagesTemporal:
context_window: int = 2,
embedding: list[float] | None = None,
observer: str | None = None,
**_kwargs: Any,
) -> list[tuple[list[models.Message], list[models.Message]]]:
_ = (
workspace_name,
@ -1502,6 +1506,7 @@ class TestExtractPreferences:
context_window: int,
embedding: list[float] | None,
observer: str | None = None,
**_kwargs: Any,
) -> list[tuple[list[models.Message], list[models.Message]]]:
_ = (limit, context_window, observer)
embedding_args.append(embedding)
@ -1848,3 +1853,63 @@ class TestObserverPeerNameWiring:
await _handle_get_messages_by_date_range(ctx, {"after_date": "2024-01-01"})
assert captured_kwargs["observer"] == ctx.observer
@pytest.mark.asyncio
class TestSessionAllowlistFailClosed:
"""A specific session_name outside the session_allowlist allowlist must fail closed.
Routes guard this too, but these CRUD/tool functions are reachable directly
from the dialectic loop, so the allowlist is enforced at the boundary.
"""
async def test_get_recent_history_respects_allowlist(
self, db_session: AsyncSession, tool_test_data: Any
):
workspace, _peer1, peer2, session, _messages, _ = tool_test_data
# session IS in the allowlist -> history returned
allowed = await get_recent_history(
db_session,
workspace_name=workspace.name,
session_name=session.name,
observed=peer2.name,
session_allowlist=[session.name],
)
assert allowed # non-empty
# session is NOT in the allowlist -> fail closed
blocked = await get_recent_history(
db_session,
workspace_name=workspace.name,
session_name=session.name,
observed=peer2.name,
session_allowlist=["some-other-session"],
)
assert blocked == []
async def test_get_observation_context_fails_closed(
self, db_session: AsyncSession, tool_test_data: Any
):
workspace, peer1, _peer2, session, messages, _ = tool_test_data
blocked = await get_observation_context(
db_session,
workspace_name=workspace.name,
session_name=session.name,
message_ids=[messages[0].id],
observer=peer1.name,
session_allowlist=["some-other-session"],
)
assert blocked == []
async def test_get_messages_by_date_range_fails_closed(
self, db_session: AsyncSession, tool_test_data: Any
):
workspace, _peer1, _peer2, session, _messages, _ = tool_test_data
blocked = await crud.get_messages_by_date_range(
db_session,
workspace_name=workspace.name,
session_name=session.name,
session_allowlist=["some-other-session"],
)
assert blocked == []