diff --git a/CHANGELOG.md b/CHANGELOG.md index e314f867..522c0985 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/sdks/python/src/honcho/aio.py b/sdks/python/src/honcho/aio.py index 29c0f445..f5148ee6 100644 --- a/sdks/python/src/honcho/aio.py +++ b/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, diff --git a/sdks/python/src/honcho/client.py b/sdks/python/src/honcho/client.py index 1527792e..dbee9478 100644 --- a/sdks/python/src/honcho/client.py +++ b/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, diff --git a/sdks/python/src/honcho/http/routes.py b/sdks/python/src/honcho/http/routes.py index 8f9ee2fb..8295b994 100644 --- a/sdks/python/src/honcho/http/routes.py +++ b/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" diff --git a/sdks/typescript/src/client.ts b/sdks/typescript/src/client.ts index 2b66b9c2..96cf57c8 100644 --- a/sdks/typescript/src/client.ts +++ b/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 { + await this._ensureWorkspace() + return this._http.post( + `/${API_VERSION}/workspaces/${workspaceId}/chat`, + { body: params } + ) + } + + private async _workspaceChatStream( + workspaceId: string, + params: Omit + ): Promise { + 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 + scope?: string | string[] + } + ): Promise { + 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 + scope?: string | string[] + } + ): Promise { + 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. * diff --git a/sdks/typescript/src/index.ts b/sdks/typescript/src/index.ts index 24ba57b4..509ba645 100644 --- a/sdks/typescript/src/index.ts +++ b/sdks/typescript/src/index.ts @@ -69,6 +69,7 @@ export type { SessionResponse, SessionSummariesResponse, SummaryResponse, + WorkspaceChatResponse, WorkspaceResponse, } from './types/api' diff --git a/sdks/typescript/src/types/api.ts b/sdks/typescript/src/types/api.ts index 10ed6037..c5855713 100644 --- a/sdks/typescript/src/types/api.ts +++ b/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 + scope?: string | string[] +} + +export interface WorkspaceChatResponse { + content: string | null +} + export interface PeerRepresentationParams { session_id?: string target?: string diff --git a/src/crud/__init__.py b/src/crud/__init__.py index 38be14da..0e920717 100644 --- a/src/crud/__init__.py +++ b/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", diff --git a/src/crud/scope.py b/src/crud/scope.py index d5b92c31..0fc8e41f 100644 --- a/src/crud/scope.py +++ b/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, diff --git a/src/crud/workspace.py b/src/crud/workspace.py index c9040f55..a5042acd 100644 --- a/src/crud/workspace.py +++ b/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 + ] diff --git a/src/dialectic/chat.py b/src/dialectic/chat.py index ba066741..47f8dff7 100644 --- a/src/dialectic/chat.py +++ b/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 diff --git a/src/dialectic/core.py b/src/dialectic/core.py index 9580b94c..786866e1 100644 --- a/src/dialectic/core.py +++ b/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(). diff --git a/src/dialectic/prompts.py b/src/dialectic/prompts.py index 948bff70..7f98a0e1 100644 --- a/src/dialectic/prompts.py +++ b/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. +""" diff --git a/src/dialectic/workspace.py b/src/dialectic/workspace.py new file mode 100644 index 00000000..7096add6 --- /dev/null +++ b/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", + ) diff --git a/src/routers/peers.py b/src/routers/peers.py index 7843081e..c5edfd4e 100644 --- a/src/routers/peers.py +++ b/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, diff --git a/src/routers/workspaces.py b/src/routers/workspaces.py index 42c111ad..5d449f9a 100644 --- a/src/routers/workspaces.py +++ b/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) diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py index b4211b99..9f414583 100644 --- a/src/schemas/__init__.py +++ b/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", diff --git a/src/schemas/api.py b/src/schemas/api.py index 55a6121a..43d91b26 100644 --- a/src/schemas/api.py +++ b/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): diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index b8af9eee..0dbdd304 100644 --- a/src/utils/agent_tools.py +++ b/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, + ) diff --git a/src/utils/scopes.py b/src/utils/scopes.py index c7b89192..5f01916b 100644 --- a/src/utils/scopes.py +++ b/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" + ) diff --git a/tests/conftest.py b/tests/conftest.py index 8a078652..7d9e81da 100644 --- a/tests/conftest.py +++ b/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, } diff --git a/tests/routes/test_scope_reads.py b/tests/routes/test_scope_reads.py index eafdecaf..23eb8d5f 100644 --- a/tests/routes/test_scope_reads.py +++ b/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 diff --git a/tests/test_security.py b/tests/test_security.py index 1785be8b..23725692 100644 --- a/tests/test_security.py +++ b/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`. diff --git a/tests/test_workspace_chat.py b/tests/test_workspace_chat.py new file mode 100644 index 00000000..8b73c4a7 --- /dev/null +++ b/tests/test_workspace_chat.py @@ -0,0 +1,1222 @@ +"""Integration tests for the workspace-level chat feature. + +Tests cover: +- Route-level: POST /workspaces/{workspace_id}/chat endpoint +- Tool handlers: workspace-specific tool handlers and executor +""" + +import asyncio +import json +from collections.abc import Callable +from contextlib import asynccontextmanager +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from typing import Any + +import pytest +from nanoid import generate as generate_nanoid +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models +from src.dialectic.chat import workspace_chat, workspace_chat_stream +from src.models import Peer, Workspace +from src.utils.agent_tools import ( + ToolContext, + _handle_get_observation_context, # pyright: ignore[reportPrivateUsage] + _handle_get_peer_card_by_name, # pyright: ignore[reportPrivateUsage] + _handle_get_reasoning_chain, # pyright: ignore[reportPrivateUsage] + _handle_get_workspace_stats, # pyright: ignore[reportPrivateUsage] + _handle_search_memory_workspace, # pyright: ignore[reportPrivateUsage] + create_workspace_tool_executor, +) +from src.utils.scopes import SCOPE_KIND, scope_peer_name + +# ============================================================================= +# Fixtures +# ============================================================================= + + +def _tool_text(result: object) -> str: + """Unwrap ToolResult (today's handler contract) or pass through str.""" + content = getattr(result, "content", None) + return content if isinstance(content, str) else str(result) + + +@pytest.fixture +async def workspace_test_data( + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], +) -> Any: + """Create comprehensive test data with multiple peers and observations. + + Sets up a workspace with: + - 3 peers (peer1 observes peer2, peer1 observes peer3) + - 1 session with messages from all peers + - Documents (observations) across different peer pairs + """ + workspace, peer1 = sample_data + + # Create additional peers + peer2 = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name) + peer3 = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name) + db_session.add_all([peer2, peer3]) + await db_session.flush() + + # Create session + session = models.Session(name=str(generate_nanoid()), workspace_name=workspace.name) + db_session.add(session) + await db_session.flush() + + # Create collections (peer1 observes peer2, peer1 observes peer3) + collection1 = models.Collection( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + ) + collection2 = models.Collection( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer3.name, + ) + db_session.add_all([collection1, collection2]) + await db_session.flush() + + # Create messages + now = datetime.now(timezone.utc) + messages: list[models.Message] = [] + for i in range(6): + peer_name = [peer1.name, peer2.name, peer3.name][i % 3] + msg = models.Message( + workspace_name=workspace.name, + session_name=session.name, + peer_name=peer_name, + content=f"Test message {i} from {peer_name}", + seq_in_session=i + 1, + token_count=10, + created_at=now - timedelta(minutes=6 - i), + ) + db_session.add(msg) + messages.append(msg) + await db_session.flush() + for msg in messages: + await db_session.refresh(msg) + + # Create documents for peer1->peer2 observations + docs_peer2: list[models.Document] = [] + for content in [ + "User likes coffee and programming", + "User works remotely from home", + ]: + doc = models.Document( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + content=content, + embedding=[0.1] * 1536, + session_name=session.name, + level="explicit", + metadata={ + "message_ids": [messages[0].id], + "message_created_at": str(messages[0].created_at), + }, + ) + db_session.add(doc) + docs_peer2.append(doc) + + # Create documents for peer1->peer3 observations + docs_peer3: list[models.Document] = [] + for content in [ + "User prefers mornings for deep work", + "User enjoys hiking on weekends", + ]: + doc = models.Document( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer3.name, + content=content, + embedding=[0.2] * 1536, + session_name=session.name, + level="explicit", + metadata={ + "message_ids": [messages[1].id], + "message_created_at": str(messages[1].created_at), + }, + ) + db_session.add(doc) + docs_peer3.append(doc) + + await db_session.flush() + for doc in docs_peer2 + docs_peer3: + await db_session.refresh(doc) + + # Commit so data is visible to independent tracked_db sessions used by + # workspace-level tool handlers. + await db_session.commit() + + yield workspace, peer1, peer2, peer3, session, messages, docs_peer2, docs_peer3 + + await db_session.rollback() + + +@pytest.fixture +def make_workspace_ctx( + workspace_test_data: Any, +) -> Callable[..., ToolContext]: + """Factory fixture to create ToolContext.""" + workspace, *_ = workspace_test_data + shared_lock = asyncio.Lock() + + def _make_ctx( + *, + session_name: str | None = None, + include_observation_ids: bool = True, + session_allowlist: list[str] | None = None, + ) -> ToolContext: + return ToolContext( + observer="", + observed="", + current_messages=None, + workspace_name=workspace.name, + session_name=session_name, + include_observation_ids=include_observation_ids, + history_token_limit=8192, + db_lock=shared_lock, + session_allowlist=session_allowlist, + ) + + return _make_ctx + + +# ============================================================================= +# Route Tests: POST /workspaces/{workspace_id}/chat +# ============================================================================= + + +class TestWorkspaceChatEndpoint: + """Tests for the workspace chat API endpoint.""" + + def test_workspace_chat_basic( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Basic non-streaming workspace chat returns DialecticResponse.""" + test_workspace, _ = sample_data + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={ + "query": "What do you know about the peers in this workspace?", + "stream": False, + }, + ) + assert response.status_code == 200 + data = response.json() + assert "content" in data + assert data["content"] == "Test workspace chat response" + + def test_workspace_chat_with_session_id( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Workspace chat accepts optional session_id parameter.""" + test_workspace, _ = sample_data + session_id = str(generate_nanoid()) + + # Create a session first + create_response = client.post( + f"/v3/workspaces/{test_workspace.name}/sessions", + json={"name": session_id}, + ) + assert create_response.status_code in (200, 201) + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={ + "query": "Tell me about recent conversations", + "session_id": session_id, + "stream": False, + }, + ) + assert response.status_code == 200 + data = response.json() + assert "content" in data + + def test_workspace_chat_with_reasoning_level( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Workspace chat accepts reasoning_level parameter.""" + test_workspace, _ = sample_data + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={ + "query": "Analyze common themes across all peers", + "stream": False, + "reasoning_level": "low", + }, + ) + assert response.status_code == 200 + data = response.json() + assert "content" in data + + def test_workspace_chat_streaming( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Streaming workspace chat returns SSE-formatted events.""" + test_workspace, _ = sample_data + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={ + "query": "What patterns do you see across the workspace?", + "stream": True, + }, + ) + assert response.status_code == 200 + assert "text/event-stream" in response.headers.get("content-type", "") + + # Parse SSE events + events: list[Any] = [] + for line in response.text.strip().split("\n\n"): + if line.startswith("data: "): + event_data = json.loads(line[6:]) + events.append(event_data) + + # Should have content events and a final done event + assert len(events) >= 2 + content_events = [e for e in events if not e.get("done")] + done_events = [e for e in events if e.get("done")] + assert len(content_events) >= 1 + assert len(done_events) == 1 + + # Content events should have delta.content + for event in content_events: + assert "delta" in event + assert "content" in event["delta"] + + def test_workspace_chat_empty_query_rejected( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Empty query should be rejected by validation.""" + test_workspace, _ = sample_data + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={ + "query": "", + "stream": False, + }, + ) + assert response.status_code == 422 + + def test_workspace_chat_missing_query_rejected( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Missing query field should be rejected.""" + test_workspace, _ = sample_data + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={"stream": False}, + ) + assert response.status_code == 422 + + def test_workspace_chat_null_content_response( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + mock_llm_call_functions: dict[str, Any], + ): + """When workspace_chat returns None, response content should be None.""" + test_workspace, _ = sample_data + mock_llm_call_functions["workspace_chat"].side_effect = None + mock_llm_call_functions["workspace_chat"].return_value = None + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={ + "query": "Some query", + "stream": False, + }, + ) + assert response.status_code == 200 + data = response.json() + assert data["content"] is None + + def test_workspace_chat_defaults( + self, + client: Any, + sample_data: tuple[Workspace, Peer], + ): + """Endpoint works with only the required query field.""" + test_workspace, _ = sample_data + + response = client.post( + f"/v3/workspaces/{test_workspace.name}/chat", + json={"query": "Hello workspace"}, + ) + assert response.status_code == 200 + data = response.json() + assert "content" in data + + +@pytest.mark.asyncio +async def test_workspace_chat_releases_preflight_session_before_agent_answer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + active_sessions = 0 + + @asynccontextmanager + async def fake_tracked_db(_: str | None = None, **_kwargs: Any): + nonlocal active_sessions + active_sessions += 1 + try: + yield object() + finally: + active_sessions -= 1 + + async def fake_get_session(*args: Any, **kwargs: Any) -> object: + _ = (args, kwargs) + assert active_sessions == 1 + return SimpleNamespace(id="session-id") + + async def fake_answer(_self: Any, query: str, **_kwargs: Any) -> str: + assert query == "What changed?" + assert active_sessions == 0 + return "ok" + + async def fake_get_workspace(*args: Any, **kwargs: Any) -> object: + _ = (args, kwargs) + assert active_sessions == 1 + return SimpleNamespace(name="workspace") + + monkeypatch.setattr("src.dialectic.chat.tracked_db", fake_tracked_db) + monkeypatch.setattr("src.dialectic.chat.crud.get_workspace", fake_get_workspace) + monkeypatch.setattr("src.dialectic.chat.crud.get_session", fake_get_session) + monkeypatch.setattr( + "src.dialectic.chat.WorkspaceDialecticAgent.answer", fake_answer + ) + + result = await workspace_chat("workspace", "session", "What changed?") + + assert result == "ok" + + +@pytest.mark.asyncio +async def test_workspace_chat_stream_releases_preflight_session_before_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + active_sessions = 0 + + @asynccontextmanager + async def fake_tracked_db(_: str | None = None, **_kwargs: Any): + nonlocal active_sessions + active_sessions += 1 + try: + yield object() + finally: + active_sessions -= 1 + + async def fake_get_session(*args: Any, **kwargs: Any) -> object: + _ = (args, kwargs) + assert active_sessions == 1 + return SimpleNamespace(id="session-id") + + async def fake_answer_stream(_self: Any, query: str, **_kwargs: Any): + assert query == "Stream it" + assert active_sessions == 0 + yield "chunk-1" + assert active_sessions == 0 + yield "chunk-2" + + async def fake_get_workspace(*args: Any, **kwargs: Any) -> object: + _ = (args, kwargs) + assert active_sessions == 1 + return SimpleNamespace(name="workspace") + + monkeypatch.setattr("src.dialectic.chat.tracked_db", fake_tracked_db) + monkeypatch.setattr("src.dialectic.chat.crud.get_workspace", fake_get_workspace) + monkeypatch.setattr("src.dialectic.chat.crud.get_session", fake_get_session) + monkeypatch.setattr( + "src.dialectic.chat.WorkspaceDialecticAgent.answer_stream", + fake_answer_stream, + ) + + chunks = [ + chunk + async for chunk in workspace_chat_stream("workspace", "session", "Stream it") + ] + + assert chunks == ["chunk-1", "chunk-2"] + + +# ============================================================================= +# Tool Handler Tests: Workspace-Specific Handlers +# ============================================================================= + + +@pytest.mark.asyncio +class TestSearchMemoryWorkspace: + """Tests for _handle_search_memory_workspace (representation-scoped).""" + + async def test_requires_observer_and_observed( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error when observer/observed params are missing.""" + ctx = make_workspace_ctx() + + result = _tool_text( + await _handle_search_memory_workspace(ctx, {"query": "coffee preferences"}) + ) + assert "ERROR" in result + assert "observer" in result + + async def test_missing_observer_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error when only observed is provided.""" + ctx = make_workspace_ctx() + + result = _tool_text( + await _handle_search_memory_workspace( + ctx, {"query": "test", "observed": "someone"} + ) + ) + assert "ERROR" in result + + async def test_missing_observed_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error when only observer is provided.""" + ctx = make_workspace_ctx() + + result = _tool_text( + await _handle_search_memory_workspace( + ctx, {"query": "test", "observer": "someone"} + ) + ) + assert "ERROR" in result + + async def test_returns_observations_for_specific_pair( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + monkeypatch: pytest.MonkeyPatch, + ): + """Returns observations scoped to a specific observer/observed pair.""" + monkeypatch.setattr("src.config.settings.VECTOR_STORE.MIGRATED", False) + _, peer1, peer2, _, _, _, _, _ = workspace_test_data + ctx = make_workspace_ctx() + + result = _tool_text( + await _handle_search_memory_workspace( + ctx, + { + "query": "coffee preferences", + "observer": peer1.name, + "observed": peer2.name, + }, + ) + ) + + assert "Found" in result + assert "observations" in result.lower() + # Should be scoped to peer1->peer2 + assert f"{peer1.name}->{peer2.name}" in result + + async def test_does_not_return_observations_from_other_pairs( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + monkeypatch: pytest.MonkeyPatch, + ): + """Does not leak observations from other peer pairs.""" + monkeypatch.setattr("src.config.settings.VECTOR_STORE.MIGRATED", False) + _, peer1, _, peer3, _, _, _, _ = workspace_test_data + ctx = make_workspace_ctx() + + result = _tool_text( + await _handle_search_memory_workspace( + ctx, + { + "query": "coffee", + "observer": peer1.name, + "observed": peer3.name, + }, + ) + ) + + # peer3 observations are about hiking/mornings, not coffee + # Should either find the hiking/mornings ones or none + assert isinstance(result, str) + + async def test_falls_back_to_message_search( + self, + db_session: AsyncSession, + sample_data: tuple[Workspace, Peer], + ): + """Falls back to message search when no observations exist for the pair.""" + workspace, _ = sample_data + + session = models.Session( + name=str(generate_nanoid()), workspace_name=workspace.name + ) + db_session.add(session) + await db_session.flush() + + observer = models.Peer( + name=str(generate_nanoid()), workspace_name=workspace.name + ) + observed = models.Peer( + name=str(generate_nanoid()), workspace_name=workspace.name + ) + db_session.add_all([observer, observed]) + await db_session.flush() + + msg = models.Message( + workspace_name=workspace.name, + session_name=session.name, + peer_name=observed.name, + content="I really like programming in Python", + seq_in_session=1, + token_count=10, + created_at=datetime.now(timezone.utc), + ) + db_session.add(msg) + await db_session.flush() + + ctx = ToolContext( + observer="", + observed="", + current_messages=None, + workspace_name=workspace.name, + session_name=session.name, + include_observation_ids=False, + history_token_limit=8192, + db_lock=asyncio.Lock(), + ) + + result = _tool_text( + await _handle_search_memory_workspace( + ctx, + { + "query": "programming", + "observer": observer.name, + "observed": observed.name, + }, + ) + ) + + assert isinstance(result, str) + assert "No observations" in result or "Found" in result + + async def test_respects_top_k( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + monkeypatch: pytest.MonkeyPatch, + ): + """Respects the top_k parameter, capped at 40.""" + monkeypatch.setattr("src.config.settings.VECTOR_STORE.MIGRATED", False) + _, peer1, peer2, _, _, _, _, _ = workspace_test_data + ctx = make_workspace_ctx() + + result = _tool_text( + await _handle_search_memory_workspace( + ctx, + { + "query": "test", + "top_k": 2, + "observer": peer1.name, + "observed": peer2.name, + }, + ) + ) + + assert isinstance(result, str) + + +@pytest.mark.asyncio +class TestGetWorkspaceStats: + """Tests for _handle_get_workspace_stats.""" + + async def test_returns_stats( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Returns workspace statistics.""" + _ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_workspace_stats(ctx, {}) + + assert "Workspace stats" in result + assert "Peers: 3" in result + assert "Sessions: 1" in result + assert "Messages: 6" in result + assert "Date range" in result + + async def test_lists_most_active_peers( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Includes the most active peers with message counts.""" + _, peer1, peer2, peer3, *_ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_workspace_stats(ctx, {}) + + assert "Most active peers" in result + assert peer1.name in result + assert peer2.name in result + assert peer3.name in result + assert "messages" in result + + async def test_empty_workspace( + self, + db_session: AsyncSession, + ): + """Returns zero counts for an empty workspace.""" + workspace = models.Workspace(name=str(generate_nanoid())) + db_session.add(workspace) + await db_session.flush() + + ctx = ToolContext( + observer="", + observed="", + current_messages=None, + workspace_name=workspace.name, + session_name=None, + include_observation_ids=False, + history_token_limit=8192, + db_lock=asyncio.Lock(), + ) + + result = await _handle_get_workspace_stats(ctx, {}) + + assert "Peers: 0" in result + assert "Messages: 0" in result + + async def test_excludes_scope_peers( + self, + db_session: AsyncSession, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + workspace, *_ = workspace_test_data + db_session.add( + models.Peer( + name=scope_peer_name("therapy"), + workspace_name=workspace.name, + internal_metadata={"kind": SCOPE_KIND}, + configuration={"observe_me": False}, + ) + ) + await db_session.commit() + + result = await _handle_get_workspace_stats(make_workspace_ctx(), {}) + + assert "Peers: 3" in result + assert "scope.therapy" not in result + + async def test_empty_session_allowlist_is_zero( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + _ = workspace_test_data + result = await _handle_get_workspace_stats( + make_workspace_ctx(session_allowlist=[]), {} + ) + + assert "Peers: 0" in result + assert "Sessions: 0" in result + assert "Messages: 0" in result + + +@pytest.mark.asyncio +class TestGetPeerCardByName: + """Tests for _handle_get_peer_card_by_name.""" + + async def test_returns_peer_card( + self, + db_session: AsyncSession, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Returns peer card when it exists.""" + workspace, peer1, peer2, *_ = workspace_test_data + + # Create a peer card + await crud.set_peer_card( + db_session, + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + peer_card=["Name: Alice", "Location: NYC"], + ) + + ctx = make_workspace_ctx() + result = await _handle_get_peer_card_by_name( + ctx, {"observer": peer1.name, "observed": peer2.name} + ) + + assert "Peer card" in result + assert "Name: Alice" in result + assert "Location: NYC" in result + + async def test_returns_not_found( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Returns appropriate message when peer card doesn't exist.""" + _, peer1, peer2, *_ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_peer_card_by_name( + ctx, {"observer": peer1.name, "observed": peer2.name} + ) + + assert "No peer card" in result + + async def test_session_allowlist_refuses( + self, + db_session: AsyncSession, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """A peer card is a cross-session aggregate, so a scoped query must not + get one — otherwise `scope` leaks facts derived outside its sessions.""" + workspace, peer1, peer2, _peer3, session, *_ = workspace_test_data + + await crud.set_peer_card( + db_session, + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + peer_card=["Secret: derived from an out-of-scope session"], + ) + + ctx = make_workspace_ctx(session_allowlist=[session.name]) + result = await _handle_get_peer_card_by_name( + ctx, {"observer": peer1.name, "observed": peer2.name} + ) + + assert "Secret" not in result + assert "unavailable for session-scoped queries" in result + + async def test_unknown_peer_is_answered_not_raised( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """The agent supplies peer names from its own routing, so a name that + doesn't exist is an expected turn, not an unhandled exception.""" + _, peer1, *_ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_peer_card_by_name( + ctx, {"observer": "no-such-peer", "observed": peer1.name} + ) + + assert "No peer named 'no-such-peer'" in result + + async def test_missing_params_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error when observer/observed params are missing.""" + ctx = make_workspace_ctx() + + result = await _handle_get_peer_card_by_name(ctx, {}) + + assert "ERROR" in result + + async def test_missing_observer_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error when only observed is provided.""" + ctx = make_workspace_ctx() + + result = await _handle_get_peer_card_by_name(ctx, {"observed": "someone"}) + + assert "ERROR" in result + + +@pytest.mark.asyncio +class TestGetObservationContextWorkspace: + """Tests for get_observation_context under the workspace executor. + + The workspace loadout routes this straight to the shared handler: its + observer="" sentinel already normalizes to None ("no perspective + scoping") at the crud boundary.""" + + async def test_retrieves_messages_by_id( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Retrieves messages by their public IDs.""" + _, _, _, _, _, messages, _, _ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_observation_context( + ctx, {"message_ids": [messages[0].public_id]} + ) + + assert "Retrieved" in result or "No messages found" in result + + async def test_nonexistent_message_ids( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns appropriate message for nonexistent IDs.""" + ctx = make_workspace_ctx() + + result = await _handle_get_observation_context( + ctx, {"message_ids": ["nonexistent_id"]} + ) + + assert "No messages found" in result + + async def test_respects_session_scope( + self, + db_session: AsyncSession, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Session-scoped context lookup should not leak snippets from other sessions.""" + workspace, _peer1, peer2, _peer3, session, messages, *_ = workspace_test_data + + other_session = models.Session( + name=str(generate_nanoid()), + workspace_name=workspace.name, + ) + db_session.add(other_session) + await db_session.flush() + + leaked_message = models.Message( + workspace_name=workspace.name, + session_name=other_session.name, + peer_name=peer2.name, + content="LEAKED_FROM_OTHER_SESSION", + seq_in_session=messages[0].seq_in_session, + token_count=10, + created_at=datetime.now(timezone.utc), + ) + db_session.add(leaked_message) + await db_session.commit() + + ctx = make_workspace_ctx(session_name=session.name) + result = await _handle_get_observation_context( + ctx, {"message_ids": [messages[0].public_id]} + ) + + assert "LEAKED_FROM_OTHER_SESSION" not in result + assert messages[0].content in result + + +@pytest.mark.asyncio +class TestGetReasoningChainWorkspace: + """Tests for _handle_get_reasoning_chain.""" + + async def test_returns_observation_chain( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Returns an observation and its chain.""" + _, _, _, _, _, _, docs_peer2, _ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_reasoning_chain( + ctx, {"observation_id": docs_peer2[0].id} + ) + + assert "Observation" in result + assert docs_peer2[0].content in result + + async def test_nonexistent_observation_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error for nonexistent observation ID.""" + ctx = make_workspace_ctx() + + result = await _handle_get_reasoning_chain( + ctx, {"observation_id": "nonexistent_id"} + ) + + assert "ERROR" in result + + async def test_missing_observation_id_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + ): + """Returns error when observation_id is missing.""" + ctx = make_workspace_ctx() + + result = await _handle_get_reasoning_chain(ctx, {}) + + assert "ERROR" in result + + async def test_invalid_direction_returns_error( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Returns error for invalid direction parameter.""" + _, _, _, _, _, _, docs_peer2, _ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_reasoning_chain( + ctx, + {"observation_id": docs_peer2[0].id, "direction": "invalid"}, + ) + + assert "ERROR" in result + + async def test_deductive_observation_shows_premises( + self, + db_session: AsyncSession, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Deductive observation shows premises in chain.""" + workspace, peer1, peer2, _, _, _, docs_peer2, _ = workspace_test_data + + # Create a deductive document with source_ids + deductive_doc = models.Document( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + content="User is probably a morning person who codes", + embedding=[0.3] * 1536, + level="deductive", + source_ids=[docs_peer2[0].id, docs_peer2[1].id], + ) + db_session.add(deductive_doc) + await db_session.commit() + await db_session.refresh(deductive_doc) + + ctx = make_workspace_ctx() + result = await _handle_get_reasoning_chain( + ctx, {"observation_id": deductive_doc.id, "direction": "premises"} + ) + + assert "Observation" in result + assert "Premises" in result + + +# ============================================================================= +# Tool Executor Tests +# ============================================================================= + + +@pytest.mark.asyncio +class TestWorkspaceToolExecutor: + """Tests for create_workspace_tool_executor.""" + + async def test_returns_callable( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """create_workspace_tool_executor returns an async callable.""" + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + + assert callable(executor) + + async def test_routes_workspace_tools( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """Workspace-specific tools are routed to workspace handlers.""" + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + + stats_result = await executor("get_workspace_stats", {}) + assert isinstance(stats_result, str) + assert "Workspace stats" in stats_result + assert "Most active peers" in stats_result + + async def test_falls_through_to_standard_handlers( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """Non-workspace tools fall through to standard handlers.""" + workspace, _, _, _, session, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + session_name=session.name, + ) + + # grep_messages is a standard handler, should fall through + result = await executor("grep_messages", {"text": "Test message"}) + + assert isinstance(result, str) + + async def test_unknown_tool_returns_error( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """Unknown tool name returns error.""" + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + + result = await executor("nonexistent_tool", {}) + + assert "Unknown tool" in result + + async def test_handles_exceptions_gracefully( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """Executor returns error strings instead of raising exceptions.""" + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + + # Missing required observer/observed/query parameters + result = await executor("search_memory", {}) + + assert isinstance(result, str) + assert "ERROR" in result + + async def test_get_peer_card_via_executor( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """get_peer_card routes through workspace handler with params.""" + workspace, peer1, peer2, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + + result = await executor( + "get_peer_card", + {"observer": peer1.name, "observed": peer2.name}, + ) + + assert isinstance(result, str) + # Should be from workspace handler (accepts observer/observed params) + assert "peer card" in result.lower() or "No peer card" in result + + +# ============================================================================= +# Regression: workspace-flat message visibility without a pinned session +# ============================================================================= + + +@pytest.mark.asyncio +class TestWorkspaceMessageToolsUnpinned: + """The workspace executor's observer='' sentinel must read as + 'no perspective scoping' (None) at the crud boundary. Under #882's + resolve_session_scope, an empty STRING is looked up as a real peer with + no session memberships and denies every result — so these tests run the + message tools with NO session_name, the primary workspace-chat shape.""" + + async def test_grep_messages_finds_content_without_session( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + result = await executor("grep_messages", {"text": "Test message"}) + + assert isinstance(result, str) + assert "No messages found" not in result + assert "Test message" in result + + async def test_date_range_finds_content_without_session( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + ) + result = await executor("get_messages_by_date_range", {"limit": 10}) + + assert isinstance(result, str) + assert "Found" in result + assert "No messages found" not in result + + async def test_session_allowlist_is_honored_when_set( + self, + db_session: AsyncSession, # pyright: ignore[reportUnusedParameter] + workspace_test_data: Any, + ): + """An allowlist naming no real session yields no results.""" + workspace, *_ = workspace_test_data + + executor = await create_workspace_tool_executor( + workspace_name=workspace.name, + session_allowlist=["no-such-session"], + ) + result = await executor("grep_messages", {"text": "Test message"}) + + assert isinstance(result, str) + assert "No messages found" in result + + +@pytest.mark.asyncio +async def test_workspace_prefetch_failure_degrades_to_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Prefetch errors must not fail the request (parity with the base + agent's try/except): the agent proceeds with no prefetched block.""" + from src.dialectic.workspace import WorkspaceDialecticAgent + + async def boom(*args: Any, **kwargs: Any) -> Any: + _ = (args, kwargs) + raise RuntimeError("stats query exploded") + + monkeypatch.setattr("src.dialectic.workspace.crud.get_workspace_stats", boom) + + agent = WorkspaceDialecticAgent(workspace_name="w") + result = await agent._prefetch_relevant_observations("q") # pyright: ignore[reportPrivateUsage] + + assert result is None diff --git a/tests/unified/README.md b/tests/unified/README.md index 0ef1d987..01f53ebb 100644 --- a/tests/unified/README.md +++ b/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 diff --git a/tests/unified/run.py b/tests/unified/run.py index c0848471..5ab4fb79 100644 --- a/tests/unified/run.py +++ b/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__": diff --git a/tests/unified/runner.py b/tests/unified/runner.py index 6d78b06b..c9e9d2e3 100644 --- a/tests/unified/runner.py +++ b/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) diff --git a/tests/unified/schema.py b/tests/unified/schema.py index b0fa84b4..31e30946 100644 --- a/tests/unified/schema.py +++ b/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[ diff --git a/tests/unified/test_cases/peer_isolation_test.json b/tests/unified/test_cases/peer_isolation_test.json new file mode 100644 index 00000000..2c87fced --- /dev/null +++ b/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 + } + ] + } + ] +} diff --git a/tests/unified/test_cases/workspace_chat_cross_peer.json b/tests/unified/test_cases/workspace_chat_cross_peer.json new file mode 100644 index 00000000..3d474c1d --- /dev/null +++ b/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 + } + ] + } + ] +} diff --git a/tests/unified/test_cases/workspace_chat_from_messages.json b/tests/unified/test_cases/workspace_chat_from_messages.json new file mode 100644 index 00000000..2fe417bd --- /dev/null +++ b/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 + } + ] + } + ] +} diff --git a/tests/unified/test_cases/workspace_chat_from_observations.json b/tests/unified/test_cases/workspace_chat_from_observations.json new file mode 100644 index 00000000..f40eae04 --- /dev/null +++ b/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 + } + ] + } + ] +} diff --git a/tests/unified/test_cases/workspace_chat_scope.json b/tests/unified/test_cases/workspace_chat_scope.json new file mode 100644 index 00000000..bf5c7243 --- /dev/null +++ b/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 + } + ] + } + ] +} diff --git a/tests/unified/test_schema.py b/tests/unified/test_schema.py new file mode 100644 index 00000000..250ff314 --- /dev/null +++ b/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()))