diff --git a/docs/memory-mcp.md b/docs/memory-mcp.md new file mode 100644 index 00000000..5896adb5 --- /dev/null +++ b/docs/memory-mcp.md @@ -0,0 +1,493 @@ +# 专家记忆 MCP Server 使用文档 + +> Octop 把**专家记忆**通过标准 **MCP（Streamable HTTP）** 暴露给外部 agent（编码 agent、机器人、 +> 其他 AI 工具），能力与进程内 `MemoryService` 对齐：外部 agent 可以召回记忆、按路径下钻读全文、 +> 写入新的事件与事实，并能区分"这条记忆是谁说的"。 +> +> 当前工具面：**11 个 `memory_*` 工具**（读取 4 个 / 写入 3 个 / 提取与审核 4 个）。 + +--- + +## 1. 连接信息 + +| 项 | 值 | +|---|---| +| Endpoint | `http(s):///mcp/memory`（MCP 侧实际请求路径为 `/mcp/memory/`） | +| 协议 | MCP Streamable HTTP（SSE + JSON） | +| 鉴权 | `Authorization: Bearer ` | +| 专家绑定 | `X-Octop-Agent-Id: ` 请求头 | +| 调用者标识 | `X-Octop-User-Id: ` 请求头（可选，见 §1.3） | + +### 1.1 鉴权（fail-closed） + +- 服务端通过环境变量 `OCTOP_MEMORY_MCP_TOKEN` 配置这条通道的独立 token；**未配置时整个 + `/mcp/memory` 端点不挂载**（安全默认）。 +- 每个请求都必须带 `Authorization: Bearer `，否则返回 `401`。 +- 建议该 token 与 Octop 登录凭据完全隔离——它是独立凭据，仅用于记忆通道。 + +### 1.2 专家绑定（按请求校验，无需重启） + +- 端点只有一个 `/mcp/memory`，URL 里不含专家 id；专家由请求头 `X-Octop-Agent-Id` 选择。 +- **每个请求都会用该 id 去专家注册表校验**（存在且处于启用状态），校验不通过返回 `404`。 + 因此新建或停用专家**立即生效**，不需要重启服务。 +- 所有读写都落在该专家的 `Memory` 实例上，工具调用时**不再**传 agent id。 +- 记忆是**专家级共享**的：同一专家下的多个外部调用者看到同一份记忆，不做按人硬隔离； + 需要按人定位时依赖 §1.3 的发送者归属。 + +### 1.3 调用者归属（`X-Octop-User-Id`） + +- 调用者身份通过 `X-Octop-User-Id` 头（或工具的显式 `user` 参数，优先级更高）传递。 +- `memory_capture` / `memory_save` / `memory_update` 会把调用者 id **拼进内容前缀** + `说：…` 后再落库。原因：`AtomCard` 没有 user 列，把发送者写进正文才能随原子一起 + 落库、被全文检索命中（查询里带上发送者名字即可命中其记忆）、并在召回时直接可见。 +- 因此**不要把名字重复写进 `content`**；内容已带同名前缀时不会重复拼接；不带该头时保持原文。 +- 未匹配到调用者身份时，写入仍然成功，只是没有前缀。 + +--- + +## 2. 工具清单（11 个） + +### 2.1 读取（4 个） + +| 工具 | 参数 | 说明 | +|---|---|---| +| `memory_recall` | `query: str`, `limit: int = 5`, `session_id?: str`, `thread_id?: str`, `user?: str` | **读入口首选**。跑完整召回管线（分词 → 路由 → FTS → 重排 → 去重），返回结构化片段 + 可直接注入 system prompt 的 markdown（`rendered`）。L2 原子优先，主题页标题并入 atom 命中，L0 原始事件仅兜底。**自动注入（hook）场景建议传 `session_id`**：管线会据此把本会话的 raw 排除，避免"注入 → 被记录 → 下轮又召回"的回声；`thread_id` 则启用共指消解（"那个项目" 靠该线程的 active-entity stack） | +| `memory_search` | `query: str`, `max_results: int = 5`, `corpus: str = "all"` | 同一套召回管线，但**不渲染 markdown**，而是给每条命中一个虚拟 `path`，交给 `memory_get` 下钻。`corpus`：`all`/`memory`（原子+原始事件同管线）、`atom`（只要 L2 原子）、`raw`（直接走 L0 全文检索，不受"有原子命中就丢 raw"的兜底影响） | +| `memory_get` | `path: str`, `start?: int`, `lines?: int` | 把命中路径解析成完整 markdown。支持 `atom/.md`、`page/.md`、`raw//.md`；长内容用 `start`/`lines` 分页（1-based）。路径非法或过期时返回 `{error, hint}` 而不是抛栈 | +| `memory_raws` | `query?: str`, `session_id?: str`, `host?: str`, `user?: str`, `limit: int = 50` | **原始事件（证据源）**。`query` 走全文检索（写入后立即可见，提取前也能查），其余字段做结构化过滤，按时间倒序返回 | + +返回形状： + +```jsonc +// memory_recall +{ "memories": [{ "source_id", "timestamp", "layer", "text" }], "count", "rendered", "caller" } +// memory_search +{ "hits": [{ "path", "layer", "snippet", "occurred_at", "source_id" }], "total", "corpus", "hint" } +// memory_get +{ "path", "kind", "content", "total_lines", "from_line", "to_line", "truncated", "metadata" } +// memory_raws +{ "events": [{ "event_id", "timestamp", "session_id", "user", "event_type", "source", "content" }], "count", "caller" } +``` + +### 2.2 写入（3 个，两条通道） + +| 工具 | 参数 | 说明 | +|---|---|---| +| `memory_capture` | `content: str`, `source: str`, `session_id?: str`, `user?: str` | **记录原始事件（L0）**：走 提取 → 候选 → 晋升 → 原子 的流水线，适合记录对话/事件原文。写入后立即可用 `memory_raws` / `memory_search(corpus="raw")` 查；晋升成原子后才能被 `memory_recall` 召回。`session_id` 缺省派生为 `ext:{source}:{user}`，用于让提取管线按会话分组蒸馏 | +| `memory_save` | `content: str`, `source: str`, `topic?: str`, `user?: str` | **直接保存事实（L2）**：不经过提取，立即可召回。适合已知的明确事实/约定 | +| `memory_update` | `atom_id: str`, `new_content: str`, `source: str`, `note: str = "mcp update"`, `user?: str` | **更新记忆**：旧原子标记 deprecated，新事实立刻可召回，并带 `supersedes` 关联。适合纠正过时事实 | + +`memory_capture` 是**幂等**的：同一 `session_id` + 同一内容重复写入时不会产生重复 L0 事件， +返回里带 `duplicate: true` 并复用已有 `event_id`（下游提取因此可以反复重跑）。 + +它还有**回声保护**：内容里带 `memory_recall` 的注入标记（`[memory] Earlier in this workspace` / +`## Memory Recall`）时**不写入**，直接返回 `{recorded: false, skipped: "recall_echo"}`。 +原因：MCP 写路径直接调 `Memory.add_raw`，绕过了 `MemoryRuntime.capture` 的 `skip_memory_echo`； +不补这层，hook 注入的召回块会被当作新事件采集，形成"注入 → 采集 → 再召回"的放大环。 + +### 2.3 提取与审核流水线（4 个） + +| 工具 | 参数 | 说明 | +|---|---|---| +| `memory_extract` | `session_id?: str`, `limit: int = 100`, `promote: bool = False` | 手动触发提取：取最近 L0 事件 → LLM 类型化提取 → 候选（pending）；`promote=True` 时直接跑晋升检查。复用**该专家进程内**的 `MemoryService`（含其配置的提取模型）；专家未运行/无记忆运行时时返回 `error` | +| `memory_candidates` | `status?: str`, `session_id?: str`, `limit: int = 50` | 列出 L1 候选（默认 pending）。`status` 可取 `pending` / `promoted` / `rejected` / `needs_review` / `conflict`，非法值报错 | +| `memory_promote` | `candidate_ids: list[str]`, `importance?: str` | 审核晋升：对指定候选跑 5 项晋升检查，通过则写入 L2 原子并记录 journal | +| `memory_reject` | `candidate_id: str`, `reason: str = "rejected by external caller"` | 拒绝候选并记录原因（不进原子层，journal 可审计） | + +--- + +## 3. 记忆分层与选工具 + +| 层级 | 内容 | 对应工具 | +|---|---|---| +| L0 | 原始事件（原话、证据） | `memory_capture` 写入；`memory_raws` / `memory_search(corpus="raw")` 读取 | +| L1 | 候选（提取产物，待审核） | `memory_extract` 产生；`memory_candidates` 查看；`memory_promote` / `memory_reject` 裁决 | +| L2 | 原子（长期记忆，可被召回） | `memory_save` / `memory_update` 直写；`memory_capture` 经流水线晋升；`memory_recall` / `memory_search` 召回 | +| L3 | 主题页（实体页） | 随晋升重新生成；可用 `memory_get(page/.md)` 读取 | + +按意图选工具： + +- 只想把相关背景拉进上下文 → `memory_recall` +- 要定位某条具体记忆并读全文 → `memory_search` 拿 `path`，再 `memory_get(path)` +- 要原话/证据，或内容刚写入还没晋升 → `memory_raws`（或 `memory_search(corpus="raw")`） +- 记录对话/事件 → `memory_capture`；记录明确事实/规则 → `memory_save`；纠正过时事实 → `memory_update` +- 处理审核队列 → `memory_candidates` → `memory_promote` / `memory_reject` + +--- + +## 4. 使用示例 + +> 下面默认客户端**直连** Octop 的 `/mcp/memory`。如果 Octop 前面还挂了一层网关/代理（例如由代理统一校验调用人身份并注入上游 token），只需把 URL 换成代理端点、按代理要求填它需要的凭证（通常是**一个** token 头）；此时 `Authorization` / `X-Octop-User-Id` 由代理负责，**不要**在客户端重复配置。 + +> 接入任一客户端都是**两步**：① 配好 MCP 服务器（§4.3 Kiro / §4.4 Claude Code / §4.5 DSH 各自的「配置」段）——这一步只让工具**可用**；② 再配**自动化**（Kiro 与 Claude Code 挂 hooks；DSH 新建并使用「共享记忆」预设）——这一步才会**自动召回、自动沉淀**。只做第 ① 步的话，每次都得手动让模型去调工具。§4.1 / §4.2 是最简的手动自测形态。 + +### 4.1 直接 HTTP（curl） + +```bash +TOKEN="" +AGENT="" +BASE="http://127.0.0.1:/mcp/memory" + +# 1) 初始化（MCP 握手） +curl -s -X POST "$BASE/" \ + -H "Authorization: Bearer $TOKEN" \ + -H "X-Octop-Agent-Id: $AGENT" \ + -H "Content-Type: application/json" \ + -H "Accept: application/json, text/event-stream" \ + -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"cli","version":"1.0"}}}' + +# 2) 记录一条原始事件（L0） +curl -s -X POST "$BASE/" \ + -H "Authorization: Bearer $TOKEN" \ + -H "X-Octop-Agent-Id: $AGENT" \ + -H "X-Octop-User-Id: alice" \ + -H "Content-Type: application/json" \ + -H "Accept: application/json, text/event-stream" \ + -d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"memory_capture","arguments":{"content":"报告页横幅没有渲染出来","source":"review-bot"}}}' +``` + +其余工具只是换 `params.name` / `params.arguments`： + +```jsonc +// 直接保存一条事实（L2，立即可召回） +{ "name": "memory_save", + "arguments": { "content": "部署约定：记忆端点挂在 /mcp/memory", "source": "planning-agent", "topic": "octop-deploy" } } + +// 召回 +{ "name": "memory_recall", "arguments": { "query": "部署约定", "limit": 5 } } + +// 检索拿 path +{ "name": "memory_search", "arguments": { "query": "部署约定", "corpus": "atom" } } + +// 读全文（path 来自上一步 hits[].path） +{ "name": "memory_get", "arguments": { "path": "atom/.md", "lines": 40 } } +``` + +### 4.2 MCP 客户端 SDK（Python） + +```python +# pip install mcp +import asyncio + +from mcp import ClientSession +from mcp.client.streamable_http import streamablehttp_client + + +async def main(): + # 注意：endpoint 以 / 结尾；鉴权与专家绑定都放在 headers 里 + async with streamablehttp_client( + "http://127.0.0.1:/mcp/memory/", + headers={ + "Authorization": "Bearer ", + "X-Octop-Agent-Id": "", + "X-Octop-User-Id": "alice", + }, + ) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + tools = await session.list_tools() + print([t.name for t in tools.tools]) # 11 个 memory_* 工具 + + await session.call_tool( + "memory_capture", + {"content": "报告页横幅没有渲染出来", "source": "review-bot"}, + ) + res = await session.call_tool("memory_recall", {"query": "报告页 横幅"}) + print(res) + + +asyncio.run(main()) +``` + +### 4.3 Kiro + +Kiro 原生支持远程 MCP（`url` + `headers`）。配置分两级，同名条目以工作区为准： + +- 工作区：`.kiro/settings/mcp.json` +- 用户级：`~/.kiro/settings/mcp.json` +- 打开方式：命令面板 `Kiro: Open workspace MCP config (JSON)` / `Kiro: Open user MCP config (JSON)`；保存后自动热重载。 + +```jsonc +{ + "mcpServers": { + "octop-memory": { + "type": "streamable_http", + "url": "https:///mcp/memory/", + "headers": { + "Authorization": "Bearer ", + "X-Octop-Agent-Id": "", + "X-Octop-User-Id": "" + }, + "disabled": false + } + } +} +``` + +Kiro 侧注意事项： + +- **远程 `url` 必须是 `https`**（只有本地回环允许 `http`）。 +- 工具名带**服务器名前缀**：服务器名 `octop-memory` → 工具 `mcp_octop_memory_memory_recall`；在 hooks 或提示词里必须写全名。 +- 工具默认不放行：在 `~/.kiro/settings/permissions.yaml` 里找到 `capability: mcp` 的 `match` 列表，按需加 `effect: allow`（只读先放 `memory_recall` / `memory_raws`，需要沉淀再放 `memory_capture` / `memory_save`）。 +- 用 `${VAR}` 引用环境变量时，要先把变量加入 Kiro 的允许列表（设置项 `Mcp Approved Env Vars`），否则不会被展开。 +- 校验：会话里执行 `/mcp`，确认服务器已连接、工具已加载。 + +#### 自动化：hooks（配好 MCP 之后必配） + +**只配 `mcp.json` 只是"工具能用"——不会自动召回、也不会自动沉淀，必须再挂 hooks。** Kiro 的 hooks 放在 `~/.kiro/hooks/*.json`，用 `trigger` + `action.type: agent` 让模型在固定时机去调对应工具。 + +提交提示词时召回 —— `~/.kiro/hooks/memory-recall-on-prompt-submit.json`： + +```jsonc +{ + "version": "v1", + "hooks": [ + { + "name": "Recall Memory on Prompt Submit", + "trigger": "UserPromptSubmit", + "action": { + "type": "agent", + "prompt": "回答前，按需主动召回相关记忆来辅助本次工作。结合用户本次输入与当前项目上下文构造查询，调用 `mcp_octop_memory_memory_recall`。若本次输入与已召回记忆无关可跳过，不要重复召回同一查询。" + }, + "description": "收到消息时自动召回相关记忆，提供上下文连续性。", + "enabled": true + } + ] +} +``` + +会话结束时沉淀 —— `~/.kiro/hooks/memory-save-on-stop.json`： + +```jsonc +{ + "version": "v1", + "hooks": [ + { + "name": "Save Memory on Session End", + "trigger": "Stop", + "action": { + "type": "agent", + "prompt": "会话结束前静默沉淀：1) 调用 `mcp_octop_memory_memory_capture` 记录本次会话（source=\"kiro-session\"，session_id 形如 \"session-YYYY-MM-DD-主题\"）；2) 对值得长期保留的事实（偏好、约定、架构决策、环境配置）调用 `mcp_octop_memory_memory_save`（source=\"kiro-session\"）。只写对未来确有价值的信息；成功后一句话告知，不要复述会话内容。" + }, + "description": "会话结束时把可复用事实沉淀到 octop-memory。", + "enabled": true + } + ] +} +``` + +改完 hooks / permissions 后重载或重启 Kiro 生效。 + +### 4.4 Claude Code + +Claude Code 直接支持远程 HTTP + 自定义请求头： + +```bash +claude mcp add --transport http octop-memory https:///mcp/memory/ \ + --header "Authorization: Bearer " \ + --header "X-Octop-Agent-Id: " \ + --header "X-Octop-User-Id: " +``` + +`--header` 可重复（短写 `-H`）。也可以直接写配置文件（`.mcp.json` 项目级 / `~/.claude/settings.json` 用户级）： + +```jsonc +{ + "mcpServers": { + "octop-memory": { + "type": "http", + "url": "https:///mcp/memory/", + "headers": { + "Authorization": "Bearer ${OCTOP_MEMORY_MCP_TOKEN}", + "X-Octop-Agent-Id": "", + "X-Octop-User-Id": "" + } + } + } +} +``` + +Claude Code 侧注意事项： + +- Streamable HTTP 写 `"type": "http"`（MCP 规范名 `streamable-http` 也接受）；**带 `url` 就必须带 `type`**，否则该条目会被当作 stdio 服务器跳过。 +- 工具名带前缀：`mcp__octop-memory__memory_recall`。 +- 排查：`claude mcp list`、`claude mcp get octop-memory`，会话内 `/mcp`。 + +#### 自动化：hooks（配好 MCP 之后必配） + +同样，**只加 MCP 服务器不会自动召回/沉淀**。Claude Code 在 `settings.json` 的 `hooks` 字段挂命令脚本（脚本放 `~/.claude/hook-events/`），两个事件各一个： + +| 脚本 | 事件 | 作用 | +|---|---|---| +| `octop-user-prompt-recall.mjs` | `UserPromptSubmit` | 向 stdout 注入「先召回」的指令；真正调用工具的是模型，脚本**不直连 MCP**，任何异常静默放行、不阻断本轮 | +| `octop-stop-capture.mjs` | `Stop` | 输出 `{"decision":"block","reason":"..."}`，让模型在追加的一轮里执行 `memory_capture`；`stop_hook_active` 为真时直接退出，防止召回/沉淀互相触发成死循环 | + +```jsonc +// ~/.claude/settings.json +{ + "hooks": { + "UserPromptSubmit": [ + { "hooks": [ { "type": "command", "command": "node ~/.claude/hook-events/octop-user-prompt-recall.mjs" } ] } + ], + "Stop": [ + { "hooks": [ { "type": "command", "command": "node ~/.claude/hook-events/octop-stop-capture.mjs" } ] } + ] + } +} +``` + +`octop-user-prompt-recall.mjs` 最小实现： + +```javascript +#!/usr/bin/env node +// UserPromptSubmit：把「先召回」指令注入本轮上下文；真正调用 MCP 工具的是模型。 +process.stdout.write( + "[octop-memory] 回答前先按需召回相关记忆：结合用户输入与当前项目上下文构造查询，" + + "调用 MCP 工具 mcp__octop-memory__memory_recall；与本轮无关可跳过，不要重复召回同一查询。\n", +); +``` + +`octop-stop-capture.mjs` 最小实现： + +```javascript +#!/usr/bin/env node +// Stop：让模型多跑一轮，把本轮可复用事实写入 L0。 +import fs from "node:fs"; + +let payload = {}; +try { + const raw = fs.readFileSync(0, "utf8"); + if (raw.trim()) payload = JSON.parse(raw); +} catch { + // 解析失败也继续：沉淀不该因为载荷缺字段而丢掉 +} + +if (payload.stop_hook_active) process.exit(0); // 防召回/沉淀互相触发 + +process.stdout.write( + JSON.stringify({ + decision: "block", + reason: + "会话结束前静默沉淀：调用 mcp__octop-memory__memory_capture 写入本轮可复用事实" + + '（source="claude-code"）；没有可沉淀内容就不要写入，成功后一句话告知即可。', + }) + "\n", +); +``` + +### 4.5 DSH（DeepSeek Harness） + +DSH 在 **Settings → MCP** 里管理 MCP 服务器（定义持久化在 `~/.dsh/storages/mcp_servers.json`）。设置页新增一条 Streamable HTTP 服务器时，对应的字段如下： + +```jsonc +{ + "serverName": "octop-memory", + "transport": "streamable-http", + "enabled": true, + "url": "https:///mcp/memory/", + "headers": [ + { "name": "Authorization", "value": "Bearer ${OCTOP_MEMORY_MCP_TOKEN}" }, + { "name": "X-Octop-Agent-Id", "value": "" }, + { "name": "X-Octop-User-Id", "value": "" } + ], + "toolCallTimeoutMs": 60000, + "failOnStartupError": true +} +``` + +注意事项： + +- `transport` 取 `streamable-http`（另支持 `stdio`）；`enabled: false` 停用整条；`failOnStartupError` 决定连接失败是否阻断启动。 +- 设置页里的 Headers 是每行一个 `名称: 值`，值支持 `${ENV}` 替换；密钥类变量交给 DSH 的全局环境变量/凭据存储，不要写进明文配置。 +- DSH 默认**按需注入** MCP 工具：会话里先检索一次，才会把 `memory_*` 挂进当前对话——所以只加服务器**更不会**自动召回/沉淀。 +- 新增/修改服务器后按提示重连；若工具列表没有刷新，重启 `dsh web`。 + +#### 自动化：共享记忆预设（配好 MCP 之后必配） + +和 Kiro / Claude Code 要挂 hooks 一样，DSH 这边还要**新增一个「共享记忆」模式的 agent preset，并用它开会话**，自动化才会发生： + +- 会话首轮：先 `memory_recall`，把相关原子记忆拉进上下文； +- 每轮结束：若本轮产生了可沉淀的事实，调用一次 `memory_capture`（`source` 标记客户端来源、`session_id` 用工作目录派生）；寒暄或已记录内容不重复写。 + +预设是一个目录（`~/.dsh/.agent-presets/<你的预设>/`），含 `preset.yml`（名称与描述）和 `agent.cordis.yml`（该预设的完整组合）： + +```text +<你的预设>/ + preset.yml + agent.cordis.yml # 共享记忆策略写在 persona 文本末尾 +``` + +`agent.cordis.yml` 的 `persona` 里追加的策略段（等价于 Kiro / Claude Code 的 hooks）： + +```text +── 共享记忆 (shared memory) ───────────────────────────── +This preset auto-reads and auto-writes the octop-memory expert store so durable facts carry across sessions and experts: memory_recall at conversation start, memory_capture at turn end. + +会话开始时(本会话首轮、尚无历史):若工具列表里没有 `memory_recall`,先按需注入 octop-memory 工具,再调用 `memory_recall(query=用户当前目标全文)`,把相关原子记忆纳入上下文。 + +每轮结束时(给出最终回复前):若本轮产生了可沉淀的事实(需求、决定、偏好、约定、结论、关键路径),调用一次 `memory_capture`: + content: 1–3 句简洁事实摘要(不写机密、不做原始转储) + source: 'dsh:shared-memory' + session_id: 'ext:dsh:<当前工作目录名>' +寒暄或已记录内容不重复记录;没有可沉淀事实就不调用。 +``` + +- 从自带预设**拷贝到自己目录再改**，不要直接改部署自带的预设（升级会覆盖）。 +- 建好后要在会话里**选择这个预设**才生效；已经在跑的会话不会自动切换。 + +### 4.6 其它 MCP 客户端 + +多数客户端共用同一份 JSON 描述一个 Streamable HTTP server： + +```jsonc +{ + "mcpServers": { + "octop-memory": { + "type": "streamable-http", + "url": "https:///mcp/memory/", + "headers": { + "Authorization": "Bearer ", + "X-Octop-Agent-Id": "", + "X-Octop-User-Id": "" + } + } + } +} +``` + +只支持 stdio 的客户端可以用 `mcp-remote` 这类桥接工具转发到远程端点；`type` 的取值随客户端而异（有的写 `http`，有的写 `streamable-http`），以客户端文档为准。自动化同样取决于该客户端有没有 hooks / 预设机制：只挂服务器通常只解决"工具可用"。 + +--- + +## 5. 部署与配置 + +| 配置 | 说明 | +|---|---| +| `OCTOP_MEMORY_MCP_TOKEN` | 必填。未设置时 `/mcp/memory` 不挂载（fail-closed） | +| `X-Octop-Agent-Id` | 每个请求必填，指向一个存在且启用的专家 | +| `X-Octop-User-Id` | 可选；用于把发送者写进记忆正文（§1.3） | +| 记忆后端 | 由专家配置 `memory.backend` 决定：默认 SQLite（`memory.sqlite`）；PostgreSQL 控制面下默认复用控制面 DSN（每专家 schema） | + +开启步骤： + +1. 设置环境变量 `OCTOP_MEMORY_MCP_TOKEN=`。 +2. 重启 Octop 服务（挂载发生在启动阶段）。 +3. 验证：不带 token 请求 `/mcp/memory/` 应返回 `401`；带 token 且带合法 + `X-Octop-Agent-Id` 时应返回 MCP 协议响应（未知/停用专家返回 `404`）。 +4. 外部 agent 按 §1 / §4 接入。 + +--- + +## 6. 行为契约与边界 + +- **写入分两条通道**：`memory_capture` 走"提取 → 候选 → 晋升"（学习型记忆，质量由流水线把关）； + `memory_save` / `memory_update` 是权威直写（规则/明确事实，立即可召回）。 + 晋升治理（提取触发、候选审核）保留在服务端，站内会话与外部写入走**同一套**记忆治理路径。 +- **capture 之后 recall 无结果属预期**：内容还在 L0，需要经提取晋升成原子才会被召回； + 想立刻看到请用 `memory_raws` 或 `memory_search(corpus="raw")`。 +- **capture 幂等**：同 `session_id` + 同内容不重复落库（见 §2.2）。 +- **召回回声有双向保护**：写入侧 `memory_capture` 丢弃带召回标记的内容（`skipped=recall_echo`）； + 读取侧建议 hook 给 `memory_recall` 传与 capture 一致的 `session_id`，把本会话的 raw 排除。 + 两侧都做，才不会出现"注入 → 采集 → 再召回"的放大环（§4.3 / §4.4 的自动召回就是这个场景）。 +- **召回是专家级共享**，不做按人隔离；按发送者定位依赖正文里的 `说：` 前缀 + 全文检索。 +- **错误形态**：`memory_get` 的坏路径返回 `{error, hint}`；`memory_search` 的非法 `corpus`、 + `memory_candidates` 的非法 `status` 直接报错（参数错误不会被静默吞掉）。 diff --git a/src/octop/api/app.py b/src/octop/api/app.py index b5e51263..9f9f0e34 100644 --- a/src/octop/api/app.py +++ b/src/octop/api/app.py @@ -273,6 +273,24 @@ async def acme_http01_challenge(token: str) -> PlainTextResponse: ], ) + # 专家记忆 MCP server（对外暴露，独立 token 鉴权，未配置 OCTOP_MEMORY_MCP_TOKEN 时不挂载） + from octop.infra.agents.memory_mcp import mount_memory_mcp + + memory_mcp_managers = mount_memory_mcp(app, server) + if memory_mcp_managers: + from collections.abc import AsyncIterator + from contextlib import AsyncExitStack, asynccontextmanager + + @asynccontextmanager + async def _memory_mcp_lifespan(application: FastAPI) -> AsyncIterator[None]: + # streamable_http_app 的 task group 依赖 lifespan，挂载后须手动并入 + async with AsyncExitStack() as stack: + for mgr in memory_mcp_managers: + await stack.enter_async_context(mgr.run()) + yield + + app.router.lifespan_context = _memory_mcp_lifespan + if enable_api_docs: @app.get("/api/docs", include_in_schema=False) diff --git a/src/octop/infra/agents/memory_mcp.py b/src/octop/infra/agents/memory_mcp.py new file mode 100644 index 00000000..202aa501 --- /dev/null +++ b/src/octop/infra/agents/memory_mcp.py @@ -0,0 +1,936 @@ +"""Expose Octop expert memory as an MCP server for external agents. + +External agents (coding agents, bots) can read/write Octop expert memory over +MCP (Streamable HTTP), aligned with the in-process ``MemoryService`` +capabilities. Every write stamps a ``source`` marker that can be traced back +on recall. + +Expert binding: the endpoint is a single ``/mcp/memory`` mount; the expert is +selected per request via the ``X-Octop-Agent-Id`` header, validated against the +agent registry on every call — the caller never passes an agent id per tool +call, and the URL itself does not leak expert ids. + +raw vs atom (aligned with ``MemoryService``): + +* ``memory_capture`` -> ``add_raw``: writes an **L0 raw event**, which goes + through the extraction pipeline (extract -> candidate -> promote -> atom). + Use it to record raw conversations / events. The record is visible + immediately via ``memory_raws``; ``memory_recall`` returns it only + after extraction promotes it to an atom. +* ``memory_save`` -> ``store``: persists a structured fact directly into the + canonical atom/tree (durable, no extraction). Use it when you already know + the exact fact to remember. + +Read surface (three tools, same storage as the in-process ``memory_search`` / +``memory_get`` exposed to Octop's own agents): + +* ``memory_recall`` -> ``recall_for_prompt``: ranked, prompt-injectable text. + L2 atoms first; ``page`` headlines are folded into atom hits; L0 raw is only + a fallback (dropped as soon as any atom matches). Takes ``session_id`` / + ``thread_id`` so an auto-inject hook can exclude the current session's raw and + use the thread's active-entity stack for co-reference. +* ``memory_search`` -> ``MemoryRuntime.memory_search`` (``corpus=raw`` uses + ``Memory.search_raw``): the same ranking, returned as hits that carry a + virtual ``path`` instead of rendered markdown. +* ``memory_get`` -> ``MemoryRuntime.memory_get``: resolve that path to the full + markdown (``atom/.md`` / ``page/.md`` / ``raw//.md``). + +Recall echo guard: ``memory_capture`` drops content carrying a recall marker +(``_RECALL_ECHO_MARKERS``), the same rule ``MemoryRuntime.capture`` applies via +``skip_memory_echo`` — the MCP write path calls ``Memory.add_raw`` directly and +would otherwise capture a hook's own injected recall block as a new event. + +Sender attribution: ``X-Octop-User-Id`` identifies the caller, and every write +(``memory_capture`` / ``memory_save`` / ``memory_update``) prefixes the content +with ``说：``. harness-memory's ``AtomCard`` has no user column, so putting +the sender into the text is what makes it reach the atom, stay FTS-searchable +(a query naming the sender matches), and remain visible on recall. + +Auth: independent token via ``OCTOP_MEMORY_MCP_TOKEN`` (fail-closed when +unset), enforced by the ASGI middleware in ``mount_memory_mcp``. +""" + +from __future__ import annotations + +import logging +import os +from contextvars import ContextVar +from typing import Any, get_args + +from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.server import Context +from mcp.server.transport_security import TransportSecuritySettings + +from octop.infra.agents.memory_backend import open_memory_kwargs +from octop.infra.server import OctopServer + +logger = logging.getLogger(__name__) + +# 当前 MCP HTTP 请求的绑定状态（由 _AgentRouter 中间件写入，工具读取）。 +# stateless streamable HTTP 下 mcp SDK 不提供 ctx.request_context，故用 contextvar +# 跨 ASGI 中间件 → 工具传递： +# - _current_agent_id: 本次请求绑定的 expert（X-Octop-Agent-Id 校验后写入） +# - _current_caller_user: 调用者 user id（供 memory_capture/save 做 per-user 追溯） +_current_agent_id: ContextVar[str] = ContextVar("octop_mcp_agent_id", default="") +_current_caller_user: ContextVar[str] = ContextVar("octop_mcp_caller_user", default="") + +_SEARCH_CORPORA: frozenset[str] = frozenset({"all", "memory", "atom", "raw"}) +"""Corpora ``memory_search`` accepts. ``atom``/``raw`` are layer filters over the +recall pipeline; ``all``/``memory`` are the same pipeline without a filter.""" + +_SNIPPET_CHARS = 200 + +_RECALL_ECHO_MARKERS: tuple[str, ...] = ( + "## Memory Recall", + "[memory] Earlier in this workspace", +) +"""Markers ``recall_for_prompt`` puts into its rendered block (legacy + current). + +``MemoryRuntime.capture`` drops events containing these (``skip_memory_echo``) so the +host's own injection cannot be captured back as a new memory. The MCP write path calls +``Memory.add_raw`` directly and therefore has to apply the same rule itself. +""" + + +def _open_memory(server: OctopServer, agent_id: str) -> Any: + """Open the agent's ``Memory`` instance (sqlite by default, postgres opt-in). + + Mirrors ``api.common.memory_client._open_memory_for_agent`` but stays in + ``infra/`` (no api dependency). Workspace is resolved from the agent + registry, falling back to the Octop default layout. + """ + from harness_memory.core import Memory # noqa: PLC0415 + + services = server.services + assert services is not None, "server.services required for memory backend" + runtime = getattr(server, "app_runtime", None) + registry = getattr(runtime, "agent_registry", None) if runtime is not None else None + if registry is not None and hasattr(registry, "resolve_workspace_dir"): + workspace = registry.resolve_workspace_dir(agent_id) + else: + paths = getattr(server, "paths", None) or services.paths + workspace = paths.ensure_agent_workspace(agent_id) + + row = services.agent_repo.get(agent_id) + cfg: dict[str, Any] = {} + if row is not None and row.config_json: + import json # noqa: PLC0415 + + try: + parsed = json.loads(row.config_json) + if isinstance(parsed, dict): + cfg = parsed + except json.JSONDecodeError: + cfg = {} + + ns, backend, backend_config = open_memory_kwargs( + agent_id=agent_id, + cfg=cfg, + octop_config=services.config, + workspace_dir=workspace, + ) + return Memory(namespace=ns, backend=backend, backend_config=backend_config) + + +def _snippet(text: str) -> str: + """Cap a raw event body so ``memory_search`` hits stay small (``memory_get`` reads the rest).""" + body = (text or "").strip() + return body if len(body) <= _SNIPPET_CHARS else body[: _SNIPPET_CHARS - 1].rstrip() + "…" + + +def _is_recall_echo(content: str) -> bool: + """True when ``content`` is our own recall block coming back as a new event. + + Mirrors ``MemoryRuntime.capture``'s anti-feedback rule; without it a hook that + injects ``memory_recall`` output and then captures the turn via MCP would feed the + injected block back in, and each round would recall (and re-capture) more of it. + """ + return any(marker in content for marker in _RECALL_ECHO_MARKERS) + + +def _attributed(content: str, caller: str) -> str: + """Prefix the sender so the caller id lives inside the text. + + harness-memory's ``AtomCard`` has no user column, so the sender is stamped + by writing ``说：`` into the content itself: it then reaches the atom's + assertion (manual writes) or its raw event (captured events), stays + FTS-searchable, and shows up verbatim on recall. Already-prefixed content is + left untouched so a re-capture cannot double it. + """ + name = (caller or "").strip() + if not name: + return content + prefix = f"{name}说：" + return content if content.startswith(prefix) else prefix + content + + +def _pipeline_hits( + memory: Any, query: str, max_results: int, *, atom_only: bool +) -> list[dict[str, Any]]: + """Run the in-process multi-source search and return its path-carrying hits. + + ``atom_only`` keeps just L2 atoms, so the request is widened first — + otherwise raw hits could crowd the atoms out before the filter runs. + """ + from harness_memory.application.runtime import MemoryRuntime # noqa: PLC0415 + + runtime = MemoryRuntime(memory) + result = runtime.memory_search( + { + "query": query, + "maxResults": max_results * 4 if atom_only else max_results, + "corpus": "memory", + } + ) + hits = list(result.get("hits") or []) + if atom_only: + hits = [hit for hit in hits if hit.get("layer") == "atom"] + return hits[:max_results] + + +def build_memory_mcp(server: OctopServer) -> FastMCP: + """Build the shared memory MCP app (expert bound per request, not per build). + + The expert is selected at request time by ``X-Octop-Agent-Id`` (validated + against the agent repo by ``_AgentRouter``) and carried to the tools via the + ``_current_agent_id`` contextvar. A single app is shared by every expert, so + agents created or disabled after process start are honored immediately — + no process restart is needed to pick up new agents. + """ + mcp = FastMCP( + "octop-memory", + # Stateless streamable HTTP: every request gets a fresh transport, no + # Mcp-Session-Id tracking. Session state is in-memory per process, so a + # server restart silently orphans every client session id and the next + # tool call fails with -32600 "Session not found". Stateless mode + # eliminates that failure class entirely (clients re-initialize per + # request); the cost is one extra initialize per tool call. + stateless_http=True, + # Octop runs behind a reverse proxy (Host is the public domain, forwarded + # by nginx), not a localhost dev scenario — the mcp SDK's localhost + # DNS-rebinding protection does not apply and would reject the Host + # with 421 unless the domain is allow-listed. + transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False), + ) + # Collapse the streamable-HTTP path to "/" so the endpoint is exactly + # /mcp/memory (the default "/mcp" would make it /mcp/memory/mcp). + mcp.settings.streamable_http_path = "/" + + def _agent_id() -> str: + """Agent bound to this request (set by ``_AgentRouter`` from the header).""" + agent_id = _current_agent_id.get() + if not agent_id: + raise RuntimeError("X-Octop-Agent-Id header not bound to this request") + return agent_id + + def _memory() -> Any: + return _open_memory(server, _agent_id()) + + def _caller_user(ctx: Any | None) -> str: + """读取当前 MCP 请求的调用者 user id。 + + 优先级：显式 ``user`` 参数 → ``X-Octop-User-Id`` header（由 + ``_AgentRouter`` 中间件写入 contextvar）。stateless HTTP 下 mcp SDK + 不提供 ``ctx.request_context``，故不依赖它。 + """ + try: + return _current_caller_user.get() or "" + except Exception: # noqa: BLE001 + return "" + + def _derive_session(source: str, user: str) -> str: + """外部调用缺省 session_id 时派生稳定会话键。 + + 规则 ``ext:{source}:{user}``：同 source 同 user 的多次 capture 落入 + 同一分组，harness 提取管线能聚合蒸馏成 atom；不同 source / 不同 user + 分开分组，避免混入彼此上下文。 + """ + return f"ext:{source or 'mcp'}:{user or 'anon'}" + + @mcp.tool() + def memory_recall( + query: str, + limit: int = 5, + session_id: str | None = None, + thread_id: str | None = None, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """**召回专家记忆（读入口首选）**：把与 query 相关的记忆召回进上下文。 + + 每次对话/任务开始前先调一次。运行完整召回管线（分词 → 路由 → FTS → + 重排 → 去重 → token 预算），返回结构化片段 + 可直接注入 system prompt 的 markdown。 + 自动注入（hook）场景建议传 ``session_id``：管线会据此把**本会话**的 raw 排除， + 避免"注入 → 被记录 → 下轮又召回"的回声。 + + 三个读工具怎么选： + - 只想把相关背景拉进上下文 → 用本工具（一次调用，``rendered`` 直接可注入）。 + - 要**定位某条具体记忆并读全文** → ``memory_search`` 拿 ``path``， + 再 ``memory_get(path)`` 读完整 markdown（支持分页）。 + - 要**原话/证据**，或 ``memory_capture`` 刚写入、还没晋升成原子的内容 → + ``memory_raws``（L0 全文检索，capture 后立即可见）。 + + 覆盖范围（与内置 ``memory_search`` 同一套管线）：L2 原子优先，``page`` + 主题页标题会并入 atom 命中；L0 原始事件只做兜底——只要有原子命中，raw 就被 + 整层丢弃。L1 候选不在召回范围内，请用 ``memory_candidates``。 + + Args: + query: 自然语言问题/关键词，整句传入（内部对中文做 n-gram 分词）。 + limit: 最多返回片段数，默认 5。 + session_id: 可选，当前会话 id。用于把本会话的 raw 从召回里排除（防回声）； + 与 ``memory_capture`` 传入的 ``session_id`` 一致才生效。 + thread_id: 可选，会话线程 id。用于共指消解（"那个项目" 靠该线程的 + active-entity stack）并把命中写回实体栈；不传则只做普通检索。 + user: 可选调用者 id（覆盖 ``X-Octop-User-Id`` 头）。 + ctx: MCP 注入的上下文（读取 ``X-Octop-User-Id`` 头）。 + """ + from harness_memory.pipeline.recall import recall_for_prompt # noqa: PLC0415 + + caller = user or _caller_user(ctx) + memory = _memory() + result = recall_for_prompt( + memory, + query, + thread_id=thread_id, + session_id=session_id, + limit=limit, + ) + return { + "memories": [ + { + "source_id": s.source_id, + "timestamp": s.timestamp_iso, + "layer": s.layer, + "text": s.text, + } + for s in result.snippets + ], + "count": len(result.snippets), + "rendered": result.rendered, + "caller": caller or None, + } + + @mcp.tool() + def memory_search( + query: str, + max_results: int = 5, + corpus: str = "all", + ) -> dict[str, Any]: + """**检索记忆（返回可下钻的 path）**：全文检索记忆，返回带虚拟路径的命中列表。 + + 与 ``memory_recall`` 走同一套召回/重排管线，区别是本工具不渲染 markdown，而是给 + 每条命中一个 ``path``，交给 ``memory_get`` 读全文。要"引用出处/读全文"用 + search + get，只想"把背景拉进上下文"用 ``memory_recall``，要"原话/证据"用 + ``memory_raws``。 + + Args: + query: 自然语言问题/关键词（中文会做 n-gram 分词）。 + max_results: 最多返回命中数，默认 5。 + corpus: 检索范围： + ``all``（默认）/``memory`` = 原子(L2)+原始事件(L0)同一套管线； + ``atom`` = 只要 L2 原子命中； + ``raw`` = 直接走 L0 全文检索，不受"有原子命中就丢 raw"的兜底策略影响， + 适合找刚 ``memory_capture``、还没晋升成原子的内容。 + """ + corpus_value = (corpus or "all").strip().lower() + if corpus_value not in _SEARCH_CORPORA: + raise ValueError( + f"invalid corpus {corpus!r}; expected one of {sorted(_SEARCH_CORPORA)}" + ) + memory = _memory() + if corpus_value == "raw": + from harness_memory.application.path_projection import raw_to_path # noqa: PLC0415 + + hits = [ + { + "path": raw_to_path(event), + "layer": "raw", + "snippet": _snippet(event.content), + "occurred_at": event.timestamp.isoformat(), + "source_id": event.id, + } + for event in memory.search_raw(query, limit=max_results) + ] + else: + hits = _pipeline_hits(memory, query, max_results, atom_only=corpus_value == "atom") + return { + "hits": hits, + "total": len(hits), + "corpus": corpus_value, + "hint": "每条命中自带 path，可交给 memory_get(path) 读全文", + } + + @mcp.tool() + def memory_get( + path: str, + start: int | None = None, + lines: int | None = None, + ) -> dict[str, Any]: + """**读取记忆全文**：把 ``memory_search`` / ``memory_recall`` 命中的虚拟路径解析成 markdown。 + + 支持的路径形态：``atom/.md``（L2 原子）、``page/.md`` + （L3 主题页）、``raw//.md``（L0 原始事件）。长内容用 + ``start`` / ``lines`` 分页（配合返回的 ``total_lines`` / ``truncated``）。 + + Args: + path: 虚拟路径，取自 ``memory_search`` 的 ``hits[].path``。 + start: 可选起始行号（1-based）。 + lines: 可选返回行数。 + """ + from harness_memory.application.runtime import MemoryRuntime # noqa: PLC0415 + + params: dict[str, Any] = {"path": path} + if start is not None: + params["from"] = start + if lines is not None: + params["lines"] = lines + try: + result = MemoryRuntime(_memory()).memory_get(params) + except Exception as exc: # stale / mistyped path from a previous call + return { + "path": path, + "error": f"{exc.__class__.__name__}: {exc}", + "hint": ( + "path 形如 atom/.md / page/.md / " + "raw//.md，取自 memory_search 的 hits[].path" + ), + } + return { + "path": result.get("path", path), + "kind": result.get("kind"), + "content": result.get("excerpt") or "", + "total_lines": result.get("total_lines"), + "from_line": result.get("from_line"), + "to_line": result.get("to_line"), + "truncated": result.get("truncated"), + "metadata": result.get("metadata") or {}, + } + + @mcp.tool() + def memory_save( + content: str, + source: str, + topic: str | None = None, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """**直接保存事实**：把一条已知事实写入原子层（跳过提取，立即可召回）。 + + 用于明确、需长期记住的事实（如用户偏好、项目约定）。日常对话内容请用 + ``memory_capture`` 交给提取管线。来源写入 ``metadata.source``，调用者写入 ``metadata.user``； + 调用者 id 还会自动拼进内容前缀（``说：…``），这样它随原子一起落库、可被检索、 + 召回时直接可见——不要把名字重复写进 ``content``。 + + Args: + content: 要记住的事实。 + source: 谁记录的（如 "coding-agent"），用于追溯。 + topic: 可选主题标签。 + user: 可选调用者 id（覆盖 ``X-Octop-User-Id`` 头）。 + ctx: MCP 注入的上下文（读取 ``X-Octop-User-Id`` 头）。 + """ + caller = user or _caller_user(ctx) + memory = _memory() + node = memory.store( + _attributed(content, caller), + topic=topic, + metadata={"source": source, **({"user": caller} if caller else {})}, + ) + return { + "node_id": node.id, + "content": node.content, + "source": source, + "user": caller or None, + } + + @mcp.tool() + def memory_capture( + content: str, + source: str, + session_id: str | None = None, + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """**记录原始事件**：把一条原始内容写入 L0，交给提取流水线。 + + 日常使用入口：记录对话/事件，经 提取 → 候选 → 晋升 → 原子 成为记忆。 + 记录后立即可用 ``memory_raws`` 查询，晋升后才可被 ``memory_recall`` 召回。 + 调用者 id（``X-Octop-User-Id``）会自动拼成内容前缀 ``说：…``：发送者由此进入 + 原子正文，既能被 FTS 直接搜到，也能在召回时一眼看出是谁说的——**不要**自己再 + 写一遍名字。 + + 回声保护：内容里带 ``memory_recall`` 注入标记（``[memory] Earlier in this + workspace`` / ``## Memory Recall``）时**不写入**，返回 ``skipped=recall_echo``。 + 拼进 prompt 的召回块被整轮回采会形成"注入 → 采集 → 再召回"的放大环，故直接丢弃。 + + Args: + content: 原始对话/事件内容（不含发送者前缀）。 + source: 谁记录的，用于追溯。 + session_id: 可选会话 id，用于提取分组；缺省派生为 ``ext:{source}:{user}``。 + 传了之后，``memory_recall`` 用同一个 id 就能把本会话的 raw 排除（防回声）。 + user: 可选调用者 id（覆盖 ``X-Octop-User-Id`` 头）。 + ctx: MCP 注入的上下文（读取 ``X-Octop-User-Id`` 头）。 + """ + caller = user or _caller_user(ctx) + if _is_recall_echo(content): + logger.info("memory_capture: dropped recall echo for agent %s", _agent_id()) + return { + "recorded": False, + "skipped": "recall_echo", + "reason": ( + "content contains a recall injection marker " + f"({_RECALL_ECHO_MARKERS[1]!r} / {_RECALL_ECHO_MARKERS[0]!r}); " + "dropped so our own recall output is not captured as a new event" + ), + "user": caller or None, + } + effective_session = session_id or _derive_session(source, caller) + memory = _memory() + stored_content = _attributed(content, caller) + + # Idempotent capture: skip if an identical raw event (same session + + # content) already exists, so re-ingesting the same conversation does + # not duplicate L0 events. Keeps downstream extraction re-runnable. + try: + for ev in memory.list_raw(session_id=effective_session, limit=1000): + if getattr(ev, "content", None) == stored_content: + return { + "event_id": ev.id, + "content": ev.content, + "source": source, + "user": caller or None, + "session_id": effective_session, + "recorded": True, + "duplicate": True, + "note": ("raw (L0) event already present; skipped (idempotent capture)"), + } + except Exception: # noqa: BLE001 + # If duplicate detection fails, fall back to recording (safe). + pass + + raw = memory.add_raw( + stored_content, + event_type="manual", + host="mcp-external", + session_id=effective_session, + user=caller or None, + payload={"source": source}, + ) + extract_scheduled = _trigger_extract(server, effective_session) + return { + "event_id": raw.id, + "content": raw.content, + "source": source, + "user": caller or None, + "session_id": effective_session, + "recorded": True, + "extract_scheduled": extract_scheduled, + "note": ( + "raw (L0) event recorded; visible now via memory_raws, " + "recallable via memory_recall after the extraction pipeline " + "promotes it to an atom" + ), + } + + @mcp.tool() + def memory_update( + atom_id: str, + new_content: str, + source: str, + note: str = "mcp update", + user: str | None = None, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """**更新记忆**：废弃旧原子并写入新事实。 + + 用于旧记忆已过时、需替换的场景（如纠正事实）。旧原子标记 deprecated， + 新事实立即可被 ``memory_recall`` 召回，带 ``supersedes`` 关联。 + 与 ``memory_save`` 一样，调用者 id 会自动拼进内容前缀（``说：…``）。 + + Args: + atom_id: 要废弃的旧原子 id。 + new_content: 替代的新事实（不含发送者前缀）。 + source: 谁更新的，用于追溯。 + note: 废弃说明，默认 "mcp update"。 + user: 可选调用者 id（覆盖 ``X-Octop-User-Id`` 头）。 + ctx: MCP 注入的上下文（读取 ``X-Octop-User-Id`` 头）。 + """ + caller = user or _caller_user(ctx) + memory = _memory() + deprecated = memory.deprecate_atom(atom_id, actor="user", note=note) + node = memory.store( + _attributed(new_content, caller), + metadata={ + "source": source, + "supersedes": atom_id, + **({"user": caller} if caller else {}), + }, + ) + return { + "deprecated": deprecated, + "deprecated_atom_id": atom_id, + "new_node_id": node.id, + "source": source, + "user": caller or None, + } + + # ─── 记忆分层查询 + 流水线调度（L0/L1/L2）───────────────────────── + # 参考 DSH 记忆工具集（memory_raws/candidates/atoms/extract/promote/reject）， + # 把记忆生产流水线的每个环节暴露为 MCP 工具，供外部调度。 + + @mcp.tool() + def memory_raws( + query: str | None = None, + session_id: str | None = None, + host: str | None = None, + user: str | None = None, + limit: int = 50, + ctx: Context | None = None, # type: ignore[type-arg] + ) -> dict[str, Any]: + """**查原始事件**：FTS 搜索或结构化过滤 L0 原始事件（证据源）。 + + ``query`` 走全文搜索（capture 后立即可见），``session_id``/``host``/``user`` + 做结构化过滤，按时间倒序返回。 + + Args: + query: FTS 关键词（可选）。 + session_id: 按会话过滤（如 ``ext:review-bot:user-alice``）。 + host: 按记录主机过滤（如 ``mcp-external``）。 + user: 按调用者过滤。 + limit: 最多返回条数，默认 50。 + ctx: MCP 注入的上下文。 + """ + caller = user or _caller_user(ctx) + memory = _memory() + events = ( + memory.search_raw(query, limit=limit) + if query + else memory.list_raw( + session_id=session_id, + host=host, + user=user or (caller or None), + limit=limit, + ) + ) + return { + "events": [ + { + "event_id": e.id, + "timestamp": e.timestamp.isoformat(), + "session_id": e.session_id, + "user": e.user, + "event_type": e.event_type, + "source": (e.payload or {}).get("source") if e.payload else None, + "content": e.content, + } + for e in events + ], + "count": len(events), + "caller": caller or None, + } + + @mcp.tool() + def memory_candidates( + status: str | None = None, + session_id: str | None = None, + limit: int = 50, + ) -> dict[str, Any]: + """**查候选记忆**：列出 L1 候选（默认 pending 队列）。 + + 候选由 ``memory_capture``/``memory_extract`` 生成。可用 ``status`` 过滤 + （pending / promoted / rejected / needs_review / conflict）。 + + Args: + status: 候选状态过滤，默认 pending。 + session_id: 按来源会话过滤。 + limit: 最多返回条数，默认 50。 + """ + memory = _memory() + from harness_memory.core import CandidateStatus # noqa: PLC0415 + + # ``CandidateStatus`` is a ``typing.Literal`` (not an Enum), so it cannot + # be instantiated: ``CandidateStatus(status)`` raises ``TypeError``. The + # backend stores status as a plain string, so validate the caller's value + # against the literal and pass the string straight through. + status_value: str | None = None + if status: + allowed = set(get_args(CandidateStatus)) + if status not in allowed: + raise ValueError( + f"invalid candidate status {status!r}; expected one of {sorted(allowed)}" + ) + status_value = status + candidates = memory.list_candidates( + status=status_value, + session_id=session_id, + limit=limit, + ) + return { + "candidates": [ + { + "candidate_id": c.id, + "status": c.status.value if hasattr(c.status, "value") else str(c.status), + "candidate_type": getattr(c, "candidate_type", None), + "assertion": getattr(c, "assertion", None), + "session_id": getattr(c, "session_id", None), + "confidence": getattr(c, "confidence", None), + } + for c in candidates + ], + "count": len(candidates), + } + + @mcp.tool() + def memory_extract( + session_id: str | None = None, + limit: int = 100, + promote: bool = False, + ) -> dict[str, Any]: + """**手动触发提取**：把最近 L0 原始事件提取为候选（可选直达原子）。 + + 取最近 ``limit`` 条 L0 事件 → LLM 类型化提取 → 候选（pending）； + ``promote=True`` 时对候选执行晋升检查（L1 → L2），跳过人工审核。 + + Args: + session_id: 仅提取该会话的事件；缺省提取最近全部。 + limit: 提取的最近原始事件数，默认 100。 + promote: 是否对候选直接晋升，默认 False。 + """ + runtime_server = server.app_runtime + assert runtime_server is not None, "app_runtime required for memory extract" + agent = runtime_server.agent_registry.get_agent(_agent_id()) + runtime = getattr(agent, "_memory_runtime", None) + service = getattr(runtime, "service", None) if runtime else None + if service is None: + return {"error": "MemoryService unavailable (agent not running / no memory runtime)"} + + eff_session = session_id or "manual" + result = service.extract(eff_session, incremental=True, promote=promote, regen_pages=False) + if not isinstance(result, dict): + return {"session_id": eff_session, "candidates": 0, "promoted": 0} + return { + "session_id": eff_session, + "events_considered": result.get("events_considered", 0), + "candidates": result.get("candidates", 0), + "promoted": result.get("promoted", 0) + if isinstance(result.get("promotion"), dict) + else 0, + "error": result.get("failure_reason"), + } + + @mcp.tool() + def memory_promote( + candidate_ids: list[str], + importance: str | None = None, + ) -> dict[str, Any]: + """**审核晋升候选**：把 L1 候选晋升为 L2 原子记忆。 + + 对指定候选执行 5 项晋升检查（规则路径），通过则写入原子并记录 journal。 + + Args: + candidate_ids: 要晋升的候选 id 列表（来自 ``memory_candidates``）。 + importance: 覆盖重要性（low/medium/high），默认保留。 + """ + memory = _memory() + candidates = memory.list_candidates(limit=1000) + by_id = {c.id: c for c in candidates} + selected = [by_id[cid] for cid in candidate_ids if cid in by_id] + if not selected: + return {"promoted": 0, "skipped": len(candidate_ids)} + result = memory.promote_candidates(selected) + return { + "promoted": result.promoted if hasattr(result, "promoted") else len(selected), + "skipped": len(candidate_ids) - len(selected), + } + + @mcp.tool() + def memory_reject( + candidate_id: str, + reason: str = "rejected by external caller", + ) -> dict[str, Any]: + """**拒绝候选**：标记候选为 rejected 并写入原因（不进原子层，可审计）。 + + Args: + candidate_id: 候选 id。 + reason: 拒绝原因，默认 "rejected by external caller"。 + """ + memory = _memory() + + # ``CandidateStatus`` is a ``typing.Literal``, not an Enum, so + # ``CandidateStatus.REJECTED`` does not exist. Pass the literal string. + ok = memory.update_candidate_status( + candidate_id, + status="rejected", + decided_by="mcp-external", + promotion_reason=reason, + ) + return {"ok": ok, "candidate_id": candidate_id, "status": "rejected"} + + return mcp + + +def _trigger_extract(server: OctopServer, session_id: str | None) -> bool: + """Best-effort: asynchronously trigger the agent's memory extraction. + + Raw events written by MCP capture are not in the harness-agent extractor's + tracked sessions, so they would never be distilled into atoms. Reuse the + agent's in-process ``MemoryService`` (with the agent's configured extraction + LLM) via ``agent._memory_runtime.service`` (no public entrypoint; + best-effort). Returns whether an extract task was scheduled. + """ + import asyncio + + agent_id = _current_agent_id.get() + if not session_id or not agent_id: + return False + try: + runtime_server = server.app_runtime + assert runtime_server is not None, "app_runtime required for memory extract" + agent = runtime_server.agent_registry.get_agent(agent_id) + runtime = getattr(agent, "_memory_runtime", None) + service = getattr(runtime, "service", None) if runtime else None + if service is None: + return False + + async def _extract() -> None: + try: + await asyncio.to_thread( + service.extract, + session_id, + incremental=True, + promote=True, + regen_pages=True, + ) + except Exception: + logger.warning("memory extract failed for session %s", session_id, exc_info=True) + + asyncio.create_task(_extract()) + return True + except Exception: + logger.debug("memory extract trigger skipped for agent %s", agent_id, exc_info=True) + return False + + +def _memory_mcp_token() -> str | None: + """Read the MCP auth token (empty string treated as unconfigured).""" + return (os.environ.get("OCTOP_MEMORY_MCP_TOKEN") or "").strip() or None + + +class _TokenAuthMiddleware: + """ASGI middleware enforcing ``Authorization: Bearer`` or ``X-Octop-Memory-Token``.""" + + def __init__(self, app: Any, token: str) -> None: + self._app = app + self._token = token + + async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: + if scope.get("type") != "http": + await self._app(scope, receive, send) + return + + headers = { + k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", []) + } + auth = headers.get("authorization", "") + provided = auth[7:].strip() if auth.startswith("Bearer ") else "" + if not provided: + provided = headers.get("x-octop-memory-token", "").strip() + + if provided != self._token: + body = b'{"error":"unauthorized"}' + await send( + { + "type": "http.response.start", + "status": 401, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode()), + ], + } + ) + await send({"type": "http.response.body", "body": body}) + return + + await self._app(scope, receive, send) + + +class _AgentRouter: + """ASGI dispatcher validating ``X-Octop-Agent-Id`` against the agent repo and + forwarding to the single shared memory MCP app. + + The agent set is NOT snapshotted at startup: every request is checked against + the agent repo (existence + ``enabled``), so agents created or disabled after + process start take effect immediately (no restart required). + """ + + def __init__(self, app: Any, server: OctopServer) -> None: + self._app = app + self._server = server + + async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: + if scope.get("type") != "http": + return # lifespan is wired into the host FastAPI manually; http only here + + headers = { + k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope.get("headers", []) + } + agent_id = headers.get("x-octop-agent-id", "").strip() + services = self._server.services + row = services.agent_repo.get(agent_id) if (services is not None and agent_id) else None + if row is None or not row.enabled: + body = b'{"error":"missing or unknown agent_id (X-Octop-Agent-Id)"}' + await send( + { + "type": "http.response.start", + "status": 404, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode()), + ], + } + ) + await send({"type": "http.response.body", "body": body}) + return + # 把本次请求绑定的 expert 与调用者 user id 写入 contextvar，供工具读取 + # （stateless HTTP 下 mcp SDK 不提供 ctx.request_context）。 + agent_cv = _current_agent_id.set(agent_id) + user = headers.get("x-octop-user-id", "").strip() + token_cv = _current_caller_user.set(user) + try: + await self._app(scope, receive, send) + finally: + _current_caller_user.reset(token_cv) + _current_agent_id.reset(agent_cv) + + +def mount_memory_mcp(app: Any, server: OctopServer) -> list[Any]: + """Mount the memory MCP endpoint at ``/mcp/memory``; the expert is selected + per request via the ``X-Octop-Agent-Id`` header (validated at request time; + the URL stays uniform and does not leak expert ids). + + Does not mount when ``OCTOP_MEMORY_MCP_TOKEN`` is unset (fail-closed). + Returns the session managers that must be initialized in the host FastAPI + lifespan (``streamable_http_app`` task groups depend on it). + """ + token = _memory_mcp_token() + if token is None: + return [] + + services = server.services + assert services is not None, "server.services required for memory MCP mount" + mcp = build_memory_mcp(server) + # IMPORTANT: _session_manager is created lazily by streamable_http_app(). + # Read it only AFTER building the ASGI app and drop None entries — in + # stateless HTTP mode there is no session manager to keep alive, and reading + # mcp._session_manager before streamable_http_app() yields None, which then + # crashes the host FastAPI lifespan with "NoneType has no attribute 'run'". + streamable_app = mcp.streamable_http_app() + managers = [mgr for mgr in (mcp._session_manager,) if mgr is not None] + + app.mount( + "/mcp/memory", + _TokenAuthMiddleware(_AgentRouter(streamable_app, server), token), + ) + return managers + + +__all__ = ["build_memory_mcp", "mount_memory_mcp"] diff --git a/src/octop/infra/agents/memory_recall_patch.py b/src/octop/infra/agents/memory_recall_patch.py new file mode 100644 index 00000000..5e4cf144 --- /dev/null +++ b/src/octop/infra/agents/memory_recall_patch.py @@ -0,0 +1,356 @@ +"""Persistent recall-quality patches for ``harness_memory``. + +The upstream ``harness-memory`` package (PyPI) ships a recall pipeline whose +retrieval quality is poor for multi-person shared-conversation datasets like +the evaluation corpus. Three defects were found during eval work and are +patched here at startup (idempotent), so the fixes survive +``uv sync`` / redeploys instead of living only in the venv: + + 1. ``router.route`` — entity hints are matched against stored aliases with + an exact string lookup, but natural-language queries attach Chinese + suffixes ("张小明的deadline" → hint "张小明的", stored alias "张小明"). + Result: no entity anchor is resolved, recall degrades to topical FTS + and cross-entity topics (everyone's deadlines) drown the requested + person's. Fix: progressively strip common suffixes before alias lookup. + + 2. ``multi_source._per_token_atom_search`` — merges per-token FTS hits + ordered by token position, so an early generic token ("李小" n-gram of + "李小婉") outranks a later *relevant* token ("recurring"). Result: + recall returns the entity's generic facts (age / company) instead of + the topical memory. Fix: rank by count of matched *strong* tokens + (full Latin words + complete Han runs + resolved anchor names), with + n-grams contributing to recall but not to the relevance count. + + 3. ``multi_source._gather_atoms`` anchor branch — when the router resolves + an entity, its "recent atoms" fill the candidate list before FTS + topical hits, so a query-relevant atom outside the recency window is + truncated before rerank. Fix: run FTS first, then anchor atoms as + fallback, and pass anchor entity names into the strong-token set so + "张小明 + deadline" outranks someone else's "deadline". + +Each patch wraps the original symbol (preserving the upstream signature and +behaviour as the fallback); ``apply_memory_recall_patch()`` is idempotent +and is invoked from ``octop.infra.server._boot_runtime`` before any agent +memory service is created. +""" + +from __future__ import annotations + +import logging +import re +from collections.abc import Sequence +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from harness_memory.pipeline.recall.router import RoutingDecision + +logger = logging.getLogger(__name__) + +_PATCHED = False + +# ── shared helpers ──────────────────────────────────────────────────────── + +_LATIN_WORD_RE = re.compile(r"[A-Za-z][A-Za-z0-9_-]{1,}") +_HAN_RUN_RE = re.compile(r"[\u4e00-\u9fff]{2,}") + +# Chinese possessive / copular suffixes that attach to a name in queries. +_SUFFIXES = ("的", "是", "在", "了", "与", "和") + + +def _strong_query_tokens(text: str) -> set[str]: + """Full-word query tokens — Latin words + complete Han runs (no n-grams).""" + out: set[str] = set() + for m in _LATIN_WORD_RE.finditer(text): + out.add(m.group(0).lower()) + for m in _HAN_RUN_RE.finditer(text): + out.add(m.group(0)) + return out + + +def _alias_candidates(alias: str) -> tuple[str, ...]: + """Ordered alias lookup candidates: exact match, then suffix-stripped.""" + out = [alias] + if alias and not alias.isascii(): + stripped = alias + for _ in range(3): + if stripped and stripped[-1] in _SUFFIXES: + stripped = stripped[:-1] + if stripped: + out.append(stripped) + else: + break + return tuple(out) + + +# ── 4. substring entity anchoring (cross-entity / conflict phrasing) ───── +# +# ``parser._entity_hints`` produces *fragmented* hints for compound +# natural-language queries: "在多位同事都提供了age的情况下，周晓东本人的age" +# yields hints like "周晓东本人的" (exact alias lookup misses "周晓东"), +# "和赵晓磊同住…王小明" yields one long blob, and "当吴晓强的…" keeps the +# leading "当". Exact-then-suffix-stripped resolution cannot recover the +# entity, so the router resolves nothing and recall degrades to topical FTS. +# Fix: also match any known entity name/alias as a *substring* of the query +# text, so the target person (王小明/周晓东/吴晓强/赵晓磊) is resolved and flows +# into the anchor branch of ``_gather_atoms`` (fixes cross-entity + conflict +# "以哪个值为准" phrasing misses). + +_ENTITY_INDEX: dict[str, dict[str, str]] = {} + + +def _entity_index_key(memory: Any) -> str: + """Key the entity-index cache by memory identity so daily / dev (two + separate agents) never share one index.""" + return str(getattr(memory, "namespace", None) or id(getattr(memory, "_backend", memory))) + + +def _entity_index_for(memory: Any) -> dict[str, str]: + key = _entity_index_key(memory) + if key not in _ENTITY_INDEX: + _ENTITY_INDEX[key] = _build_entity_index(memory) + return _ENTITY_INDEX[key] + + +def _build_entity_index(memory: Any, *, limit: int = 800) -> dict[str, str]: + """Map every entity canonical name + alias -> entity id.""" + idx: dict[str, str] = {} + try: + for e in memory.list_entities(limit=limit): + names = [e.canonical_name, *(getattr(e, "aliases", None) or [])] + for name in names: + name = str(name).strip() + if len(name) >= 2: + idx.setdefault(name, e.id) + # The alias table may hold names not denormalized onto the entity. + for a in memory.list_aliases(limit=limit * 2): + al = str(getattr(a, "alias", "")).strip() + if len(al) >= 2: + idx.setdefault(al, a.entity_id) + except Exception: # noqa: BLE001 + # A bare/mock backend may lack these; fall back to an empty index. + pass + return idx + + +def _substring_entity_hits(text: str, idx: dict[str, str]) -> list[str]: + """Entity ids whose name/alias appears as a substring of ``text``. + + De-duped by entity id, order-preserving. + """ + hits: dict[str, str] = {} + for name, eid in idx.items(): + if len(name) >= 2 and name in text: + hits.setdefault(eid, name) + return list(dict.fromkeys(hits)) + + +# ── 1. router.route: suffix-tolerant entity resolution ──────────────────── + + +def _patched_route( + memory: Any, + parsed: Any, + *, + thread_id: str | None = None, +) -> RoutingDecision: + from harness_memory.pipeline.recall.router import _orig_route # noqa: PLC0415 + + # Re-resolve hints with suffix stripping: try each candidate alias in + # order, resolve the first hit, and inject it back into the parsed + # query so the original route() picks it up. + resolved: list[str] = [] + for alias in parsed.entity_hints or (): + for candidate in _alias_candidates(alias): + try: + entity = memory.find_entity_by_alias(candidate) + except Exception: # noqa: BLE001 + continue + if entity is not None: + resolved.append(entity.id) + break + + # Substring anchoring: the parser's hint fragments (e.g. "当吴晓强的", + # "周晓东本人的", "和赵晓磊同住…王小明") fail exact alias lookup, so the + # router resolves nothing for these compound queries. Match any known + # entity name/alias as a substring of the whole query text to recover the + # target entity. + try: + for eid in _substring_entity_hits(parsed.text, _entity_index_for(memory)): + if eid not in resolved: + resolved.append(eid) + except Exception: # noqa: BLE001 + pass + + decision = _orig_route(memory, parsed, thread_id=thread_id) + # Union our anchors (suffix + substring) with whatever the original route + # resolved; never drop the original resolution. Only rebuild the decision + # when we actually resolved something extra. + union = list(dict.fromkeys(resolved + list(decision.resolved_entity_ids))) + if union and tuple(union) != tuple(decision.resolved_entity_ids): + from dataclasses import replace # noqa: PLC0415 + + decision = replace(decision, resolved_entity_ids=tuple(union)) + return decision + + +# ── 2. multi_source._per_token_atom_search: strong-token ranking ────────── + + +def _patched_per_token_atom_search( + memory: Any, + parsed: Any, + *, + limit: int, + anchor_names: Sequence[str] = (), +) -> list[Any]: + """Per-token FTS merge ranked by matched *strong* token count. + + Mirrors the upstream function but counts only strong tokens (full words + + anchor entity names), so a hit matching "张小明" + "deadline" outranks + a single-token "deadline" hit on a different entity. + """ + strong = _strong_query_tokens(parsed.text) + for name in anchor_names: + norm = str(name).strip() + if len(norm) >= 2: + strong.add(norm.lower() if norm.isascii() else norm) + seen: dict[str, dict[str, Any]] = {} + tokens = parsed.raw_tokens or (parsed.text,) + for token_idx, token in enumerate(tokens): + if not str(token).strip(): + continue + try: + # Search deep: FTS5 rank is a global corpus score, so the + # relevant hit for a common token (e.g. "recurring" shared by + # many entities) can sit far past the top-N. A shallow pool + # truncates it before our strong-token ranking can promote it. + atoms = memory.search_atoms(token, limit=limit * 8) + except Exception: + continue + is_strong = token in strong + for atom_idx, atom in enumerate(atoms): + entry = seen.get(atom.id) + if entry is None: + seen[atom.id] = { + "count": 1 if is_strong else 0, + "first": (token_idx, atom_idx), + "atom": atom, + } + elif is_strong: + entry["count"] += 1 + ranked = sorted( + seen.values(), + key=lambda e: (-e["count"], e["first"][0], e["first"][1]), + ) + return [e["atom"] for e in ranked[:limit]] + + +# ── 3. multi_source._gather_atoms anchor branch: FTS first + anchor names ── + +# item 3: decision/cause/pitfall/experience intent reorder. The corpus stores +# "技术决策" atoms as e.g. "「X」做过一个技术决策：用 G6 而非 D3.js：因为…" and +# "踩坑/经验复用" atoms as "「X」的pitfalls有更新…" / "可复用经验…". For a query +# asking "技术决策及原因 / 为什么 / 经验复用 / 踩坑", those atoms are added as +# *anchor fillers after* the FTS "技术选型" matches and get truncated by +# `limit`. Reorder them ahead of generic selection atoms when the query shows +# that intent. +_DECISION_INTENT_RE = re.compile( + r"技术决策|决策及原因|原因|为什么|若非|为何|经验|复用|踩|坑|规避|如何解决" +) +_DECISION_MARK_RE = re.compile(r"技术决策|而非|因为|坑|pitfall|经验|可复用|规避|决策|N\+1") + + +def _decision_first(atoms: list[Any]) -> list[Any]: + """Stable-partition: decision/cause/pitfall atoms first, others after.""" + dec = [a for a in atoms if _DECISION_MARK_RE.search(getattr(a, "assertion", "") or "")] + rest = [a for a in atoms if not _DECISION_MARK_RE.search(getattr(a, "assertion", "") or "")] + return dec + rest + + +def _patched_gather_atoms( + memory: Any, + parsed: Any, + *, + limit: int, + anchor_ids: Sequence[str] = (), +) -> list[Any]: + """Anchor-narrowed atom gather: FTS topical hits first, anchor fillers + after, with anchor entity names promoted into the strong-token set.""" + + if anchor_ids: + anchor_names: list[str] = [] + for eid in anchor_ids: + ent = memory.get_entity(eid) + if ent is not None and ent.canonical_name: + anchor_names.append(ent.canonical_name) + out: dict[str, Any] = {} + for atom in _patched_per_token_atom_search( + memory, parsed, limit=limit, anchor_names=anchor_names + ): + out.setdefault(atom.id, atom) + if parsed.time_window is not None: + for eid in anchor_ids: + for atom in memory.search_atoms_by_time_range( + start=parsed.time_window.start, + end=parsed.time_window.end, + entity_id=eid, + limit=limit, + ): + out.setdefault(atom.id, atom) + else: + for eid in anchor_ids: + for atom in memory.list_atoms(entity_id=eid, limit=limit): + out.setdefault(atom.id, atom) + atoms = list(out.values()) + # Reorder the WHOLE gathered set (FTS hits + anchor fillers) so the + # decision/cause/pitfall atom is not truncated by `limit` before it + # can be promoted ahead of generic "技术选型" atoms. + if _DECISION_INTENT_RE.search(parsed.text or ""): + atoms = _decision_first(atoms) + return atoms[:limit] + + if parsed.time_window is not None: + return list( + memory.search_atoms_by_time_range( + start=parsed.time_window.start, + end=parsed.time_window.end, + limit=limit, + ) + ) + return _patched_per_token_atom_search(memory, parsed, limit=limit) + + +# ── apply ───────────────────────────────────────────────────────────────── + + +def apply_memory_recall_patch() -> None: + """Install the recall-quality patches (idempotent).""" + global _PATCHED + if _PATCHED: + return + + try: + import harness_memory.pipeline.recall.multi_source as ms # noqa: PLC0415 + import harness_memory.pipeline.recall.router as router # noqa: PLC0415 + + if not hasattr(router, "_orig_route"): + router._orig_route = router.route + router.route = _patched_route + + if not hasattr(ms, "_orig_per_token_atom_search"): + ms._orig_per_token_atom_search = ms._per_token_atom_search + ms._per_token_atom_search = _patched_per_token_atom_search + + if not hasattr(ms, "_orig_gather_atoms"): + ms._orig_gather_atoms = ms._gather_atoms + ms._gather_atoms = _patched_gather_atoms + + _PATCHED = True + logger.info("memory recall patches applied (router suffix + strong-token ranking)") + except Exception: # noqa: BLE001 + # Never fail startup because a recall patch could not be installed — + # upstream behaviour is the safe fallback. + logger.warning( + "memory recall patch installation failed; using upstream recall", + exc_info=True, + ) diff --git a/src/octop/infra/server.py b/src/octop/infra/server.py index 65f43367..eaf45f01 100644 --- a/src/octop/infra/server.py +++ b/src/octop/infra/server.py @@ -361,6 +361,14 @@ async def _boot_runtime(self, config: OctopConfig) -> None: configure_browser_idle_timeout(config.browser_idle_timeout_minutes) + # harness_memory recall 质量补丁（实体后缀解析 + 强 token 排序）： + # 在任意 MemoryService / MCP 记忆工具创建前应用，保证评测与生产召回一致。 + from octop.infra.agents.memory_recall_patch import ( + apply_memory_recall_patch, + ) + + apply_memory_recall_patch() + registry = AgentManager( repos=self.services.repos, paths=self.paths, diff --git a/tests/unit/agents/test_memory_mcp.py b/tests/unit/agents/test_memory_mcp.py new file mode 100644 index 00000000..b9232d37 --- /dev/null +++ b/tests/unit/agents/test_memory_mcp.py @@ -0,0 +1,610 @@ +"""Unit tests for the expert memory MCP server (infra/agents/memory_mcp).""" + +from __future__ import annotations + +from unittest import mock + +import pytest + +from octop.infra.agents import memory_mcp as mm + + +@pytest.fixture +def fake_memory(monkeypatch): + mem = mock.MagicMock() + mem.recall.return_value = [] + node = mock.MagicMock() + node.id = "node1" + node.content = "remember X" + mem.store.return_value = node + + def _add_raw(content, **_kwargs): + raw = mock.MagicMock(id="evt1") + raw.content = content + return raw + + mem.add_raw.side_effect = _add_raw + mem.deprecate_atom.return_value = True + monkeypatch.setattr(mm, "_open_memory", lambda server, agent_id: mem) + return mem + + +@pytest.fixture +def bind_agent(): + """Bind the request-scoped agent contextvar (main resolves the expert per request).""" + token = mm._current_agent_id.set("A1") + yield "A1" + mm._current_agent_id.reset(token) + + +@pytest.fixture +def bind_user(): + """Bind the ``X-Octop-User-Id`` caller contextvar (sender attribution).""" + token = mm._current_caller_user.set("alice") + yield "alice" + mm._current_caller_user.reset(token) + + +def _tools(mcp): + return mcp._tool_manager._tools + + +def test_build_resolves_agent_per_request(monkeypatch): + """Expert is bound per request (contextvar), not captured at build time.""" + captured = {} + + def fake_open(server, agent_id): + captured["agent_id"] = agent_id + mem = mock.MagicMock() + mem.store.return_value = mock.MagicMock(id="n1", content="x") + return mem + + monkeypatch.setattr(mm, "_open_memory", fake_open) + mcp = mm.build_memory_mcp(mock.MagicMock()) + token = mm._current_agent_id.set("EXPERT42") + try: + _tools(mcp)["memory_save"].fn(content="x", source="s") + finally: + mm._current_agent_id.reset(token) + assert captured["agent_id"] == "EXPERT42" + + +def test_build_registers_eleven_tools(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock()) + assert set(_tools(mcp)) == { + "memory_recall", + "memory_search", + "memory_get", + "memory_save", + "memory_capture", + "memory_update", + "memory_raws", + "memory_candidates", + "memory_extract", + "memory_promote", + "memory_reject", + } + + +def test_memory_recall_uses_full_pipeline(fake_memory, monkeypatch, bind_agent): + """memory_recall runs the full recall pipeline (recall_for_prompt).""" + import harness_memory.pipeline.recall as _recall + + class _Snippet: + source_id = "atom-1" + timestamp_iso = "2026-08-19T00:00:00+00:00" + layer = "atom" + text = "billing-migration is the local clone" + + fake_result = mock.MagicMock() + fake_result.snippets = [_Snippet()] + fake_result.rendered = "markdown" + monkeypatch.setattr(_recall, "recall_for_prompt", lambda m, q, **kw: fake_result) + + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_recall"].fn(query="billing-migration", limit=3) + assert result["count"] == 1 + assert result["memories"][0]["text"] == "billing-migration is the local clone" + assert result["rendered"] == "markdown" + + +def test_memory_recall_forwards_session_and_thread(fake_memory, monkeypatch, bind_agent): + """Hook callers can scope recall to a session/thread (echo guard + co-reference).""" + import harness_memory.pipeline.recall as _recall + + captured = {} + + def _fake(memory, query, **kwargs): + captured["memory"] = memory + captured["query"] = query + captured.update(kwargs) + result = mock.MagicMock() + result.snippets = [] + result.rendered = "" + return result + + monkeypatch.setattr(_recall, "recall_for_prompt", _fake) + mcp = mm.build_memory_mcp(mock.MagicMock()) + _tools(mcp)["memory_recall"].fn( + query="那个项目", + limit=4, + session_id="sess-1", + thread_id="thr-1", + ) + assert captured["memory"] is fake_memory + assert captured["query"] == "那个项目" + assert captured["session_id"] == "sess-1" + assert captured["thread_id"] == "thr-1" + assert captured["limit"] == 4 + + +def test_memory_recall_without_scope_passes_none(fake_memory, monkeypatch, bind_agent): + import harness_memory.pipeline.recall as _recall + + captured = {} + + def _fake(memory, query, **kwargs): + captured.update(kwargs) + result = mock.MagicMock() + result.snippets = [] + result.rendered = "" + return result + + monkeypatch.setattr(_recall, "recall_for_prompt", _fake) + mcp = mm.build_memory_mcp(mock.MagicMock()) + _tools(mcp)["memory_recall"].fn(query="q") + assert captured["session_id"] is None + assert captured["thread_id"] is None + + +def test_memory_save_goes_store(fake_memory, bind_agent): + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_save"].fn(content="remember X", source="coding-agent") + kwargs = fake_memory.store.call_args.kwargs + assert kwargs["topic"] is None + assert kwargs["metadata"] == {"source": "coding-agent"} + assert result["source"] == "coding-agent" + + +def test_memory_capture_goes_add_raw(fake_memory, bind_agent): + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_capture"].fn( + content="raw conversation", source="review-bot", session_id="review-1" + ) + kwargs = fake_memory.add_raw.call_args.kwargs + assert kwargs["event_type"] == "manual" + assert kwargs["host"] == "mcp-external" + assert kwargs["session_id"] == "review-1" + assert kwargs["payload"] == {"source": "review-bot"} + assert result["recorded"] is True + assert "raw (L0)" in result["note"] + + +def test_memory_capture_is_idempotent(fake_memory, bind_agent): + """Re-capturing the same session + content reuses the existing L0 event.""" + existing = mock.MagicMock(id="evt-existing") + existing.content = "raw conversation" + fake_memory.list_raw.return_value = [existing] + + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_capture"].fn( + content="raw conversation", source="review-bot", session_id="review-1" + ) + + fake_memory.add_raw.assert_not_called() + assert result["event_id"] == "evt-existing" + assert result["duplicate"] is True + assert "idempotent capture" in result["note"] + + +def test_memory_raws_queries_l0_with_query(fake_memory, bind_agent): + class _Evt: + id = "evt1" + timestamp = __import__("datetime").datetime(2026, 8, 19) + session_id = "review-1" + user = "u1" + event_type = "manual" + payload = {"source": "review-bot"} + content = "report panel banner hidden" + + fake_memory.search_raw.return_value = [_Evt()] + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_raws"].fn(query="report panel banner", limit=5) + fake_memory.search_raw.assert_called_once_with("report panel banner", limit=5) + assert result["count"] == 1 + assert result["events"][0]["event_id"] == "evt1" + assert result["events"][0]["source"] == "review-bot" + + +def test_memory_update_deprecates_and_saves(fake_memory, bind_agent): + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_update"].fn( + atom_id="atom1", new_content="new fact", source="review-bot" + ) + fake_memory.deprecate_atom.assert_called_once_with("atom1", actor="user", note="mcp update") + assert fake_memory.store.call_args.kwargs["metadata"] == { + "source": "review-bot", + "supersedes": "atom1", + } + assert result["deprecated"] is True + + +def _asgi_scope(headers: list[tuple[bytes, bytes]] | None = None) -> dict: + return {"type": "http", "headers": headers or []} + + +@pytest.mark.asyncio +async def test_token_middleware_rejects_bad_token(): + inner_called = False + + async def _inner(scope, receive, send): + nonlocal inner_called + inner_called = True + + mw = mm._TokenAuthMiddleware(_inner, "secret") + sent = [] + scope = _asgi_scope([(b"authorization", b"Bearer wrong")]) + + async def _send(msg): + sent.append(msg) + + await mw(scope, lambda: {}, _send) + assert inner_called is False + assert sent[0]["status"] == 401 + + +@pytest.mark.asyncio +async def test_token_middleware_accepts_bearer(): + inner_called = False + + async def _inner(scope, receive, send): + nonlocal inner_called + inner_called = True + + mw = mm._TokenAuthMiddleware(_inner, "secret") + scope = _asgi_scope([(b"authorization", b"Bearer secret")]) + await mw(scope, lambda: {}, lambda msg: None) + assert inner_called is True + + +def test_mount_fail_closed_without_token(monkeypatch): + monkeypatch.delenv("OCTOP_MEMORY_MCP_TOKEN", raising=False) + app = mock.MagicMock() + assert mm.mount_memory_mcp(app, mock.MagicMock()) == [] + app.mount.assert_not_called() + + +def test_mount_unified_path_with_shared_app(monkeypatch): + """A single shared MCP app is mounted once at /mcp/memory.""" + from types import SimpleNamespace + + monkeypatch.setenv("OCTOP_MEMORY_MCP_TOKEN", "secret") + app = mock.MagicMock() + server = SimpleNamespace(services=SimpleNamespace(agent_repo=mock.MagicMock())) + managers = mm.mount_memory_mcp(app, server) + assert len(managers) == 1 + app.mount.assert_called_once() + assert app.mount.call_args.args[0] == "/mcp/memory" + + +@pytest.mark.asyncio +async def test_agent_router_routes_by_header(): + """_AgentRouter validates the agent, binds the contextvar, forwards to the shared app.""" + from types import SimpleNamespace + + seen = {} + + class _FakeApp: + async def __call__(self, scope, receive, send): + seen["agent"] = mm._current_agent_id.get() + + rows = { + "A1": SimpleNamespace(agent_id="A1", enabled=True), + "A2": SimpleNamespace(agent_id="A2", enabled=True), + } + server = SimpleNamespace( + services=SimpleNamespace(agent_repo=mock.MagicMock(get=lambda aid: rows.get(aid))) + ) + router = mm._AgentRouter(_FakeApp(), server) + scope = {"type": "http", "headers": [(b"x-octop-agent-id", b"A2")]} + await router(scope, lambda: {}, lambda msg: None) + assert seen["agent"] == "A2" + + +@pytest.mark.asyncio +async def test_agent_router_404_unknown_agent(): + """Unknown agent_id returns 404 before dispatching.""" + from types import SimpleNamespace + + server = SimpleNamespace( + services=SimpleNamespace(agent_repo=mock.MagicMock(get=lambda aid: None)) + ) + router = mm._AgentRouter(mock.MagicMock(), server) + scope = {"type": "http", "headers": [(b"x-octop-agent-id", b"NOPE")]} + sent = [] + + async def _send(msg): + sent.append(msg) + + await router(scope, lambda: {}, _send) + assert sent[0]["status"] == 404 + + +def test_trigger_extract_no_session_returns_false(): + """No session_id -> no extraction trigger.""" + assert mm._trigger_extract(mock.MagicMock(), None) is False + + +def test_trigger_extract_no_service_returns_false(): + """Agent without memory runtime/service -> silently skipped.""" + agent = mock.MagicMock() + runtime = mock.MagicMock() + runtime.service = None + agent._memory_runtime = runtime + registry = mock.MagicMock(get_agent=lambda aid: agent) + server = mock.MagicMock() + server.app_runtime.agent_registry = registry + token = mm._current_agent_id.set("A1") + try: + assert mm._trigger_extract(server, "kiro-chat") is False + finally: + mm._current_agent_id.reset(token) + + +def test_trigger_extract_schedules_service(monkeypatch): + """With a service, asynchronously schedule extract and return True.""" + import asyncio + import time + + agent = mock.MagicMock() + service = mock.MagicMock() + runtime = mock.MagicMock() + runtime.service = service + agent._memory_runtime = runtime + registry = mock.MagicMock(get_agent=lambda aid: agent) + server = mock.MagicMock() + server.app_runtime.agent_registry = registry + + async def _run(): + return mm._trigger_extract(server, "kiro-chat") + + token = mm._current_agent_id.set("A1") + try: + assert asyncio.run(_run()) is True + finally: + mm._current_agent_id.reset(token) + time.sleep(0.1) + service.extract.assert_called() + assert service.extract.call_args.args[0] == "kiro-chat" + + +def _with_agent(agent_id: str): + """Bind the request-scoped agent contextvar for a single tool call.""" + return mm._current_agent_id.set(agent_id) + + +def test_memory_candidates_passes_status_string(fake_memory): + """``status`` is a typing.Literal of strings: pass the raw value, don't + instantiate it (previously crashed with "Cannot instantiate typing.Literal").""" + fake_memory.list_candidates.return_value = [] + mcp = mm.build_memory_mcp(mock.MagicMock()) + token = _with_agent("A1") + try: + result = _tools(mcp)["memory_candidates"].fn(status="pending", limit=5) + finally: + mm._current_agent_id.reset(token) + assert result["candidates"] == [] + assert fake_memory.list_candidates.call_args.kwargs["status"] == "pending" + + +def test_memory_candidates_not_a_status_raises(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock()) + token = _with_agent("A1") + try: + with pytest.raises(ValueError): + _tools(mcp)["memory_candidates"].fn(status="no-such-status") + finally: + mm._current_agent_id.reset(token) + + +def test_memory_reject_passes_literal_status(fake_memory): + mcp = mm.build_memory_mcp(mock.MagicMock()) + token = _with_agent("A1") + try: + result = _tools(mcp)["memory_reject"].fn(candidate_id="c1", reason="dup") + finally: + mm._current_agent_id.reset(token) + assert result["status"] == "rejected" + assert fake_memory.update_candidate_status.call_args.kwargs["status"] == "rejected" + + +def _stub_runtime(monkeypatch, *, hits=None, get_result=None, captured=None): + """Patch ``MemoryRuntime`` so search/get run against a fake runtime.""" + + class _FakeRuntime: + def __init__(self, memory): + if captured is not None: + captured["memory"] = memory + + def memory_search(self, params): + if captured is not None: + captured["search_params"] = params + return {"hits": list(hits or []), "total": len(hits or []), "empty_reason": None} + + def memory_get(self, params): + if captured is not None: + captured["get_params"] = params + return dict(get_result or {}) + + import harness_memory.application.runtime as _runtime + + monkeypatch.setattr(_runtime, "MemoryRuntime", _FakeRuntime) + + +def test_memory_search_projects_hit_paths(fake_memory, monkeypatch, bind_agent): + """memory_search runs the shared pipeline and returns path-carrying hits.""" + captured = {} + _stub_runtime( + monkeypatch, + hits=[ + {"path": "atom/a1.md", "layer": "atom", "snippet": "s1", "source_id": "a1"}, + {"path": "raw/2026-08-19/r1.md", "layer": "raw", "snippet": "s2", "source_id": "r1"}, + ], + captured=captured, + ) + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_search"].fn(query="billing", max_results=5) + assert captured["memory"] is fake_memory + assert captured["search_params"] == {"query": "billing", "maxResults": 5, "corpus": "memory"} + assert [hit["path"] for hit in result["hits"]] == ["atom/a1.md", "raw/2026-08-19/r1.md"] + assert result["total"] == 2 + assert result["corpus"] == "all" + + +def test_memory_search_atom_corpus_widens_then_filters(fake_memory, monkeypatch, bind_agent): + """``corpus=atom`` asks for a wider pool, then keeps only L2 atoms.""" + captured = {} + _stub_runtime( + monkeypatch, + hits=[ + {"path": "raw/2026-08-19/r1.md", "layer": "raw", "snippet": "s2", "source_id": "r1"}, + {"path": "atom/a1.md", "layer": "atom", "snippet": "s1", "source_id": "a1"}, + {"path": "atom/a2.md", "layer": "atom", "snippet": "s3", "source_id": "a2"}, + ], + captured=captured, + ) + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_search"].fn(query="billing", max_results=2, corpus="atom") + assert captured["search_params"]["maxResults"] == 8 + assert [hit["path"] for hit in result["hits"]] == ["atom/a1.md", "atom/a2.md"] + assert result["corpus"] == "atom" + + +def test_memory_search_raw_corpus_uses_l0_fts(fake_memory, bind_agent): + """``corpus=raw`` bypasses the atom-first fallback and searches L0 directly.""" + from datetime import datetime + + event = mock.MagicMock() + event.id = "r1" + event.content = "nginx 需要 proxy /api/memory-mcp" + event.timestamp = datetime(2026, 8, 19, 12, 0) + fake_memory.search_raw.return_value = [event] + + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_search"].fn(query="proxy", max_results=3, corpus="raw") + fake_memory.search_raw.assert_called_once_with("proxy", limit=3) + assert result["hits"][0]["path"] == "raw/2026-08-19/r1.md" + assert result["hits"][0]["layer"] == "raw" + + +def test_memory_search_rejects_unknown_corpus(fake_memory, bind_agent): + mcp = mm.build_memory_mcp(mock.MagicMock()) + with pytest.raises(ValueError): + _tools(mcp)["memory_search"].fn(query="billing", corpus="wiki") + + +def test_memory_get_returns_excerpt_as_content(fake_memory, monkeypatch, bind_agent): + """memory_get surfaces the excerpt under ``content`` and forwards paging.""" + captured = {} + _stub_runtime( + monkeypatch, + get_result={ + "path": "atom/a1.md", + "kind": "atom", + "excerpt": "# body", + "metadata": {"id": "a1"}, + "total_lines": 3, + "from_line": 1, + "to_line": 3, + "truncated": False, + }, + captured=captured, + ) + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_get"].fn(path="atom/a1.md", start=1, lines=10) + assert captured["get_params"] == {"path": "atom/a1.md", "from": 1, "lines": 10} + assert result["content"] == "# body" + assert result["kind"] == "atom" + assert result["total_lines"] == 3 + + +def test_memory_get_returns_hint_for_bad_path(fake_memory, monkeypatch, bind_agent): + """A stale/mistyped path degrades to an error payload instead of raising.""" + + class _Boom: + def __init__(self, memory): + pass + + def memory_get(self, params): + raise ValueError("path must be non-empty and unpadded: 'nope'") + + import harness_memory.application.runtime as _runtime + + monkeypatch.setattr(_runtime, "MemoryRuntime", _Boom) + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_get"].fn(path="nope") + assert "path must be non-empty" in result["error"] + assert "atom/.md" in result["hint"] + + +def test_memory_capture_prefixes_sender(fake_memory, bind_agent, bind_user): + """The header user id is folded into the captured text, not just the payload.""" + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_capture"].fn(content="接口先不要动", source="review-bot") + assert fake_memory.add_raw.call_args.args[0] == "alice说：接口先不要动" + assert fake_memory.add_raw.call_args.kwargs["user"] == "alice" + assert result["content"] == "alice说：接口先不要动" + assert result["user"] == "alice" + + +def test_memory_capture_does_not_double_prefix(fake_memory, bind_agent, bind_user): + mcp = mm.build_memory_mcp(mock.MagicMock()) + _tools(mcp)["memory_capture"].fn(content="alice说：接口先不要动", source="review-bot") + assert fake_memory.add_raw.call_args.args[0] == "alice说：接口先不要动" + + +def test_memory_save_prefixes_sender(fake_memory, bind_agent, bind_user): + mcp = mm.build_memory_mcp(mock.MagicMock()) + _tools(mcp)["memory_save"].fn(content="部署约定：端点挂在 /mcp/memory", source="coding-agent") + assert fake_memory.store.call_args.args[0] == "alice说：部署约定：端点挂在 /mcp/memory" + assert fake_memory.store.call_args.kwargs["metadata"]["user"] == "alice" + + +def test_memory_update_prefixes_sender(fake_memory, bind_agent, bind_user): + mcp = mm.build_memory_mcp(mock.MagicMock()) + _tools(mcp)["memory_update"].fn( + atom_id="a1", new_content="端点改到 /api/memory-mcp", source="s" + ) + assert fake_memory.store.call_args.args[0] == "alice说：端点改到 /api/memory-mcp" + + +def test_write_without_caller_keeps_content(fake_memory, bind_agent): + """No ``X-Octop-User-Id`` -> no attribution prefix.""" + mcp = mm.build_memory_mcp(mock.MagicMock()) + _tools(mcp)["memory_save"].fn(content="no sender", source="coding-agent") + assert fake_memory.store.call_args.args[0] == "no sender" + + +@pytest.mark.parametrize( + "content", + [ + "[memory] Earlier in this workspace, related to your question:\n- [atom] x", + "结论见下：\n## Memory Recall\n- [atom] y\n[/memory]", + ], +) +def test_memory_capture_drops_recall_echo(fake_memory, bind_agent, content): + """Injected recall blocks are not captured back (mirrors skip_memory_echo).""" + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_capture"].fn(content=content, source="hook") + assert result["recorded"] is False + assert result["skipped"] == "recall_echo" + assert "recall_echo" in result["skipped"] + fake_memory.add_raw.assert_not_called() + + +def test_memory_capture_still_records_normal_content(fake_memory, bind_agent, bind_user): + """The echo guard must not block ordinary captures.""" + mcp = mm.build_memory_mcp(mock.MagicMock()) + result = _tools(mcp)["memory_capture"].fn(content="接口先不要动", source="hook") + assert result["recorded"] is True + assert fake_memory.add_raw.call_args.args[0] == "alice说：接口先不要动" diff --git a/tests/unit/agents/test_memory_recall_patch.py b/tests/unit/agents/test_memory_recall_patch.py new file mode 100644 index 00000000..e88242e1 --- /dev/null +++ b/tests/unit/agents/test_memory_recall_patch.py @@ -0,0 +1,42 @@ +"""Unit tests for the harness_memory recall-quality patch (infra/agents/memory_recall_patch). + +The patch monkey-patches three symbols inside the upstream ``harness-memory`` +package at startup, so these tests only assert the wiring contract: the three +symbols are replaced, the originals are kept as the fallback, and a second call +is a no-op. The patched behaviour itself is covered by the memory eval corpus, +not here. +""" + +from __future__ import annotations + +import harness_memory.pipeline.recall.multi_source as multi_source +import harness_memory.pipeline.recall.router as router + +from octop.infra.agents import memory_recall_patch as patch + + +def test_apply_patches_the_three_recall_symbols(): + """router.route + the two multi_source helpers are replaced, originals kept.""" + patch.apply_memory_recall_patch() + + assert router.route is patch._patched_route + assert multi_source._per_token_atom_search is patch._patched_per_token_atom_search + assert multi_source._gather_atoms is patch._patched_gather_atoms + + # The upstream implementations stay reachable as the fallback path. + assert router._orig_route is not patch._patched_route + assert multi_source._orig_per_token_atom_search is not patch._patched_per_token_atom_search + assert multi_source._orig_gather_atoms is not patch._patched_gather_atoms + + +def test_apply_is_idempotent(): + """A second call must not re-wrap the originals (that would nest the patch).""" + patch.apply_memory_recall_patch() + original_route = router._orig_route + original_gather = multi_source._orig_gather_atoms + + patch.apply_memory_recall_patch() + + assert router._orig_route is original_route + assert multi_source._orig_gather_atoms is original_gather + assert router.route is patch._patched_route