Implement hosted Breach/Compliance Record publication (WP-0006-T09)

migrations/0004_breach_records.sql models a case's lifecycle as
append-only events (alleged/cured/determined/terminated) grouped by
case_id rather than one mutable row - resolution is always a new,
later event, never an edit. A CHECK constraint makes the
anonymized-default rule (License V1C1 §7.4) a database fact:
named_entitlement_holder can be set if and only if anonymized = false.

src/target_revenue/breach_record.py's publish_breach_event() enforces
per-Licensor phase ownership and rejects named-disclosure requests
that don't also set named_disclosure_authorized_under_cua: true - the
Trust Service records the Licensor's assertion that the CUA's naming
clause authorizes it, it never verifies the underlying CUA text
itself. Signs every event with the same instance Ed25519 key already
used for Ledger entries and Attestations.

Adds POST/GET /phases/{id}/breach-records. Guarded the .registry
import behind a lazy in-function import (matching attestation.py's
TYPE_CHECKING pattern) so tests/test_breach_record.py (7 tests) runs
under plain system Python with no psycopg dependency. 5 new
Docker-gated tests cover the default-anonymized lifecycle, the
named-disclosure authorization gate, cross-Licensor rejection,
signature verification, and DB-level UPDATE/DELETE rejection.

This closes WP-0006 again - all 9 tasks done.

src/target_revenue/breach_record.py

