REUSE-WP-0019-T04: reuse telemetry store and recording

Implements the hub side of the shared reuse-event schema (already drafted
in WP-0018-T01, schemas/reuse-event.schema.json): a SQLite reuse_events
table, POST /v1/reuse-events (token-auth), GET /v1/reuse-events?capability_id=
(read-only).

reuse_surface/plan_check.py: refactored record_outcome around a new
shared post_or_fallback_reuse_event() helper -- tries the hub first, falls
back to the local JSONL only on failure/unreachability, never both. New
record_manual_reuse_event() backs a new CLI command, reuse-surface
record-reuse, for retroactive facts recorded outside plan-check.

Privacy/scope (repo slugs and capability ids only, no code, no secrets) is
enforced structurally via the schema's additionalProperties: false, not
just by convention.

21 new pytest cases, 145 total pass. Live-verified against a real running
hub instance: POST/GET /v1/reuse-events directly, record-reuse and
plan-check --record-outcome both posting successfully to the hub, and --
after actually killing the hub process -- confirmed the fallback path
writes correctly to the local JSONL instead of erroring.

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

reuse_surface/cli.py

@@ -28,6 +28,7 @@ from reuse_surface.plan_check import (
     load_query_from_intent,
     load_query_from_workplan,
     maybe_file_capability_request,
+    record_manual_reuse_event,
     record_outcome,
     run_plan_check,
 )
@@ -663,8 +664,9 @@ def cmd_plan_check(args: argparse.Namespace) -> int:
         llm_url=args.llm_url,
     )
 
+    recorded = None
     if args.record_outcome:
-        record_outcome(result, args.record_outcome, consumer_repo=args.consumer_repo)
+        recorded = record_outcome(result, args.record_outcome, consumer_repo=args.consumer_repo)
 
     filed = None
     if args.file_request and result["verdict"] == "new":
@@ -677,6 +679,8 @@ def cmd_plan_check(args: argparse.Namespace) -> int:
     if args.format == "json":
         if filed is not None:
             result["filed_capability_request"] = filed
+        if recorded is not None:
+            result["recorded_reuse_event"] = recorded
         print(format_plan_check_json(result))
     else:
         print(format_plan_check_markdown(result), end="")
@@ -685,6 +689,32 @@ def cmd_plan_check(args: argparse.Namespace) -> int:
                 print(f"\nFiled State Hub capability request: {filed.get('id')}")
             else:
                 print("\nState Hub unreachable — capability request not filed.")
+        if recorded is not None:
+            if recorded["recorded_to"] == "hub":
+                print("\nReuse event recorded to hub.")
+            else:
+                print("\nReuse hub unreachable — recorded to local JSONL fallback.")
+    return 0
+
+
+def cmd_record_reuse(args: argparse.Namespace) -> int:
+    try:
+        recorded = record_manual_reuse_event(
+            consumer_repo=args.consumer_repo,
+            capability_id=args.capability_id,
+            verdict=args.verdict,
+            outcome=args.outcome,
+            hub_url=args.hub_url,
+        )
+    except ValueError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 1
+
+    if args.format == "json":
+        print(json.dumps(recorded, indent=2, sort_keys=True))
+    else:
+        destination = "hub" if recorded["recorded_to"] == "hub" else f"local ({recorded['path']})"
+        print(f"ok: recorded reuse event to {destination}")
     return 0
 
 
@@ -880,7 +910,8 @@ def main(argv: list[str] | None = None) -> int:
     plan_check.add_argument(
         "--record-outcome",
         choices=["reused", "extended", "new", "skipped"],
-        help="append this outcome to registry/telemetry/plan-check-events.jsonl",
+        help="record this outcome as a reuse event: POST /v1/reuse-events if the "
+        "hub is reachable, else append to registry/telemetry/plan-check-events.jsonl",
     )
     plan_check.add_argument(
         "--consumer-repo", default="reuse-surface",
@@ -903,6 +934,26 @@ def main(argv: list[str] | None = None) -> int:
     )
     plan_check.set_defaults(func=cmd_plan_check)
 
+    record_reuse = subparsers.add_parser(
+        "record-reuse",
+        help="manually record a retroactive reuse fact (REUSE-WP-0019-T04)",
+    )
+    record_reuse.add_argument("--consumer-repo", required=True, help="repo slug that reused/extended/built new")
+    record_reuse.add_argument(
+        "--capability-id", help="capability id involved, if any (omit for a pure 'new' fact)"
+    )
+    record_reuse.add_argument(
+        "--verdict", required=True, choices=["reuse", "extend", "new"],
+    )
+    record_reuse.add_argument(
+        "--outcome", choices=["reused", "extended", "new", "skipped"],
+    )
+    record_reuse.add_argument(
+        "--hub-url", help="hub base URL (or REUSE_SURFACE_URL)",
+    )
+    record_reuse.add_argument("--format", choices=["markdown", "json"], default="markdown")
+    record_reuse.set_defaults(func=cmd_record_reuse)
+
     catalog = subparsers.add_parser(
         "catalog", help="generate human-readable capability catalog"
     )

