diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ed8832f..5afdb99a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ All notable changes to vouch are documented here. Format follows ## [Unreleased] ### Added +- **read-only kb subscriptions — federate search without copying** (#610): + sharing knowledge between KBs meant copying it — `vouch hub` export/import + moves artifacts through the receiving gate and they become local, + duplicated, and drift independently, re-review needed on every upstream + edit. `vouch subscribe ` / `unsubscribe` / `subscriptions list` add + ditto-style subscription instead: a subscribed KB's approved knowledge + joins local `kb.search` / `kb.context` results, live, read-only, never + copied. Federated hits are namespaced (`:`) so they + never resolve locally, tagged with `origin_kb_id`/`origin_kb_name`/ + `trust_level` so a reader can't mistake them for locally-reviewed + knowledge, and capped by a `budget_share` knob that only fills leftover + result slots — a subscription can never crowd out local knowledge. One + hop only: a federated call never itself federates. + - **correction capture — the pushback becomes a proposal** (#430): the adapter captured tool *outcomes* passively but never the single highest-signal event in a session, the user correcting the agent ("no, we deploy from `main` not diff --git a/schemas/context-item.schema.json b/schemas/context-item.schema.json index 2362104d..a9d1b55e 100644 --- a/schemas/context-item.schema.json +++ b/schemas/context-item.schema.json @@ -51,6 +51,19 @@ "title": "Summary", "type": "string" }, + "trust_level": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "vouch: the subscribed KB's trust level ('unverified' or 'trusted'), set only for hits federated in via a read-only subscription. None for local knowledge and for gated-import origin hits (those aren't federated, they're locally-owned copies).", + "title": "Trust Level" + }, "type": { "enum": [ "claim", diff --git a/schemas/context-pack.schema.json b/schemas/context-pack.schema.json index 3616896e..d6395478 100644 --- a/schemas/context-pack.schema.json +++ b/schemas/context-pack.schema.json @@ -51,6 +51,19 @@ "title": "Summary", "type": "string" }, + "trust_level": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "description": "vouch: the subscribed KB's trust level ('unverified' or 'trusted'), set only for hits federated in via a read-only subscription. None for local knowledge and for gated-import origin hits (those aren't federated, they're locally-owned copies).", + "title": "Trust Level" + }, "type": { "enum": [ "claim", diff --git a/src/vouch/capture.py b/src/vouch/capture.py index 9aeb98b8..a66f15c5 100644 --- a/src/vouch/capture.py +++ b/src/vouch/capture.py @@ -75,9 +75,11 @@ def load_config(store: KBStore) -> CaptureConfig: return CaptureConfig( enabled=coerce_bool(raw.get("enabled", DEFAULT_ENABLED), DEFAULT_ENABLED), realtime=coerce_bool(raw.get("realtime", DEFAULT_REALTIME), DEFAULT_REALTIME), - min_observations=int(raw.get("min_observations", DEFAULT_MIN_OBSERVATIONS)), - dedup_window_seconds=float( - raw.get("dedup_window_seconds", DEFAULT_DEDUP_WINDOW_SECONDS) + min_observations=coerce_numeric( + raw.get("min_observations"), DEFAULT_MIN_OBSERVATIONS, int + ), + dedup_window_seconds=coerce_numeric( + raw.get("dedup_window_seconds"), DEFAULT_DEDUP_WINDOW_SECONDS, float ), answer_mode=answer_mode, ) diff --git a/src/vouch/cli.py b/src/vouch/cli.py index 9dba52bd..8cc90bc1 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -52,6 +52,7 @@ from . import sessions as sess_mod from . import skills as skills_mod from . import stats as stats_mod +from . import subscriptions as subscriptions_mod from . import sync as sync_mod from . import synthesize as synth from . import trust as trust_mod @@ -393,6 +394,60 @@ def hub_unregister(token: str) -> None: click.echo(f"unregistered {removed.name} ({removed.kb_id})") +@cli.command("subscribe") +@click.argument("kb_ref") +@click.option( + "--trust-level", + default=subscriptions_mod.DEFAULT_TRUST_LEVEL, + show_default=True, + type=click.Choice(subscriptions_mod.TRUST_LEVELS), + help="How federated hits from this KB are labeled downstream.", +) +def subscribe_cmd(kb_ref: str, trust_level: str) -> None: + """Subscribe to another KB's approved knowledge, read-only.""" + store = _load_store() + with _cli_errors(): + try: + sub = subscriptions_mod.subscribe(store, kb_ref, trust_level=trust_level) + except subscriptions_mod.SubscriptionError as e: + raise click.ClickException(str(e)) from e + click.echo(f"subscribed to {sub.name} ({sub.kb_id}) trust={sub.trust_level}") + + +@cli.command("unsubscribe") +@click.argument("kb_ref") +def unsubscribe_cmd(kb_ref: str) -> None: + """Stop federating search/context with a subscribed KB.""" + store = _load_store() + with _cli_errors(): + removed = subscriptions_mod.unsubscribe(store, kb_ref) + if not removed: + raise click.ClickException(f"no subscription matches {kb_ref!r}") + click.echo(f"unsubscribed from {kb_ref}") + + +@cli.group(name="subscriptions") +def subscriptions_group() -> None: + """Read-only KBs federated into this KB's search and context.""" + + +@subscriptions_group.command("list") +@click.option("--json", "as_json", is_flag=True, help="Emit subscriptions as JSON.") +def subscriptions_list(as_json: bool) -> None: + """List this KB's subscriptions.""" + store = _load_store() + with _cli_errors(): + subs = subscriptions_mod.list_subscriptions(store) + if as_json: + _emit_json({"subscriptions": [s.to_dict() for s in subs]}) + return + if not subs: + click.echo("no subscriptions") + return + for s in subs: + click.echo(f"{s.name} [{s.trust_level}] {s.path} ({s.kb_id})") + + def _init_personal_kb(fallback: bool | None) -> Path: """Create + register the personal catch-all KB; shared by the two entry points (`vouch hub init-personal` and `install-mcp --global`'s opt-in) diff --git a/src/vouch/context.py b/src/vouch/context.py index f6d1c315..a7a638a4 100644 --- a/src/vouch/context.py +++ b/src/vouch/context.py @@ -22,6 +22,7 @@ from . import graph, hot_memory, index_db, retrieval_events from . import pins as pins_mod from . import strategy as strategy_mod +from . import subscriptions as subscriptions_mod from .config_coerce import coerce_bool from .embeddings.fusion import rrf_fuse from .models import ( @@ -481,6 +482,58 @@ def _retrieve( return [(k, i, s, sc, "substring") for k, i, s, sc in filtered] +def _add_federated_hits( + store: KBStore, + local_hits: list[dict[str, Any]], + *, + query: str, + limit: int, + backend: str | None, + min_score: float, +) -> list[dict[str, Any]]: + """Merge in read-only hits from subscribed KBs, budget-capped. + + Federated hits only ever fill leftover slots after local hits -- the + budget knob caps how many slots subscriptions may use, it never lets a + subscription displace an already-ranked local hit. A large or noisy + subscribed KB therefore cannot crowd out local knowledge; at worst it + fills the remainder of the pack. + """ + subs = subscriptions_mod.list_subscriptions(store) + if not subs: + return local_hits + slack = limit - len(local_hits) + if slack <= 0: + return local_hits + cfg = subscriptions_mod.load_subscriptions_config(store) + budget = max(0, min(slack, round(limit * cfg.budget_share))) + if budget <= 0: + return local_hits + + federated: list[dict[str, Any]] = [] + for sub in subs: + sub_store = subscriptions_mod.open_subscribed_store(sub) + if sub_store is None: + continue # moved/deleted since subscribing -- contribute nothing + sub_result = search_kb( + sub_store, query=query, limit=budget, backend=backend, + min_score=min_score, _federate=False, + ) + for hit in sub_result.get("hits", []): + federated.append( + { + **hit, + "id": f"{sub.kb_id}:{hit['id']}", + "origin_kb_id": sub.kb_id, + "origin_kb_name": sub.name, + "trust_level": sub.trust_level, + "federated": True, + } + ) + federated.sort(key=lambda h: h["score"], reverse=True) + return local_hits + federated[:budget] + + def search_kb( store: KBStore, *, @@ -490,6 +543,7 @@ def search_kb( min_score: float = 0.0, project: str | None = None, agent: str | None = None, + _federate: bool = True, ) -> dict[str, Any]: """The one `kb.search` implementation every surface delegates to. @@ -580,6 +634,17 @@ def search_kb( "viewer": {"project": viewer.project, "agent": viewer.agent}, "hits": hits_list, } + # Federate: subscribed KBs are queried alongside the local one and + # merged in, read-only. One hop only -- a federated call never itself + # federates (subscriptions_mod never reads a foreign KB's own + # subscriptions.json either, so this flag is belt-and-suspenders). + if _federate: + hits_list = _add_federated_hits( + store, hits_list, query=query, limit=limit, backend=backend, + min_score=min_score, + ) + result["hits"] = hits_list + # The single search path serves both agent-facing surfaces (MCP + JSONL), # so the hot-memory sidebar (#261) is attached here rather than duplicated # at each call site. @@ -762,6 +827,58 @@ def _origin_from_tags(tags: list[str]) -> str | None: return None +def _federated_context_items( + store: KBStore, query: str, viewer: ViewerContext, budget: int, +) -> list[ContextItem]: + """Read-only ContextItems from every subscribed KB, budget-capped. + + Mirrors the per-hit processing in build_context_pack (retracted-claim + and dead-page filtering) but against each subscription's own store, and + scoped by that store's own viewer/config -- a subscription is queried on + its own terms, same as the local KB is. Always tagged with the + subscribing KB's kb_id and trust_level; a federated hit's own origin tag + (if any) is ignored, since subscriptions are one hop only. + """ + subs = subscriptions_mod.list_subscriptions(store) + if not subs: + return [] + federated: list[ContextItem] = [] + for sub in subs: + sub_store = subscriptions_mod.open_subscribed_store(sub) + if sub_store is None: + continue # moved/deleted since subscribing -- contributes nothing + sub_viewer = viewer_from(config_path=sub_store.config_path) + hits = _retrieve(sub_store, query, budget, sub_viewer) + for kind, hid, summary, score, backend in hits: + cites: list[str] = [] + if kind == "claim": + try: + claim = sub_store.get_claim(hid) + except ArtifactNotFoundError: + continue + if claim.status in _RETRACTED_CLAIM_STATUSES: + continue + cites = list(claim.evidence) + elif kind == "page" and not _page_is_live(sub_store, hid): + continue + summary = _enrich_summary(sub_store, kind, hid, summary) + federated.append( + ContextItem( + id=f"{sub.kb_id}:{hid}", + type=cast(ContextItemKind, kind), + summary=summary, + score=score, + backend=backend, + citations=cites, + freshness="unknown", + origin=sub.kb_id, + trust_level=sub.trust_level, + ) + ) + federated.sort(key=lambda i: i.score, reverse=True) + return federated[:budget] + + def build_context_pack( store: KBStore, *, @@ -837,6 +954,16 @@ def build_context_pack( ) ) + # Federate: subscribed KBs fill only the leftover slots after local + # hits, capped by the budget knob -- they never displace a ranked local + # item, only occupy what's unused. + slack = limit - len(items) + if slack > 0: + cfg = subscriptions_mod.load_subscriptions_config(store) + fed_budget = max(0, min(slack, round(limit * cfg.budget_share))) + if fed_budget > 0: + items.extend(_federated_context_items(store, query, viewer, fed_budget)) + items = _dedupe_near_duplicates(items) # Pins go in front of everything retrieval chose (#615): the working set is diff --git a/src/vouch/models.py b/src/vouch/models.py index 2118602b..a8b80428 100644 --- a/src/vouch/models.py +++ b/src/vouch/models.py @@ -532,6 +532,13 @@ class ContextItem(BaseModel): "gated federation import (from the claim's origin: tag). None for " "locally-authored knowledge.", ) + trust_level: str | None = Field( + default=None, + description="vouch: the subscribed KB's trust level ('unverified' or " + "'trusted'), set only for hits federated in via a read-only " + "subscription. None for local knowledge and for gated-import origin " + "hits (those aren't federated, they're locally-owned copies).", + ) class ContextQuality(BaseModel): diff --git a/src/vouch/subscriptions.py b/src/vouch/subscriptions.py new file mode 100644 index 00000000..c7a3e552 --- /dev/null +++ b/src/vouch/subscriptions.py @@ -0,0 +1,219 @@ +"""Read-only KB subscriptions -- federate search without copying (issue #610). + +`vouch hub` export/import (#536) moves artifacts through the receiving gate +and they become local, forked copies -- correct for "make this mine", wrong +for "let me read yours" (duplicates, drifts independently, forces re-review +on every upstream edit). Subscription is the safer, ditto-style primitive +(heyditto.ai/docs/knowledge-graph-sharing): a subscribed KB's approved +knowledge joins local search/context results, live, read-only, never +copied, never locally proposable/approvable. + +Read-only is mostly free by construction, not a separate enforced gate: a +federated hit's id is namespaced ":" and never resolves +against the local store, so store.get_claim()/propose()/approve() on a +federated id simply finds nothing local to act on. Wanting to *own* a +federated fact means importing it through the existing gated hub path, +unchanged. + +One hop only: a subscribed KB's own subscriptions.json is never read (this +module never recurses into a foreign KB's subscriptions). + +Storage: subscriptions.json, a small committed list (mirrors the hub +registry's shape, but per-KB rather than per-machine) -- additive, no +schema change to stored artifacts. The origin/trust tag lives on the +*result*, never the artifact. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +import yaml + +from . import hub +from .storage import KBStore + +logger = logging.getLogger(__name__) + +FILENAME = "subscriptions.json" +TRUST_LEVELS = ("unverified", "trusted") +DEFAULT_TRUST_LEVEL = "unverified" +DEFAULT_BUDGET_SHARE = 0.3 + + +class SubscriptionError(Exception): + """Raised for invalid subscription operations (unresolvable ref, self- + subscribe, duplicate subscription).""" + + +@dataclass(frozen=True) +class Subscription: + kb_id: str + path: str + name: str + trust_level: str + subscribed_at: str + + def to_dict(self) -> dict[str, Any]: + return { + "kb_id": self.kb_id, + "path": self.path, + "name": self.name, + "trust_level": self.trust_level, + "subscribed_at": self.subscribed_at, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Subscription: + return cls( + kb_id=str(data["kb_id"]), + path=str(data["path"]), + name=str(data.get("name") or data["kb_id"]), + trust_level=str(data.get("trust_level") or DEFAULT_TRUST_LEVEL), + subscribed_at=str(data.get("subscribed_at", "")), + ) + + +@dataclass(frozen=True) +class SubscriptionsConfig: + budget_share: float = DEFAULT_BUDGET_SHARE + + +def load_subscriptions_config(store: KBStore) -> SubscriptionsConfig: + """Read ``retrieval.subscriptions`` from config.yaml; fall back to defaults.""" + try: + loaded = yaml.safe_load(store.config_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError): + return SubscriptionsConfig() + if not isinstance(loaded, dict): + return SubscriptionsConfig() + retrieval = loaded.get("retrieval") + raw = retrieval.get("subscriptions") if isinstance(retrieval, dict) else None + if not isinstance(raw, dict): + return SubscriptionsConfig() + try: + budget_share = float(raw.get("budget_share", DEFAULT_BUDGET_SHARE)) + except (TypeError, ValueError): + budget_share = DEFAULT_BUDGET_SHARE + if not (0.0 <= budget_share <= 1.0): + budget_share = DEFAULT_BUDGET_SHARE + return SubscriptionsConfig(budget_share=budget_share) + + +def _subscriptions_path(store: KBStore) -> Path: + return store.kb_dir / FILENAME + + +def _read_subscriptions(store: KBStore) -> list[Subscription]: + path = _subscriptions_path(store) + if not path.exists(): + return [] + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return [] + if not isinstance(data, list): + return [] + out: list[Subscription] = [] + for entry in data: + if not isinstance(entry, dict): + continue + try: + out.append(Subscription.from_dict(entry)) + except (KeyError, ValueError): + continue + return out + + +def _write_subscriptions(store: KBStore, subs: list[Subscription]) -> None: + path = _subscriptions_path(store) + path.write_text( + json.dumps([s.to_dict() for s in subs], indent=2) + "\n", encoding="utf-8" + ) + + +def _resolve_ref(ref: str) -> tuple[str, Path] | None: + """A registered kb_id (vouch hub registry), or a filesystem path, to + (kb_id, root). None if neither resolves to a KB that exists on disk.""" + for entry in hub.load_registry(): + if entry.kb_id == ref: + root = Path(entry.path) + return (entry.kb_id, root) if (root / ".vouch").is_dir() else None + root = Path(ref).expanduser().resolve() + if not (root / ".vouch").is_dir(): + return None + identity = KBStore(root).identity() + if identity is None: + return None + kb_id, _name = identity + return kb_id, root + + +def subscribe( + store: KBStore, kb_ref: str, *, trust_level: str = DEFAULT_TRUST_LEVEL +) -> Subscription: + """Subscribe to another KB's approved knowledge, read-only. + + kb_ref is a registered kb_id (vouch hub registry) or a filesystem path. + Raises SubscriptionError if it doesn't resolve to an existing KB, is + this KB's own identity, or is already subscribed. + """ + if trust_level not in TRUST_LEVELS: + raise SubscriptionError( + f"unknown trust level {trust_level!r} (use {'/'.join(TRUST_LEVELS)})" + ) + resolved = _resolve_ref(kb_ref) + if resolved is None: + raise SubscriptionError(f"no KB found at {kb_ref!r}") + kb_id, root = resolved + + own_identity = store.identity() + if own_identity is not None and own_identity[0] == kb_id: + raise SubscriptionError("cannot subscribe to this KB's own identity") + + existing = _read_subscriptions(store) + if any(s.kb_id == kb_id for s in existing): + raise SubscriptionError(f"already subscribed to {kb_id!r}") + + foreign_identity = KBStore(root).identity() + name = foreign_identity[1] if foreign_identity else root.name + + sub = Subscription( + kb_id=kb_id, + path=str(root), + name=name, + trust_level=trust_level, + subscribed_at=datetime.now(UTC).isoformat(), + ) + existing.append(sub) + _write_subscriptions(store, existing) + return sub + + +def unsubscribe(store: KBStore, kb_ref: str) -> bool: + """Remove a subscription by kb_id or path. Returns True if found.""" + existing = _read_subscriptions(store) + filtered = [s for s in existing if s.kb_id != kb_ref and s.path != kb_ref] + if len(filtered) == len(existing): + return False + _write_subscriptions(store, filtered) + return True + + +def list_subscriptions(store: KBStore) -> list[Subscription]: + return _read_subscriptions(store) + + +def open_subscribed_store(sub: Subscription) -> KBStore | None: + """Open a subscribed KB read-only, or None if it's no longer there + (moved/deleted since subscribing) -- a dead subscription degrades to + silently contributing nothing, not an error.""" + root = Path(sub.path) + if not (root / ".vouch").is_dir(): + return None + return KBStore(root) diff --git a/tests/test_cli_maintenance.py b/tests/test_cli_maintenance.py index e7107891..1a704d98 100644 --- a/tests/test_cli_maintenance.py +++ b/tests/test_cli_maintenance.py @@ -11,7 +11,6 @@ import json from pathlib import Path -import numpy as np import pytest from click.testing import CliRunner, Result @@ -22,6 +21,8 @@ from vouch.proposals import propose_claim from vouch.storage import KBStore +np = pytest.importorskip("numpy") + class _HashEmbedder(Embedder): name = "mock" diff --git a/tests/test_index_db_embeddings.py b/tests/test_index_db_embeddings.py index 16f9b09c..08cafda6 100644 --- a/tests/test_index_db_embeddings.py +++ b/tests/test_index_db_embeddings.py @@ -13,7 +13,6 @@ import sys from pathlib import Path -import numpy as np import pytest from vouch import index_db @@ -22,6 +21,8 @@ from vouch.models import Claim, Entity, Page from vouch.storage import KBStore +np = pytest.importorskip("numpy") + class _HashEmbedder(Embedder): name = "mock" diff --git a/tests/test_jsonl_server_surface.py b/tests/test_jsonl_server_surface.py index 814f78a7..7767997e 100644 --- a/tests/test_jsonl_server_surface.py +++ b/tests/test_jsonl_server_surface.py @@ -12,7 +12,6 @@ from pathlib import Path from typing import Any -import numpy as np import pytest from vouch import bundle @@ -22,6 +21,8 @@ from vouch.models import Claim, Entity, Page from vouch.storage import KBStore +np = pytest.importorskip("numpy") + class _HashEmbedder(Embedder): name = "mock" diff --git a/tests/test_server_tool_surface.py b/tests/test_server_tool_surface.py index 66abd77d..b61200ec 100644 --- a/tests/test_server_tool_surface.py +++ b/tests/test_server_tool_surface.py @@ -14,7 +14,6 @@ from pathlib import Path from typing import Any -import numpy as np import pytest from vouch import server @@ -24,6 +23,8 @@ from vouch.proposals import propose_claim from vouch.storage import KBStore +np = pytest.importorskip("numpy") + class _HashEmbedder(Embedder): name = "mock" diff --git a/tests/test_subscriptions.py b/tests/test_subscriptions.py new file mode 100644 index 00000000..8984235e --- /dev/null +++ b/tests/test_subscriptions.py @@ -0,0 +1,196 @@ +"""Read-only KB subscriptions (#610) — federate search/context without copying.""" + +from __future__ import annotations + +import shutil +from pathlib import Path + +import pytest +import yaml + +from vouch import health, hub, subscriptions +from vouch.context import build_context_pack, search_kb +from vouch.models import Claim +from vouch.storage import KBStore + + +@pytest.fixture(autouse=True) +def _isolated_machine(tmp_path_factory, monkeypatch): + """Fake $HOME so subscribe's registry lookup never touches the real machine.""" + fake_home = tmp_path_factory.mktemp("home") + monkeypatch.setattr(Path, "home", classmethod(lambda cls: fake_home)) + monkeypatch.setenv(hub.REGISTRY_ENV, str(fake_home / "registry.yaml")) + monkeypatch.delenv("VOUCH_KB_PATH", raising=False) + monkeypatch.delenv("VOUCH_PROJECT_DIR", raising=False) + return fake_home + + +@pytest.fixture +def store(tmp_path: Path) -> KBStore: + return KBStore.init(tmp_path / "local") + + +def _kb_with_claim(root: Path, *, claim_id: str, text: str) -> KBStore: + kb = KBStore.init(root) + src = kb.put_source(text.encode(), title="doc") + kb.put_claim(Claim(id=claim_id, text=text, evidence=[src.id])) + health.rebuild_index(kb) + return kb + + +# --- subscribe / unsubscribe / list ---------------------------------------- + + +def test_subscribe_by_path(store: KBStore, tmp_path: Path) -> None: + foreign = _kb_with_claim(tmp_path / "foreign", claim_id="f1", text="foreign fact") + sub = subscriptions.subscribe(store, str(tmp_path / "foreign")) + assert sub.kb_id == foreign.identity()[0] + assert sub.trust_level == "unverified" + listed = subscriptions.list_subscriptions(store) + assert [s.kb_id for s in listed] == [sub.kb_id] + + +def test_subscribe_rejects_own_identity(store: KBStore) -> None: + with pytest.raises(subscriptions.SubscriptionError): + subscriptions.subscribe(store, str(store.kb_dir.parent)) + + +def test_subscribe_rejects_duplicate(store: KBStore, tmp_path: Path) -> None: + _kb_with_claim(tmp_path / "foreign", claim_id="f1", text="foreign fact") + subscriptions.subscribe(store, str(tmp_path / "foreign")) + with pytest.raises(subscriptions.SubscriptionError): + subscriptions.subscribe(store, str(tmp_path / "foreign")) + + +def test_subscribe_rejects_nonexistent_kb(store: KBStore, tmp_path: Path) -> None: + with pytest.raises(subscriptions.SubscriptionError): + subscriptions.subscribe(store, str(tmp_path / "nowhere")) + + +def test_subscribe_rejects_unknown_trust_level(store: KBStore, tmp_path: Path) -> None: + _kb_with_claim(tmp_path / "foreign", claim_id="f1", text="foreign fact") + with pytest.raises(subscriptions.SubscriptionError): + subscriptions.subscribe(store, str(tmp_path / "foreign"), trust_level="bogus") + + +def test_unsubscribe_removes_entry(store: KBStore, tmp_path: Path) -> None: + foreign = _kb_with_claim(tmp_path / "foreign", claim_id="f1", text="foreign fact") + subscriptions.subscribe(store, str(tmp_path / "foreign")) + assert subscriptions.unsubscribe(store, foreign.identity()[0]) is True + assert subscriptions.list_subscriptions(store) == [] + + +def test_unsubscribe_missing_returns_false(store: KBStore) -> None: + assert subscriptions.unsubscribe(store, "nope") is False + + +def test_open_subscribed_store_returns_none_when_moved( + store: KBStore, tmp_path: Path, +) -> None: + foreign_root = tmp_path / "foreign" + _kb_with_claim(foreign_root, claim_id="f1", text="foreign fact") + sub = subscriptions.subscribe(store, str(foreign_root)) + shutil.rmtree(foreign_root) + assert subscriptions.open_subscribed_store(sub) is None + + +# --- federation: search ----------------------------------------------------- + + +def test_search_kb_includes_federated_hit_when_local_has_slack( + store: KBStore, tmp_path: Path, +) -> None: + foreign = _kb_with_claim( + tmp_path / "foreign", claim_id="f1", text="rust ownership rules", + ) + subscriptions.subscribe(store, str(tmp_path / "foreign")) + health.rebuild_index(store) + result = search_kb(store, query="rust ownership", limit=10) + federated = [h for h in result["hits"] if h.get("federated")] + assert len(federated) == 1 + assert federated[0]["origin_kb_id"] == foreign.identity()[0] + assert federated[0]["trust_level"] == "unverified" + assert federated[0]["id"] == f"{foreign.identity()[0]}:f1" + + +def test_search_kb_federated_hits_never_displace_local( + store: KBStore, tmp_path: Path, +) -> None: + for i in range(10): + src = store.put_source(f"e{i}".encode()) + store.put_claim( + Claim(id=f"local{i}", text=f"local claim about widgets {i}", evidence=[src.id]), + ) + health.rebuild_index(store) + _kb_with_claim(tmp_path / "foreign", claim_id="f1", text="widgets foreign fact") + subscriptions.subscribe(store, str(tmp_path / "foreign")) + result = search_kb(store, query="widgets", limit=10) + assert len(result["hits"]) == 10 # local already fills the whole limit + assert not any(h.get("federated") for h in result["hits"]) + + +def test_search_kb_federation_is_one_hop(store: KBStore, tmp_path: Path) -> None: + # store -> subscribes to B -> subscribes to C. Searching store must not + # surface C's data. + _kb_with_claim(tmp_path / "c", claim_id="c1", text="deep transitive fact") + kb_b = KBStore.init(tmp_path / "b") + subscriptions.subscribe(kb_b, str(tmp_path / "c")) + subscriptions.subscribe(store, str(tmp_path / "b")) + health.rebuild_index(store) + result = search_kb(store, query="deep transitive", limit=10) + assert result["hits"] == [] + + +def test_search_kb_respects_budget_share(store: KBStore, tmp_path: Path) -> None: + # Empty local KB -> all 10 slots are slack; budget_share caps how many + # federation may actually take. + cfg = yaml.safe_load(store.config_path.read_text(encoding="utf-8")) + cfg.setdefault("retrieval", {})["subscriptions"] = {"budget_share": 0.2} + store.config_path.write_text(yaml.safe_dump(cfg), encoding="utf-8") + + foreign = KBStore.init(tmp_path / "foreign") + for i in range(10): + src = foreign.put_source(f"e{i}".encode()) + foreign.put_claim( + Claim(id=f"f{i}", text=f"gadget fact number {i}", evidence=[src.id]), + ) + health.rebuild_index(foreign) + subscriptions.subscribe(store, str(tmp_path / "foreign")) + health.rebuild_index(store) + + result = search_kb(store, query="gadget", limit=10) + federated = [h for h in result["hits"] if h.get("federated")] + assert len(federated) == 2 # round(10 * 0.2) + + +def test_dead_subscription_contributes_nothing(store: KBStore, tmp_path: Path) -> None: + foreign_root = tmp_path / "foreign" + _kb_with_claim(foreign_root, claim_id="f1", text="soon to vanish fact") + subscriptions.subscribe(store, str(foreign_root)) + shutil.rmtree(foreign_root) + health.rebuild_index(store) + result = search_kb(store, query="vanish", limit=10) + assert result["hits"] == [] + + +# --- federation: context pack ----------------------------------------------- + + +def test_build_context_pack_tags_federated_item_origin_and_trust( + store: KBStore, tmp_path: Path, +) -> None: + foreign = _kb_with_claim( + tmp_path / "foreign", claim_id="f1", text="graph databases model relations", + ) + subscriptions.subscribe(store, str(tmp_path / "foreign"), trust_level="trusted") + health.rebuild_index(store) + pack = build_context_pack(store, query="graph databases", limit=10) + items = pack["items"] if isinstance(pack, dict) else pack.items + fed = [ + i for i in items + if (i.get("origin") if isinstance(i, dict) else i.origin) == foreign.identity()[0] + ] + assert len(fed) == 1 + fed_item = fed[0] + trust = fed_item.get("trust_level") if isinstance(fed_item, dict) else fed_item.trust_level + assert trust == "trusted"