diff --git a/CHANGELOG.md b/CHANGELOG.md index 986dcb0..6f9fd95 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ All notable changes to vouch are documented here. Format follows ## [Unreleased] +### Added +- `vouch source usage` — a read-only report that reads the source→claim + direction the existing surfaces don't: most-cited sources (resolving both + direct citations and citations made through an evidence span) and orphan + sources that no claim cites and no page references, safe to prune. writes + nothing and logs no audit event; `--format json|markdown` for cron and + dashboards. + ## [1.2.2] — 2026-07-07 ### Packaging diff --git a/src/vouch/cli.py b/src/vouch/cli.py index 198d32f..4b660cc 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -39,6 +39,7 @@ from . import provenance as prov_mod from . import recall as recall_mod from . import sessions as sess_mod +from . import source_usage as source_usage_mod from . import stats as stats_mod from . import sync as sync_mod from . import synthesize as synth @@ -1783,6 +1784,34 @@ def source_verify(fail_on_issue: bool) -> None: sys.exit(1) +@source.command("usage") +@click.option( + "--limit", + default=source_usage_mod.DEFAULT_LIMIT, + show_default=True, + type=int, + help="Cap the most-cited and orphaned sections.", +) +@click.option( + "--format", + "fmt", + default="text", + show_default=True, + type=click.Choice(["text", "json", "markdown"]), +) +def source_usage(limit: int, fmt: str) -> None: + """Read-only source-usage report: most-cited sources and orphans that back + no claim. Writes nothing — safe to run from cron.""" + store = _load_store() + report = source_usage_mod.build(store, limit=limit) + if fmt == "json": + _emit_json(report.to_dict()) + elif fmt == "markdown": + click.echo(source_usage_mod.render_markdown(report)) + else: + click.echo(source_usage_mod.render_text(report)) + + @cli.command(name="inbox") @click.option( "--dir", "directory", required=True, diff --git a/src/vouch/source_usage.py b/src/vouch/source_usage.py new file mode 100644 index 0000000..652e823 --- /dev/null +++ b/src/vouch/source_usage.py @@ -0,0 +1,213 @@ +"""Read-only source-usage report: which registered sources back real claims, +and which back nothing. + +vouch's core invariant runs claim -> source: every claim cites a source. This +viewport reads it from the other end — source -> claim — to answer questions +the existing surfaces don't. `vouch lint` / `health` flag *dangling* citations +(a claim pointing at a source that isn't there); `vouch stats` reports the +coverage *rate*. Neither surfaces the inverse: a source that is registered, +still on disk, and cited by nothing. Those orphans are dead provenance — safe +to prune, or a prompt to mine the material for the claim it should support. + +A source counts as cited when a claim rests on it through either citation path +vouch allows (Claim.evidence holds "Source ids OR Evidence ids"): directly, by +naming the source id, or indirectly, by naming an Evidence span that points +into the source. Both are resolved here, mirroring how `stats.citation_summary` +treats `ref in sources_present or ref in evidence_present`. A source is an +*orphan* when no claim cites it by either path AND no page references it in +`page.sources`; a source that a page references but no claim cites is left out +of both the cited and orphan counts — it is neither, so it appears in neither +actionable list, only in `sources_total`. + +Strictly a viewport: it composes `store.list_sources`, `store.list_claims`, +`store.list_pages` and `store.list_evidence`, writes nothing, logs no audit +event, and never touches a proposal — there is nothing here for the review +gate to gate. +""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import asdict, dataclass, field +from datetime import UTC, datetime +from typing import Any + +from .storage import KBStore + +DEFAULT_LIMIT = 20 + + +@dataclass(frozen=True) +class SourceUsage: + id: str + title: str | None + type: str + locator: str + created_at: str + # Distinct claims that cite this source directly (by its id) or indirectly + # (by an Evidence span that points into it). A claim doing both is counted + # once. + claim_citations: int + # Pages listing the source id in `page.sources`. + page_references: int + # Evidence records whose `source_id` is this source, whether or not any + # claim cites them — informational, not part of the orphan test. + evidence_spans: int + is_orphan: bool + + +@dataclass(frozen=True) +class SourceUsageReport: + """Stable `to_dict()` schema — the `--format json` contract.""" + + generated_at: str + limit: int + sources_total: int + cited_total: int + orphan_total: int + most_cited: list[SourceUsage] = field(default_factory=list) + orphans: list[SourceUsage] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def _as_utc(dt: datetime | None) -> datetime | None: + if dt is None: + return None + return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt.astimezone(UTC) + + +def build( + store: KBStore, + *, + limit: int = DEFAULT_LIMIT, + now: datetime | None = None, +) -> SourceUsageReport: + """Compose the source-usage report. Read-only by construction.""" + now = _as_utc(now) or datetime.now(UTC) + + sources = store.list_sources() + claims = store.list_claims() + pages = store.list_pages() + evidence = store.list_evidence() + + source_ids = {s.id for s in sources} + evidence_to_source = {ev.id: ev.source_id for ev in evidence} + + spans_by_source: dict[str, int] = defaultdict(int) + for ev in evidence: + spans_by_source[ev.source_id] += 1 + + # Distinct claims per source, resolving both citation paths. A ref is an + # Evidence id (resolve to its source) or a Source id; a ref that matches + # neither is a dangling citation `health.lint` already reports, so skip it + # rather than inventing a phantom source row for it. + citing_claims: dict[str, set[str]] = defaultdict(set) + for c in claims: + for ref in c.evidence: + if ref in evidence_to_source: + citing_claims[evidence_to_source[ref]].add(c.id) + elif ref in source_ids: + citing_claims[ref].add(c.id) + + page_refs: dict[str, int] = defaultdict(int) + for p in pages: + for sid in p.sources: + page_refs[sid] += 1 + + rows: list[SourceUsage] = [] + for s in sources: + claim_count = len(citing_claims.get(s.id, ())) + page_count = page_refs.get(s.id, 0) + rows.append( + SourceUsage( + id=s.id, + title=s.title, + type=s.type.value, + locator=s.locator, + created_at=(_as_utc(s.created_at) or now).isoformat(timespec="seconds"), + claim_citations=claim_count, + page_references=page_count, + evidence_spans=spans_by_source.get(s.id, 0), + is_orphan=claim_count == 0 and page_count == 0, + ) + ) + + most_cited = sorted( + (r for r in rows if r.claim_citations > 0), + key=lambda r: (-r.claim_citations, -r.page_references, r.id), + ) + orphans = sorted( + (r for r in rows if r.is_orphan), + key=lambda r: (r.created_at, r.id), + ) + + return SourceUsageReport( + generated_at=now.isoformat(timespec="seconds"), + limit=limit, + sources_total=len(rows), + cited_total=sum(1 for r in rows if r.claim_citations > 0), + orphan_total=len(orphans), + most_cited=most_cited[:limit], + orphans=orphans[:limit], + ) + + +def _label(r: SourceUsage) -> str: + return r.title or r.locator or r.id + + +def render_text(report: SourceUsageReport) -> str: + lines = [ + f"source usage @ {report.generated_at}", + f"sources: {report.sources_total} cited: {report.cited_total} " + f"orphaned: {report.orphan_total}", + "", + f"most cited ({len(report.most_cited)})", + ] + for r in report.most_cited: + via = f", {r.evidence_spans} span(s)" if r.evidence_spans else "" + lines.append( + f" {r.claim_citations:>4} claim(s){via} {r.id[:12]} [{r.type}] {_label(r)}" + ) + if not report.most_cited: + lines.append(" none") + if report.cited_total > len(report.most_cited): + lines.append(f" ... and {report.cited_total - len(report.most_cited)} more cited") + lines.append("") + lines.append(f"orphaned sources ({report.orphan_total})") + for r in report.orphans: + lines.append(f" {r.id[:12]} [{r.type}] {r.created_at[:10]} {_label(r)}") + if not report.orphans: + lines.append(" none") + if report.orphan_total > len(report.orphans): + lines.append(f" ... and {report.orphan_total - len(report.orphans)} more") + return "\n".join(lines) + + +def render_markdown(report: SourceUsageReport) -> str: + lines = [ + f"# source usage — {report.generated_at}", + "", + f"sources: {report.sources_total} · cited: {report.cited_total} · " + f"orphaned: {report.orphan_total}", + "", + f"## most cited ({len(report.most_cited)})", + ] + lines += [ + f"- `{r.id[:12]}` [{r.type}] {_label(r)} — {r.claim_citations} claim(s)" + + (f", {r.evidence_spans} span(s)" if r.evidence_spans else "") + for r in report.most_cited + ] or ["- none"] + if report.cited_total > len(report.most_cited): + lines.append(f"- ... and {report.cited_total - len(report.most_cited)} more cited") + lines.append("") + lines.append(f"## orphaned sources ({report.orphan_total})") + lines += [ + f"- `{r.id[:12]}` [{r.type}] {_label(r)} — registered {r.created_at[:10]}" + for r in report.orphans + ] or ["- none"] + if report.orphan_total > len(report.orphans): + lines.append(f"- ... and {report.orphan_total - len(report.orphans)} more") + return "\n".join(lines) diff --git a/tests/test_source_usage.py b/tests/test_source_usage.py new file mode 100644 index 0000000..fabc08c --- /dev/null +++ b/tests/test_source_usage.py @@ -0,0 +1,208 @@ +"""Read-only source-usage report — vouch source usage.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from vouch import source_usage +from vouch.cli import cli +from vouch.models import Claim, Evidence, Page +from vouch.storage import KBStore + + +@pytest.fixture +def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> KBStore: + s = KBStore.init(tmp_path) + monkeypatch.chdir(s.root) + return s + + +def _row(report: source_usage.SourceUsageReport, source_id: str) -> source_usage.SourceUsage: + for r in report.most_cited + report.orphans: + if r.id == source_id: + return r + raise AssertionError(f"{source_id} not in report") + + +def test_direct_citation_counts(store: KBStore) -> None: + src = store.put_source(b"backing material", title="doc") + store.put_claim(Claim(id="c1", text="a", evidence=[src.id])) + store.put_claim(Claim(id="c2", text="b", evidence=[src.id])) + + report = source_usage.build(store) + row = _row(report, src.id) + assert row.claim_citations == 2 + assert row.is_orphan is False + assert report.cited_total == 1 + assert report.orphan_total == 0 + + +def test_via_evidence_citation(store: KBStore) -> None: + src = store.put_source(b"spanned", title="doc") + store.put_evidence(Evidence(id="ev1", source_id=src.id, locator="L1-L2", quote="q")) + store.put_claim(Claim(id="c1", text="rests on the span", evidence=["ev1"])) + + report = source_usage.build(store) + row = _row(report, src.id) + assert row.claim_citations == 1 + assert row.evidence_spans == 1 + assert row.is_orphan is False + + +def test_direct_and_via_evidence_dedup(store: KBStore) -> None: + src = store.put_source(b"both paths", title="doc") + store.put_evidence(Evidence(id="ev1", source_id=src.id, locator="L1")) + # one claim naming both the source id and an evidence span into it — the + # same claim must not be counted twice for the same source. + store.put_claim(Claim(id="c1", text="cites both ways", evidence=[src.id, "ev1"])) + + report = source_usage.build(store) + assert _row(report, src.id).claim_citations == 1 + + +def test_page_reference_is_not_orphan(store: KBStore) -> None: + src = store.put_source(b"page material", title="doc") + store.put_page(Page(id="p1", title="write-up", sources=[src.id])) + + report = source_usage.build(store) + # a page reference keeps the source out of the orphan set even though no + # claim cites it — it is neither cited nor orphaned, a distinct state. + assert report.sources_total == 1 + assert report.cited_total == 0 + assert report.orphan_total == 0 + assert [r.id for r in report.orphans] == [] + + +def test_orphan_detection(store: KBStore) -> None: + cited = store.put_source(b"cited", title="cited-doc") + orphan = store.put_source(b"orphan", title="orphan-doc") + store.put_claim(Claim(id="c1", text="a", evidence=[cited.id])) + + report = source_usage.build(store) + assert report.sources_total == 2 + assert report.orphan_total == 1 + assert [r.id for r in report.orphans] == [orphan.id] + assert _row(report, orphan.id).is_orphan is True + + +def test_orphan_with_unused_evidence_span_stays_orphan(store: KBStore) -> None: + # a source can have Evidence spans defined yet be cited by no claim — the + # spans are informational and do not rescue it from the orphan set. + orphan = store.put_source(b"has a span nobody cites", title="doc") + store.put_evidence(Evidence(id="ev1", source_id=orphan.id, locator="L1")) + + report = source_usage.build(store) + row = _row(report, orphan.id) + assert row.is_orphan is True + assert row.evidence_spans == 1 + assert row.claim_citations == 0 + + +def test_dangling_citation_is_ignored(store: KBStore) -> None: + orphan = store.put_source(b"real orphan", title="doc") + # a claim citing a source id that isn't on disk — a dangling citation + # health.lint reports. Written directly to bypass put_claim's existence + # guard. The report must not crash or invent a row for the missing id. + (store.kb_dir / "claims" / "dangling.yaml").write_text( + "id: dangling\n" + 'text: "cites nothing real"\n' + "type: fact\n" + "status: working\n" + "confidence: 0.7\n" + f"evidence: ['{'a' * 64}']\n", + encoding="utf-8", + ) + + report = source_usage.build(store) + assert report.sources_total == 1 + assert _row(report, orphan.id).is_orphan is True + + +def test_most_cited_ordering_and_limit(store: KBStore) -> None: + heavy = store.put_source(b"heavy", title="heavy") + light = store.put_source(b"light", title="light") + store.put_claim(Claim(id="c1", text="a", evidence=[heavy.id])) + store.put_claim(Claim(id="c2", text="b", evidence=[heavy.id])) + store.put_claim(Claim(id="c3", text="c", evidence=[light.id])) + + report = source_usage.build(store, limit=1) + assert len(report.most_cited) == 1 + assert report.most_cited[0].id == heavy.id + assert report.most_cited[0].claim_citations == 2 + + +def test_truncation_is_not_silent(store: KBStore) -> None: + for i in range(3): + src = store.put_source(f"doc {i}".encode(), title=f"doc-{i}") + store.put_claim(Claim(id=f"c{i}", text="t", evidence=[src.id])) + + report = source_usage.build(store, limit=1) + assert report.cited_total == 3 + assert len(report.most_cited) == 1 + assert "... and 2 more cited" in source_usage.render_text(report) + assert "... and 2 more cited" in source_usage.render_markdown(report) + + +def test_report_to_dict_schema(store: KBStore) -> None: + store.put_source(b"x", title="doc") + body = source_usage.build(store).to_dict() + assert set(body) == { + "generated_at", + "limit", + "sources_total", + "cited_total", + "orphan_total", + "most_cited", + "orphans", + } + # nested rows carry the stable per-source contract. + assert set(body["orphans"][0]) == { + "id", + "title", + "type", + "locator", + "created_at", + "claim_citations", + "page_references", + "evidence_spans", + "is_orphan", + } + + +def test_empty_kb(store: KBStore) -> None: + report = source_usage.build(store) + assert report.sources_total == 0 + assert report.most_cited == [] + assert report.orphans == [] + assert "orphaned sources (0)" in source_usage.render_text(report) + + +def test_cli_source_usage_json(store: KBStore) -> None: + src = store.put_source(b"x", title="doc") + store.put_claim(Claim(id="c1", text="t", evidence=[src.id])) + + result = CliRunner().invoke(cli, ["source", "usage", "--format", "json"]) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["cited_total"] == 1 + assert data["most_cited"][0]["claim_citations"] == 1 + + +def test_cli_source_usage_text_lists_orphans(store: KBStore) -> None: + store.put_source(b"orphan", title="orphan-doc") + result = CliRunner().invoke(cli, ["source", "usage"]) + assert result.exit_code == 0, result.output + assert "orphaned sources (1)" in result.output + assert "orphan-doc" in result.output + + +def test_cli_source_usage_markdown(store: KBStore) -> None: + src = store.put_source(b"x", title="doc") + store.put_claim(Claim(id="c1", text="t", evidence=[src.id])) + result = CliRunner().invoke(cli, ["source", "usage", "--format", "markdown"]) + assert result.exit_code == 0, result.output + assert "## most cited" in result.output