reuse_surface/hub/app.py

@@ -200,6 +200,21 @@ def create_app() -> FastAPI:
 
         return JSONResponse(content={"accepted": True, "composed_at": composed_at})
 
+    @app.post("/v1/reuse-events", status_code=201, dependencies=[Depends(_require_auth)])
+    async def record_reuse_event(request: Request) -> dict[str, Any]:
+        payload = await request.json()
+        try:
+            return store.record_reuse_event(payload)
+        except ValueError as exc:
+            raise _http_error(400, "validation_error", str(exc)) from exc
+
+    @app.get("/v1/reuse-events")
+    def list_reuse_events(
+        capability_id: str | None = Query(default=None),
+    ) -> dict[str, Any]:
+        events = store.list_reuse_events(capability_id)
+        return {"count": len(events), "events": events}
+
     return app
 
 

reuse_surface/hub/store.py

@@ -10,6 +10,11 @@ import yaml
 from jsonschema import Draft202012Validator
 
 SCHEMA_PATH = Path(__file__).resolve().parent / "hub-registration.schema.yaml"
+REUSE_EVENT_SCHEMA_PATH = Path(__file__).resolve().parent.parent.parent / "schemas" / "reuse-event.schema.json"
+
+
+def _load_reuse_event_schema() -> dict[str, Any]:
+    return json.loads(REUSE_EVENT_SCHEMA_PATH.read_text(encoding="utf-8"))
 
 
 def _utc_now() -> str:
@@ -78,6 +83,67 @@ class HubStore:
             conn.execute(
                 "INSERT OR IGNORE INTO compose_state (id, composed_at, stale) VALUES (1, NULL, 0)"
             )
