diff --git a/docs/v3/documentation/features/chat.mdx b/docs/v3/documentation/features/chat.mdx index 6aab6996..83324b2e 100644 --- a/docs/v3/documentation/features/chat.mdx +++ b/docs/v3/documentation/features/chat.mdx @@ -94,43 +94,6 @@ for await (const chunk of responseStream.iter_text()) { Streaming is useful for displaying real-time responses in chat interfaces or when asking complex questions that require longer answers. -## Structured Outputs - -When your application needs a machine-readable answer instead of prose, pass a schema as `response_format` and the answer is guaranteed to conform to it: - - -```python Python -from pydantic import BaseModel - -class OnboardingStatus(BaseModel): - completed: bool - remaining_steps: list[str] - -status = peer.chat( - "Has the user completed the onboarding flow?", - response_format=OnboardingStatus, -) -# status is a parsed OnboardingStatus instance -``` - -```typescript TypeScript -import { z } from 'zod'; - -const OnboardingStatus = z.object({ - completed: z.boolean(), - remainingSteps: z.array(z.string()), -}); - -const status = await peer.chat( - "Has the user completed the onboarding flow?", - { responseFormat: OnboardingStatus }, -); -// status is typed as z.infer -``` - - -The agent runs its full reasoning loop either way — only the final answer is formatted to your schema. See [Structured Outputs](/v3/documentation/features/advanced/structured-outputs) for the supported schema subset, streaming behavior, and best practices. - ## Integration Patterns ### Dynamic Prompt Enhancement @@ -226,6 +189,48 @@ const goals = await peer.chat("What are the user's main goals or objectives?"); ``` +## Workspace-Level Chat + +While `peer.chat()` queries knowledge about a single peer, `honcho.chat()` searches across **all peers and observations** in the workspace. This is useful for cross-peer analysis, discovering common themes, or asking workspace-wide questions. + + +```python Python +from honcho import Honcho + +honcho = Honcho() + +# Ask about the entire workspace +answer = honcho.chat("What are common themes across all users?") +print(answer) + +# With streaming +stream = honcho.chat_stream("Summarize all peer activity this week.") +for chunk in stream: + print(chunk, end="", flush=True) + +# Async +answer = await honcho.aio.chat("Which users have discussed topic X?") +``` + +```typescript TypeScript +import { Honcho } from '@honcho-ai/sdk'; + +const honcho = new Honcho({}); + +// Ask about the entire workspace +const answer = await honcho.chat("What are common themes across all users?"); +console.log(answer); + +// With streaming +const stream = await honcho.chatStream("Summarize all peer activity this week."); +for await (const chunk of stream) { + process.stdout.write(chunk); +} +``` + + +Workspace chat accepts `reasoning_level` and optional `session` scoping. For streaming, use the separate `chat_stream()` / `chatStream()` method rather than a `stream` parameter. + ## How Honcho Answers When you call `peer.chat(query)`: diff --git a/sdks/python/src/honcho/aio.py b/sdks/python/src/honcho/aio.py index 51797c72..025be152 100644 --- a/sdks/python/src/honcho/aio.py +++ b/sdks/python/src/honcho/aio.py @@ -399,6 +399,73 @@ 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, + ) -> 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 + 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, + ) -> 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 + 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 06a2887c..e021ee67 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 +from collections.abc import Generator, Mapping from typing import Any, Literal import httpx @@ -27,9 +27,10 @@ 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 .types import DialecticStreamResponse from .session import Session -from .utils import normalize_peers_to_dict, resolve_id +from .utils import normalize_peers_to_dict, parse_sse_stream, resolve_id logger = logging.getLogger(__name__) @@ -582,6 +583,90 @@ 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, + ) -> 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). + + 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 + 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, + ) -> 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 + 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 3fdbd677..1cf7d300 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 db8e237a..fadf83bc 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' @@ -12,6 +16,7 @@ import type { QueueStatusParams, QueueStatusResponse, SessionResponse, + WorkspaceChatResponse, WorkspaceResponse, } from './types/api' import { resolveId, transformQueueStatus } from './utils' @@ -49,6 +54,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. @@ -361,6 +367,43 @@ export class Honcho { ) } + private async _workspaceChat( + workspaceId: string, + params: { + query: string + stream?: boolean + session_id?: string + reasoning_level?: ReasoningLevel + } + ): Promise { + await this._ensureWorkspace() + return this._http.post( + `/${API_VERSION}/workspaces/${workspaceId}/chat`, + { body: params } + ) + } + + private async _workspaceChatStream( + workspaceId: string, + params: { + query: string + session_id?: string + reasoning_level?: ReasoningLevel + } + ): Promise { + await this._ensureWorkspace() + return this._http.stream( + 'POST', + `/${API_VERSION}/workspaces/${workspaceId}/chat`, + { + body: { + ...params, + stream: true, + }, + } + ) + } + // =========================================================================== // Public Methods // =========================================================================== @@ -803,6 +846,92 @@ 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. + * @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 + } + ): 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, + }) + 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. + * @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 + } + ): 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, + }) + + 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 90bff9f2..fa2545f0 100644 --- a/sdks/typescript/src/index.ts +++ b/sdks/typescript/src/index.ts @@ -55,6 +55,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 dbe4378d..71de5003 100644 --- a/sdks/typescript/src/types/api.ts +++ b/sdks/typescript/src/types/api.ts @@ -81,6 +81,17 @@ export interface PeerChatResponse { content: string | null } +export interface WorkspaceChatParams { + query: string + stream?: boolean + session_id?: string + reasoning_level?: 'minimal' | 'low' | 'medium' | 'high' | 'max' +} + +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 b34e4d17..09e8699f 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -66,16 +66,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", diff --git a/src/crud/workspace.py b/src/crud/workspace.py index c9040f55..640cbfaf 100644 --- a/src/crud/workspace.py +++ b/src/crud/workspace.py @@ -1,6 +1,7 @@ """CRUD helpers for workspace records and workspace deletion checks.""" from dataclasses import dataclass +from datetime import datetime from logging import getLogger from typing import Any @@ -535,3 +536,137 @@ 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, +) -> WorkspaceStats: + """Get aggregate statistics for a workspace. + + Args: + db: Database session + workspace_name: Name of the workspace + + Returns: + WorkspaceStats with peer, session, and message counts plus date range + """ + peer_count = int( + await db.scalar( + select(func.count(models.Peer.id)).where( + models.Peer.workspace_name == workspace_name + ) + ) + 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(models.Message.workspace_name == workspace_name) + ) + ).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, + ) + + +async def get_active_peers( + db: AsyncSession, + workspace_name: str, + limit: int = 20, + sort_by: str = "recent_activity", +) -> list[ActivePeer]: + """Get the most active peers in a workspace. + + Args: + db: Database session + workspace_name: Name of the workspace + limit: Maximum number of peers to return (default 20, max 50) + sort_by: Sort order — "recent_activity" (default) or "message_count" + + Returns: + List of ActivePeer objects with message counts and last-active dates + """ + if limit <= 0: + return [] + limit = min(limit, 50) + + # Subquery: aggregate messages per peer + 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(models.Message.workspace_name == workspace_name) + .group_by(models.Message.peer_name) + .subquery() + ) + + # Join with Peer to get only valid peers + stmt = ( + select( + models.Peer.name, + func.coalesce(subq.c.msg_count, 0).label("msg_count"), + subq.c.last_msg_at, + ) + .outerjoin(subq, models.Peer.name == subq.c.peer_name) + .where(models.Peer.workspace_name == workspace_name) + ) + + if sort_by == "message_count": + stmt = stmt.order_by(func.coalesce(subq.c.msg_count, 0).desc()) + else: + # Default: recent_activity — peers with most recent messages first + stmt = stmt.order_by(subq.c.last_msg_at.desc().nulls_last()) + + 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 d2f7e741..cfa6dac3 100644 --- a/src/dialectic/chat.py +++ b/src/dialectic/chat.py @@ -14,6 +14,7 @@ from src import crud, schemas 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.utils.config_helpers import get_configuration logger = logging.getLogger(__name__) @@ -161,3 +162,69 @@ 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, +) -> str: + """ + Answer a query across all peers in a workspace (workspace-level dialectic). + + Args: + workspace_name: Workspace identifier + session_name: Optional session scope for message tools + query: The question to answer about the workspace + reasoning_level: Level of reasoning to apply + response_model: Optional Pydantic model the answer must conform to. + + Returns: + The synthesized answer string + """ + 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, + ) + 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, +): + """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, + ) + 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..f38cb8bf 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,16 @@ class DialecticAgent: parent_category="dialectic", ) - return tool_executor, task_name, run_id, start_time + def _trace_name(self) -> str: + """Langfuse trace name for this agent's LLM calls.""" + return "dialectic_chat" + + 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(). @@ -487,7 +502,7 @@ class DialecticAgent: max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, messages=self.messages, max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, - trace_name="dialectic_chat", + trace_name=self._trace_name(), telemetry=self._telemetry_context(track_name="Dialectic Agent"), response_model=response_model, ), @@ -563,7 +578,7 @@ class DialecticAgent: max_tool_iterations=level_settings.MAX_TOOL_ITERATIONS, messages=self.messages, max_input_tokens=settings.DIALECTIC.MAX_INPUT_TOKENS, - trace_name="dialectic_chat", + trace_name=self._trace_name(), telemetry=self._telemetry_context(track_name="Dialectic Agent Stream"), response_model=response_model, ), diff --git a/src/dialectic/prompts.py b/src/dialectic/prompts.py index 948bff70..9f097784 100644 --- a/src/dialectic/prompts.py +++ b/src/dialectic/prompts.py @@ -235,3 +235,78 @@ 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. + +Unlike a peer-level agent that knows about one specific peer, you can search, compare, and correlate information about any and all peers — but you must discover relevant peers first and then query each peer relationship individually. + +## AVAILABLE TOOLS + +**Discovery Tools:** +- `get_workspace_stats`: Get workspace-level counts (peers, sessions, messages) and date range. Use this to orient yourself. +- `get_active_peers`: Get the most active peers ranked by recent activity or message count. Use this to 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 are provided in your query context. Use `get_active_peers` if you need to discover which peers are relevant, 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..3f40f44d --- /dev/null +++ b/src/dialectic/workspace.py @@ -0,0 +1,169 @@ +"""Workspace-level dialectic agent (DEV-1326). + +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, +) + +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, + ): + super().__init__( + workspace_name=workspace_name, + session_name=session_name, + observer="", + observed="", + metric_key=metric_key, + reasoning_level=reasoning_level, + session_id=session_id, + ) + # 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 + async with tracked_db("dialectic.workspace_prefetch", read_only=True) as db: + stats = await crud.get_workspace_stats(db, self.workspace_name) + if stats.peer_count == 0: + return None + peers = await crud.get_active_peers( + db, self.workspace_name, limit=_PREFETCH_ACTIVE_PEERS + ) + cards: dict[str, list[str]] = {} + 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 + + lines: list[str] = [ + 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: + date_range = f"{stats.oldest_message_at:%Y-%m-%d} to {stats.newest_message_at:%Y-%m-%d}" + lines.append(f"Date range: {date_range}") + 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.get(peer.name, [])[:8]: + lines.append(f" - {fact}") + return "\n".join(lines) + + def _prefetch_intro(self) -> str: + return ( + "Workspace overview and most-active peers with their 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_active_peers 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: reasoning chains traverse + # provenance across sessions, so the tool can't honor an allowlist. + if self.session_allowlist is not None: + tools = [t for t in tools if t.get("name") != "get_reasoning_chain"] + 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, + history_token_limit=settings.DIALECTIC.HISTORY_TOKEN_LIMIT, + run_id=self._run_id, + agent_type="workspace_dialectic", + parent_category="dialectic", + ) + + def _trace_name(self) -> str: + return "workspace_chat" + + 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/workspaces.py b/src/routers/workspaces.py index b7e15a71..86d9dbf2 100644 --- a/src/routers/workspaces.py +++ b/src/routers/workspaces.py @@ -1,8 +1,10 @@ """FastAPI routes for workspace resources and workspace-scoped operations.""" +import json import logging 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 sqlalchemy.ext.asyncio import AsyncSession @@ -11,8 +13,11 @@ from src import crud, schemas from src.config import settings from src.dependencies import db, read_db from src.deriver.enqueue import enqueue_deletion, enqueue_dream -from src.exceptions import AuthenticationException +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.schema_conversion import json_response_schema_to_pydantic from src.utils.search import search logger = logging.getLogger(__name__) @@ -249,3 +254,75 @@ 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": {}, + }, + }, + }, + dependencies=[Depends(require_auth(workspace_name="workspace_id"))], +) +async def chat( + workspace_id: str = Path(...), + options: schemas.WorkspaceChatOptions = Body(...), +): + """ + Query the entire workspace using natural language. Performs agentic search + and reasoning across ALL peers' representations and messages -- discovering + relevant peers first, then querying their memory -- to answer + workspace-wide questions ("what themes are common across users?", + "which peers discussed X?"). + """ + from typing import Any as _Any + + from pydantic import BaseModel as _BaseModel + + 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: _Any) -> _Any: + 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, + ) + ), + 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, + ) + return schemas.DialecticResponse(content=response if response else None) diff --git a/src/schemas/__init__.py b/src/schemas/__init__.py index 85d5da44..ca35e23a 100644 --- a/src/schemas/__init__.py +++ b/src/schemas/__init__.py @@ -49,6 +49,7 @@ from src.schemas.api import ( WebhookEndpointCreate, Workspace, WorkspaceBase, + WorkspaceChatOptions, WorkspaceCreate, WorkspaceGet, WorkspaceUpdate, @@ -108,6 +109,7 @@ __all__ = [ "ConclusionQuery", "DialecticOptions", "DialecticResponse", + "WorkspaceChatOptions", "DialecticStreamChunk", "DialecticStreamDelta", "Message", diff --git a/src/schemas/api.py b/src/schemas/api.py index 78e5a125..ea911d92 100644 --- a/src/schemas/api.py +++ b/src/schemas/api.py @@ -610,6 +610,37 @@ class DialecticOptions(BaseModel): 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") + ] + 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." + ), + ) + # scopes(#897): the observer-swap `scope` read option lands here once the + # scopes facade merges; workspace chat is the peer-unanchored read and + # scope becomes its narrowing parameter. + + @field_validator("query", mode="after") + @classmethod + def sanitize_query(cls, v: str) -> str: + return v.replace("\x00", "") + + class DialecticResponse(BaseModel): content: str | None diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index 9f214009..88f6a01a 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 @@ -785,6 +785,79 @@ 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, and date range of messages. Use this to understand the scale of the workspace before deciding how to explore it.", + "input_schema": { + "type": "object", + "properties": {}, + }, + }, + "get_active_peers": { + "name": "get_active_peers", + "description": "Get the most active peers in the workspace, ranked by recent activity or message count. Returns peer names with their message counts and last-active timestamps. Use this to discover which peers are most relevant.", + "input_schema": { + "type": "object", + "properties": { + "limit": { + "type": "integer", + "description": "Maximum number of peers to return (default: 20, max: 50)", + "default": 20, + }, + "sort_by": { + "type": "string", + "enum": ["recent_activity", "message_count"], + "description": "Sort order: 'recent_activity' for most recently active, 'message_count' for most messages (default: recent_activity)", + "default": "recent_activity", + }, + }, + }, + }, + "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 +879,32 @@ 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["get_active_peers"], + 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_active_peers"], + 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) @@ -2511,6 +2610,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. @@ -2596,7 +2696,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 @@ -2794,3 +2894,156 @@ def _estimate_tokens_safe(text: str | None) -> int | None: if not text: return None return _estimate_tokens(text) + + +# --------------------------------------------------------------------------- +# Workspace-level tool handlers (workspace chat, DEV-1326) +# +# 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="" which crud treats as "no +# perspective scoping" -- correct for a workspace-level read. +# --------------------------------------------------------------------------- + + +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) + return await _handle_get_peer_card(pair_ctx, tool_input) + + +async def _handle_get_workspace_stats( + ctx: ToolContext, tool_input: dict[str, Any] +) -> str: + """Workspace-level counts and message date range.""" + _ = 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) + 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}" + ) + return "Workspace stats:\n" + "\n".join(lines) + + +async def _handle_get_active_peers(ctx: ToolContext, tool_input: dict[str, Any]) -> str: + """Most active peers, ranked by recency or message count.""" + limit = min(_safe_int(tool_input.get("limit"), 20), 50) + sort_by = tool_input.get("sort_by", "recent_activity") + if sort_by not in ("recent_activity", "message_count"): + sort_by = "recent_activity" + async with tracked_db("workspace_tool.get_active_peers", read_only=True) as db: + peers = await crud.get_active_peers( + db, ctx.workspace_name, limit=limit, sort_by=sort_by + ) + if not peers: + return "No peers found in this workspace." + lines: list[str] = [] + 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})") + return f"Found {len(peers)} active peers (sorted by {sort_by}):\n" + "\n".join( + lines + ) + + +async def _handle_get_observation_context_workspace( + ctx: ToolContext, tool_input: dict[str, Any] +) -> "str | ToolResult": + """get_observation_context without perspective scoping (workspace read).""" + return await _handle_get_observation_context(replace(ctx, observer=""), tool_input) + + +# 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_active_peers": _handle_get_active_peers, + "get_peer_card": _handle_get_peer_card_by_name, + "get_observation_context": _handle_get_observation_context_workspace, + "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( + { + "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, + 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. + """ + return await create_tool_executor( + workspace_name=workspace_name, + observer="", + observed="", + session_name=session_name, + 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/representation.py b/src/utils/representation.py index 01e4b70a..fcd10081 100644 --- a/src/utils/representation.py +++ b/src/utils/representation.py @@ -725,3 +725,44 @@ def _safe_datetime_from_metadata( if isinstance(message_created_at, datetime): return _strip_microseconds_and_timezone(message_created_at) return _strip_microseconds_and_timezone(fallback_datetime) + + +def format_documents_with_attribution( + documents: Sequence[models.Document], + include_ids: bool = False, +) -> str: + """ + Format documents grouped by observer/observed pair with attribution headers. + + Groups documents by their (observer, observed) pair and formats each group + as a separate Representation section with a descriptive header. + + Args: + documents: Sequence of Document models to format + include_ids: Whether to include observation IDs + + Returns: + Formatted markdown string with attribution headers + """ + if not documents: + return "No observations found." + + # Group documents by (observer, observed) + groups: dict[tuple[str, str], list[models.Document]] = {} + for doc in documents: + key = (doc.observer, doc.observed) + if key not in groups: + groups[key] = [] + groups[key].append(doc) + + parts: list[str] = [] + for (observer, observed), group_docs in groups.items(): + if observer == observed: + parts.append(f"### About {observed}\n") + else: + parts.append(f"### About {observed} (observed by {observer})\n") + + rep = Representation.from_documents(group_docs) + parts.append(rep.format_as_markdown(include_ids=include_ids)) + + return "\n".join(parts) diff --git a/tests/conftest.py b/tests/conftest.py index 1ec64055..57cb799d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -791,6 +791,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" @@ -806,11 +812,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/test_workspace_chat.py b/tests/test_workspace_chat.py new file mode 100644 index 00000000..33a8db2d --- /dev/null +++ b/tests/test_workspace_chat.py @@ -0,0 +1,1198 @@ +"""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 +- Utility: format_documents_with_attribution() +""" + +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_active_peers, # pyright: ignore[reportPrivateUsage] + _handle_get_observation_context_workspace, # 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.representation import format_documents_with_attribution + +# ============================================================================= +# 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, + ) -> 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, + ) + + 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_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 + + +@pytest.mark.asyncio +class TestGetActivePeers: + """Tests for _handle_get_active_peers.""" + + async def test_returns_active_peers( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Returns active peers with message counts.""" + _, peer1, peer2, peer3, *_ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_active_peers(ctx, {}) + + assert "Found" in result + assert "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_returns_no_peers( + self, + db_session: AsyncSession, + ): + """Returns appropriate message for workspace with no peers.""" + 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_active_peers(ctx, {}) + + assert "No peers found" in result + + async def test_respects_limit( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Respects the limit parameter.""" + _ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_active_peers(ctx, {"limit": 1}) + + # Should only return 1 peer + assert "Found 1 active peers" in result + + async def test_sort_by_message_count( + self, + make_workspace_ctx: Callable[..., ToolContext], + workspace_test_data: Any, + ): + """Supports sorting by message count.""" + _ = workspace_test_data + ctx = make_workspace_ctx() + + result = await _handle_get_active_peers(ctx, {"sort_by": "message_count"}) + + assert "sorted by message_count" 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_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 _handle_get_observation_context_workspace.""" + + 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_workspace( + 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_workspace( + 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_workspace( + 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, + 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, + 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, + ) + + # Test get_workspace_stats + stats_result = await executor("get_workspace_stats", {}) + assert isinstance(stats_result, str) + assert "Workspace stats" in stats_result + + # Test get_active_peers + peers_result = await executor("get_active_peers", {}) + assert isinstance(peers_result, str) + assert "peers" in peers_result.lower() + + async def test_falls_through_to_standard_handlers( + self, + db_session: AsyncSession, + 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, + 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, + 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, + 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 + + +# ============================================================================= +# Utility Tests: format_documents_with_attribution() +# ============================================================================= + + +class TestFormatDocumentsWithAttribution: + """Tests for format_documents_with_attribution().""" + + def test_empty_documents(self): + """Returns fallback message for empty list.""" + result = format_documents_with_attribution([]) + assert result == "No observations found." + + def test_groups_by_peer_pair( + self, + workspace_test_data: Any, + ): + """Groups documents by observer/observed pair.""" + _, peer1, peer2, peer3, _, _, docs_peer2, docs_peer3 = workspace_test_data + + result = format_documents_with_attribution(docs_peer2 + docs_peer3) + + # Should have headers for both peer pairs + assert f"About {peer2.name}" in result + assert f"About {peer3.name}" in result + assert f"observed by {peer1.name}" in result + + def test_self_observation_header(self): + """Self-observation (observer==observed) uses simpler header.""" + doc = models.Document( + workspace_name="test", + observer="alice", + observed="alice", + content="I like coffee", + level="explicit", + embedding=[0.1] * 1536, + internal_metadata={ + "message_ids": [1], + "message_created_at": "2025-01-01T00:00:00Z", + }, + ) + doc.id = "test_id" + + result = format_documents_with_attribution([doc]) + + assert "### About alice\n" in result + # Should NOT have "observed by" for self-observation + assert "observed by" not in result + + def test_include_ids( + self, + workspace_test_data: Any, + ): + """include_ids=True includes observation IDs in output.""" + _, _, _, _, _, _, docs_peer2, _ = workspace_test_data + + result_with_ids = format_documents_with_attribution( + docs_peer2, include_ids=True + ) + result_without_ids = format_documents_with_attribution( + docs_peer2, include_ids=False + ) + + # With IDs should be longer or have [id: markers + assert len(result_with_ids) >= len(result_without_ids) + + def test_contains_document_content( + self, + workspace_test_data: Any, + ): + """Output contains the actual document content.""" + _, _, _, _, _, _, docs_peer2, _ = workspace_test_data + + result = format_documents_with_attribution(docs_peer2) + + assert "coffee" in result.lower() or "programming" in result.lower() + assert "remotely" in result.lower() or "works" in result.lower() 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..56e6b7e4 --- /dev/null +++ b/tests/unified/test_cases/workspace_chat_from_messages.json @@ -0,0 +1,59 @@ +{ + "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": { + "deriver": 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 + } + ] + } + ] +}