diff --git a/pyproject.toml b/pyproject.toml
index d7905fc6..586f0c45 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -32,7 +32,7 @@ dependencies = [
     "typing-extensions>=4.11.0",
     "json-repair>=0.49.0",
     "turbopuffer>=1.8.1",
-    "lancedb>=0.25.3",
+    "lancedb>=0.25.3; sys_platform != \"darwin\" or platform_machine != \"x86_64\"",
     "pyarrow>=19.0.0",
     "redis>=7.0.0,<8.0.0",
     "cashews[redis]==7.5.0",
diff --git a/src/cache/client.py b/src/cache/client.py
index 29d8c187..1ad8e3f5 100644
--- a/src/cache/client.py
+++ b/src/cache/client.py
@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 import logging
 from typing import Any, cast
+from urllib.parse import urlparse, urlunparse
 
 import sentry_sdk
 from cashews import cache
@@ -20,10 +21,99 @@ from src.config import settings
 
 logger = logging.getLogger(__name__)
 
-
 _cache_lock = asyncio.Lock()
 
 
+# Query parameters that carry secrets when configured via URL:
+# redis-py accepts ``?password=`` (all querystring options become client
+# kwargs) and cashews accepts ``?secret=`` (HMAC key for value signing).
+_SENSITIVE_QUERY_PARAMS = frozenset({"password", "secret"})
+
+
+def _mask_sensitive_query(query: str) -> str:
+    """Mask values of secret-bearing query parameters.
+
+    Operates on the raw query string (no decode/re-encode round trip)
+    so non-secret parameters are preserved byte-for-byte.
+
+    Args:
+        query: The raw query string from a parsed URL.
+
+    Returns:
+        The query string with sensitive values replaced by ``***``, or
+        the original string if no sensitive parameter is present.
+    """
+    if not query:
+        return query
+    parts: list[str] = []
+    changed = False
+    for part in query.split("&"):
+        name, sep, _value = part.partition("=")
+        if sep and name.lower() in _SENSITIVE_QUERY_PARAMS:
+            parts.append(f"{name}=***")
+            changed = True
+        else:
+            parts.append(part)
+    return "&".join(parts) if changed else query
+
+
+def _redact_cache_url(url: str) -> str:
+    """Mask credentials in a Redis connection URL before logging.
+
+    Given ``redis://:password@host:port/db`` returns
+    ``redis://:***@host:port/db``; secret-bearing query parameters
+    (``?password=``, ``?secret=``) are masked as well.  A URL carrying
+    no credentials is returned unchanged.  This function never raises
+    and never returns a credential: an invalid port is omitted from
+    the output, and a URL that cannot be parsed at all is replaced by
+    a generic placeholder rather than echoed back, so that logging
+    inside ``except`` blocks can neither crash startup nor leak the
+    secrets this helper exists to hide.
+
+    Args:
+        url: The Redis connection URL to redact.
+
+    Returns:
+        The URL with its credentials masked, the original URL if it
+        carries none, or ``"<redacted-unparseable-url>"`` if parsing
+        fails entirely.
+    """
+    try:
+        parsed = urlparse(url)
+        query = _mask_sensitive_query(parsed.query)
+        # .password only splits netloc and never raises, unlike .port
+        if parsed.password is None and query == parsed.query:
+            # A string with an "@" but no parsed authority (e.g. a URL
+            # missing its scheme, ":pass@host:6379/0") may still carry
+            # userinfo that urlparse could not see — never echo it.
+            if "@" in url and not parsed.netloc:
+                return "<redacted-unparseable-url>"
+            return url
+        netloc = parsed.netloc
+        if parsed.password is not None:
+            userinfo = parsed.username or ""
+            hostname = parsed.hostname or ""
+            # Preserve IPv6 brackets (urlparse strips them from .hostname)
+            if hostname and ":" in hostname and not hostname.startswith("["):
+                hostname = f"[{hostname}]"
+            netloc = f"{userinfo}:***@{hostname}"
+            try:
+                port = parsed.port
+            except ValueError:
+                # Invalid or out-of-range port: omit it rather than let
+                # the outer fallback echo the raw URL (and its password)
+                # back.
+                port = None
+            if port is not None:
+                netloc += f":{port}"
+        parsed = parsed._replace(netloc=netloc, query=query)
+        return urlunparse(parsed)
+    except (ValueError, TypeError):
+        # Unparseable URL: never return the raw input — it may contain
+        # the very password this helper exists to hide.
+        return "<redacted-unparseable-url>"
+
+
 def is_cache_enabled() -> bool:
     return settings.CACHE.ENABLED
 
@@ -59,7 +149,7 @@ async def init_cache() -> None:
         except Exception as setup_err:
             logger.warning(
                 "Cache setup failed for %s: %s. Falling back to in-memory cache",
-                settings.CACHE.URL,
+                _redact_cache_url(settings.CACHE.URL),
                 setup_err,
             )
             if settings.SENTRY.ENABLED:
@@ -87,7 +177,10 @@ async def init_cache() -> None:
                 with attempt:
                     async with asyncio.timeout(2):
                         await cache.ping()
-                        logger.info("Connected to cache at %s", settings.CACHE.URL)
+                        logger.info(
+                            "Connected to cache at %s",
+                            _redact_cache_url(settings.CACHE.URL),
+                        )
         except (
             redis_exc.TimeoutError,
             redis_exc.ConnectionError,
@@ -96,7 +189,7 @@ async def init_cache() -> None:
         ) as e:
             logger.warning(
                 "Failed to connect to cache at %s: %s. Falling back to in-memory cache",
-                settings.CACHE.URL,
+                _redact_cache_url(settings.CACHE.URL),
                 e,
             )
             if settings.SENTRY.ENABLED:
@@ -107,7 +200,7 @@ async def init_cache() -> None:
         except Exception as e:
             logger.warning(
                 "Unexpected cache error at %s: %s. Falling back to in-memory cache",
-                settings.CACHE.URL,
+                _redact_cache_url(settings.CACHE.URL),
                 e,
             )
             if settings.SENTRY.ENABLED:
diff --git a/src/crud/message.py b/src/crud/message.py
index e66b8f49..c53be851 100644
--- a/src/crud/message.py
+++ b/src/crud/message.py
@@ -56,11 +56,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)
@@ -68,10 +85,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]]:
@@ -699,21 +786,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(
@@ -768,6 +863,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.
@@ -778,12 +874,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)
@@ -809,6 +912,7 @@ async def search_messages(
         context_window=context_window,
         operation_name="message.search_messages",
         observer=observer,
+        session_allowlist=session_allowlist,
     )
 
 
@@ -856,6 +960,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).
@@ -866,25 +971,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,
@@ -908,6 +1017,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.
@@ -916,24 +1026,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)
 
@@ -967,6 +1081,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.
@@ -977,6 +1092,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
@@ -985,6 +1104,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)
@@ -1011,4 +1133,5 @@ async def search_messages_temporal(
         context_window=context_window,
         operation_name="message.search_messages_temporal",
         observer=observer,
+        session_allowlist=session_allowlist,
     )
diff --git a/src/crud/representation.py b/src/crud/representation.py
index 3061c2d9..d4b86ffb 100644
--- a/src/crud/representation.py
+++ b/src/crud/representation.py
@@ -19,9 +19,11 @@ from src.telemetry.events import EmbeddingCallPurpose
 from src.telemetry.logging import accumulate_metric
 from src.utils.formatting import format_datetime_utc
 from src.utils.representation import (
+    ALLOWLIST_SAFE_LEVELS,
     DeductiveObservation,
     ExplicitObservation,
     Representation,
+    allowlist_safe_levels,
 )
 from src.utils.types import embedding_call_purpose
 
@@ -213,7 +215,7 @@ class RepresentationManager:
         self,
         *,
         db: AsyncSession | None = None,
-        session_name: 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,10 @@ class RepresentationManager:
         Args:
             db: Optional database session. If provided, uses it directly;
                 otherwise creates a new session via tracked_db.
-            session_name: Optional session to filter by
+            session_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.
             include_semantic_query: Query for semantic search
             embedding: Pre-computed embedding for the semantic query.
             semantic_search_top_k: Number of semantic results
@@ -267,7 +272,7 @@ class RepresentationManager:
         if db is not None:
             return await self._get_working_representation_internal(
                 db,
-                session_name=session_name,
+                session_allowlist=session_allowlist,
                 include_semantic_query=include_semantic_query,
                 embedding=embedding,
                 semantic_search_top_k=semantic_search_top_k,
@@ -281,7 +286,7 @@ class RepresentationManager:
         ) as new_db:
             return await self._get_working_representation_internal(
                 new_db,
-                session_name=session_name,
+                session_allowlist=session_allowlist,
                 include_semantic_query=include_semantic_query,
                 embedding=embedding,
                 semantic_search_top_k=semantic_search_top_k,
@@ -296,7 +301,7 @@ class RepresentationManager:
         self,
         db: AsyncSession,
         *,
-        session_name: 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,
@@ -305,6 +310,12 @@ class RepresentationManager:
         max_observations: int = settings.DERIVER.WORKING_REPRESENTATION_MAX_OBSERVATIONS,
     ) -> Representation:
         """Internal implementation of get_working_representation."""
+        # Fail closed on an empty allowlist. This must short-circuit before
+        # any query: downstream stores drop an `IN ()` clause with an empty
+        # list (lancedb), which would silently widen the scope instead.
+        if session_allowlist is not None and not session_allowlist:
+            return Representation()
+
         total = max_observations
 
         # Calculate how many observations to get from each source
@@ -345,6 +356,7 @@ class RepresentationManager:
                 top_k=semantic_observations,
                 max_distance=semantic_search_max_distance,
                 embedding=embedding,
+                session_allowlist=session_allowlist,
             )
             representation.merge_representation(
                 Representation.from_documents(semantic_docs)
@@ -353,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
+                db, top_k=top_observations, session_allowlist=session_allowlist
             )
             representation.merge_representation(
                 Representation.from_documents(derived_docs)
@@ -361,7 +373,7 @@ class RepresentationManager:
 
         # Get recent observations
         recent_docs = await self._query_documents_recent(
-            db, top_k=recent_observations, session_name=session_name
+            db, top_k=recent_observations, session_allowlist=session_allowlist
         )
 
         representation.merge_representation(Representation.from_documents(recent_docs))
@@ -376,6 +388,7 @@ class RepresentationManager:
         max_distance: float | None = None,
         level: str | None = None,
         embedding: list[float] | None = None,
+        session_allowlist: list[str] | None = None,
     ) -> list[models.Document]:
         """Query documents by semantic similarity."""
         try:
@@ -387,6 +400,7 @@ class RepresentationManager:
                     top_k,
                     max_distance,
                     embedding=embedding,
+                    session_allowlist=session_allowlist,
                 )
             else:
                 documents = await crud.query_documents(
@@ -398,6 +412,10 @@ class RepresentationManager:
                     max_distance=max_distance,
                     top_k=top_k,
                     embedding=embedding,
+                    filters=self._build_filter_conditions(
+                        session_allowlist=session_allowlist
+                    )
+                    or None,
                 )
                 db.expunge_all()
                 return list(documents)