+            # Reuse telemetry (REUSE-WP-0019-T04): append-only facts, never
+            # hand-edited. Schema shared with the local JSONL fallback in
+            # plan_check.py's record_outcome() -- see schemas/reuse-event.schema.json.
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS reuse_events (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    ts TEXT NOT NULL,
+                    consumer_repo TEXT NOT NULL,
+                    capability_id TEXT,
+                    verdict TEXT NOT NULL,
+                    outcome TEXT,
+                    source TEXT NOT NULL
+                )
+                """
+            )
+
+    def record_reuse_event(self, payload: dict[str, Any]) -> dict[str, Any]:
+        errors = [
+            error.message
+            for error in Draft202012Validator(_load_reuse_event_schema()).iter_errors(payload)
+        ]
+        if errors:
+            raise ValueError("; ".join(errors))
+        with self._connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO reuse_events (ts, consumer_repo, capability_id, verdict, outcome, source)
+                VALUES (?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    payload["ts"],
+                    payload["consumer_repo"],
+                    payload.get("capability_id"),
+                    payload["verdict"],
+                    payload.get("outcome"),
+                    payload["source"],
+                ),
+            )
+        return payload
+
+    def list_reuse_events(self, capability_id: str | None = None) -> list[dict[str, Any]]:
+        query = "SELECT ts, consumer_repo, capability_id, verdict, outcome, source FROM reuse_events"
+        params: tuple[Any, ...] = ()
+        if capability_id:
+            query += " WHERE capability_id = ?"
+            params = (capability_id,)
+        query += " ORDER BY ts"
+        with self._connect() as conn:
+            rows = conn.execute(query, params).fetchall()
+        return [
+            {
+                "ts": row["ts"],
+                "consumer_repo": row["consumer_repo"],
+                "capability_id": row["capability_id"],
+                "verdict": row["verdict"],
+                "outcome": row["outcome"],
+                "source": row["source"],
+            }
+            for row in rows
+        ]
 
     def list_repos(self) -> list[dict[str, Any]]:
         with self._connect() as conn:

reuse_surface/hub_client.py

@@ -83,4 +83,27 @@ def hub_update(
         f"{service_base_url(base_url)}/v1/repos/{repo}",
         token=token,
         body=payload,
-    )
+    )
+
+
+def hub_record_reuse_event(
+    payload: dict[str, Any], base_url: str | None = None
+) -> tuple[int, Any]:
+    token = service_token()
+    if not token:
+        raise ValueError("REUSE_SURFACE_TOKEN is required for record-reuse")
+    return _request(
+        "POST",
+        f"{service_base_url(base_url)}/v1/reuse-events",
+        token=token,
+        body=payload,
+    )
+
+
+def hub_list_reuse_events(
+    capability_id: str | None = None, base_url: str | None = None
+) -> tuple[int, Any]:
+    url = f"{service_base_url(base_url)}/v1/reuse-events"
+    if capability_id:
+        url += f"?capability_id={capability_id}"
+    return _request("GET", url)

reuse_surface/plan_check.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import re
+import urllib.error
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
@@ -11,6 +12,7 @@ import yaml
 
 from jsonschema import Draft202012Validator
 
+from reuse_surface import hub_client
 from reuse_surface.federation import FEDERATED_INDEX_PATH
 from reuse_surface.llm_bridge import execute_prompt, extract_json_object
 from reuse_surface.overlaps import TOKEN_RE
@@ -19,6 +21,7 @@ from reuse_surface.statehub_bridge import file_capability_request
 
 TELEMETRY_PATH = ROOT / "registry" / "telemetry" / "plan-check-events.jsonl"
 RERANK_SCHEMA_PATH = ROOT / "schemas" / "plan-check-rerank.schema.json"
+REUSE_EVENT_SCHEMA_PATH = ROOT / "schemas" / "reuse-event.schema.json"
 
 STALE_DAYS = 14
 DEFAULT_REUSE_THRESHOLD = 0.45
@@ -371,22 +374,87 @@ def maybe_file_capability_request(
     )
 
 
+def load_reuse_event_schema() -> dict[str, Any]:
+    return json.loads(REUSE_EVENT_SCHEMA_PATH.read_text(encoding="utf-8"))
+
+
+def validate_reuse_event(event: dict[str, Any]) -> None:
+    validator = Draft202012Validator(load_reuse_event_schema())
+    errors = sorted(validator.iter_errors(event), key=lambda err: list(err.path))
+    if errors:
+        messages = "; ".join(error.message for error in errors[:3])
+        raise ValueError(f"reuse event schema validation failed: {messages}")
+
+
+def append_local_reuse_event(event: dict[str, Any]) -> Path:
+    TELEMETRY_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with TELEMETRY_PATH.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(event, sort_keys=True) + "\n")
+    return TELEMETRY_PATH
+
+
+def post_or_fallback_reuse_event(
+    event: dict[str, Any], *, hub_url: str | None = None
+) -> dict[str, Any]:
+    """Records one reuse event: the shared schema (schemas/reuse-event.schema.json)
+    is the same whether the fact ends up in the hub's reuse_events table or the
+    local JSONL fallback. Tries POST /v1/reuse-events first; on any
+    misconfiguration or unreachability, falls back to the local file rather
+    than losing the fact -- never both, per the design ("+ local JSONL
+    fallback when hub unreachable", not a dual-write)."""
+    validate_reuse_event(event)
+
+    try:
+        status, _payload = hub_client.hub_record_reuse_event(event, hub_url)
+    except (ValueError, urllib.error.URLError, TimeoutError, OSError):
+        status = None
+
+    if status == 201:
+        return {"event": event, "recorded_to": "hub", "path": None}
+
+    path = append_local_reuse_event(event)
+    return {"event": event, "recorded_to": "local", "path": str(path)}
+
+
 def record_outcome(
     result: dict[str, Any],
     outcome: str,
     *,
     consumer_repo: str = "reuse-surface",
-) -> Path:
-    TELEMETRY_PATH.parent.mkdir(parents=True, exist_ok=True)
+    capability_id: str | None = None,
+    verdict: str | None = None,
+    source: str = "plan-check",
+    hub_url: str | None = None,
+) -> dict[str, Any]:
     top_match = result["matches"][0]["id"] if result.get("matches") else None
     event = {
         "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
         "consumer_repo": consumer_repo,
-        "capability_id": top_match,
-        "verdict": result["verdict"],
+        "capability_id": capability_id if capability_id is not None else top_match,
+        "verdict": verdict if verdict is not None else result["verdict"],
         "outcome": outcome,
-        "source": "plan-check",
+        "source": source,
     }
-    with TELEMETRY_PATH.open("a", encoding="utf-8") as handle:
-        handle.write(json.dumps(event, sort_keys=True) + "\n")
-    return TELEMETRY_PATH
+    return post_or_fallback_reuse_event(event, hub_url=hub_url)
+
+
+def record_manual_reuse_event(
+    *,
+    consumer_repo: str,
+    capability_id: str | None,
+    verdict: str,
+    outcome: str | None = None,
+    hub_url: str | None = None,
+) -> dict[str, Any]:
+    """Retroactive manual fact recording (`reuse-surface record-reuse`) --
+    for reuse decisions made outside plan-check, e.g. discovered after the
+    fact during a review."""
+    event = {
+        "ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "consumer_repo": consumer_repo,
+        "capability_id": capability_id,
+        "verdict": verdict,
+        "outcome": outcome,
+        "source": "manual",
+    }
+    return post_or_fallback_reuse_event(event, hub_url=hub_url)