attest: schema v2 — Fingerprint.cites + blast_radius over audit graph

A Fingerprint may declare an optional ``cites: tuple[str, ...]`` of
other entities' chain hashes it vouches for. Citation is a claim of
dependency, not a verification — arborist does not re-verify the
cited chain at commit time. When a cited chain hash is later
falsified (KS / Mahalanobis drift, manual quarantine), the audit
graph yields a deterministic blast radius:

    blast_radius(conn, falsified_chain_hash) -> list[entity_id]

Pure read over ``audit_events`` where ``event_type='attest_fingerprint'``;
returns distinct, sorted entity_ids whose fingerprints cite the
falsified hash. Routing (quarantine, retest, downgrade) belongs to
a higher layer — blast_radius produces the graph, not the action.

Schema bumps ``fingerprint-v1 → fingerprint-v2``. Legacy v1 bodies
in storage (no ``cites`` key) reconstruct with ``cites=()`` via
``body.get("cites", ())`` in ``entity_chain_iter``; the
schema-version mismatch guard in ``commit_fingerprint`` still
rejects v1 commits going forward, as intended.

arborist/attest/__init__.py

@@ -18,6 +18,7 @@ sample inputs. No model is in the proof path.
 """
 
 from arborist.attest.chain import (
+    blast_radius,
     commit_fingerprint,
     entity_chain_hash,
     entity_chain_iter,
@@ -41,6 +42,7 @@ from arborist.attest.fingerprint import (
 __all__ = [
     "Fingerprint",
     "KSResult",
+    "blast_radius",
     "canonical_bytes",
     "commit_fingerprint",
     "entity_chain_hash",

arborist/attest/chain.py

@@ -154,6 +154,7 @@ def entity_chain_iter(
             skew_q10000=int(body["skew_q10000"]),
             error_taxonomy={k: int(v) for k, v in body.get("error_taxonomy", {}).items()},
             stress_params={k: int(v) for k, v in body.get("stress_params", {}).items()},
+            cites=tuple(body.get("cites", ())),
             schema_version=body["schema_version"],
         )
         yield n, event_hash, fp
@@ -165,3 +166,36 @@ def verify_entity_chain(
     """Return True iff the derived chain hash matches the claim."""
     actual = entity_chain_hash(conn, entity_id)
     return actual is not None and actual == claimed_chain_hash
+
+
+def blast_radius(conn: sqlite3.Connection, falsified_chain_hash: str) -> list[str]:
+    """Distinct entity_ids whose fingerprints cite the falsified chain hash.
+
+    When an entity's chain hash is later falsified (KS / Mahalanobis
+    drift, manual quarantine, etc.), this returns the set of entities
+    who had declared a citation on that exact chain state via
+    ``Fingerprint.cites``. Their trust posture is now contingent on
+    the falsification and should be re-examined.
+
+    Pure read; does NOT mutate state. Routing decisions (quarantine,
+    retest, downgrade) belong to a caller higher up the stack — the
+    blast graph is the input, not the action.
+
+    Linear scan over ``audit_events`` of type ``attest_fingerprint``;
+    O(N) per call. Acceptable while N is small; if it ever isn't, a
+    derived ``fingerprint_cites(falsified_chain, citing_entity_id)``
+    index is the next step. Don't build that until the pressure is
+    real.
+    """
+    import json
+
+    affected: set[str] = set()
+    cur = conn.execute(
+        "SELECT body FROM audit_events WHERE event_type=?",
+        (EVENT_TYPE,),
+    )
+    for (body_json,) in cur:
+        body = json.loads(body_json)
+        if falsified_chain_hash in body.get("cites", ()):
+            affected.add(body["entity_id"])
+    return sorted(affected)

arborist/attest/fingerprint.py

@@ -1,6 +1,6 @@
 """Fingerprint schema + canonical encoder + Merkle leaf hash.
 
-Schema version: ``fingerprint-v1``.
+Schema version: ``fingerprint-v2``.
 
 All numeric fields are integers. Latency lives in microseconds.
 Skew is quantized to a 4-decimal-place integer (``skew_q10000``).
@@ -13,6 +13,16 @@ the quantization boundary at unlucky moments, two hosts could differ
 by ±1 in the last digit of ``skew_q10000`` or one of the std/mean
 fields. Acceptable for v1; a future revision can compute moments in
 integer arithmetic if required.
+
+**v2 delta: optional ``cites`` field.** A fingerprint may declare a
+``cites`` tuple of other entities' chain hashes that it vouches for
+or depends on. The citation is a *claim of dependency*, not a
+verification — at commit time arborist does not re-verify the cited
+chain. The point is that if a cited chain hash is later falsified
+(KS / Mahalanobis drift, manual quarantine, etc.), the audit graph
+yields a deterministic blast radius via ``chain.blast_radius()``.
+Default empty tuple → equivalent to v1 semantics. Legacy v1 bodies
+in storage (no ``cites`` key) are reconstructed with ``cites=()``.
 """
 
 from __future__ import annotations
@@ -25,7 +35,7 @@ from typing import Mapping, Sequence
 
 from arborist.merkle import hash_leaf
 
-SCHEMA_VERSION = "fingerprint-v1"
+SCHEMA_VERSION = "fingerprint-v2"
 
 
 @dataclass(frozen=True)
@@ -58,6 +68,12 @@ class Fingerprint:
     - ``stress_params``: stress parameters in effect during the
       session (``concurrent_tasks``, ``time_limit_s``, ``interrupts``,
       etc.); caller-defined keys, integer values only.
+    - ``cites``: optional tuple of other entities' chain hashes this
+      fingerprint declares a dependency on. Order is significant
+      (caller decides) — citations are NOT sorted at canonicalization,
+      because the order itself carries meaning ("I observed A then B").
+      Default empty tuple. See module docstring for blast-radius
+      semantics.
     - ``schema_version``: pinned to ``SCHEMA_VERSION``.
     """
 