@@ -407,7 +425,7 @@ class RepresentationManager:
             return []
 
     async def _query_documents_recent(
-        self, db: AsyncSession, top_k: int, session_name: str | None = None
+        self, db: AsyncSession, top_k: int, session_allowlist: list[str] | None = None
     ) -> list[models.Document]:
         """Query most recent documents."""
         stmt = (
@@ -419,8 +437,13 @@ class RepresentationManager:
                 models.Document.observed == self.observed,
                 models.Document.deleted_at.is_(None),
                 *(
-                    [models.Document.session_name == session_name]
-                    if session_name is not None
+                    [
+                        models.Document.session_name.in_(session_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 []
                 ),
             )
@@ -433,7 +456,7 @@ class RepresentationManager:
         return list(documents)
 
     async def _query_documents_most_derived(
-        self, db: AsyncSession, top_k: int
+        self, db: AsyncSession, top_k: int, session_allowlist: list[str] | None = None
     ) -> list[models.Document]:
         """Query most derived documents."""
         stmt = (
@@ -444,6 +467,16 @@ class RepresentationManager:
                 models.Document.observer == self.observer,
                 models.Document.observed == self.observed,
                 models.Document.deleted_at.is_(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 []
+                ),
             )
             .order_by(
                 models.Document.times_derived.desc(),
@@ -480,6 +513,7 @@ class RepresentationManager:
         count: int,
         max_distance: float | None = None,
         embedding: list[float] | None = None,
+        session_allowlist: list[str] | None = None,
     ) -> list[models.Document]:
         """Query documents for a specific level."""
         documents = await crud.query_documents(
@@ -490,7 +524,9 @@ class RepresentationManager:
             query=query,
             max_distance=max_distance,
             top_k=count,
-            filters=self._build_filter_conditions(level),
+            filters=self._build_filter_conditions(
+                level, session_allowlist=session_allowlist
+            ),
             embedding=embedding,
         )
 
@@ -503,17 +539,32 @@ class RepresentationManager:
     def _build_filter_conditions(
         self,
         level: 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_allowlist list — empty allowlists
+        fail closed before any query is issued (see
+        _get_working_representation_internal).
         """
         filters: dict[str, Any] = {}
 
         if level:
             filters["level"] = level
 
+        # `is not None` (not truthiness): an explicit empty allowlist must emit
+        # an empty `in` so downstream stores fail closed, matching
+        # _query_documents_recent / _query_documents_most_derived. Truthiness
+        # here would silently drop the filter and widen scope.
+        if session_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
 
 
@@ -526,7 +577,7 @@ async def get_working_representation(
     db: AsyncSession | None = None,
     observer: str,
     observed: str,
-    session_name: 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,
@@ -559,7 +610,7 @@ async def get_working_representation(
     )
     return await manager.get_working_representation(
         db=db,
-        session_name=session_name,
+        session_allowlist=session_allowlist,
         include_semantic_query=include_semantic_query,
         embedding=embedding,
         semantic_search_top_k=semantic_search_top_k,
diff --git a/src/dialectic/chat.py b/src/dialectic/chat.py
index 40d991b4..d2f7e741 100644
--- a/src/dialectic/chat.py
+++ b/src/dialectic/chat.py
@@ -26,6 +26,7 @@ async def agentic_chat(
     observer: str,
     observed: str,
     reasoning_level: ReasoningLevel = "low",
+    session_allowlist: list[str] | None = None,
     response_model: type[BaseModel] | None = None,
 ) -> str:
     """
@@ -38,6 +39,7 @@ async def agentic_chat(
         observer: The peer making the query
         observed: The peer being queried about
         reasoning_level: Level of reasoning to apply
+        session_allowlist: Optional session allowlist restricting all recall
         response_model: Optional Pydantic model the answer must conform to.
             When set, the returned string is JSON matching the model's schema.
 
@@ -82,6 +84,7 @@ async def agentic_chat(
         observer_peer_card=observer_peer_card,
         observed_peer_card=observed_peer_card,
         reasoning_level=reasoning_level,
+        session_allowlist=session_allowlist,
     )
 
     return await agent.answer(query, response_model=response_model)
@@ -94,6 +97,7 @@ async def agentic_chat_stream(
     observer: str,
     observed: str,
     reasoning_level: ReasoningLevel = "low",
+    session_allowlist: list[str] | None = None,
     response_model: type[BaseModel] | None = None,
 ) -> AsyncIterator[str]:
     """
@@ -106,6 +110,7 @@ async def agentic_chat_stream(
         observer: The peer making the query
         observed: The peer being queried about
         reasoning_level: Level of reasoning to apply
+        session_allowlist: Optional session allowlist restricting all recall
         response_model: Optional Pydantic model the answer must conform to.
             When set, the streamed text accumulates to JSON matching the
             model's schema.
@@ -151,6 +156,7 @@ async def agentic_chat_stream(
         observer_peer_card=observer_peer_card,
         observed_peer_card=observed_peer_card,
         reasoning_level=reasoning_level,
+        session_allowlist=session_allowlist,
     )
 
     async for chunk in agent.answer_stream(query, response_model=response_model):
diff --git a/src/dialectic/core.py b/src/dialectic/core.py
index 757423e0..9580b94c 100644
--- a/src/dialectic/core.py
+++ b/src/dialectic/core.py
@@ -70,6 +70,7 @@ class DialecticAgent:
         metric_key: str | None = None,
         reasoning_level: ReasoningLevel = "low",
         session_id: str | None = None,
+        session_allowlist: list[str] | None = None,
     ):
         """
         Initialize the dialectic agent.
@@ -84,9 +85,13 @@ class DialecticAgent:
             metric_key: Optional key for logging metrics (if provided, agent won't log separately)
             reasoning_level: Level of reasoning to apply
             session_id: ID used for grouping traces (not session_name)
+            session_allowlist: Optional session allowlist restricting all recall
+                (conclusions and messages) to these sessions; empty list
+                fails closed
         """
         self.workspace_name: str = workspace_name
         self.session_name: str | None = session_name
+        self.session_allowlist: list[str] | None = session_allowlist
         self.session_id: str | None = session_id
         self.observer: str = observer
         self.observed: str = observed
@@ -108,6 +113,24 @@ class DialecticAgent:
         self._prefetched_conclusion_count: int = 0
         self._run_id: str = generate_nanoid()  # Always generate for event correlation
 
+    def _select_tools(self) -> list[dict[str, Any]]:
+        """Pick the toolset for this query.
+
+        Minimal reasoning uses a reduced set to reduce cost. Under a session
+        allowlist `get_reasoning_chain` is dropped entirely rather than left in
+        to fail at call time: chains traverse provenance across sessions, so it
+        can't be scoped, and offering it costs both the schema in context and a
+        wasted turn when the model tries it.
+        """
+        tools = (
+            DIALECTIC_TOOLS_MINIMAL
+            if self.reasoning_level == "minimal"
+            else DIALECTIC_TOOLS
+        )
+        if self.session_allowlist is not None:
+            tools = [t for t in tools if t.get("name") != "get_reasoning_chain"]
+        return tools
+
     async def _initialize_session_history(self) -> None:
         """Fetch and inject session history into the system prompt if configured."""
         if self._session_history_initialized:
@@ -197,6 +220,7 @@ class DialecticAgent:
                 limit=prefetch_limit,
                 levels=["explicit"],
                 embedding=query_embedding,
+                session_allowlist=self.session_allowlist,
             )
 
             derived_repr = await search_memory(
@@ -207,6 +231,7 @@ class DialecticAgent:
                 limit=prefetch_limit,
                 levels=["deductive", "inductive", "contradiction"],
                 embedding=query_embedding,
+                session_allowlist=self.session_allowlist,
             )
 
             if explicit_repr.is_empty() and derived_repr.is_empty():
@@ -296,6 +321,7 @@ class DialecticAgent:
         ] = await create_tool_executor(
             workspace_name=self.workspace_name,
             session_name=self.session_name,
+            session_allowlist=self.session_allowlist,
             observer=self.observer,
             observed=self.observed,
             history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT,
@@ -438,12 +464,7 @@ class DialecticAgent:
         # Get level-specific settings
         level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level]
 
-        # Use minimal tools for minimal reasoning to reduce cost
-        tools = (
-            DIALECTIC_TOOLS_MINIMAL
-            if self.reasoning_level == "minimal"
-            else DIALECTIC_TOOLS
-        )
+        tools = self._select_tools()
         # Use level-specific max_output_tokens if set, otherwise global default
         max_tokens = (
             level_settings.MAX_OUTPUT_TOKENS
@@ -520,12 +541,7 @@ class DialecticAgent:
         # Get level-specific settings
         level_settings = settings.DIALECTIC.LEVELS[self.reasoning_level]
 
-        # Use minimal tools for minimal reasoning to reduce cost
-        tools = (
-            DIALECTIC_TOOLS_MINIMAL
-            if self.reasoning_level == "minimal"
-            else DIALECTIC_TOOLS
-        )
+        tools = self._select_tools()
         # Use level-specific max_output_tokens if set, otherwise global default
         max_tokens = (
             level_settings.MAX_OUTPUT_TOKENS
diff --git a/src/main.py b/src/main.py
index 29d641b4..75eac455 100644
--- a/src/main.py
+++ b/src/main.py
@@ -4,15 +4,12 @@ import time
 import uuid
 from collections.abc import Awaitable, Callable
 from contextlib import asynccontextmanager
-from typing import TYPE_CHECKING
 
 import sentry_sdk
 from fastapi import FastAPI, Request, Response
-from fastapi.exceptions import RequestValidationError
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi_pagination import add_pagination
-from pydantic import ValidationError
 from sentry_sdk.integrations.fastapi import FastApiIntegration
 from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration
 from sentry_sdk.integrations.starlette import StarletteIntegration
@@ -43,9 +40,6 @@ from src.telemetry import (
 from src.telemetry.logging import get_route_template
 from src.telemetry.sentry import initialize_sentry
 
-if TYPE_CHECKING:
-    from sentry_sdk._types import Event, Hint
-
 
 def get_log_level() -> int:
     """
@@ -89,30 +83,10 @@ class MetricsAccessFilter(logging.Filter):
 logging.getLogger("uvicorn.access").addFilter(MetricsAccessFilter())
 
 
-def before_send(event: "Event", hint: "Hint | None") -> "Event | None":
-    """Filter out events raised from known non-actionable exceptions before Sentry sees them."""
-    if not hint:
-        return event
-
-    exc_info = hint.get("exc_info")
-    if not exc_info:
-        return event
-
-    _, exc_value, _ = exc_info
-    if isinstance(exc_value, HonchoException):
-        return None
-
-    # Filters out ValidationErrors and RequestValidationErrors (typically coming from Pydantic)
-    if isinstance(exc_value, ValidationError | RequestValidationError):
-        logger.info(f"Filtering out validation error from Sentry: {exc_value}")
-        return None
-
-    return event
-
-
 # Sentry Setup
 SENTRY_ENABLED = settings.SENTRY.ENABLED
 if SENTRY_ENABLED:
+    # before_send defaults to sentry.default_before_send (shared with the deriver).
     initialize_sentry(
         integrations=[
             StarletteIntegration(
@@ -124,7 +98,6 @@ if SENTRY_ENABLED:
             # Explicit so DB-query spans are not reliant on auto-enabling.
             SqlalchemyIntegration(),
         ],
-        before_send=before_send,
     )
 
 
diff --git a/src/routers/peers.py b/src/routers/peers.py
index 0b17eef8..0de468df 100644
--- a/src/routers/peers.py
+++ b/src/routers/peers.py
@@ -15,6 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from src import crud, schemas
 from src.config import settings
+from src.crud.message import get_peer_session_names
 from src.crud.session import is_peer_in_session
 from src.dependencies import db, read_db, tracked_db
 from src.dialectic.chat import agentic_chat, agentic_chat_stream
@@ -27,6 +28,7 @@ from src.exceptions import (
 from src.security import JWTParams, require_auth
 from src.telemetry import prometheus_metrics
 from src.telemetry.events import EmbeddingCallPurpose, GetContextEvent, emit
+from src.utils.filter import extract_session_allowlist
 from src.utils.schema_conversion import json_response_schema_to_pydantic
 from src.utils.scopes import is_scope_peer_name, validate_no_scope_peer_names
 from src.utils.search import search
@@ -220,6 +222,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:
@@ -265,6 +286,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,
                 )
             ),
@@ -280,6 +302,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,
     )
 
@@ -321,6 +344,12 @@ async def get_representation(
             "Scope peers cannot be a representation target: no representation is formed of a scope."
         )
 
+    # 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:
@@ -339,7 +368,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_allowlist=[options.session_id]
+            if options.session_id is not None
+            else session_allowlist,
             include_semantic_query=options.search_query,
             embedding=embedding,
             semantic_search_top_k=options.search_top_k,
@@ -502,7 +533,7 @@ async def get_peer_context(
             workspace_id,
             observer=peer_id,
             observed=observed,
-            session_name=None,  # Peer context is global, not session-scoped
+            session_allowlist=None,  # Peer context is global, not session-scoped
             include_semantic_query=search_query,
             embedding=embedding,
             semantic_search_top_k=search_top_k,
diff --git a/src/routers/sessions.py b/src/routers/sessions.py
index f7e57b0e..e59bbf76 100644
--- a/src/routers/sessions.py
+++ b/src/routers/sessions.py
@@ -53,7 +53,7 @@ async def _get_working_representation_task(
     *,
     observer: str,
     observed: str,
-    session_name: str | None,
+    session_allowlist: list[str] | None,
     search_top_k: int | None,
     search_max_distance: float | None,
     include_most_derived: bool,
@@ -69,7 +69,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_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
@@ -84,7 +84,7 @@ async def _get_working_representation_task(
         db=db,
         observer=observer,
         observed=observed,
-        session_name=session_name,
+        session_allowlist=session_allowlist,
         include_semantic_query=last_message,
         semantic_search_top_k=search_top_k,
         semantic_search_max_distance=search_max_distance,
@@ -793,7 +793,7 @@ async def get_session_context(
         search_query,
         observer=observer,
         observed=observed,
-        session_name=session_id if limit_to_session else None,
+        session_allowlist=[session_id] if limit_to_session else None,
         search_top_k=search_top_k,
         search_max_distance=search_max_distance,
         include_most_derived=include_most_frequent,
diff --git a/src/schemas/api.py b/src/schemas/api.py
index 73d2ef7e..d14e6d57 100644
--- a/src/schemas/api.py
+++ b/src/schemas/api.py
@@ -211,6 +211,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",
@@ -676,6 +685,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/telemetry/sentry.py b/src/telemetry/sentry.py
index d84837ed..7b17e156 100644
--- a/src/telemetry/sentry.py
+++ b/src/telemetry/sentry.py
@@ -9,19 +9,58 @@ from functools import wraps
 from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast
 
 import sentry_sdk
+from fastapi.exceptions import RequestValidationError
+from pydantic import ValidationError
+from sqlalchemy.exc import OperationalError
 
 from src.config import settings
+from src.exceptions import HonchoException
 
 P = ParamSpec("P")
 T = TypeVar("T")
 
 if TYPE_CHECKING:
-    from sentry_sdk._types import EventProcessor
+    from sentry_sdk._types import Event, EventProcessor, Hint
     from sentry_sdk.integrations import Integration
 
 logger = logging.getLogger(__name__)
 
 
+def default_before_send(event: Event, hint: Hint | None) -> Event | None:
+    """Filter/regroup known non-actionable events before Sentry ingests them.
+
+    Shared by every entrypoint (API + deriver) so filtering is process-agnostic.
+    """
+    if not hint:
+        return event
+
+    exc_info = hint.get("exc_info")
+    if not exc_info:
+        return event
+
+    _, exc_value, _ = exc_info
+    if isinstance(exc_value, HonchoException):
+        return None
+
+    # Filters out ValidationErrors and RequestValidationErrors (typically from Pydantic)
+    if isinstance(exc_value, ValidationError | RequestValidationError):
+        logger.info(f"Filtering out validation error from Sentry: {exc_value}")
+        return None
+
+    # DB connection-pool checkout timeouts are a fleet-wide saturation symptom, not a
+    # per-transaction bug. Collapse every occurrence into one issue (Sentry would otherwise
+    # split by transaction/endpoint) and drop to warning so it stops tripping error alerts.
+    # Watch it via a rate/spike metric alert instead. Root cause tracked in DEV-1852.
+    if isinstance(exc_value, OperationalError) and "connection timeout expired" in str(
+        exc_value
+    ):
+        event["fingerprint"] = ["honcho-db-connection-timeout"]
+        event["level"] = "warning"
+        return event
+
+    return event
+
+
 # Paths whose transactions carry no debugging value but are hit constantly
 # (health checks, Prometheus scrapes, OpenAPI schema, docs). Tracing them at the
 # same rate as real traffic drowns the signal and burns tracing/profiling quota.
@@ -81,13 +120,14 @@ def traces_sampler(sampling_context: dict[str, Any]) -> float:
 def initialize_sentry(
     *,
     integrations: Sequence[Integration],
-    before_send: EventProcessor | None = None,
+    before_send: EventProcessor | None = default_before_send,
 ) -> None:
     """Initialize Sentry SDK with project settings.
 
     Args:
         integrations: Sentry SDK integrations to enable (e.g., Starlette, FastAPI).
-        before_send: Optional event filter callback to suppress specific exceptions.
+        before_send: Event filter override. Defaults to ``default_before_send`` so
+            every entrypoint gets the shared filters; pass ``None`` to opt out.
     """
     sentry_sdk.init(
         dsn=settings.SENTRY.DSN,
diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py
index c5a99207..9f214009 100644
--- a/src/utils/agent_tools.py
+++ b/src/utils/agent_tools.py
@@ -1,7 +1,7 @@
 import asyncio
 import logging
 import weakref
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any, cast
@@ -30,7 +30,11 @@ from src.utils.formatting import (
     parse_datetime_iso,
     utc_now_iso,
 )
-from src.utils.representation import Representation
+from src.utils.representation import (
+    ALLOWLIST_SAFE_LEVELS,
+    Representation,
+    allowlist_safe_levels,
+)
 from src.utils.types import ToolResult, embedding_call_purpose, get_current_iteration
 
 logger = logging.getLogger(__name__)
@@ -1030,6 +1034,7 @@ async def get_recent_history(
     session_name: str | None,
     observed: str | None = None,
     token_limit: int = 8192,
+    session_allowlist: list[str] | None = None,
 ) -> list[models.Message]:
     """
     Retrieve recent conversation history.
@@ -1042,6 +1047,10 @@ async def get_recent_history(
         db: Database session
         workspace_name: Workspace identifier
         session_name: Session identifier (optional)
+            Deprecated for *scoping*: prefer session_allowlist, which
+            intersects with observer membership. This parameter also pins
+            the query to one session and bypasses observer scoping, so it
+            is not a drop-in equivalent and is not removed.
         observed: Peer name to filter by when no session specified (optional)
         token_limit: Maximum tokens to retrieve (default: 8192)
 
@@ -1049,6 +1058,9 @@ async def get_recent_history(
         List of messages in chronological order
     """
     if session_name:
+        # Fail closed: a specific session outside the allowlist is not readable.
+        if session_allowlist is not None and session_name not in session_allowlist:
+            return []
         # Get messages from a specific session
         messages_stmt = await crud.get_messages(
             workspace_name=workspace_name,
@@ -1061,7 +1073,11 @@ async def get_recent_history(
         # Return in chronological order
         return list(reversed(messages))
     elif observed:
+        # Fail closed on an empty allowlist
+        if session_allowlist is not None and not session_allowlist:
+            return []
         # Get recent messages from the observed peer across all sessions
+        # (restricted to the session allowlist when one is provided)
         stmt = (
             select(models.Message)
             .where(models.Message.workspace_name == workspace_name)
@@ -1069,6 +1085,8 @@ async def get_recent_history(
             .order_by(models.Message.created_at.desc())
             .limit(50)  # Limit to recent messages
         )
+        if session_allowlist is not None:
+            stmt = stmt.where(models.Message.session_name.in_(session_allowlist))
         result = await db.execute(stmt)
         messages = list(result.scalars().all())
         # Return in chronological order
@@ -1086,6 +1104,7 @@ async def search_memory(
     limit: int,
     levels: list[str] | None = None,
     embedding: list[float] | None = None,
+    session_allowlist: list[str] | None = None,
 ) -> Representation:
     """
     Search for observations in memory using semantic similarity.
@@ -1106,10 +1125,22 @@ async def search_memory(
     Returns:
         Representation object containing relevant observations
     """
-    # Build filter for levels if specified
-    filters: dict[str, Any] | None = None
+    # Fail closed on an empty allowlist — downstream stores drop empty IN
+    # clauses, which would silently widen scope.
+    if session_allowlist is not None and not session_allowlist:
+        return Representation()
+
+    if session_allowlist is not None:
+        levels = allowlist_safe_levels(levels)
+        if not levels:
+            return Representation()
+
+    # Build filters for levels / session allowlist if specified
+    filters: dict[str, Any] = {}
     if levels:
-        filters = {"level": {"in": levels}}
+        filters["level"] = {"in": levels}
+    if session_allowlist is not None:
+        filters["session_name"] = {"in": session_allowlist}
 
     documents = await crud.query_documents(
         db=None,
@@ -1118,7 +1149,7 @@ async def search_memory(
         observed=observed,
         query=query,
         top_k=limit,
-        filters=filters,
+        filters=filters or None,
         embedding=embedding,
     )
 
@@ -1131,6 +1162,7 @@ async def get_observation_context(
     session_name: str | None,
     message_ids: list[str],
     observer: str | None = None,
+    session_allowlist: list[str] | None = None,
 ) -> list[models.Message]:
     """
     Retrieve messages for given message IDs along with surrounding context.
@@ -1143,9 +1175,16 @@ async def get_observation_context(
         db: Database session
         workspace_name: Workspace identifier
         session_name: Session identifier (optional)
+            Deprecated for *scoping*: prefer session_allowlist, which
+            intersects with observer membership. This parameter also pins
+            the query to one session and bypasses observer scoping, so it
+            is not a drop-in equivalent and is not removed.
         message_ids: List of message IDs to retrieve
         observer: When provided and session_name is None, scope results
             to sessions this peer belongs to
+        session_allowlist: Optional session allowlist. None is unrestricted; an
+            empty list fails closed (empty result); a populated list is
+            intersected with the observer's session scope when observer is set
 
     Returns:
         List of messages in chronological order, including the requested messages and surrounding context
@@ -1153,16 +1192,13 @@ async def get_observation_context(
     if not message_ids:
         return []
 
-    # Pre-fetch peer session scope if needed
-    allowed_session_names: list[str] | None = None
-    if observer and not session_name:
-        from src.crud.message import get_peer_session_names
+    from src.crud.message import resolve_session_scope
 
-        allowed_session_names = await get_peer_session_names(
-            db, workspace_name, observer
-        )
-        if not allowed_session_names:
-            return []
+    allowed_session_names, deny = await resolve_session_scope(
+        db, workspace_name, session_name, session_allowlist, observer
+    )
+    if deny:
+        return []
 
     # Use a CTE to get seq_in_session values for target messages
     stmt = (
@@ -1221,9 +1257,16 @@ async def extract_preferences(
     Args:
         workspace_name: Workspace identifier
         session_name: Session identifier (optional)
+            Deprecated for *scoping*: prefer session_allowlist, which
+            intersects with observer membership. This parameter also pins
+            the query to one session and bypasses observer scoping, so it
+            is not a drop-in equivalent and is not removed.
         observed: The peer whose preferences to extract
         observer: When provided and session_name is None, scope results
             to sessions this peer belongs to
+        session_allowlist: Optional session allowlist. None is unrestricted; an
+            empty list fails closed (empty result); a populated list is
+            intersected with the observer's session scope when observer is set
 
     Returns:
         Dict with 'messages' list containing potentially relevant messages
@@ -1303,6 +1346,10 @@ class ToolContext:
     db_lock: asyncio.Lock
     # Optional resolved configuration for checking feature flags
     configuration: ResolvedConfiguration | None = None
+    # Optional session allowlist (dialectic filters). When set, message and
+    # conclusion recall is restricted to these sessions (intersected with
+    # observer membership); empty list fails closed.
+    session_allowlist: list[str] | None = None
     # Telemetry context fields
     run_id: str | None = None
     agent_type: str | None = None  # "dialectic", "deriver", "dreamer"
@@ -1724,6 +1771,7 @@ async def _handle_get_recent_history(
             session_name=ctx.session_name,
             observed=ctx.observed,
             token_limit=ctx.history_token_limit,
+            session_allowlist=ctx.session_allowlist,
         )
         if not history:
             return "No conversation history available"
@@ -1769,15 +1817,28 @@ async def _handle_search_memory(
         "query_tokens": _estimate_tokens_safe(query),
     }
 
-    documents = await crud.query_documents(
-        db=None,
-        workspace_name=ctx.workspace_name,
-        observer=ctx.observer,
-        observed=ctx.observed,
-        query=query,
-        top_k=top_k,
-        embedding=query_embedding,
-    )
+    # Restrict conclusion recall to the session allowlist when one is set.
+    # Empty allowlist fails closed (downstream stores drop empty IN clauses),
+    # and only levels with a trustworthy session stamp are served.
+    documents: Sequence[models.Document]
+    if ctx.session_allowlist is not None and not ctx.session_allowlist:
+        documents = []
+    else:
+        documents = await crud.query_documents(
+            db=None,
+            workspace_name=ctx.workspace_name,
+            observer=ctx.observer,
+            observed=ctx.observed,
+            query=query,
+            top_k=top_k,
+            embedding=query_embedding,
+            filters={
+                "session_name": {"in": ctx.session_allowlist},
+                "level": {"in": list(ALLOWLIST_SAFE_LEVELS)},
+            }
+            if ctx.session_allowlist is not None
+            else None,
+        )
     mem = Representation.from_documents(documents)
     total_count = mem.len()
     if total_count == 0:
@@ -1799,6 +1860,7 @@ async def _handle_search_memory(
                 context_window=0,
                 embedding=query_embedding,
                 observer=ctx.observer,
+                session_allowlist=ctx.session_allowlist,
             )
             if snippets:
                 message_output = _format_message_snippets(
@@ -1840,6 +1902,7 @@ async def _handle_get_observation_context(
             session_name=ctx.session_name,
             message_ids=tool_input["message_ids"],
             observer=ctx.observer,
+            session_allowlist=ctx.session_allowlist,
         )
         if not messages:
             return f"No messages found for IDs {tool_input['message_ids']}"
@@ -1882,6 +1945,7 @@ async def _handle_search_messages(
         context_window=2,
         embedding=query_embedding,
         observer=ctx.observer,
+        session_allowlist=ctx.session_allowlist,
     )
     search_meta: dict[str, Any] = {
         "top_k": limit,
@@ -1918,6 +1982,7 @@ async def _handle_grep_messages(
         limit=limit,
         context_window=context_window,
         observer=ctx.observer,
+        session_allowlist=ctx.session_allowlist,
     )
     if not snippets:
         return f"No messages found containing '{text}'"
@@ -1982,6 +2047,7 @@ async def _handle_get_messages_by_date_range(
             limit=limit,
             order=order,
             observer=ctx.observer,
+            session_allowlist=ctx.session_allowlist,
         )
         msg_count = len(messages)
         messages_text = (
@@ -2054,6 +2120,7 @@ async def _handle_search_messages_temporal(
         before_date=before_date,
         limit=limit,
         context_window=context_window,
+        session_allowlist=ctx.session_allowlist,
         embedding=query_embedding,
         observer=ctx.observer,
     )
@@ -2310,6 +2377,14 @@ async def _handle_get_reasoning_chain(
     ctx: ToolContext, tool_input: dict[str, Any]
 ) -> str:
     """Handle get_reasoning_chain tool."""
+    # Reasoning chains traverse provenance across sessions by design, so a
+    # session allowlist cannot be enforced on the traversal without exposing
+    # out-of-scope premises/conclusions. Fail closed rather than leak.
+    if ctx.session_allowlist is not None:
+        return (
+            "Reasoning-chain traversal is unavailable for session-scoped "
+            "queries. Use search_memory and message tools instead."
+        )
     observation_id = tool_input.get("observation_id")
     if not observation_id:
         return "ERROR: 'observation_id' is required"
@@ -2435,6 +2510,7 @@ async def create_tool_executor(
     run_id: str | None = None,
     agent_type: str | None = None,
     parent_category: str | None = None,
+    session_allowlist: list[str] | None = None,
 ) -> Callable[[str, dict[str, Any]], Any]:
     """
     Create a unified tool executor function for all agent operations.
@@ -2475,6 +2551,7 @@ async def create_tool_executor(
         history_token_limit=history_token_limit,
         db_lock=shared_lock,
         configuration=configuration,
+        session_allowlist=session_allowlist,
         run_id=run_id,
         agent_type=agent_type,
         parent_category=parent_category,
diff --git a/src/utils/filter.py b/src/utils/filter.py
index 1394a8c6..8df3590c 100644
--- a/src/utils/filter.py
+++ b/src/utils/filter.py
@@ -1,7 +1,8 @@
 import datetime
-from collections.abc import Callable
+from collections.abc import Callable, Sequence
 from logging import getLogger
 from typing import Any, TypeVar
+from typing import cast as typing_cast
 
 from sqlalchemy import ColumnElement, Select, and_, case, cast, literal, not_, or_
 from sqlalchemy.types import Numeric
@@ -28,6 +29,10 @@ COMPARISON_OPERATORS = {
 
 NUMERIC_OPERATORS = {"gte", "lte", "gt", "lt", "ne"}
 
+# JSONB columns keep containment semantics: bare lists are not membership
+# sugar, and dict values map to nested-metadata conditions rather than IN/Eq.
+JSONB_COLUMNS = ("h_metadata", "configuration", "internal_metadata")
+
 ALLOWED_EXTERNAL_TO_INTERNAL_COLUMN_MAPPING = {
     "id": "name",
     "created_at": "created_at",
@@ -56,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]]:
@@ -216,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
@@ -238,6 +327,12 @@ def _build_field_condition(
     if value == "*":
         return None
 
+    # Bare-list sugar on regular columns: {"session_id": ["a", "b"]} is
+    # shorthand for {"session_id": {"in": ["a", "b"]}}. JSONB columns are
+    # excluded — a bare list there keeps JSONB containment semantics.
+    if isinstance(value, list | tuple | set) and column_name not in JSONB_COLUMNS:
+        value = {"in": list(typing_cast(Sequence[Any], value))}
+
     # Handle comparison operators vs regular values
     if isinstance(value, dict):
         # Check if this is a comparison operators dict by looking for known operators
@@ -248,12 +343,12 @@ def _build_field_condition(
         else:
             # This is a regular value that happens to be a dict
             # For JSONB fields (metadata, configuration), check if it contains nested comparison operators
-            if column_name in ("h_metadata", "configuration", "internal_metadata"):
+            if column_name in JSONB_COLUMNS:
                 return _build_nested_metadata_conditions(column, value)  # pyright: ignore
             else:
                 return column == value
     else:
-        if column_name in ("h_metadata", "configuration", "internal_metadata"):
+        if column_name in JSONB_COLUMNS:
             return column.contains(value)
         else:
             return column == value
diff --git a/src/utils/representation.py b/src/utils/representation.py
index 674b25bb..01e4b70a 100644
--- a/src/utils/representation.py
+++ b/src/utils/representation.py
@@ -7,6 +7,35 @@ from pydantic import BaseModel, Field, field_validator
 from src import models
 from src.utils.formatting import parse_datetime_iso
 
+# Conclusion levels whose `session_name` stamp is trustworthy enough to scope on.
+#
+# Explicit conclusions come from the deriver over a single session's message
+# batch, so their stamp is authoritative. Deductive/inductive conclusions are
+# produced by the dreamer, which reads across *all* sessions (its discovery
+# tools default to session_only=False) but stamps its output with one session —
+# whichever holds the most recent explicit conclusion, see
+# dreamer/dream_scheduler.py. Serving those under a session allowlist would leak
+# conclusions synthesized from sessions outside it.
+#
+# ponytail: whole-level exclusion rather than per-conclusion provenance. The
+# reasoning trees already link each conclusion to its premises, so the real fix
+# is an authoritative source-session set per conclusion; until that exists this
+# fails closed. Tracked in DEV-2201.
+ALLOWLIST_SAFE_LEVELS = ("explicit",)
+
+
+def allowlist_safe_levels(levels: list[str] | None) -> list[str]:
+    """Narrow a level filter to those safe to serve under a session allowlist.
+
+    Returns the intersection with :data:`ALLOWLIST_SAFE_LEVELS`; ``None`` means
+    "no level filter requested" and yields the full safe set. An empty result
+    means the caller asked only for levels we can't scope, and should receive
+    nothing rather than unscoped conclusions.
+    """
+    if levels is None:
+        return list(ALLOWLIST_SAFE_LEVELS)
+    return [level for level in levels if level in ALLOWLIST_SAFE_LEVELS]
+
 
 def _strip_microseconds_and_timezone(timestamp: datetime) -> datetime:
     """
diff --git a/src/vector_store/__init__.py b/src/vector_store/__init__.py
index f0fccb63..85fdd9ff 100644
--- a/src/vector_store/__init__.py
+++ b/src/vector_store/__init__.py
@@ -202,7 +202,16 @@ def _create_store_by_type(store_type: str) -> VectorStore:
 
         return TurbopufferVectorStore()
     elif store_type == "lancedb":
-        from src.vector_store.lancedb import LanceDBVectorStore
+        try:
+            from src.vector_store.lancedb import LanceDBVectorStore
+        except ImportError as exc:
+            raise RuntimeError(
+                "VECTOR_STORE.TYPE is set to 'lancedb', but the 'lancedb' package "
+                "is not installed (for example on macOS Intel, where it is omitted "
+                "from dependencies because PyPI has no wheel). "
+                "Use TYPE 'pgvector' or 'turbopuffer', or install lancedb manually. "
+                f"Original import error: {exc}"
+            ) from exc
 
         return LanceDBVectorStore()
     else:
diff --git a/src/vector_store/lancedb.py b/src/vector_store/lancedb.py
index 0b621971..1b4c4880 100644
--- a/src/vector_store/lancedb.py
+++ b/src/vector_store/lancedb.py
@@ -298,10 +298,15 @@ class LanceDBVectorStore(VectorStore):
             if not _VALID_IDENTIFIER_PATTERN.match(key):
                 raise ValueError(f"Invalid filter key: {key!r}")
 
-            # Check if value is a dict with "in" operator
-            if isinstance(value, dict) and "in" in value:
-                # IN clause for list membership
-                in_values = cast(Sequence[Any], value["in"])
+            # Membership: dict form {"in": [...]} or bare-list sugar
+            if (isinstance(value, dict) and "in" in value) or isinstance(
+                value, list | tuple | set
+            ):
+                in_values = (
+                    cast(Sequence[Any], value["in"])
+                    if isinstance(value, dict)
+                    else list(cast(Sequence[Any], value))
+                )
                 if in_values:
                     escaped_values = [
                         f"'{str(v).replace(chr(39), chr(39) + chr(39))}'"
@@ -310,6 +315,11 @@ class LanceDBVectorStore(VectorStore):
                         for v in in_values
                     ]
                     conditions.append(f"{key} IN ({', '.join(escaped_values)})")
+                else:
+                    # An empty membership list matches nothing. Emitting no
+                    # condition would silently widen the result set
+                    # (fail-open); force an always-false condition instead.
+                    conditions.append("1 = 0")
             # Handle string values with proper quoting
             elif isinstance(value, str):
                 # Escape single quotes in the value
diff --git a/src/vector_store/turbopuffer.py b/src/vector_store/turbopuffer.py
index ded0c691..e7825310 100644
--- a/src/vector_store/turbopuffer.py
+++ b/src/vector_store/turbopuffer.py
@@ -20,9 +20,7 @@ from . import VectorQueryResult, VectorRecord, VectorStore
 
 logger = logging.getLogger(__name__)
 
-# Type aliases for Turbopuffer's filter formats
-EqFilter = tuple[str, Literal["Eq"], Any]
-InFilter = tuple[str, Literal["In"], Sequence[Any]]
+# Type alias for Turbopuffer's AND filter format
 AndFilter = tuple[Literal["And"], Sequence[Filter]]
 
 DISTANCE_METRIC = "cosine_distance"
@@ -245,13 +243,17 @@ class TurbopufferVectorStore(VectorStore):
         if not filters:
             return None
 
-        filter_list: list[EqFilter | InFilter] = []
+        filter_list: list[Filter] = []
         for key, value in filters.items():
             # Check if value is a dict with "in" operator
             if isinstance(value, dict) and "in" in value:
                 # Membership filter using "In" operator
-                in_values = cast(Sequence[Any], value["in"])
-                filter_list.append((key, "In", in_values))
+                in_values = list(cast(Sequence[Any], value["in"]))
+                filter_list.append(self._membership_filter(key, in_values))
+            elif isinstance(value, list | tuple | set):
+                # Bare-list sugar: same membership semantics as {"in": [...]}
+                in_values = list(cast(Sequence[Any], value))
+                filter_list.append(self._membership_filter(key, in_values))
             else:
                 # Simple equality filter using "Eq" operator
                 filter_list.append((key, "Eq", cast(Any, value)))
@@ -266,6 +268,20 @@ class TurbopufferVectorStore(VectorStore):
         and_filter: AndFilter = ("And", filter_list)
         return and_filter
 
+    @staticmethod
+    def _membership_filter(key: str, values: list[Any]) -> Filter:
+        """Build an "In" membership filter, failing closed on an empty list.
+
+        Turbopuffer's empty-"In" semantics are undocumented, so an empty
+        allowlist emits an explicit contradiction (`Eq(x) AND NotEq(x)` is
+        false for every document) rather than risk a fail-open widening.
+        Mirrors lancedb's `1 = 0` guard.
+        """
+        if not values:
+            never: AndFilter = ("And", [(key, "Eq", ""), (key, "NotEq", "")])
+            return never
+        return (key, "In", values)
+
     async def delete_many(self, namespace: str, ids: list[str]) -> None:
         """
         Delete multiple vectors from Turbopuffer.
diff --git a/tests/crud/test_representation_manager.py b/tests/crud/test_representation_manager.py
index f4be22dd..f551de76 100644
--- a/tests/crud/test_representation_manager.py
+++ b/tests/crud/test_representation_manager.py
@@ -233,6 +233,243 @@ class TestRepresentationManagerSoftDelete:
         assert contents[1:] == ["tie 2", "tie 1", "tie 0"]
 
 
+class TestRepresentationManagerSessionScoping:
+    """Tests that the session allowlist is applied uniformly to every query path.
+
+    Regression for DEV-1994: session_name used to be applied only to the
+    recent-documents query; the semantic and most-derived paths ignored it,
+    so limit_to_session leaked cross-session conclusions.
+    """
+
+    async def _setup(
+        self,
+        db_session: AsyncSession,
+        test_workspace: models.Workspace,
+        test_peer: models.Peer,
+    ) -> tuple[models.Session, models.Session, RepresentationManager]:
+        """Create two sessions and documents in each, plus a session-less doc."""
+        test_peer2 = models.Peer(
+            name=str(generate_nanoid()), workspace_name=test_workspace.name
+        )
+        db_session.add(test_peer2)
+        await db_session.flush()
+
+        session_a = models.Session(
+            name=str(generate_nanoid()), workspace_name=test_workspace.name
+        )
+        session_b = models.Session(
+            name=str(generate_nanoid()), workspace_name=test_workspace.name
+        )
+        db_session.add_all([session_a, session_b])
+        await db_session.flush()
+
+        collection = models.Collection(
+            workspace_name=test_workspace.name,
+            observer=test_peer.name,
+            observed=test_peer2.name,
+        )
+        db_session.add(collection)
+        await db_session.flush()
+
+        db_session.add_all(
+            [
+                models.Document(
+                    workspace_name=test_workspace.name,
+                    observer=test_peer.name,
+                    observed=test_peer2.name,
+                    content="in-scope observation",
+                    session_name=session_a.name,
+                    times_derived=1,
+                ),
+                models.Document(
+                    workspace_name=test_workspace.name,
+                    observer=test_peer.name,
+                    observed=test_peer2.name,
+                    content="out-of-scope observation",
+                    session_name=session_b.name,
+                    times_derived=100,
+                ),
+                # Dream-produced documents have no session_name; a session
+                # allowlist must exclude them (fail-closed).
+                models.Document(
+                    workspace_name=test_workspace.name,
+                    observer=test_peer.name,
+                    observed=test_peer2.name,
+                    content="sessionless dream observation",
+                    session_name=None,
+                    times_derived=50,
+                ),
+            ]
+        )
+        await db_session.flush()
+
+        manager = RepresentationManager(
+            test_workspace.name,
+            observer=test_peer.name,
+            observed=test_peer2.name,
+        )
+        return session_a, session_b, manager
+
+    @pytest.mark.asyncio
+    async def test_recent_respects_session_allowlist(
+        self,
+        db_session: AsyncSession,
+        sample_data: tuple[models.Workspace, models.Peer],
+    ):
+        test_workspace, test_peer = sample_data
+        session_a, _, manager = await self._setup(db_session, test_workspace, test_peer)
+
+        results = await manager._query_documents_recent(  # pyright: ignore[reportPrivateUsage]
+            db_session, top_k=10, session_allowlist=[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_allowlist=[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_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]},
+            # Scoped recall serves only levels with a trustworthy session
+            # stamp (ALLOWLIST_SAFE_LEVELS / DEV-2201).
+            "level": {"in": ["explicit"]},
+        }
+
+    @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_allowlist=[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_allowlist=[],
+            include_most_derived=True,
+        )
+
+        assert representation.explicit == []
+        assert representation.deductive == []
+
+    def test_build_filter_conditions_empty_allowlist_fails_closed(self):
+        """The filter-builder layer itself must fail closed, independent of the
+        early-return guard in _get_working_representation_internal. An empty
+        allowlist emits an empty `in` (renders as always-false downstream), not
+        an omitted filter."""
+        manager = RepresentationManager(
+            "workspace", observer="observer", observed="observed"
+        )
+
+        # 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 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"
+        }
+
+
 class TestRepresentationManagerSave:
     @pytest.mark.asyncio
     async def test_save_representation_filters_blank_observations_before_embedding(
diff --git a/tests/telemetry/test_sentry_before_send.py b/tests/telemetry/test_sentry_before_send.py
new file mode 100644
index 00000000..7e63b555
--- /dev/null
+++ b/tests/telemetry/test_sentry_before_send.py
@@ -0,0 +1,82 @@
+"""Tests for the shared Sentry before_send filter.
+
+default_before_send runs in every entrypoint (API + deriver). It drops known
+non-actionable exceptions and collapses DB connection-pool checkout timeouts
+into a single warning-level issue so they stop spawning a fresh error issue per
+transaction (fleet-wide saturation symptom, tracked in DEV-1852).
+"""
+
+from typing import TYPE_CHECKING, cast
+
+import pytest
+import sentry_sdk
+from fastapi.exceptions import RequestValidationError
+from pydantic import ValidationError
+from sqlalchemy.exc import OperationalError
+
+from src.exceptions import ResourceNotFoundException
+from src.telemetry.sentry import default_before_send, initialize_sentry
+
+if TYPE_CHECKING:
+    from sentry_sdk._types import Event, Hint
+
+
+def _hint(exc: BaseException) -> "Hint":
+    return cast("Hint", {"exc_info": (type(exc), exc, None)})
+
+
+def _event(**kwargs: object) -> "Event":
+    return cast("Event", cast(object, dict(kwargs)))
+
+
+def test_connection_timeout_is_consolidated_and_downgraded() -> None:
+    exc = OperationalError("SELECT 1", {}, Exception("connection timeout expired"))
+    out = default_before_send({}, _hint(exc))
+    assert out == {
+        "fingerprint": ["honcho-db-connection-timeout"],
+        "level": "warning",
+    }
+
+
+def test_unrelated_operational_error_passes_through() -> None:
+    exc = OperationalError("SELECT 1", {}, Exception("some other db failure"))
+    event = _event(level="error")
+    assert default_before_send(event, _hint(exc)) == {"level": "error"}
+
+
+def test_honcho_and_validation_errors_are_dropped() -> None:
+    assert default_before_send({}, _hint(ResourceNotFoundException("nope"))) is None
+    assert (
+        default_before_send({}, _hint(ValidationError.from_exception_data("x", [])))
+        is None
+    )
+    assert default_before_send({}, _hint(RequestValidationError([]))) is None
+
+
+def test_events_without_exc_info_pass_through() -> None:
+    event = _event(release="1.0")
+    assert default_before_send(event, None) == {"release": "1.0"}
+    assert default_before_send(event, cast("Hint", {})) == {"release": "1.0"}
+
+
+def _captured_before_send(monkeypatch: pytest.MonkeyPatch, **kwargs: object) -> object:
+    captured: dict[str, object] = {}
+
+    def fake_init(**init_kwargs: object) -> None:
+        captured.update(init_kwargs)
+
+    monkeypatch.setattr(sentry_sdk, "init", fake_init)
+    initialize_sentry(integrations=[], **kwargs)  # pyright: ignore[reportArgumentType]
+    return captured["before_send"]
+
+
+def test_initialize_sentry_defaults_to_shared_filter(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    assert _captured_before_send(monkeypatch) is default_before_send
+
+
+def test_initialize_sentry_explicit_none_bypasses_shared_filter(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    assert _captured_before_send(monkeypatch, before_send=None) is None
diff --git a/tests/test_advanced_filters.py b/tests/test_advanced_filters.py
index 768f29bc..3b187470 100644
--- a/tests/test_advanced_filters.py
+++ b/tests/test_advanced_filters.py
@@ -235,6 +235,86 @@ async def test_comparison_operators_filters(
             ), f"Unexpected message '{message_config['content']}' found in results for {description}"
 
 
+@pytest.mark.asyncio
+async def test_bare_list_membership_sugar(
+    client: TestClient,
+    sample_data: tuple[Workspace, Peer],
+):
+    """A bare list on a regular column is shorthand for {"in": [...]}.
+
+    JSONB metadata columns are excluded from the sugar: a bare list there
+    keeps JSONB containment semantics.
+    """
+    test_workspace, test_peer = sample_data
+
+    # Second peer so peer_id membership has something to exclude
+    peer2_name = str(generate_nanoid())
+    client.post(
+        f"/v3/workspaces/{test_workspace.name}/peers",
+        json={"id": peer2_name},
+    )
+
+    session_id = str(generate_nanoid())
+    session_response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/sessions",
+        json={
+            "id": session_id,
+            "peer_names": {test_peer.name: {}, peer2_name: {}},
+        },
+    )
+    assert session_response.status_code == 201
+
+    message_configs = [
+        {
+            "content": "From peer one",
+            "peer_id": test_peer.name,
+            "metadata": {"tags": ["important", "urgent"]},
+        },
+        {
+            "content": "From peer two",
+            "peer_id": peer2_name,
+            "metadata": {"tags": ["normal"]},
+        },
+    ]
+    messages_response = client.post(
+        f"/v3/workspaces/{test_workspace.name}/sessions/{session_id}/messages",
+        json={"messages": message_configs},
+    )
+    assert messages_response.status_code == 201
+
+    def list_contents(filter_config: dict[str, Any]) -> list[str]:
+        response = client.post(
+            f"/v3/workspaces/{test_workspace.name}/sessions/{session_id}/messages/list",
+            json={"filters": filter_config},
+        )
+        assert response.status_code == 200
+        return [item["content"] for item in response.json()["items"]]
+
+    # Bare list == membership on a regular column
+    assert list_contents({"peer_id": [test_peer.name]}) == ["From peer one"]
+
+    # Multiple values
+    assert sorted(list_contents({"peer_id": [test_peer.name, peer2_name]})) == [
+        "From peer one",
+        "From peer two",
+    ]
+
+    # Equivalent to the explicit {"in": [...]} form
+    assert list_contents({"peer_id": [test_peer.name]}) == list_contents(
+        {"peer_id": {"in": [test_peer.name]}}
+    )
+
+    # Empty list matches nothing (fail-closed), never everything
+    assert list_contents({"peer_id": []}) == []
+
+    # JSONB metadata keeps containment semantics for bare lists:
+    # matches arrays containing ALL listed elements, not membership.
+    assert list_contents({"metadata": {"tags": ["important", "urgent"]}}) == [
+        "From peer one"
+    ]
+    assert list_contents({"metadata": {"tags": ["important", "missing"]}}) == []
+
+
 @pytest.mark.asyncio
 async def test_wildcard_filters(
     client: TestClient, sample_data: tuple[Workspace, Peer]
diff --git a/tests/test_cache_redaction.py b/tests/test_cache_redaction.py
new file mode 100644
index 00000000..0ff2bbc7
--- /dev/null
+++ b/tests/test_cache_redaction.py
@@ -0,0 +1,126 @@
+"""Unit tests for the cache client's _redact_cache_url helper."""
+
+import pytest
+
+from src.cache.client import _redact_cache_url
+
+
+class TestRedactCacheUrl:
+    """Tests for _redact_cache_url — a security-relevant logging helper
+    that must never raise and must never leak a password."""
+
+    # --- Password masking ---
+
+    def test_password_only_userinfo(self):
+        assert (
+            _redact_cache_url("redis://:secret@localhost:6379/0")
+            == "redis://:***@localhost:6379/0"
+        )
+
+    def test_user_and_password(self):
+        result = _redact_cache_url("redis://user:s3cret@10.0.0.1:6380/2")
+        assert "***" in result
+        assert "s3cret" not in result
+        assert "user" in result
+
+    def test_rediss_protocol(self):
+        result = _redact_cache_url("rediss://:secret@redis.example.com:6380")
+        assert result.startswith("rediss://")
+        assert "***" in result
+        assert "secret" not in result
+
+    def test_complex_password(self):
+        result = _redact_cache_url("redis://:p%40ssw0rd!%24@host:6379/0")
+        assert "***" in result
+        assert "p%40ssw0rd" not in result
+
+    def test_password_never_leaked(self):
+        """The original password must never appear in the redacted output."""
+        for url in [
+            "redis://:hunter2@localhost:6379/0",
+            "redis://admin:hunter2@localhost:6379/0",
+            "rediss://:hunter2@[::1]:6380/1",
+        ]:
+            assert "hunter2" not in _redact_cache_url(url)
+
+    # --- Secrets in query parameters ---
+    # redis-py accepts ?password= (querystring options become client
+    # kwargs) and cashews accepts ?secret= (HMAC signing key), so both
+    # are real configuration paths that must not reach the logs.
+
+    @pytest.mark.parametrize("param", ["password", "secret", "PASSWORD"])
+    def test_query_param_secret_masked(self, param: str):
+        result = _redact_cache_url(f"redis://host:6379/0?{param}=s3cret")
+        assert "s3cret" not in result
+        assert f"{param}=***" in result
+
+    def test_query_param_masking_preserves_other_params(self):
+        result = _redact_cache_url("redis://host:6379/0?db=1&password=s3cret&ssl=true")
+        assert "s3cret" not in result
+        assert "db=1" in result
+        assert "ssl=true" in result
+
+    def test_userinfo_and_query_secret_both_masked(self):
+        result = _redact_cache_url("redis://:hunter2@host:6379/0?secret=s3cret")
+        assert "hunter2" not in result
+        assert "s3cret" not in result
+
+    def test_non_secret_query_params_unchanged(self):
+        url = "redis://localhost:6379/0?suppress=true"
+        assert _redact_cache_url(url) == url
+
+    # --- No-password URLs (returned unchanged) ---
+
+    def test_user_without_password_unchanged(self):
+        assert (
+            _redact_cache_url("redis://user@localhost:6379/0")
+            == "redis://user@localhost:6379/0"
+        )
+
+    def test_in_memory_url_unchanged(self):
+        assert _redact_cache_url("mem://") == "mem://"
+
+    # --- IPv6 ---
+
+    def test_ipv6_brackets_preserved(self):
+        result = _redact_cache_url("rediss://:secret@[::1]:6380/1")
+        assert "[::1]" in result
+        assert "***" in result
+        assert "secret" not in result
+
+    # --- Malformed URLs (must NOT raise) ---
+
+    def test_invalid_port_redacts_password(self):
+        """Regression test for two review findings: accessing
+        ``parsed.port`` on a URL with a non-numeric port raises
+        ``ValueError`` (must not crash startup inside an except block),
+        and the fallback must never echo the raw URL back — the
+        password has to be masked even when the port is unparseable.
+        """
+        result = _redact_cache_url("redis://:pass@host:notaport/0")
+        assert "pass" not in result
+        assert "***" in result
+
+    def test_out_of_range_port_redacts_password(self):
+        result = _redact_cache_url("redis://:supersecret@host:99999/0")
+        assert "supersecret" not in result
+        assert "***" in result
+
+    def test_unparseable_url_never_echoed(self):
+        # Unbalanced IPv6 bracket makes urlparse itself raise; the
+        # fallback must return a placeholder, not the raw input.
+        result = _redact_cache_url("redis://:secret@[::1:6379/0")
+        assert "secret" not in result
+
+    def test_missing_scheme_never_echoed(self):
+        # Without "redis://" urlparse sees no netloc, so the userinfo
+        # (and its password) is invisible to .password — the string
+        # must not be echoed back.
+        result = _redact_cache_url(":hunter2@host:6379/0")
+        assert "hunter2" not in result
+
+    def test_garbage_input_does_not_raise(self):
+        assert isinstance(_redact_cache_url("not a url at all"), str)
+
+    def test_empty_string_does_not_raise(self):
+        assert isinstance(_redact_cache_url(""), str)
diff --git a/tests/test_session_allowlist.py b/tests/test_session_allowlist.py
new file mode 100644
index 00000000..aa6077cb
--- /dev/null
+++ b/tests/test_session_allowlist.py
@@ -0,0 +1,679 @@
+"""
+Tests for the session allowlist (DEV-1995).
+
+Covers the constrained `filters` surface on dialectic/representation
+(extract_session_allowlist), fail-closed conclusion recall (search_memory),
+and the strict allowlist ∩ membership intersection in message cruds.
+"""
+
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi.testclient import TestClient
+from nanoid import generate as generate_nanoid
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src import crud, models
+from src.config import settings
+from src.crud.message import resolve_session_scope
+from src.exceptions import FilterError
+from src.models import Peer, Workspace
+from src.security import JWTParams, create_jwt
+from src.utils.agent_tools import search_memory
+from src.utils.filter import (
+    MAX_SESSION_ALLOWLIST_ENTRIES,
+    extract_session_allowlist,
+)
+
+
+class TestExtractSessionAllowlist:
+    def test_none_passthrough(self):
+        assert extract_session_allowlist(None) is None
+
+    def test_single_id(self):
+        assert extract_session_allowlist({"session_id": "s1"}) == ["s1"]
+
+    def test_bare_list(self):
+        assert extract_session_allowlist({"session_id": ["s1", "s2"]}) == ["s1", "s2"]
+
+    def test_in_operator(self):
+        assert extract_session_allowlist({"session_id": {"in": ["s1"]}}) == ["s1"]
+
+    def test_dedupes_preserving_order(self):
+        assert extract_session_allowlist({"session_id": ["s2", "s1", "s2"]}) == [
+            "s2",
+            "s1",
+        ]
+
+    def test_empty_list_preserved_for_fail_closed(self):
+        assert extract_session_allowlist({"session_id": []}) == []
+
+    def test_unsupported_key_rejected(self):
+        with pytest.raises(FilterError, match="Unsupported filter key"):
+            extract_session_allowlist({"peer_id": ["a"], "session_id": ["s1"]})
+
+    def test_missing_session_id_rejected(self):
+        with pytest.raises(FilterError, match="must contain"):
+            extract_session_allowlist({})
+
+    def test_bad_shapes_rejected(self):
+        for bad in [123, {"gte": "x"}, {"in": "s1"}, [1, 2], [""], None]:
+            with pytest.raises(FilterError):
+                extract_session_allowlist({"session_id": bad})
+
+    def test_cap_enforced(self):
+        too_many = [f"s{i}" for i in range(MAX_SESSION_ALLOWLIST_ENTRIES + 1)]
+        with pytest.raises(FilterError, match="at most"):
+            extract_session_allowlist({"session_id": too_many})
+
+    def test_must_include_satisfied(self):
+        assert extract_session_allowlist(
+            {"session_id": ["s1", "s2"]}, must_include="s2"
+        ) == ["s1", "s2"]
+
+    def test_must_include_missing_rejected(self):
+        with pytest.raises(FilterError, match="must be included"):
+            extract_session_allowlist({"session_id": ["s1"]}, must_include="s2")
+
+    def test_must_include_ignored_without_filters(self):
+        assert extract_session_allowlist(None, must_include="s1") is None
+
+    def test_must_include_none_is_no_constraint(self):
+        assert extract_session_allowlist({"session_id": ["s1"]}, must_include=None) == [
+            "s1"
+        ]
+
+
+class TestSearchMemoryAllowlist:
+    @pytest.mark.asyncio
+    async def test_allowlist_pushed_down_as_filters(self):
+        with patch(
+            "src.crud.query_documents", new=AsyncMock(return_value=[])
+        ) as mock_query:
+            await search_memory(
+                workspace_name="w",
+                observer="o",
+                observed="o",
+                query="q",
+                limit=5,
+                levels=["explicit"],
+                embedding=[0.1],
+                session_allowlist=["s1", "s2"],
+            )
+        assert mock_query.await_args is not None
+        assert mock_query.await_args.kwargs["filters"] == {
+            "level": {"in": ["explicit"]},
+            "session_name": {"in": ["s1", "s2"]},
+        }
+
+    @pytest.mark.asyncio
+    async def test_allowlist_narrows_levels_to_allowlist_safe(self):
+        """Only levels with a trustworthy session stamp survive scoping.
+
+        Dream-derived levels are stamped with one session but synthesized
+        across many (DEV-2201), so they can't be served under an allowlist.
+        """
+        with patch(
+            "src.crud.query_documents", new=AsyncMock(return_value=[])
+        ) as mock_query:
+            await search_memory(
+                workspace_name="w",
+                observer="o",
+                observed="o",
+                query="q",
+                limit=5,
+                levels=["explicit", "inductive"],
+                embedding=[0.1],
+                session_allowlist=["s1"],
+            )
+        assert mock_query.await_args is not None
+        assert mock_query.await_args.kwargs["filters"]["level"] == {"in": ["explicit"]}
+
+    @pytest.mark.asyncio
+    async def test_allowlist_defaults_to_explicit_when_no_levels_requested(self):
+        with patch(
+            "src.crud.query_documents", new=AsyncMock(return_value=[])
+        ) as mock_query:
+            await search_memory(
+                workspace_name="w",
+                observer="o",
+                observed="o",
+                query="q",
+                limit=5,
+                embedding=[0.1],
+                session_allowlist=["s1"],
+            )
+        assert mock_query.await_args is not None
+        assert mock_query.await_args.kwargs["filters"]["level"] == {"in": ["explicit"]}
+
+    @pytest.mark.asyncio
+    async def test_derived_only_request_under_allowlist_returns_empty(self):
+        """The dialectic's derived prefetch short-circuits instead of querying."""
+        with patch(
+            "src.crud.query_documents", new=AsyncMock(return_value=[])
+        ) as mock_query:
+            result = await search_memory(
+                workspace_name="w",
+                observer="o",
+                observed="o",
+                query="q",
+                limit=5,
+                levels=["deductive", "inductive", "contradiction"],
+                embedding=[0.1],
+                session_allowlist=["s1"],
+            )
+        mock_query.assert_not_awaited()
+        assert result.is_empty()
+
+    @pytest.mark.asyncio
+    async def test_levels_untouched_without_allowlist(self):
+        """No allowlist means no level narrowing — unscoped recall is unchanged."""
+        with patch(
+            "src.crud.query_documents", new=AsyncMock(return_value=[])
+        ) as mock_query:
+            await search_memory(
+                workspace_name="w",
+                observer="o",
+                observed="o",
+                query="q",
+                limit=5,
+                levels=["deductive", "inductive"],
+                embedding=[0.1],
+            )
+        assert mock_query.await_args is not None
+        assert mock_query.await_args.kwargs["filters"] == {
+            "level": {"in": ["deductive", "inductive"]}
+        }
+
+    @pytest.mark.asyncio
+    async def test_empty_allowlist_fails_closed_without_querying(self):
+        with patch(
+            "src.crud.query_documents", new=AsyncMock(return_value=[])
+        ) as mock_query:
+            result = await search_memory(
+                workspace_name="w",
+                observer="o",
+                observed="o",
+                query="q",
+                limit=5,
+                embedding=[0.1],
+                session_allowlist=[],
+            )
+        mock_query.assert_not_awaited()
+        assert result.is_empty()
+
+
+class TestMessageCrudAllowlistIntersection:
+    """allowlist ∩ observer-membership, fail-closed on empty intersection."""
+
+    async def _setup_two_sessions(
+        self,
+        client: TestClient,
+        workspace: Workspace,
+        peer: Peer,
+    ) -> tuple[str, str]:
+        ids: list[str] = []
+        for marker in ("alpha", "beta"):
+            session_id = str(generate_nanoid())
+            resp = client.post(
+                f"/v3/workspaces/{workspace.name}/sessions",
+                json={"id": session_id, "peer_names": {peer.name: {}}},
+            )
+            assert resp.status_code == 201
+            resp = client.post(
+                f"/v3/workspaces/{workspace.name}/sessions/{session_id}/messages",
+                json={
+                    "messages": [
+                        {
+                            "content": f"needle in {marker}",
+                            "peer_id": peer.name,
+                        }
+                    ]
+                },
+            )
+            assert resp.status_code == 201
+            ids.append(session_id)
+        return ids[0], ids[1]
+
+    @pytest.mark.asyncio
+    async def test_grep_messages_intersects_allowlist(
+        self,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        workspace, peer = sample_data
+        session_a, session_b = await self._setup_two_sessions(client, workspace, peer)
+
+        snippets = await crud.grep_messages(
+            workspace_name=workspace.name,
+            session_name=None,
+            text="needle",
+            observer=peer.name,
+            session_allowlist=[session_a],
+        )
+        contents = [m.content for matches, _ in snippets for m in matches]
+        assert contents == ["needle in alpha"]
+
+        # A session the observer is NOT a member of contributes nothing,
+        # even when allowlisted (strict intersection).
+        foreign = str(generate_nanoid())
+        snippets = await crud.grep_messages(
+            workspace_name=workspace.name,
+            session_name=None,
+            text="needle",
+            observer=peer.name,
+            session_allowlist=[foreign],
+        )
+        assert snippets == []
+
+        # Both sessions allowlisted -> both found
+        snippets = await crud.grep_messages(
+            workspace_name=workspace.name,
+            session_name=None,
+            text="needle",
+            observer=peer.name,
+            session_allowlist=[session_a, session_b],
+        )
+        assert len(snippets) == 2
+
+    @pytest.mark.asyncio
+    async def test_get_messages_by_date_range_intersects_allowlist(
+        self,
+        client: TestClient,
+        db_session: AsyncSession,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        workspace, peer = sample_data
+        session_a, _session_b = await self._setup_two_sessions(client, workspace, peer)
+
+        messages = await crud.get_messages_by_date_range(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=None,
+            observer=peer.name,
+            session_allowlist=[session_a],
+        )
+        assert [m.content for m in messages] == ["needle in alpha"]
+
+        # Empty allowlist fails closed
+        messages = await crud.get_messages_by_date_range(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=None,
+            observer=peer.name,
+            session_allowlist=[],
+        )
+        assert messages == []
+
+
+class TestPeerScopedJWTAllowlistGate:
+    """A peer-scoped key may only allowlist sessions its peer actively belongs to.
+
+    The gate uses `active_only=True` so it agrees with the `is_peer_in_session`
+    check on `options.session_id` — a peer that has left a session is denied by
+    both, not just one.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _enable_auth(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setattr(settings.AUTH, "USE_AUTH", True)
+        monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret")
+
+    def _chat_as(
+        self,
+        client: TestClient,
+        workspace: Workspace,
+        peer: Peer,
+        token: str,
+        body: dict[str, Any],
+    ):
+        return client.post(
+            f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat",
+            json={"query": "what do you know?", **body},
+            headers={"Authorization": f"Bearer {token}"},
+        )
+
+    async def _session_with(
+        self, client: TestClient, workspace: Workspace, peer: Peer
+    ) -> str:
+        session_id = str(generate_nanoid())
+        resp = client.post(
+            f"/v3/workspaces/{workspace.name}/sessions",
+            json={"id": session_id, "peer_names": {peer.name: {}}},
+        )
+        assert resp.status_code == 201
+        return session_id
+
+    @pytest.mark.asyncio
+    async def test_member_sessions_allowed(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        session_id = await self._session_with(client, workspace, peer)
+        token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
+
+        with patch(
+            "src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")
+        ) as mock_chat:
+            resp = self._chat_as(
+                client,
+                workspace,
+                peer,
+                token,
+                {"filters": {"session_id": [session_id]}},
+            )
+        assert resp.status_code == 200
+        # The allowlist reaches the agent rather than being dropped at the gate.
+        assert mock_chat.await_args is not None
+        assert mock_chat.await_args.kwargs["session_allowlist"] == [session_id]
+
+    @pytest.mark.asyncio
+    async def test_non_member_session_denied(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        session_id = await self._session_with(client, workspace, peer)
+        token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
+
+        # One allowlisted session the peer belongs to, one it doesn't:
+        # membership must hold for *every* entry.
+        resp = self._chat_as(
+            client,
+            workspace,
+            peer,
+            token,
+            {"filters": {"session_id": [session_id, str(generate_nanoid())]}},
+        )
+        assert resp.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_left_session_denied(
+        self,
+        client: TestClient,
+        db_session: AsyncSession,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        """The regression this gate's `active_only` flag exists to prevent.
+
+        With the loose membership definition a peer that left a session still
+        passed here, while the adjacent `session_id` check rejected it.
+        """
+        workspace, peer = sample_data
+        session_id = await self._session_with(client, workspace, peer)
+        token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
+
+        await crud.remove_peers_from_session(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=session_id,
+            peer_names={peer.name},
+        )
+        await db_session.commit()
+
+        resp = self._chat_as(
+            client, workspace, peer, token, {"filters": {"session_id": [session_id]}}
+        )
+        assert resp.status_code == 401
+
+        # ...and the single-session gate agrees, which is the whole point.
+        resp = self._chat_as(client, workspace, peer, token, {"session_id": session_id})
+        assert resp.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_workspace_scoped_key_bypasses_gate(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        """Workspace keys are trusted callers — the allowlist passes as given."""
+        workspace, peer = sample_data
+        token = create_jwt(JWTParams(w=workspace.name))
+        foreign = str(generate_nanoid())
+
+        with patch("src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")):
+            resp = self._chat_as(
+                client, workspace, peer, token, {"filters": {"session_id": [foreign]}}
+            )
+        assert resp.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_empty_allowlist_still_gated(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        """`filters={"session_id": []}` is a real allowlist, not an absent one.
+
+        It must reach the gate (and pass trivially, since the empty set is a
+        subset of anything) rather than being skipped by a truthiness check.
+        """
+        workspace, peer = sample_data
+        token = create_jwt(JWTParams(w=workspace.name, p=peer.name))
+
+        with patch(
+            "src.routers.peers.agentic_chat", new=AsyncMock(return_value="ok")
+        ) as mock_chat:
+            resp = self._chat_as(
+                client, workspace, peer, token, {"filters": {"session_id": []}}
+            )
+        assert resp.status_code == 200
+        assert mock_chat.await_args is not None
+        assert mock_chat.await_args.kwargs["session_allowlist"] == []
+
+
+class TestResolveSessionScope:
+    """The tri-state contract the four message-crud call sites depend on."""
+
+    @pytest.mark.asyncio
+    async def test_unrestricted_when_no_observer_and_no_allowlist(
+        self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, _ = sample_data
+        assert await resolve_session_scope(
+            db_session, workspace.name, None, None, None
+        ) == (None, False)
+
+    @pytest.mark.asyncio
+    async def test_pinned_session_inside_allowlist_passes_through(
+        self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, _ = sample_data
+        # None (not [s1]) — the query filters on session_name directly.
+        assert await resolve_session_scope(
+            db_session, workspace.name, "s1", ["s1", "s2"], None
+        ) == (None, False)
+
+    @pytest.mark.asyncio
+    async def test_pinned_session_outside_allowlist_denies(
+        self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, _ = sample_data
+        assert await resolve_session_scope(
+            db_session, workspace.name, "s3", ["s1", "s2"], None
+        ) == (None, True)
+
+    @pytest.mark.asyncio
+    async def test_empty_allowlist_denies_rather_than_returning_empty_list(
+        self, db_session: AsyncSession, sample_data: tuple[Workspace, Peer]
+    ):
+        """Never returns [] — downstream stores drop an empty IN clause."""
+        workspace, _ = sample_data
+        allowed, deny = await resolve_session_scope(
+            db_session, workspace.name, None, [], None
+        )
+        assert (allowed, deny) == (None, True)
+
+    @pytest.mark.asyncio
+    async def test_no_db_touched_when_no_observer_lookup_needed(self):
+        """Callers pass db=None on the external-vector-store path.
+
+        The helper must not open a session of its own unless it actually needs
+        an observer lookup, or the external semantic lookup stops being the
+        first thing that happens (see
+        tests/integration/test_message_embeddings.py).
+        """
+        with patch("src.crud.message.tracked_db") as mock_tracked_db:
+            # No observer: pinned session, unrestricted, and plain allowlist.
+            assert await resolve_session_scope(None, "w", "s1", None, None) == (
+                None,
+                False,
+            )
+            assert await resolve_session_scope(None, "w", None, None, None) == (
+                None,
+                False,
+            )
+            assert await resolve_session_scope(None, "w", None, ["s1"], None) == (
+                ["s1"],
+                False,
+            )
+        mock_tracked_db.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_observer_scope_intersected_with_allowlist(
+        self,
+        client: TestClient,
+        db_session: AsyncSession,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        workspace, peer = sample_data
+        session_id = str(generate_nanoid())
+        resp = client.post(
+            f"/v3/workspaces/{workspace.name}/sessions",
+            json={"id": session_id, "peer_names": {peer.name: {}}},
+        )
+        assert resp.status_code == 201
+
+        allowed, deny = await resolve_session_scope(
+            db_session, workspace.name, None, [session_id], peer.name
+        )
+        assert (allowed, deny) == ([session_id], False)
+
+        # Allowlisting only a session the observer isn't in denies outright.
+        allowed, deny = await resolve_session_scope(
+            db_session, workspace.name, None, [str(generate_nanoid())], peer.name
+        )
+        assert (allowed, deny) == (None, True)
+
+
+class TestChatRouteFilterValidation:
+    """Filter validation happens before any LLM work — safe to exercise."""
+
+    def _chat(
+        self,
+        client: TestClient,
+        workspace: Workspace,
+        peer: Peer,
+        body: dict[str, Any],
+    ):
+        return client.post(
+            f"/v3/workspaces/{workspace.name}/peers/{peer.name}/chat",
+            json={"query": "what do you know?", **body},
+        )
+
+    def test_unsupported_filter_key_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        resp = self._chat(client, workspace, peer, {"filters": {"peer_id": ["x"]}})
+        assert resp.status_code == 422
+
+    def test_bad_filter_shape_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        resp = self._chat(client, workspace, peer, {"filters": {"session_id": 42}})
+        assert resp.status_code == 422
+
+    def test_session_id_not_in_allowlist_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        resp = self._chat(
+            client,
+            workspace,
+            peer,
+            {"session_id": "s-outside", "filters": {"session_id": ["s1", "s2"]}},
+        )
+        assert resp.status_code == 422
+
+    def test_allowlist_cap_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        too_many = [f"s{i}" for i in range(MAX_SESSION_ALLOWLIST_ENTRIES + 1)]
+        resp = self._chat(
+            client, workspace, peer, {"filters": {"session_id": too_many}}
+        )
+        assert resp.status_code == 422
+
+
+class TestRepresentationRouteFilters:
+    @pytest.mark.asyncio
+    async def test_representation_scoped_by_filters(
+        self,
+        client: TestClient,
+        db_session: AsyncSession,
+        sample_data: tuple[Workspace, Peer],
+    ):
+        workspace, peer = sample_data
+
+        session_a = models.Session(
+            name=str(generate_nanoid()), workspace_name=workspace.name
+        )
+        session_b = models.Session(
+            name=str(generate_nanoid()), workspace_name=workspace.name
+        )
+        db_session.add_all([session_a, session_b])
+        await db_session.flush()
+
+        collection = models.Collection(
+            workspace_name=workspace.name,
+            observer=peer.name,
+            observed=peer.name,
+        )
+        db_session.add(collection)
+        await db_session.flush()
+
+        db_session.add_all(
+            [
+                models.Document(
+                    workspace_name=workspace.name,
+                    observer=peer.name,
+                    observed=peer.name,
+                    content="fact from session a",
+                    session_name=session_a.name,
+                ),
+                models.Document(
+                    workspace_name=workspace.name,
+                    observer=peer.name,
+                    observed=peer.name,
+                    content="fact from session b",
+                    session_name=session_b.name,
+                ),
+                models.Document(
+                    workspace_name=workspace.name,
+                    observer=peer.name,
+                    observed=peer.name,
+                    content="sessionless dream fact",
+                    session_name=None,
+                ),
+            ]
+        )
+        await db_session.commit()
+
+        resp = client.post(
+            f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation",
+            json={"filters": {"session_id": [session_a.name]}},
+        )
+        assert resp.status_code == 200
+        representation = resp.json()["representation"]
+        assert "fact from session a" in representation
+        assert "fact from session b" not in representation
+        assert "sessionless dream fact" not in representation
+
+    def test_session_id_not_in_allowlist_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, peer = sample_data
+        resp = client.post(
+            f"/v3/workspaces/{workspace.name}/peers/{peer.name}/representation",
+            json={"session_id": "s-out", "filters": {"session_id": ["s-in"]}},
+        )
+        assert resp.status_code == 422
diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py
index 9bfb3990..71b93eb0 100644
--- a/tests/utils/test_agent_tools.py
+++ b/tests/utils/test_agent_tools.py
@@ -39,6 +39,8 @@ from src.utils.agent_tools import (
     create_observations,
     create_tool_executor,
     extract_preferences,
+    get_observation_context,
+    get_recent_history,
 )
 
 # =============================================================================
@@ -801,6 +803,7 @@ class TestSearchMemory:
             context_window: int = 2,
             embedding: list[float] | None = None,
             observer: str | None = None,
+            **_kwargs: Any,
         ) -> list[tuple[list[models.Message], list[models.Message]]]:
             _ = (workspace_name, session_name, query, limit, context_window, observer)
             fallback_embeddings.append(embedding)
@@ -909,6 +912,7 @@ class TestSearchMessagesTemporal:
             context_window: int = 2,
             embedding: list[float] | None = None,
             observer: str | None = None,
+            **_kwargs: Any,
         ) -> list[tuple[list[models.Message], list[models.Message]]]:
             _ = (
                 workspace_name,
@@ -1502,6 +1506,7 @@ class TestExtractPreferences:
             context_window: int,
             embedding: list[float] | None,
             observer: str | None = None,
+            **_kwargs: Any,
         ) -> list[tuple[list[models.Message], list[models.Message]]]:
             _ = (limit, context_window, observer)
             embedding_args.append(embedding)
@@ -1848,3 +1853,63 @@ class TestObserverPeerNameWiring:
         await _handle_get_messages_by_date_range(ctx, {"after_date": "2024-01-01"})
 
         assert captured_kwargs["observer"] == ctx.observer
+
+
+@pytest.mark.asyncio
+class TestSessionAllowlistFailClosed:
+    """A specific session_name outside the session_allowlist allowlist must fail closed.
+
+    Routes guard this too, but these CRUD/tool functions are reachable directly
+    from the dialectic loop, so the allowlist is enforced at the boundary.
+    """
+
+    async def test_get_recent_history_respects_allowlist(
+        self, db_session: AsyncSession, tool_test_data: Any
+    ):
+        workspace, _peer1, peer2, session, _messages, _ = tool_test_data
+
+        # session IS in the allowlist -> history returned
+        allowed = await get_recent_history(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=session.name,
+            observed=peer2.name,
+            session_allowlist=[session.name],
+        )
+        assert allowed  # non-empty
+
+        # session is NOT in the allowlist -> fail closed
+        blocked = await get_recent_history(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=session.name,
+            observed=peer2.name,
+            session_allowlist=["some-other-session"],
+        )
+        assert blocked == []
+
+    async def test_get_observation_context_fails_closed(
+        self, db_session: AsyncSession, tool_test_data: Any
+    ):
+        workspace, peer1, _peer2, session, messages, _ = tool_test_data
+        blocked = await get_observation_context(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=session.name,
+            message_ids=[messages[0].id],
+            observer=peer1.name,
+            session_allowlist=["some-other-session"],
+        )
+        assert blocked == []
+
+    async def test_get_messages_by_date_range_fails_closed(
+        self, db_session: AsyncSession, tool_test_data: Any
+    ):
+        workspace, _peer1, _peer2, session, _messages, _ = tool_test_data
+        blocked = await crud.get_messages_by_date_range(
+            db_session,
+            workspace_name=workspace.name,
+            session_name=session.name,
+            session_allowlist=["some-other-session"],
+        )
+        assert blocked == []
diff --git a/tests/vector_store/test_lancedb.py b/tests/vector_store/test_lancedb.py
index 0c502e4c..9b52708a 100644
--- a/tests/vector_store/test_lancedb.py
+++ b/tests/vector_store/test_lancedb.py
@@ -37,6 +37,27 @@ def store() -> LanceDBVectorStore:
     return LanceDBVectorStore()
 
 
+def test_build_where_clause_membership(store: LanceDBVectorStore) -> None:
+    """Both the dict `in` form and the bare-list sugar produce an IN clause."""
+    assert (
+        store._build_where_clause({"session_name": {"in": ["s1", "s2"]}})  # pyright: ignore[reportPrivateUsage]
+        == "session_name IN ('s1', 's2')"
+    )
+    assert (
+        store._build_where_clause({"session_name": ["s1", "s2"]})  # pyright: ignore[reportPrivateUsage]
+        == "session_name IN ('s1', 's2')"
+    )
+
+
+def test_build_where_clause_empty_membership_fails_closed(
+    store: LanceDBVectorStore,
+) -> None:
+    """An empty membership list must emit an always-false predicate, never an
+    omitted condition that would widen scope (fail-open)."""
+    assert store._build_where_clause({"session_name": {"in": []}}) == "1 = 0"  # pyright: ignore[reportPrivateUsage]
+    assert store._build_where_clause({"session_name": []}) == "1 = 0"  # pyright: ignore[reportPrivateUsage]
+
+
 @pytest.mark.asyncio
 async def test_query_returns_empty_when_table_missing(
     store: LanceDBVectorStore,
diff --git a/tests/vector_store/test_turbopuffer.py b/tests/vector_store/test_turbopuffer.py
index ca8c4ff3..cdae44d5 100644
--- a/tests/vector_store/test_turbopuffer.py
+++ b/tests/vector_store/test_turbopuffer.py
@@ -53,6 +53,30 @@ async def test_upsert_many_raises_vector_store_error_on_5xx(
     namespace_mock.write.assert_awaited_once()
 
 
+def test_build_filters_membership(store: TurbopufferVectorStore) -> None:
+    """Both the dict `in` form and the bare-list sugar produce an In filter."""
+    assert store._build_filters({"session_name": {"in": ["s1", "s2"]}}) == (  # pyright: ignore[reportPrivateUsage]
+        "session_name",
+        "In",
+        ["s1", "s2"],
+    )
+    assert store._build_filters({"session_name": ["s1", "s2"]}) == (  # pyright: ignore[reportPrivateUsage]
+        "session_name",
+        "In",
+        ["s1", "s2"],
+    )
+
+
+def test_build_filters_empty_membership_fails_closed(
+    store: TurbopufferVectorStore,
+) -> None:
+    """An empty membership list must produce an always-false filter, never an
+    omitted/empty In that could widen scope (fail-open)."""
+    never = ("And", [("session_name", "Eq", ""), ("session_name", "NotEq", "")])
+    assert store._build_filters({"session_name": {"in": []}}) == never  # pyright: ignore[reportPrivateUsage]
+    assert store._build_filters({"session_name": []}) == never  # pyright: ignore[reportPrivateUsage]
+
+
 @pytest.mark.asyncio
 async def test_upsert_many_short_circuits_on_empty(
     store: TurbopufferVectorStore,
diff --git a/uv.lock b/uv.lock
index c27d054c..ffaa6255 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1171,7 +1171,7 @@ dependencies = [
     { name = "greenlet" },
     { name = "httpx" },
     { name = "json-repair" },
-    { name = "lancedb" },
+    { name = "lancedb", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'" },
     { name = "langfuse" },
     { name = "nanoid" },
     { name = "openai" },
@@ -1225,7 +1225,7 @@ requires-dist = [
     { name = "greenlet", specifier = ">=3.0.3" },
     { name = "httpx", specifier = ">=0.27.0" },
     { name = "json-repair", specifier = ">=0.49.0" },
-    { name = "lancedb", specifier = ">=0.25.3" },
+    { name = "lancedb", marker = "platform_machine != 'x86_64' or sys_platform != 'darwin'", specifier = ">=0.25.3" },
     { name = "langfuse", specifier = ">=3.3.2" },
     { name = "nanoid", specifier = ">=2.0.0" },
     { name = "openai", specifier = ">=1.99.7" },