From d1cbdcc36a1bb43b9b74edc07aa2c08ebd8a66cc Mon Sep 17 00:00:00 2001
From: boskodev790 <233739930+boskodev790@users.noreply.github.com>
Date: Fri, 10 Jul 2026 04:52:41 -0400
Subject: [PATCH] =?UTF-8?q?feat(entities):=20entity-connectivity=20report?=
 =?UTF-8?q?=20=E2=80=94=20hubs=20and=20orphans?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`health.status` counts entities and `vouch neighbors` walks the edges
around one node; neither says, in one pass over the whole graph, which
entities are load-bearing and which anchor nothing.

`vouch entities` reports the most-connected hubs and the orphan entities
that no claim mentions, no relation joins, and no page references —
proposed once, then never used, so a curator can prune or merge them.
connectivity sums claim.entities mentions, relation endpoints (a self-loop
counts once), and page.entities references.

strictly a viewport: composes store.list_entities / list_claims /
list_relations / list_pages, writes nothing, logs no audit event. cli-only,
no kb.* method surface. adds src/vouch/entity_graph.py and
tests/test_entity_graph.py (11 tests).
---
 CHANGELOG.md               |   8 ++
 src/vouch/cli.py           |  29 ++++++
 src/vouch/entity_graph.py  | 198 +++++++++++++++++++++++++++++++++++++
 tests/test_entity_graph.py | 171 ++++++++++++++++++++++++++++++++
 4 files changed, 406 insertions(+)
 create mode 100644 src/vouch/entity_graph.py
 create mode 100644 tests/test_entity_graph.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 986dcb0e..25aeffbe 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,14 @@ All notable changes to vouch are documented here. Format follows
 
 ## [Unreleased]
 
+### Added
+- `vouch entities` — a read-only entity-connectivity report over the whole
+  graph: most-connected hubs and orphan entities that no claim mentions, no
+  relation joins, and no page references (proposed once, then never used).
+  connectivity sums `claim.entities` mentions, relation endpoints (a self-loop
+  counts once), and `page.entities` references. `--format text|json|markdown`;
+  writes nothing.
+
 ## [1.2.2] — 2026-07-07
 
 ### Packaging
diff --git a/src/vouch/cli.py b/src/vouch/cli.py
index 198d32f4..81ab553a 100644
--- a/src/vouch/cli.py
+++ b/src/vouch/cli.py
@@ -28,6 +28,7 @@
 from . import codex_rollout as codex_rollout_mod
 from . import compile as compile_mod
 from . import digest as digest_mod
+from . import entity_graph as entity_graph_mod
 from . import fetch as fetch_mod
 from . import inbox as inbox_mod
 from . import install_adapter as install_mod
@@ -365,6 +366,34 @@ def digest_cmd(since: str, stale_days: int, limit: int, fmt: str) -> None:
         click.echo(digest_mod.render_text(d))
 
 
