Merge branch 'main' into vineeth/dev-2418

# Conflicts:
#	CONTRIBUTING.md
#	SECURITY.md
#	docs/v3/contributing/guidelines.mdx

CHANGELOG.md

@@ -176,7 +176,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - Vector store queries no longer fetch embedding vectors — only document metadata is returned, reducing payload size and DB load (pgvector, lancedb, turbopuffer) (#682)
 - Langfuse trace metadata now includes `namespace`, `model`, and `provider` so traces can be filtered by deployment slice (#565)
 - Deriver: model-aware tokenizer (replaces the previously hardcoded encoding) and explicit guard on empty message content (#647)
-- Dialectic level defaults now merge correctly with per-level overrides in `src/config` (DEV-1733) (#656)
+- Dialectic level defaults now merge correctly with per-level overrides in `src/config` (#656)
 - Default dialectic tool choice switched from forced/required to `auto` (#630)
 - Vector sync given a substantial retry budget to tolerate transient embedding provider outages (#604)
 - `AgentToolConclusionsDeletedEvent` payload now carries `levels` for parity with the rest of the conclusion event surface (#612)
@@ -194,7 +194,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 - Removed strict parameter validation for thinking params on Anthropic and OpenAI transports — was rejecting valid per-transport configs (#686)
 - `reverse` query parameter is now honored on the v3 workspace list (`POST /v3/workspaces/list`), peer list (`POST /v3/workspaces/{workspace_id}/peers/list`), workspace-scoped session list (`POST /v3/workspaces/{workspace_id}/sessions/list`), and peer-scoped session list (`POST /v3/workspaces/{workspace_id}/peers/{peer_id}/sessions`). Honcho SDKs at 2.1.0+ were already sending `reverse=true` for these routes but the server silently ignored it. Ties on `created_at` now fall back to the internal nanoid `id` so ordering remains stable across pages (#685)
 - LLM client factories now receive `base_url` from `LLMSettings` for default providers — previously the override path honored `base_url` but the default path didn't, so operators pointing at OpenAI-compatible proxies via `LLM__OPENAI_BASE_URL` were ignored (#643, fixes #641)
-- Internal N+1 query in dialectic agent tool execution (DEV-1721) — collapsed per-iteration DB lookups into a single fetch (#652)
+- Internal N+1 query in dialectic agent tool execution — collapsed per-iteration DB lookups into a single fetch (#652)
 - Dreamer threshold and time-guard semantics: `check_and_schedule_dream` count filter now includes only `documents.level == 'explicit'` (dreamer-created levels are output, not input, and were inflating the threshold and creating a feedback loop); `last_dream_at` write relocated from `enqueue_dream` into `process_dream` so duplicate enqueues or failed runs no longer reset the 8-hour time guard (#573)
 - Deriver: blank observations are filtered out before embedding (previously triggered noisy embedding calls and persisted empty rows); blank-observation filtering unified across tool paths (#615)
 - Surprisal module: filter for level observations changed from `{"level": levels}` to `{"level": {"in": levels}}` — `apply_filter()` requires operator syntax, so the prior call silently returned 0 results and made the entire Surprisal phase of the Dream cycle a no-op (#581, fixes #559)

SECURITY.md

@@ -15,6 +15,7 @@ Include as much of the following as you have:
 
 - **Version** — a git commit SHA, or the release tag you are running
 - **Deployment** — self-hosted or the managed service at `api.honcho.dev`
+- **Affected component** — API, deriver, dialectic, auth/JWT, an SDK, or the managed offering
 - **Reproduction** — the exact steps, requests, or script that trigger it
 - **Proof of concept** — the smallest thing that demonstrates the issue actually works
 - **Impact** — what an attacker gains, and what they need to already have to get it

sdks/python/src/honcho/aio.py

@@ -499,6 +499,79 @@ class HonchoAio(AsyncMetadataConfigMixin):
         """Delete a workspace asynchronously."""
         await self._honcho._async_http_client.delete(routes.workspace(workspace_id))
 
+    @validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+    async def chat(
+        self,
+        query: str = Field(..., min_length=1, description="The natural language query"),
+        *,
+        session: str | SessionBase | None = None,
+        reasoning_level: Literal["minimal", "low", "medium", "high", "max"]
+        | None = None,
+        response_format: type[BaseModel] | dict[str, Any] | None = None,
+        scope: str | list[str] | None = None,
+    ) -> BaseModel | str | None:
+        """Query the entire workspace asynchronously (see Honcho.chat)."""
+        await self._honcho._ensure_workspace_async()
+        resolved_session_id = resolve_id(session)
+        body: dict[str, Any] = {"query": query, "stream": False}
+        if resolved_session_id:
+            body["session_id"] = resolved_session_id
+        if reasoning_level:
+            body["reasoning_level"] = reasoning_level
+        if scope is not None:
+            body["scope"] = scope
+        response_format_schema = serialize_response_format(response_format)
+        if response_format_schema is not None:
+            body["response_format"] = response_format_schema
+
+        data = await self._honcho._async_http_client.post(
+            routes.workspace_chat(self._honcho.workspace_id),
+            body=body,
+        )
+        content = data.get("content")
+        if not content:
+            return None
+        if isinstance(response_format, type):
+            return response_format.model_validate_json(content)
+        return content
+
+    @validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+    async def chat_stream(
+        self,
+        query: str = Field(..., min_length=1, description="The natural language query"),
+        *,
+        session: str | SessionBase | None = None,
+        reasoning_level: Literal["minimal", "low", "medium", "high", "max"]
+        | None = None,
+        response_format: type[BaseModel] | dict[str, Any] | None = None,
+        scope: str | list[str] | None = None,
+    ) -> AsyncDialecticStreamResponse:
+        """Streaming variant of :meth:`chat` (async)."""
+        await self._honcho._ensure_workspace_async()
+        resolved_session_id = resolve_id(session)
+        body: dict[str, Any] = {"query": query, "stream": True}
+        if resolved_session_id:
+            body["session_id"] = resolved_session_id
+        if reasoning_level:
+            body["reasoning_level"] = reasoning_level
+        if scope is not None:
+            body["scope"] = scope
+        response_format_schema = serialize_response_format(response_format)
+        if response_format_schema is not None:
+            body["response_format"] = response_format_schema
+
+        async def stream_response() -> AsyncGenerator[str, None]:
+            async for chunk in parse_sse_astream(
+                self._honcho._async_http_client.stream(
+                    "POST",
+                    routes.workspace_chat(self._honcho.workspace_id),
+                    body=body,
+                )
+            ):
+                yield chunk
+
+        return AsyncDialecticStreamResponse(stream_response())
+
     @validate_call
     async def search(
         self,

sdks/python/src/honcho/client.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import logging
 import os
-from collections.abc import Mapping, Sequence
+from collections.abc import Generator, Mapping, Sequence
 from typing import Any, Literal
 
 import httpx
@@ -28,10 +28,16 @@ from .http import AsyncHonchoHTTPClient, HonchoHTTPClient, routes
 from .message import Message
 from .mixins import MetadataConfigMixin
 from .pagination import SyncPage
-from .peer import Peer
+from .peer import Peer, serialize_response_format
 from .scope import Scope
 from .session import Session
-from .utils import normalize_peers_to_dict, resolve_id, validate_scope_id
+from .types import DialecticStreamResponse
+from .utils import (
+    normalize_peers_to_dict,
+    parse_sse_stream,
+    resolve_id,
+    validate_scope_id,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -686,6 +692,98 @@ class Honcho(BaseModel, MetadataConfigMixin):  # pyright: ignore[reportUnsafeMul
         """
         self._http.delete(routes.workspace(workspace_id))
 
+    @validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+    def chat(
+        self,
+        query: str = Field(..., min_length=1, description="The natural language query"),
+        *,
+        session: str | SessionBase | None = None,
+        reasoning_level: Literal["minimal", "low", "medium", "high", "max"]
+        | None = None,
+        response_format: type[BaseModel] | dict[str, Any] | None = None,
+        scope: str | list[str] | None = None,
+    ) -> BaseModel | str | None:
+        """
+        Query the entire workspace with a natural language question.
+
+        Unlike peer.chat(), which queries a single peer's representation, this
+        searches across ALL peers and observations in the workspace — use it
+        for cross-peer analysis, common themes, or workspace-wide questions.
+
+        Args:
+            query: The natural language question to ask.
+            session: Optional session to scope message retrieval to.
+            reasoning_level: Optional reasoning level: "minimal", "low",
+                             "medium", "high", or "max" (default "low").
+            response_format: Optional structure for the answer: a Pydantic
+                             model class (returns a parsed instance) or a raw
+                             JSON Schema dict (returns a JSON string).
+            scope: Optional scope name(s) restricting recall to those scopes'
+                   member sessions. Mutually exclusive with `session`.
+
+        Returns:
+            The synthesized answer, or None if no relevant information.
+        """
+        self._ensure_workspace()
+        resolved_session_id = resolve_id(session)
+        body: dict[str, Any] = {"query": query, "stream": False}
+        if resolved_session_id:
+            body["session_id"] = resolved_session_id
+        if reasoning_level:
+            body["reasoning_level"] = reasoning_level
+        if scope is not None:
+            body["scope"] = scope
+        response_format_schema = serialize_response_format(response_format)
+        if response_format_schema is not None:
+            body["response_format"] = response_format_schema
+
+        data = self._http.post(
+            routes.workspace_chat(self.workspace_id),
+            body=body,
+        )
+        content = data.get("content")
+        if not content:
+            return None
+        if isinstance(response_format, type):
+            return response_format.model_validate_json(content)
+        return content
+
+    @validate_call(config=ConfigDict(arbitrary_types_allowed=True))
+    def chat_stream(
+        self,
+        query: str = Field(..., min_length=1, description="The natural language query"),
+        *,
+        session: str | SessionBase | None = None,
+        reasoning_level: Literal["minimal", "low", "medium", "high", "max"]
+        | None = None,
+        response_format: type[BaseModel] | dict[str, Any] | None = None,
+        scope: str | list[str] | None = None,
+    ) -> DialecticStreamResponse:
+        """Streaming variant of :meth:`chat`. See chat() for argument docs."""
+        self._ensure_workspace()
+        resolved_session_id = resolve_id(session)
+        body: dict[str, Any] = {"query": query, "stream": True}
+        if resolved_session_id:
+            body["session_id"] = resolved_session_id
+        if reasoning_level:
+            body["reasoning_level"] = reasoning_level
+        if scope is not None:
+            body["scope"] = scope
+        response_format_schema = serialize_response_format(response_format)
+        if response_format_schema is not None:
+            body["response_format"] = response_format_schema
+
+        def stream_response() -> Generator[str, None, None]:
+            yield from parse_sse_stream(
+                self._http.stream(
+                    "POST",
+                    routes.workspace_chat(self.workspace_id),
+                    body=body,
+                )
+            )
+
+        return DialecticStreamResponse(stream_response())
+
     @validate_call
     def search(
         self,

sdks/python/src/honcho/http/routes.py

@@ -16,6 +16,10 @@ def workspace(workspace_id: str) -> str:
     return f"/{API_VERSION}/workspaces/{workspace_id}"
 
 
+def workspace_chat(workspace_id: str) -> str:
+    return f"/{API_VERSION}/workspaces/{workspace_id}/chat"
+
+
 def workspace_search(workspace_id: str) -> str:
     return f"/{API_VERSION}/workspaces/{workspace_id}/search"
 

sdks/typescript/src/client.ts

@@ -1,5 +1,9 @@
 import { API_VERSION } from './api-version'
 import { HonchoHTTPClient } from './http/client'
+import {
+  createDialecticStream,
+  type DialecticStreamResponse,
+} from './http/streaming'
 import { Message } from './message'
 import { Page } from './pagination'
 import { Peer } from './peer'
@@ -14,6 +18,8 @@ import type {
   QueueStatusResponse,
   ScopeResponse,
   SessionResponse,
+  WorkspaceChatParams,
+  WorkspaceChatResponse,
   WorkspaceResponse,
 } from './types/api'
 import { resolveId, transformQueueStatus } from './utils'
@@ -53,6 +59,7 @@ import {
 } from './validation'
 
 const DEFAULT_BASE_URL = 'https://api.honcho.dev'
+type ReasoningLevel = 'minimal' | 'low' | 'medium' | 'high' | 'max'
 
 /**
  * Main client for the Honcho TypeScript SDK.
@@ -401,6 +408,34 @@ export class Honcho {
     )
   }
 
+  private async _workspaceChat(
+    workspaceId: string,
+    params: WorkspaceChatParams
+  ): Promise<WorkspaceChatResponse> {
+    await this._ensureWorkspace()
+    return this._http.post<WorkspaceChatResponse>(
+      `/${API_VERSION}/workspaces/${workspaceId}/chat`,
+      { body: params }
+    )
+  }
+
+  private async _workspaceChatStream(
+    workspaceId: string,
+    params: Omit<WorkspaceChatParams, 'stream'>
+  ): Promise<Response> {
+    await this._ensureWorkspace()
+    return this._http.stream(
+      'POST',
+      `/${API_VERSION}/workspaces/${workspaceId}/chat`,
+      {
+        body: {
+          ...params,
+          stream: true,
+        },
+      }
+    )
+  }
+
   // ===========================================================================
   // Public Methods
   // ===========================================================================
@@ -948,6 +983,106 @@ export class Honcho {
     return response.map(Message.fromApiResponse)
   }
 
+  /**
+   * Query the workspace's collective knowledge using natural language.
+   *
+   * Performs agentic search and reasoning across ALL peers and observations
+   * in the workspace to synthesize a comprehensive answer. Useful for
+   * cross-peer analysis, discovering common themes, and workspace-wide queries.
+   *
+   * @param query - The natural language question to ask
+   * @param options.session - Optional session to scope message search to. Can be a session
+   *                          ID string or a Session object.
+   * @param options.reasoningLevel - Optional reasoning level for the query: "minimal", "low",
+   *                                 "medium", "high", or "max". Defaults to "low" if not provided.
+   * @param options.responseFormat - Optional JSON Schema (root type "object") the response
+   *                                 must conform to. When provided, the response content is a
+   *                                 JSON string matching this schema.
+   * @returns Promise resolving to the response string, or null if no relevant information
+   *
+   * @example
+   * ```typescript
+   * const response = await honcho.chat('What are common themes across all users?')
+   * ```
+   */
+  async chat(
+    query: string,
+    options?: {
+      session?: string | Session
+      reasoningLevel?: ReasoningLevel
+      responseFormat?: Record<string, unknown>
+      scope?: string | string[]
+    }
+  ): Promise<string | null> {
+    const validatedQuery = SearchQuerySchema.parse(query)
+    const resolvedSessionId = options?.session
+      ? resolveId(options.session)
+      : undefined
+
+    const response = await this._workspaceChat(this.workspaceId, {
+      query: validatedQuery,
+      stream: false,
+      session_id: resolvedSessionId,
+      reasoning_level: options?.reasoningLevel,
+      response_format: options?.responseFormat,
+      scope: options?.scope,
+    })
+    if (!response.content) {
+      return null
+    }
+    return response.content
+  }
+
+  /**
+   * Query the workspace's collective knowledge with streaming response.
+   *
+   * Performs agentic search and reasoning across ALL peers and observations
+   * in the workspace to synthesize a comprehensive answer, streaming the
+   * response as it is generated.
+   *
+   * @param query - The natural language question to ask
+   * @param options.session - Optional session to scope message search to. Can be a session
+   *                          ID string or a Session object.
+   * @param options.reasoningLevel - Optional reasoning level for the query: "minimal", "low",
+   *                                 "medium", "high", or "max". Defaults to "low" if not provided.
+   * @param options.responseFormat - Optional JSON Schema (root type "object") the response
+   *                                 must conform to. When provided, the response content is a
+   *                                 JSON string matching this schema.
+   * @returns Promise resolving to a DialecticStreamResponse that can be iterated over
+   *
+   * @example
+   * ```typescript
+   * const stream = await honcho.chatStream('What do all peers have in common?')
+   * for await (const chunk of stream) {
+   *   process.stdout.write(chunk)
+   * }
+   * ```
+   */
+  async chatStream(
+    query: string,
+    options?: {
+      session?: string | Session
+      reasoningLevel?: ReasoningLevel
+      responseFormat?: Record<string, unknown>
+      scope?: string | string[]
+    }
+  ): Promise<DialecticStreamResponse> {
+    const validatedQuery = SearchQuerySchema.parse(query)
+    const resolvedSessionId = options?.session
+      ? resolveId(options.session)
+      : undefined
+
+    const response = await this._workspaceChatStream(this.workspaceId, {
+      query: validatedQuery,
+      session_id: resolvedSessionId,
+      reasoning_level: options?.reasoningLevel,
+      response_format: options?.responseFormat,
+      scope: options?.scope,
+    })
+
+    return createDialecticStream(response)
+  }
+
   /**
    * Get the queue processing status, optionally scoped to an observer, sender, and/or session.
    *

sdks/typescript/src/index.ts

@@ -69,6 +69,7 @@ export type {
   SessionResponse,
   SessionSummariesResponse,
   SummaryResponse,
+  WorkspaceChatResponse,
   WorkspaceResponse,
 } from './types/api'
 

sdks/typescript/src/types/api.ts

@@ -81,6 +81,19 @@ export interface PeerChatResponse {
   content: string | null
 }
 
+export interface WorkspaceChatParams {
+  query: string
+  stream?: boolean
+  session_id?: string
+  reasoning_level?: 'minimal' | 'low' | 'medium' | 'high' | 'max'
+  response_format?: Record<string, unknown>
+  scope?: string | string[]
+}
+
+export interface WorkspaceChatResponse {
+  content: string | null
+}
+
 export interface PeerRepresentationParams {
   session_id?: string
   target?: string

src/crud/__init__.py

@@ -58,6 +58,7 @@ from .scope import (
     invalidate_scope_peer_cache,
     remove_session_from_scope,
     resolve_scope_peers,
+    resolve_scope_session_union,
     update_scope_backfill_status,
 )
 from .session import (
@@ -81,16 +82,24 @@ from .webhook import (
     list_webhook_endpoints,
 )
 from .workspace import (
+    ActivePeer,
     WorkspaceDeletionResult,
+    WorkspaceStats,
     check_no_active_sessions,
     delete_workspace,
+    get_active_peers,
     get_all_workspaces,
     get_or_create_workspace,
     get_workspace,
+    get_workspace_stats,
     update_workspace,
 )
 
 __all__ = [
+    "get_workspace_stats",
+    "get_active_peers",
+    "WorkspaceStats",
+    "ActivePeer",
     # Collection
     "get_collection",
     "get_or_create_collection",
@@ -150,6 +159,7 @@ __all__ = [
     "invalidate_scope_peer_cache",
     "remove_session_from_scope",
     "resolve_scope_peers",
+    "resolve_scope_session_union",
     "update_scope_backfill_status",
     # Session
     "SessionDeletionResult",

src/crud/scope.py

@@ -319,6 +319,26 @@ async def resolve_scope_peers(
     return resolved
 
 
+async def resolve_scope_session_union(
+    db: AsyncSession,
+    workspace_name: str,
+    scope_names: Sequence[str],
+) -> list[str]:
+    """Return the union of member sessions across the given scopes."""
+    from src.crud.message import get_peer_session_names
+
+    union: list[str] = []
+    seen: set[str] = set()
+    for scope_peer in await resolve_scope_peers(db, workspace_name, scope_names):
+        for session_name in await get_peer_session_names(
+            db, workspace_name, scope_peer
+        ):
+            if session_name not in seen:
+                seen.add(session_name)
+                union.append(session_name)
+    return union
+
+
 async def get_scope_sessions(
     workspace_name: str,
     scope_name: str,

src/crud/workspace.py

@@ -1,6 +1,8 @@
 """CRUD helpers for workspace records and workspace deletion checks."""
 
+from collections.abc import Sequence
 from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
 from logging import getLogger
 from typing import Any
 
@@ -535,3 +537,206 @@ async def delete_workspace(
         messages_deleted=messages_count,
         conclusions_deleted=conclusions_count,
     )
+
+
+@dataclass
+class WorkspaceStats:
+    """Workspace-level aggregate statistics."""
+
+    peer_count: int
+    session_count: int
+    message_count: int
+    oldest_message_at: datetime | None
+    newest_message_at: datetime | None
+
+
+@dataclass
+class ActivePeer:
+    """A peer with activity metrics."""
+
+    name: str
+    message_count: int
+    last_message_at: datetime | None
+
+
+async def get_workspace_stats(
+    db: AsyncSession,
+    workspace_name: str,
+    session_names: Sequence[str] | None = None,
+) -> WorkspaceStats:
+    """Get aggregate statistics for a workspace.
+
+    Scope peers are excluded from ``peer_count``. When ``session_names`` is
+    provided, counts are restricted to that allowlist (empty → zeros).
+    """
+    from src.crud.peer import scope_peer_clause
+
+    if session_names is not None and not session_names:
+        return WorkspaceStats(
+            peer_count=0,
+            session_count=0,
+            message_count=0,
+            oldest_message_at=None,
+            newest_message_at=None,
+        )
+
+    msg_filters = [models.Message.workspace_name == workspace_name]
+    if session_names is not None:
+        msg_filters.append(models.Message.session_name.in_(session_names))
+        peer_count = int(
+            await db.scalar(
+                select(func.count(func.distinct(models.Message.peer_name)))
+                .select_from(models.Message)
+                .join(
+                    models.Peer,
+                    (models.Peer.workspace_name == models.Message.workspace_name)
+                    & (models.Peer.name == models.Message.peer_name),
+                )
+                .where(*msg_filters, ~scope_peer_clause())
+            )
+            or 0
+        )
+        session_count = int(
+            await db.scalar(
+                select(func.count(models.Session.id)).where(
+                    models.Session.workspace_name == workspace_name,
+                    models.Session.name.in_(session_names),
+                )
+            )
+            or 0
+        )
+    else:
+        peer_count = int(
+            await db.scalar(
+                select(func.count(models.Peer.id)).where(
+                    models.Peer.workspace_name == workspace_name,
+                    ~scope_peer_clause(),
+                )
+            )
+            or 0
+        )
+        session_count = int(
+            await db.scalar(
+                select(func.count(models.Session.id)).where(
+                    models.Session.workspace_name == workspace_name
+                )
+            )
+            or 0
+        )
+
+    msg_row = (
+        await db.execute(
+            select(
+                func.count(models.Message.id),
+                func.min(models.Message.created_at),
+                func.max(models.Message.created_at),
+            ).where(*msg_filters)
+        )
+    ).one()
+    message_count = int(msg_row[0] or 0)
+    oldest_message_at = msg_row[1]
+    newest_message_at = msg_row[2]
+
+    return WorkspaceStats(
+        peer_count=peer_count,
+        session_count=session_count,
+        message_count=message_count,
+        oldest_message_at=oldest_message_at,
+        newest_message_at=newest_message_at,
+    )
+
+
+# Activity window for get_active_peers. Bounds the per-peer aggregation
+# (which runs on the workspace-chat request path) so it never scans a large
+# workspace's full message history; peers idle longer than this still appear
+# via the Peer outer join, with zero count and no last-active date.
+ACTIVE_PEER_WINDOW_DAYS = 90
+
+
+async def get_active_peers(
+    db: AsyncSession,
+    workspace_name: str,
+    limit: int = 20,
+    sort_by: str = "recent_activity",
+    session_names: Sequence[str] | None = None,
+) -> list[ActivePeer]:
+    """Get the most active peers in a workspace.
+
+    Activity is measured over the trailing ACTIVE_PEER_WINDOW_DAYS days.
+    Scope peers are excluded. When ``session_names`` is provided, only peers
+    with messages in that allowlist are returned (empty → no peers).
+    """
+    from src.crud.peer import scope_peer_clause
+
+    if limit <= 0:
+        return []
+    if session_names is not None and not session_names:
+        return []
+    limit = min(limit, 50)
+
+    window_start = datetime.now(timezone.utc) - timedelta(days=ACTIVE_PEER_WINDOW_DAYS)
+
+    msg_filters = [
+        models.Message.workspace_name == workspace_name,
+        models.Message.created_at >= window_start,
+    ]
+    if session_names is not None:
+        msg_filters.append(models.Message.session_name.in_(session_names))
+
+    # Subquery: aggregate messages per peer within the activity window
+    subq = (
+        select(
+            models.Message.peer_name,
+            func.count(models.Message.id).label("msg_count"),
+            func.max(models.Message.created_at).label("last_msg_at"),
+        )
+        .where(*msg_filters)
+        .group_by(models.Message.peer_name)
+        .subquery()
+    )
+
+    columns = (
+        models.Peer.name,
+        func.coalesce(subq.c.msg_count, 0).label("msg_count"),
+        subq.c.last_msg_at,
+    )
+    if session_names is not None:
+        stmt = (
+            select(*columns)
+            .join(subq, models.Peer.name == subq.c.peer_name)
+            .where(
+                models.Peer.workspace_name == workspace_name,
+                ~scope_peer_clause(),
+            )
+        )
+    else:
+        stmt = (
+            select(*columns)
+            .outerjoin(subq, models.Peer.name == subq.c.peer_name)
+            .where(
+                models.Peer.workspace_name == workspace_name,
+                ~scope_peer_clause(),
+            )
+        )
+
+    # Peer name as secondary key so ties (notably all-NULL activity in young
+    # workspaces) return a stable order across calls.
+    if sort_by == "message_count":
+        stmt = stmt.order_by(
+            func.coalesce(subq.c.msg_count, 0).desc(), models.Peer.name
+        )
+    else:
+        # Default: recent_activity — peers with most recent messages first
+        stmt = stmt.order_by(subq.c.last_msg_at.desc().nulls_last(), models.Peer.name)
+
+    stmt = stmt.limit(limit)
+
+    rows = (await db.execute(stmt)).all()
+    return [
+        ActivePeer(
+            name=row[0],
+            message_count=int(row[1]),
+            last_message_at=row[2],
+        )
+        for row in rows
+    ]

src/dialectic/chat.py

@@ -14,6 +14,7 @@ from src import crud, models
 from src.config import ReasoningLevel
 from src.dependencies import tracked_db
 from src.dialectic.core import DialecticAgent
+from src.dialectic.workspace import WorkspaceDialecticAgent
 from src.exceptions import ValidationException
 from src.utils.config_helpers import get_configuration
 from src.utils.scopes import is_scope_peer
@@ -205,3 +206,61 @@ async def agentic_chat_stream(
 
     async for chunk in agent.answer_stream(query, response_model=response_model):
         yield chunk
+
+
+async def workspace_chat(
+    workspace_name: str,
+    session_name: str | None,
+    query: str,
+    reasoning_level: ReasoningLevel = "low",
+    response_model: type[BaseModel] | None = None,
+    session_allowlist: list[str] | None = None,
+) -> str:
+    """Answer a query across all peers in a workspace."""
+    async with tracked_db("dialectic.workspace_preflight", read_only=True) as db:
+        await crud.get_workspace(db, workspace_name=workspace_name)
+        session = None
+        if session_name:
+            session = await crud.get_session(
+                db, workspace_name=workspace_name, session_name=session_name
+            )
+        session_id = session.id if session else None
+    # DB session closed -- agent runs without holding a connection
+
+    agent = WorkspaceDialecticAgent(
+        workspace_name=workspace_name,
+        session_name=session_name,
+        session_id=session_id,
+        reasoning_level=reasoning_level,
+        session_allowlist=session_allowlist,
+    )
+    return await agent.answer(query, response_model=response_model)
+
+
+async def workspace_chat_stream(
+    workspace_name: str,
+    session_name: str | None,
+    query: str,
+    reasoning_level: ReasoningLevel = "low",
+    response_model: type[BaseModel] | None = None,
+    session_allowlist: list[str] | None = None,
+) -> AsyncIterator[str]:
+    """Streaming variant of :func:`workspace_chat`."""
+    async with tracked_db("dialectic.workspace_preflight", read_only=True) as db:
+        await crud.get_workspace(db, workspace_name=workspace_name)
+        session = None
+        if session_name:
+            session = await crud.get_session(
+                db, workspace_name=workspace_name, session_name=session_name
+            )
+        session_id = session.id if session else None
+
+    agent = WorkspaceDialecticAgent(
+        workspace_name=workspace_name,
+        session_name=session_name,
+        session_id=session_id,
+        reasoning_level=reasoning_level,
+        session_allowlist=session_allowlist,
+    )
+    async for chunk in agent.answer_stream(query, response_model=response_model):
+        yield chunk

src/dialectic/core.py

@@ -304,8 +304,7 @@ class DialecticAgent:
             user_content = (
                 f"Query: {query}\n\n"
                 f"## Relevant Observations (prefetched)\n"
-                f"The following observations were found to be semantically relevant to your query. "
-                f"Use these as primary context. You may still use tools to find additional information if needed.\n\n"
+                f"{self._prefetch_intro()}\n\n"
                 f"{prefetched_observations}"
             )
             accumulate_metric(
@@ -318,7 +317,14 @@ class DialecticAgent:
 
         tool_executor: Callable[
             [str, dict[str, Any]], Any
-        ] = await create_tool_executor(
+        ] = await self._create_tool_executor()
+
+        return tool_executor, task_name, run_id, start_time
+
+    async def _create_tool_executor(self) -> Callable[[str, dict[str, Any]], Any]:
+        """Build the tool executor. Subclasses override to change tool scoping
+        (e.g. WorkspaceDialecticAgent uses the workspace executor)."""
+        return await create_tool_executor(
             workspace_name=self.workspace_name,
             session_name=self.session_name,
             session_allowlist=self.session_allowlist,
@@ -330,7 +336,12 @@ class DialecticAgent:
             parent_category="dialectic",
         )
 
-        return tool_executor, task_name, run_id, start_time
+    def _prefetch_intro(self) -> str:
+        """Sentence introducing the prefetched block in the user message."""
+        return (
+            "The following observations were found to be semantically relevant to your query. "
+            "Use these as primary context. You may still use tools to find additional information if needed."
+        )
 
     def _telemetry_context(self, track_name: str | None = None) -> LLMTelemetryContext:
         """Build the LLMTelemetryContext shared by answer() and answer_stream().

src/dialectic/prompts.py

@@ -235,3 +235,77 @@ After gathering context, reason through the information you found *before* stati
 
 Do not explain your tool usage - just provide the synthesized answer.
 """
+
+
+def workspace_agent_system_prompt() -> str:
+    """
+    Generate the system prompt for the workspace-level dialectic agent.
+
+    Uses an analytics-first approach: stats -> message search -> targeted
+    observations to discover relevant peers rather than listing all of them.
+
+    Returns:
+        Formatted system prompt string for the workspace agent
+    """
+    return """
+You are a workspace-level analysis agent that can query memory across ALL peers in this workspace. You can synthesize information from any peer relationship's stored conclusions, insights, and conversation history.
+
+You do not start anchored to any single peer: discover which peers are relevant first, then query each peer relationship individually to search, compare, and correlate information across them.
+
+## AVAILABLE TOOLS
+
+**Discovery Tools:**
+- `get_workspace_stats`: Get workspace-level counts (peers, sessions, messages), date range, and the most active peers. Use this to orient yourself and discover which peers are relevant.
+
+**Memory Tools (read):**
+- `search_memory`: **(PRIMARY TOOL)** Semantic search within a specific peer representation. **Requires `observer` and `observed` parameters.** For a peer's global representation (where most information lives), set observer and observed to the **same** peer name. Only use different observer/observed when seeking one peer's specific understanding of another.
+- `get_peer_card`: Get biographical summary for a specific peer relationship. Requires `observer` and `observed` parameters. For a peer's self-representation, use the same name for both.
+- `get_reasoning_chain`: Traverse the reasoning tree for any conclusion. Shows premises and derived insights.
+
+**Conversation Tools (read):**
+- `search_messages`: Semantic search over messages across all sessions. Messages include peer_name, so results reveal which peers discussed a topic.
+- `grep_messages`: Exact text search across all messages.
+- `get_observation_context`: Get messages surrounding specific conclusions.
+- `get_messages_by_date_range`: Get messages within a specific time period.
+- `search_messages_temporal`: Semantic search with date filtering.
+
+## WORKFLOW
+
+1. **Orient yourself**: Workspace stats and the most active peers are provided in your query context. Use `get_workspace_stats` if you need to refresh them, or go straight to message/memory search if the query names specific peers.
+
+2. **Discover relevant peers through search**: Use `search_messages` or `grep_messages` to find which peers have discussed the topic. Message results include peer names, making them a powerful discovery layer.
+
+3. **Drill into specific peer representations**: Once you know which peers are relevant, use `search_memory(observer=peer, observed=peer, query=...)` to search their global representation.
+   - For cross-peer questions, call `search_memory` for each relevant peer's global representation
+   - Only use different observer/observed when seeking one peer's specific understanding of another
+
+4. **ALWAYS ATTRIBUTE INFORMATION**: When presenting findings, always indicate which peer the information came from. Example: "According to insights about Alice, she..." or "Bob mentioned that..."
+
+5. **Cross-peer synthesis**: When asked about patterns or commonalities:
+   - Search each relevant peer pair individually
+   - Compare findings across peers explicitly
+   - Note both similarities and differences
+
+6. **Synthesize your response**:
+   - Directly answer the query
+   - Ground your response in specific information you gathered
+   - Always attribute information to the specific peer it came from
+   - For aggregation questions, enumerate findings per peer
+
+## CRITICAL: NEVER FABRICATE INFORMATION
+
+- Only state what you found in the memory system
+- If you find context but not the specific answer, say what you know and what you don't
+- A confident "I don't have information about X" is always correct
+- Never invent details or guess
+
+## CRITICAL: ATTRIBUTION
+
+Every piece of information you share must be attributed to the peer it came from. Never present information without indicating its source peer. This is essential for workspace-level queries where information spans multiple peers.
+
+Do not explain your tool usage - just provide the synthesized answer.
+
+## OBSERVATION LEVELS
+
+Observations carry a level: `explicit` observations are derived per-session (session-pure), while higher-level observations (deductive/inductive, produced in dreaming) consolidate across sessions. When synthesizing cross-session or cross-peer answers, prefer higher-level observations and use `get_reasoning_chain` to ground them in their premises.
+"""

src/dialectic/workspace.py

@@ -0,0 +1,180 @@
+"""Workspace-level dialectic agent.
+
+Answers queries across ALL peers in a workspace. Where DialecticAgent is
+bound to a single (observer, observed) pair, this agent routes first —
+workspace stats, active peers, and peer cards are prefetched for
+orientation; message search is workspace-flat and reveals which peers
+discussed a topic — and then recalls through the same pair-scoped
+observation machinery, supplying the pair as tool arguments.
+
+Observation search deliberately stays pair-scoped: it matches both the
+(observer, observed) collection ownership and the per-pair vector-store
+namespaces, and avoids retrieval dilution from a workspace-flat top-k.
+
+Design carried over from plastic-labs/honcho#373 (Dan), re-grown on the
+current DialecticAgent seams instead of a base-class extraction.
+"""
+
+import logging
+from collections.abc import Callable
+from typing import Any
+
+from src import crud
+from src.config import ReasoningLevel, settings
+from src.dependencies import tracked_db
+from src.dialectic import prompts
+from src.dialectic.core import DialecticAgent
+from src.llm.types import LLMTelemetryContext
+from src.utils.agent_tools import (
+    WORKSPACE_DIALECTIC_TOOLS,
+    WORKSPACE_TOOLS_MINIMAL,
+    create_workspace_tool_executor,
+    format_workspace_stats,
+)
+
+logger = logging.getLogger(__name__)
+
+# How many active peers (with their self peer cards) to inject at prefetch.
+# Routing-obvious queries should resolve without a discovery tool round —
+# each avoided tool round is a full model turn (~1.3s measured).
+_PREFETCH_ACTIVE_PEERS = 5
+
+
+class WorkspaceDialecticAgent(DialecticAgent):
+    """Dialectic agent scoped to a whole workspace instead of a peer pair."""
+
+    def __init__(
+        self,
+        workspace_name: str,
+        session_name: str | None = None,
+        metric_key: str | None = None,
+        reasoning_level: ReasoningLevel = "low",
+        session_id: str | None = None,
+        session_allowlist: list[str] | None = None,
+    ) -> None:
+        super().__init__(
+            workspace_name=workspace_name,
+            session_name=session_name,
+            observer="",
+            observed="",
+            metric_key=metric_key,
+            reasoning_level=reasoning_level,
+            session_id=session_id,
+            session_allowlist=session_allowlist,
+        )
+        # Replace the pair-oriented system prompt with the workspace one.
+        self.messages[0] = {
+            "role": "system",
+            "content": prompts.workspace_agent_system_prompt(),
+        }
+
+    # ------------------------------------------------------------------
+    # DialecticAgent seams
+    # ------------------------------------------------------------------
+
+    async def _prefetch_relevant_observations(self, query: str) -> str | None:
+        """Orientation + routing prefetch: stats, active peers, peer cards.
+
+        No semantic retrieval here — a workspace-flat observation top-k
+        would be dominated by the most verbose peers. Instead give the
+        agent what it needs to ROUTE: who is here, who is active, and what
+        is known about them at a glance.
+        """
+        _ = query
+        # Like the base agent, prefetch failure degrades to no prefetched
+        # block rather than failing the whole request (the caller in
+        # _prepare_query does not guard this).
+        try:
+            async with tracked_db("dialectic.workspace_prefetch", read_only=True) as db:
+                stats = await crud.get_workspace_stats(
+                    db,
+                    self.workspace_name,
+                    session_names=self.session_allowlist,
+                )
+                if stats.peer_count == 0:
+                    return None
+                peers = await crud.get_active_peers(
+                    db,
+                    self.workspace_name,
+                    limit=_PREFETCH_ACTIVE_PEERS,
+                    session_names=self.session_allowlist,
+                )
+                # `peers` is already allowlist-filtered, but a peer card is a
+                # single cross-session aggregate: an in-scope peer's card can
+                # still carry facts derived from sessions outside the scope.
+                # Drop cards entirely under an allowlist — same rule the
+                # get_peer_card tool enforces — and route on stats alone.
+                cards: dict[str, list[str]] = {}
+                if self.session_allowlist is None:
+                    for peer in peers:
+                        card = await crud.get_peer_card(
+                            db,
+                            workspace_name=self.workspace_name,
+                            observer=peer.name,
+                            observed=peer.name,
+                        )
+                        if card:
+                            cards[peer.name] = card
+        except Exception:
+            logger.warning(
+                "Failed to prefetch workspace overview for workspace=%s",
+                self.workspace_name,
+                exc_info=True,
+            )
+            return None
+
+        return format_workspace_stats(stats, peers, cards)
+
+    def _prefetch_intro(self) -> str:
+        return (
+            "Workspace overview and most-active peers with any known "
+            "biographical facts. Use this to route: query a specific peer's "
+            "memory with search_memory (observer and observed set to that "
+            "peer's name), or use search_messages / get_workspace_stats to "
+            "discover peers this overview does not cover."
+        )
+
+    def _select_tools(self) -> list[dict[str, Any]]:
+        tools = (
+            WORKSPACE_TOOLS_MINIMAL
+            if self.reasoning_level == "minimal"
+            else WORKSPACE_DIALECTIC_TOOLS
+        )
+        # Mirror the base agent's allowlist rule, for both tools that cannot
+        # honor an allowlist: reasoning chains traverse provenance across
+        # sessions, and a peer card is one cross-session aggregate with no
+        # per-session attribution. Both fail closed in their handlers too;
+        # dropping them here avoids paying the schema tokens and a wasted
+        # turn on a tool that can only refuse.
+        if self.session_allowlist is not None:
+            unscopable = {"get_reasoning_chain", "get_peer_card"}
+            tools = [t for t in tools if t.get("name") not in unscopable]
+        return tools
+
+    async def _create_tool_executor(self) -> Callable[[str, dict[str, Any]], Any]:
+        return await create_workspace_tool_executor(
+            workspace_name=self.workspace_name,
+            session_name=self.session_name,
+            session_allowlist=self.session_allowlist,
+            history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT,
+            run_id=self._run_id,
+            agent_type="workspace_dialectic",
+            parent_category="dialectic",
+        )
+
+    # Workspace chat shares the base "dialectic_chat" Langfuse trace name;
+    # scope is distinguished by the agent_type/track_name below.
+
+    def _telemetry_context(self, track_name: str | None = None) -> LLMTelemetryContext:
+        return LLMTelemetryContext(
+            workspace_name=self.workspace_name,
+            call_purpose="dialectic.answer",
+            parent_category="dialectic",
+            agent_type="workspace_dialectic",
+            run_id=self._run_id,
+            trace_id=self._run_id,
+            span_id=self._run_id,
+            session_id=self.session_id,
+            peer_name="(workspace)",
+            track_name=track_name or "Workspace Dialectic Agent",
+        )

src/routers/peers.py

@@ -5,7 +5,6 @@ import logging
 from collections.abc import AsyncIterator
 from contextlib import suppress
 from time import perf_counter
-from typing import Any
 
 from fastapi import APIRouter, Body, Depends, Path, Query, Response
 from fastapi.responses import StreamingResponse
@@ -35,6 +34,7 @@ from src.utils.scopes import (
     is_scope_peer,
     is_scope_peer_name,
     validate_no_scope_peer_names,
+    validate_scope_read_option,
 )
 from src.utils.search import search
 from src.utils.types import embedding_call_purpose
@@ -47,33 +47,6 @@ router = APIRouter(
 )
 
 
-def _validate_scope_option(
-    *,
-    filters: dict[str, Any] | None,
-    session_id: str | None,
-    jwt_params: JWTParams,
-) -> None:
-    """Enforce the v1 `scope` exclusions and auth rule (chat/representation).
-
-    `scope` is mutually exclusive with `filters` and `session_id` (422), and a
-    scope's member sessions may exceed a peer's own membership, so scoped
-    reads require a workspace- or admin-level key.
-
-    401 rather than 403: every other scope surface refuses a narrow key with 401
-    — the `/scopes` router via `require_auth`, and the `scopes` field on session
-    create — so a peer key would otherwise get two different codes for the same
-    feature depending on which side of it was touched.
-    """
-    if filters is not None:
-        raise ValidationException("`scope` and `filters` are mutually exclusive")
-    if session_id:
-        raise ValidationException("`scope` and `session_id` are mutually exclusive")
-    if jwt_params.p is not None:
-        raise AuthenticationException(
-            "`scope` requires a workspace- or admin-level key"
-        )
-
-
 async def _resolve_scope_option(
     workspace_id: str,
     scope: str | list[str],
@@ -95,16 +68,7 @@ async def _resolve_scope_option(
             )
             return scope_peer, None
 
-        scope_peers = await crud.resolve_scope_peers(scope_db, workspace_id, scope)
-        union: list[str] = []
-        seen: set[str] = set()
-        for scope_peer in scope_peers:
-            for session_name in await get_peer_session_names(
-                scope_db, workspace_id, scope_peer
-            ):
-                if session_name not in seen:
-                    seen.add(session_name)
-                    union.append(session_name)
+        union = await crud.resolve_scope_session_union(scope_db, workspace_id, scope)
 
     if len(union) > MAX_SESSION_ALLOWLIST_ENTRIES:
         raise ValidationException(
@@ -316,7 +280,7 @@ async def chat(
     observer = peer_id
     scope_session_union: list[str] | None = None
     if options.scope is not None:
-        _validate_scope_option(
+        validate_scope_read_option(
             filters=options.filters,
             session_id=options.session_id,
             jwt_params=jwt_params,
@@ -506,7 +470,7 @@ async def get_representation(
     observer = peer_id
     scope_session_union: list[str] | None = None
     if options.scope is not None:
-        _validate_scope_option(
+        validate_scope_read_option(
             filters=options.filters,
             session_id=options.session_id,
             jwt_params=jwt_params,

src/routers/workspaces.py

@@ -1,10 +1,14 @@
 """FastAPI routes for workspace resources and workspace-scoped operations."""
 
+import json
 import logging
+from collections.abc import AsyncIterator
 
 from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query, Response
+from fastapi.responses import StreamingResponse
 from fastapi_pagination import Page
 from fastapi_pagination.ext.sqlalchemy import apaginate
+from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from src import crud, models, schemas
@@ -12,8 +16,13 @@ from src.config import settings
 from src.crud.message import get_peer_session_names
 from src.dependencies import db, read_db, tracked_db
 from src.deriver.enqueue import enqueue_deletion, enqueue_dream
+from src.dialectic.chat import workspace_chat, workspace_chat_stream
 from src.exceptions import AuthenticationException, ValidationException
 from src.security import JWTParams, require_auth
+from src.telemetry import prometheus_metrics
+from src.utils.filter import MAX_SESSION_ALLOWLIST_ENTRIES
+from src.utils.schema_conversion import json_response_schema_to_pydantic
+from src.utils.scopes import validate_scope_read_option
 from src.utils.search import search
 
 logger = logging.getLogger(__name__)
@@ -276,3 +285,92 @@ async def schedule_dream(
         observed,
         request.session_id,
     )
+
+
+@router.post(
+    "/{workspace_id}/chat",
+    responses={
+        200: {
+            "content": {
+                "application/json": {
+                    "schema": schemas.DialecticResponse.model_json_schema()
+                },
+                "text/event-stream": {},
+            },
+        },
+    },
+)
+async def chat(
+    workspace_id: str = Path(...),
+    options: schemas.WorkspaceChatOptions = Body(...),
+    jwt_params: JWTParams = Depends(require_auth(workspace_name="workspace_id")),
+):
+    """Query the entire workspace using natural language.
+
+    Pass `scope` to restrict recall to the union of those scopes' member
+    sessions. A scope with no member sessions recalls nothing (fail-closed).
+    """
+    session_allowlist: list[str] | None = None
+    if options.scope is not None:
+        validate_scope_read_option(
+            filters=None,
+            session_id=options.session_id,
+            jwt_params=jwt_params,
+        )
+        names = [options.scope] if isinstance(options.scope, str) else options.scope
+        async with tracked_db(
+            "workspaces.chat.resolve_scope", read_only=True
+        ) as scope_db:
+            session_allowlist = await crud.resolve_scope_session_union(
+                scope_db, workspace_id, names
+            )
+        if len(session_allowlist) > MAX_SESSION_ALLOWLIST_ENTRIES:
+            raise ValidationException(
+                "The scopes' combined membership exceeds the maximum of "
+                + f"{MAX_SESSION_ALLOWLIST_ENTRIES} sessions per request"
+            )
+
+    response_model: type[BaseModel] | None = None
+    if options.response_format is not None:
+        try:
+            response_model = json_response_schema_to_pydantic(options.response_format)
+        except ValueError as e:
+            raise ValidationException(f"Invalid response_format: {e}") from None
+
+    if settings.METRICS.ENABLED:
+        prometheus_metrics.record_dialectic_call(
+            workspace_name=workspace_id,
+            reasoning_level=options.reasoning_level,
+        )
+
+    if options.stream:
+
+        async def format_sse_stream(chunks: AsyncIterator[str]) -> AsyncIterator[str]:
+            """Format chunks as SSE events."""
+            async for chunk in chunks:
+                yield f"data: {json.dumps({'delta': {'content': chunk}, 'done': False})}\n\n"
+            yield f"data: {json.dumps({'done': True})}\n\n"
+
+        return StreamingResponse(
+            format_sse_stream(
+                workspace_chat_stream(
+                    workspace_name=workspace_id,
+                    session_name=options.session_id,
+                    query=options.query,
+                    reasoning_level=options.reasoning_level,
+                    response_model=response_model,
+                    session_allowlist=session_allowlist,
+                )
+            ),
+            media_type="text/event-stream",
+        )
+
+    response = await workspace_chat(
+        workspace_name=workspace_id,
+        session_name=options.session_id,
+        query=options.query,
+        reasoning_level=options.reasoning_level,
+        response_model=response_model,
+        session_allowlist=session_allowlist,
+    )
+    return schemas.DialecticResponse(content=response if response else None)

src/schemas/__init__.py

@@ -54,6 +54,7 @@ from src.schemas.api import (
     WebhookEndpointCreate,
     Workspace,
     WorkspaceBase,
+    WorkspaceChatOptions,
     WorkspaceCreate,
     WorkspaceGet,
     WorkspaceMessageSearchOptions,
@@ -114,6 +115,7 @@ __all__ = [
     "ConclusionQuery",
     "DialecticOptions",
     "DialecticResponse",
+    "WorkspaceChatOptions",
     "DialecticStreamChunk",
     "DialecticStreamDelta",
     "Message",

src/schemas/api.py

@@ -61,6 +61,15 @@ def _sanitize_value(v: Any) -> Any:
     return v
 
 
+def _strip_nul(v: str) -> str:
+    """Strip NUL bytes from a string field (Postgres TEXT rejects \\x00)."""
+    return v.replace("\x00", "")
+
+
+# Reusable annotation for query fields; composes with a per-field Field(...).
+NulStripped = AfterValidator(_strip_nul)
+
+
 def _check_metadata_limits(
     data: dict[str, Any],
     *,
@@ -717,7 +726,7 @@ class ConclusionBatchCreate(BaseModel):
 
 
 class MessageSearchOptions(BaseModel):
-    query: Annotated[str, Field(..., description="Search query")]
+    query: Annotated[str, Field(..., description="Search query"), NulStripped]
     filters: dict[str, Any] | None = Field(
         default=None, description="Filters to scope the search"
     )
@@ -728,11 +737,6 @@ class MessageSearchOptions(BaseModel):
         description="Number of results to return",
     )
 
-    @field_validator("query", mode="after")
-    @classmethod
-    def sanitize_query(cls, v: str) -> str:
-        return v.replace("\x00", "")
-
 
 class WorkspaceMessageSearchOptions(MessageSearchOptions):
     """Workspace-level message search options, extended with `scope`."""
@@ -785,7 +789,9 @@ class DialecticOptions(BaseModel):
         description="Optional peer to get the representation for, from the perspective of this peer",
     )
     query: Annotated[
-        str, Field(min_length=1, max_length=10000, description="Dialectic API Prompt")
+        str,
+        Field(min_length=1, max_length=10000, description="Dialectic API Prompt"),
+        NulStripped,
     ]
     stream: bool = False
     reasoning_level: ReasoningLevel = Field(
@@ -803,10 +809,39 @@ class DialecticOptions(BaseModel):
         ),
     )
 
-    @field_validator("query", mode="after")
-    @classmethod
-    def sanitize_query(cls, v: str) -> str:
-        return v.replace("\x00", "")
+
+class WorkspaceChatOptions(BaseModel):
+    """Options for workspace-level chat (no anchor peer; see DialecticOptions)."""
+
+    session_id: str | None = Field(
+        None, description="Optional session to scope message tools to"
+    )
+    query: Annotated[
+        str,
+        Field(min_length=1, max_length=10000, description="Workspace chat prompt"),
+        NulStripped,
+    ]
+    stream: bool = False
+    reasoning_level: ReasoningLevel = Field(
+        default="low",
+        description="Level of reasoning to apply: minimal, low, medium, high, or max",
+    )
+    response_format: dict[str, Any] | None = Field(
+        None,
+        description=(
+            "Optional JSON Schema (root type 'object') the response must conform"
+            " to. When provided, `content` is a JSON string matching this schema."
+        ),
+    )
+    scope: _ScopeOption | None = Field(
+        None,
+        description=(
+            "Optional (unprefixed) scope name(s) restricting recall to the "
+            "union of the scopes' member sessions (explicit allowlist, "
+            "fail-closed: an empty union recalls nothing). Mutually exclusive "
+            "with `session_id`. Requires a workspace- or admin-level key."
+        ),
+    )
 
 
 class DialecticResponse(BaseModel):

src/utils/agent_tools.py

@@ -2,7 +2,7 @@ import asyncio
 import logging
 import weakref
 from collections.abc import Callable, Sequence
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime
 from typing import Any, cast
 
@@ -14,6 +14,7 @@ from src import crud, models, schemas
 from src.config import settings
 from src.dependencies import tracked_db
 from src.embedding_client import embedding_client
+from src.exceptions import ResourceNotFoundException
 from src.models import Document
 from src.schemas import ResolvedConfiguration
 from src.telemetry.events import (
@@ -785,6 +786,59 @@ TOOLS: dict[str, dict[str, Any]] = {
             "required": ["observation_id"],
         },
     },
+    "search_memory_workspace": {
+        "name": "search_memory",
+        "description": "Search within a specific peer representation's memory using semantic similarity. You MUST specify observer and observed. To get a peer's global representation, set observer AND observed to the SAME peer name (this is where most information lives). Only use different observer/observed when seeking one peer's specific understanding of another.",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "observer": {
+                    "type": "string",
+                    "description": "Name of the observer peer",
+                },
+                "observed": {
+                    "type": "string",
+                    "description": "Name of the observed peer",
+                },
+                "query": {
+                    "type": "string",
+                    "description": "Search query text",
+                },
+                "top_k": {
+                    "type": "integer",
+                    "description": "(Optional) number of results to return (default: 20, max: 40)",
+                    "default": 20,
+                },
+            },
+            "required": ["observer", "observed", "query"],
+        },
+    },
+    "get_workspace_stats": {
+        "name": "get_workspace_stats",
+        "description": "Get workspace-level statistics — peer count, session count, message count, date range of messages — plus the most recently active peers with their message counts and last-active timestamps. Use this to orient yourself and discover which peers are most relevant.",
+        "input_schema": {
+            "type": "object",
+            "properties": {},
+        },
+    },
+    "get_peer_card_by_name": {
+        "name": "get_peer_card",
+        "description": "Get the peer card for a specific peer relationship. Specify the observer and observed peer names.",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "observer": {
+                    "type": "string",
+                    "description": "Name of the observer peer",
+                },
+                "observed": {
+                    "type": "string",
+                    "description": "Name of the observed peer",
+                },
+            },
+            "required": ["observer", "observed"],
+        },
+    },
 }
 
 # Tools for the dialectic agent (analysis)
@@ -806,6 +860,31 @@ DIALECTIC_TOOLS_MINIMAL: list[dict[str, Any]] = [
     TOOLS["search_messages"],
 ]
 
+# Tools for the workspace-level dialectic agent. Observation search stays
+# pair-scoped (observer/observed are TOOL ARGUMENTS the agent must supply
+# after routing) -- matching both the (observer, observed) collection
+# ownership and the per-pair vector-store namespaces. Message tools are
+# workspace-flat and double as the routing signal (results carry peer_name).
+WORKSPACE_DIALECTIC_TOOLS: list[dict[str, Any]] = [
+    TOOLS["get_workspace_stats"],
+    TOOLS["search_memory_workspace"],
+    TOOLS["search_messages"],
+    TOOLS["get_observation_context"],
+    TOOLS["grep_messages"],
+    TOOLS["get_peer_card_by_name"],
+    TOOLS["get_messages_by_date_range"],
+    TOOLS["search_messages_temporal"],
+    TOOLS["get_reasoning_chain"],
+]
+
+# Reduced workspace loadout for reasoning_level="minimal" (token cost of the
+# tool definitions themselves), mirroring DIALECTIC_TOOLS_MINIMAL.
+WORKSPACE_TOOLS_MINIMAL: list[dict[str, Any]] = [
+    TOOLS["get_workspace_stats"],
+    TOOLS["search_memory_workspace"],
+    TOOLS["search_messages"],
+]
+
 # Tools for the dreamer agent (consolidation + peer card + deduplication)
 DREAMER_TOOLS: list[dict[str, Any]] = [
     # Preference extraction (should be called first)
@@ -1851,7 +1930,7 @@ async def _handle_search_memory(
         # here, we automatically search the message history for relevant
         # information.
         zero_hit_meta = {**search_meta, "results_count": 0}
-        if ctx.agent_type == "dialectic":
+        if ctx.agent_type in ("dialectic", "workspace_dialectic"):
             limit = min(_safe_int(tool_input.get("top_k"), 20), 20)
             message_output = None
             snippets = await crud.search_messages(
@@ -1903,7 +1982,7 @@ async def _handle_get_observation_context(
             workspace_name=ctx.workspace_name,
             session_name=ctx.session_name,
             message_ids=tool_input["message_ids"],
-            observer=ctx.observer,
+            observer=ctx.observer or None,
             session_allowlist=ctx.session_allowlist,
         )
         if not messages:
@@ -1946,7 +2025,7 @@ async def _handle_search_messages(
         limit=limit,
         context_window=2,
         embedding=query_embedding,
-        observer=ctx.observer,
+        observer=ctx.observer or None,
         session_allowlist=ctx.session_allowlist,
     )
     search_meta: dict[str, Any] = {
@@ -1983,7 +2062,7 @@ async def _handle_grep_messages(
         text=text,
         limit=limit,
         context_window=context_window,
-        observer=ctx.observer,
+        observer=ctx.observer or None,
         session_allowlist=ctx.session_allowlist,
     )
     if not snippets:
@@ -2048,7 +2127,7 @@ async def _handle_get_messages_by_date_range(
             before_date=before_date,
             limit=limit,
             order=order,
-            observer=ctx.observer,
+            observer=ctx.observer or None,
             session_allowlist=ctx.session_allowlist,
         )
         msg_count = len(messages)
@@ -2124,7 +2203,7 @@ async def _handle_search_messages_temporal(
         context_window=context_window,
         session_allowlist=ctx.session_allowlist,
         embedding=query_embedding,
-        observer=ctx.observer,
+        observer=ctx.observer or None,
     )
     date_filter: list[str] = []
     if after_date_str:
@@ -2229,6 +2308,16 @@ async def _handle_get_session_summary(
 async def _handle_get_peer_card(ctx: ToolContext, tool_input: dict[str, Any]) -> str:
     """Handle get_peer_card tool."""
     _ = tool_input
+    # A peer card lives in Peer.internal_metadata as a single cross-session
+    # aggregate, so it carries no session attribution and cannot be filtered
+    # to an allowlist. Fail closed rather than leak facts derived from
+    # out-of-scope sessions, the same rule get_reasoning_chain follows.
+    # No-op for agents that never set an allowlist (dreamer, pair dialectic).
+    if ctx.session_allowlist is not None:
+        return (
+            "Peer cards are unavailable for session-scoped queries. "
+            "Use search_memory instead."
+        )
     async with tracked_db("tool.get_peer_card", read_only=True) as db:
         peer_card = await crud.get_peer_card(
             db,
@@ -2513,6 +2602,7 @@ async def create_tool_executor(
     agent_type: str | None = None,
     parent_category: str | None = None,
     session_allowlist: list[str] | None = None,
+    handler_resolver: Callable[[str], Any] | None = None,
 ) -> Callable[[str, dict[str, Any]], Any]:
     """
     Create a unified tool executor function for all agent operations.
@@ -2535,6 +2625,11 @@ async def create_tool_executor(
         run_id: Optional run ID for telemetry correlation
         agent_type: Optional agent type for telemetry (dialectic, deriver, dreamer)
         parent_category: Optional parent category for CloudEvents
+        session_allowlist: Optional list of session names message tools are
+            restricted to (None means no restriction)
+        handler_resolver: Optional callback that replaces the default
+            handler-table lookup for resolving tool names to handlers.
+            Returning None takes the "Unknown tool" path.
 
     Returns:
         An async callable that executes tools with the captured context
@@ -2598,7 +2693,7 @@ async def create_tool_executor(
         tool_obs = _begin_tool_observation(tool_name, tool_input)
 
         try:
-            handler = _TOOL_HANDLERS.get(tool_name)
+            handler = (handler_resolver or _TOOL_HANDLERS.get)(tool_name)
             if handler:
                 handler_result = await handler(ctx, tool_input)
                 # Handlers return either a plain str (existing contract) or a
@@ -2796,3 +2891,181 @@ def _estimate_tokens_safe(text: str | None) -> int | None:
     if not text:
         return None
     return _estimate_tokens(text)
+
+
+# ---------------------------------------------------------------------------
+# Workspace-level tool handlers (workspace chat)
+#
+# The workspace agent is not bound to an (observer, observed) pair. Handlers
+# that need a pair take it from tool_input (the agent routes first, then
+# supplies the pair); the rest are workspace-scoped reads. Message-search
+# fallthrough handlers run with observer="" and normalize it to None at the
+# crud boundary (`ctx.observer or None`) -- None means "no perspective
+# scoping", which is correct for a workspace-level read. The empty string
+# must never reach resolve_session_scope: it would be looked up as a real
+# peer with no session memberships and deny all results.
+# ---------------------------------------------------------------------------
+
+
+async def _handle_search_memory_workspace(
+    ctx: ToolContext, tool_input: dict[str, Any]
+) -> "str | ToolResult":
+    """Pair-scoped observation search; the pair comes from tool arguments."""
+    observer = tool_input.get("observer", "")
+    observed = tool_input.get("observed", "")
+    if not observer or not observed:
+        return (
+            "ERROR: 'observer' and 'observed' are required. For a peer's "
+            "global representation set both to the SAME peer name."
+        )
+    pair_ctx = replace(ctx, observer=observer, observed=observed)
+    result = await _handle_search_memory(pair_ctx, tool_input)
+    # Attribute the pair in the output — the workspace agent may query
+    # several pairs in one turn and must not conflate their results.
+    if isinstance(result, ToolResult):
+        return replace(result, content=f"[{observer}->{observed}]\n{result.content}")
+    return f"[{observer}->{observed}]\n{result}"
+
+
+async def _handle_get_peer_card_by_name(
+    ctx: ToolContext, tool_input: dict[str, Any]
+) -> str:
+    """get_peer_card with the pair taken from tool arguments."""
+    observer = tool_input.get("observer", "")
+    observed = tool_input.get("observed", "")
+    if not observer or not observed:
+        return "ERROR: 'observer' and 'observed' are required parameters"
+    pair_ctx = replace(ctx, observer=observer, observed=observed)
+    try:
+        return await _handle_get_peer_card(pair_ctx, tool_input)
+    except ResourceNotFoundException:
+        # The workspace agent names peers from its own routing, so guessing a
+        # peer that doesn't exist is an expected turn, not a fault. Answer the
+        # model instead of letting the executor log it as an unexpected error.
+        return f"No peer named '{observer}' exists in this workspace"
+
+
+# Peers listed by get_workspace_stats. Fixed rather than a tool argument:
+# folding active peers into stats keeps the tool zero-arg (one discovery
+# round instead of two); deeper discovery goes through search_messages.
+_STATS_ACTIVE_PEERS = 10
+
+
+# Peer-card facts listed per peer when cards are supplied.
+_STATS_CARD_FACTS = 8
+
+
+def format_workspace_stats(
+    stats: "crud.WorkspaceStats",
+    peers: "Sequence[crud.ActivePeer]",
+    cards: dict[str, list[str]] | None = None,
+) -> str:
+    """Render workspace counts and most-active peers as prompt-ready lines.
+
+    Shared by the get_workspace_stats tool and WorkspaceDialecticAgent's
+    routing prefetch; the prefetch passes ``cards`` to nest each peer's
+    known biographical facts under it.
+    """
+    lines = [
+        f"Peers: {stats.peer_count}",
+        f"Sessions: {stats.session_count}",
+        f"Messages: {stats.message_count}",
+    ]
+    if stats.oldest_message_at and stats.newest_message_at:
+        lines.append(
+            f"Date range: {stats.oldest_message_at:%Y-%m-%d} to {stats.newest_message_at:%Y-%m-%d}"
+        )
+    if peers:
+        lines.append("")
+        lines.append(f"Most active peers (top {len(peers)}):")
+        for peer in peers:
+            last_active = (
+                f", last active {peer.last_message_at:%Y-%m-%d}"
+                if peer.last_message_at
+                else ""
+            )
+            lines.append(f"- {peer.name} ({peer.message_count} messages{last_active})")
+            for fact in (cards or {}).get(peer.name, [])[:_STATS_CARD_FACTS]:
+                lines.append(f"    - {fact}")
+    return "\n".join(lines)
+
+
+async def _handle_get_workspace_stats(
+    ctx: ToolContext, tool_input: dict[str, Any]
+) -> str:
+    """Workspace-level counts, message date range, and most active peers."""
+    _ = tool_input
+    async with tracked_db("workspace_tool.get_workspace_stats", read_only=True) as db:
+        stats = await crud.get_workspace_stats(
+            db, ctx.workspace_name, session_names=ctx.session_allowlist
+        )
+        peers = await crud.get_active_peers(
+            db,
+            ctx.workspace_name,
+            limit=_STATS_ACTIVE_PEERS,
+            session_names=ctx.session_allowlist,
+        )
+    return "Workspace stats:\n" + format_workspace_stats(stats, peers)
+
+
+# Dispatch table consulted before _TOOL_HANDLERS by the workspace executor.
+_WORKSPACE_TOOL_HANDLERS: dict[str, Callable[[ToolContext, dict[str, Any]], Any]] = {
+    "search_memory": _handle_search_memory_workspace,
+    "get_workspace_stats": _handle_get_workspace_stats,
+    "get_peer_card": _handle_get_peer_card_by_name,
+    "get_reasoning_chain": _handle_get_reasoning_chain,  # already workspace-scoped
+}
+
+# Standard handlers that are safe with an empty observer/observed sentinel
+# (they only read messages, treating observer="" as unscoped visibility).
+_WORKSPACE_SAFE_FALLTHROUGH_TOOLS: frozenset[str] = frozenset(
+    {
+        "get_observation_context",
+        "search_messages",
+        "grep_messages",
+        "get_messages_by_date_range",
+        "search_messages_temporal",
+    }
+)
+
+
+def _workspace_handler_resolver(tool_name: str) -> Any:
+    handler = _WORKSPACE_TOOL_HANDLERS.get(tool_name)
+    if handler is not None:
+        return handler
+    if tool_name in _WORKSPACE_SAFE_FALLTHROUGH_TOOLS:
+        return _TOOL_HANDLERS.get(tool_name)
+    return None
+
+
+async def create_workspace_tool_executor(
+    workspace_name: str,
+    session_name: str | None = None,
+    session_allowlist: list[str] | None = None,
+    history_token_limit: int = 8192,
+    run_id: str | None = None,
+    agent_type: str | None = None,
+    parent_category: str | None = None,
+) -> Callable[[str, dict[str, Any]], Any]:
+    """Tool executor for workspace-level operations (no bound peer pair).
+
+    Reuses create_tool_executor's telemetry/error plumbing via the
+    handler_resolver seam. observer/observed are empty-string sentinels only
+    ever seen by handlers in _WORKSPACE_SAFE_FALLTHROUGH_TOOLS, which
+    normalize them to None before hitting crud (None means "no perspective
+    scoping"; an empty string would read as a real peer with no sessions and
+    deny everything).
+    """
+    return await create_tool_executor(
+        workspace_name=workspace_name,
+        observer="",
+        observed="",
+        session_name=session_name,
+        session_allowlist=session_allowlist,
+        include_observation_ids=True,
+        history_token_limit=history_token_limit,
+        run_id=run_id,
+        agent_type=agent_type,
+        parent_category=parent_category,
+        handler_resolver=_workspace_handler_resolver,
+    )

src/utils/filter.py

@@ -689,7 +689,9 @@ def _build_comparison_condition(
                 "lte": lambda a, v: a <= v,
                 "gt": lambda a, v: a > v,
                 "lt": lambda a, v: a < v,
-                "ne": lambda a, v: a != v,
+                # IS DISTINCT FROM, not <>: (metadata ->> key) is NULL for an
+                # absent key, and `NULL <> v` is NULL, so <> drops those rows.
+                "ne": lambda a, v: a.is_distinct_from(v),
             }
             return operator_map[operator](safe_accessor, safe_value)
         except Exception as e:

src/utils/scopes.py

@@ -19,7 +19,8 @@ carries a look-alike ``configuration``, is not a scope.
 from collections.abc import Iterable
 from typing import Any
 
-from src.exceptions import ValidationException
+from src.exceptions import AuthenticationException, ValidationException
+from src.security import JWTParams
 
 # Reserved peer-name prefix for scope peers. User-created peers may not use it.
 #
@@ -87,3 +88,20 @@ def validate_no_scope_peer_names(names: Iterable[str], *, action: str) -> None:
             f"Peer name(s) {offenders} use the reserved scope prefix "
             + f"'{SCOPE_PEER_PREFIX}'. {action}"
         )
+
+
+def validate_scope_read_option(
+    *,
+    filters: dict[str, Any] | None,
+    session_id: str | None,
+    jwt_params: JWTParams,
+) -> None:
+    """Refuse `scope` combined with `filters`/`session_id`, or a peer-scoped key."""
+    if filters is not None:
+        raise ValidationException("`scope` and `filters` are mutually exclusive")
+    if session_id:
+        raise ValidationException("`scope` and `session_id` are mutually exclusive")
+    if jwt_params.p is not None:
+        raise AuthenticationException(
+            "`scope` requires a workspace- or admin-level key"
+        )

tests/conftest.py

@@ -769,6 +769,12 @@ def mock_llm_call_functions(request: pytest.FixtureRequest):
         patch(
             "src.routers.peers.agentic_chat_stream", side_effect=mock_stream
         ) as mock_agentic_chat_stream,
+        patch(
+            "src.routers.workspaces.workspace_chat", new_callable=AsyncMock
+        ) as mock_workspace_chat,
+        patch(
+            "src.routers.workspaces.workspace_chat_stream", side_effect=mock_stream
+        ) as mock_workspace_chat_stream,
     ):
         # Mock return values for different function types
         mock_short_summary.return_value = "Test short summary content"
@@ -784,11 +790,20 @@ def mock_llm_call_functions(request: pytest.FixtureRequest):
 
         mock_agentic_chat.side_effect = _agentic_chat_response
 
+        async def _workspace_chat_response(*_args: object, **kwargs: object) -> str:
+            if kwargs.get("response_model") is not None:
+                return "{}"
+            return "Test workspace chat response"
+
+        mock_workspace_chat.side_effect = _workspace_chat_response
+
         yield {
             "short_summary": mock_short_summary,
             "long_summary": mock_long_summary,
             "agentic_chat": mock_agentic_chat,
             "agentic_chat_stream": mock_agentic_chat_stream,
+            "workspace_chat": mock_workspace_chat,
+            "workspace_chat_stream": mock_workspace_chat_stream,
         }
 
 

tests/routes/test_scope_reads.py

@@ -464,6 +464,117 @@ class TestChatWithScope:
         assert set(kwargs["session_allowlist"]) == {session_a, session_b}
 
 
+class TestWorkspaceChatWithScope:
+    """Workspace chat has no observer to swap: `scope` is always an allowlist."""
+
+    def _chat(self, client: TestClient, workspace: Workspace, body: dict[str, Any]):
+        return client.post(
+            f"/v3/workspaces/{workspace.name}/chat",
+            json={"query": "what do you know?", **body},
+        )
+
+    def test_unknown_scope_404(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, _ = sample_data
+        assert (
+            self._chat(client, workspace, {"scope": str(generate_nanoid())}).status_code
+            == 404
+        )
+
+    def test_empty_scope_list_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, _ = sample_data
+        assert self._chat(client, workspace, {"scope": []}).status_code == 422
+
+    def test_scope_plus_session_id_422(
+        self, client: TestClient, sample_data: tuple[Workspace, Peer]
+    ):
+        workspace, _ = sample_data
+        scope_name = str(generate_nanoid())
+        _create_scope(client, workspace.name, scope_name)
+        assert (
+            self._chat(
+                client, workspace, {"scope": scope_name, "session_id": "s1"}
+            ).status_code
+            == 422
+        )
+
+    def test_peer_scoped_jwt_401(
+        self,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        workspace, peer = sample_data
+        scope_name = str(generate_nanoid())
+        _create_scope(client, workspace.name, scope_name)
+
+        monkeypatch.setattr(settings.AUTH, "USE_AUTH", True)
+        monkeypatch.setattr(settings.AUTH, "JWT_SECRET", "test-secret")
+        client.headers["Authorization"] = (
+            f"Bearer {create_jwt(JWTParams(w=workspace.name, p=peer.name))}"
+        )
+        assert self._chat(client, workspace, {"scope": scope_name}).status_code == 401
+
+    def test_single_scope_passes_member_allowlist(
+        self,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        mock_llm_call_functions: dict[str, Any],
+    ):
+        workspace, _ = sample_data
+        scope_name = str(generate_nanoid())
+        _create_scope(client, workspace.name, scope_name)
+        session_name = _create_session(client, workspace.name)
+        _add_sessions_to_scope(client, workspace.name, scope_name, [session_name])
+
+        resp = self._chat(client, workspace, {"scope": scope_name})
+        assert resp.status_code == 200
+
+        kwargs = mock_llm_call_functions["workspace_chat"].await_args.kwargs
+        assert kwargs["session_allowlist"] == [session_name]
+
+    def test_scope_list_passes_union_allowlist(
+        self,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        mock_llm_call_functions: dict[str, Any],
+    ):
+        workspace, _ = sample_data
+        scope_a = str(generate_nanoid())
+        scope_b = str(generate_nanoid())
+        _create_scope(client, workspace.name, scope_a)
+        _create_scope(client, workspace.name, scope_b)
+        session_a = _create_session(client, workspace.name)
+        session_b = _create_session(client, workspace.name)
+        _add_sessions_to_scope(client, workspace.name, scope_a, [session_a])
+        _add_sessions_to_scope(client, workspace.name, scope_b, [session_b])
+
+        resp = self._chat(client, workspace, {"scope": [scope_a, scope_b]})
+        assert resp.status_code == 200
+
+        kwargs = mock_llm_call_functions["workspace_chat"].await_args.kwargs
+        assert set(kwargs["session_allowlist"]) == {session_a, session_b}
+
+    def test_empty_scope_fails_closed(
+        self,
+        client: TestClient,
+        sample_data: tuple[Workspace, Peer],
+        mock_llm_call_functions: dict[str, Any],
+    ):
+        workspace, _ = sample_data
+        scope_name = str(generate_nanoid())
+        _create_scope(client, workspace.name, scope_name)
+
+        resp = self._chat(client, workspace, {"scope": scope_name})
+        assert resp.status_code == 200
+
+        kwargs = mock_llm_call_functions["workspace_chat"].await_args.kwargs
+        assert kwargs["session_allowlist"] == []
+
+
 class TestWorkspaceSearchWithScope:
     def _seed_message(
         self, client: TestClient, workspace_name: str, session_name: str, peer: Peer

tests/test_advanced_filters.py

@@ -4,7 +4,7 @@ comparison operators, and wildcards across multiple models.
 """
 
 from datetime import datetime, timedelta, timezone
-from typing import Any
+from typing import Any, TypedDict
 
 import pytest
 from fastapi.testclient import TestClient
@@ -13,6 +13,12 @@ from nanoid import generate as generate_nanoid
 from src.models import Peer, Workspace
 
 
+class MessageConfig(TypedDict):
+    content: str
+    peer_id: str
+    metadata: dict[str, Any]
+
+
 @pytest.mark.parametrize(
     "filter_config,expected_peer_indices,description",
     [
@@ -235,6 +241,83 @@ async def test_comparison_operators_filters(
             ), f"Unexpected message '{message_config['content']}' found in results for {description}"
 
 
+@pytest.mark.asyncio
+async def test_nested_metadata_ne_includes_missing_and_empty_metadata(
+    client: TestClient,
+    sample_data: tuple[Workspace, Peer],
+):
+    """`ne` on a nested metadata key must not silently drop rows where the
+    key is absent. Under SQL's three-valued logic, comparing NULL (a missing
+    key or empty metadata) with `<>` yields NULL, which excludes the row —
+    the filter builds and executes cleanly either way, so this has to be
+    checked by counting the rows actually returned.
+    """
+    test_workspace, test_peer = sample_data
+
+    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: {}}},
+    )
+    assert session_response.status_code == 201
+
+    message_configs: list[MessageConfig] = [
+        {
+            "content": "High priority, score 10",
+            "peer_id": test_peer.name,
+            "metadata": {"priority": "high", "score": 10},
+        },
+        {
+            "content": "Low priority, score 5",
+            "peer_id": test_peer.name,
+            "metadata": {"priority": "low", "score": 5},
+        },
+        {
+            "content": "Metadata present, no priority or score key",
+            "peer_id": test_peer.name,
+            "metadata": {"other": "value"},
+        },
+        {
+            "content": "Empty metadata",
+            "peer_id": test_peer.name,
+            "metadata": {},
+        },
+    ]
+    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"]]
+
+    # String value: excludes only the row where priority actually equals "high"
+    string_ne_contents = list_contents({"metadata": {"priority": {"ne": "high"}}})
+    assert sorted(string_ne_contents) == sorted(
+        [
+            message_configs[1]["content"],
+            message_configs[2]["content"],
+            message_configs[3]["content"],
+        ]
+    )
+
+    # Numeric value: excludes only the row where score actually equals 5
+    numeric_ne_contents = list_contents({"metadata": {"score": {"ne": 5}}})
+    assert sorted(numeric_ne_contents) == sorted(
+        [
+            message_configs[0]["content"],
+            message_configs[2]["content"],
+            message_configs[3]["content"],
+        ]
+    )
+
+
 @pytest.mark.asyncio
 async def test_bare_list_membership_sugar(
     client: TestClient,

tests/test_security.py

@@ -1,4 +1,4 @@
-"""Auth scope tests — DEV-1736 regression coverage.
+"""Auth scope tests — regression coverage.
 
 Prior to this fix `auth()` walked the route's declared scope first and fell
 through to a workspace check, so a `{w, p}` token authorized any peer in `w`.

tests/unified/README.md

@@ -48,19 +48,21 @@ Tests are defined in JSON files. A test definition consists of a name, optional
 
 4. **Querying & Assertions**:
     * `query`: Perform an action and assert on the result.
-        * `target`: "chat", "get_context", "get_peer_card", "get_representation"
-        * `scope`: confine the read to a scope (or, for chat/representation, to
-          the union of several). Valid for "chat", "get_representation" and
-          "get_context"; the latter takes a single scope and requires
-          `observed_peer_id`.
+        * `target`: "chat", "get_context", "get_peer_card", "get_representation",
+          "workspace_chat"
+        * `scope`: confine the read to a scope (or, for chat/representation/
+          workspace_chat, to the union of several). Valid for "chat",
+          "get_representation", "get_context", and "workspace_chat"; get_context
+          takes a single scope and requires `observed_peer_id`.
 
 ### Raw HTTP vs the SDK
 
-Most steps drive the Honcho Python SDK. `create_scope` and any query carrying
-`scope` go over raw HTTP instead, because the published SDK trails the API and
-exposes neither. Calling the API directly also tests the contract the SDK is
-generated from, so a wrong status code or response shape surfaces here rather
-than being masked by client-side validation.
+Most steps drive the Honcho Python SDK. `create_scope` and scoped `chat` /
+`get_representation` / `get_context` queries go over raw HTTP instead, because
+the published SDK trails the API and exposes neither. Scoped `workspace_chat`
+uses the SDK `scope` argument. Calling the API directly also tests the contract
+the SDK is generated from, so a wrong status code or response shape surfaces
+here rather than being masked by client-side validation.
 
 ### Assertions
 

tests/unified/run.py

@@ -57,7 +57,9 @@ async def main():
             tests_dir=test_dir, honcho_port=args.port, api_port=args.api_port
         )
 
-    await runner.run()
+    # Non-zero on any failed or unrunnable test, so CI fails on results.
+    if await runner.run():
+        sys.exit(1)
 
 
 if __name__ == "__main__":

tests/unified/runner.py

@@ -386,6 +386,17 @@ class UnifiedTestExecutor:
         raise TimeoutError("Deriver queue did not empty within timeout")
 
     async def perform_query(self, step: QueryAction) -> Any:
+        if step.target == "workspace_chat":
+            if step.input is None:
+                raise ValueError("input required for workspace_chat")
+            return await self.client.aio.chat(
+                step.input,
+                session=step.session_id,
+                reasoning_level=step.reasoning_level,
+                response_format=step.response_format,
+                scope=step.scope,
+            )
+
         if step.scope is not None:
             return await self._perform_scoped_query(step)
 
@@ -626,7 +637,8 @@ class UnifiedTestRunner:
             AsyncAnthropic(api_key=self.api_key) if self.api_key else None
         )
 
-    async def run(self):
+    async def run(self) -> int:
+        """Run the suite and return the number of tests that did not pass."""
         try:
             # 1. Start Harness
             logger.info("Starting Honcho Harness...")
@@ -784,6 +796,8 @@ class UnifiedTestRunner:
 
                 await send_discord_message(discord_webhook_url, message)
 
+            return failed_count
+
         finally:
             # 7. Cleanup
             logger.info("Cleaning up harness...")
@@ -798,4 +812,4 @@ if __name__ == "__main__":
     args = parser.parse_args()
 
     runner = UnifiedTestRunner(Path(args.test_dir))
-    asyncio.run(runner.run())
+    sys.exit(1 if asyncio.run(runner.run()) else 0)

tests/unified/schema.py

@@ -149,7 +149,13 @@ class JsonMatchAssertion(Assertion):
 
 class QueryAction(TestStep):
     step_type: Literal["query"] = "query"
-    target: Literal["chat", "get_context", "get_peer_card", "get_representation"]
+    target: Literal[
+        "chat",
+        "get_context",
+        "get_peer_card",
+        "get_representation",
+        "workspace_chat",
+    ]
 
     session_id: str | None = None
 
@@ -168,9 +174,9 @@ class QueryAction(TestStep):
     # for chat - optional JSON Schema the response must conform to
     response_format: dict[str, Any] | None = None
 
-    # Confine the read to one scope (observer swap) or to the union of several
-    # scopes' member sessions. Forces the raw-HTTP path, since the SDK has no
-    # `scope` parameter. Valid for chat, get_representation and get_context.
+    # Confine the read to one scope (observer swap on peer chat) or to the
+    # union of several scopes' member sessions. Peer-chat/representation/
+    # context go over raw HTTP; workspace_chat uses the SDK `scope` argument.
     scope: str | list[str] | None = None
 
     assertions: list[

tests/unified/test_cases/peer_isolation_test.json

@@ -0,0 +1,148 @@
+{
+  "description": "Test that peer Z cannot access information from a session between X and Y that Z was not part of. This validates session membership scoping - peers should only see messages from sessions they participated in.",
+  "workspace_config": {},
+  "steps": [
+    {
+      "step_type": "create_session",
+      "session_id": "xy_private_session",
+      "peer_configs": {
+        "peer_x": {
+          "observe_me": true,
+          "observe_others": true
+        },
+        "peer_y": {
+          "observe_me": true,
+          "observe_others": true
+        },
+        "agent": {
+          "observe_me": false,
+          "observe_others": true
+        }
+      }
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "xy_private_session",
+      "messages": [
+        {
+          "peer_id": "peer_x",
+          "content": "Hey peer_y, I need to tell you something important. My secret password is 'strawberry_unicorn_42' - please remember this for later."
+        },
+        {
+          "peer_id": "agent",
+          "content": "Got it, I'll remember that."
+        },
+        {
+          "peer_id": "peer_y",
+          "content": "Thanks for sharing that, peer_x. I've noted your password: strawberry_unicorn_42. I'll keep it safe."
+        },
+        {
+          "peer_id": "peer_x",
+          "content": "Perfect. Also, I'm planning a surprise party for peer_z next Friday at 7pm at the Blue Whale restaurant downtown. Don't tell peer_z!"
+        },
+        {
+          "peer_id": "peer_y",
+          "content": "Amazing! I'll be there. The surprise party at Blue Whale will be great. My lips are sealed!"
+        }
+      ]
+    },
+    {
+      "step_type": "wait",
+      "target": "queue_empty",
+      "flush": true
+    },
+    {
+      "step_type": "create_session",
+      "session_id": "z_separate_session",
+      "peer_configs": {
+        "peer_z": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "agent": {
+          "observe_me": false,
+          "observe_others": true
+        }
+      }
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "z_separate_session",
+      "messages": [
+        {
+          "peer_id": "peer_z",
+          "content": "Hi! I'm peer_z. I like going to restaurants and attending parties."
+        },
+        {
+          "peer_id": "agent",
+          "content": "That's nice! Do you have any upcoming plans?"
+        }
+      ]
+    },
+    {
+      "step_type": "wait",
+      "target": "queue_empty",
+      "flush": true
+    },
+    {
+      "step_type": "query",
+      "target": "chat",
+      "session_id": "z_separate_session",
+      "observer_peer_id": "agent",
+      "observed_peer_id": "peer_z",
+      "input": "What is peer_x's secret password?",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "not_contains",
+          "text": "strawberry_unicorn_42",
+          "case_sensitive": false
+        },
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response indicate that the agent DOES NOT KNOW the password, or that the information is not available? The response should NOT reveal the password 'strawberry_unicorn_42'. Return true if the agent correctly says they don't know or the info isn't available.",
+          "pass_if": true
+        }
+      ]
+    },
+    {
+      "step_type": "query",
+      "target": "chat",
+      "session_id": "z_separate_session",
+      "observer_peer_id": "agent",
+      "observed_peer_id": "peer_z",
+      "input": "Is there a surprise party being planned? If so, where and when?",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "not_contains",
+          "text": "Blue Whale",
+          "case_sensitive": false
+        },
+        {
+          "assertion_type": "not_contains",
+          "text": "Friday at 7",
+          "case_sensitive": false
+        },
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response indicate that the agent DOES NOT KNOW about any surprise party details? The response should NOT mention the Blue Whale restaurant or Friday at 7pm. Return true if the agent correctly says they don't have this information.",
+          "pass_if": true
+        }
+      ]
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "input": "What did peer_x tell peer_y about their password?",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "The workspace chat should be able to answer this question since it can access information from all sessions. Does the response mention the password 'strawberry_unicorn_42' or indicate that peer_x shared a password with peer_y? Return true if the workspace-level query CAN see this information.",
+          "pass_if": true
+        }
+      ]
+    }
+  ]
+}

tests/unified/test_cases/workspace_chat_cross_peer.json

@@ -0,0 +1,88 @@
+{
+  "description": "Test that workspace-level chat can synthesize information across multiple peers in a single response. Three peers have distinct attributes; workspace chat should be able to compare or list them.",
+  "workspace_config": {},
+  "steps": [
+    {
+      "step_type": "create_session",
+      "session_id": "ws_cross_session",
+      "peer_configs": {
+        "dan": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "emma": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "frank": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "agent": {
+          "observe_me": false,
+          "observe_others": false
+        }
+      }
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "ws_cross_session",
+      "messages": [
+        {
+          "peer_id": "dan",
+          "content": "I'm a software engineer specializing in Rust and systems programming. I work at a startup building distributed databases."
+        },
+        {
+          "peer_id": "agent",
+          "content": "Interesting! What kind of databases?"
+        },
+        {
+          "peer_id": "dan",
+          "content": "We're building a distributed time-series database optimized for IoT sensor data."
+        },
+        {
+          "peer_id": "emma",
+          "content": "I'm a marine biologist studying coral reef ecosystems in the Great Barrier Reef. I've been doing field research there for 3 years."
+        },
+        {
+          "peer_id": "agent",
+          "content": "That must be fascinating work. What's your focus?"
+        },
+        {
+          "peer_id": "emma",
+          "content": "I'm specifically studying the impact of ocean temperature changes on coral bleaching patterns."
+        },
+        {
+          "peer_id": "frank",
+          "content": "I'm a pastry chef at a Michelin-starred restaurant in Lyon, France. My specialty is chocolate souffles."
+        },
+        {
+          "peer_id": "agent",
+          "content": "That's impressive! How did you get into pastry?"
+        },
+        {
+          "peer_id": "frank",
+          "content": "I trained at Le Cordon Bleu in Paris and then apprenticed under Chef Pierre Herme for two years."
+        }
+      ]
+    },
+    {
+      "step_type": "wait",
+      "target": "queue_empty",
+      "flush": true
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "input": "What professions do the people in this workspace have? List each person and what they do.",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response mention at least 2 of the 3 peers (Dan, Emma, Frank) along with their professions? Dan is a software engineer, Emma is a marine biologist, and Frank is a pastry chef. The response should identify at least 2 of these 3 profession/person pairs correctly.",
+          "pass_if": true
+        }
+      ]
+    }
+  ]
+}

tests/unified/test_cases/workspace_chat_from_messages.json

@@ -0,0 +1,61 @@
+{
+  "description": "Test that workspace-level chat can fall back to searching messages directly. Deriver is disabled so no observations are created, forcing the agent to find information from raw message history.",
+  "workspace_config": {
+    "reasoning": {
+      "enabled": false
+    }
+  },
+  "steps": [
+    {
+      "step_type": "create_session",
+      "session_id": "ws_msg_session",
+      "peer_configs": {
+        "carol": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "agent": {
+          "observe_me": false,
+          "observe_others": false
+        }
+      }
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "ws_msg_session",
+      "messages": [
+        {
+          "peer_id": "carol",
+          "content": "I just finished writing my third novel. It's a mystery set in 1920s Paris called 'The Montmartre Cipher'."
+        },
+        {
+          "peer_id": "agent",
+          "content": "Congratulations! That's a great achievement. What inspired the setting?"
+        },
+        {
+          "peer_id": "carol",
+          "content": "I lived in Paris for two years and fell in love with the history of Montmartre. The artists and writers who gathered there in the 1920s were fascinating."
+        }
+      ]
+    },
+    {
+      "step_type": "wait",
+      "duration": 2,
+      "target": "queue_empty"
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "input": "What can you tell me about Carol's writing?",
+      "session_id": "ws_msg_session",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response reference Carol writing novels or a book? It should mention something about her being a writer/author or her novel. Mentioning 'The Montmartre Cipher' or Paris or mystery is a bonus but not required. The key point is that the system found information about Carol's writing from the message history.",
+          "pass_if": true
+        }
+      ]
+    }
+  ]
+}

tests/unified/test_cases/workspace_chat_from_observations.json

@@ -0,0 +1,85 @@
+{
+  "description": "Test that workspace-level chat can gather information about peers from their global representations (observer==observed). Two peers share distinct facts; after deriver processes them, workspace chat should retrieve info about each peer.",
+  "workspace_config": {},
+  "steps": [
+    {
+      "step_type": "create_session",
+      "session_id": "ws_obs_session",
+      "peer_configs": {
+        "alice": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "bob": {
+          "observe_me": true,
+          "observe_others": false
+        },
+        "agent": {
+          "observe_me": false,
+          "observe_others": false
+        }
+      }
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "ws_obs_session",
+      "messages": [
+        {
+          "peer_id": "alice",
+          "content": "I'm a professional violinist and I perform with the Chicago Symphony Orchestra every Friday evening."
+        },
+        {
+          "peer_id": "agent",
+          "content": "That's wonderful! How long have you been playing?"
+        },
+        {
+          "peer_id": "alice",
+          "content": "I've been playing violin since I was 5 years old, so about 25 years now."
+        },
+        {
+          "peer_id": "bob",
+          "content": "I just got back from a scuba diving trip in Belize. I'm a certified rescue diver."
+        },
+        {
+          "peer_id": "agent",
+          "content": "That sounds amazing! Do you dive often?"
+        },
+        {
+          "peer_id": "bob",
+          "content": "Yes, I try to go diving at least twice a month. My favorite dive site is the Great Blue Hole."
+        }
+      ]
+    },
+    {
+      "step_type": "wait",
+      "target": "queue_empty",
+      "flush": true
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "input": "What do you know about Alice's musical background?",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response mention that Alice is a violinist or plays violin? It should reference her connection to music/violin performance. The specific detail about the Chicago Symphony Orchestra is a bonus but not required.",
+          "pass_if": true
+        }
+      ]
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "input": "What are Bob's hobbies?",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response mention that Bob is into scuba diving or diving? It should reference his diving hobby. The specific detail about Belize or the Great Blue Hole is a bonus but not required.",
+          "pass_if": true
+        }
+      ]
+    }
+  ]
+}

tests/unified/test_cases/workspace_chat_scope.json

@@ -0,0 +1,97 @@
+{
+  "description": "Workspace chat with `scope` only recalls the scope's member sessions.",
+  "workspace_config": {
+    "reasoning": {
+      "enabled": true
+    }
+  },
+  "steps": [
+    {
+      "step_type": "create_session",
+      "session_id": "in_scope",
+      "peer_configs": {
+        "alice": {
+          "observe_me": true,
+          "observe_others": false
+        }
+      }
+    },
+    {
+      "step_type": "create_session",
+      "session_id": "out_of_scope",
+      "peer_configs": {
+        "bob": {
+          "observe_me": true,
+          "observe_others": false
+        }
+      }
+    },
+    {
+      "step_type": "create_scope",
+      "scope_id": "therapy",
+      "session_ids": ["in_scope"]
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "in_scope",
+      "messages": [
+        {
+          "peer_id": "alice",
+          "content": "My favorite tea is jasmine green tea from Hangzhou."
+        }
+      ]
+    },
+    {
+      "step_type": "add_messages",
+      "session_id": "out_of_scope",
+      "messages": [
+        {
+          "peer_id": "bob",
+          "content": "The vault code is 7491-orange-lantern."
+        }
+      ]
+    },
+    {
+      "step_type": "wait",
+      "duration": 2,
+      "target": "queue_empty"
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "scope": "therapy",
+      "input": "What facts do you know about people in this workspace? Mention any codes or secrets if you have them.",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "not_contains",
+          "text": "7491-orange-lantern",
+          "case_sensitive": false
+        },
+        {
+          "assertion_type": "not_contains",
+          "text": "scope.therapy",
+          "case_sensitive": false
+        },
+        {
+          "assertion_type": "llm_judge",
+          "prompt": "Does the response mention Alice's jasmine tea (or that Alice likes tea), and does it NOT mention a vault code or 7491? Return true only if both are true.",
+          "pass_if": true
+        }
+      ]
+    },
+    {
+      "step_type": "query",
+      "target": "workspace_chat",
+      "input": "List every peer in this workspace and how many messages each has.",
+      "reasoning_level": "low",
+      "assertions": [
+        {
+          "assertion_type": "not_contains",
+          "text": "scope.therapy",
+          "case_sensitive": false
+        }
+      ]
+    }
+  ]
+}

tests/unified/test_schema.py

@@ -0,0 +1,15 @@
+"""Assert every unified JSON case still parses against the schema."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+_CASES = sorted(Path(__file__).parent.joinpath("test_cases").glob("*.json"))
+
+
+@pytest.mark.parametrize("path", _CASES, ids=lambda p: p.name)
+def test_unified_case_parses(path: Path) -> None:
+    from tests.unified.schema import TestDefinition
+
+    TestDefinition(**json.loads(path.read_text()))

tests/utils/test_filter.py

@@ -228,6 +228,20 @@ def test_ne_is_null_safe():
     assert "IS DISTINCT FROM" in where
 
 
+def test_nested_metadata_ne_string_is_null_safe():
+    """`ne` on a JSONB metadata key went through the operator map as plain <>,
+    unlike the scalar path, so a row missing that key was silently dropped."""
+    where = _where(Document, {"metadata": {"priority": {"ne": "high"}}})
+    assert "IS DISTINCT FROM" in where
+    assert "!=" not in where
+
+
+def test_nested_metadata_ne_numeric_is_null_safe():
+    where = _where(Document, {"metadata": {"score": {"ne": 5}}})
+    assert "IS DISTINCT FROM" in where
+    assert "!=" not in where
+
+
 def test_not_is_null_safe_over_a_compound_condition():
     """Negation has to survive nesting, not just single comparisons."""
     where = _where(