attest: per-entity FingerprintChain on top of audit_events

commit_fingerprint() wraps append_audit with event_type='attest_fingerprint'
and subject_root=entity_id. Per-entity chain hash is derived at query time
by folding sha256(prev_chain || event_hash) over the entity's filtered
event sequence, with genesis prev_chain = ZERO_HASH.

Two layers of tamper-evidence: mutating any audit row breaks the global
event_hash chain (existing invariant); mutating, reordering, or deleting
a fingerprint breaks the derived per-entity chain hash.

No new SQL table — head-pointer optimization deferred until long chains
warrant it.

15 chain tests + 13 fingerprint tests = 28 pass.

arborist/attest/__init__.py

@@ -17,6 +17,12 @@ fingerprint distributions are deterministic functions of their
 sample inputs. No model is in the proof path.
 """
 
+from arborist.attest.chain import (
+    commit_fingerprint,
+    entity_chain_hash,
+    entity_chain_iter,
+    verify_entity_chain,
+)
 from arborist.attest.fingerprint import (
     Fingerprint,
     canonical_bytes,
@@ -24,4 +30,13 @@ from arborist.attest.fingerprint import (
     leaf_hash,
 )
 
-__all__ = ["Fingerprint", "canonical_bytes", "from_samples", "leaf_hash"]
+__all__ = [
+    "Fingerprint",
+    "canonical_bytes",
+    "commit_fingerprint",
+    "entity_chain_hash",
+    "entity_chain_iter",
+    "from_samples",
+    "leaf_hash",
+    "verify_entity_chain",
+]

arborist/attest/chain.py

@@ -0,0 +1,167 @@
+"""Per-entity fingerprint chain on top of `audit_events`.
+
+A FingerprintChain piggybacks on the existing append-only audit chain
+via `arborist.store.append_audit`. Each fingerprint commit becomes
+one `audit_events` row with `event_type='attest_fingerprint'` and
+`subject_root=fp.entity_id`.
+
+Per-entity chain hash is derived at query time by folding
+``sha256(prev_chain || event_hash)`` over the entity's filtered
+event sequence, with the genesis ``prev_chain`` fixed at 32 zero
+bytes (``arborist.merkle.ZERO_HASH``). Two layers of
+tamper-evidence:
+
+1. Mutating any audit row breaks the global ``event_hash`` chain
+   (existing arborist invariant).
+2. Mutating, reordering, or deleting a fingerprint breaks the
+   derived per-entity chain hash.
+
+No new SQL table at this phase. The per-entity head is computable
+from existing columns; adding a head-pointer table is a forward
+optimization for entities with very long chains.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import sqlite3
+from dataclasses import asdict
+from typing import Iterator
+
+from arborist.attest.fingerprint import Fingerprint, SCHEMA_VERSION
+from arborist.merkle import ZERO_HASH
+from arborist.store import append_audit
+
+EVENT_TYPE = "attest_fingerprint"
+GENESIS_PREV_CHAIN = ZERO_HASH.hex()
+
+
+def _fold(prev_chain_hex: str, event_hash_hex: str) -> str:
+    """One step of the per-entity chain hash."""
+    h = hashlib.sha256()
+    h.update(bytes.fromhex(prev_chain_hex))
+    h.update(bytes.fromhex(event_hash_hex))
+    return h.hexdigest()
+
+
+def commit_fingerprint(
+    conn: sqlite3.Connection,
+    fp: Fingerprint,
+    *,
+    ts: int | None = None,
+) -> dict:
+    """Append a Fingerprint to the entity's chain.
+
+    Returns ``{"event_hash", "entity_chain_hash", "seq"}`` where
+    ``seq`` is the 1-indexed per-entity sequence number after this
+    insert.
+
+    Raises ``ValueError`` if the fingerprint's ``schema_version``
+    does not match this module's pinned ``SCHEMA_VERSION``. This
+    guards against silently committing fingerprints encoded under
+    a future schema while the chain math still uses v1
+    assumptions.
+    """
+    if fp.schema_version != SCHEMA_VERSION:
+        raise ValueError(
+            f"fingerprint schema_version {fp.schema_version!r} != "
+            f"chain SCHEMA_VERSION {SCHEMA_VERSION!r}"
+        )
+
+    body = asdict(fp)
+    body["error_taxonomy"] = dict(body["error_taxonomy"])
+    body["stress_params"] = dict(body["stress_params"])
+
+    event_hash = append_audit(
+        conn,
+        event_type=EVENT_TYPE,
+        body=body,
+        subject_root=fp.entity_id,
+        ts=ts,
+    )
+
+    new_chain_hash = entity_chain_hash(conn, fp.entity_id)
+    assert new_chain_hash is not None, "chain just got an event; must not be None"
+    seq = _entity_count(conn, fp.entity_id)
+    return {
+        "event_hash": event_hash,
+        "entity_chain_hash": new_chain_hash,
+        "seq": seq,
+    }
+
+
+def _entity_count(conn: sqlite3.Connection, entity_id: str) -> int:
+    row = conn.execute(
+        "SELECT COUNT(*) FROM audit_events "
+        "WHERE event_type=? AND subject_root=?",
+        (EVENT_TYPE, entity_id),
+    ).fetchone()
+    return int(row[0])
+
+
+def _iter_entity_events(
+    conn: sqlite3.Connection, entity_id: str
+) -> Iterator[tuple[int, str, str]]:
+    """Yield ``(seq, event_hash, body_json)`` for the entity in chain order."""
+    cur = conn.execute(
+        "SELECT seq, event_hash, body FROM audit_events "
+        "WHERE event_type=? AND subject_root=? ORDER BY seq",
+        (EVENT_TYPE, entity_id),
+    )
+    for row in cur:
+        yield int(row[0]), row[1], row[2]
+
+
+def entity_chain_hash(conn: sqlite3.Connection, entity_id: str) -> str | None:
+    """Per-entity chain hash, derived from the entity's audit-event sequence.
+
+    Returns None if the entity has no fingerprints committed yet.
+    """
+    chain = GENESIS_PREV_CHAIN
+    any_event = False
+    for _seq, event_hash, _body in _iter_entity_events(conn, entity_id):
+        chain = _fold(chain, event_hash)
+        any_event = True
+    return chain if any_event else None
+
+
+def entity_chain_iter(
+    conn: sqlite3.Connection, entity_id: str
+) -> Iterator[tuple[int, str, Fingerprint]]:
+    """Yield ``(entity_seq, event_hash, Fingerprint)`` in chain order.
+
+    ``entity_seq`` is 1-indexed per entity (NOT the global audit
+    ``seq``). Reconstructs the Fingerprint from canonical body JSON.
+    """
+    import json
+
+    n = 0
+    for _global_seq, event_hash, body_json in _iter_entity_events(conn, entity_id):
+        n += 1
+        body = json.loads(body_json)
+        fp = Fingerprint(
+            entity_id=body["entity_id"],
+            session_id=body["session_id"],
+            timestamp_utc=body["timestamp_utc"],
+            domain=body["domain"],
+            host_hash=body["host_hash"],
+            count=int(body["count"]),
+            mean_us=int(body["mean_us"]),
+            std_us=int(body["std_us"]),
+            p50_us=int(body["p50_us"]),
+            p90_us=int(body["p90_us"]),
+            p99_us=int(body["p99_us"]),
+            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()},
+            schema_version=body["schema_version"],
+        )
+        yield n, event_hash, fp
+
+
+def verify_entity_chain(
+    conn: sqlite3.Connection, entity_id: str, claimed_chain_hash: str
+) -> bool:
+    """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

tests/test_attest_chain.py

@@ -0,0 +1,179 @@
+"""Tests for `arborist.attest.chain`."""
+
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+
+from arborist.attest import (
+    Fingerprint,
+    commit_fingerprint,
+    entity_chain_hash,
+    entity_chain_iter,
+    verify_entity_chain,
+)
+from arborist.attest.chain import (
+    EVENT_TYPE,
+    GENESIS_PREV_CHAIN,
+    _fold,
+)
+from arborist.attest.fingerprint import SCHEMA_VERSION
+from arborist.store import connect
+
+
+def _fp(entity_id="alice", session_id="s-1", mean_us=1000):
+    return Fingerprint(
+        entity_id=entity_id,
+        session_id=session_id,
+        timestamp_utc="2026-06-05T15:30:00Z",
+        domain="python-debug",
+        host_hash="0" * 64,
+        count=3,
+        mean_us=mean_us,
+        std_us=100,
+        p50_us=mean_us,
+        p90_us=mean_us + 100,
+        p99_us=mean_us + 200,
+        skew_q10000=0,
+        error_taxonomy={},
+        stress_params={},
+        schema_version=SCHEMA_VERSION,
+    )
+
+
+def test_commit_returns_event_and_chain_hash(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    out = commit_fingerprint(conn, _fp())
+    assert set(out) == {"event_hash", "entity_chain_hash", "seq"}
+    assert len(out["event_hash"]) == 64
+    assert len(out["entity_chain_hash"]) == 64
+    assert out["seq"] == 1
+
+
+def test_first_chain_hash_folds_genesis_and_event(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    out = commit_fingerprint(conn, _fp())
+    expected = _fold(GENESIS_PREV_CHAIN, out["event_hash"])
+    assert out["entity_chain_hash"] == expected
+
+
+def test_second_chain_hash_folds_prev_and_event(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    out1 = commit_fingerprint(conn, _fp(session_id="s-1"))
+    out2 = commit_fingerprint(conn, _fp(session_id="s-2", mean_us=1100))
+    assert out2["seq"] == 2
+    expected = _fold(out1["entity_chain_hash"], out2["event_hash"])
+    assert out2["entity_chain_hash"] == expected
+
+
+def test_entity_chain_hash_independent_across_entities(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    a1 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-1"))
+    b1 = commit_fingerprint(conn, _fp(entity_id="bob", session_id="b-1"))
+    assert a1["entity_chain_hash"] != b1["entity_chain_hash"]
+    assert a1["seq"] == 1 and b1["seq"] == 1
+
+
+def test_entity_chain_hash_independent_of_global_audit_position(tmp_path):
+    """Per-entity chain only counts that entity's attest events.
+
+    Even though alice and bob's events are interleaved in the global
+    audit_events table, alice's chain hash should be a function only
+    of alice's filtered event sequence.
+    """
+    conn = connect(tmp_path / "t.db")
+    a1 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-1"))
+    _b1 = commit_fingerprint(conn, _fp(entity_id="bob", session_id="b-1"))
+    a2 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-2", mean_us=1100))
+    expected = _fold(a1["entity_chain_hash"], a2["event_hash"])
+    assert a2["entity_chain_hash"] == expected
+    assert a2["seq"] == 2  # second alice event, not third global
+
+
+def test_entity_chain_iter_yields_in_order(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    commit_fingerprint(conn, _fp(session_id="s-1", mean_us=1000))
+    commit_fingerprint(conn, _fp(session_id="s-2", mean_us=1100))
+    commit_fingerprint(conn, _fp(session_id="s-3", mean_us=1200))
+    seen = list(entity_chain_iter(conn, "alice"))
+    assert [n for n, _h, _fp in seen] == [1, 2, 3]
+    assert [fp.session_id for _n, _h, fp in seen] == ["s-1", "s-2", "s-3"]
+    assert [fp.mean_us for _n, _h, fp in seen] == [1000, 1100, 1200]
+
+
+def test_entity_chain_iter_yields_nothing_for_unknown_entity(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    commit_fingerprint(conn, _fp(entity_id="alice"))
+    assert list(entity_chain_iter(conn, "carol")) == []
+
+
+def test_entity_chain_hash_returns_none_for_unknown_entity(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    commit_fingerprint(conn, _fp(entity_id="alice"))
+    assert entity_chain_hash(conn, "carol") is None
+
+
+def test_verify_entity_chain_accepts_correct(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    out = commit_fingerprint(conn, _fp())
+    assert verify_entity_chain(conn, "alice", out["entity_chain_hash"])
+
+
+def test_verify_entity_chain_rejects_wrong(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    commit_fingerprint(conn, _fp())
+    assert not verify_entity_chain(conn, "alice", "f" * 64)
+
+
+def test_commit_rejects_unknown_schema_version(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    base = _fp()
+    bad = Fingerprint(**{**base.__dict__, "schema_version": "fingerprint-v999"})
+    with pytest.raises(ValueError):
+        commit_fingerprint(conn, bad)
+
+
+def test_fingerprint_round_trips_through_chain(tmp_path):
+    conn = connect(tmp_path / "t.db")
+    original = _fp(session_id="s-1", mean_us=4242)
+    commit_fingerprint(conn, original)
+    recovered = next(entity_chain_iter(conn, "alice"))[2]
+    # Frozen dataclass equality covers all fields including the dict ones.
+    assert recovered == original
+
+
+def test_event_type_is_namespaced(tmp_path):
+    """Other event types with same subject_root must not pollute the chain."""
+    from arborist.store import append_audit
+
+    conn = connect(tmp_path / "t.db")
+    commit_fingerprint(conn, _fp(entity_id="alice"))
+    # Inject an unrelated event with the same subject_root.
+    append_audit(conn, event_type="ingest", body={"x": 1}, subject_root="alice")
+    commit_fingerprint(conn, _fp(entity_id="alice", session_id="s-2", mean_us=1100))
+    # Chain still counts only the two attest_fingerprint events.
+    seen = list(entity_chain_iter(conn, "alice"))
+    assert len(seen) == 2
+    assert [fp.session_id for _n, _h, fp in seen] == ["s-1", "s-2"]
+
+
+def test_event_hash_matches_global_audit_chain_invariant(tmp_path):
+    """sha256(prev_global || canonical_body) — the rule in store.py:1665-1675."""
+    import json
+    from dataclasses import asdict
+
+    conn = connect(tmp_path / "t.db")
+    out = commit_fingerprint(conn, _fp())
+    body = asdict(_fp())
+    body["error_taxonomy"] = dict(body["error_taxonomy"])
+    body["stress_params"] = dict(body["stress_params"])
+    body_json = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+    expected = hashlib.sha256(
+        body_json.encode("utf-8", errors="surrogatepass")
+    ).hexdigest()
+    assert out["event_hash"] == expected
+
+
+def test_event_type_constant_is_pinned():
+    assert EVENT_TYPE == "attest_fingerprint"