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

Add workspace-level chat (DEV-1326) 

POST /v3/workspaces/{workspace_id}/chat: agentic dialectic over the whole
workspace instead of a single (observer, observed) pair. Salvaged from
plastic-labs/honcho#373 and re-grown on today's DialecticAgent:

- WorkspaceDialecticAgent subclasses DialecticAgent via four new seams
  (_get_tools, _create_tool_executor, _prefetch_intro, _trace_name) instead
  of a base-class extraction; observer/observed use empty-string sentinels.
- Routing-accelerated prefetch: workspace stats + top-5 active peers with
  their self peer-cards (pure DB, ~7ms measured) so routing-obvious queries
  resolve without a discovery tool round.
- Observation search stays pair-scoped (matches per-pair vector namespaces;
  avoids workspace-flat top-k dilution): search_memory/get_peer_card take
  observer/observed as tool arguments, with pair attribution in results.
- workspace_chat / workspace_chat_stream orchestrators, WorkspaceChatOptions
  schema (scope param seam left for the #897 scopes facade), SSE streaming,
  structured output via response_format.
- crud: get_workspace_stats, get_active_peers; format_documents_with_attribution.
- SDKs: Python Honcho.chat/chat_stream + HonchoAio mirrors; TypeScript
  honcho.chat/chatStream.
- 46 tests (route, orchestrator preflight, tool handlers, executor routing,
  attribution formatting) + unified test cases + docs.

Co-Authored-By: doria <93405247+dr-frmr@users.noreply.github.com>
Co-Authored-By: Benjamin McCormick <docterformer@protonmail.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
adavyas 2026-07-23 16:47:29 -04:00
parent 4d3ab1c36b
commit fc9de0f0d9
24 changed files with 2817 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
docs/v3/documentation/features/chat.mdx
View File

@ -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:
<CodeGroup>
```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<typeof OnboardingStatus>
```
</CodeGroup>
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?");
```
</CodeGroup>
## 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.
<CodeGroup>
```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);
}
```
</CodeGroup>
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)`:

67
sdks/python/src/honcho/aio.py
View File

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

91
sdks/python/src/honcho/client.py
View File

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

4
sdks/python/src/honcho/http/routes.py
View File

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

129
sdks/typescript/src/client.ts
View File

@ -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<WorkspaceChatResponse> {
await this._ensureWorkspace()
return this._http.post<WorkspaceChatResponse>(
`/${API_VERSION}/workspaces/${workspaceId}/chat`,
{ body: params }
)
}
private async _workspaceChatStream(
workspaceId: string,
params: {
query: string
session_id?: string
reasoning_level?: ReasoningLevel
}
): Promise<Response> {
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<string | null> {
const validatedQuery = SearchQuerySchema.parse(query)
const resolvedSessionId = options?.session
? resolveId(options.session)
: undefined
const response = await this._workspaceChat(this.workspaceId, {
query: validatedQuery,
stream: false,
session_id: resolvedSessionId,
reasoning_level: options?.reasoningLevel,
})
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<DialecticStreamResponse> {
const validatedQuery = SearchQuerySchema.parse(query)
const resolvedSessionId = options?.session
? resolveId(options.session)
: undefined
const response = await this._workspaceChatStream(this.workspaceId, {
query: validatedQuery,
session_id: resolvedSessionId,
reasoning_level: options?.reasoningLevel,
})
return createDialecticStream(response)
}
/**
* Get the queue processing status, optionally scoped to an observer, sender, and/or session.
*

1
sdks/typescript/src/index.ts
View File

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

11
sdks/typescript/src/types/api.ts
View File

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

8
src/crud/__init__.py
View File

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

135
src/crud/workspace.py
View File

@ -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
]

67
src/dialectic/chat.py
View File

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

27
src/dialectic/core.py
View File

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

75
src/dialectic/prompts.py
View File

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

169
src/dialectic/workspace.py Normal file
View File

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

79
src/routers/workspaces.py
View File

@ -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)

2
src/schemas/__init__.py
View File

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

31
src/schemas/api.py
View File

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

257
src/utils/agent_tools.py
View File

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

41
src/utils/representation.py
View File

@ -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)

15
tests/conftest.py
View File

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

1198
tests/test_workspace_chat.py Normal file
View File

File diff suppressed because it is too large Load Diff

148
tests/unified/test_cases/peer_isolation_test.json Normal file
View File

@ -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
}
]
}
]
}

88
tests/unified/test_cases/workspace_chat_cross_peer.json Normal file
View File

@ -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
}
]
}
]
}

59
tests/unified/test_cases/workspace_chat_from_messages.json Normal file
View File

@ -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
}
]
}
]
}

85
tests/unified/test_cases/workspace_chat_from_observations.json Normal file
View File

@ -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
}
]
}
]
}