From 3c88675c8b0e52ce412004744f45d90aebd3b0a7 Mon Sep 17 00:00:00 2001 From: dale053 Date: Wed, 8 Jul 2026 10:50:19 -0400 Subject: [PATCH] feat(eval): add kb effectiveness outcome scoring --- CHANGELOG.md | 12 ++ docs/effectiveness.md | 34 ++++ src/vouch/capabilities.py | 1 + src/vouch/cli.py | 60 +++++++ src/vouch/context.py | 15 ++ src/vouch/eval/__init__.py | 2 + src/vouch/eval/effectiveness.py | 260 +++++++++++++++++++++++++++ src/vouch/index_db.py | 69 +++++++ src/vouch/jsonl_server.py | 11 ++ src/vouch/openclaw/context_engine.py | 1 + src/vouch/server.py | 15 ++ tests/test_effectiveness.py | 150 ++++++++++++++++ 12 files changed, 630 insertions(+) create mode 100644 docs/effectiveness.md create mode 100644 src/vouch/eval/effectiveness.py create mode 100644 tests/test_effectiveness.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 523eb817..00bbc880 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,18 @@ All notable changes to vouch are documented here. Format follows ## [Unreleased] +### Added +- `vouch health effectiveness` and `kb.effectiveness`: a read-only + per-artifact outcome signal that correlates context-pack surfacing with + session outcomes over a time window. Reports surfaced good/bad counts, + lift vs baseline, 95% wilson interval, and conservative verdicts + (`useful`/`harmful` only when confidence clears baseline and sample size is + sufficient; otherwise `unverified`/`insufficient`). +- derived `context_surface` cache in `.vouch/state.db`, populated from + `build_context_pack` and cleared by `index_db.reset`, to record which + artifacts were surfaced to which sessions without changing durable yaml/audit + authority. + ## [1.1.0] — 2026-07-03 ### Added diff --git a/docs/effectiveness.md b/docs/effectiveness.md new file mode 100644 index 00000000..79f8aedc --- /dev/null +++ b/docs/effectiveness.md @@ -0,0 +1,34 @@ +# Effectiveness schema + +`vouch health effectiveness --format json` and `kb.effectiveness` share one +stable machine-readable schema. + +## Top-level fields + +- `schema_version` (int): schema revision for compatibility checks. +- `window` (object): `{spec, since, until, generated_at}`. +- `min_samples` (int): minimum surfaced samples required for confident verdicts. +- `sessions` (object): `{classified, good, bad, baseline_rate}`. +- `artifacts` (array): ranked per-artifact rows. + +## Artifact row + +- `artifact_kind` (string) +- `artifact_id` (string) +- `samples` (int): surfaced sessions with classified outcomes. +- `surfaced` (object): `{good, bad}` counts for surfaced sessions. +- `not_surfaced` (object): `{good, bad}` counts for classified sessions where + this artifact was not surfaced. +- `rate` (float): surfaced good rate. +- `baseline_rate` (float): global good rate across classified sessions. +- `lift` (float): `rate - baseline_rate`. +- `wilson_95` (object): `{low, high}` confidence interval on surfaced good rate. +- `verdict` (string): one of `useful`, `harmful`, `unverified`, `insufficient`. +- `earned_value` (float): `lift * samples`, used for ranking. + +## Verdict rules + +- `insufficient`: `samples < min_samples` +- `useful`: interval lower bound is greater than baseline +- `harmful`: interval upper bound is lower than baseline +- `unverified`: otherwise diff --git a/src/vouch/capabilities.py b/src/vouch/capabilities.py index 2efc39a3..92cff511 100644 --- a/src/vouch/capabilities.py +++ b/src/vouch/capabilities.py @@ -21,6 +21,7 @@ "kb.search", "kb.neighbors", "kb.context", + "kb.effectiveness", "kb.synthesize", "kb.read_page", "kb.read_claim", diff --git a/src/vouch/cli.py b/src/vouch/cli.py index f8b4a0c8..911cf3b2 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -40,6 +40,7 @@ from . import verify as verify_mod from .capabilities import capabilities as build_caps from .context import build_context_pack +from .eval.effectiveness import compute_effectiveness from .lifecycle import LifecycleError from .logging_config import configure_logging from .models import Proposal, ProposalKind, ProposalStatus @@ -320,6 +321,65 @@ def fsck() -> None: sys.exit(0 if report.ok else 1) +@cli.group(name="health") +def health_group() -> None: + """Health-oriented diagnostics and evaluation commands.""" + + +@health_group.command("effectiveness") +@click.option( + "--window", + default="90d", + show_default=True, + help="Time window (e.g. 90d, 12h, 2w, or ISO date).", +) +@click.option( + "--min-samples", + default=5, + show_default=True, + type=int, + help="Minimum surfaced sessions before a confident verdict is allowed.", +) +@click.option( + "--format", + "output_format", + type=click.Choice(["text", "json"]), + default="text", + show_default=True, +) +def health_effectiveness(window: str, min_samples: int, output_format: str) -> None: + """Measure whether surfaced artifacts correlate with better outcomes.""" + store = _load_store() + with _cli_errors(): + report = compute_effectiveness( + store, + window=window, + min_samples=min_samples, + ) + if output_format == "json": + _emit_json(report) + return + + sessions = report["sessions"] + base = sessions["baseline_rate"] + base_str = "n/a" if base is None else f"{base * 100:.1f}%" + click.echo( + f"effectiveness window={report['window']['spec']} " + f"classified={sessions['classified']} baseline={base_str}" + ) + if not report["artifacts"]: + click.echo("no surfaced artifacts with enough classified sessions") + return + for row in report["artifacts"]: + lo = row["wilson_95"]["low"] + hi = row["wilson_95"]["high"] + click.echo( + f"{row['artifact_kind']}/{row['artifact_id']} " + f"verdict={row['verdict']} samples={row['samples']} " + f"lift={row['lift']:+.3f} ci95=[{lo:.3f},{hi:.3f}]" + ) + + @cli.group(invoke_without_command=True) @click.option("--check", "check_only", is_flag=True, help="Only check if a migration is needed.") @click.option("--dry-run", is_flag=True, help="Show planned format changes without writing.") diff --git a/src/vouch/context.py b/src/vouch/context.py index 6e9b08ce..4535db51 100644 --- a/src/vouch/context.py +++ b/src/vouch/context.py @@ -214,6 +214,7 @@ def build_context_pack( graph_depth: int = 1, graph_limit: int = 20, graph_rel_types: list[str] | None = None, + session_id: str | None = None, ) -> ContextPack | dict[str, Any]: viewer = viewer_from( config_path=store.config_path, @@ -305,6 +306,20 @@ def build_context_pack( ) pack = ContextPack(query=query, items=items, quality=quality, warnings=warnings) + if items: + try: + with index_db.open_db(store.kb_dir) as conn: + index_db.index_context_surface( + conn, + session_id=session_id, + query=query, + surfaced_at=pack.generated_at.isoformat(), + items=((item.type, item.id) for item in items), + ) + except sqlite3.Error: + # The surfacing log is derived cache only; retrieval must still work + # if state.db is unavailable or stale. + pass result: dict[str, Any] = pack.model_dump() result["viewer"] = { "project": viewer.project, diff --git a/src/vouch/eval/__init__.py b/src/vouch/eval/__init__.py index de519c17..ec637a3c 100644 --- a/src/vouch/eval/__init__.py +++ b/src/vouch/eval/__init__.py @@ -13,10 +13,12 @@ run_recall, score_query, ) +from .effectiveness import compute_effectiveness __all__ = [ "compare_baseline", "load_queries", "run_recall", "score_query", + "compute_effectiveness", ] diff --git a/src/vouch/eval/effectiveness.py b/src/vouch/eval/effectiveness.py new file mode 100644 index 00000000..0b79ae97 --- /dev/null +++ b/src/vouch/eval/effectiveness.py @@ -0,0 +1,260 @@ +"""Per-artifact effectiveness signal from surfaced context vs session outcomes. + +Read-only evaluation: +- surfaced artifacts come from the derived `context_surface` cache in state.db +- session outcomes come from append-only audit events + +This module never mutates durable KB state. +""" + +from __future__ import annotations + +import re +from collections import defaultdict +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +from .. import index_db, metrics +from ..audit import read_events +from ..models import AuditEvent +from ..storage import KBStore + +SCHEMA_VERSION = 1 +_Z_95 = 1.96 +_REJECT_RE = re.compile(r"^proposal\.[a-z]+\.reject$") +_GOOD_EVENTS = frozenset({"claim.confirm"}) +_BAD_EVENTS = frozenset({"claim.contradict"}) + + +@dataclass(frozen=True) +class SessionWindow: + session_id: str + actor: str + started_at: datetime + ended_at: datetime | None + + +def _as_utc(dt: datetime) -> datetime: + return dt if dt.tzinfo is not None else dt.replace(tzinfo=UTC) + + +def _in_window(ts: datetime, since: datetime | None, until: datetime | None) -> bool: + if since is not None and ts < since: + return False + return until is None or ts <= until + + +def _window_contains(win: SessionWindow, ts: datetime) -> bool: + if ts < win.started_at: + return False + if win.ended_at is None: + return True + return ts <= win.ended_at + + +def _wilson_95(successes: int, total: int) -> tuple[float, float]: + if total <= 0: + return (0.0, 0.0) + p = successes / total + z2 = _Z_95 ** 2 + denom = 1 + (z2 / total) + center = (p + (z2 / (2 * total))) / denom + margin = ( + _Z_95 + * (((p * (1 - p)) / total + (z2 / (4 * total * total))) ** 0.5) + / denom + ) + return (max(0.0, center - margin), min(1.0, center + margin)) + + +def _build_sessions(events: list[AuditEvent], *, now: datetime) -> dict[str, SessionWindow]: + sessions: dict[str, SessionWindow] = {} + ends: dict[str, datetime] = {} + for ev in events: + if not ev.object_ids: + continue + sid = ev.object_ids[0] + ts = _as_utc(ev.created_at) + if ev.event == "session.start": + sessions[sid] = SessionWindow( + session_id=sid, + actor=ev.actor, + started_at=ts, + ended_at=None, + ) + elif ev.event == "session.end": + ends[sid] = ts + for sid, end_ts in ends.items(): + win = sessions.get(sid) + if win is None: + continue + sessions[sid] = SessionWindow( + session_id=win.session_id, + actor=win.actor, + started_at=win.started_at, + ended_at=end_ts, + ) + # Keep deterministic "still open" end as None; consumers can use now. + _ = now + return sessions + + +def _session_for_event( + event: AuditEvent, + *, + sessions: dict[str, SessionWindow], + by_actor: dict[str, list[SessionWindow]], + proposal_to_session: dict[str, str], +) -> str | None: + if event.event.startswith("session."): + return None + # proposal..reject points at proposal id in object_ids[0]. + if _REJECT_RE.match(event.event) and event.object_ids: + sid = proposal_to_session.get(event.object_ids[0]) + if sid is not None: + return sid + # If the event explicitly references a session id, trust it. + for oid in event.object_ids: + if oid.startswith("sess-") and oid in sessions: + return oid + # Otherwise, infer from actor + timestamp overlap. + ts = _as_utc(event.created_at) + for win in by_actor.get(event.actor, []): + if _window_contains(win, ts): + return win.session_id + return None + + +def compute_effectiveness( + store: KBStore, + *, + window: str = "90d", + min_samples: int = 5, + now: datetime | None = None, +) -> dict[str, Any]: + """Return per-artifact outcome association with conservative verdicts.""" + if min_samples < 1: + raise metrics.MetricsError("--min-samples must be >= 1") + now_utc = _as_utc(now or datetime.now(UTC)) + since = metrics.parse_since(window, now=now_utc) + until = now_utc + + events = list(read_events(store.kb_dir)) + sessions = _build_sessions(events, now=now_utc) + by_actor: dict[str, list[SessionWindow]] = defaultdict(list) + for s in sessions.values(): + by_actor[s.actor].append(s) + for rows in by_actor.values(): + rows.sort(key=lambda w: w.started_at) + + proposal_to_session = { + p.id: p.session_id + for p in store.list_proposals() + if p.session_id + } + + session_outcome: dict[str, str] = {} + activity: dict[str, set[str]] = defaultdict(set) + for ev in events: + ts = _as_utc(ev.created_at) + if not _in_window(ts, since, until): + continue + sid = _session_for_event( + ev, + sessions=sessions, + by_actor=by_actor, + proposal_to_session=proposal_to_session, + ) + if sid is None: + continue + if ev.event in _GOOD_EVENTS: + activity[sid].add("good") + elif ev.event in _BAD_EVENTS or _REJECT_RE.match(ev.event): + activity[sid].add("bad") + + for sid, marks in activity.items(): + if marks == {"good"}: + session_outcome[sid] = "good" + elif marks == {"bad"}: + session_outcome[sid] = "bad" + + classified_sessions = set(session_outcome) + total_good = sum(1 for v in session_outcome.values() if v == "good") + total_bad = sum(1 for v in session_outcome.values() if v == "bad") + baseline_total = total_good + total_bad + baseline_rate = (total_good / baseline_total) if baseline_total else None + + surfaced_rows = index_db.read_context_surfaces( + store.kb_dir, + since=since, + until=until, + ) + by_artifact: dict[tuple[str, str], set[str]] = defaultdict(set) + for row in surfaced_rows: + sid = row["session_id"] + if not sid or sid not in classified_sessions: + continue + by_artifact[(row["artifact_kind"], row["artifact_id"])].add(sid) + + artifacts: list[dict[str, Any]] = [] + for (kind, artifact_id), surfaced_sessions in sorted(by_artifact.items()): + surfaced_good = sum(1 for sid in surfaced_sessions if session_outcome[sid] == "good") + surfaced_bad = sum(1 for sid in surfaced_sessions if session_outcome[sid] == "bad") + surfaced_total = surfaced_good + surfaced_bad + if surfaced_total == 0 or baseline_rate is None: + continue + + exposed_rate = surfaced_good / surfaced_total + ci_low, ci_high = _wilson_95(surfaced_good, surfaced_total) + lift = exposed_rate - baseline_rate + non_exposed_good = total_good - surfaced_good + non_exposed_bad = total_bad - surfaced_bad + + if surfaced_total < min_samples: + verdict = "insufficient" + elif ci_low > baseline_rate: + verdict = "useful" + elif ci_high < baseline_rate: + verdict = "harmful" + else: + verdict = "unverified" + + artifacts.append( + { + "artifact_kind": kind, + "artifact_id": artifact_id, + "samples": surfaced_total, + "surfaced": {"good": surfaced_good, "bad": surfaced_bad}, + "not_surfaced": {"good": non_exposed_good, "bad": non_exposed_bad}, + "rate": exposed_rate, + "baseline_rate": baseline_rate, + "lift": lift, + "wilson_95": {"low": ci_low, "high": ci_high}, + "verdict": verdict, + "earned_value": lift * surfaced_total, + } + ) + + artifacts.sort( + key=lambda row: (row["earned_value"], row["samples"], row["artifact_id"]), + reverse=True, + ) + + return { + "schema_version": SCHEMA_VERSION, + "window": { + "spec": window, + "since": since.isoformat() if since else None, + "until": until.isoformat(), + "generated_at": now_utc.isoformat(), + }, + "min_samples": min_samples, + "sessions": { + "classified": baseline_total, + "good": total_good, + "bad": total_bad, + "baseline_rate": baseline_rate, + }, + "artifacts": artifacts, + } diff --git a/src/vouch/index_db.py b/src/vouch/index_db.py index 146d9f52..281f725d 100644 --- a/src/vouch/index_db.py +++ b/src/vouch/index_db.py @@ -86,6 +86,18 @@ CREATE INDEX IF NOT EXISTS prov_edges_dst ON prov_edges(dst_id); CREATE INDEX IF NOT EXISTS prov_edges_kind ON prov_edges(kind); + +CREATE TABLE IF NOT EXISTS context_surface ( + session_id TEXT, + artifact_kind TEXT NOT NULL, + artifact_id TEXT NOT NULL, + query TEXT NOT NULL DEFAULT '', + surfaced_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS context_surface_session ON context_surface(session_id); +CREATE INDEX IF NOT EXISTS context_surface_artifact ON context_surface(artifact_kind, artifact_id); +CREATE INDEX IF NOT EXISTS context_surface_time ON context_surface(surfaced_at); """ @@ -126,6 +138,7 @@ def reset(kb_dir: Path) -> None: "DELETE FROM query_embedding_cache;" "DELETE FROM embedding_dupes;" "DELETE FROM prov_edges;" + "DELETE FROM context_surface;" "DELETE FROM index_meta WHERE key LIKE 'embedding_%';" "DELETE FROM index_meta WHERE key LIKE 'prov_%';" ) @@ -214,6 +227,62 @@ def read_prov_edges(kb_dir: Path) -> list[tuple[str, str, str, str, str | None]] return [(r[0], r[1], r[2], r[3], r[4]) for r in rows] +def index_context_surface( + conn: sqlite3.Connection, + *, + session_id: str | None, + query: str, + surfaced_at: str, + items: Iterable[tuple[str, str]], +) -> None: + rows = [ + (session_id, kind, artifact_id, query, surfaced_at) + for kind, artifact_id in items + ] + if not rows: + return + conn.executemany( + "INSERT INTO context_surface " + "(session_id, artifact_kind, artifact_id, query, surfaced_at) " + "VALUES (?, ?, ?, ?, ?)", + rows, + ) + + +def read_context_surfaces( + kb_dir: Path, + *, + since: _dt.datetime | None = None, + until: _dt.datetime | None = None, +) -> list[dict[str, str | None]]: + clauses: list[str] = [] + params: list[str] = [] + if since is not None: + clauses.append("surfaced_at >= ?") + params.append(since.isoformat()) + if until is not None: + clauses.append("surfaced_at <= ?") + params.append(until.isoformat()) + where = f"WHERE {' AND '.join(clauses)}" if clauses else "" + with open_db(kb_dir) as conn: + rows = conn.execute( + "SELECT session_id, artifact_kind, artifact_id, query, surfaced_at " + f"FROM context_surface {where} " + "ORDER BY surfaced_at ASC, artifact_kind ASC, artifact_id ASC", + params, + ).fetchall() + return [ + { + "session_id": row[0], + "artifact_kind": row[1], + "artifact_id": row[2], + "query": row[3], + "surfaced_at": row[4], + } + for row in rows + ] + + def set_meta(conn: sqlite3.Connection, key: str, value: str) -> None: conn.execute( "INSERT OR REPLACE INTO index_meta (key, value) VALUES (?, ?)", diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index 5f42e16b..8fbbfc06 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -36,6 +36,7 @@ from . import verify as verify_mod from .capabilities import capabilities as build_caps from .context import build_context_pack +from .eval.effectiveness import compute_effectiveness from .logging_config import configure_logging from .models import ProposalStatus from .proposals import ( @@ -205,10 +206,19 @@ def _h_context(p: dict) -> dict: graph_depth=int(p.get("graph_depth", 1)), graph_limit=int(p.get("graph_limit", 20)), graph_rel_types=p.get("graph_rel_types"), + session_id=session_id, ) return salience_mod.attach_salience(result, store, session_id, cfg) +def _h_effectiveness(p: dict) -> dict: + return compute_effectiveness( + _store(), + window=str(p.get("window", "90d")), + min_samples=int(p.get("min_samples", 5)), + ) + + def _h_synthesize(p: dict) -> dict: return synthesize( _store(), @@ -680,6 +690,7 @@ def _h_propose_theme(p: dict) -> dict: "kb.search": _h_search, "kb.neighbors": _h_neighbors, "kb.context": _h_context, + "kb.effectiveness": _h_effectiveness, "kb.synthesize": _h_synthesize, "kb.read_page": _h_read_page, "kb.read_claim": _h_read_claim, diff --git a/src/vouch/openclaw/context_engine.py b/src/vouch/openclaw/context_engine.py index 323bfaa8..50d3e09f 100644 --- a/src/vouch/openclaw/context_engine.py +++ b/src/vouch/openclaw/context_engine.py @@ -244,6 +244,7 @@ def assemble(self, params: AssembleParams | dict[str, Any]) -> AssembleResult: require_citations=False, project=project, agent=agent, + session_id=session_id, ) salience_mod.attach_salience(pack, store, session_id, cfg) salience = (pack.get("_meta") or {}).get("vouch_salience") or [] diff --git a/src/vouch/server.py b/src/vouch/server.py index 81fa5c23..5bc4ed6d 100644 --- a/src/vouch/server.py +++ b/src/vouch/server.py @@ -27,6 +27,7 @@ from . import verify as verify_mod from .capabilities import capabilities as build_caps from .context import build_context_pack +from .eval.effectiveness import compute_effectiveness from .logging_config import configure_logging from .models import ProposalStatus from .proposals import ( @@ -230,10 +231,24 @@ def kb_context( min_items=min_items, require_citations=require_citations, project=project, agent=agent, expand_graph=expand_graph, graph_depth=graph_depth, graph_limit=graph_limit, + session_id=session_id, ) return salience_mod.attach_salience(result, store, session_id, cfg) +@mcp.tool() +def kb_effectiveness( + window: str = "90d", + min_samples: int = 5, +) -> dict[str, Any]: + """Per-artifact effect estimate from surfaced context vs session outcomes.""" + return compute_effectiveness( + _store(), + window=window, + min_samples=min_samples, + ) + + @mcp.tool() def kb_synthesize( query: str, diff --git a/tests/test_effectiveness.py b/tests/test_effectiveness.py new file mode 100644 index 00000000..7daa6f9b --- /dev/null +++ b/tests/test_effectiveness.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from click.testing import CliRunner + +from vouch import index_db +from vouch.audit import log_event, read_events +from vouch.cli import cli +from vouch.eval.effectiveness import compute_effectiveness +from vouch.models import Claim +from vouch.storage import KBStore + +NOW = datetime(2026, 7, 8, 12, 0, 0, tzinfo=UTC) + + +def _ev(kb_dir: Path, event: str, actor: str, object_ids: list[str], ts: datetime) -> None: + log_event(kb_dir, event=event, actor=actor, object_ids=object_ids) + path = kb_dir / "audit.log.jsonl" + lines = path.read_text(encoding="utf-8").splitlines() + payload = json.loads(lines[-1]) + payload["created_at"] = ts.isoformat() + lines[-1] = json.dumps(payload, separators=(",", ":"), sort_keys=True) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _seed_artifacts(store: KBStore) -> None: + src = store.put_source(b"effectiveness fixture") + store.put_claim(Claim(id="c-good", text="good artifact", evidence=[src.id])) + store.put_claim(Claim(id="c-noisy", text="noisy artifact", evidence=[src.id])) + + +def test_effectiveness_insufficient_sample_path(tmp_path: Path) -> None: + store = KBStore.init(tmp_path) + _seed_artifacts(store) + base = NOW - timedelta(days=1) + _ev(store.kb_dir, "session.start", "a1", ["sess-g1"], base) + _ev(store.kb_dir, "claim.confirm", "a1", ["c-good"], base + timedelta(minutes=1)) + _ev(store.kb_dir, "session.end", "a1", ["sess-g1"], base + timedelta(minutes=2)) + _ev(store.kb_dir, "session.start", "a2", ["sess-b1"], base + timedelta(hours=1)) + _ev( + store.kb_dir, + "claim.contradict", + "a2", + ["c-good", "c-noisy"], + base + timedelta(hours=1, minutes=1), + ) + _ev(store.kb_dir, "session.end", "a2", ["sess-b1"], base + timedelta(hours=1, minutes=2)) + with index_db.open_db(store.kb_dir) as conn: + index_db.index_context_surface( + conn, + session_id="sess-g1", + query="q", + surfaced_at=(base + timedelta(minutes=1)).isoformat(), + items=[("claim", "c-good")], + ) + + report = compute_effectiveness(store, window="7d", min_samples=3, now=NOW) + row = next(r for r in report["artifacts"] if r["artifact_id"] == "c-good") + assert row["samples"] == 1 + assert row["verdict"] == "insufficient" + + +def test_effectiveness_clear_signal_with_fixed_clock(tmp_path: Path) -> None: + store = KBStore.init(tmp_path) + _seed_artifacts(store) + base = NOW - timedelta(days=2) + with index_db.open_db(store.kb_dir) as conn: + for i in range(12): + sid = f"sess-{i}" + actor = f"agent-{i}" + start = base + timedelta(minutes=i * 10) + _ev(store.kb_dir, "session.start", actor, [sid], start) + if i < 6: + _ev(store.kb_dir, "claim.confirm", actor, ["c-good"], start + timedelta(minutes=2)) + index_db.index_context_surface( + conn, + session_id=sid, + query="good-query", + surfaced_at=(start + timedelta(minutes=1)).isoformat(), + items=[("claim", "c-good")], + ) + else: + _ev( + store.kb_dir, + "claim.contradict", + actor, + ["c-good", "c-noisy"], + start + timedelta(minutes=2), + ) + _ev(store.kb_dir, "session.end", actor, [sid], start + timedelta(minutes=5)) + + report = compute_effectiveness(store, window="30d", min_samples=5, now=NOW) + row = next(r for r in report["artifacts"] if r["artifact_id"] == "c-good") + assert report["sessions"]["baseline_rate"] == 0.5 + assert row["samples"] == 6 + assert row["verdict"] == "useful" + assert row["wilson_95"]["low"] > report["sessions"]["baseline_rate"] + + +def test_effectiveness_is_read_only_and_reset_clears_surface_table(tmp_path: Path) -> None: + store = KBStore.init(tmp_path) + _seed_artifacts(store) + base = NOW - timedelta(hours=5) + _ev(store.kb_dir, "session.start", "r", ["sess-r"], base) + _ev(store.kb_dir, "claim.confirm", "r", ["c-good"], base + timedelta(minutes=1)) + _ev(store.kb_dir, "session.end", "r", ["sess-r"], base + timedelta(minutes=2)) + with index_db.open_db(store.kb_dir) as conn: + index_db.index_context_surface( + conn, + session_id="sess-r", + query="read-only", + surfaced_at=(base + timedelta(minutes=1)).isoformat(), + items=[("claim", "c-good")], + ) + events_before = len(list(read_events(store.kb_dir))) + rows_before = len(index_db.read_context_surfaces(store.kb_dir)) + + _ = compute_effectiveness(store, window="7d", min_samples=1, now=NOW) + + assert len(list(read_events(store.kb_dir))) == events_before + assert len(index_db.read_context_surfaces(store.kb_dir)) == rows_before + + index_db.reset(store.kb_dir) + assert index_db.read_context_surfaces(store.kb_dir) == [] + + +def test_cli_health_effectiveness_json(tmp_path: Path, monkeypatch) -> None: + store = KBStore.init(tmp_path) + _seed_artifacts(store) + base = NOW - timedelta(hours=4) + _ev(store.kb_dir, "session.start", "a", ["sess-a"], base) + _ev(store.kb_dir, "claim.confirm", "a", ["c-good"], base + timedelta(minutes=1)) + _ev(store.kb_dir, "session.end", "a", ["sess-a"], base + timedelta(minutes=2)) + with index_db.open_db(store.kb_dir) as conn: + index_db.index_context_surface( + conn, + session_id="sess-a", + query="cli", + surfaced_at=(base + timedelta(minutes=1)).isoformat(), + items=[("claim", "c-good")], + ) + monkeypatch.chdir(store.root) + res = CliRunner().invoke(cli, ["health", "effectiveness", "--format", "json"]) + assert res.exit_code == 0, res.output + payload = json.loads(res.output) + assert payload["schema_version"] == 1 + assert "artifacts" in payload