@@ -0,0 +1,171 @@
+"""Hosted Breach/Compliance Record publication (WP-0006-T09).
+
+Publishes the Licensor's own breach and termination determinations for a
+Phase, per `specs/TargetRevenueSourceLicense-V1C1.md` §7.4 and
+`specs/TechnicalSpecificationDocument.md` §4.1. This module never decides
+whether a breach occurred, was cured, or was terminated — it only records,
+signs, and publishes the Licensor's assertion of one of those facts,
+exactly like `attestation.py` records an already-true conversion rather
+than deciding one. The distinction between `alleged` and `determined`
+(and the further `cured`/`terminated` outcomes) is the Licensor's call, or
+a Commercial Use Agreement's dispute process's call — never this module's.
+
+Naming: default anonymized (Phase and category only, no party identified).
+Named disclosure requires the caller to affirmatively assert that the
+applicable Commercial Use Agreement's naming/disclosure clause (License
+§7.4, CUA V1C1 §9) authorizes it — this module records that assertion, it
+does not, and cannot, verify the underlying CUA text itself.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from . import hashing
+
+if TYPE_CHECKING:
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from psycopg import Connection
+
+    from .registry import Licensor
+
+_EVENT_TYPES = frozenset({"alleged", "cured", "determined", "terminated"})
+_REQUIRED_FIELDS = ("id", "case_id", "category", "event_at")
+
+
+def _validate(event_input: dict[str, Any]) -> list[str]:
+    errors = []
+    for field in _REQUIRED_FIELDS:
+        if not event_input.get(field):
+            errors.append(f"{field}: required")
+    event_type = event_input.get("event_type")
+    if event_type not in _EVENT_TYPES:
+        errors.append(
+            f"event_type: must be one of {sorted(_EVENT_TYPES)}, got {event_type!r}"
+        )
+    anonymized = event_input.get("anonymized", True)
+    named_holder = event_input.get("named_entitlement_holder")
+    authorized = event_input.get("named_disclosure_authorized_under_cua", False)
+    if not anonymized:
+        if not named_holder:
+            errors.append(
+                "named_entitlement_holder: required when anonymized is false"
+            )
+        if not authorized:
+            errors.append(
+                "named_disclosure_authorized_under_cua: must be true to publish a "
+                "named record — the Trust Service records this assertion, it does "
+                "not verify the underlying Commercial Use Agreement text "
+                "(License V1C1 §7.4)"
+            )
+    elif named_holder:
+        errors.append(
+            "named_entitlement_holder: must not be set when anonymized is true "
+            "(default)"
+        )
+    return errors
+
+
+def publish_breach_event(
+    conn: "Connection",
+    licensor: "Licensor",
+    phase_id: str,
+    event_input: dict[str, Any],
+    signing_key: "Ed25519PrivateKey",
+) -> dict[str, Any]:
+    """Validate, sign, and append one breach/compliance lifecycle event.
+
+    Never mutates a prior event for the same `case_id` — resolution
+    (cure, determination, termination) is always a new, separately dated
+    event, per License V1C1 §7.4's "update the record promptly upon
+    resolution," which this module implements as append, not edit
+    (`migrations/0004_breach_records.sql` grants trf_app no UPDATE/DELETE
+    on this table at all).
+    """
+    from .registry import RegistrationError, get_phase_manifest
+
+    manifest = get_phase_manifest(conn, phase_id)
+    if manifest is None:
+        raise RegistrationError(f"phase {phase_id!r} is not registered")
+
+    manifest_licensor = conn.execute(
+        "SELECT licensor_id FROM phase_manifests WHERE phase_id = %s", (phase_id,)
+    ).fetchone()[0]
+    if manifest_licensor != licensor.licensor_id:
+        raise RegistrationError(
+            "this token is not authorized to publish breach records for this Phase"
+        )
+
+    errors = _validate(event_input)
+    if errors:
+        raise RegistrationError("breach event rejected: " + "; ".join(errors))
+
+    anonymized = event_input.get("anonymized", True)
+    record = {
+        "id": event_input["id"],
+        "case_id": event_input["case_id"],
+        "phase": phase_id,
+        "event_type": event_input["event_type"],
+        "category": event_input["category"],
+        "event_at": event_input["event_at"],
+        "anonymized": anonymized,
+        "named_entitlement_holder": event_input.get("named_entitlement_holder") if not anonymized else None,
+        "evidence_reference": event_input.get("evidence_reference"),
+    }
+    signed = {**record, "signature": hashing.sign_record(record, signing_key)}
+
+    from psycopg.errors import UniqueViolation
+
+    try:
+        conn.execute(
+            """
+            INSERT INTO breach_records
+                (record_id, case_id, phase_id, event_type, category, event_at,
+                 anonymized, named_entitlement_holder, evidence_reference,
+                 published_by, signature)
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+            """,
+            (
+                signed["id"],
+                signed["case_id"],
+                phase_id,
+                signed["event_type"],
+                signed["category"],
+                signed["event_at"],
+                signed["anonymized"],
+                signed["named_entitlement_holder"],
+                signed["evidence_reference"],
+                licensor.licensor_id,
+                signed["signature"],
+            ),
+        )
+    except UniqueViolation as exc:
+        raise RegistrationError(f"breach record id {signed['id']!r} already exists") from exc
+
+    return signed
+
+
+def get_breach_records(conn: "Connection", phase_id: str) -> list[dict[str, Any]]:
+    """Public read: the Phase's full breach/compliance event history.
+
+    Anonymized events never carry `named_entitlement_holder` in the first
+    place (enforced at publish time and by the table's own CHECK
+    constraint); this function does not need to additionally strip it, but
+    the caller must never bypass this function to read the table directly
+    for a public-facing response.
+    """
+    rows = conn.execute(
+        """
+        SELECT record_id, case_id, event_type, category, event_at, anonymized,
+               named_entitlement_holder, evidence_reference, signature
+          FROM breach_records
+         WHERE phase_id = %s
+         ORDER BY sequence ASC
+        """,
+        (phase_id,),
+    ).fetchall()
+    columns = (
+        "id", "case_id", "event_type", "category", "event_at", "anonymized",
+        "named_entitlement_holder", "evidence_reference", "signature",
+    )
+    return [dict(zip(columns, row)) for row in rows]

src/target_revenue/service/app.py

@@ -18,7 +18,7 @@ from fastapi import Depends, FastAPI, HTTPException, Request
 from psycopg import Connection
 from psycopg_pool import ConnectionPool
 
-from .. import attestation, ledger, metrics, registry
+from .. import attestation, breach_record, ledger, metrics, registry
 from . import keys
 
 app = FastAPI(title="Target Revenue Trust Service — Registries", version="0.1.0")
@@ -174,3 +174,32 @@ def read_attestation(
         return attestation.publish_attestation(conn, manifest, entries, signing_key)
     except attestation.NotConvertedError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
+
+
+@app.post("/phases/{phase_id}/breach-records", status_code=201)
+def publish_breach_record(
+    phase_id: str,
+    event: dict[str, Any],
+    licensor: registry.Licensor = Depends(get_licensor),
+    conn: Connection = Depends(get_connection),
+    signing_key=Depends(get_signing_key),
+) -> dict[str, Any]:
+    """Publishes the Licensor's own breach/compliance determination — this
+    endpoint never adjudicates whether a breach occurred (TSD §4.1
+    Breach/Compliance Record row, License V1C1 §7.4)."""
+    try:
+        return breach_record.publish_breach_event(conn, licensor, phase_id, event, signing_key)
+    except registry.RegistrationError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+
+
+@app.get("/phases/{phase_id}/breach-records")
+def read_breach_records(
+    phase_id: str, conn: Connection = Depends(get_connection)
+) -> list[dict[str, Any]]:
+    """Public, unauthenticated: anonymized by default per License V1C1
+    §7.4; a named record only appears if it was published with the CUA
+    §9 opt-in authorization asserted."""
+    if registry.get_phase_manifest(conn, phase_id) is None:
+        raise HTTPException(status_code=404, detail="phase not found")
+    return breach_record.get_breach_records(conn, phase_id)