REUSE-WP-0019-T05: reuse telemetry aggregation into R-axis evidence

reuse_surface/reports.py: collect_reuse_events() merges the hub's
GET /v1/reuse-events (if reachable) with this repo's local JSONL fallback,
deduped. collect_reuse_report() aggregates per-capability consumer counts,
outcome breakdown, and last-used. collect_reused_by_suggestions() proposes
evidence-gated relation_add patches -- only for capabilities this repo
owns, only for consumer repos not already listed -- reusing the existing
patches.py:apply_patches mechanism (relation_add already isn't in
SAFE_DETERMINISTIC_KINDS, so it was already never auto-applied by
maintain --auto).

New CLI: reuse-surface report reuse [--capability-id] [--format]
[--suggest-relations] [--apply]. --apply requires --suggest-relations and
is the only thing that writes -- nothing happens automatically from
telemetry alone.

schemas/capability.schema.yaml: added relations.reused_by as a new
repoSlugList type, distinct from the existing capability-id relations,
since reused-by targets are consumer repo slugs.

specs/CapabilityMaturityStandard.md Sec8.9: what observed-reuse evidence
counts toward R2->R3 (single corroborating consumer) vs R3->R4+ (multiple
independent consumers) and what it never substitutes for.

19 new pytest cases, 162 total pass. Live-verified with synthetic local
events against a real capability entry: --suggest-relations --apply
correctly wrote relations.reused_by via the real apply_patches path
(reverted after, since it was a smoke test).

Deliberately deferred: surfacing consumer counts in the catalog/graph --
graph.py's relation model is capability-to-capability edges, a different
namespace than repo-slug reused_by targets; catalog.py doesn't currently
parse full front matter per entry. Left for a follow-up.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

reuse_surface/cli.py