+@cli.command(name="entities")
+@click.option(
+    "--limit",
+    default=entity_graph_mod.DEFAULT_LIMIT,
+    show_default=True,
+    type=int,
+    help="Cap the most-connected and orphaned sections.",
+)
+@click.option(
+    "--format",
+    "fmt",
+    default="text",
+    show_default=True,
+    type=click.Choice(["text", "json", "markdown"]),
+)
+def entities_cmd(limit: int, fmt: str) -> None:
+    """Read-only entity-connectivity report: most-connected hubs and orphan
+    entities that anchor nothing. Writes nothing — safe to run from cron."""
+    store = _load_store()
+    report = entity_graph_mod.build(store, limit=limit)
+    if fmt == "json":
+        _emit_json(report.to_dict())
+    elif fmt == "markdown":
+        click.echo(entity_graph_mod.render_markdown(report))
+    else:
+        click.echo(entity_graph_mod.render_text(report))
+
+
 def _findings_json(report) -> list[dict[str, Any]]:
     return [
         {
diff --git a/src/vouch/entity_graph.py b/src/vouch/entity_graph.py
new file mode 100644
index 00000000..dd2ce5e8
--- /dev/null
+++ b/src/vouch/entity_graph.py
@@ -0,0 +1,198 @@
+"""Read-only entity-connectivity report: which entities anchor the graph, and
+which anchor nothing.
+
+An entity exists to be connected — vouch's own model note says entities "anchor
+relations and aggregate the claims that mention them." `health.status` counts
+entities; `vouch neighbors` walks the edges around one node. Neither answers, in
+one pass over the whole graph, which entities are load-bearing hubs and which
+are dead: proposed once, then never mentioned by a claim, never joined by a
+relation, never referenced by a page.
+
+An entity's connectivity here is the count of everything that points at it:
+claims that mention it (`claim.entities`), relations with it at either endpoint
+(a self-loop counts once), and pages that reference it (`page.entities`). An
+entity is an *orphan* when that count is zero — dead weight a curator can prune
+or merge into a live one.
+
+Strictly a viewport: it composes `store.list_entities`, `list_claims`,
+`list_relations` and `list_pages`, writes nothing, logs no audit event, and
+never touches a proposal.
+"""
+
+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 EntityConnectivity:
+    id: str
+    name: str
+    type: str
+    created_at: str
+    # Distinct claims whose `entities` list names this entity.
+    claim_mentions: int
+    # Relations with this entity at either endpoint; a self-loop counts once.
+    relations: int
+    # Pages whose `entities` list names this entity.
+    page_references: int
+    # claim_mentions + relations + page_references.
+    connections: int
+    is_orphan: bool
+
+
+@dataclass(frozen=True)
+class EntityGraphReport:
+    """Stable `to_dict()` schema — the `--format json` contract."""
+
+    generated_at: str
+    limit: int
+    entities_total: int
+    connected_total: int
+    orphan_total: int
+    most_connected: list[EntityConnectivity] = field(default_factory=list)
+    orphans: list[EntityConnectivity] = 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,
+) -> EntityGraphReport:
+    """Compose the entity-connectivity report. Read-only by construction."""
+    now = _as_utc(now) or datetime.now(UTC)
+
+    entities = store.list_entities()
+    entity_ids = {e.id for e in entities}
+
+    claim_mentions: dict[str, int] = defaultdict(int)
+    for c in store.list_claims():
+        # A claim can list the same entity twice; count the entity once per
+        # claim by collapsing to a set first.
+        for eid in set(c.entities):
+            if eid in entity_ids:
+                claim_mentions[eid] += 1
+
+    relation_degree: dict[str, int] = defaultdict(int)
+    for r in store.list_relations():
+        # A self-loop (source == target) is one edge on the node, not two.
+        for eid in {r.source, r.target}:
+            if eid in entity_ids:
+                relation_degree[eid] += 1
+
+    page_references: dict[str, int] = defaultdict(int)
+    for p in store.list_pages():
+        for eid in set(p.entities):
+            if eid in entity_ids:
+                page_references[eid] += 1
+
+    rows: list[EntityConnectivity] = []
+    for e in entities:
+        mentions = claim_mentions.get(e.id, 0)
+        rels = relation_degree.get(e.id, 0)
+        pages = page_references.get(e.id, 0)
+        total = mentions + rels + pages
+        rows.append(
+            EntityConnectivity(
+                id=e.id,
+                name=e.name,
+                type=e.type.value,
+                created_at=(_as_utc(e.created_at) or now).isoformat(timespec="seconds"),
+                claim_mentions=mentions,
+                relations=rels,
+                page_references=pages,
+                connections=total,
+                is_orphan=total == 0,
+            )
+        )
+
+    most_connected = sorted(
+        (r for r in rows if r.connections > 0),
+        key=lambda r: (-r.connections, -r.relations, r.id),
+    )
+    orphans = sorted(
+        (r for r in rows if r.is_orphan),
+        key=lambda r: (r.created_at, r.id),
+    )
+
+    return EntityGraphReport(
+        generated_at=now.isoformat(timespec="seconds"),
+        limit=limit,
+        entities_total=len(rows),
+        connected_total=sum(1 for r in rows if r.connections > 0),
+        orphan_total=len(orphans),
+        most_connected=most_connected[:limit],
+        orphans=orphans[:limit],
+    )
+
+
+def render_text(report: EntityGraphReport) -> str:
+    lines = [
+        f"entity graph @ {report.generated_at}",
+        f"entities: {report.entities_total}  connected: {report.connected_total}  "
+        f"orphaned: {report.orphan_total}",
+        "",
+        f"most connected ({len(report.most_connected)})",
+    ]
+    for r in report.most_connected:
+        lines.append(
+            f"  {r.connections:>4} conn  ({r.claim_mentions} claim / {r.relations} rel / "
+            f"{r.page_references} page)  {r.id}  [{r.type}]  {r.name}"
+        )
+    if not report.most_connected:
+        lines.append("  none")
+    if report.connected_total > len(report.most_connected):
+        lines.append(f"  ... and {report.connected_total - len(report.most_connected)} more")
+    lines.append("")
+    lines.append(f"orphaned entities ({report.orphan_total})")
+    for r in report.orphans:
+        lines.append(f"  {r.id}  [{r.type}]  {r.created_at[:10]}  {r.name}")
+    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: EntityGraphReport) -> str:
+    lines = [
+        f"# entity graph — {report.generated_at}",
+        "",
+        f"entities: {report.entities_total} · connected: {report.connected_total} · "
+        f"orphaned: {report.orphan_total}",
+        "",
+        f"## most connected ({len(report.most_connected)})",
+    ]
+    lines += [
+        f"- `{r.id}` [{r.type}] {r.name} — {r.connections} conn "
+        f"({r.claim_mentions} claim / {r.relations} rel / {r.page_references} page)"
+        for r in report.most_connected
+    ] or ["- none"]
+    if report.connected_total > len(report.most_connected):
+        lines.append(f"- ... and {report.connected_total - len(report.most_connected)} more")
+    lines.append("")
+    lines.append(f"## orphaned entities ({report.orphan_total})")
+    lines += [
+        f"- `{r.id}` [{r.type}] {r.name} — 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_entity_graph.py b/tests/test_entity_graph.py
new file mode 100644
index 00000000..a75fdf09
--- /dev/null
+++ b/tests/test_entity_graph.py
@@ -0,0 +1,171 @@
+"""Read-only entity-connectivity report — vouch entities."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from vouch import entity_graph
+from vouch.cli import cli
+from vouch.models import Claim, Entity, EntityType, Page, Relation, RelationType
+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 _ent(store: KBStore, eid: str, name: str) -> None:
+    store.put_entity(Entity(id=eid, name=name, type=EntityType.CONCEPT))
+
+
+def _rel(store: KBStore, rid: str, src: str, rel: RelationType, tgt: str) -> None:
+    store.put_relation(Relation(id=rid, source=src, relation=rel, target=tgt))
+
+
+def _row(report: entity_graph.EntityGraphReport, eid: str) -> entity_graph.EntityConnectivity:
+    for r in report.most_connected + report.orphans:
+        if r.id == eid:
+            return r
+    raise AssertionError(f"{eid} not in report")
+
+
+def test_hub_connectivity_sums_all_sources(store: KBStore) -> None:
+    src = store.put_source(b"x")
+    _ent(store, "hub", "Hub")
+    _ent(store, "other", "Other")
+    store.put_claim(Claim(id="c1", text="a", evidence=[src.id], entities=["hub"]))
+    store.put_claim(Claim(id="c2", text="b", evidence=[src.id], entities=["hub"]))
+    _rel(store, "r1", "hub", RelationType.RELATES_TO, "other")
+    store.put_page(Page(id="p1", title="w", entities=["hub"]))
+
+    row = _row(entity_graph.build(store), "hub")
+    assert row.claim_mentions == 2
+    assert row.relations == 1
+    assert row.page_references == 1
+    assert row.connections == 4
+    assert row.is_orphan is False
+
+
+def test_self_loop_counts_once(store: KBStore) -> None:
+    _ent(store, "e", "E")
+    _rel(store, "r1", "e", RelationType.SIMILAR_TO, "e")
+    assert _row(entity_graph.build(store), "e").relations == 1
+
+
+def test_relation_increments_both_endpoints(store: KBStore) -> None:
+    _ent(store, "a", "A")
+    _ent(store, "b", "B")
+    _rel(store, "r1", "a", RelationType.DEPENDS_ON, "b")
+    report = entity_graph.build(store)
+    assert _row(report, "a").relations == 1
+    assert _row(report, "b").relations == 1
+
+
+def test_non_entity_endpoint_is_not_counted_as_entity(store: KBStore) -> None:
+    src = store.put_source(b"x")
+    _ent(store, "e", "E")
+    store.put_claim(Claim(id="c1", text="a", evidence=[src.id]))
+    # a relation from an entity to a claim: the entity gets the edge, the claim
+    # id must not surface as a phantom entity row.
+    _rel(store, "r1", "e", RelationType.MENTIONS, "c1")
+    report = entity_graph.build(store)
+    assert report.entities_total == 1
+    assert _row(report, "e").relations == 1
+
+
+def test_claim_mention_deduped_per_claim(store: KBStore) -> None:
+    src = store.put_source(b"x")
+    _ent(store, "e", "E")
+    # a claim naming the same entity twice counts the entity once.
+    store.put_claim(Claim(id="c1", text="a", evidence=[src.id], entities=["e", "e"]))
+    assert _row(entity_graph.build(store), "e").claim_mentions == 1
+
+
+def test_orphan_entity(store: KBStore) -> None:
+    src = store.put_source(b"x")
+    _ent(store, "live", "Live")
+    _ent(store, "orphan", "Orphan")
+    store.put_claim(Claim(id="c1", text="a", evidence=[src.id], entities=["live"]))
+
+    report = entity_graph.build(store)
+    assert report.entities_total == 2
+    assert report.connected_total == 1
+    assert report.orphan_total == 1
+    assert [r.id for r in report.orphans] == ["orphan"]
+    assert _row(report, "orphan").is_orphan is True
+
+
+def test_most_connected_ordering_and_truncation(store: KBStore) -> None:
+    src = store.put_source(b"x")
+    for i in range(3):
+        _ent(store, f"e{i}", f"E{i}")
+        store.put_claim(Claim(id=f"c{i}", text="t", evidence=[src.id], entities=[f"e{i}"]))
+    # give e0 an extra connection so it ranks first.
+    _ent(store, "z", "Z")
+    _rel(store, "r0", "e0", RelationType.RELATES_TO, "z")
+
+    report = entity_graph.build(store, limit=1)
+    assert report.most_connected[0].id == "e0"
+    assert len(report.most_connected) == 1
+    assert "... and" in entity_graph.render_text(report)
+
+
+def test_to_dict_schema(store: KBStore) -> None:
+    _ent(store, "e", "E")
+    body = entity_graph.build(store).to_dict()
+    assert set(body) == {
+        "generated_at",
+        "limit",
+        "entities_total",
+        "connected_total",
+        "orphan_total",
+        "most_connected",
+        "orphans",
+    }
+    assert set(body["orphans"][0]) == {
+        "id",
+        "name",
+        "type",
+        "created_at",
+        "claim_mentions",
+        "relations",
+        "page_references",
+        "connections",
+        "is_orphan",
+    }
+
+
+def test_empty_kb(store: KBStore) -> None:
+    report = entity_graph.build(store)
+    assert report.entities_total == 0
+    assert report.most_connected == []
+    assert "orphaned entities (0)" in entity_graph.render_text(report)
+
+
+def test_cli_entities_json(store: KBStore) -> None:
+    src = store.put_source(b"x")
+    _ent(store, "e", "E")
+    store.put_claim(Claim(id="c1", text="t", evidence=[src.id], entities=["e"]))
+    result = CliRunner().invoke(cli, ["entities", "--format", "json"])
+    assert result.exit_code == 0, result.output
+    data = json.loads(result.output)
+    assert data["connected_total"] == 1
+    assert data["most_connected"][0]["id"] == "e"
+
+
+def test_cli_entities_text_and_markdown(store: KBStore) -> None:
+    _ent(store, "lonely", "Lonely")
+    text = CliRunner().invoke(cli, ["entities"])
+    assert text.exit_code == 0, text.output
+    assert "orphaned entities (1)" in text.output
+    assert "Lonely" in text.output
+    md = CliRunner().invoke(cli, ["entities", "--format", "markdown"])
+    assert md.exit_code == 0, md.output
+    assert "# entity graph" in md.output