Merge origin/main into abigail/dev-2219

Bring in scope backfill jobs so the document_sources migration can be reconciled with the new source_ids property.
This commit is contained in:
ajspig 2026-08-18 13:31:08 -04:00
commit 1395407da3
42 changed files with 3105 additions and 233 deletions

View File

@ -77,6 +77,8 @@ model = "text-embedding-3-small"
# Optional provider request input cap. Useful for OpenAI-compatible embedding
# APIs with smaller limits, such as DashScope text-embedding-v4.
# max_batch_size = 10
# Optional client HTTP timeout in seconds (OpenAI + Gemini).
# timeout = 90.0
# Optional module-level endpoint overrides
# [embedding.model_config.overrides]

View File

@ -267,6 +267,7 @@ EMBEDDING_MAX_TOKENS_PER_REQUEST=300000
EMBEDDING_MODEL_CONFIG__TRANSPORT=openai # openai, gemini
EMBEDDING_MODEL_CONFIG__MODEL=text-embedding-3-small
EMBEDDING_MODEL_CONFIG__MAX_BATCH_SIZE=10 # optional per-request input cap
EMBEDDING_MODEL_CONFIG__TIMEOUT=90.0 # optional client HTTP timeout (seconds)
# Optional endpoint overrides
EMBEDDING_MODEL_CONFIG__OVERRIDES__BASE_URL=http://localhost:8000/v1
@ -279,6 +280,13 @@ document a per-request limit. Set it when an OpenAI-compatible embedding
provider accepts fewer inputs per request, such as DashScope
`text-embedding-v4` with a limit of 10.
`EMBEDDING_MODEL_CONFIG__TIMEOUT` is an optional client HTTP timeout in
seconds. OpenAI-compatible transports receive it as the SDK `timeout` kwarg
(omitted when unset, so the SDK default applies). Gemini converts it to
milliseconds on `http_options.timeout`, and keeps its existing 10-minute
default when unset. The value is validated at config load the same way as
LLM `provider_params.timeout` (positive, finite number of seconds).
Forwarding `dimensions=` to OpenAI-compatible providers is controlled by `EMBEDDING_MODEL_CONFIG__DIMENSIONS_MODE`:
- `auto` (default): forwards `dimensions=` when **the operator has explicitly set `EMBEDDING_VECTOR_DIMENSIONS`** — provenance, not value — and the configured model is not on the known-rejecting list (currently `text-embedding-ada-002`). Explicit `EMBEDDING_VECTOR_DIMENSIONS=1536` *does* trigger the forward; this is how `text-embedding-3-large` truncation to 1536 is expressed. Deployments that leave the setting unset get their existing behavior (`dimensions=` is not forwarded).

View File