@@ -35,11 +35,16 @@ from reuse_surface.plan_check import (
 from reuse_surface.reports import (
     cohort_filters_from_args,
     collect_gap_report,
+    collect_reuse_events,
+    collect_reuse_report,
+    collect_reused_by_suggestions,
     default_roster_path,
     format_cohort_json,
     format_cohort_markdown,
     format_gap_json,
     format_gap_markdown,
+    format_reuse_report_json,
+    format_reuse_report_markdown,
     select_cohort,
 )
 from reuse_surface.establish import (
@@ -753,6 +758,46 @@ def cmd_report_gaps(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_report_reuse(args: argparse.Namespace) -> int:
+    if args.apply and not args.suggest_relations:
+        print("error: --apply requires --suggest-relations", file=sys.stderr)
+        return 1
+
+    events = collect_reuse_events(hub_url=args.hub_url)
+    report = collect_reuse_report(events, capability_id=args.capability_id)
+
+    suggestions: list[dict[str, Any]] = []
+    if args.suggest_relations:
+        suggestions = collect_reused_by_suggestions(report)
+
+    applied: list[str] = []
+    if args.apply and suggestions:
+        from reuse_surface.patches import apply_patches
+        from reuse_surface.registry import ROOT
+
+        applied = apply_patches(ROOT, suggestions)
+
+    if args.format == "json":
+        payload = dict(report)
+        if args.suggest_relations:
+            payload["reused_by_suggestions"] = suggestions
+        if args.apply:
+            payload["applied"] = applied
+        print(format_reuse_report_json(payload))
+    else:
+        print(format_reuse_report_markdown(report), end="")
+        if args.suggest_relations:
+            print(f"\n## relations.reused_by suggestions ({len(suggestions)})\n")
+            if not suggestions:
+                print("_None — all observed reuse already reflected, or no capabilities owned here._")
+            else:
+                for s in suggestions:
+                    print(f"- `{s['capability_id']}`: {s['detail']}")
+            if args.apply:
+                print(f"\nApplied {len(applied)} patch(es).")
+    return 0
+
+
 def cmd_export(args: argparse.Namespace) -> int:
     index = load_index()
     bundle: dict[str, Any] = {
@@ -1073,6 +1118,27 @@ def main(argv: list[str] | None = None) -> int:
     )
     gaps.set_defaults(func=cmd_report_gaps)
 
+    reuse = report_sub.add_parser(
+        "reuse",
+        help="reuse telemetry aggregation: consumer counts, outcomes, last-used (REUSE-WP-0019-T05)",
+    )
+    reuse.add_argument("--capability-id", help="filter to one capability")
+    reuse.add_argument("--hub-url", help="hub base URL (or REUSE_SURFACE_URL)")
+    reuse.add_argument("--format", choices=["markdown", "json"], default="markdown")
+    reuse.add_argument(
+        "--suggest-relations",
+        action="store_true",
+        help="also list evidence-gated relations.reused_by suggestions for "
+        "capabilities this repo owns",
+    )
+    reuse.add_argument(
+        "--apply",
+        action="store_true",
+        help="apply --suggest-relations suggestions (requires --suggest-relations; "
+        "never applies without this explicit flag)",
+    )
+    reuse.set_defaults(func=cmd_report_reuse)
+
     stats = subparsers.add_parser("stats", help="registry maturity and federation stats")
     stats.add_argument("--path", help="repo root (default: cwd)")
     stats.add_argument(

reuse_surface/reports.py

@@ -6,7 +6,15 @@ from typing import Any
 
 import yaml
 
-from reuse_surface.registry import ROOT, level_at_least, load_index, parse_vector
+from reuse_surface.registry import (
+    ROOT,
+    level_at_least,
+    load_index,
+    load_index_at,
+    parse_front_matter,
+    parse_vector,
+    registry_paths,
+)
 
 
 def _availability_at_most(current: str, maximum: str) -> bool:
@@ -228,4 +236,127 @@ def format_gap_json(report: dict[str, Any]) -> str:
 
 
 def default_roster_path() -> Path:
-    return ROOT / "registry/federation/local-repo-roster.yaml"
+    return ROOT / "registry/federation/local-repo-roster.yaml"
+
+
+def collect_reuse_events(
+    *, repo_root: Path = ROOT, hub_url: str | None = None
+) -> list[dict[str, Any]]:
+    """Merges the hub's reuse_events (ecosystem-wide, if reachable) with this
+    repo's local JSONL fallback (covers facts recorded while the hub was
+    down), deduped by (ts, consumer_repo, capability_id, source)."""
+    import urllib.error
+
+    from reuse_surface import hub_client
+
+    events: list[dict[str, Any]] = []
+    try:
+        status, payload = hub_client.hub_list_reuse_events(base_url=hub_url)
+        if status == 200:
+            events.extend(payload.get("events", []))
+    except (ValueError, urllib.error.URLError, TimeoutError, OSError):
+        pass
+
+    telemetry_path = repo_root / "registry" / "telemetry" / "plan-check-events.jsonl"
+    if telemetry_path.exists():
+        for line in telemetry_path.read_text(encoding="utf-8").splitlines():
+            if line.strip():
+                events.append(json.loads(line))
+
+    seen: set[tuple[Any, ...]] = set()
+    deduped: list[dict[str, Any]] = []
+    for event in events:
+        key = (event.get("ts"), event.get("consumer_repo"), event.get("capability_id"), event.get("source"))
+        if key not in seen:
+            seen.add(key)
+            deduped.append(event)
+    return deduped
+
+
+def collect_reuse_report(
+    events: list[dict[str, Any]], *, capability_id: str | None = None
+) -> dict[str, Any]:
+    """Per-capability consumer counts, outcome breakdown, last-used."""
+    by_capability: dict[str, dict[str, Any]] = {}
+    for event in events:
+        cap_id = event.get("capability_id")
+        if not cap_id or (capability_id and cap_id != capability_id):
+            continue
+        bucket = by_capability.setdefault(
+            cap_id,
+            {"consumer_repos": set(), "outcomes": {}, "last_used": None},
+        )
+        bucket["consumer_repos"].add(event["consumer_repo"])
+        outcome = event.get("outcome") or "unspecified"
+        bucket["outcomes"][outcome] = bucket["outcomes"].get(outcome, 0) + 1
+        ts = event.get("ts")
+        if ts and (bucket["last_used"] is None or ts > bucket["last_used"]):
+            bucket["last_used"] = ts
+
+    rows = [
+        {
+            "capability_id": cap_id,
+            "consumer_count": len(bucket["consumer_repos"]),
+            "consumer_repos": sorted(bucket["consumer_repos"]),
+            "outcomes": bucket["outcomes"],
+            "last_used": bucket["last_used"],
+        }
+        for cap_id, bucket in sorted(by_capability.items())
+    ]
+    return {"capabilities": rows, "total_events": len(events)}
+
+
+def format_reuse_report_markdown(report: dict[str, Any]) -> str:
+    lines = ["# Reuse telemetry report", ""]
+    if not report["capabilities"]:
+        lines.append("_No reuse events recorded yet._")
+        return "\n".join(lines) + "\n"
+    for row in report["capabilities"]:
+        outcomes = ", ".join(f"{k}: {v}" for k, v in sorted(row["outcomes"].items()))
+        lines.append(f"## `{row['capability_id']}`")
+        lines.append(f"- Consumers ({row['consumer_count']}): {', '.join(row['consumer_repos'])}")
+        lines.append(f"- Outcomes: {outcomes}")
+        lines.append(f"- Last used: {row['last_used']}")
+        lines.append("")
+    lines.append(f"**{report['total_events']}** total event(s) across **{len(report['capabilities'])}** capability(ies).")
+    return "\n".join(lines) + "\n"
+
+
+def format_reuse_report_json(report: dict[str, Any]) -> str:
+    return json.dumps(report, indent=2, sort_keys=True)
+
+
+def collect_reused_by_suggestions(
+    report: dict[str, Any], *, repo_root: Path = ROOT
+) -> list[dict[str, Any]]:
+    """Evidence-gated relation_add suggestions (REUSE-WP-0019-T05): only for
+    capabilities this repo actually owns (in its own index), and only for
+    consumer repos not already listed in relations.reused_by. Suggestions
+    only -- applying them is a separate, explicit step
+    (`report reuse --suggest-relations --apply`), never automatic."""
+    paths = registry_paths(repo_root)
+    if not paths["index"].exists():
+        return []
+    index = load_index_at(paths["index"])
+    index_by_id = {row["id"]: row for row in index.get("capabilities", [])}
+
+    suggestions: list[dict[str, Any]] = []
+    for row in report["capabilities"]:
+        cap_id = row["capability_id"]
+        entry_row = index_by_id.get(cap_id)
+        if not entry_row:
+            continue
+        front_matter = parse_front_matter(repo_root / entry_row["path"])
+        existing = set(front_matter.get("relations", {}).get("reused_by", []))
+        for consumer_repo in row["consumer_repos"]:
+            if consumer_repo in existing:
+                continue
+            suggestions.append(
+                {
+                    "capability_id": cap_id,
+                    "kind": "relation_add",
+                    "detail": f"observed reuse by {consumer_repo} (outcomes: {row['outcomes']})",
+                    "value": {"type": "reused_by", "target": consumer_repo},
+                }
+            )
+    return suggestions