@@ -75,6 +91,7 @@ class Fingerprint:
     skew_q10000: int
     error_taxonomy: Mapping[str, int] = field(default_factory=dict)
     stress_params: Mapping[str, int] = field(default_factory=dict)
+    cites: tuple[str, ...] = ()
     schema_version: str = SCHEMA_VERSION
 
 
@@ -135,6 +152,7 @@ def from_samples(
     latency_samples_us: Sequence[int],
     error_taxonomy: Mapping[str, int] | None = None,
     stress_params: Mapping[str, int] | None = None,
+    cites: Sequence[str] = (),
 ) -> Fingerprint:
     """Compute a Fingerprint from a list of integer microsecond latencies.
 
@@ -165,6 +183,7 @@ def from_samples(
         skew_q10000=int(round(skew_f * 10000)),
         error_taxonomy=dict(error_taxonomy or {}),
         stress_params=dict(stress_params or {}),
+        cites=tuple(cites),
     )
 
 

tests/test_attest_chain.py

@@ -8,6 +8,7 @@ import pytest
 
 from arborist.attest import (
     Fingerprint,
+    blast_radius,
     commit_fingerprint,
     entity_chain_hash,
     entity_chain_iter,
@@ -22,7 +23,7 @@ from arborist.attest.fingerprint import SCHEMA_VERSION
 from arborist.store import connect
 
 
-def _fp(entity_id="alice", session_id="s-1", mean_us=1000):
+def _fp(entity_id="alice", session_id="s-1", mean_us=1000, cites=()):
     return Fingerprint(
         entity_id=entity_id,
         session_id=session_id,
@@ -38,6 +39,7 @@ def _fp(entity_id="alice", session_id="s-1", mean_us=1000):
         skew_q10000=0,
         error_taxonomy={},
         stress_params={},
+        cites=tuple(cites),
         schema_version=SCHEMA_VERSION,
     )
 
@@ -177,3 +179,93 @@ def test_event_hash_matches_global_audit_chain_invariant(tmp_path):
 
 def test_event_type_constant_is_pinned():
     assert EVENT_TYPE == "attest_fingerprint"
+
+
+def test_blast_radius_empty_when_no_events(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    assert blast_radius(conn, "a" * 64) == []
+
+
+def test_blast_radius_empty_when_no_citations(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    out = commit_fingerprint(conn, _fp(entity_id="alice"))
+    assert blast_radius(conn, out["entity_chain_hash"]) == []
+
+
+def test_blast_radius_finds_single_citer(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    a = commit_fingerprint(conn, _fp(entity_id="alice"))
+    commit_fingerprint(
+        conn,
+        _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)),
+    )
+    assert blast_radius(conn, a["entity_chain_hash"]) == ["bob"]
+
+
+def test_blast_radius_finds_multiple_citers_distinct_and_sorted(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    a = commit_fingerprint(conn, _fp(entity_id="alice"))
+    commit_fingerprint(
+        conn,
+        _fp(entity_id="carol", session_id="c-1", cites=(a["entity_chain_hash"],)),
+    )
+    commit_fingerprint(
+        conn,
+        _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)),
+    )
+    assert blast_radius(conn, a["entity_chain_hash"]) == ["bob", "carol"]
+
+
+def test_blast_radius_deduplicates_repeated_citers(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    a = commit_fingerprint(conn, _fp(entity_id="alice"))
+    commit_fingerprint(
+        conn,
+        _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)),
+    )
+    commit_fingerprint(
+        conn,
+        _fp(entity_id="bob", session_id="b-2", mean_us=1100, cites=(a["entity_chain_hash"],)),
+    )
+    # Same entity citing the same chain hash from two fingerprints → one row.
+    assert blast_radius(conn, a["entity_chain_hash"]) == ["bob"]
+
+
+def test_blast_radius_misses_unrelated_chain_hash(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    a = commit_fingerprint(conn, _fp(entity_id="alice"))
+    commit_fingerprint(
+        conn,
+        _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)),
+    )
+    assert blast_radius(conn, "f" * 64) == []
+
+
+def test_legacy_v1_body_without_cites_reconstructs_with_empty_tuple(tmp_path):
+    """v1 events in storage (no `cites` key) must round-trip through entity_chain_iter."""
+    from arborist.store import append_audit
+
+    conn = connect(tmp_path / "t.db")
+    legacy_body = {
+        "entity_id": "alice",
+        "session_id": "s-legacy",
+        "timestamp_utc": "2026-06-05T15:30:00Z",
+        "domain": "python-debug",
+        "host_hash": "0" * 64,
+        "count": 3,
+        "mean_us": 1000,
+        "std_us": 100,
+        "p50_us": 1000,
+        "p90_us": 1100,
+        "p99_us": 1200,
+        "skew_q10000": 0,
+        "error_taxonomy": {},
+        "stress_params": {},
+        "schema_version": "fingerprint-v1",
+    }
+    append_audit(conn, event_type=EVENT_TYPE, body=legacy_body, subject_root="alice")
+    seen = list(entity_chain_iter(conn, "alice"))
+    assert len(seen) == 1
+    fp = seen[0][2]
+    assert fp.cites == ()
+    assert fp.schema_version == "fingerprint-v1"