diff --git a/CHECKLIST.md b/CHECKLIST.md index 6f6212e..02ab60e 100644 --- a/CHECKLIST.md +++ b/CHECKLIST.md @@ -76,17 +76,34 @@ - [x] p0-3.6 Go 的 `search_atoms` / `compile_context` 改走 `POST /api/search`,删掉 Go 侧第二套分词器(`store.SearchAtoms` / `ftsQuery` / `tokenPattern`)——两面不可能再漂开,因为只剩一个排序实现 - [ ] p0-3.7 100k atom 下 p50 <100ms:`python -m bench.scale_check` → `bench/reports/SCALE.md`。基线是 p50 20.16s / 165,973 atoms(`bench/reports/BASELINE.md`)。**未测不写数**:机器负载 3.7–17.9 期间测出来的数不算数 - 验收:`pytest backend/tests/test_text_index.py backend/tests/test_search_api.py` 绿(18 条,真 FTS5 表 + TestClient):「钥匙」命中「我们的钥匙放在玄关抽屉里」、`search atoms` 命中 `func SearchAtoms(ctx context.Context)`、`forg` 命中 forgotten、全角 `Search Atoms` 命中、游标翻页无重复、supersede 后旧记忆不再应答;MCP 一致性由 `ely-eye-mcp/internal/mcpserver/search_tools_test.go`(3 条)+ `bash scripts/mcp_write_smoke.sh` PASS 证明;偏差:checklist 写的 `tests/test_retrieval_parity.py` 不存在,实际在 `backend/tests/` 与 Go 测试里 -- [ ] **p0-4** 卡带变成可挂载作用域 - - [ ] p0-4.1 `mount` / `unmount` / `switch` / `append` 四动词 - - [ ] p0-4.2 检索按已挂载卡带过滤(作用域,不再是全局) - - [ ] p0-4.3 `kv_fingerprint`:model / tokenizer / chat template / adapter hash / rope / 量化 / 引擎版本 - - [ ] p0-4.4 fingerprint 不匹配则拒绝挂载,错误信息点名差异字段 - - 验收:`pytest tests/test_cartridge_scope.py tests/test_kv_fingerprint.py` 绿 -- [ ] **p0-5** holder registry + who_knows - - [ ] p0-5.1 `holders` 表 + registry CRUD - - [ ] p0-5.2 `who_knows` 工具:返回 holder 与置信度,非空结果 - - [ ] p0-5.3 转介:visited-set + 深度上限 + 终止态 `NO_HOLDER_KNOWS` - - [ ] p0-5.4 默认 answer-only 转介,不外泄原文 +- [x] **p0-4** 卡带变成可挂载作用域 + - [x] p0-4.1 `mount` / `unmount` / `switch` / `append` 四动词(`cartridge.py:CartridgeService`;schema v2→v3 新增 `cartridge_atoms` 成员表与 `cartridge_mounts` 挂载表,重开时自动重建——挂载是机器状态,不是数据事实) + - [x] p0-4.2 检索按已挂载卡带过滤(`retrieval.py:search` 默认作作域 = 当前挂载集,显式 `cartridge_ids` 覆盖挂载;什么都没挂时仍是全库,空库行为与以前一致)——偏差:`forget` 同时从所有卡带成员表里掉出,否则“忘了”还能从卡带里搜到 + - [x] p0-4.3 `kv_fingerprint`(`kv_fingerprint.py`):model / tokenizer / chat template hash、adapter hash、RoPE 块(factor / original_max_position_embeddings / mrope_interleaved / mrope_section / partial_rotary_factor)、context 长度、引擎版本、指纹格式版本;读不到的项写进 `unverifiable_fields`,不留空 + - [x] p0-4.4 fingerprint 不匹配则拒绝挂载,错误信息逐字段点名差异(`home_x was built for a different runtime; refusing to mount: rope: factor: cartridge 4.0 vs local 2.0; …`) + - 验收:`pytest backend/tests/test_cartridge_scope.py backend/tests/test_kv_fingerprint.py` 绿(26 条:16 作用域 + 10 指纹);端到端 `bash scripts/mcp_write_smoke.sh` → `PASS: remember, supersede, search, forget and cartridge mounting work through MCP`(挂载后 work 的记忆搜不到、上传后新写的记忆要 append 才能检索、unmount 后全库恢复、改 RoPE factor 后 `mount_cartridge` 报错含 `rope`);偏差:checklist 写的 `tests/` 不存在,实际在 `backend/tests/` + - 偏差记录:四动词同时有 HTTP(`/api/mounts`、`/api/cartridges/{id}/{mount,switch,unmount,append}`)、CLI(`ely-eye mount/mounts/unmount/append`)和 5 个 MCP 工具;`append` 重写 `atoms.parquet` + `sparse_index.bm25.zst` 并重算 DNA,文件先写、后写库,失败不会留半个卡带;**量化没进指纹**——量化是引擎启动参数(`profiles.py` 的 `--cache-type-k q4_0` 等),不在模型快照里,指纹只比快照真说得出的东西,否则就是拿不存在的字段做门禁 + - 跑出来的两个真缺陷:`atoms.parquet` 根本导不出记忆(pyarrow 拒绝无子字段的 struct,`metadata={}` 炸掉整表 → 六个 JSON 列改存 JSON 文本,读取端 `load_cartridge_atoms` 解码);`materialize()` 先写成员表后写 cartridges 行,外键拒绝所有卡带 + - `api.py` 超 500 行,卡带/挂载路由拆到 `api_cartridges.py`;那个模块不能用 `from __future__ import annotations`——FastAPI 会拿模块 globals 去解 `Depends(settings_dep)`,而它只是 `register()` 的参数,结果每条路由都要 `settings` 查询参数 +- [x] **p0-5** holder registry + who_knows + - [x] p0-5.1 `holders` 表 + registry CRUD + - 验收:`pytest backend/tests/test_transactive.py -k "registry or kind or blank or holding" -q` 通过;schema v5 在旧库上补表。 + - 事实:`migrations.py` SCHEMA_VERSION 4→5,`holders` 进 `_BASE_TABLES`(holder_id 主键、kind CHECK、knows_about_json、contact_json、created/updated)。CRUD 在 `holder_store.py`(`db.py` 已 498 行,逼近 500 上限,故另立模块)。删除保护:`holder 'x' still holds N memories; forget them before deleting the holder` → HTTP 409。 + - 偏差:方法名 `list` 在类内遮蔽内置 `list`,mypy 判 `list[Holder]` 注解非法(`Function ... is not valid as a type`)→ 改名 `list_holders()`,不是加 cast 糊过去。 + - [x] p0-5.2 `who_knows` 工具:返回 holder 与置信度,非空结果 + - 验收:`pytest backend/tests/test_transactive.py -k "directory or declared" -q`;`bash scripts/mcp_write_smoke.sh` 里 who_knows("护照") 首位是 wife 且带 contact.channel=signal。 + - 事实:目录**不按可见性过滤**("她知道"就是答案),回答才按可见性过滤。置信度是序数的:`0.60*相对得分 + 0.25*min(1, matched/3) + 0.15*声明过这个话题`,相对 = top_score / 候选集最高 top_score(避免 BM25 魔法常数:同一 query 在 3 文档库上量级 1e-7)。每条命中附 matched / top_score / latest_time 与 basis 说明。 + - 根因修复:FTS5 的 `bm25()` 不能聚合(`MAX(-bm25())` + GROUP BY 报 `unable to use function bm25 in the requested context`),子查询/窗口形式也被拒 → 用 FTS5 自带的 `rank` 列(`-MAX(rank)`,实测与 `MAX(-bm25())` 相等,9.1786e-07)。 + - 根因修复:`fts_query("钥匙在哪里")` 生成与问句等长的短语 → 0 命中(实测 钥匙 ✓ / 钥匙在哪里 ✗)。`text_index.py` 新增 `_cjk_phrase`:整段短语 OR 每个相邻二元组短语,相邻性保留(p0-3 的反精确测试仍成立),等长不再是条件。拉丁路径未动 ⇒ 冻结的 BASELINE 数字不受影响。 + - [x] p0-5.3 转介:visited-set + 深度上限 + 终止态 `NO_HOLDER_KNOWS` + - 验收:`pytest backend/tests/test_transactive.py -k "walks or hop or nobody" -q`。 + - 事实:`ask()` 先查 asker 可见范围(hop 0),再按 who_knows 命中依次跳转,`asked` 列表即 visited-set,不重复问同一 holder;`max_hops` 限 1..10,越界 `ValueError("ask needs max_hops between 1 and 10, got N")`;终止态 `answered` / `referred` / `no_holder_knows`(无人知道是结果,不是错误:`basis=["no holder has a memory that answers this"]`)。 + - 根因修复:`if len(asked) >= max_hops: break` 把 asker 自己算作一跳 ⇒ `max_hops=1` 谁也没问就返回 no_holder_knows → 改 `len(asked) - 1 >= max_hops`,docstring 写明「max_hops 数的是 asker 之外被问的人」。 + - [x] p0-5.4 默认 answer-only 转介,不外泄原文 + - 验收:`bash scripts/mcp_write_smoke.sh` → `PASS: remember, supersede, search, forget, cartridge mounting and the transactive referral work through MCP`(两个进程、一个 SQLite:husband 通过 MCP `ask` 得到 `status=referred`、`answered_by=wife`、`answer` 为空且整份 JSON 里搜不到妻子的原文)。 + - 事实:`include_text` 只对可读所有者(asker 本人或无主的 library 条目)显示原文;`referred` 只带 contact 渠道。 + - 根因修复:`_answered()` 里 `readable = owner in (as_holder, LIBRARY)` 是 SQL 可见性子句之外的第二份、更严的规则,与 `visible_to=[husband]` 冲突(共享记忆被错误隐藏)→ 删掉这份重复判断,SQL 的 `as_holder` 可见性子句是唯一权威。 + - 偏差:Go 侧 `limit`/`max_hops` 零值曾当真实数字发出,被 API 以 `greater_than` 拒 → 未设置的可选字段一律省略,由服务端定默认值。 - 验收:`pytest tests/test_transactive.py` 绿;钥匙测试场景通过 - [ ] **p0-6** 证明套件去自证 - [ ] p0-6.1 删 arena 答案入墨(`evals.py:1408-1409` 连接线、`evals.py:1417-1426` 白圈) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 8edb8ab..6036e2d 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -51,3 +51,7 @@ where = ["src"] line-length = 100 target-version = "py312" +[tool.ruff.lint] +# Typer requires an option default to be a typer.Option call, which is what B008 forbids. +ignore = ["B008"] + diff --git a/backend/src/ely_eye/api.py b/backend/src/ely_eye/api.py index 113b579..36560a6 100644 --- a/backend/src/ely_eye/api.py +++ b/backend/src/ely_eye/api.py @@ -11,6 +11,7 @@ from pydantic import BaseModel from . import __version__ from .api_cartridges import register as register_cartridge_routes +from .api_holders import register as register_holder_routes from .benchmarks import ( benchmark_prediction_report, benchmark_registry_report, @@ -246,6 +247,7 @@ def create_app() -> FastAPI: raise HTTPException(status_code=422, detail=str(exc)) from exc register_cartridge_routes(app, settings_dep) + register_holder_routes(app, settings_dep) @app.get("/api/adapters", response_model=list[AdapterStatus]) def adapters(settings: Annotated[Settings, Depends(settings_dep)]) -> list[AdapterStatus]: diff --git a/backend/src/ely_eye/api_holders.py b/backend/src/ely_eye/api_holders.py new file mode 100644 index 0000000..8abfcfb --- /dev/null +++ b/backend/src/ely_eye/api_holders.py @@ -0,0 +1,118 @@ +"""Holder registry, who-knows, and referral routes: the directory side of the store. + +No `from __future__ import annotations` here: FastAPI reads these route signatures, and string +annotations would be resolved against module globals, where the injected settings_dep exists +only as a parameter of register(). +""" + +from collections.abc import Callable +from typing import Annotated + +from fastapi import Depends, FastAPI, HTTPException + +from .config import Settings +from .db import Database +from .holder_store import HolderStore +from .schemas import utc_now +from .transactive import TransactiveService +from .transactive_schemas import ( + AskRequest, + Holder, + HolderUpsertRequest, + ReferralAnswer, + WhoKnowsReport, + WhoKnowsRequest, +) + + +def register(app: FastAPI, settings_dep: Callable[[], Settings]) -> None: + """Register the transactive routes. The server owns the timestamps, the caller does not.""" + + def store(settings: Settings) -> HolderStore: + return HolderStore(Database(settings)) + + @app.get("/api/holders", response_model=list[Holder]) + def holders(settings: Annotated[Settings, Depends(settings_dep)]) -> list[Holder]: + return store(settings).list_holders() + + @app.post("/api/holders/{holder_id}", response_model=Holder) + def put_holder( + holder_id: str, + request: HolderUpsertRequest, + settings: Annotated[Settings, Depends(settings_dep)], + ) -> Holder: + now = utc_now() + entry = Holder( + holder_id=holder_id, + created_at=now, + updated_at=now, + **request.model_dump(), + ) + try: + return store(settings).upsert(entry) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + @app.get("/api/holders/{holder_id}", response_model=Holder) + def get_holder( + holder_id: str, + settings: Annotated[Settings, Depends(settings_dep)], + ) -> Holder: + entry = store(settings).get(holder_id) + if entry is None: + raise HTTPException(status_code=404, detail=f"no holder {holder_id!r} is registered") + return entry + + @app.delete("/api/holders/{holder_id}", response_model=Holder) + def delete_holder( + holder_id: str, + settings: Annotated[Settings, Depends(settings_dep)], + ) -> Holder: + registry = store(settings) + entry = registry.get(holder_id) + if entry is None: + raise HTTPException(status_code=404, detail=f"no holder {holder_id!r} is registered") + try: + registry.delete(holder_id) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return entry + + @app.get("/api/holders/{holder_id}/knows", response_model=WhoKnowsReport) + def holder_knows( + holder_id: str, + query: str, + settings: Annotated[Settings, Depends(settings_dep)], + ) -> WhoKnowsReport: + """What this holder knows about, as a ranked directory answer.""" + try: + return TransactiveService(settings).who_knows(query, as_holder=holder_id) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + @app.post("/api/who_knows", response_model=WhoKnowsReport) + def who_knows( + request: WhoKnowsRequest, + settings: Annotated[Settings, Depends(settings_dep)], + ) -> WhoKnowsReport: + try: + return TransactiveService(settings).who_knows( + request.query, as_holder=request.as_holder, limit=request.limit + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + @app.post("/api/ask", response_model=ReferralAnswer) + def ask( + request: AskRequest, + settings: Annotated[Settings, Depends(settings_dep)], + ) -> ReferralAnswer: + try: + return TransactiveService(settings).ask( + request.question, + as_holder=request.as_holder, + max_hops=request.max_hops, + include_text=request.include_text, + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc diff --git a/backend/src/ely_eye/cli.py b/backend/src/ely_eye/cli.py index 6da06af..c880fef 100644 --- a/backend/src/ely_eye/cli.py +++ b/backend/src/ely_eye/cli.py @@ -16,6 +16,7 @@ from .benchmarks import ( ) from .cartridge import CartridgeService from .cartridge_assets import CartridgeAssetService +from .cli_transactive import register as register_transactive_commands from .config import get_settings from .db import Database from .embeddings import EmbeddingService @@ -38,6 +39,9 @@ def json_dump(value: object) -> str: return json.dumps(value, ensure_ascii=False, indent=2) +register_transactive_commands(app, lambda: get_settings()) + + @app.command() def serve(host: str = "127.0.0.1", port: int = 8765) -> None: uvicorn.run(create_app(), host=host, port=port) @@ -202,7 +206,7 @@ def unmount(cartridge_id: str) -> None: @app.command() def append( cartridge_id: str, - atom_id: list[str] = typer.Option(..., "--atom-id"), # noqa: B008 # typer reads option defaults itself + atom_id: list[str] = typer.Option(..., "--atom-id"), ) -> None: """Add memories to an existing cartridge; the bundle and its DNA follow.""" service = CartridgeService(get_settings()) @@ -352,7 +356,7 @@ def build_visual_training_data( prd_path: Path = Path("PRD.md"), output_path: Path | None = None, cartridge_id: str | None = None, - image_path: list[Path] | None = typer.Option(None, "--image-path"), # noqa: B008 # typer reads option defaults itself + image_path: list[Path] | None = typer.Option(None, "--image-path"), ) -> None: path = TrainingService(get_settings()).build_visual_dataset( prd_path=prd_path, diff --git a/backend/src/ely_eye/cli_transactive.py b/backend/src/ely_eye/cli_transactive.py new file mode 100644 index 0000000..b685693 --- /dev/null +++ b/backend/src/ely_eye/cli_transactive.py @@ -0,0 +1,100 @@ +"""CLI face for the directory: register holders, ask who knows, ask and get referred. + +Registered onto the main Typer app so the command names stay flat (`ely-eye who-knows`), while +the transactive surface stays in one file. +""" + +from __future__ import annotations + +from collections.abc import Callable + +import typer + +from .config import Settings +from .db import Database +from .holder_store import HolderStore +from .schemas import utc_now +from .transactive import TransactiveService +from .transactive_schemas import Holder, HolderKind + + +def register(app: typer.Typer, settings: Callable[[], Settings]) -> None: + """Add the holder and referral commands. The caller passes a settings thunk so one patch + in a test redirects every command, the same way the cartridge routes do.""" + + def store() -> HolderStore: + return HolderStore(Database(settings())) + + def asker(as_holder: str) -> str: + """An empty --as-holder means the local principal, the same identity `remember` writes as.""" + return as_holder or settings().default_holder + + @app.command("holders") + def holders() -> None: + """List registered holders: the directory a referral walks.""" + for entry in store().list_holders(): + typer.echo(entry.model_dump_json(indent=2)) + + @app.command("holder-set") + def holder_set( + holder_id: str, + name: str = "", + kind: HolderKind = "person", + knows_about: list[str] = typer.Option(None, "--knows-about"), + contact: list[str] = typer.Option(None, "--contact", help="key=value, repeatable"), + ) -> None: + """Create or replace one holder entry.""" + now = utc_now() + entry = Holder( + holder_id=holder_id, + name=name, + kind=kind, + knows_about=list(knows_about or []), + contact=_pairs(contact), + created_at=now, + updated_at=now, + ) + typer.echo(store().upsert(entry).model_dump_json(indent=2)) + + @app.command("holder-rm") + def holder_rm(holder_id: str) -> None: + """Delete a holder. Refuses while that holder still owns memories.""" + store().delete(holder_id) + typer.echo(f'{{"deleted": "{holder_id}"}}') + + @app.command("who-knows") + def who_knows(query: str, as_holder: str = "", limit: int = 8) -> None: + """Rank the holders whose memories answer this query.""" + report = TransactiveService(settings()).who_knows( + query, as_holder=asker(as_holder), limit=limit + ) + typer.echo(report.model_dump_json(indent=2)) + + @app.command() + def ask( + question: str, + as_holder: str = "", + max_hops: int = 3, + include_text: bool = typer.Option( + False, "--include-text", help="show the memory text when the asker may already read it" + ), + ) -> None: + """Ask the store, then whoever knows. An unanswered question says nobody knows.""" + answer = TransactiveService(settings()).ask( + question, + as_holder=asker(as_holder), + max_hops=max_hops, + include_text=include_text, + ) + typer.echo(answer.model_dump_json(indent=2)) + + +def _pairs(values: list[str] | None) -> dict[str, str]: + """Turn repeatable key=value options into a contact map, naming any entry that is not one.""" + contact: dict[str, str] = {} + for value in values or []: + key, separator, detail = value.partition("=") + if not separator or not key.strip(): + raise ValueError(f"--contact needs key=value, got {value!r}") + contact[key.strip()] = detail + return contact diff --git a/backend/src/ely_eye/db.py b/backend/src/ely_eye/db.py index defb298..9b28179 100644 --- a/backend/src/ely_eye/db.py +++ b/backend/src/ely_eye/db.py @@ -188,12 +188,18 @@ class Database: until: datetime | None = None, min_trust: float | None = None, cartridge_ids: list[str] | None = None, + as_holder: str | None = None, ) -> FtsPage: """BM25 over the FTS index joined to memories that are open right now. Every filter is a `? IS NULL OR` predicate in one literal statement: the ranking never depends on string-built SQL, and the planner sees one shape. A memory with no event time cannot answer a time filter, so it drops out when one is asked for. + + `as_holder` applies the sharing rule: a memory is readable by its owner, by anyone named + in its visible_to list, by everyone when it is public, and by anyone when nobody owns it. + A memory with no holder cannot be private to a holder that does not exist. Without an + asker the query runs over the whole library, which is what an operator debugging wants. """ with self.connect() as connection: rows = connection.execute( @@ -221,6 +227,13 @@ class Database: WHERE ca.atom_id = f.atom_id AND ca.cartridge_id IN (SELECT value FROM json_each(?)) )) + AND (? IS NULL OR a.holder_id = '' + OR json_extract(a.visibility_json, '$.scope') = 'everyone' + OR a.holder_id = ? + OR EXISTS ( + SELECT 1 FROM json_each(json_extract(a.visibility_json, '$.holders')) + WHERE value = ? + )) ORDER BY score LIMIT ? OFFSET ? """, @@ -242,6 +255,9 @@ class Database: min_trust, None if cartridge_ids is None else to_json(cartridge_ids), None if cartridge_ids is None else to_json(cartridge_ids), + as_holder, + as_holder, + as_holder, limit + 1, offset, ), diff --git a/backend/src/ely_eye/holder_store.py b/backend/src/ely_eye/holder_store.py new file mode 100644 index 0000000..ad24d0b --- /dev/null +++ b/backend/src/ely_eye/holder_store.py @@ -0,0 +1,173 @@ +"""The holder registry and the directory query: who holds a memory, and who knows a topic. + +A transactive memory system stores the directory, not the content: "I do not know where the +keys are, I know my wife knows, and I can find my wife." That requires a holder entity the +store can answer questions about, and it requires the answer to be a pointer with a reason, +never a guess. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass +from datetime import datetime +from typing import cast + +from .db import Database +from .row_codec import from_json, to_json +from .schemas import utc_now +from .text_index import fts_query +from .transactive_schemas import Holder, HolderKind + + +@dataclass(frozen=True) +class WhoKnowsCount: + """One holder's evidence for one query, straight from the index.""" + + holder_id: str + matched: int + top_score: float + latest_time: str | None + + +class HolderStore: + """CRUD for registered holders plus the raw counts who_knows is built from.""" + + def __init__(self, database: Database) -> None: + self.database = database + + def upsert(self, holder: Holder) -> Holder: + """Insert or replace a registry entry, keeping the original creation time.""" + now = utc_now().isoformat() + with self.database.connect() as connection: + existing = connection.execute( + "SELECT created_at FROM holders WHERE holder_id = ?", (holder.holder_id,) + ).fetchone() + created_at = str(existing["created_at"]) if existing is not None else now + stored = holder.model_copy(update={"created_at": _moment(created_at), "updated_at": _moment(now)}) + connection.execute( + """ + INSERT INTO holders (holder_id, name, kind, knows_about_json, contact_json, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(holder_id) DO UPDATE SET + name = excluded.name, + kind = excluded.kind, + knows_about_json = excluded.knows_about_json, + contact_json = excluded.contact_json, + updated_at = excluded.updated_at + """, + ( + stored.holder_id, + stored.name, + stored.kind, + to_json(stored.knows_about), + to_json(stored.contact), + stored.created_at.isoformat(), + stored.updated_at.isoformat(), + ), + ) + return stored + + def get(self, holder_id: str) -> Holder | None: + with self.database.connect() as connection: + row = connection.execute("SELECT * FROM holders WHERE holder_id = ?", (holder_id,)).fetchone() + return None if row is None else _row_to_holder(row) + + def list_holders(self) -> list[Holder]: + with self.database.connect() as connection: + rows = connection.execute("SELECT * FROM holders ORDER BY holder_id").fetchall() + return [_row_to_holder(row) for row in rows] + + def delete(self, holder_id: str) -> None: + """Remove a registry entry. Refuses while that holder still owns memories: forgetting a + phone number must not silently orphan the memories it points to.""" + holding = self.atom_count(holder_id) + if holding: + raise ValueError( + f"holder {holder_id!r} still holds {holding} memories; forget them before deleting the holder" + ) + with self.database.connect() as connection: + connection.execute("DELETE FROM holders WHERE holder_id = ?", (holder_id,)) + + def atom_count(self, holder_id: str) -> int: + with self.database.connect() as connection: + row = connection.execute( + "SELECT COUNT(*) AS n FROM v_atoms_current WHERE holder_id = ?", (holder_id,) + ).fetchone() + return int(row["n"]) + + def known_holders(self) -> list[str]: + """Every holder with at least one open memory, registered or not. + + A person who has memories is a holder whether or not anybody registered them; the + registry adds declared topics and contact details, it does not grant existence. + """ + with self.database.connect() as connection: + rows = connection.execute( + "SELECT DISTINCT holder_id FROM v_atoms_current WHERE holder_id <> '' ORDER BY holder_id" + ).fetchall() + return [str(row["holder_id"]) for row in rows] + + def who_knows_counts(self, query: str, limit: int) -> list[WhoKnowsCount]: + """Rank holders by how well their own memories answer the query. + + The directory is deliberately unfiltered by visibility: the whole point of transactive + memory is knowing that someone knows, even when their memory is private to them. + + `rank` is FTS5's own bm25 column, and it is the only way to aggregate the score: SQLite + refuses bm25() inside MAX(), and refuses it in a subquery or window over a match. + """ + with self.database.connect() as connection: + rows = connection.execute( + """ + SELECT a.holder_id, + COUNT(*) AS matched, + -MAX(rank) AS top_score, + MAX(COALESCE(a.event_time, a.ingested_at)) AS latest_time + FROM atom_fts AS f + JOIN v_atoms_current AS a ON a.atom_id = f.atom_id + WHERE atom_fts MATCH ? + AND a.holder_id <> '' + GROUP BY a.holder_id + ORDER BY top_score DESC, matched DESC + LIMIT ? + """, + (fts_query(query), limit), + ).fetchall() + return [ + WhoKnowsCount( + holder_id=str(row["holder_id"]), + matched=int(row["matched"]), + top_score=float(row["top_score"]), + latest_time=None if row["latest_time"] is None else str(row["latest_time"]), + ) + for row in rows + ] + + +def _moment(value: str) -> datetime: + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + raise ValueError(f"holder timestamps need an explicit offset, stored {value!r}") + return parsed + + +_HOLDER_KINDS: tuple[HolderKind, ...] = ("person", "agent", "service") + + +def _kind(value: str) -> HolderKind: + if value not in _HOLDER_KINDS: + raise ValueError(f"stored holder kind {value!r} is not one of person, agent, service") + return cast(HolderKind, value) + + +def _row_to_holder(row: sqlite3.Row) -> Holder: + return Holder( + holder_id=str(row["holder_id"]), + name=str(row["name"]), + kind=_kind(str(row["kind"])), + knows_about=from_json(row["knows_about_json"], []), + contact=from_json(row["contact_json"], {}), + created_at=_moment(str(row["created_at"])), + updated_at=_moment(str(row["updated_at"])), + ) diff --git a/backend/src/ely_eye/migrations.py b/backend/src/ely_eye/migrations.py index 1eefd02..105b095 100644 --- a/backend/src/ely_eye/migrations.py +++ b/backend/src/ely_eye/migrations.py @@ -14,7 +14,7 @@ import sqlite3 from .text_index import index_text -SCHEMA_VERSION = 4 +SCHEMA_VERSION = 5 ATOM_COLUMNS_V2 = """ atom_id TEXT PRIMARY KEY, @@ -82,6 +82,21 @@ _BASE_TABLES = """ mounted_at TEXT NOT NULL ); + -- The directory of who may hold or answer a memory. A transactive system answers + -- "who knows this?" before it answers "what is this?", and it cannot do that without + -- an entity to point at. Contact details are free-form on purpose: the transport is + -- whoever wired this store together, not the schema. + CREATE TABLE IF NOT EXISTS holders ( + holder_id TEXT PRIMARY KEY, + name TEXT NOT NULL DEFAULT '', + kind TEXT NOT NULL DEFAULT 'person' + CHECK (kind IN ('person', 'agent', 'service')), + knows_about_json TEXT NOT NULL DEFAULT '[]', + contact_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS cache_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, diff --git a/backend/src/ely_eye/schemas.py b/backend/src/ely_eye/schemas.py index 069b00b..27f4a1a 100644 --- a/backend/src/ely_eye/schemas.py +++ b/backend/src/ely_eye/schemas.py @@ -157,10 +157,12 @@ class SearchFilters(BaseModel): """Who may answer, in what setting, and how fresh the memory has to be. A memory with no event time cannot satisfy since/until, so it drops out when a time bound is - asked for rather than passing as unknown. + asked for rather than passing as unknown. `holder` restricts the answer to one holder's + memories; `as_holder` is the identity asking, and applies the sharing rule instead. """ holder: str | None = None + as_holder: str | None = None scenario: str | None = None source_id: str | None = None modality: Modality | None = None diff --git a/backend/src/ely_eye/text_index.py b/backend/src/ely_eye/text_index.py index acd426d..5731e6b 100644 --- a/backend/src/ely_eye/text_index.py +++ b/backend/src/ely_eye/text_index.py @@ -42,7 +42,7 @@ def fts_query(query: str) -> str: terms = _bare(query).split() if not terms: raise ValueError(f"search needs at least one searchable word, got {query!r}") - return " ".join(_phrase(term) if _is_cjk(term) else f"{_quote(term)}*" for term in terms) + return " ".join(_cjk_phrase(term) if _is_cjk(term) else f"{_quote(term)}*" for term in terms) def _bare(text: str) -> str: @@ -59,5 +59,17 @@ def _phrase(run: str) -> str: return '"' + " ".join(run) + '"' +def _cjk_phrase(run: str) -> str: + """A CJK run stays adjacent, and a long question also reaches the pairs inside it. + + The whole run as one phrase is the precise reading, and it is what a two-character word + needs. A question is usually longer than the sentence that answers it: 钥匙在哪里 will not + appear inside 我们的钥匙放在玄关抽屉里. Offering every adjacent pair as an alternative keeps + adjacency — 钥 and 匙 still have to touch — without demanding equal length. + """ + phrases = list(dict.fromkeys([_phrase(run)] + [_phrase(run[start : start + 2]) for start in range(len(run) - 1)])) + return phrases[0] if len(phrases) == 1 else "(" + " OR ".join(phrases) + ")" + + def _quote(term: str) -> str: return '"' + term.replace('"', '""') + '"' diff --git a/backend/src/ely_eye/transactive.py b/backend/src/ely_eye/transactive.py new file mode 100644 index 0000000..1115fb1 --- /dev/null +++ b/backend/src/ely_eye/transactive.py @@ -0,0 +1,223 @@ +"""Ask the store who knows something, then ask them. + +The transactive rule this service implements: not knowing is fine when the directory knows +where to send you. Three things follow, and each one is a failure mode if left implicit. + +- The directory is public, the content is not. `who_knows` counts every holder's matching + memories, including memories the asker may not read, because "she knows" is the answer. +- A referral never invents an answer. If the asker may not read the holder's memories, the + result is a pointer plus contact details, and the text stays where it is. +- Referral must terminate. Every hop records whom it asked, and the chain stops at the hop + limit or when every candidate has been asked once. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Literal + +from .config import Settings, get_settings +from .db import Database +from .holder_store import HolderStore, WhoKnowsCount +from .retrieval import RetrievalService +from .schemas import SearchFilters +from .text_index import index_text +from .transactive_schemas import ( + ReferralAnswer, + WhoKnowsHit, + WhoKnowsReport, +) + +NO_HOLDER_KNOWS: Literal["no_holder_knows"] = "no_holder_knows" +LIBRARY = "library" +WHO_KNOWS_LIMIT = 8 +MAX_HOPS_LIMIT = 10 + +# Confidence is ordinal: it orders candidates, it does not estimate a probability. The weights +# say what the evidence is worth relative to itself, and every hit carries the raw numbers too. +CONFIDENCE_SATURATE = 3 +WEIGHT_STRENGTH = 0.60 +WEIGHT_COVERAGE = 0.25 +WEIGHT_DECLARED = 0.15 + + +class TransactiveService: + """who_knows and ask, built on the holder registry and the same retrieval as search.""" + + def __init__(self, settings: Settings | None = None, database: Database | None = None) -> None: + self.settings = settings or get_settings() + self.db = database or Database(self.settings) + self.holders = HolderStore(self.db) + self.retrieval = RetrievalService(self.settings, self.db) + + def who_knows( + self, query: str, *, as_holder: str | None = None, limit: int = WHO_KNOWS_LIMIT + ) -> WhoKnowsReport: + """Rank the holders whose own memories answer this query. + + An empty result is the honest answer, not a fallback: nobody holds this, so the caller + should say so instead of paraphrasing something unrelated. + """ + if not query.strip(): + raise ValueError("who_knows needs a query") + if limit <= 0: + raise ValueError(f"who_knows needs a positive limit, got {limit}") + counts = self.holders.who_knows_counts(query, limit) + if not counts: + return WhoKnowsReport(query=query, as_holder=as_holder, status=NO_HOLDER_KNOWS, hits=[]) + registered = {holder.holder_id: holder for holder in self.holders.list_holders()} + best = max(count.top_score for count in counts) + return WhoKnowsReport( + query=query, + as_holder=as_holder, + status="found", + hits=[ + self._hit(count, best, registered.get(count.holder_id), query) + for count in counts + ], + ) + + def ask( + self, + question: str, + *, + as_holder: str | None = None, + max_hops: int = 3, + include_text: bool = False, + ) -> ReferralAnswer: + """Answer a question by asking the asker first, then the directory's best candidates. + + `answered_by` is the asker when their own memory answered, `library` when an unowned + artifact did, and the holder whose memory answered otherwise. `referred` means somebody + knows but their memory is not readable by the asker. + + `max_hops` counts the holders asked besides the asker: the asker's own memory is hop 0, + so max_hops=1 still asks the best candidate. + """ + if not question.strip(): + raise ValueError("ask needs a question") + if not 1 <= max_hops <= MAX_HOPS_LIMIT: + raise ValueError(f"ask needs max_hops between 1 and {MAX_HOPS_LIMIT}, got {max_hops}") + + own = self._search(question, as_holder=as_holder, holder=None) + if own: + return self._answered( + question, as_holder, asked=[as_holder or LIBRARY], hops=0, hits=own, include_text=include_text + ) + + asked: list[str] = [as_holder or LIBRARY] + for hit in self.who_knows(question, as_holder=as_holder).hits: + if hit.holder_id in asked: + continue + if len(asked) - 1 >= max_hops: + break + asked.append(hit.holder_id) + readable = self._search(question, as_holder=as_holder, holder=hit.holder_id) + if readable: + return self._answered( + question, + as_holder, + asked=asked, + hops=len(asked) - 1, + hits=readable, + include_text=include_text, + ) + return self._pointer(question, as_holder, asked, hit) + return ReferralAnswer( + question=question, + as_holder=as_holder, + status=NO_HOLDER_KNOWS, + asked=asked, + hops=len(asked) - 1, + basis=["no holder has a memory that answers this"], + ) + + def _search(self, query: str, *, as_holder: str | None, holder: str | None): + result = self.retrieval.search( + query, + top_k=5, + filters=SearchFilters(holder=holder, as_holder=as_holder), + ) + return result.hits + + def _answered(self, question, as_holder, *, asked, hops, hits, include_text) -> ReferralAnswer: + """The hits already passed the visibility clause in SQL, so they may be shown.""" + top = hits[0] + owner = top.atom.holder_id or LIBRARY + return ReferralAnswer( + question=question, + as_holder=as_holder, + status="answered", + answered_by=owner, + asked=asked, + hops=hops, + atom_ids=[hit.atom.atom_id for hit in hits], + answer=top.atom.text if include_text else None, + basis=[ + f"{len(hits)} memories readable by {as_holder or 'anyone'}", + f"best match scores {top.final_score:.2f}", + f"held by {owner}", + ], + ) + + def _pointer(self, question, as_holder, asked, hit: WhoKnowsHit) -> ReferralAnswer: + """The candidate knows but the asker cannot read it: return the contact, not the text.""" + contact = hit.contact + return ReferralAnswer( + question=question, + as_holder=as_holder, + status="referred", + answered_by=hit.holder_id, + asked=asked, + hops=len(asked) - 1, + basis=[ + f"{hit.holder_id} holds {hit.matched} matching memories the asker may not read", + f"best match scores {hit.top_score:.2f}", + f"reachable via {contact}" if contact else "no contact details registered", + ], + ) + + def _hit(self, count: WhoKnowsCount, best: float, holder, query: str) -> WhoKnowsHit: + declared = _declared_topics(holder.knows_about if holder else [], query) + strength = count.top_score / best if best > 0 else 0.0 + coverage = min(1.0, count.matched / CONFIDENCE_SATURATE) + confidence = ( + WEIGHT_STRENGTH * strength + WEIGHT_COVERAGE * coverage + WEIGHT_DECLARED * (1.0 if declared else 0.0) + ) + basis = [ + f"holds {count.matched} matching memories", + f"best match scores {count.top_score:.2f}", + ] + if count.latest_time: + basis.append(f"most recent memory is from {count.latest_time}") + if declared: + basis.append(f"declared topics: {', '.join(declared)}") + contact = dict(holder.contact) if holder else {} + if contact: + basis.append(f"reachable via {contact}") + return WhoKnowsHit( + holder_id=count.holder_id, + name=holder.name if holder else "", + confidence=round(confidence, 3), + matched=count.matched, + top_score=round(count.top_score, 3), + latest_time=None if count.latest_time is None else _as_moment(count.latest_time), + basis=basis, + contact=contact, + ) + + +def _as_moment(value: str) -> datetime: + """A stored timestamp that will not parse is a corrupt row, not a missing one.""" + try: + return datetime.fromisoformat(value) + except ValueError as error: + raise ValueError(f"stored timestamp {value!r} is unreadable: {error}") from error + + +def _declared_topics(topics: list[str], query: str) -> list[str]: + """Declared topics that share a searchable token with the query, using the index's own split.""" + query_tokens = set(index_text(query).split()) + if not query_tokens: + return [] + return [topic for topic in topics if query_tokens & set(index_text(topic).split())] diff --git a/backend/src/ely_eye/transactive_schemas.py b/backend/src/ely_eye/transactive_schemas.py new file mode 100644 index 0000000..3e361cf --- /dev/null +++ b/backend/src/ely_eye/transactive_schemas.py @@ -0,0 +1,115 @@ +"""Models for the directory side of memory: who exists, who knows what, and who answered. + +Confidence here is ordinal, not calibrated: it orders candidates by the evidence in front of +us. Every model that carries a score also carries the raw evidence behind it, so a caller can +disagree with the number without re-running the query. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, Field, field_validator + +HolderKind = Literal["person", "agent", "service"] +"""The registry's own vocabulary; migrations.py enforces the same set in the database.""" + + +class Holder(BaseModel): + """A registered person, agent, or service that can hold or answer memories.""" + + holder_id: str + name: str = "" + kind: HolderKind = "person" + knows_about: list[str] = Field(default_factory=list) + contact: dict[str, str] = Field(default_factory=dict) + created_at: datetime + updated_at: datetime + + @field_validator("holder_id") + @classmethod + def _holder_id_is_usable(cls, value: str) -> str: + if not value.strip(): + raise ValueError("holder_id cannot be blank; a directory entry has to name somebody") + return value.strip() + + +class HolderUpsertRequest(BaseModel): + """What a caller may set about a holder. The id comes from the path, never the body.""" + + name: str = "" + kind: HolderKind = "person" + knows_about: list[str] = Field(default_factory=list) + contact: dict[str, str] = Field(default_factory=dict) + + +class WhoKnowsHit(BaseModel): + """One candidate holder, the evidence for them, and how to reach them.""" + + holder_id: str + name: str = "" + confidence: float + matched: int + top_score: float + latest_time: datetime | None = None + basis: list[str] = Field(default_factory=list) + contact: dict[str, str] = Field(default_factory=dict) + + +class WhoKnowsReport(BaseModel): + """The directory's answer to a question. `no_holder_knows` is a result, not a failure.""" + + query: str + as_holder: str | None = None + status: Literal["found", "no_holder_knows"] + hits: list[WhoKnowsHit] = Field(default_factory=list) + + +class WhoKnowsRequest(BaseModel): + """Ask the directory who holds an answer.""" + + query: str + as_holder: str | None = None + limit: int = Field(default=8, gt=0) + + @field_validator("query") + @classmethod + def _query_is_usable(cls, value: str) -> str: + if not value.strip(): + raise ValueError("who_knows needs a query") + return value.strip() + + +class AskRequest(BaseModel): + """Ask a question on behalf of one holder, allowing a bounded chain of referrals.""" + + question: str + as_holder: str | None = None + max_hops: int = 3 + include_text: bool = False + + @field_validator("question") + @classmethod + def _question_is_usable(cls, value: str) -> str: + if not value.strip(): + raise ValueError("ask needs a question") + return value.strip() + + +class ReferralAnswer(BaseModel): + """Who answered, who was asked along the way, and what may be shown. + + `referred` means the directory found someone who knows but the asker may not read their + memory: the answer is the pointer and the contact, never the text. + """ + + question: str + as_holder: str | None = None + status: Literal["answered", "referred", "no_holder_knows"] + answered_by: str | None = None + asked: list[str] = Field(default_factory=list) + hops: int = 0 + atom_ids: list[str] = Field(default_factory=list) + answer: str | None = None + basis: list[str] = Field(default_factory=list) diff --git a/backend/tests/test_transactive.py b/backend/tests/test_transactive.py new file mode 100644 index 0000000..2292185 --- /dev/null +++ b/backend/tests/test_transactive.py @@ -0,0 +1,249 @@ +"""The transactive test: I do not know where the keys are, I know my wife knows, I find my wife.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from pydantic import ValidationError +from typer.testing import CliRunner + +import ely_eye.api as api_module +import ely_eye.cli as cli_module +from ely_eye.cli import app +from ely_eye.config import Settings +from ely_eye.db import Database +from ely_eye.ingestion import IngestionService +from ely_eye.memory import MemoryService +from ely_eye.migrations import SCHEMA_VERSION +from ely_eye.schemas import MemoryOrigin, Visibility, utc_now +from ely_eye.transactive import LIBRARY, NO_HOLDER_KNOWS, TransactiveService +from ely_eye.transactive_schemas import Holder + +HOME_KEYS = "我们的钥匙放在玄关抽屉里" +WORK_KEYS = "备用钥匙在办公室第二个抽屉里" +QUERY = "钥匙" + + +class Store: + """One SQLite file, the writer and the directory that read it.""" + + def __init__(self, settings: Settings) -> None: + self.settings = settings + self.memory = MemoryService(settings=settings) + self.db = self.memory.db + self.transactive = TransactiveService(settings=settings, database=self.db) + + def wife_remembers(self, text: str, visibility: Visibility | None = None) -> str: + atom = self.memory.remember( + text, holder_id="wife", origin=MemoryOrigin.self_, scenario="home", visibility=visibility + ) + return atom.atom_id + + def husband_remembers(self, text: str) -> str: + atom = self.memory.remember(text, holder_id="husband", origin=MemoryOrigin.self_, scenario="home") + return atom.atom_id + + def register(self, holder_id: str, **fields: Any) -> Holder: + moment = utc_now() + return self.transactive.holders.upsert( + Holder(holder_id=holder_id, created_at=moment, updated_at=moment, **fields) + ) + + +@pytest.fixture +def store(tmp_path: Path) -> Store: + return Store(Settings(home=tmp_path / "home", default_holder="husband")) + + +def test_the_directory_names_the_holder_and_says_why(store: Store) -> None: + store.wife_remembers(HOME_KEYS) + report = store.transactive.who_knows(QUERY, as_holder="husband") + assert report.status == "found" + assert report.hits[0].holder_id == "wife" + assert report.hits[0].confidence > 0.0 + assert any("matching memories" in line for line in report.hits[0].basis) + + +def test_the_directory_counts_a_memory_the_asker_cannot_read(store: Store) -> None: + """The whole point: knowing that she knows does not require reading her memory.""" + store.wife_remembers(HOME_KEYS) + assert store.transactive.who_knows(QUERY, as_holder="husband").hits[0].matched == 1 + assert store.transactive._search(QUERY, as_holder="husband", holder="wife") == [] + + +def test_a_private_memory_is_referred_and_not_leaked(store: Store) -> None: + store.register("wife", name="小美", contact={"channel": "signal", "address": "+86-555"}) + store.wife_remembers(HOME_KEYS) + answer = store.transactive.ask("钥匙在哪里", as_holder="husband") + assert answer.status == "referred" + assert answer.answered_by == "wife" + assert answer.hops == 1 + assert answer.asked == ["husband", "wife"] + assert answer.answer is None + assert answer.atom_ids == [] + assert any("may not read" in line for line in answer.basis) + assert any("signal" in line for line in answer.basis) + + +def test_the_askers_own_memory_answers_without_a_referral(store: Store) -> None: + store.wife_remembers(HOME_KEYS) + store.husband_remembers("我的工牌挂在显示器边上") + answer = store.transactive.ask("工牌", as_holder="husband") + assert answer.status == "answered" + assert answer.answered_by == "husband" + assert answer.hops == 0 + assert answer.asked == ["husband"] + assert answer.answer is None + + +def test_text_shown_only_when_the_asker_may_already_read_it(store: Store) -> None: + store.husband_remembers("我的工牌挂在显示器边上") + with_text = store.transactive.ask("工牌", as_holder="husband", include_text=True) + assert with_text.answer == "我的工牌挂在显示器边上" + + store.register("wife") + store.wife_remembers(HOME_KEYS) + referred = store.transactive.ask("钥匙在哪里", as_holder="husband", include_text=True) + assert referred.status == "referred" + assert referred.answer is None + + +def test_a_shared_memory_is_readable_without_a_referral(store: Store) -> None: + """A memory the asker may already read answers at hop 0: no referral, the text is shown.""" + store.register("wife") + store.wife_remembers(HOME_KEYS, visibility=Visibility(scope="visible_to", holders=["husband"])) + answer = store.transactive.ask("钥匙在哪里", as_holder="husband", include_text=True) + assert answer.status == "answered" + assert answer.answered_by == "wife" + assert answer.hops == 0 + assert answer.answer == HOME_KEYS + + +def test_an_unowned_artifact_answers_as_the_library(store: Store) -> None: + notebook = store.settings.home / "notebook.txt" + notebook.write_text("the garage door code is written on the fuse box", encoding="utf-8") + IngestionService(store.settings).ingest_path(notebook) + answer = store.transactive.ask("garage door code", as_holder="wife", include_text=True) + assert answer.status == "answered" + assert answer.answered_by == LIBRARY + assert "fuse box" in (answer.answer or "") + + +def test_nobody_knows_is_a_result_and_not_a_fabrication(store: Store) -> None: + store.wife_remembers(HOME_KEYS) + report = store.transactive.who_knows("潜水镜", as_holder="husband") + assert report.status == NO_HOLDER_KNOWS + assert report.hits == [] + + answer = store.transactive.ask("潜水镜在哪里", as_holder="husband") + assert answer.status == NO_HOLDER_KNOWS + assert answer.answered_by is None + assert answer.answer is None + assert answer.basis == ["no holder has a memory that answers this"] + + +def test_the_referral_walks_once_and_never_asks_the_same_holder_twice(store: Store) -> None: + store.register("wife") + store.register("daughter") + store.wife_remembers(HOME_KEYS) + store.memory.remember(WORK_KEYS, holder_id="daughter", origin=MemoryOrigin.self_, scenario="work") + answer = store.transactive.ask("钥匙在哪里", as_holder="husband", max_hops=1) + assert answer.status in ("referred", "answered") + assert len(answer.asked) == 2 + assert len(set(answer.asked)) == len(answer.asked) + assert answer.asked[0] == "husband" + + +def test_the_hop_limit_is_enforced(store: Store) -> None: + for bad in (0, 11): + with pytest.raises(ValueError, match="max_hops between 1 and 10"): + store.transactive.ask(QUERY, as_holder="husband", max_hops=bad) + + +def test_declared_topics_raise_the_confidence_and_say_so(store: Store) -> None: + store.wife_remembers(HOME_KEYS) + before = store.transactive.who_knows(QUERY, as_holder="husband").hits[0] + store.register("wife", knows_about=["家里的钥匙"]) + after = store.transactive.who_knows(QUERY, as_holder="husband").hits[0] + assert after.confidence > before.confidence + assert any("declared topics" in line for line in after.basis) + + +def test_the_registry_refuses_to_delete_a_holder_who_still_holds(store: Store) -> None: + store.register("wife") + atom_id = store.wife_remembers(HOME_KEYS) + with pytest.raises(ValueError, match="still holds 1 memories"): + store.transactive.holders.delete("wife") + store.memory.forget(atom_id=atom_id) + store.transactive.holders.delete("wife") + assert store.transactive.holders.list_holders() == [] + + +def test_an_unknown_holder_kind_is_refused_at_the_door() -> None: + with pytest.raises(ValidationError, match="'person', 'agent' or 'service'"): + Holder.model_validate({"holder_id": "ghost", "kind": "ghost", "created_at": utc_now(), "updated_at": utc_now()}) + + +def test_a_blank_holder_id_is_refused() -> None: + with pytest.raises(ValidationError, match="directory entry has to name somebody"): + Holder.model_validate({"holder_id": " ", "created_at": utc_now(), "updated_at": utc_now()}) + + +def test_a_store_without_holders_gains_the_table_on_reopen(tmp_path: Path) -> None: + settings = Settings(home=tmp_path / "home", default_holder="husband") + database = Database(settings) + with database.connect() as connection: + connection.execute("DROP TABLE holders") + connection.execute("PRAGMA user_version = 4") + reopened = Database(settings) + with reopened.connect() as connection: + assert connection.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION + assert connection.execute("SELECT COUNT(*) AS n FROM holders").fetchone()["n"] == 0 + + +def test_the_http_surface_serves_the_directory(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + settings = Settings(home=tmp_path / "home", default_holder="husband") + monkeypatch.setattr(api_module, "get_settings", lambda: settings) + client = TestClient(api_module.create_app()) + + store = Store(settings) + store.wife_remembers(HOME_KEYS) + + created = client.post( + "/api/holders/wife", + json={"name": "小美", "knows_about": ["家里的钥匙"], "contact": {"channel": "signal"}}, + ) + assert created.status_code == 200 + assert created.json()["holder_id"] == "wife" + assert client.get("/api/holders").json()[0]["name"] == "小美" + assert client.get("/api/holders/nobody").status_code == 404 + + knows = client.get("/api/holders/wife/knows", params={"query": QUERY}).json() + assert knows["status"] == "found" and knows["hits"][0]["holder_id"] == "wife" + + asked = client.post("/api/ask", json={"question": "钥匙在哪里", "as_holder": "husband"}).json() + assert asked["status"] == "referred" and asked["answered_by"] == "wife" + + who = client.post("/api/who_knows", json={"query": "钥匙", "as_holder": "husband"}).json() + assert who["hits"][0]["basis"] + + blocked = client.delete("/api/holders/wife") + assert blocked.status_code == 409 and "still holds" in blocked.json()["detail"] + + +def test_the_cli_asks_and_reports_the_referral(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + settings = Settings(home=tmp_path / "home", default_holder="husband") + monkeypatch.setattr(cli_module, "get_settings", lambda: settings) + runner = CliRunner() + Store(settings).wife_remembers(HOME_KEYS) + + assert runner.invoke(app, ["holder-set", "wife", "--name", "小美", "--contact", "channel=signal"]).exit_code == 0 + who = runner.invoke(app, ["who-knows", QUERY]) + assert who.exit_code == 0 and '"holder_id": "wife"' in who.stdout + asked = runner.invoke(app, ["ask", "钥匙在哪里"]) + assert asked.exit_code == 0 and '"status": "referred"' in asked.stdout + assert runner.invoke(app, ["holders"]).stdout.count("wife") >= 1 + assert runner.invoke(app, ["holder-set", "wife", "--contact", "not-a-pair"]).exit_code != 0 diff --git a/ely-eye-mcp/internal/domain/types.go b/ely-eye-mcp/internal/domain/types.go index 915d270..f4bd700 100644 --- a/ely-eye-mcp/internal/domain/types.go +++ b/ely-eye-mcp/internal/domain/types.go @@ -207,3 +207,43 @@ type ProofSuiteDetail struct { Path string `json:"path"` HashHop HashHopReport `json:"hashhop"` } + +type Holder struct { + HolderID string `json:"holder_id"` + Name string `json:"name"` + Kind string `json:"kind"` + KnowsAbout []string `json:"knows_about"` + Contact map[string]string `json:"contact"` + CreatedAt string `json:"created_at"` + UpdatedAt string `json:"updated_at"` +} + +type WhoKnowsHit struct { + HolderID string `json:"holder_id"` + Name string `json:"name"` + Confidence float64 `json:"confidence"` + Matched int64 `json:"matched"` + TopScore float64 `json:"top_score"` + LatestTime string `json:"latest_time,omitempty"` + Basis []string `json:"basis"` + Contact map[string]string `json:"contact"` +} + +type WhoKnowsReport struct { + Query string `json:"query"` + AsHolder string `json:"as_holder,omitempty"` + Status string `json:"status"` + Hits []WhoKnowsHit `json:"hits"` +} + +type ReferralAnswer struct { + Question string `json:"question"` + AsHolder string `json:"as_holder,omitempty"` + Status string `json:"status"` + AnsweredBy string `json:"answered_by,omitempty"` + Asked []string `json:"asked"` + Hops int64 `json:"hops"` + AtomIDs []string `json:"atom_ids"` + Answer string `json:"answer,omitempty"` + Basis []string `json:"basis"` +} diff --git a/ely-eye-mcp/internal/mcpserver/directory_tools.go b/ely-eye-mcp/internal/mcpserver/directory_tools.go new file mode 100644 index 0000000..ad88fde --- /dev/null +++ b/ely-eye-mcp/internal/mcpserver/directory_tools.go @@ -0,0 +1,102 @@ +package mcpserver + +import ( + "context" + "errors" + + "github.com/ZacharyZhang-NY/ely-eye/ely-eye-mcp/internal/domain" + "github.com/modelcontextprotocol/go-sdk/mcp" +) + +type WhoKnowsInput struct { + Query string `json:"query" jsonschema:"what the asker wants to know, in the asker's own words"` + AsHolder string `json:"as_holder,omitempty" jsonschema:"who is asking; decides which memories may be shown"` + Limit int `json:"limit,omitempty" jsonschema:"how many candidates to return, default 8"` +} + +type AskInput struct { + Question string `json:"question" jsonschema:"the question to answer from memory"` + AsHolder string `json:"as_holder,omitempty" jsonschema:"who is asking; their own memory is checked first"` + MaxHops int `json:"max_hops,omitempty" jsonschema:"how many holders besides the asker may be asked, default 3"` + IncludeText bool `json:"include_text,omitempty" jsonschema:"show the memory text when the asker may already read it"` +} + +// registerDirectoryTools exposes the transactive surface: who knows, and ask-and-get-referred. +func (s *Server) registerDirectoryTools(server *mcp.Server) { + mcp.AddTool(server, &mcp.Tool{ + Title: "Who knows", + Name: "who_knows", + Description: "Rank the holders whose memories answer a query, with the evidence and how to reach them. The directory counts memories the asker cannot read; that is the point. no_holder_knows is a result, not a failure.", + }, s.whoKnows) + + mcp.AddTool(server, &mcp.Tool{ + Title: "Ask and get referred", + Name: "ask", + Description: "Ask the asker's own memory, then the directory. answered means somebody's memory matched; referred means the best holder's memory is private, so the answer is the pointer and contact, never the text.", + }, s.askMemory) + + mcp.AddTool(server, &mcp.Tool{ + Title: "List holders", + Name: "list_holders", + Description: "List registered holders: the people, agents, and services a referral can walk to.", + }, s.listHolders) +} + +type whoKnowsOutput struct { + Report domain.WhoKnowsReport `json:"report" jsonschema:"the ranked candidates and their evidence"` +} + +type askOutput struct { + Answer domain.ReferralAnswer `json:"answer" jsonschema:"who answered, who was asked, and what may be shown"` +} + +type holdersOutput struct { + Holders []domain.Holder `json:"holders" jsonschema:"the registered holders"` +} + +func (s *Server) whoKnows(ctx context.Context, _ *mcp.CallToolRequest, in WhoKnowsInput) (*mcp.CallToolResult, whoKnowsOutput, error) { + if in.Query == "" { + return nil, whoKnowsOutput{}, errors.New("query is required") + } + var report domain.WhoKnowsReport + body := map[string]any{"query": in.Query} + if in.AsHolder != "" { + body["as_holder"] = in.AsHolder + } + if in.Limit > 0 { + body["limit"] = in.Limit + } + if err := s.api.Post(ctx, "/api/who_knows", body, &report); err != nil { + return nil, whoKnowsOutput{}, err + } + return nil, whoKnowsOutput{Report: report}, nil +} + +func (s *Server) askMemory(ctx context.Context, _ *mcp.CallToolRequest, in AskInput) (*mcp.CallToolResult, askOutput, error) { + if in.Question == "" { + return nil, askOutput{}, errors.New("question is required") + } + var answer domain.ReferralAnswer + body := map[string]any{"question": in.Question} + if in.AsHolder != "" { + body["as_holder"] = in.AsHolder + } + if in.MaxHops > 0 { + body["max_hops"] = in.MaxHops + } + if in.IncludeText { + body["include_text"] = true + } + if err := s.api.Post(ctx, "/api/ask", body, &answer); err != nil { + return nil, askOutput{}, err + } + return nil, askOutput{Answer: answer}, nil +} + +func (s *Server) listHolders(ctx context.Context, _ *mcp.CallToolRequest, _ struct{}) (*mcp.CallToolResult, holdersOutput, error) { + var holders []domain.Holder + if err := s.api.Get(ctx, "/api/holders", &holders); err != nil { + return nil, holdersOutput{}, err + } + return nil, holdersOutput{Holders: holders}, nil +} diff --git a/ely-eye-mcp/internal/mcpserver/directory_tools_test.go b/ely-eye-mcp/internal/mcpserver/directory_tools_test.go new file mode 100644 index 0000000..ce9508b --- /dev/null +++ b/ely-eye-mcp/internal/mcpserver/directory_tools_test.go @@ -0,0 +1,133 @@ +package mcpserver + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/ZacharyZhang-NY/ely-eye/ely-eye-mcp/internal/apiclient" +) + +func directoryServer(t *testing.T, body string, seen *[]string) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + payload, _ := io.ReadAll(r.Body) + *seen = append(*seen, r.Method+" "+r.URL.Path+" "+string(payload)) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(body)) + })) +} + +func TestWhoKnowsNamesTheHolderAndKeepsTheEvidence(t *testing.T) { + var seen []string + server := directoryServer(t, `{"query":"钥匙","as_holder":"husband","status":"found","hits":[{ + "holder_id":"wife","name":"小美","confidence":0.91,"matched":2,"top_score":0.02, + "basis":["wife holds 2 matching memories the asker may not read"],"contact":{"channel":"signal"}}]}`, &seen) + defer server.Close() + + _, out, err := (&Server{api: apiclient.New(server.URL)}).whoKnows( + context.Background(), nil, WhoKnowsInput{Query: "钥匙", AsHolder: "husband"}) + if err != nil { + t.Fatalf("who_knows failed: %v", err) + } + if out.Report.Status != "found" || len(out.Report.Hits) != 1 { + t.Fatalf("report = %+v, want one found candidate", out.Report) + } + hit := out.Report.Hits[0] + if hit.HolderID != "wife" || hit.Confidence != 0.91 || hit.Contact["channel"] != "signal" { + t.Errorf("hit = %+v, want the holder, her confidence, and how to reach her", hit) + } + if len(seen) != 1 || !strings.Contains(seen[0], "POST /api/who_knows") || !strings.Contains(seen[0], `"query":"钥匙"`) { + t.Fatalf("requests = %v, want one POST carrying the query", seen) + } +} + +func TestWhoKnowsRefusesAnEmptyQueryBeforeTheNetwork(t *testing.T) { + var seen []string + server := directoryServer(t, `{}`, &seen) + defer server.Close() + + if _, _, err := (&Server{api: apiclient.New(server.URL)}).whoKnows(context.Background(), nil, WhoKnowsInput{}); err == nil { + t.Fatal("an empty query must not reach the directory") + } + if len(seen) != 0 { + t.Errorf("requests = %v, want none", seen) + } +} + +func TestAskReturnsThePointerAndNotThePrivateText(t *testing.T) { + var seen []string + server := directoryServer(t, `{"question":"钥匙在哪里","as_holder":"husband","status":"referred", + "answered_by":"wife","asked":["husband","wife"],"hops":1,"atom_ids":[],"basis":["wife holds 1 matching memories the asker may not read"]}`, &seen) + defer server.Close() + + _, out, err := (&Server{api: apiclient.New(server.URL)}).askMemory( + context.Background(), nil, AskInput{Question: "钥匙在哪里", AsHolder: "husband", MaxHops: 2}) + if err != nil { + t.Fatalf("ask failed: %v", err) + } + if out.Answer.Status != "referred" || out.Answer.Answer != "" { + t.Errorf("answer = %+v, want a referral with no text", out.Answer) + } + if out.Answer.AnsweredBy != "wife" || out.Answer.Hops != 1 { + t.Errorf("answer = %+v, want wife at hop 1", out.Answer) + } + if !strings.Contains(seen[0], `"max_hops":2`) || !strings.Contains(seen[0], `"as_holder":"husband"`) { + t.Errorf("request = %s, want the hop budget and the asker", seen[0]) + } +} + +func TestAskWithoutAnAskerOmitsTheIdentity(t *testing.T) { + var seen []string + server := directoryServer(t, `{"question":"garage code","status":"no_holder_knows","asked":[],"hops":0}`, &seen) + defer server.Close() + + if _, _, err := (&Server{api: apiclient.New(server.URL)}).askMemory(context.Background(), nil, AskInput{Question: "garage code"}); err != nil { + t.Fatalf("ask failed: %v", err) + } + var body map[string]any + parts := strings.SplitN(seen[0], " ", 3) + if err := json.Unmarshal([]byte(parts[2]), &body); err != nil { + t.Fatalf("request body is not JSON: %s", seen[0]) + } + if _, present := body["as_holder"]; present { + t.Errorf("body = %v, want as_holder omitted so the server applies its own default", body) + } + if _, present := body["max_hops"]; present { + t.Errorf("body = %v, want max_hops omitted; a Go zero is not a hop budget", body) + } +} + +func TestAskRefusesAnEmptyQuestion(t *testing.T) { + var seen []string + server := directoryServer(t, `{}`, &seen) + defer server.Close() + + if _, _, err := (&Server{api: apiclient.New(server.URL)}).askMemory(context.Background(), nil, AskInput{}); err == nil { + t.Fatal("an empty question must not be asked") + } + if len(seen) != 0 { + t.Errorf("requests = %v, want none", seen) + } +} + +func TestListHoldersReadsTheRegistry(t *testing.T) { + var seen []string + server := directoryServer(t, `[{"holder_id":"wife","name":"小美","kind":"person","knows_about":["家里的钥匙"],"contact":{"channel":"signal"},"created_at":"2026-05-01T00:00:00+00:00","updated_at":"2026-05-01T00:00:00+00:00"}]`, &seen) + defer server.Close() + + _, out, err := (&Server{api: apiclient.New(server.URL)}).listHolders(context.Background(), nil, struct{}{}) + if err != nil { + t.Fatalf("list_holders failed: %v", err) + } + if len(out.Holders) != 1 || out.Holders[0].KnowsAbout[0] != "家里的钥匙" { + t.Fatalf("holders = %+v, want the declared topics", out.Holders) + } + if len(seen) != 1 || !strings.HasPrefix(seen[0], "GET /api/holders") { + t.Errorf("requests = %v, want one GET", seen) + } +} diff --git a/ely-eye-mcp/internal/mcpserver/server.go b/ely-eye-mcp/internal/mcpserver/server.go index 5a50415..ed9f359 100644 --- a/ely-eye-mcp/internal/mcpserver/server.go +++ b/ely-eye-mcp/internal/mcpserver/server.go @@ -37,6 +37,7 @@ func Run(ctx context.Context, cfg config.Config, version string) error { }) server.registerTools(mcpServer) server.registerWriteTools(mcpServer) + server.registerDirectoryTools(mcpServer) server.registerMountTools(mcpServer) server.registerResources(mcpServer) if err := mcpServer.Run(ctx, &mcp.StdioTransport{}); err != nil { diff --git a/scripts/mcp_write_smoke.sh b/scripts/mcp_write_smoke.sh index 9e96b15..8e82d33 100755 --- a/scripts/mcp_write_smoke.sh +++ b/scripts/mcp_write_smoke.sh @@ -25,8 +25,10 @@ printf '{"metadata": {}, "weight_map": {"model-00001.safetensors": "0"}}' > "$mo export ELY_EYE_MODEL_DIR="$model_dir" api="" +wife_api="" cleanup() { if [ -n "$api" ]; then kill "$api" 2>/dev/null || true; fi + if [ -n "$wife_api" ]; then kill "$wife_api" 2>/dev/null || true; fi rm -rf "$home" } trap cleanup EXIT @@ -47,6 +49,27 @@ fi (cd "$root/ely-eye-mcp" && go build -o "$home/ely-eye-mcp" ./cmd/ely-eye-mcp) +# A second principal on the same SQLite file. The server, not the client, decides who a memory +# belongs to, so the only way to write as the wife is to ask a server whose principal is the wife. +wife_port=$((port + 1)) +ELY_EYE_DEFAULT_HOLDER=wife "$python_bin" -m ely_eye.cli serve --port "$wife_port" >"$home/wife-api.log" 2>&1 & +wife_api=$! +ready=0 +for _ in $(seq 1 80); do + if curl -sf "http://127.0.0.1:${wife_port}/api/status" >/dev/null 2>&1; then ready=1; break; fi + sleep 0.5 +done +if [ "$ready" -ne 1 ]; then + echo "FAIL: the wife's Ely-Eye API did not start" + cat "$home/wife-api.log" + exit 1 +fi +secret="护照在书房第二个抽屉里" +curl -sf -X POST "http://127.0.0.1:${wife_port}/api/memory" -H 'content-type: application/json' \ + -d "{\"text\":\"${secret}\",\"scenario\":\"home\"}" >/dev/null +curl -sf -X POST "http://127.0.0.1:${wife_port}/api/holders/wife" -H 'content-type: application/json' \ + -d '{"name":"小美","kind":"person","knows_about":["护照"],"contact":{"channel":"signal","address":"+86-555"}}' >/dev/null + printf 'the pantry keeps flour\n' > "$home/pantry.txt" uploaded="$(curl -sf -X POST "http://127.0.0.1:${port}/api/ingest/upload?cartridge_name=home" \ -F "files=@${home}/pantry.txt;type=text/plain")" @@ -56,7 +79,7 @@ if [ -z "$cartridge_id" ] || [ "$cartridge_id" = "None" ]; then exit 1 fi -"$python_bin" - "$home/ely-eye-mcp" "$cartridge_id" "$model_dir" <<'PY' +"$python_bin" - "$home/ely-eye-mcp" "$cartridge_id" "$model_dir" "$secret" <<'PY' import json import pathlib import subprocess @@ -65,6 +88,7 @@ import sys binary = sys.argv[1] cartridge_id = sys.argv[2] model_dir = pathlib.Path(sys.argv[3]) +secret = sys.argv[4] proc = subprocess.Popen( [binary, "server"], stdin=subprocess.PIPE, @@ -108,7 +132,8 @@ send({"jsonrpc": "2.0", "method": "notifications/initialized"}) names = [item["name"] for item in call(2, "tools/list", {})["tools"]] for required in ("remember", "supersede_memory", "forget_memory", "mounted_cartridges", - "mount_cartridge", "switch_cartridge", "unmount_cartridge", "append_to_cartridge"): + "mount_cartridge", "switch_cartridge", "unmount_cartridge", "append_to_cartridge", + "who_knows", "ask", "list_holders"): if required not in names: raise SystemExit(f"FAIL: {required} is not registered; tools are {names}") @@ -182,5 +207,38 @@ refused = call(3, "tools/call", {"name": "mount_cartridge", "arguments": {"cartr if not refused.get("isError") or "rope" not in refused["content"][0]["text"]: raise SystemExit(f"FAIL: a cartridge built for RoPE factor 4.0 mounted on a factor 2.0 machine: {refused}") -print("PASS: remember, supersede, search, forget and cartridge mounting work through MCP") +# The transactive proof. The asker is this machine's principal; the wife's memory is private to +# her, so the directory may say she knows without showing what she knows. +holders = tool("list_holders", {})["holders"] +if not any(item["holder_id"] == "wife" for item in holders): + raise SystemExit(f"FAIL: the registry does not list the wife: {holders}") + +directory = tool("who_knows", {"query": "护照", "as_holder": "local"})["report"] +if directory["status"] != "found" or not directory["hits"]: + raise SystemExit(f"FAIL: the directory found nobody who knows: {directory}") +top = directory["hits"][0] +if top["holder_id"] != "wife" or top["contact"].get("channel") != "signal": + raise SystemExit(f"FAIL: the directory did not name the wife and how to reach her: {top}") +if not any("declared topics" in line for line in top["basis"]): + raise SystemExit(f"FAIL: her declared topic did not raise the confidence: {top['basis']}") + +referred = tool("ask", {"question": "护照在哪里", "as_holder": "local", "include_text": True})["answer"] +if referred["status"] != "referred" or referred["answered_by"] != "wife": + raise SystemExit(f"FAIL: the question was not referred to the wife: {referred}") +if referred.get("answer"): + raise SystemExit(f"FAIL: a private memory leaked into the referral: {referred['answer']}") +if secret in json.dumps(referred, ensure_ascii=False): + raise SystemExit(f"FAIL: the referral carried the private text: {referred}") +if "wife" not in referred["asked"]: + raise SystemExit(f"FAIL: the referral never asked the wife: {referred}") + +mine = tool("ask", {"question": "pantry flour", "as_holder": "local", "include_text": True})["answer"] +if mine["status"] != "answered" or not mine["atom_ids"]: + raise SystemExit(f"FAIL: the asker's own memory did not answer: {mine}") + +nobody = tool("ask", {"question": "潜水镜", "as_holder": "local"})["answer"] +if nobody["status"] != "no_holder_knows" or nobody.get("answer"): + raise SystemExit(f"FAIL: nobody knowing was reported as an answer: {nobody}") + +print("PASS: remember, supersede, search, forget, cartridge mounting and the transactive referral work through MCP") PY