@ -37,7 +37,7 @@ source ~/.zshrc # or ~/.bashrc
### Step 3: Install the Plugin
<Note>
This plugin requires [Bun](https://bun.sh). If you don't have it: `curl -fsSL https://bun.sh/install | bash`
This plugin requires [Node.js](https://nodejs.org) on your PATH.
</Note>
Open Claude Code and run:
@ -74,10 +74,11 @@ Claude will interview you about your personal preferences to kickstart a represe
- **Persistent Memory** — Claude remembers your preferences, projects, and context across sessions
- **Survives Context Wipes** — Even when Claude's context window resets, memory persists
- **Configurable Memory Injection** — Choose exactly what context is injected at session start and per turn
- **Git Awareness** — Detects branch switches, commits, and changes made outside Claude
- **Flexible Sessions** — Map sessions per directory, per git branch, or per chat instance
- **AI Self-Awareness** — Claude knows what it was working on, even after restarts
- **Cross-Tool Context** — Link workspaces across Claude Code, Cursor, and other hosts so context flows between tools
- **Secret Redaction** — Built-in patterns (plus your own) scrub secrets from tool summaries before upload
- **Team Support** — Multiple people can share a workspace and build context together
- **MCP Tools** — Search memory, query knowledge about you, and save insights
@ -97,8 +98,7 @@ All configuration lives in a single global file at `~/.honcho/config.json`. You
"hosts": {
"claude_code": {
"workspace": "claude_code", // Workspace for Claude Code sessions
"aiPeer": "claude", // AI identity in this workspace
"linkedHosts": ["cursor"] // Read context from other hosts (optional)
"aiPeer": "claude" // AI identity in this workspace
},
"cursor": {
"workspace": "cursor",
@ -112,6 +112,8 @@ All configuration lives in a single global file at `~/.honcho/config.json`. You
// Message handling
"saveMessages": true,
"saveToolUse": false, // Save [Tool] action summaries (default: false)
"saveGitEvents": false, // Save [Git External] state-change events (default: false)
"messageUpload": {
"maxUserTokens": null, // Truncate user messages (null = no limit)
"maxAssistantTokens": null, // Truncate assistant messages (null = no limit)
@ -124,6 +126,19 @@ All configuration lives in a single global file at `~/.honcho/config.json`. You
"ttlSeconds": 300, // Cache TTL for context
"skipDialectic": false // Skip dialectic chat() calls in user-prompt hook
},
"reasoningLevel": "medium", // Default dialectic reasoning tier: "minimal" | "low" | "medium" | "high" | "max"
// Memory injection (see "Memory Injection" below)
"injection": {
"sessionStart": ["directives", "summary", "peerCard"],
"perTurn": ["userContext"]
},
// On-demand recall tool (see "The honcho_remember Tool" below)
"rememberTool": false,
// Observation mode
"observationMode": "unified", // "unified" (default) | "directional"
// Endpoint
"endpoint": {
@ -132,7 +147,8 @@ All configuration lives in a single global file at `~/.honcho/config.json`. You
},
// Miscellaneous
"localContext": { "maxEntries": 50 },
"redactPatterns": [], // Extra regexes redacted from tool summaries (additive to built-in secret patterns)
"statusline": "on", // Memory statusline visibility: "on" | "off"
"enabled": true,
"logging": true,
@ -141,7 +157,82 @@ All configuration lives in a single global file at `~/.honcho/config.json`. You
}
```
### Session Strategies
## Memory Injection
The `injection` config block controls exactly what memory is injected into Claude's context, on two surfaces: **once at session start** and **per prompt**. Each surface selects zero or more components; retrieval knobs shape what those components emit.
Configure it interactively with `/honcho:config` (under the memory injection settings), by asking Claude to use `set_config`, or by editing `~/.honcho/config.json` directly.
### Session-Start Components
Injected once when a session opens. Default: `["directives", "summary", "peerCard"]`.
| Component | What it injects |
| --- | --- |
| `directives` | Static memory-usage guidance — tells Claude to treat injected memory as background, use `chat`/`search` for recall, and save insights with `create_conclusion` |
| `summary` | The session's long summary narrative (skipped on a fresh session) |
| `peerCard` | Your peer card — a structured identity/attribute list |
| `peerRepresentation` | Your full derived representation, injected at full length |
| `briefing` | A nudge for Claude to call the `get_briefing` MCP tool instead of injecting the summary and peer card inline. The tool call renders as an expandable row in the UI, so you can see exactly what was loaded. Use it *in place of* `summary`/`peerCard`, not alongside them |
```json
{ "injection": { "sessionStart": ["directives", "briefing"] } }
```
### Per-Turn Components
Injected with each non-trivial prompt. Default: `["userContext"]`.
| Component | What it injects |
| --- | --- |
| `userContext` | A fresh, prompt-scoped context fetch for *you* — conclusions selected by semantic search over your representation, shaped by the retrieval knobs below |
| `assistantContext` | The same context fetch, but for the AI peer — what Honcho has derived about the assistant itself |
| `sessionContext` | Recent raw messages from the currently mapped Honcho session, which can span other Claude instances sharing the session name |
| `dialectic` | A reasoned `chat()` answer over your representation, seeded from `dialecticTemplate`. Off by default — it is much slower than a context fetch, so it runs on its own time budget |
### Retrieval Knobs
| Field | Default | Description |
| --- | --- | --- |
| `searchTopK` | `10` | Top-K conclusions pulled by the context fetch's semantic search |
| `maxConclusions` | `15` | Max conclusions injected per context fetch |
| `searchMaxDistance` | `0.6` | Max cosine distance for the semantic search — lower is stricter |
| `searchQuerySource` | `"prompt"` | What drives the per-turn search: the raw `"prompt"` or extracted `"topics"` |
| `sessionContextTokens` | `1500` | Token budget for the `sessionContext` message fetch |
| `dialecticTemplate` | compact factual recall | Query template for the `dialectic` component; the user's prompt is substituted into `%{user_query}` |
| `dialecticReasoning` | `"medium"` | Reasoning tier for the per-turn `dialectic` call — kept separate from the top-level `reasoningLevel` so per-turn dialectic can stay cheap |
If injected context feels off-topic, lower `searchMaxDistance` (stricter relevance); if it feels too sparse, raise it or bump `searchTopK`.
### Injection Visibility
By default, per-turn components report a one-line summary in the terminal instead of printing their full contents. To see exactly what a component injects, list it in `showContents`:
```json
{ "injection": { "showContents": ["userContext", "sessionContext"] } }
```
Components not listed still inject — they just stay quiet about it.
## The `honcho_remember` Tool
An experimental on-demand recall tool. When enabled, Claude gets a `honcho_remember` MCP tool that fans out up to 5 parallel dialectic queries about you and returns per-question answers — useful before starting a task, when catching up ("where were we?"), or whenever your history could shape the response.
To enable it, just ask Claude: *"Set my Honcho rememberTool config to true"* (it uses the `set_config` tool). Or set it in `~/.honcho/config.json`:
```json
{
"hosts": {
"claude_code": { "rememberTool": true }
}
}
```
Then restart Claude Code — MCP tools register at startup.
When it's on, the injected session-start directives steer Claude to use it proactively as the primary recall path.
## Session Strategies
Session strategy controls how Honcho maps your conversations to sessions:
@ -153,7 +244,20 @@ Session strategy controls how Honcho maps your conversations to sessions:
Session names are prefixed with your `peerName` by default (e.g., `alice-my-project`). Set `sessionPeerPrefix: false` if you're the only user and want shorter names.
### Host-Aware Configuration
Linked git worktrees resolve to their main repository's session, so a worktree shares memory with the repo it belongs to.
## Observation Mode
Controls how Honcho stores and retrieves conclusions about you. Change it via `set_config` or edit `config.json` directly. Requires a Claude Code restart.
| Mode | Behavior | Best for |
| --- | --- | --- |
| `unified` (default) | All agents write to your self-observation collection (`observer=you, observed=you`). Conclusions are portable — switch between agents without losing memory. | Most users — a unified context hub across agents |
| `directional` | Each AI peer keeps its own separate view of you (`observer=aiPeer, observed=you`). | Multi-peer workspaces where you want isolated per-agent representations |
Switching modes doesn't automatically migrate existing conclusions — each mode reads from a different collection. The [plugin repository](https://github.com/plastic-labs/claude-honcho) ships a `migrate-observations.py` script to copy conclusions between collections.
## Host-Aware Configuration
The plugin auto-detects which tool is running it (Claude Code, Cursor, etc.) and reads the matching block from `hosts`. Each host gets its own workspace and AI peer name, so data stays separated by default.
@ -163,28 +267,7 @@ The plugin auto-detects which tool is running it (Claude Code, Cursor, etc.) and
3. `CURSOR_PROJECT_DIR` env var (Cursor child process)
4. Default: `claude_code`
### Linking Hosts for Cross-Tool Context
If you use both Claude Code and Cursor, you can link them so context from one is readable in the other. Writes always stay in the current host's workspace — linking only adds read access.
```jsonc
{
"hosts": {
"claude_code": {
"workspace": "claude_code",
"aiPeer": "claude",
"linkedHosts": ["cursor"] // Claude Code can read Cursor's context
},
"cursor": {
"workspace": "cursor",
"aiPeer": "cursor",
"linkedHosts": ["claude_code"] // Cursor can read Claude Code's context
}
}
}
```
Or use `/honcho:config` and select **Workspace > Linking** to set this up interactively.
Each host block can also carry its own `apiKey` (useful when different tools authenticate against different Honcho orgs) and override most settings — `sessionStrategy`, `injection`, `rememberTool`, `observationMode`, and more.
### Global Override
@ -237,6 +320,14 @@ Multiple people can share context by pointing to the same workspace. Each person
Both Alice and Bob write to the `team-acme` workspace. Their sessions are namespaced (e.g., `alice-my-project`, `bob-my-project`) so data doesn't collide, but Honcho's dialectic reasoning can draw on context from both users.
## Secret Redaction
Tool-capture summaries are scrubbed against built-in secret patterns (API keys, tokens, credentials) before upload. Add your own patterns with `redactPatterns` — an array of regexes applied on top of the defaults:
```json
{ "redactPatterns": ["internal-[a-z0-9]+", "ACME_SECRET_\\w+"] }
```
## Logging
The plugin logs activity to `~/.honcho/` and to Claude Code's verbose mode, so you can see exactly how Honcho is being used — what context is loaded at session start, what messages are saved, and what context is injected into Claude's prompts. Set `logging` to `false` in your config (or `HONCHO_LOGGING=false`) to disable file logging.
@ -247,20 +338,30 @@ The plugin provides these tools via MCP:
| Tool | Description |
|------|-------------|
| `search` | Semantic search across session messages |
| `chat` | Query Honcho's knowledge about the user |
| `search` | Semantic search across session messages and saved conclusions |
| `chat` | Query Honcho's knowledge about the user (dialectic reasoning) |
| `create_conclusion` | Save insights about the user to memory |
| `list_conclusions` | List saved conclusions |
| `query_conclusions` | Semantic search over saved conclusions |
| `delete_conclusion` | Delete a conclusion by ID |
| `get_briefing` | Load the session briefing: session summary + peer card |
| `get_context` | Retrieve the full context object (representation + peer card) |
| `get_representation` | Retrieve the user's representation string |
| `get_config` | View current configuration and status |
| `set_config` | Change any configuration field programmatically |
| `honcho_remember` | Fan-out dialectic recall (only registered when `rememberTool: true`) |
## Skills (Slash Commands)
| Command | Description |
| ------- | ----------- |
| `/honcho:status` | Show current memory status and connection info |
| `/honcho:config` | Interactive configuration menu |
| `/honcho:config` | Interactive configuration menu (including memory injection settings) |
| `/honcho:setup` | First-time setup — validate API key and create config |
| `/honcho:interview` | Interview to capture stable, cross-project user preferences |
| `/honcho:briefing` | Load the session briefing via a visible tool call |
| `/honcho:import` | Backfill past Claude Code sessions into Honcho memory |
| `/honcho:insights` | Distill memory into CLAUDE.md edits, style rules, and skill ideas |
### The Interview
@ -289,6 +390,8 @@ Environment variables work for initial bootstrap (before a config file exists).
| `HONCHO_ENDPOINT` | No | `production` | `production`, `local`, or a full URL |
| `HONCHO_ENABLED` | No | `true` | Set to `false` to disable |
| `HONCHO_SAVE_MESSAGES` | No | `true` | Set to `false` to stop saving messages |
| `HONCHO_SAVE_TOOL_USE` | No | `false` | Set to `true` to save tool action summaries |
| `HONCHO_SAVE_GIT_EVENTS` | No | `false` | Set to `true` to save external git state-change events |
| `HONCHO_LOGGING` | No | `true` | Set to `false` to disable file logging to `~/.honcho/` |
### Using a local Honcho instance

View File

@ -239,38 +239,11 @@ To teach Goose the recommended memory flow, save the [instructions](https://raw.
---
## Optional Configuration
## Workspace
You can target a specific workspace by adding an extra header. It's optional.
Every workspace-scoped tool takes a `workspace_id` argument. You can also set `X-Honcho-Workspace-ID` on the connection; that value fills `workspace_id` when the argument is omitted.
| Header | Default | Description |
|--------|---------|-------------|
| `Authorization` | *required* | `Bearer hch-your-key-here` |
| `X-Honcho-Workspace-ID` | `"default"` | Isolate memory per project |
Example with all headers (Claude Desktop format):
```json
{
"mcpServers": {
"honcho": {
"command": "npx",
"args": [
"mcp-remote",
"https://mcp.honcho.dev",
"--header",
"Authorization:${AUTH_HEADER}",
"--header",
"X-Honcho-Workspace-ID:${WORKSPACE_ID}"
],
"env": {
"AUTH_HEADER": "Bearer hch-your-key-here",
"WORKSPACE_ID": "my-project"
}
}
}
}
```
Use `list_workspaces` to discover IDs (each result includes metadata and `created_at`), or `create_workspace` if none fit, then reuse the same ID for subsequent tool calls.
---

View File

@ -26,15 +26,11 @@ A Cloudflare Worker that implements the [Model Context Protocol (MCP)](https://m
}
```
### Optional Headers
| Header | Default | Description |
| --- | --- | --- |
| `X-Honcho-Workspace-ID` | `"default"` | Workspace to operate in |
Every workspace-scoped tool takes a `workspace_id` argument. If you set `X-Honcho-Workspace-ID` on the connection, that value fills `workspace_id` when the argument is omitted. Use `list_workspaces` to discover IDs.
## Available Tools
**Workspace:** `inspect_workspace` (aggregates metadata, configuration, and peer/session IDs), `list_workspaces` (enumerates accessible workspaces), `search` (semantic search scoped by optional peer/session params), `get_metadata`, `set_metadata`
**Workspace:** `list_workspaces` (id, metadata, created_at), `create_workspace` (get-or-create with optional metadata), `inspect_workspace` (aggregates metadata, configuration, and peer/session IDs), `search` (semantic search scoped by optional peer/session params), `get_metadata`, `set_metadata`
**Peers:** `create_peer`, `list_peers`, `chat`, `get_peer_card`, `set_peer_card`, `get_peer_context`, `get_representation`
@ -50,7 +46,7 @@ A Cloudflare Worker that implements the [Model Context Protocol (MCP)](https://m
src/
index.ts # Worker entry point — parse config, delegate to MCP handler
server.ts # createServer() — registers all tools on an McpServer
config.ts # HonchoConfig, parseConfig(), createClient()
config.ts # HonchoConfig, parseConfig(), createClientFactory()
types.ts # ToolContext, result helpers
tools/
workspace.ts # inspect, list, search, metadata

View File

@ -5,7 +5,7 @@
"": {
"name": "honcho-mcp-proxy",
"dependencies": {
"@honcho-ai/sdk": "^2.0.0",
"@honcho-ai/sdk": "^2.2.0",
"@modelcontextprotocol/sdk": "^1.26.0",
"agents": "^0.4.0",
"nanoid": "^5.1.7",
@ -107,7 +107,7 @@
"@esbuild/win32-x64": ["@esbuild/win32-x64@0.25.4", "", { "os": "win32", "cpu": "x64" }, "sha512-nOT2vZNw6hJ+z43oP1SPea/G/6AbN6X+bGNhNuq8NtRHy4wsMhw765IKLNmnjek7GvjWBYQ8Q5VBoYTFg9y1UQ=="],
"@honcho-ai/sdk": ["@honcho-ai/sdk@2.0.1", "", { "dependencies": { "@types/node": "^24.0.1", "zod": "4.0.0" } }, "sha512-y/Wk49C0N1miI9BZTNWFIbzdUkMZfP4Do/EJ1q4lEIK+FAOKxQgces/zET3kPKV3zF9sOUl2pXrFb/XKYayeYw=="],
"@honcho-ai/sdk": ["@honcho-ai/sdk@2.2.0", "", { "dependencies": { "zod": "4.0.0" } }, "sha512-SyygN+BrpUB2fRjhwcYmT+tcEhHrKmbj9nOZLVUFY7M5YBswJ+mZb/CeLpNbRh+QQTU8F8JWY3lQat95c6nwmA=="],
"@hono/node-server": ["@hono/node-server@1.19.9", "", { "peerDependencies": { "hono": "^4" } }, "sha512-vHL6w3ecZsky+8P5MD+eFfaGTyCeOHUIFYMGpQGbrBTSmNNoxv0if69rEZ5giu36weC5saFuznL411gRX7bJDw=="],
@ -177,8 +177,6 @@
"@types/lodash": ["@types/lodash@4.17.23", "", {}, "sha512-RDvF6wTulMPjrNdCoYRC8gNR880JNGT8uB+REUpC2Ns4pRqQJhGz90wh7rgdXDPpCczF3VGktDuFGVnz8zP7HA=="],
"@types/node": ["@types/node@24.1.0", "", { "dependencies": { "undici-types": "~7.8.0" } }, "sha512-ut5FthK5moxFKH2T1CUOC6ctR67rQRvvHdFLCD2Ql6KXmMuCrjsSsRI9UsLCm9M18BMwClv4pn327UvB7eeO1w=="],
"@vercel/oidc": ["@vercel/oidc@3.1.0", "", {}, "sha512-Fw28YZpRnA3cAHHDlkt7xQHiJ0fcL+NRcIqsocZQUSmbzeIKRpwttJjik5ZGanXP+vlA4SbTg+AbA3bP363l+w=="],
"accepts": ["accepts@2.0.0", "", { "dependencies": { "mime-types": "^3.0.0", "negotiator": "^1.0.0" } }, "sha512-5cvg6CtKwfgdmVqY1WIiXKc3Q1bkRqGLi+2W/6ao+6Y7gu/RCwRuAhGEzh5B4KlszSuTLgZYuqFqo5bImjNKng=="],
@ -471,8 +469,6 @@
"undici": ["undici@7.12.0", "", {}, "sha512-GrKEsc3ughskmGA9jevVlIOPMiiAHJ4OFUtaAH+NhfTUSiZ1wMPIQqQvAJUrJspFXJt3EBWgpAeoHEDVT1IBug=="],
"undici-types": ["undici-types@7.8.0", "", {}, "sha512-9UJ2xGDvQ43tYyVMpuHlsgApydB8ZKfVYTsLDhXkFL/6gfkp+U8xTGdh8pMJv1SpZna0zxG1DwsKZsreLbXBxw=="],
"unenv": ["unenv@2.0.0-rc.17", "", { "dependencies": { "defu": "^6.1.4", "exsolve": "^1.0.4", "ohash": "^2.0.11", "pathe": "^2.0.3", "ufo": "^1.6.1" } }, "sha512-B06u0wXkEd+o5gOCMl/ZHl5cfpYbDZKAT+HWTL+Hws6jWu7dCiqBBXXXzMFcFVJb8D4ytAnYmxJA83uwOQRSsg=="],
"unpipe": ["unpipe@1.0.0", "", {}, "sha512-pjy2bYhSsufwWlKwPc+l3cN7+wuJlK6uz0YdJEOlQDbl6jo/YlPi4mb8agUkVC8BF7V8NuzeyPNqRksA3hztKQ=="],

View File

@ -4,12 +4,21 @@
The simplest way to use Honcho for a standard user/assistant conversation. Three steps using the general tools.
Every workspace-scoped tool takes `workspace_id`. The simplest setup is for the client to set `X-Honcho-Workspace-ID` on the connection — then omit `workspace_id` on every call. Do not list or create a workspace just to rediscover a header that is already set.
If the header is unset and you don't already know the workspace:
1. Call `list_workspaces` and pick the workspace whose id or metadata best matches this work.
2. If none fit, call `create_workspace` with a descriptive id (and optional metadata like `{ "project": "...", "purpose": "..." }`).
3. Reuse that same `workspace_id` for the rest of the conversation.
### 1. Start a conversation (once per conversation)
Create a session and set up the user and assistant peers:
```
create_session
workspace_id: "<workspace-id>"
session_id: "<unique-id>"
```
@ -17,12 +26,15 @@ Then add peers to the session:
```
create_peer
workspace_id: "<workspace-id>"
peer_id: "<user-name>"
create_peer
workspace_id: "<workspace-id>"
peer_id: "Assistant"
add_peers_to_session
workspace_id: "<workspace-id>"
session_id: "<session_id>"
peers:
- peer_id: "<user-name>"
@ -39,6 +51,7 @@ Store the `session_id` for the rest of this conversation.
```
chat
workspace_id: "<workspace-id>"
peer_id: "Assistant"
query: "What communication style does this user prefer?"
target_peer_id: "<user-name>"
@ -58,6 +71,7 @@ This calls Honcho's reasoning system to answer your question about the user, gro
```
add_messages_to_session
workspace_id: "<workspace-id>"
session_id: "<session_id>"
messages:
- peer_id: "<user-name>"
@ -88,8 +102,9 @@ The full API for advanced use cases.
| Tool | When to use |
| --- | --- |
| `inspect_workspace` | Inspect a single workspace's details |
| `list_workspaces` | Enumerate available workspaces |
| `list_workspaces` | Discover available workspaces (id, metadata, created_at). No `workspace_id` needed. |
| `create_workspace` | Get or create a workspace when none of the listed ones fit |
| `inspect_workspace` | Inspect a single workspace's details. Requires `workspace_id`. |
| `search` | Semantic search across messages — scope with optional `peer_id` or `session_id` params |
| `get_metadata` | Read metadata for workspace, peer, or session (scope with optional `peer_id` or `session_id`) |
| `set_metadata` | Store metadata for workspace, peer, or session (scope with optional `peer_id` or `session_id`) |

View File

@ -15,7 +15,7 @@
"deploy:staging": "wrangler deploy --env staging"
},
"dependencies": {
"@honcho-ai/sdk": "^2.1.0",
"@honcho-ai/sdk": "^2.2.0",
"@modelcontextprotocol/sdk": "^1.26.0",
"agents": "^0.4.0",
"nanoid": "^5.1.7",

View File

@ -21,7 +21,7 @@
},
{
"name": "X-Honcho-Workspace-ID",
"description": "Optional. Target Honcho workspace; defaults to 'default' when omitted.",
"description": "Optional. Default Honcho workspace for tool calls. When set, it fills workspace_id on tools.",
"isRequired": false,
"isSecret": false
}

View File

@ -3,11 +3,13 @@ import { Honcho } from "@honcho-ai/sdk";
export interface HonchoConfig {
apiKey: string;
baseUrl: string;
workspaceId: string;
/** From X-Honcho-Workspace-ID when set. */
workspaceId?: string;
}
export interface Env {
HONCHO_API_URL?: string;
ALERT_WEBHOOK_URL?: string;
}
/**
@ -19,6 +21,9 @@ export interface Env {
* instance (see the "Self-Hosted Honcho" section in README.md). It is
* intentionally not exposed as a request header: routing public requests
* to an internal URL would be a latency and security regression.
*
* Optional `X-Honcho-Workspace-ID` becomes the default `workspace_id` on
* tools. If the header is omitted, each tool call must pass `workspace_id`.
*/
export function parseConfig(request: Request, env: Env = {}): HonchoConfig {
const authHeader = request.headers.get("Authorization");
@ -33,17 +38,60 @@ export function parseConfig(request: Request, env: Env = {}): HonchoConfig {
throw new Error("Authorization header is empty after 'Bearer '.");
}
const workspaceId =
request.headers.get("X-Honcho-Workspace-ID")?.trim() || undefined;
return {
apiKey,
baseUrl: env.HONCHO_API_URL?.trim() || "https://api.honcho.dev",
workspaceId: request.headers.get("X-Honcho-Workspace-ID")?.trim() || "default",
workspaceId,
};
}
export function createClient(config: HonchoConfig): Honcho {
export const MISSING_WORKSPACE_ID_MESSAGE =
"Missing workspace_id. Pass workspace_id on the next tool call, or set the X-Honcho-Workspace-ID header on the connection so it is used automatically.";
export function resolveWorkspaceId(
config: HonchoConfig,
workspaceId?: string,
): string {
const id = workspaceId?.trim() || config.workspaceId?.trim();
if (!id) {
throw new Error(MISSING_WORKSPACE_ID_MESSAGE);
}
return id;
}
export function createClient(
config: HonchoConfig,
workspaceId: string,
): Honcho {
return new Honcho({
apiKey: config.apiKey,
baseURL: config.baseUrl,
workspaceId: config.workspaceId,
workspaceId,
});
}
/** Client used only for credential-scoped ops (list workspaces). */
export function createUnscopedClient(config: HonchoConfig): Honcho {
return new Honcho({
apiKey: config.apiKey,
baseURL: config.baseUrl,
});
}
export function createClientFactory(
config: HonchoConfig,
): (workspaceId?: string) => Honcho {
const cache = new Map<string, Honcho>();
return (workspaceId?: string) => {
const id = resolveWorkspaceId(config, workspaceId);
let client = cache.get(id);
if (!client) {
client = createClient(config, id);
cache.set(id, client);
}
return client;
};
}

View File

@ -1,5 +1,10 @@
import { createMcpHandler } from "agents/mcp";
import { parseConfig, createClient, type Env } from "./config.js";
import {
parseConfig,
createClientFactory,
createUnscopedClient,
type Env,
} from "./config.js";
import { createServer } from "./server.js";
const CORS_ORIGIN = "*";
@ -15,6 +20,7 @@ const CORS_HEADERS = {
};
const PROTECTED_RESOURCE_PATH = "/.well-known/oauth-protected-resource";
const HEALTHCHECK_TIMEOUT_MS = 15_000;
function resourceUrl(request: Request): string {
return new URL(request.url).origin;
@ -25,6 +31,41 @@ function authorizationServer(env: Env): string {
}
export default {
// Probes the authorization server API to confirm it is reachable and healthy.
async scheduled(
_controller: ScheduledController,
env: Env,
_executionCtx: ExecutionContext,
): Promise<void> {
const upstream = `${authorizationServer(env)}/health`;
let failure: string | null = null;
try {
const response = await fetch(upstream, {
signal: AbortSignal.timeout(HEALTHCHECK_TIMEOUT_MS),
});
if (!response.ok) failure = `HTTP ${response.status}`;
} catch (e) {
failure = e instanceof Error ? e.message : String(e);
}
if (!failure) return;
console.error(`healthcheck failed: ${upstream} — ${failure}`);
if (!env.ALERT_WEBHOOK_URL) return;
const alertResponse = await fetch(env.ALERT_WEBHOOK_URL, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
content: `🚨 honcho-mcp cannot reach upstream API (${upstream}): ${failure}`,
}),
signal: AbortSignal.timeout(HEALTHCHECK_TIMEOUT_MS),
});
if (!alertResponse.ok) {
throw new Error(`alert webhook failed: HTTP ${alertResponse.status}`);
}
},
async fetch(
request: Request,
env: Env,
@ -42,6 +83,7 @@ export default {
resource: resourceUrl(request),
authorization_servers: [authorizationServer(env)],
bearer_methods_supported: ["header"],
scopes_supported: ["read", "write"],
},
{ headers: CORS_HEADERS },
);
@ -66,8 +108,11 @@ export default {
}
try {
const honcho = createClient(config);
const server = createServer({ honcho, config });
const server = createServer({
config,
clientFor: createClientFactory(config),
unscoped: createUnscopedClient(config),
});
const handler = createMcpHandler(server, {
route: "/",
corsOptions: {

View File

@ -1,7 +1,7 @@
import { z } from "zod";
import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import type { ToolContext } from "../types.js";
import { textResult, errorResult } from "../types.js";
import { textResult, errorResult, workspaceIdSchema } from "../types.js";
export function register(server: McpServer, ctx: ToolContext) {
// ── list_conclusions ────────────────────────────────────────────────
@ -14,6 +14,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns conclusion objects with pagination metadata.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
target_peer_id: z
.string()
@ -23,9 +24,9 @@ export function register(server: McpServer, ctx: ToolContext) {
),
},
},
async ({ peer_id, target_peer_id }) => {
async ({ workspace_id, peer_id, target_peer_id }) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const scope = target_peer_id
? peer.conclusionsOf(target_peer_id)
: peer.conclusions;
@ -61,6 +62,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns an array of matching conclusions ranked by relevance.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
query: z.string().describe("Semantic search query."),
target_peer_id: z
@ -71,19 +73,26 @@ export function register(server: McpServer, ctx: ToolContext) {
.number()
.optional()
.describe("Max results to return."),
filters: z
.record(z.string(), z.unknown())
.optional()
.describe(
'Optional: filter criteria, e.g. {"level": ["deductive", "inductive"]} to only return conclusions derived during dreaming. Levels: explicit (extracted directly from messages), deductive, inductive, contradiction. See https://honcho.dev/docs/v3/documentation/features/advanced/using-filters',
),
},
},
async ({ peer_id, query, target_peer_id, top_k }) => {
async ({ workspace_id, peer_id, query, target_peer_id, top_k, filters }) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const scope = target_peer_id
? peer.conclusionsOf(target_peer_id)
: peer.conclusions;
const conclusions = await scope.query(query, top_k);
const conclusions = await scope.query(query, top_k, undefined, filters);
return textResult(
conclusions.map((c) => ({
id: c.id,
content: c.content,
level: c.level,
observer_id: c.observerId,
observed_id: c.observedId,
session_id: c.sessionId,
@ -108,6 +117,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the number of conclusions created.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
target_peer_id: z
.string()
@ -123,9 +133,15 @@ export function register(server: McpServer, ctx: ToolContext) {
),
},
},
async ({ peer_id, target_peer_id, conclusions, session_id }) => {
async ({
workspace_id,
peer_id,
target_peer_id,
conclusions,
session_id,
}) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const scope = peer.conclusionsOf(target_peer_id);
const params = conclusions.map((content) => ({
content,
@ -149,9 +165,11 @@ export function register(server: McpServer, ctx: ToolContext) {
{
description: [
"Delete a specific conclusion by ID.",
"Use query_conclusions or list_conclusions to find the ID first.",
"Use this to remove incorrect or outdated knowledge.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
target_peer_id: z
.string()
@ -159,9 +177,9 @@ export function register(server: McpServer, ctx: ToolContext) {
conclusion_id: z.string().describe("The conclusion to delete."),
},
},
async ({ peer_id, target_peer_id, conclusion_id }) => {
async ({ workspace_id, peer_id, target_peer_id, conclusion_id }) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const scope = peer.conclusionsOf(target_peer_id);
await scope.delete(conclusion_id);
return textResult("Conclusion deleted successfully");

View File

@ -1,7 +1,7 @@
import { z } from "zod";
import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import type { ToolContext } from "../types.js";
import { textResult, errorResult } from "../types.js";
import { textResult, errorResult, workspaceIdSchema } from "../types.js";
export function register(server: McpServer, ctx: ToolContext) {
// ── create_peer ─────────────────────────────────────────────────────
@ -14,6 +14,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the peer ID and any configuration that was set.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("Unique identifier for the peer."),
configuration: z
.object({
@ -25,9 +26,11 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Optional peer configuration."),
},
},
async ({ peer_id, configuration }) => {
async ({ workspace_id, peer_id, configuration }) => {
try {
const peer = await ctx.honcho.peer(peer_id, { configuration });
const peer = await ctx.clientFor(workspace_id).peer(peer_id, {
configuration,
});
return textResult({ peer_id: peer.id, configuration: peer.configuration });
} catch (e) {
return errorResult(
@ -42,15 +45,17 @@ export function register(server: McpServer, ctx: ToolContext) {
"list_peers",
{
description: [
"List peers in the current workspace (paginated).",
"List peers in the given workspace (paginated).",
"Use this to discover which users and agents exist.",
"Returns peer IDs with pagination metadata.",
].join("\n"),
inputSchema: {},
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
},
},
async () => {
async ({ workspace_id }) => {
try {
const page = await ctx.honcho.peers();
const page = await ctx.clientFor(workspace_id).peers();
return textResult({
peers: page.items.map((p) => ({ id: p.id })),
total: page.total,
@ -75,6 +80,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns a natural-language answer, or 'None' if no relevant information exists.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The peer to query about."),
query: z.string().describe("Natural-language question."),
target_peer_id: z
@ -93,9 +99,16 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Reasoning effort. Higher = more detailed but slower."),
},
},
async ({ peer_id, query, target_peer_id, session_id, reasoning_level }) => {
async ({
workspace_id,
peer_id,
query,
target_peer_id,
session_id,
reasoning_level,
}) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const result = await peer.chat(query, {
target: target_peer_id,
session: session_id,
@ -120,6 +133,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns an array of fact strings, or null if no card exists yet.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
target_peer_id: z
.string()
@ -129,9 +143,9 @@ export function register(server: McpServer, ctx: ToolContext) {
),
},
},
async ({ peer_id, target_peer_id }) => {
async ({ workspace_id, peer_id, target_peer_id }) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const card = await peer.getCard(target_peer_id);
return textResult(card ?? "No peer card found.");
} catch (e) {
@ -152,6 +166,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the updated peer card.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
peer_card: z
.array(z.string())
@ -164,9 +179,9 @@ export function register(server: McpServer, ctx: ToolContext) {
),
},
},
async ({ peer_id, peer_card, target_peer_id }) => {
async ({ workspace_id, peer_id, peer_card, target_peer_id }) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const result = await peer.setCard(peer_card, target_peer_id);
return textResult(result ?? "Peer card set successfully");
} catch (e) {
@ -187,6 +202,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns an object with representation, peer_card, peer_id, and target_id.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
target_peer_id: z
.string()
@ -202,9 +218,15 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Optional: max number of conclusions to include."),
},
},
async ({ peer_id, target_peer_id, search_query, max_conclusions }) => {
async ({
workspace_id,
peer_id,
target_peer_id,
search_query,
max_conclusions,
}) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const context = await peer.context({
target: target_peer_id,
searchQuery: search_query,
@ -234,6 +256,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns a formatted string of conclusions.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer."),
target_peer_id: z
.string()
@ -254,6 +277,7 @@ export function register(server: McpServer, ctx: ToolContext) {
},
},
async ({
workspace_id,
peer_id,
target_peer_id,
session_id,
@ -261,7 +285,7 @@ export function register(server: McpServer, ctx: ToolContext) {
max_conclusions,
}) => {
try {
const peer = await ctx.honcho.peer(peer_id);
const peer = await ctx.clientFor(workspace_id).peer(peer_id);
const rep = await peer.representation({
target: target_peer_id,
session: session_id,

View File

@ -7,6 +7,7 @@ import {
formatMessage,
formatMessages,
formatSessionSummaries,
workspaceIdSchema,
} from "../types.js";
export function register(server: McpServer, ctx: ToolContext) {
@ -20,12 +21,13 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the session ID.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("Unique identifier for the session."),
},
},
async ({ session_id }) => {
async ({ workspace_id, session_id }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
return textResult({ session_id: session.id });
} catch (e) {
return errorResult(
@ -40,15 +42,17 @@ export function register(server: McpServer, ctx: ToolContext) {
"list_sessions",
{
description: [
"List sessions in the current workspace (paginated).",
"List sessions in the given workspace (paginated).",
"Use this to discover existing conversations.",
"Returns session IDs with pagination metadata.",
].join("\n"),
inputSchema: {},
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
},
},
async () => {
async ({ workspace_id }) => {
try {
const page = await ctx.honcho.sessions();
const page = await ctx.clientFor(workspace_id).sessions();
return textResult({
sessions: page.items.map((s) => ({ id: s.id })),
total: page.total,
@ -72,12 +76,13 @@ export function register(server: McpServer, ctx: ToolContext) {
"This cannot be undone.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to delete."),
},
},
async ({ session_id }) => {
async ({ workspace_id, session_id }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
await session.delete();
return textResult("Session deleted successfully");
} catch (e) {
@ -98,6 +103,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the new cloned session ID.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to clone."),
message_id: z
.string()
@ -107,9 +113,9 @@ export function register(server: McpServer, ctx: ToolContext) {
),
},
},
async ({ session_id, message_id }) => {
async ({ workspace_id, session_id, message_id }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
const cloned = await session.clone(message_id);
return textResult({ session_id: cloned.id });
} catch (e) {
@ -129,6 +135,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Use this to bring participants into a conversation.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to add peers to."),
peers: z
.array(
@ -152,9 +159,10 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Peers to add — plain IDs or objects with per-session config."),
},
},
async ({ session_id, peers }) => {
async ({ workspace_id, session_id, peers }) => {
try {
const session = await ctx.honcho.session(session_id);
const honcho = ctx.clientFor(workspace_id);
const session = await honcho.session(session_id);
const additions = peers.map((p) => {
if (typeof p === "string") return p;
const config: { observeMe?: boolean | null; observeOthers?: boolean | null } = {};
@ -182,15 +190,16 @@ export function register(server: McpServer, ctx: ToolContext) {
"Remove one or more peers from a session.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to remove peers from."),
peer_ids: z
.array(z.string())
.describe("Peer IDs to remove."),
},
},
async ({ session_id, peer_ids }) => {
async ({ workspace_id, session_id, peer_ids }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
await session.removePeers(peer_ids);
return textResult("Peers removed from session successfully");
} catch (e) {
@ -211,12 +220,13 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns an array of peer IDs.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to query."),
},
},
async ({ session_id }) => {
async ({ workspace_id, session_id }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
const peers = await session.peers();
return textResult(peers.map((p) => p.id));
} catch (e) {
@ -237,12 +247,13 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns a single JSON object.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to inspect."),
},
},
async ({ session_id }) => {
async ({ workspace_id, session_id }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
const [peers, messagePage, summaries] = await Promise.all([
session.peers(),
session.messages(),
@ -273,6 +284,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Each message must specify the peer_id of the author.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to add messages to."),
messages: z
.array(
@ -288,15 +300,16 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Messages to add."),
},
},
async ({ session_id, messages }) => {
async ({ workspace_id, session_id, messages }) => {
try {
const session = await ctx.honcho.session(session_id);
const peerCache = new Map<string, Awaited<ReturnType<typeof ctx.honcho.peer>>>();
const honcho = ctx.clientFor(workspace_id);
const session = await honcho.session(session_id);
const peerCache = new Map<string, Awaited<ReturnType<typeof honcho.peer>>>();
const sessionMessages = [];
for (const msg of messages) {
let peer = peerCache.get(msg.peer_id);
if (!peer) {
peer = await ctx.honcho.peer(msg.peer_id);
peer = await honcho.peer(msg.peer_id);
peerCache.set(msg.peer_id, peer);
}
sessionMessages.push(
@ -325,6 +338,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the first page of messages with pagination metadata.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to get messages from."),
filters: z
.record(z.string(), z.unknown())
@ -332,9 +346,9 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Optional metadata filter criteria."),
},
},
async ({ session_id, filters }) => {
async ({ workspace_id, session_id, filters }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
const page = await session.messages(filters);
return textResult({
messages: formatMessages(page.items),
@ -360,13 +374,14 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns the message object.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session the message belongs to."),
message_id: z.string().describe("The message ID to fetch."),
},
},
async ({ session_id, message_id }) => {
async ({ workspace_id, session_id, message_id }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
const message = await session.getMessage(message_id);
return textResult(formatMessage(message));
} catch (e) {
@ -388,6 +403,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Returns messages, summary, and session ID.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
session_id: z.string().describe("The session to get context for."),
summary: z
.boolean()
@ -399,9 +415,9 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Target token budget for the context window."),
},
},
async ({ session_id, summary, tokens }) => {
async ({ workspace_id, session_id, summary, tokens }) => {
try {
const session = await ctx.honcho.session(session_id);
const session = await ctx.clientFor(workspace_id).session(session_id);
const context = await session.context({ summary, tokens });
return textResult({
session_id: context.sessionId,

View File

@ -1,7 +1,7 @@
import { z } from "zod";
import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import type { ToolContext } from "../types.js";
import { textResult, errorResult } from "../types.js";
import { textResult, errorResult, workspaceIdSchema } from "../types.js";
export function register(server: McpServer, ctx: ToolContext) {
// ── schedule_dream ──────────────────────────────────────────────────
@ -14,6 +14,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"Use this after a long conversation to improve Honcho's memory quality.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z.string().describe("The observer peer to dream for."),
target_peer_id: z
.string()
@ -27,9 +28,9 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Optional: scope the dream to a session."),
},
},
async ({ peer_id, target_peer_id, session_id }) => {
async ({ workspace_id, peer_id, target_peer_id, session_id }) => {
try {
await ctx.honcho.scheduleDream({
await ctx.clientFor(workspace_id).scheduleDream({
observer: peer_id,
observed: target_peer_id,
session: session_id,
@ -52,11 +53,13 @@ export function register(server: McpServer, ctx: ToolContext) {
"Use this to check if Honcho is still processing messages before querying for insights.",
"Returns work unit counts: total, completed, in-progress, and pending.",
].join("\n"),
inputSchema: {},
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
},
},
async () => {
async ({ workspace_id }) => {
try {
const status = await ctx.honcho.queueStatus();
const status = await ctx.clientFor(workspace_id).queueStatus();
return textResult(status);
} catch (e) {
return errorResult(

View File

@ -1,7 +1,36 @@
import { z } from "zod";
import {
BadRequestError,
HonchoError,
UnprocessableEntityError,
type PageResponse,
type WorkspaceResponse,
} from "@honcho-ai/sdk";
import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import type { ToolContext } from "../types.js";
import { textResult, errorResult, formatMessages } from "../types.js";
import { resolveWorkspaceId } from "../config.js";
import {
textResult,
errorResult,
formatMessages,
workspaceIdSchema,
} from "../types.js";
function withAdminKeyHint(prefix: string, e: unknown): string {
const message = e instanceof Error ? e.message : String(e);
const denied =
e instanceof HonchoError && (e.status === 401 || e.status === 403);
if (!denied) return `${prefix}: ${message}`;
return `${prefix}: ${message}. This operation is only possible with an admin API key.`;
}
function formatWorkspace(workspace: WorkspaceResponse) {
return {
id: workspace.id,
metadata: workspace.metadata ?? {},
created_at: workspace.created_at,
};
}
export function register(server: McpServer, ctx: ToolContext) {
// ── inspect_workspace ───────────────────────────────────────────────
@ -9,23 +38,27 @@ export function register(server: McpServer, ctx: ToolContext) {
"inspect_workspace",
{
description: [
"Inspect the current workspace at a glance.",
"Inspect a workspace at a glance.",
"Aggregates workspace metadata, configuration, peer IDs, and session IDs.",
"Returns the first page of peers/sessions with total counts.",
].join("\n"),
inputSchema: {},
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
},
},
async () => {
async ({ workspace_id }) => {
try {
const [metadata, configuration, peerPage, sessionPage] = await Promise.all([
ctx.honcho.getMetadata(),
ctx.honcho.getConfiguration(),
ctx.honcho.peers(),
ctx.honcho.sessions(),
]);
const honcho = ctx.clientFor(workspace_id);
const [metadata, configuration, peerPage, sessionPage] =
await Promise.all([
honcho.getMetadata(),
honcho.getConfiguration(),
honcho.peers(),
honcho.sessions(),
]);
return textResult({
workspace_id: ctx.honcho.workspaceId,
workspace_id: honcho.workspaceId,
metadata,
configuration,
peer_count: peerPage.total,
@ -47,24 +80,82 @@ export function register(server: McpServer, ctx: ToolContext) {
{
description: [
"List workspaces accessible to the current credentials (paginated).",
"Use this to discover available workspaces before selecting or switching context.",
"Returns workspace IDs with pagination metadata.",
"Skip this if the connection already set X-Honcho-Workspace-ID — that header is the workspace; omit workspace_id on other tools.",
"Use this only when the header is unset and you don't already know the workspace ID.",
"Returns each workspace's id, metadata, and created_at. If none fit, call create_workspace.",
].join("\n"),
inputSchema: {},
inputSchema: {
page: z
.number()
.int()
.min(1)
.optional()
.describe("Page number (1-indexed)."),
size: z
.number()
.int()
.min(1)
.max(100)
.optional()
.describe("Results per page (max 100)."),
},
},
async () => {
async ({ page, size }) => {
try {
const page = await ctx.honcho.workspaces();
const result = await ctx.unscoped.http.post<
PageResponse<WorkspaceResponse>
>("/v3/workspaces/list", {
body: {},
query: { page, size },
});
return textResult({
workspaces: page.items.map((id) => ({ id })),
total: page.total,
page: page.page,
pages: page.pages,
workspaces: result.items.map(formatWorkspace),
total: result.total,
page: result.page,
pages: result.pages,
});
} catch (e) {
return errorResult(
`Failed to list workspaces: ${e instanceof Error ? e.message : String(e)}`,
return errorResult(withAdminKeyHint("Failed to list workspaces", e));
}
},
);
// ── create_workspace ────────────────────────────────────────────────
server.registerTool(
"create_workspace",
{
description: [
"Get or create a workspace with the given ID.",
"Skip this if the connection already set X-Honcho-Workspace-ID — that header pins the workspace without a create call.",
"Use this only when the header is unset and list_workspaces has no suitable workspace.",
"Optional metadata helps future list_workspaces calls identify what the workspace is for.",
"Returns the workspace id, metadata, and created_at.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
metadata: z
.record(z.string(), z.unknown())
.optional()
.describe(
"Optional key-value metadata to store on the workspace (e.g. project, purpose).",
),
},
},
async ({ workspace_id, metadata }) => {
try {
const id = resolveWorkspaceId(ctx.config, workspace_id);
const workspace = await ctx.unscoped.http.post<WorkspaceResponse>(
"/v3/workspaces",
{
body: {
id,
metadata,
},
},
);
return textResult(formatWorkspace(workspace));
} catch (e) {
return errorResult(withAdminKeyHint("Failed to create workspace", e));
}
},
);
@ -74,13 +165,16 @@ export function register(server: McpServer, ctx: ToolContext) {
"search",
{
description: [
"Semantic search across messages. Scope is determined by which optional params are provided:",
"Semantic search across messages and, when peer_id is given, that peer's saved conclusions.",
"Message scope is determined by which optional params are provided:",
"- No scope params: search all messages in the workspace.",
"- peer_id only: search messages authored by that peer across all sessions.",
"- session_id only: search messages within that session.",
"Returns an array of matching messages with their content, peer, and session info.",
"Conclusions require peer_id (self-conclusions are searched; conclusion IDs are usable with delete_conclusion).",
"Returns {messages, conclusions}.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
query: z.string().describe("Search query."),
peer_id: z
.string()
@ -90,21 +184,95 @@ export function register(server: McpServer, ctx: ToolContext) {
.string()
.optional()
.describe("Optional: scope search to messages in this session."),
message_limit: z
.number()
.optional()
.describe("Optional: max message results (1-100, default 10)."),
message_filters: z
.record(z.string(), z.unknown())
.optional()
.describe(
'Optional: filters for the message search, e.g. {"created_at": {"gte": "2026-01-01"}}. See https://honcho.dev/docs/v3/documentation/features/advanced/using-filters',
),
conclusion_top_k: z
.number()
.optional()
.describe("Optional: max conclusion results (default 10)."),
conclusion_filters: z
.record(z.string(), z.unknown())
.optional()
.describe(
'Optional: filters for the conclusion search, e.g. {"level": ["deductive", "inductive"]} to only return conclusions derived during dreaming. Levels: explicit (extracted directly from messages), deductive, inductive, contradiction. The session_id param does not scope conclusions; use {"session_id": ...} here for that.',
),
},
},
async ({ query, peer_id, session_id }) => {
async ({
workspace_id,
query,
peer_id,
session_id,
message_limit,
message_filters,
conclusion_top_k,
conclusion_filters,
}) => {
try {
let messages;
if (session_id) {
const session = await ctx.honcho.session(session_id);
messages = await session.search(query);
} else if (peer_id) {
const peer = await ctx.honcho.peer(peer_id);
messages = await peer.search(query);
} else {
messages = await ctx.honcho.search(query);
}
return textResult(formatMessages(messages));
const honcho = ctx.clientFor(workspace_id);
const peer = peer_id ? await honcho.peer(peer_id) : null;
const messageOptions = {
filters: message_filters,
limit: message_limit,
};
const searchMessages = async () => {
if (session_id) {
const session = await honcho.session(session_id);
return session.search(query, messageOptions);
}
if (peer) {
return peer.search(query, messageOptions);
}
return honcho.search(query, messageOptions);
};
// Conclusion search needs an (observer, observed) pair, so it only
// runs when peer_id is given.
const searchConclusions = async () => {
if (!peer) {
return [];
}
try {
return await peer.conclusions.query(
query,
conclusion_top_k,
undefined,
conclusion_filters,
);
} catch (e) {
if (
conclusion_filters &&
(e instanceof BadRequestError ||
e instanceof UnprocessableEntityError)
) {
throw e;
}
return [];
}
};
const [messages, conclusions] = await Promise.all([
searchMessages(),
searchConclusions(),
]);
return textResult({
messages: formatMessages(messages),
conclusions: conclusions.map((c) => ({
id: c.id,
content: c.content,
level: c.level,
created_at: c.createdAt,
})),
});
} catch (e) {
return errorResult(
`Search failed: ${e instanceof Error ? e.message : String(e)}`,
@ -124,6 +292,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"- session_id only: get session metadata.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
peer_id: z
.string()
.optional()
@ -134,17 +303,18 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Optional: get metadata for this session."),
},
},
async ({ peer_id, session_id }) => {
async ({ workspace_id, peer_id, session_id }) => {
try {
const honcho = ctx.clientFor(workspace_id);
let metadata;
if (session_id) {
const session = await ctx.honcho.session(session_id);
const session = await honcho.session(session_id);
metadata = await session.getMetadata();
} else if (peer_id) {
const peer = await ctx.honcho.peer(peer_id);
const peer = await honcho.peer(peer_id);
metadata = await peer.getMetadata();
} else {
metadata = await ctx.honcho.getMetadata();
metadata = await honcho.getMetadata();
}
return textResult(metadata);
} catch (e) {
@ -167,6 +337,7 @@ export function register(server: McpServer, ctx: ToolContext) {
"- session_id only: set session metadata.",
].join("\n"),
inputSchema: {
workspace_id: workspaceIdSchema(ctx),
metadata: z
.record(z.string(), z.unknown())
.describe("Key-value pairs to set as metadata."),
@ -180,18 +351,19 @@ export function register(server: McpServer, ctx: ToolContext) {
.describe("Optional: set metadata for this session."),
},
},
async ({ metadata, peer_id, session_id }) => {
async ({ workspace_id, metadata, peer_id, session_id }) => {
try {
const honcho = ctx.clientFor(workspace_id);
if (session_id) {
const session = await ctx.honcho.session(session_id);
const session = await honcho.session(session_id);
await session.setMetadata(metadata);
return textResult("Session metadata set successfully");
} else if (peer_id) {
const peer = await ctx.honcho.peer(peer_id);
const peer = await honcho.peer(peer_id);
await peer.setMetadata(metadata);
return textResult("Peer metadata set successfully");
} else {
await ctx.honcho.setMetadata(metadata);
await honcho.setMetadata(metadata);
return textResult("Workspace metadata set successfully");
}
} catch (e) {

View File

@ -1,10 +1,27 @@
import { z } from "zod";
import type { Honcho, Message, Summary, SessionSummaries } from "@honcho-ai/sdk";
import type { CallToolResult } from "@modelcontextprotocol/sdk/types.js";
import type { HonchoConfig } from "./config.js";
export interface ToolContext {
honcho: Honcho;
config: HonchoConfig;
/** Return a Honcho client scoped to the given workspace (or the header default). */
clientFor: (workspaceId?: string) => Honcho;
/** Client used only for credential-scoped ops (list workspaces). */
unscoped: Honcho;
}
/**
* Always optional at the schema layer so a missing value reaches clientFor,
* which returns a clear error (header or workspace_id on the next call).
*/
export function workspaceIdSchema(ctx: ToolContext) {
const fromHeader = ctx.config.workspaceId;
const description = fromHeader
? `Workspace to operate in. The connection already set X-Honcho-Workspace-ID=${fromHeader}; omit this argument unless you need a different workspace.`
: "Workspace to operate in. Prefer the client setting X-Honcho-Workspace-ID on the connection — then you can omit this on every call. Only pass it (or use list_workspaces / create_workspace) when the header is unset.";
const field = z.string().optional().describe(description);
return fromHeader ? field.default(fromHeader) : field;
}
export function textResult(

View File

@ -8,6 +8,10 @@ rules = [
{ type = "Text", globs = ["**/*.md"], fallthrough = true },
]
# Executes the scheduled() handler
[triggers]
crons = ["*/5 * * * *"]
[env.production]
name = "honcho-mcp"

View File

@ -60,7 +60,7 @@ If you're unsure, list your available tools and look for Honcho memory tools (an
You need a Honcho API key — get one free at <https://app.honcho.dev> (starts with `hch-`). Then connect via the path you picked above — a purpose-built integration (recommended), or a raw connection:
- **MCP** — point your client at `https://mcp.honcho.dev` with two headers: `Authorization: Bearer hch-your-key-here` and `X-Honcho-User-Name: YourName` (what Honcho should call the user). Optional: `X-Honcho-Assistant-Name` (default `Assistant`) and `X-Honcho-Workspace-ID` (default `default`; set it to isolate memory per project). Restart the client fully after adding config. Per-client config snippets (Claude Desktop, Cursor, Codex, Windsurf, VS Code, Cline, Zed) are in the [MCP integration guide](https://honcho.dev/docs/v3/guides/integrations/mcp.md). Once connected, the server tells your assistant how to use the tools automatically.
- **MCP** — point your client at `https://mcp.honcho.dev` with `Authorization: Bearer hch-your-key-here`. Optional `X-Honcho-Workspace-ID` fills the `workspace_id` tool argument when omitted; otherwise pass `workspace_id` on each call (use `list_workspaces` to discover IDs). Restart the client fully after adding config. Per-client config snippets (Claude Desktop, Cursor, Codex, Windsurf, VS Code, Cline, Zed) are in the [MCP integration guide](https://honcho.dev/docs/v3/guides/integrations/mcp.md). Once connected, the server tells your assistant how to use the tools automatically.
- **CLI** — use the `honcho-cli` skill.
---

View File

@ -394,6 +394,15 @@ class ConfiguredEmbeddingModelSettings(BaseModel):
dimensions_mode: EmbeddingDimensionsMode = "auto"
encoding_format_mode: EmbeddingEncodingFormatMode = "auto"
max_batch_size: Annotated[int, Field(gt=0)] | None = None
# Client HTTP timeout in seconds. OpenAI receives seconds; Gemini converts to ms.
timeout: float | None = None
@field_validator("timeout", mode="before")
@classmethod
def _validate_timeout(cls, v: Any) -> float | None:
if v is None:
return None
return coerce_provider_timeout(v)
@model_validator(mode="before")
@classmethod
@ -431,6 +440,15 @@ class EmbeddingModelConfig(BaseModel):
api_key: str | None = None
base_url: str | None = None
max_batch_size: Annotated[int, Field(gt=0)] | None = None
# Client HTTP timeout in seconds. OpenAI receives seconds; Gemini converts to ms.
timeout: float | None = None
@field_validator("timeout", mode="before")
@classmethod
def _validate_timeout(cls, v: Any) -> float | None:
if v is None:
return None
return coerce_provider_timeout(v)
@model_validator(mode="before")
@classmethod
@ -556,6 +574,7 @@ def resolve_embedding_model_config(
api_key=api_key,
base_url=configured.overrides.base_url,
max_batch_size=configured.max_batch_size,
timeout=configured.timeout,
)

View File

@ -49,12 +49,16 @@ from .representation import (
)
from .scope import (
add_sessions_to_scope,
clear_scope_backfill_status,
get_or_create_scopes,
get_scope_backfill_status,
get_scope_or_raise,
get_scope_sessions,
get_scopes,
invalidate_scope_peer_cache,
remove_session_from_scope,
resolve_scope_peers,
update_scope_backfill_status,
)
from .session import (
SessionDeletionResult,
@ -137,12 +141,16 @@ __all__ = [
"get_working_representation",
# Scope
"add_sessions_to_scope",
"clear_scope_backfill_status",
"get_or_create_scopes",
"get_scope_backfill_status",
"get_scope_or_raise",
"get_scope_sessions",
"get_scopes",
"invalidate_scope_peer_cache",
"remove_session_from_scope",
"resolve_scope_peers",
"update_scope_backfill_status",
# Session
"SessionDeletionResult",
"get_sessions",

View File

@ -7,17 +7,26 @@ authoritative, user-unwritable flag) and ``{"observe_me": false}`` in
and never speaks.
See ``src/utils/scopes.py`` for the namespace helpers.
Membership only affects messages ingested *after* a session is added to a
scope. Conclusions already derived are neither backfilled on add nor
reconciled on removal.
Messages ingested after a membership change flow to the scope via the normal
deriver fan-out. Retroactive changes are handled by queue jobs: adding a session
that already has messages enqueues a ``scope_backfill`` task (copy of the
session's explicit documents into the scope's collections) and removing a
session enqueues a ``scope_removal`` task (soft-delete of the session's
documents plus dependent derived documents).
"""
from collections.abc import Sequence
from datetime import datetime, timezone
from logging import getLogger
from typing import Any, Literal
from typing import cast as py_cast
from sqlalchemy import Select, select
from sqlalchemy import Select, Text, cast, select, update
from sqlalchemy.dialects.postgresql import ARRAY, JSONB, array
from sqlalchemy.engine import CursorResult
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.functions import func
from src import models, schemas
from src.cache.client import safe_cache_delete
@ -38,11 +47,21 @@ from .workspace import get_or_create_workspace
logger = getLogger(__name__)
# Key inside the scope peer's internal_metadata that holds per-session
# backfill job status: {<session_name>: {state, updated_at[, docs_copied]}}.
BACKFILL_STATUS_KEY = "backfill_status"
ScopeBackfillState = Literal["pending", "completed", "failed"]
# Internal metadata stamped on every scope peer at creation. `kind` is the
# authoritative scope flag and lives here — NOT in `configuration` — because
# `configuration` is user-writable (`PeerCreate`/`PeerUpdate` accept a free-form
# dict, and `update_peer` replaces it wholesale), so a user could forge or clear
# the flag. `internal_metadata` appears in no API schema at all.
#
# Backfill job status shares this field under BACKFILL_STATUS_KEY below. Every
# write to it must be a JSONB merge scoped to that key, never a wholesale
# replacement, or the `kind` flag goes with it and the peer stops being a scope.
SCOPE_PEER_INTERNAL_METADATA: dict[str, str] = {
"kind": SCOPE_KIND,
}
@ -355,9 +374,9 @@ async def add_sessions_to_scope(
Each membership is a ``session_peers`` row for the scope peer with
``observe_others=true, observe_me=false`` — exactly what a hand-built
observer peer would carry. No backfill happens here: membership only affects
messages ingested after this call, and conclusions already derived are left
as they are.
observer peer would carry. Sessions that already have messages get a
``scope_backfill`` queue task so their existing explicit documents are
copied into the scope's collections; fresh sessions need nothing.
Args:
db: Database session
@ -400,6 +419,23 @@ async def add_sessions_to_scope(
await db.commit()
# Backfill: only sessions that already have messages need it.
# Imported lazily: src.deriver.enqueue imports crud at module level.
from src.deriver.enqueue import enqueue_scope_backfill
msg_result = await db.execute(
select(models.Message.session_name)
.where(models.Message.workspace_name == workspace_name)
.where(models.Message.session_name.in_(requested))
.distinct()
)
for session_with_messages in sorted({row[0] for row in msg_result.all()}):
await enqueue_scope_backfill(
workspace_name,
scope_peer=scope_peer_name(scope_name),
session_name=session_with_messages,
)
async def remove_session_from_scope(
db: AsyncSession,
@ -411,8 +447,9 @@ async def remove_session_from_scope(
Remove a session from a scope by ending the scope peer's membership.
Ends the membership the same way the generic remove-peer path does (sets
``left_at``). Conclusions derived while the session was a member are left in
place — nothing reconciles them.
``left_at``), then enqueues a ``scope_removal`` reconciliation task that
soft-deletes the session's documents from the scope's collections along
with any derived documents whose support left the scope.
Args:
db: Database session
@ -424,6 +461,8 @@ async def remove_session_from_scope(
ResourceNotFoundException: If the scope or session does not exist
"""
# Lazy import for the same circular-import reason as add_sessions_to_scope.
from src.deriver.enqueue import enqueue_scope_removal
from .session import remove_peers_from_session
await get_scope_or_raise(db, workspace_name, scope_name)
@ -436,3 +475,124 @@ async def remove_session_from_scope(
# This *is* the supported path for ending scope membership.
_allow_scope_peers=True,
)
await enqueue_scope_removal(
workspace_name,
scope_peer=scope_peer_name(scope_name),
session_name=session_name,
)
async def update_scope_backfill_status(
db: AsyncSession,
workspace_name: str,
scope_peer: str,
session_name: str,
*,
state: ScopeBackfillState,
docs_copied: int | None = None,
) -> None:
"""
Merge one session's backfill status into the scope peer's internal_metadata.
Uses a single-statement nested JSONB merge so concurrent writers updating
*different* session keys never clobber each other (the merge is computed
from the row's current committed value under the row lock, never from a
stale Python-side read):
internal_metadata || {"backfill_status":
coalesce(internal_metadata->'backfill_status', '{}') || {<session>: <entry>}}
Does not commit; the caller owns the transaction. Callers should
invalidate the peer cache after committing (``peer_cache_key``).
"""
entry: dict[str, Any] = {
"state": state,
"updated_at": datetime.now(timezone.utc).isoformat(),
}
if docs_copied is not None:
entry["docs_copied"] = docs_copied
# NOTE: the JSONB operand must be a Python dict, not a json.dumps() string.
# cast(<already-serialized-str>, JSONB) double-encodes: psycopg's JSONB
# bind adapter serializes the *string* again, producing a JSONB string
# scalar instead of an object. `||` between two non-array jsonb scalars
# doesn't merge keys — it silently wraps both sides into a 2-element
# array, corrupting backfill_status into a list and later crashing
# clear_scope_backfill_status's `#-` path delete (which then sees an
# array where it expects an object).
merged_status = func.coalesce(
models.Peer.internal_metadata.op("->")(BACKFILL_STATUS_KEY),
cast({}, JSONB),
).op("||")(cast({session_name: entry}, JSONB))
stmt = (
update(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == scope_peer)
.values(
internal_metadata=models.Peer.internal_metadata.op("||")(
func.jsonb_build_object(BACKFILL_STATUS_KEY, merged_status)
)
)
)
result = py_cast(CursorResult[Any], await db.execute(stmt))
if result.rowcount == 0:
raise ResourceNotFoundException(
f"Scope peer {scope_peer} not found in workspace {workspace_name}"
)
async def clear_scope_backfill_status(
db: AsyncSession,
workspace_name: str,
scope_peer: str,
session_name: str,
) -> None:
"""
Drop one session's entry from the scope peer's backfill status.
Called by the removal reconciliation job: once a session leaves a scope its
backfill status is moot, and clearing it keeps add→remove→re-add cycles
honest (the re-add starts from a fresh ``pending``). Single-statement
``#-`` delete, safe against concurrent per-key merges. Does not commit.
Missing peers are a no-op (removal is idempotent).
"""
stmt = (
update(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == scope_peer)
.values(
internal_metadata=models.Peer.internal_metadata.op("#-")(
cast(array([BACKFILL_STATUS_KEY, session_name]), ARRAY(Text))
)
)
)
await db.execute(stmt)
async def invalidate_scope_peer_cache(workspace_name: str, scope_peer: str) -> None:
"""Invalidate the cached peer row after a status write commits."""
await safe_cache_delete(peer_cache_key(workspace_name, scope_peer))
async def get_scope_backfill_status(
db: AsyncSession,
workspace_name: str,
scope_name: str,
) -> dict[str, Any]:
"""
Read the per-session backfill job status map for a scope.
Returns:
``{<session_name>: {state, updated_at[, docs_copied]}}`` (empty when no
backfill has ever been enqueued for the scope)
Raises:
ResourceNotFoundException: If the scope does not exist
"""
peer = await get_scope_or_raise(db, workspace_name, scope_name)
status = peer.internal_metadata.get(BACKFILL_STATUS_KEY, {})
if not isinstance(status, dict):
return {}
return py_cast(dict[str, Any], status)

View File

@ -320,8 +320,8 @@ async def get_or_create_session(
# Add the session to any requested scopes: create-or-get each scope peer
# and record an observer membership (observe_others=true, observe_me=false).
# No backfill happens here — membership only affects messages ingested
# after this point.
# If the session already has messages, a backfill task is enqueued after
# commit (below) so its existing documents are copied into the scope.
scopes_result = None
if session.scopes:
scopes_result = await get_or_create_scopes(
@ -352,6 +352,30 @@ async def get_or_create_session(
if scopes_result is not None:
await scopes_result.post_commit()
# Backfill (DEV-1999): a pre-existing session added to scopes at
# create-or-get time may already have messages; those need a
# backfill-by-copy task per scope. Fresh sessions need nothing.
if session.scopes:
has_messages = await db.scalar(
select(
exists(
select(models.Message.id)
.where(models.Message.workspace_name == workspace_name)
.where(models.Message.session_name == session.name)
)
)
)
if has_messages:
# Imported lazily: src.deriver.enqueue imports crud at module level.
from src.deriver.enqueue import enqueue_scope_backfill
for scope_name in session.scopes:
await enqueue_scope_backfill(
workspace_name,
scope_peer=scope_peer_name(scope_name),
session_name=session.name,
)
# Only update cache if session data changed or was newly created
if needs_cache_update:
cache_key = session_cache_key(workspace_name, session.name)

View File

@ -8,6 +8,10 @@ from sqlalchemy import select
from src import crud, models
from src.dependencies import tracked_db
from src.deriver.deriver import process_representation_tasks_batch
from src.deriver.scope_backfill import (
process_scope_backfill,
process_scope_removal,
)
from src.dreamer import process_dream
from src.exceptions import ResourceNotFoundException, ValidationException
from src.models import Message
@ -26,6 +30,8 @@ from src.utils.queue_payload import (
DeletionPayload,
DreamPayload,
ReconcilerPayload,
ScopeBackfillPayload,
ScopeRemovalPayload,
SummaryPayload,
WebhookPayload,
)
@ -150,6 +156,36 @@ async def process_item(queue_item: models.QueueItem) -> None:
raise ValueError(f"Invalid payload structure: {str(e)}") from e
await process_deletion(validated, workspace_name)
elif task_type == "scope_backfill":
with sentry_sdk.start_transaction(
name="process_scope_backfill_task", op="deriver"
):
try:
validated = ScopeBackfillPayload(**queue_payload)
except ValidationError as e:
logger.error(
"Invalid scope_backfill payload received: %s. Payload: %s",
str(e),
queue_payload,
)
raise ValueError(f"Invalid payload structure: {str(e)}") from e
await process_scope_backfill(validated, workspace_name)
elif task_type == "scope_removal":
with sentry_sdk.start_transaction(
name="process_scope_removal_task", op="deriver"
):
try:
validated = ScopeRemovalPayload(**queue_payload)
except ValidationError as e:
logger.error(
"Invalid scope_removal payload received: %s. Payload: %s",
str(e),
queue_payload,
)
raise ValueError(f"Invalid payload structure: {str(e)}") from e
await process_scope_removal(validated, workspace_name)
else:
raise ValueError(f"Invalid task type: {task_type}")

View File

@ -16,6 +16,7 @@ from src.utils.queue_payload import (
create_deletion_payload,
create_dream_payload,
create_payload,
create_scope_task_payload,
)
from src.utils.work_unit import construct_work_unit_key
@ -569,6 +570,168 @@ async def enqueue_dream(
raise
def create_scope_task_record(
workspace_name: str,
*,
task_type: Literal["scope_backfill", "scope_removal"],
scope_peer: str,
session_name: str,
) -> dict[str, Any]:
"""
Create a queue record for a scope backfill / removal task (DEV-1999).
Args:
workspace_name: Name of the workspace
task_type: "scope_backfill" or "scope_removal"
scope_peer: Prefixed name of the peer backing the scope
session_name: Name of the session whose membership changed
Returns:
Queue record dictionary ready for insertion into the queue
"""
payload = create_scope_task_payload(
task_type, scope_peer=scope_peer, session_name=session_name
)
return {
"work_unit_key": construct_work_unit_key(workspace_name, payload),
"payload": payload,
"session_id": None,
"task_type": task_type,
"workspace_name": workspace_name,
"message_id": None,
}
async def _enqueue_scope_task(
workspace_name: str,
*,
task_type: Literal["scope_backfill", "scope_removal"],
scope_peer: str,
session_name: str,
) -> None:
"""
Enqueue a scope backfill / removal task, deduplicating like enqueue_dream.
If a task with the same work_unit_key is already in progress (has an
ActiveQueueSession) or pending in the queue, the enqueue is skipped — the
handlers are idempotent, so a queued task already covers this membership
change.
For backfill tasks, the scope peer's per-session status entry is written
to "pending" in the same transaction as the queue insert, so the status
surface never claims a job exists that was never enqueued (or vice versa).
"""
async with tracked_db("scope_task_enqueue") as db_session:
try:
record = create_scope_task_record(
workspace_name,
task_type=task_type,
scope_peer=scope_peer,
session_name=session_name,
)
work_unit_key = record["work_unit_key"]
is_in_progress = await db_session.scalar(
select(
exists(
select(models.ActiveQueueSession.id).where(
models.ActiveQueueSession.work_unit_key == work_unit_key
)
)
)
)
if is_in_progress:
logger.debug(
"Skipping %s enqueue - already in progress: %s",
task_type,
work_unit_key,
)
return
is_pending = await db_session.scalar(
select(
exists(
select(QueueItem.id).where(
QueueItem.work_unit_key == work_unit_key,
QueueItem.processed == False, # noqa: E712
)
)
)
)
if is_pending:
logger.debug(
"%s already pending in queue: %s", task_type, work_unit_key
)
return
stmt = insert(QueueItem).returning(QueueItem)
await db_session.execute(stmt, [record])
if task_type == "scope_backfill":
await crud.update_scope_backfill_status(
db_session,
workspace_name,
scope_peer,
session_name,
state="pending",
)
await db_session.commit()
await crud.invalidate_scope_peer_cache(workspace_name, scope_peer)
logger.info(
"Enqueued %s task for %s/%s/%s",
task_type,
workspace_name,
scope_peer,
session_name,
)
except Exception as e:
logger.exception("Failed to enqueue %s task!", task_type)
if settings.SENTRY.ENABLED:
import sentry_sdk
sentry_sdk.capture_exception(e)
raise
async def enqueue_scope_backfill(
workspace_name: str,
*,
scope_peer: str,
session_name: str,
) -> None:
"""
Enqueue a backfill-by-copy task for a session newly added to a scope.
Only call this for sessions that already have messages — fresh sessions
need nothing (the deriver fan-out covers everything ingested after the
membership change).
"""
await _enqueue_scope_task(
workspace_name,
task_type="scope_backfill",
scope_peer=scope_peer,
session_name=session_name,
)
async def enqueue_scope_removal(
workspace_name: str,
*,
scope_peer: str,
session_name: str,
) -> None:
"""Enqueue a removal reconciliation task for a session removed from a scope."""
await _enqueue_scope_task(
workspace_name,
task_type="scope_removal",
scope_peer=scope_peer,
session_name=session_name,
)
def create_deletion_record(
workspace_name: str,
deletion_type: Literal["session", "observation", "workspace"],

View File

@ -71,10 +71,14 @@ RULES:
- Extract ALL observations from the target peer's messages, using others as context.
- Contextualize each observation sufficiently (e.g. "Ann is nervous about the job interview at the pharmacy" not just "Ann is nervous")
<examples>
These examples are fabricated illustrations of the output format. Never emit a conclusion for which content comes from these examples. Every conclusion must be supported by the <messages> block only.
EXAMPLES (using `alice` as the target peer id):
- EXPLICIT: "I just turned 25" → "alice is 25 years old"
- EXPLICIT: "I took my dog for a walk in NYC" → "alice has a dog", "alice walked her dog in NYC"
- EXPLICIT: "I've lived in NYC for six years" → "alice lives in NYC", "alice has lived in NYC for six years"
</examples>
{custom_instructions_section}

View File

@ -4,7 +4,7 @@ import random
import signal
import time
from asyncio import Task
from collections.abc import Sequence
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from logging import getLogger
@ -476,6 +476,18 @@ class QueueManager:
"""Snap the polling interval back to the base after finding work."""
self._current_poll_interval = settings.DERIVER.POLLING_SLEEP_INTERVAL_SECONDS
@staticmethod
def _is_tenant_work(work_unit_keys: Iterable[str]) -> bool:
"""True if any claimed work unit is real tenant work, not housekeeping.
"""
for key in work_unit_keys:
try:
if parse_work_unit_key(key).task_type != "reconciler":
return True
except ValueError:
return True
return False
def _jitter(self, seconds: float) -> float:
"""Scatter a sleep by +/- POLLING_JITTER_RATIO to avoid lockstep polling.
@ -542,7 +554,8 @@ class QueueManager:
await self._maybe_cleanup_stale_work_units()
claimed_work_units = await self.get_and_claim_work_units()
if claimed_work_units:
self._reset_poll_interval()
if self._is_tenant_work(claimed_work_units):
self._reset_poll_interval()
for work_unit_key, aqs_id in claimed_work_units.items():
# Create a new task for processing this work unit
if not self.shutdown_event.is_set():

View File

@ -0,0 +1,558 @@
"""Scope membership reconciliation jobs (DEV-1999, part of the Scopes RFC).
Two queue task handlers, dispatched from ``consumer.process_item``:
- ``scope_backfill`` — a session was added to a scope that already had
messages. Copy the session's *explicit* documents from each sender peer's
global ``(P, P)`` collection into the scope's ``(scope_peer, P)`` collection,
then enqueue a manual omni dream per touched collection to rebuild the
scope's higher-order layer and card.
- ``scope_removal`` — a session was removed from a scope. Soft-delete the
session's explicit documents from the scope's collections, cascade the
soft-delete to derived documents whose support left the scope (fail-closed),
then enqueue a ``card_refresh`` dream with ``rebuild=True`` plus a manual
omni dream per touched collection.
Zero LLM re-derivation: explicit-level documents are session-pure and
identical across observer collections (the DEV-2000 invariant), so retroactive
membership is pure row copying. The only external call is an embedding lookup
for source rows whose embedding column is NULL — embedding API only, never an
LLM.
DB sessions are never held across embedding or vector-store calls: the
handlers run in phases (plan → embed → write → sync), each phase opening its
own short-lived session.
"""
import logging
from dataclasses import dataclass
from typing import Any
from sqlalchemy import select, update
from sqlalchemy.dialects.postgresql import array
from sqlalchemy.sql.functions import func
from src import crud, models
from src.config import settings
from src.crud.scope import ScopeBackfillState
from src.crud.session import is_peer_in_session
from src.dependencies import tracked_db
from src.embedding_client import embedding_client
from src.schemas import DreamType
from src.utils.queue_payload import ScopeBackfillPayload, ScopeRemovalPayload
from src.utils.scopes import is_scope_peer_name
from src.vector_store import VectorRecord, get_external_vector_store
logger = logging.getLogger(__name__)
# internal_metadata key linking a backfilled copy to the global document it
# was copied from. The presence of this key is the idempotency marker.
COPIED_FROM_KEY = "copied_from"
def _store_embeddings_in_postgres() -> bool:
"""Whether document embeddings are persisted to the postgres column.
True when TYPE=pgvector OR still migrating (dual-write) — mirrors
``crud.document.create_documents``.
"""
return (
settings.VECTOR_STORE.TYPE == "pgvector" or not settings.VECTOR_STORE.MIGRATED
)
def _embedding_as_list(embedding: Any) -> list[float] | None:
"""Normalize a pgvector column value (numpy array or None) to a list."""
if embedding is None:
return None
return list(embedding)
@dataclass
class _CopySpec:
"""A planned copy of one global explicit document into a scope collection.
``restore_document_id`` points at an existing soft-deleted copy to restore
when set; for new copies it is filled with the inserted row's id after the
write phase.
"""
observed: str
source_id: str
content: str
embedding: list[float] | None
internal_metadata: dict[str, Any]
times_derived: int
source_ids: list[str] | None
session_name: str
restore_document_id: str | None = None
# ---------------------------------------------------------------------------
# Backfill (session added to a scope)
# ---------------------------------------------------------------------------
async def process_scope_backfill(
payload: ScopeBackfillPayload, workspace_name: str
) -> None:
"""Process a ``scope_backfill`` queue task.
Idempotent: a live copy (matched by ``internal_metadata.copied_from``) is
never duplicated, and a soft-deleted copy left by an earlier removal is
restored rather than re-inserted, so add→remove→re-add converges on
exactly one live copy per source document.
"""
scope_peer = payload.scope_peer
session_name = payload.session_name
try:
result = await _run_backfill(workspace_name, scope_peer, session_name)
except Exception:
await _write_backfill_status(
workspace_name, scope_peer, session_name, state="failed"
)
raise
if result is None:
# The session left the scope before this task was drained: removal
# already cleared the copies and the status entry, so there is nothing
# to backfill and nothing to record.
logger.info(
"Scope backfill skipped for %s/%s/%s: membership is no longer active",
workspace_name,
scope_peer,
session_name,
)
return
docs_copied, touched_observed = result
# One manual (gate-bypassing) omni dream per touched collection builds the
# scope's higher-order layer and card. enqueue_dream dedupes on the
# work-unit key, so a batch of backfills collapses to one dream each.
from src.deriver.enqueue import enqueue_dream
for observed in sorted(touched_observed):
await enqueue_dream(
workspace_name,
observer=scope_peer,
observed=observed,
dream_type=DreamType.OMNI,
trigger_reason="scope_backfill",
)
await _write_backfill_status(
workspace_name,
scope_peer,
session_name,
state="completed",
docs_copied=docs_copied,
)
logger.info(
"Scope backfill complete for %s/%s/%s: %d documents copied across %d collections",
workspace_name,
scope_peer,
session_name,
docs_copied,
len(touched_observed),
)
async def _run_backfill(
workspace_name: str, scope_peer: str, session_name: str
) -> tuple[int, set[str]] | None:
"""Run the copy itself.
Returns (docs copied or restored, touched observed peers), or ``None`` if
the session is no longer a member of the scope.
"""
# Phase 1 (DB): plan. Read the session's explicit documents from each
# sender's global (P, P) collection, and any existing copies (live or
# soft-deleted) already in the scope's collections.
plans: list[_CopySpec] = []
async with tracked_db("scope_backfill.plan") as db:
source_result = await db.execute(
select(models.Document).where(
models.Document.workspace_name == workspace_name,
models.Document.session_name == session_name,
models.Document.level == "explicit",
models.Document.observer == models.Document.observed,
models.Document.deleted_at.is_(None),
)
)
source_docs = [
doc
for doc in source_result.scalars().all()
# Scope peers never speak and are never observed, but stay
# defensive: never treat another scope's rows as a source.
if not is_scope_peer_name(doc.observer)
]
if not source_docs:
return 0, set()
# Existing copies in the scope's collections for this session, keyed
# by (observed, copied_from). Includes soft-deleted rows: those are
# restore candidates, not blockers.
copies_result = await db.execute(
select(models.Document).where(
models.Document.workspace_name == workspace_name,
models.Document.observer == scope_peer,
models.Document.session_name == session_name,
models.Document.internal_metadata.has_key(COPIED_FROM_KEY),
)
)
live_copies: set[tuple[str, str]] = set()
soft_deleted_copies: dict[tuple[str, str], str] = {}
for copy_doc in copies_result.scalars().all():
key = (copy_doc.observed, str(copy_doc.internal_metadata[COPIED_FROM_KEY]))
if copy_doc.deleted_at is None:
live_copies.add(key)
else:
soft_deleted_copies.setdefault(key, copy_doc.id)
for source in source_docs:
key = (source.observed, source.id)
if key in live_copies:
continue
plans.append(
_CopySpec(
observed=source.observed,
source_id=source.id,
content=source.content,
embedding=_embedding_as_list(source.embedding),
internal_metadata=dict(source.internal_metadata),
times_derived=source.times_derived,
source_ids=list(source.source_ids)
if source.source_ids is not None
else None,
session_name=session_name,
restore_document_id=soft_deleted_copies.get(key),
)
)
if not plans:
return 0, set()
# Phase 2 (no DB): fill missing embeddings. Source rows have NULL
# embeddings on external-store deployments (and soft-deleted copies may
# have lost their vectors) — re-embed via the embedding API only; no LLM.
missing = [spec for spec in plans if spec.embedding is None]
if missing:
logger.info(
"Scope backfill re-embedding %d documents with NULL embeddings for %s/%s/%s (embedding API only, no LLM)",
len(missing),
workspace_name,
scope_peer,
session_name,
)
embeddings = await embedding_client.simple_batch_embed(
[spec.content for spec in missing]
)
for spec, embedding in zip(missing, embeddings, strict=True):
spec.embedding = embedding
# Phase 3 (DB): write the copies.
store_in_postgres = _store_embeddings_in_postgres()
touched_observed = {spec.observed for spec in plans}
new_rows: list[models.Document] = []
async with tracked_db("scope_backfill.write") as db:
# scope_backfill and scope_removal carry different work-unit keys, so
# nothing orders them: a removal enqueued right after the add (or one
# that landed while phase 2 was embedding) can sweep the scope before
# these copies exist. Re-checking membership here, in the transaction
# that inserts, keeps a removed session from being copied back in.
if not await is_peer_in_session(db, workspace_name, session_name, scope_peer):
return None
for observed in sorted(touched_observed):
await crud.get_or_create_collection(
db, workspace_name, observer=scope_peer, observed=observed
)
restore_ids: list[str] = []
for spec in plans:
if spec.restore_document_id is not None:
restore_ids.append(spec.restore_document_id)
continue
row = models.Document(
workspace_name=workspace_name,
observer=scope_peer,
observed=spec.observed,
content=spec.content,
level="explicit",
times_derived=spec.times_derived,
internal_metadata={
**spec.internal_metadata,
COPIED_FROM_KEY: spec.source_id,
},
session_name=spec.session_name,
source_ids=spec.source_ids,
embedding=spec.embedding if store_in_postgres else None,
)
row.sync_state = "pending"
new_rows.append(row)
db.add_all(new_rows)
if restore_ids:
await db.execute(
update(models.Document)
.where(models.Document.id.in_(restore_ids))
.values(deleted_at=None, sync_state="pending")
)
await db.commit()
# IDs are generated client-side and remain accessible post-commit
# (expire_on_commit=False), mirroring crud.document.create_documents.
for spec, row in zip(
[s for s in plans if s.restore_document_id is None], new_rows, strict=True
):
spec.restore_document_id = row.id
copied_ids = [
spec.restore_document_id
for spec in plans
if spec.restore_document_id is not None
]
# Phase 4: sync to the external vector store (or mark synced in pgvector
# mode). Failures leave rows in sync_state='pending' for the reconciler.
await _sync_copies_to_vector_store(workspace_name, scope_peer, plans, copied_ids)
return len(plans), touched_observed
async def _sync_copies_to_vector_store(
workspace_name: str,
scope_peer: str,
plans: list[_CopySpec],
copied_ids: list[str],
) -> None:
"""Mirror the document-create sync path for the backfilled copies."""
external_vector_store = get_external_vector_store()
if external_vector_store is None:
# pgvector mode: embeddings live in the postgres column; nothing to sync.
async with tracked_db("scope_backfill.mark_synced") as db:
await db.execute(
update(models.Document)
.where(models.Document.id.in_(copied_ids))
.values(sync_state="synced", last_sync_at=func.now(), sync_attempts=0)
)
await db.commit()
return
by_observed: dict[str, list[_CopySpec]] = {}
for spec in plans:
by_observed.setdefault(spec.observed, []).append(spec)
synced_ids: list[str] = []
failed_ids: list[str] = []
for observed, specs in by_observed.items():
namespace = external_vector_store.get_vector_namespace(
"document", workspace_name, scope_peer, observed
)
records = [
VectorRecord(
id=spec.restore_document_id,
embedding=spec.embedding,
metadata={
"workspace_name": workspace_name,
"observer": scope_peer,
"observed": observed,
"session_name": spec.session_name,
"level": "explicit",
},
)
for spec in specs
if spec.restore_document_id is not None and spec.embedding is not None
]
ids = [record.id for record in records]
try:
await external_vector_store.upsert_many(namespace, records)
synced_ids.extend(ids)
except Exception:
logger.exception(
"Failed to upsert backfilled vectors to %s; leaving docs pending for the reconciler",
namespace,
)
failed_ids.extend(ids)
async with tracked_db("scope_backfill.sync_state") as db:
if synced_ids:
await db.execute(
update(models.Document)
.where(models.Document.id.in_(synced_ids))
.values(sync_state="synced", last_sync_at=func.now(), sync_attempts=0)
)
if failed_ids:
await db.execute(
update(models.Document)
.where(models.Document.id.in_(failed_ids))
.values(
sync_attempts=models.Document.sync_attempts + 1,
last_sync_at=func.now(),
)
)
await db.commit()
# ---------------------------------------------------------------------------
# Removal reconciliation (session removed from a scope)
# ---------------------------------------------------------------------------
async def process_scope_removal(
payload: ScopeRemovalPayload, workspace_name: str
) -> None:
"""Process a ``scope_removal`` queue task.
Soft-deletes (fail-closed) rather than hard-deletes: the reconciler's
soft-delete sweep handles vector cleanup and eventual hard deletion, and a
later re-add restores the same rows.
"""
scope_peer = payload.scope_peer
session_name = payload.session_name
removed_by_observed: dict[str, list[str]] = {}
async with tracked_db("scope_removal") as db:
observed_result = await db.execute(
select(models.Collection.observed).where(
models.Collection.workspace_name == workspace_name,
models.Collection.observer == scope_peer,
)
)
for observed in [row[0] for row in observed_result.all()]:
explicit_stmt = (
update(models.Document)
.where(
models.Document.workspace_name == workspace_name,
models.Document.observer == scope_peer,
models.Document.observed == observed,
models.Document.session_name == session_name,
models.Document.level == "explicit",
models.Document.deleted_at.is_(None),
)
.values(deleted_at=func.now())
.returning(models.Document.id)
)
frontier = [row[0] for row in (await db.execute(explicit_stmt)).all()]
all_removed = list(frontier)
# Fail-closed cascade: soft-delete derived documents whose support
# (source_ids) intersects anything removed, transitively — a
# deduction resting on removed evidence must leave with it, and so
# must an induction resting on that deduction.
while frontier:
derived_stmt = (
update(models.Document)
.where(
models.Document.workspace_name == workspace_name,
models.Document.observer == scope_peer,
models.Document.observed == observed,
models.Document.level != "explicit",
models.Document.deleted_at.is_(None),
models.Document.source_ids.has_any(array(frontier)),
)
.values(deleted_at=func.now())
.returning(models.Document.id)
)
frontier = [row[0] for row in (await db.execute(derived_stmt)).all()]
all_removed.extend(frontier)
if all_removed:
removed_by_observed[observed] = all_removed
# The session left the scope, so its backfill status entry is moot;
# clearing it keeps a later re-add starting from a fresh "pending".
await crud.clear_scope_backfill_status(
db, workspace_name, scope_peer, session_name
)
await db.commit()
await crud.invalidate_scope_peer_cache(workspace_name, scope_peer)
# Delete the vectors eagerly so recall can't surface removed memory while
# waiting for the reconciler sweep (which remains the backstop on failure).
external_vector_store = get_external_vector_store()
if external_vector_store is not None:
for observed, removed_ids in removed_by_observed.items():
namespace = external_vector_store.get_vector_namespace(
"document", workspace_name, scope_peer, observed
)
try:
await external_vector_store.delete_many(namespace, removed_ids)
except Exception:
logger.exception(
"Failed to delete removed vectors from %s; reconciler sweep will retry",
namespace,
)
# Rebuild what remains: the card must be regenerated from remaining
# evidence only (rebuild=True drops the stale card from the prompt), and a
# manual omni dream rebuilds the higher-order structure.
from src.deriver.enqueue import enqueue_dream
for observed in sorted(removed_by_observed):
await enqueue_dream(
workspace_name,
observer=scope_peer,
observed=observed,
dream_type=DreamType.CARD_REFRESH,
rebuild=True,
trigger_reason="scope_removal",
)
await enqueue_dream(
workspace_name,
observer=scope_peer,
observed=observed,
dream_type=DreamType.OMNI,
trigger_reason="scope_removal",
)
logger.info(
"Scope removal reconciliation complete for %s/%s/%s: %d documents soft-deleted across %d collections",
workspace_name,
scope_peer,
session_name,
sum(len(ids) for ids in removed_by_observed.values()),
len(removed_by_observed),
)
# ---------------------------------------------------------------------------
# Shared helpers
# ---------------------------------------------------------------------------
async def _write_backfill_status(
workspace_name: str,
scope_peer: str,
session_name: str,
*,
state: ScopeBackfillState,
docs_copied: int | None = None,
) -> None:
"""Best-effort status write in its own short-lived session."""
try:
async with tracked_db("scope_backfill.status") as db:
await crud.update_scope_backfill_status(
db,
workspace_name,
scope_peer,
session_name,
state=state,
docs_copied=docs_copied,
)
await db.commit()
await crud.invalidate_scope_peer_cache(workspace_name, scope_peer)
except Exception:
# Never mask the underlying failure (or fail a completed backfill)
# over a status bookkeeping write.
logger.exception(
"Failed to write scope backfill status %s for %s/%s/%s",
state,
workspace_name,
scope_peer,
session_name,
)

View File

@ -195,13 +195,13 @@ class _EmbeddingClient:
from google import genai
from google.genai import types as genai_types
# 10-minute HTTP timeout, in lockstep with the LLM registry's Gemini
# client (`src/llm/registry.py:_build_gemini_http_options`). Without
# this, a stalled Gemini embedding socket wedges the deriver worker
# exactly the way #785 describes for the LLM client.
# Default 10-minute HTTP timeout matches the LLM registry Gemini client.
timeout_ms = (
int(config.timeout * 1000) if config.timeout is not None else 600_000
)
http_options = genai_types.HttpOptions(
base_url=config.base_url,
timeout=600_000,
timeout=timeout_ms,
)
self.client: genai.Client | AsyncOpenAI = genai.Client(
api_key=config.api_key,
@ -216,10 +216,14 @@ class _EmbeddingClient:
raise ValueError("OpenAI API key is required")
from openai import AsyncOpenAI
self.client = AsyncOpenAI(
api_key=config.api_key,
base_url=config.base_url,
)
# Omit timeout when unset so the OpenAI SDK keeps its own default.
client_kwargs: dict[str, Any] = {
"api_key": config.api_key,
"base_url": config.base_url,
}
if config.timeout is not None:
client_kwargs["timeout"] = config.timeout
self.client = AsyncOpenAI(**client_kwargs)
self.max_embedding_tokens = max_input_tokens
self.max_batch_size = config.max_batch_size or 2048

View File

@ -8,9 +8,10 @@ that keeps the observer/observed mechanics hidden.
All scopes routes require a workspace-level (or admin) key: scopes are an
app-level admin surface, so peer- and session-scoped keys are rejected.
Note: scope membership only affects messages ingested *after* the membership
change. Conclusions already derived are neither backfilled on add nor
reconciled on removal.
Retroactive membership changes are handled asynchronously: adding a session
that already has messages enqueues a backfill-by-copy job, removing a session
enqueues a removal-reconciliation job. Track backfill progress via
``GET /scopes/{scope_id}/status``.
"""
import logging
@ -109,8 +110,9 @@ async def add_sessions_to_scope(
is already a member is a no-op. List the resulting membership with
`POST /scopes/{scope_id}/sessions/list`.
Note: membership applies only to messages ingested after this call;
conclusions already derived are not backfilled.
Note: any added session that already has messages triggers an asynchronous
backfill-by-copy of its existing documents into the scope; track progress
via ``GET /scopes/{scope_id}/status``.
"""
await crud.add_sessions_to_scope(
db,
@ -135,8 +137,10 @@ async def remove_session_from_scope(
"""
Remove a Session from a Scope.
Note: conclusions already derived while the session was a member are left in
place.
Note: documents copied/derived while the session was a member are
reconciled asynchronously — the session's explicit copies are soft-deleted
from the scope, dependent derived documents follow (fail-closed), and the
scope's card is rebuilt from the remaining evidence.
"""
await crud.remove_session_from_scope(
db,
@ -171,3 +175,24 @@ async def get_scope_sessions(
workspace_name=workspace_id, scope_name=scope_id, reverse=reverse
),
)
@router.get(
"/{scope_id}/status",
response_model=schemas.ScopeStatus,
dependencies=[Depends(require_auth(workspace_name="workspace_id"))],
)
async def get_scope_status(
workspace_id: str = Path(...),
scope_id: str = Path(...),
db: AsyncSession = read_db,
):
"""
Get the backfill/reconciliation job status for a Scope.
Returns a per-session map of the backfill job state (pending / completed /
failed) with the number of documents copied once complete. Empty when no
backfill has ever been enqueued for the scope.
"""
backfill_status = await crud.get_scope_backfill_status(db, workspace_id, scope_id)
return schemas.ScopeStatus(backfill_status=backfill_status)

View File

@ -39,6 +39,7 @@ from src.schemas.api import (
Scope,
ScopeCreate,
ScopeSessionsAdd,
ScopeStatus,
Session,
SessionBase,
SessionContext,
@ -140,6 +141,7 @@ __all__ = [
"Scope",
"ScopeCreate",
"ScopeSessionsAdd",
"ScopeStatus",
"Session",
"SessionBase",
"SessionContext",

View File

@ -428,9 +428,9 @@ class SessionCreate(SessionBase):
max_length=100,
description=(
"Optional list of (unprefixed) scope names to add this session to. "
"Each scope is created if it does not exist yet. Membership applies "
"only to messages ingested after the session is added to the scope; "
"conclusions already derived are not backfilled."
"Each scope is created if it does not exist yet. If the session "
"already has messages, its existing documents are backfilled into "
"the scope asynchronously."
),
)
@ -591,6 +591,18 @@ class ScopeSessionsAdd(BaseModel):
)
class ScopeStatus(BaseModel):
"""Per-session backfill/reconciliation job status for a scope.
``backfill_status`` maps each session that has had a backfill enqueued to
its current job state: ``{state, updated_at[, docs_copied]}`` where
``state`` is ``pending``/``completed``/``failed`` and ``docs_copied`` is
present once a backfill completes.
"""
backfill_status: dict[str, dict[str, Any]] = Field(default_factory=dict)
# ---------------------------------------------------------------------------
# Conclusion schemas
# ---------------------------------------------------------------------------

View File

@ -73,6 +73,26 @@ class DreamPayload(BasePayload):
rebuild: bool = False
class ScopeBackfillPayload(BasePayload):
"""Payload for scope backfill tasks (session added to a scope, DEV-1999).
workspace_name lives in the QueueItem's dedicated column, mirroring the
other task payloads.
"""
task_type: Literal["scope_backfill"] = "scope_backfill"
scope_peer: str
session_name: str
class ScopeRemovalPayload(BasePayload):
"""Payload for scope removal reconciliation tasks (session removed from a scope)."""
task_type: Literal["scope_removal"] = "scope_removal"
scope_peer: str
session_name: str
class DeletionPayload(BasePayload):
"""Payload for deletion tasks."""
@ -124,6 +144,22 @@ def create_dream_payload(
).model_dump(mode="json", exclude_none=True)
def create_scope_task_payload(
task_type: Literal["scope_backfill", "scope_removal"],
*,
scope_peer: str,
session_name: str,
) -> dict[str, Any]:
"""Create a scope backfill / removal payload."""
if task_type == "scope_backfill":
return ScopeBackfillPayload(
scope_peer=scope_peer, session_name=session_name
).model_dump(mode="json", exclude_none=True)
return ScopeRemovalPayload(
scope_peer=scope_peer, session_name=session_name
).model_dump(mode="json", exclude_none=True)
def create_deletion_payload(
deletion_type: Literal["session", "observation", "workspace"],
resource_id: str,

View File

@ -251,7 +251,14 @@ class GetOrCreateResult(Generic[T]):
TaskType = Literal[
"webhook", "summary", "representation", "dream", "deletion", "reconciler"
"webhook",
"summary",
"representation",
"dream",
"deletion",
"reconciler",
"scope_backfill",
"scope_removal",
]
VectorSyncState = Literal["synced", "pending", "failed"]
DocumentLevel = Literal["explicit", "deductive", "inductive", "contradiction"]

View File

@ -74,6 +74,15 @@ def construct_work_unit_key(
raise ValueError("reconciler_type is required for reconciler tasks")
return f"reconciler:{reconciler_type}"
if task_type in ("scope_backfill", "scope_removal"):
scope_peer = payload.get("scope_peer")
session_name = payload.get("session_name")
if not scope_peer or not session_name:
raise ValueError(
f"scope_peer and session_name are required for {task_type} tasks"
)
return f"{task_type}:{workspace_name}:{scope_peer}:{session_name}"
raise ValueError(f"Invalid task type: {task_type}")
@ -183,4 +192,19 @@ def parse_work_unit_key(work_unit_key: str) -> ParsedWorkUnit:
observed=None,
)
if task_type in ("scope_backfill", "scope_removal"):
# {task_type}:{workspace}:{scope_peer}:{session}
if len(parts) != 4:
raise ValueError(
f"Invalid work_unit_key format for task_type {task_type}: {work_unit_key}"
)
return ParsedWorkUnit(
task_type=task_type,
workspace_name=parts[1],
session_name=parts[3],
# The scope peer is the observer of every collection the task touches.
observer=parts[2],
observed=None,
)
raise ValueError(f"Invalid task type in work_unit_key: {task_type}")

View File

@ -957,6 +957,7 @@ def mock_tracked_db(request: pytest.FixtureRequest):
"src.dialectic.core.tracked_db",
"src.dreamer.specialists.tracked_db",
"src.dreamer.surprisal.tracked_db",
"src.deriver.scope_backfill.tracked_db",
]
with ExitStack() as stack:
for target in tracked_db_targets:

View File

@ -0,0 +1,958 @@
"""Tests for scope backfill-by-copy and removal reconciliation (DEV-1999).
A scope is an observer peer (``scope.<name>``). Adding a session with
pre-existing messages to a scope enqueues a ``scope_backfill`` task; the
handler (``src.deriver.scope_backfill``) copies each observed peer's
explicit-level documents from their global ``(P, P)`` collection into the
scope's ``(scope_peer, P)`` collection, stamping ``copied_from`` for
idempotency, then enqueues a manual omni dream. Removal enqueues
``scope_removal``, which soft-deletes the copies (cascading to dependent
derived documents) and enqueues a card_refresh (rebuild) + omni dream.
These tests exercise the handlers directly (``process_scope_backfill`` /
``process_scope_removal``) against real Collection/Document/Peer rows,
mirroring the fixture style in tests/crud/test_document.py and
tests/dreamer/test_card_refresh.py: rows are created directly via
``db_session`` (never through the cache-backed ``crud.get_or_create_collection``,
which the ``mock_crud_collection_operations`` autouse fixture stubs out to an
unpersisted object for every other test). Fixture data must be *committed*
(not merely flushed) because the handlers run their DB work through
``tracked_db``, which in tests opens a separate session bound to the same
engine (see ``mock_tracked_db_context`` in conftest.py) — a different
connection that cannot see another session's uncommitted writes.
"""
from typing import Any
import pytest
from fastapi.testclient import TestClient
from nanoid import generate as generate_nanoid
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from src import crud, models
from src.deriver.scope_backfill import (
COPIED_FROM_KEY,
process_scope_backfill,
process_scope_removal,
)
from src.schemas import DreamType
from src.utils.queue_payload import ScopeBackfillPayload, ScopeRemovalPayload
from src.utils.scopes import is_scope_peer, scope_peer_name
_EMBEDDING_DIM = 1536
def _embedding(seed: float = 0.5) -> list[float]:
return [seed] * _EMBEDDING_DIM
async def _create_peer(db_session: AsyncSession, workspace_name: str) -> models.Peer:
peer = models.Peer(name=str(generate_nanoid()), workspace_name=workspace_name)
db_session.add(peer)
await db_session.commit()
return peer
async def _create_scope_peer(
db_session: AsyncSession, workspace_name: str, scope_name: str
) -> models.Peer:
peer = models.Peer(
name=scope_peer_name(scope_name),
workspace_name=workspace_name,
# The authoritative kind flag lives in internal_metadata, which is not
# user-writable; `configuration` carries only the observe_me knob.
internal_metadata={"kind": "scope"},
configuration={"observe_me": False},
)
db_session.add(peer)
await db_session.commit()
return peer
async def _create_session(
db_session: AsyncSession, workspace_name: str
) -> models.Session:
session = models.Session(name=str(generate_nanoid()), workspace_name=workspace_name)
db_session.add(session)
await db_session.commit()
return session
async def _join_scope(
db_session: AsyncSession, workspace_name: str, session_name: str, scope_peer: str
) -> None:
"""Record the scope peer's membership, as the scopes routes do.
The backfill handler refuses to copy into a scope the session has left, so
tests driving the handler directly must stand the membership row up.
"""
db_session.add(
models.SessionPeer(
workspace_name=workspace_name,
session_name=session_name,
peer_name=scope_peer,
)
)
await db_session.commit()
async def _create_collection(
db_session: AsyncSession, workspace_name: str, observer: str, observed: str
) -> models.Collection:
collection = models.Collection(
workspace_name=workspace_name, observer=observer, observed=observed
)
db_session.add(collection)
await db_session.commit()
return collection
async def _create_document(
db_session: AsyncSession,
workspace_name: str,
*,
observer: str,
observed: str,
session_name: str | None,
content: str = "some observation",
level: str = "explicit",
embedding: list[float] | None = None,
internal_metadata: dict[str, Any] | None = None,
source_ids: list[str] | None = None,
) -> models.Document:
doc = models.Document(
workspace_name=workspace_name,
observer=observer,
observed=observed,
content=content,
level=level,
session_name=session_name,
embedding=embedding if embedding is not None else _embedding(),
internal_metadata=internal_metadata or {},
source_ids=source_ids,
)
db_session.add(doc)
await db_session.commit()
return doc
async def _get_docs(
db_session: AsyncSession,
workspace_name: str,
*,
observer: str,
observed: str | None = None,
include_deleted: bool = True,
) -> list[models.Document]:
stmt = select(models.Document).where(
models.Document.workspace_name == workspace_name,
models.Document.observer == observer,
)
if observed is not None:
stmt = stmt.where(models.Document.observed == observed)
if not include_deleted:
stmt = stmt.where(models.Document.deleted_at.is_(None))
result = await db_session.execute(stmt)
return list(result.scalars().all())
async def _dream_items(
db_session: AsyncSession, workspace_name: str
) -> list[models.QueueItem]:
result = await db_session.execute(
select(models.QueueItem).where(
models.QueueItem.workspace_name == workspace_name,
models.QueueItem.task_type == "dream",
)
)
return list(result.scalars().all())
# ---------------------------------------------------------------------------
# 1. Backfill copies exactly the target session's explicit docs
# ---------------------------------------------------------------------------
async def test_backfill_copies_only_target_session_explicit_docs(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
target_session = await _create_session(db_session, workspace_name)
other_session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, target_session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
# Destination collection: crud.get_or_create_collection is stubbed to an
# unpersisted object by the autouse mock_crud_collection_operations
# fixture, so the scope's own collection must already exist for the
# copied Document rows' FK to resolve.
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
# In-scope: the target session's explicit doc.
target_doc = await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=target_session.name,
content="target session explicit fact",
embedding=_embedding(0.7),
)
# Out-of-scope: another session's explicit doc.
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=other_session.name,
content="other session explicit fact",
)
# Out-of-scope: a derived (non-explicit) doc for the target session.
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=target_session.name,
content="deductive fact",
level="deductive",
)
await process_scope_backfill(
ScopeBackfillPayload(
scope_peer=scope_peer.name, session_name=target_session.name
),
workspace_name,
)
copies = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
assert len(copies) == 1
copy = copies[0]
assert copy.content == "target session explicit fact"
assert copy.level == "explicit"
assert copy.session_name == target_session.name
assert copy.internal_metadata[COPIED_FROM_KEY] == target_doc.id
assert list(copy.embedding) == pytest.approx( # pyright: ignore[reportUnknownMemberType]
_embedding(0.7)
)
assert copy.deleted_at is None
# ---------------------------------------------------------------------------
# 2. Idempotency
# ---------------------------------------------------------------------------
async def test_backfill_processed_twice_is_idempotent(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session.name,
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
payload = ScopeBackfillPayload(
scope_peer=scope_peer.name, session_name=session.name
)
await process_scope_backfill(payload, workspace_name)
await process_scope_backfill(payload, workspace_name)
copies = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
assert len(copies) == 1
async def test_add_remove_readd_converges_on_one_live_copy(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session.name,
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
backfill_payload = ScopeBackfillPayload(
scope_peer=scope_peer.name, session_name=session.name
)
removal_payload = ScopeRemovalPayload(
scope_peer=scope_peer.name, session_name=session.name
)
# add
await process_scope_backfill(backfill_payload, workspace_name)
# remove
await process_scope_removal(removal_payload, workspace_name)
live = await _get_docs(
db_session,
workspace_name,
observer=scope_peer.name,
observed=sender.name,
include_deleted=False,
)
assert live == []
# re-add
await process_scope_backfill(backfill_payload, workspace_name)
all_copies = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
live_copies = [d for d in all_copies if d.deleted_at is None]
assert len(all_copies) == 1 # restored, not duplicated
assert len(live_copies) == 1
async def test_backfill_skips_a_session_that_left_the_scope(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
"""A removal that lands first must not be undone by a queued backfill.
scope_backfill and scope_removal carry different work-unit keys, so nothing
orders them: add-then-remove can leave a backfill queued after removal has
already swept the scope.
"""
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session.name,
)
# The session leaves the scope before the queued backfill is drained.
await db_session.execute(
update(models.SessionPeer)
.where(
models.SessionPeer.workspace_name == workspace_name,
models.SessionPeer.session_name == session.name,
models.SessionPeer.peer_name == scope_peer.name,
)
.values(left_at=func.now())
)
await db_session.commit()
await process_scope_backfill(
ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
assert (
await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
== []
)
# No status entry either: removal cleared it, and a skipped backfill must
# not resurrect the session in the scope's status map.
# Names held as plain strings: expire_all() below would make reading them
# off the ORM instances trigger a lazy reload mid-assertion.
scope_peer_name_str, session_name = scope_peer.name, session.name
db_session.expire_all()
peer = await db_session.scalar(
select(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == scope_peer_name_str)
)
assert peer is not None
assert session_name not in peer.internal_metadata.get("backfill_status", {})
# ---------------------------------------------------------------------------
# 3. Multi-peer session
# ---------------------------------------------------------------------------
async def test_backfill_multi_peer_session_copies_into_right_collections(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, peer_a = sample_data
workspace_name = test_workspace.name
peer_b = await _create_peer(db_session, workspace_name)
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
for peer in (peer_a, peer_b):
await _create_collection(
db_session, workspace_name, observer=peer.name, observed=peer.name
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=peer.name
)
await _create_document(
db_session,
workspace_name,
observer=peer.name,
observed=peer.name,
session_name=session.name,
content=f"fact about {peer.name}",
)
await process_scope_backfill(
ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
copies_a = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=peer_a.name
)
copies_b = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=peer_b.name
)
assert len(copies_a) == 1
assert copies_a[0].content == f"fact about {peer_a.name}"
assert len(copies_b) == 1
assert copies_b[0].content == f"fact about {peer_b.name}"
# ---------------------------------------------------------------------------
# 4. Removal cascade
# ---------------------------------------------------------------------------
async def test_removal_cascades_to_dependent_derived_docs_only(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session.name,
)
await process_scope_backfill(
ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
[copy] = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
# A derived doc resting on the copy's evidence -> must be cascaded.
dependent = await _create_document(
db_session,
workspace_name,
observer=scope_peer.name,
observed=sender.name,
session_name=None,
content="deduction resting on removed evidence",
level="deductive",
source_ids=[copy.id],
)
# An unrelated derived doc in the same collection -> must survive.
unrelated = await _create_document(
db_session,
workspace_name,
observer=scope_peer.name,
observed=sender.name,
session_name=None,
content="unrelated deduction",
level="deductive",
source_ids=["some-other-doc-id-not-removed"],
)
copy_id, dependent_id, unrelated_id = copy.id, dependent.id, unrelated.id
await process_scope_removal(
ScopeRemovalPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
# process_scope_removal runs on a separate tracked_db session (a
# different connection). Query raw columns rather than full ORM entities
# so this session's identity map (holding the pre-removal `copy` /
# `dependent` / `unrelated` instances) can't hand back stale, expired
# attributes.
result = await db_session.execute(
select(models.Document.id, models.Document.deleted_at).where(
models.Document.workspace_name == workspace_name,
models.Document.observer == scope_peer.name,
models.Document.observed == sender.name,
)
)
deleted_at_by_id = {row[0]: row[1] for row in result.all()}
assert deleted_at_by_id[copy_id] is not None
assert deleted_at_by_id[dependent_id] is not None
assert deleted_at_by_id[unrelated_id] is None
# ---------------------------------------------------------------------------
# 5. Dream enqueues
# ---------------------------------------------------------------------------
async def test_backfill_enqueues_manual_omni_dream(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session.name,
)
await process_scope_backfill(
ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
dreams = await _dream_items(db_session, workspace_name)
assert len(dreams) == 1
payload = dreams[0].payload
assert payload["dream_type"] == DreamType.OMNI.value
assert payload["observer"] == scope_peer.name
assert payload["observed"] == sender.name
assert payload["trigger_reason"] == "scope_backfill"
assert payload.get("rebuild", False) is False
async def test_removal_enqueues_card_refresh_rebuild_and_omni_dream(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
"""Removal's own dream enqueues, isolated from backfill's.
The scope's copy is created directly (as if an earlier backfill already
ran and its dream was drained by the deriver) rather than by calling
process_scope_backfill first: enqueue_dream dedupes on work_unit_key, so
a still-pending omni dream from an immediately-preceding backfill would
silently swallow removal's own omni enqueue and make this test couple to
that unrelated dedup behavior instead of testing removal in isolation.
"""
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=scope_peer.name,
observed=sender.name,
session_name=session.name,
internal_metadata={COPIED_FROM_KEY: "some-source-doc-id"},
)
await process_scope_removal(
ScopeRemovalPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
dreams = await _dream_items(db_session, workspace_name)
removal_dreams = [
d for d in dreams if d.payload.get("trigger_reason") == "scope_removal"
]
assert len(removal_dreams) == 2
by_type = {d.payload["dream_type"]: d.payload for d in removal_dreams}
assert DreamType.CARD_REFRESH.value in by_type
assert DreamType.OMNI.value in by_type
card_refresh_payload = by_type[DreamType.CARD_REFRESH.value]
assert card_refresh_payload["rebuild"] is True
assert card_refresh_payload["observer"] == scope_peer.name
assert card_refresh_payload["observed"] == sender.name
# ---------------------------------------------------------------------------
# 6. Status endpoint
# ---------------------------------------------------------------------------
async def test_status_reflects_pending_then_completed(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
response = client.post(
f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name}
)
assert response.status_code == 201
scope_peer_full_name = scope_peer_name(scope_name)
session_name = str(generate_nanoid())
response = client.post(
f"/v3/workspaces/{workspace_name}/sessions",
json={"id": session_name, "peers": {sender.name: {}}},
)
assert response.status_code == 201
message = models.Message(
workspace_name=workspace_name,
session_name=session_name,
peer_name=sender.name,
content="hello from before the scope existed",
public_id=generate_nanoid(),
seq_in_session=1,
token_count=5,
)
db_session.add(message)
await db_session.commit()
# The message's explicit document (normally produced by the deriver) —
# created directly since the deriver isn't run in this test.
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session_name,
)
response = client.post(
f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions",
json={"session_ids": [session_name]},
)
assert response.status_code == 204, response.text
# Destination collection: crud.get_or_create_collection is stubbed to an
# unpersisted object by the autouse fixture, so it must pre-exist.
await _create_collection(
db_session, workspace_name, observer=scope_peer_full_name, observed=sender.name
)
status_url = f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/status"
response = client.get(status_url)
assert response.status_code == 200, response.text
backfill_status = response.json()["backfill_status"]
assert backfill_status[session_name]["state"] == "pending"
# Simulate the deriver picking up the enqueued task.
await process_scope_backfill(
ScopeBackfillPayload(
scope_peer=scope_peer_full_name, session_name=session_name
),
workspace_name,
)
response = client.get(status_url)
assert response.status_code == 200, response.text
backfill_status = response.json()["backfill_status"]
assert backfill_status[session_name]["state"] == "completed"
assert backfill_status[session_name]["docs_copied"] == 1
async def test_backfill_re_embeds_sources_with_null_embeddings(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
"""Source rows carry no embedding on external-store deployments.
Phase 2 re-embeds those (embedding API only) and pairs results back with
strict=True, so a mis-pairing would raise rather than silently mismatch.
"""
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
await _join_scope(db_session, workspace_name, session.name, scope_peer.name)
await _create_collection(
db_session, workspace_name, observer=sender.name, observed=sender.name
)
await _create_collection(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
source = await _create_document(
db_session,
workspace_name,
observer=sender.name,
observed=sender.name,
session_name=session.name,
content="fact whose vector lives in the external store",
)
source.embedding = None
await db_session.commit()
await process_scope_backfill(
ScopeBackfillPayload(scope_peer=scope_peer.name, session_name=session.name),
workspace_name,
)
[copy] = await _get_docs(
db_session, workspace_name, observer=scope_peer.name, observed=sender.name
)
assert copy.embedding is not None
assert len(copy.embedding) == _EMBEDDING_DIM
async def test_backfill_failure_records_failed_status(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
monkeypatch: pytest.MonkeyPatch,
):
test_workspace, _ = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
scope_peer = await _create_scope_peer(db_session, workspace_name, scope_name)
session = await _create_session(db_session, workspace_name)
# Plain strings: the ORM instances are expired below (see the same guard in
# test_backfill_status_writes_preserve_the_scope_kind_flag).
scope_peer_name_str, session_name = scope_peer.name, session.name
async def boom(*_args: Any, **_kwargs: Any) -> None:
raise RuntimeError("copy phase blew up")
monkeypatch.setattr("src.deriver.scope_backfill._run_backfill", boom)
with pytest.raises(RuntimeError):
await process_scope_backfill(
ScopeBackfillPayload(
scope_peer=scope_peer_name_str, session_name=session_name
),
workspace_name,
)
db_session.expire_all()
peer = await db_session.scalar(
select(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == scope_peer_name_str)
)
assert peer is not None
assert peer.internal_metadata["backfill_status"][session_name]["state"] == "failed"
# ---------------------------------------------------------------------------
# 7. Route wiring: add-sessions enqueues backfill only when messages exist
# ---------------------------------------------------------------------------
async def test_add_sessions_enqueues_backfill_only_when_session_has_messages(
client: TestClient,
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
test_workspace, sender = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
assert (
client.post(
f"/v3/workspaces/{workspace_name}/scopes", json={"id": scope_name}
).status_code
== 201
)
# Session with a pre-existing message.
session_with_messages = str(generate_nanoid())
assert (
client.post(
f"/v3/workspaces/{workspace_name}/sessions",
json={"id": session_with_messages, "peers": {sender.name: {}}},
).status_code
== 201
)
message = models.Message(
workspace_name=workspace_name,
session_name=session_with_messages,
peer_name=sender.name,
content="already said something",
public_id=generate_nanoid(),
seq_in_session=1,
token_count=5,
)
db_session.add(message)
await db_session.commit()
# Empty session, no messages.
empty_session = str(generate_nanoid())
assert (
client.post(
f"/v3/workspaces/{workspace_name}/sessions",
json={"id": empty_session},
).status_code
== 201
)
response = client.post(
f"/v3/workspaces/{workspace_name}/scopes/{scope_name}/sessions",
json={"session_ids": [session_with_messages, empty_session]},
)
assert response.status_code == 204, response.text
result = await db_session.execute(
select(models.QueueItem).where(
models.QueueItem.workspace_name == workspace_name,
models.QueueItem.task_type == "scope_backfill",
)
)
backfill_items = list(result.scalars().all())
assert len(backfill_items) == 1
assert backfill_items[0].payload["session_name"] == session_with_messages
# ---------------------------------------------------------------------------
# The scope `kind` flag and the backfill status map share the scope peer's
# internal_metadata. Every write to that column must be a JSONB merge scoped to
# the backfill key; a wholesale assignment would drop the flag and silently turn
# the peer back into an ordinary one — invisible until some later read stopped
# recognising it as a scope.
# ---------------------------------------------------------------------------
async def test_backfill_status_writes_preserve_the_scope_kind_flag(
db_session: AsyncSession,
sample_data: tuple[models.Workspace, models.Peer],
):
"""Status writes must not clobber the authoritative kind flag.
Both live in internal_metadata, so this pins the one property that makes
them able to coexist. Covers the whole lifecycle, because a wholesale write
could be introduced at any single step: pending, completed, then cleared.
"""
test_workspace, _ = sample_data
workspace_name = test_workspace.name
scope_name = str(generate_nanoid())
session_name = str(generate_nanoid())
await _create_scope_peer(db_session, workspace_name, scope_name)
# Held as a plain string: the ORM instance is expired below on every check,
# so reading an attribute off it would trigger a reload mid-assertion.
backing_peer = scope_peer_name(scope_name)
async def assert_still_a_scope(stage: str) -> dict[str, Any]:
db_session.expire_all()
refreshed = await db_session.scalar(
select(models.Peer)
.where(models.Peer.workspace_name == workspace_name)
.where(models.Peer.name == backing_peer)
)
assert refreshed is not None
assert is_scope_peer(refreshed.name, refreshed.internal_metadata), (
f"the peer stopped being a scope after {stage}: "
f"internal_metadata={refreshed.internal_metadata!r}"
)
# And the facade still resolves it, which is what actually breaks:
# get_scope_or_raise 404s on a peer that has lost the flag.
resolved = await crud.get_scope_or_raise(db_session, workspace_name, scope_name)
assert resolved.name == backing_peer
return refreshed.internal_metadata
await assert_still_a_scope("creation")
await crud.update_scope_backfill_status(
db_session,
workspace_name,
backing_peer,
session_name,
state="pending",
)
await db_session.commit()
metadata = await assert_still_a_scope("a pending status write")
assert metadata["backfill_status"][session_name]["state"] == "pending"
await crud.update_scope_backfill_status(
db_session,
workspace_name,
backing_peer,
session_name,
state="completed",
docs_copied=3,
)
await db_session.commit()
metadata = await assert_still_a_scope("a completed status write")
assert metadata["backfill_status"][session_name]["docs_copied"] == 3
await crud.clear_scope_backfill_status(
db_session, workspace_name, backing_peer, session_name
)
await db_session.commit()
metadata = await assert_still_a_scope("clearing the status")
assert session_name not in metadata.get("backfill_status", {})

View File

@ -68,5 +68,5 @@ Coverage by provider:
- OpenAI transport → OpenRouter non-reasoning models (e.g. `inception/mercury-2`): non-chat / diffusion architectures must stay on `max_tokens`, no `reasoning_effort`, tool-calling parameter-schema compatibility is the canary for exotic OR-served providers
- Gemini 2.5/3.0 classes: structured outputs, cached-content reuse, thought signatures, multi-turn tool replay
- Gemini 3.1 class: thinking and tool replay coverage by default; structured-output/caching coverage should only be added once Google documents support for that path
- Embeddings (`test_live_embeddings.py`): single embed, batched embed, batch-vs-single alignment, and chunk-to-id mapping for every configured embedding model. `gemini-embedding-2*` is the reason this exists — those models collapse a list of bare strings into one document (#745), and only a live call catches it
- Embeddings (`test_live_embeddings.py`): single embed, batched embed, batch-vs-single alignment, and chunk-to-id mapping for every configured embedding model. `gemini-embedding-2*` is the reason this exists — those models collapse a list of bare strings into one document (#745), and only a live call catches it. Also covers first-class `EmbeddingModelConfig.timeout` plumbing (one representative model per transport): configured timeout lands on the SDK client, and a near-zero timeout aborts before the provider answers
- OpenAI-compatible embedding providers (e.g. OpenRouter's `google/gemini-embedding-001`): the #932 surface. Those providers reject a base64 embedding request outright (HTTP 400) or answer HTTP 200 with empty data, so the whole matrix fails without `encoding_format="float"`. Real OpenAI accepts base64 happily, so only a third-party provider catches it. Note that OpenRouter load-balances across upstreams, so the base64 failure is per-attempt rather than guaranteed: a retry can land on an endpoint that accepts it. `test_live_openai_float_encoding_matches_base64` covers the other side, that the float switch must not move vectors on real OpenAI

View File

@ -66,14 +66,23 @@ def require_embedding_key(spec: LiveEmbeddingSpec) -> str:
return key
_EMBEDDING_CONFIG_OVERRIDE_KEYS = frozenset({"timeout", "max_batch_size"})
def make_embedding_client(
spec: LiveEmbeddingSpec, **overrides: Any
) -> _EmbeddingClient:
"""Build a live embedding client for one matrix entry.
Bypasses the `EmbeddingClient` singleton so each spec gets its own client
without mutating global settings.
without mutating global settings. `timeout` and `max_batch_size` land on
`EmbeddingModelConfig`; remaining kwargs go to `_EmbeddingClient`.
"""
config_overrides = {
key: overrides.pop(key)
for key in _EMBEDDING_CONFIG_OVERRIDE_KEYS
if key in overrides
}
kwargs: dict[str, Any] = {
"vector_dimensions": spec.dimensions,
"max_input_tokens": 2048,
@ -90,6 +99,7 @@ def make_embedding_client(
model=spec.model,
api_key=require_embedding_key(spec),
base_url=spec.base_url,
**config_overrides,
),
**kwargs,
)

View File

@ -1,10 +1,15 @@
from __future__ import annotations
import time
from typing import Any, cast
import httpx
import openai
import pytest
from openai import AsyncOpenAI
from src.config import EmbeddingTransport
from .conftest import cosine_similarity, make_embedding_client
from .embedding_matrix import LiveEmbeddingSpec, get_live_embedding_specs
@ -25,6 +30,52 @@ OPENAI_NATIVE_SPECS = tuple(
spec for spec in ALL_SPECS if spec.family == "openai_embedding"
)
GENEROUS_TIMEOUT_SECONDS = 120.0
TIGHT_TIMEOUT_SECONDS = 0.01
# Well under the client defaults; generous enough to absorb SDK retries.
TIGHT_TIMEOUT_WALL_CLOCK_LIMIT_SECONDS = 30
EMBEDDING_TIMEOUT_EXCEPTIONS: dict[
EmbeddingTransport, tuple[type[BaseException], ...]
] = {
"openai": (openai.APITimeoutError,),
# google-genai raises httpx or aiohttp timeouts depending on its transport;
# aiohttp surfaces as asyncio.TimeoutError (== builtins.TimeoutError).
"gemini": (httpx.TimeoutException, TimeoutError),
}
TRANSPORT_MARKS = {
"openai": pytest.mark.requires_openai,
"gemini": pytest.mark.requires_gemini,
}
def representative_embedding_specs() -> list[Any]:
"""One spec per transport — timeout plumbing is client-level, not model-level."""
params: list[Any] = []
for transport in ("openai", "gemini"):
specs = get_live_embedding_specs(transport=transport)
if not specs:
continue
# Prefer the native family over openai-compatible proxies.
family = f"{transport}_embedding"
native = next((s for s in specs if s.family == family), specs[0])
params.append(
pytest.param(native, marks=TRANSPORT_MARKS[transport], id=native.id)
)
return params
def assert_embedding_timeout_on_client(
client: Any, transport: EmbeddingTransport, timeout_seconds: float
) -> None:
if transport == "gemini":
http_options = client.client._api_client._http_options
assert http_options.timeout == int(timeout_seconds * 1000)
return
openai_client = cast(AsyncOpenAI, client.client)
assert openai_client.timeout == timeout_seconds
@pytest.mark.asyncio
@pytest.mark.parametrize("spec", ALL_SPECS, ids=lambda spec: spec.id)
@ -160,3 +211,36 @@ async def test_live_gemini_batch_embed_survives_batch_split(
assert len(embeddings) == len(BATCH_TEXTS)
assert len({tuple(embedding) for embedding in embeddings}) == len(BATCH_TEXTS)
@pytest.mark.asyncio
@pytest.mark.parametrize("spec", representative_embedding_specs())
async def test_live_embedding_timeout_reaches_the_client(
spec: LiveEmbeddingSpec,
) -> None:
"""Configured embedding timeout lands on the provider SDK client."""
client = make_embedding_client(spec, timeout=GENEROUS_TIMEOUT_SECONDS)
embedding = await client.embed(BATCH_TEXTS[0])
assert len(embedding) == spec.dimensions
assert_embedding_timeout_on_client(client, spec.transport, GENEROUS_TIMEOUT_SECONDS)
@pytest.mark.asyncio
@pytest.mark.parametrize("spec", representative_embedding_specs())
async def test_live_tight_embedding_timeout_aborts_request(
spec: LiveEmbeddingSpec,
) -> None:
"""A near-zero embedding timeout aborts before the provider can answer."""
client = make_embedding_client(spec, timeout=TIGHT_TIMEOUT_SECONDS)
started = time.monotonic()
with pytest.raises(EMBEDDING_TIMEOUT_EXCEPTIONS[spec.transport]):
await client.embed(BATCH_TEXTS[0])
elapsed = time.monotonic() - started
assert elapsed < TIGHT_TIMEOUT_WALL_CLOCK_LIMIT_SECONDS, (
f"tight embedding timeout took {elapsed:.1f}s — client timeout "
f"likely not applied"
)

View File

@ -65,7 +65,13 @@ async def test_openai_embedding_client_uses_configured_model_and_dimensions(
fake_embeddings = FakeOpenAIEmbeddingsAPI([0.1] * 8)
class FakeOpenAIClient:
def __init__(self, *, api_key: str | None, base_url: str | None) -> None:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.api_key: str | None = api_key
self.base_url: str | None = base_url
self.embeddings: FakeOpenAIEmbeddingsAPI = fake_embeddings
@ -104,7 +110,13 @@ async def test_openai_embedding_client_rejects_dimension_mismatch(
fake_embeddings = FakeOpenAIEmbeddingsAPI([0.1] * 7)
class FakeOpenAIClient:
def __init__(self, *, api_key: str | None, base_url: str | None) -> None:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.embeddings: FakeOpenAIEmbeddingsAPI = fake_embeddings
monkeypatch.setattr("openai.AsyncOpenAI", FakeOpenAIClient)
@ -220,6 +232,118 @@ async def test_gemini_embedding_client_keeps_timeout_without_base_url(
assert gemini_client.http_options.timeout == 600_000
@pytest.mark.asyncio
async def test_openai_embedding_client_forwards_timeout(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Configured embedding timeout reaches the OpenAI-compatible client."""
class FakeOpenAIClient:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.api_key: str | None = api_key
self.base_url: str | None = base_url
self.timeout: float | None = timeout
self.embeddings: FakeOpenAIEmbeddingsAPI = FakeOpenAIEmbeddingsAPI(
[0.1] * 8
)
monkeypatch.setattr("openai.AsyncOpenAI", FakeOpenAIClient)
client = _EmbeddingClient(
EmbeddingModelConfig(
transport="openai",
model="text-embedding-3-small",
api_key="test-key",
timeout=45,
),
vector_dimensions=8,
max_input_tokens=8192,
max_tokens_per_request=300_000,
send_dimensions=False,
)
openai_client = cast(Any, client.client)
assert openai_client.timeout == 45.0
@pytest.mark.asyncio
async def test_openai_embedding_client_omits_timeout_when_unset(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Unset timeout omits the kwarg so the OpenAI SDK keeps its default."""
missing = object()
class FakeOpenAIClient:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: object = missing,
) -> None:
self.api_key: str | None = api_key
self.base_url: str | None = base_url
self.timeout: object = timeout
self.embeddings: FakeOpenAIEmbeddingsAPI = FakeOpenAIEmbeddingsAPI(
[0.1] * 8
)
monkeypatch.setattr("openai.AsyncOpenAI", FakeOpenAIClient)
client = _EmbeddingClient(
EmbeddingModelConfig(
transport="openai",
model="text-embedding-3-small",
api_key="test-key",
),
vector_dimensions=8,
max_input_tokens=8192,
max_tokens_per_request=300_000,
send_dimensions=False,
)
openai_client = cast(Any, client.client)
assert openai_client.timeout is missing
@pytest.mark.asyncio
async def test_gemini_embedding_client_forwards_timeout(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Configured embedding timeout reaches Gemini as milliseconds."""
class FakeGeminiClient:
def __init__(self, *, api_key: str | None, http_options: Any) -> None:
self.api_key: str | None = api_key
self.http_options: Any = http_options
self.aio: Any = SimpleNamespace(models=SimpleNamespace())
monkeypatch.setattr("google.genai.Client", FakeGeminiClient)
client = _EmbeddingClient(
EmbeddingModelConfig(
transport="gemini",
model="gemini-embedding-001",
api_key="gemini-key",
timeout=45,
),
vector_dimensions=8,
max_input_tokens=4096,
max_tokens_per_request=300_000,
send_dimensions=False,
)
gemini_client = cast(Any, client.client)
assert gemini_client.http_options.timeout == 45_000
def _build_openai_client(
monkeypatch: pytest.MonkeyPatch,
*,
@ -233,7 +357,13 @@ def _build_openai_client(
fake_embeddings = FakeOpenAIEmbeddingsAPI(embedding)
class FakeOpenAIClient:
def __init__(self, *, api_key: str | None, base_url: str | None) -> None:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.api_key: str | None = api_key
self.base_url: str | None = base_url
self.embeddings: FakeOpenAIEmbeddingsAPI = fake_embeddings
@ -707,7 +837,13 @@ async def test_simple_batch_embed_respects_token_budget_per_request(
fake_embeddings = FakeOpenAIEmbeddingsAPI([0.5] * 4)
class FakeOpenAIClient:
def __init__(self, *, api_key: str | None, base_url: str | None) -> None:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.embeddings: FakeOpenAIEmbeddingsAPI = fake_embeddings
monkeypatch.setattr("openai.AsyncOpenAI", FakeOpenAIClient)
@ -745,7 +881,13 @@ async def test_simple_batch_embed_rejects_oversized_input(
fake_embeddings = FakeOpenAIEmbeddingsAPI([0.1] * 4)
class FakeOpenAIClient:
def __init__(self, *, api_key: str | None, base_url: str | None) -> None:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.embeddings: FakeOpenAIEmbeddingsAPI = fake_embeddings
monkeypatch.setattr("openai.AsyncOpenAI", FakeOpenAIClient)
@ -775,7 +917,13 @@ def test_prepare_chunks_returns_ordered_chunks(
fake_embeddings = FakeOpenAIEmbeddingsAPI([0.1] * 4)
class FakeOpenAIClient:
def __init__(self, *, api_key: str | None, base_url: str | None) -> None:
def __init__(
self,
*,
api_key: str | None,
base_url: str | None,
timeout: float | None = None,
) -> None:
self.embeddings: FakeOpenAIEmbeddingsAPI = fake_embeddings
monkeypatch.setattr("openai.AsyncOpenAI", FakeOpenAIClient)
@ -818,6 +966,31 @@ def test_embedding_model_config_parses_max_batch_size_from_env(
assert resolved.max_batch_size == 10
def test_embedding_model_config_parses_timeout_from_env(
monkeypatch: pytest.MonkeyPatch,
) -> None:
s = _build_embedding_settings(
{"EMBEDDING_MODEL_CONFIG__TIMEOUT": "90.0"},
monkeypatch,
)
assert s.MODEL_CONFIG.timeout == 90.0
resolved = resolve_embedding_model_config(s.MODEL_CONFIG)
assert resolved.timeout == 90.0
def test_embedding_model_config_rejects_invalid_timeout() -> None:
with pytest.raises(
ValueError, match=r"provider_params\.timeout must be a positive number"
):
EmbeddingModelConfig(
transport="openai",
model="text-embedding-3-small",
timeout=-1,
)
@pytest.mark.asyncio
async def test_gemini_process_batch_wraps_contents_as_content_part(
monkeypatch: pytest.MonkeyPatch,

View File

@ -175,6 +175,117 @@ async def test_polling_loop_idle_sleeps_once_per_cycle(
assert sleeps == [1.0, 2.0, 4.0, 8.0, 8.0]
def test_is_tenant_work_ignores_reconciler_only_batches() -> None:
from src.deriver.queue_manager import QueueManager
assert not QueueManager._is_tenant_work(["reconciler:sync_vectors"]) # pyright: ignore[reportPrivateUsage]
assert not QueueManager._is_tenant_work( # pyright: ignore[reportPrivateUsage]
["reconciler:sync_vectors", "reconciler:cleanup_queue"]
)
# A mixed batch is tenant work: real work is present alongside housekeeping.
assert QueueManager._is_tenant_work( # pyright: ignore[reportPrivateUsage]
["reconciler:sync_vectors", "representation:ws:sess:peer"]
)
# Unparseable keys count as tenant work so an unknown key can't strand the
# loop in a long sleep.
assert QueueManager._is_tenant_work(["not-a-real-key"]) # pyright: ignore[reportPrivateUsage]
@pytest.mark.asyncio
async def test_reconciler_work_does_not_reset_backoff(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The reconciler enqueues sweeps on its own timer. Claiming one must not
look like a busy queue, or the backoff resets every cycle and the pooler
never releases an idle tenant's connection."""
monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_ENABLED", True)
monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_INTERVAL_SECONDS", 1.0)
monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_MULTIPLIER", 2.0)
monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_MAX_INTERVAL_SECONDS", 64.0)
monkeypatch.setattr(settings.DERIVER, "POLLING_JITTER_RATIO", 0.0)
import asyncio
from src.deriver import queue_manager as qm_mod
qm = qm_mod.QueueManager()
sleeps: list[float] = []
polls = {"n": 0}
async def fake_cleanup() -> None:
return None
async def fake_claim() -> dict[str, str]:
polls["n"] += 1
if polls["n"] >= 5:
qm.shutdown_event.set()
# Third poll hands back a reconciler sweep; the rest are empty.
return {"reconciler:sync_vectors": "aqs-1"} if polls["n"] == 3 else {}
async def fake_process(_work_unit_key: str, _worker_id: str) -> None:
return None
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
monkeypatch.setattr(qm, "cleanup_stale_work_units", fake_cleanup)
monkeypatch.setattr(qm, "get_and_claim_work_units", fake_claim)
monkeypatch.setattr(qm, "process_work_unit", fake_process)
monkeypatch.setattr(asyncio, "sleep", fake_sleep)
await qm.polling_loop()
# Polls 1 and 2 sleep 1 and 2. Poll 3 claims the sweep, so it neither sleeps
# nor resets. Polls 4 and 5 resume the schedule at 4 -- not back at 1.
assert sleeps == [1.0, 2.0, 4.0, 8.0]
@pytest.mark.asyncio
async def test_tenant_work_still_resets_backoff(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Real tenant work must still snap the interval back for fast pickup."""
monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_ENABLED", True)
monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_INTERVAL_SECONDS", 1.0)
monkeypatch.setattr(settings.DERIVER, "POLLING_BACKOFF_MULTIPLIER", 2.0)
monkeypatch.setattr(settings.DERIVER, "POLLING_SLEEP_MAX_INTERVAL_SECONDS", 64.0)
monkeypatch.setattr(settings.DERIVER, "POLLING_JITTER_RATIO", 0.0)
import asyncio
from src.deriver import queue_manager as qm_mod
qm = qm_mod.QueueManager()
sleeps: list[float] = []
polls = {"n": 0}
async def fake_cleanup() -> None:
return None
async def fake_claim() -> dict[str, str]:
polls["n"] += 1
if polls["n"] >= 5:
qm.shutdown_event.set()
return {"representation:ws:sess:peer": "aqs-1"} if polls["n"] == 3 else {}
async def fake_process(_work_unit_key: str, _worker_id: str) -> None:
return None
async def fake_sleep(seconds: float) -> None:
sleeps.append(seconds)
monkeypatch.setattr(qm, "cleanup_stale_work_units", fake_cleanup)
monkeypatch.setattr(qm, "get_and_claim_work_units", fake_claim)
monkeypatch.setattr(qm, "process_work_unit", fake_process)
monkeypatch.setattr(asyncio, "sleep", fake_sleep)
await qm.polling_loop()
# Poll 3 finds tenant work, so polls 4 and 5 start over from the base
# interval rather than continuing from 4.
assert sleeps == [1.0, 2.0, 1.0, 2.0]
def test_inflight_gauge_no_drift(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(settings.METRICS, "NAMESPACE", "test")
child: Any = db_queries_in_flight_gauge.labels(instance_type="api")