attest: drop v2 schema bump — no v1 records exist, cites is just a v1 field

The v1→v2 bump in 30aba0e was ceremony with no historical witnesses.
A sweep of every local SQLite (~/.arborist/shards/*.db,
~/.arborist/*.db) found zero ``attest_fingerprint`` audit rows; the
module was introduced on this branch with no commits outside test
fixtures. With no v1 bodies in the wild there is no canonical-bytes
divergence to mark — v2 is just better v1.

- ``SCHEMA_VERSION`` returns to ``fingerprint-v1``.
- ``cites: tuple[str, ...] = ()`` stays on the dataclass; it was the
  actual feature, not the version bump.
- ``entity_chain_iter`` reads ``body["cites"]`` directly — no
  ``.get(..., ())`` defensive fallback for a missing-field case that
  cannot arise.
- ``test_legacy_v1_body_without_cites_reconstructs_with_empty_tuple``
  removed — it tested a backwards-compat path with no caller.
- ``blast_radius`` unchanged.

Net: -31 lines of compat scaffolding for a population of zero
records. Per CLAUDE.md "don't add backwards-compatibility shims when
you can just change the code" and the five-step algorithm step 2
("delete the part"). 66 attest tests pass.

arborist/attest/__init__.py

@@ -0,0 +1,58 @@
+"""Consistency attestation primitives.
+
+A `Fingerprint` is a canonical, integer-only summary of a session's
+observable behavior (latency moments, error taxonomy, stress params,
+host hash). Fingerprints hash via the project Merkle conventions
+(`arborist.merkle.hash_leaf`) and chain into the existing audit
+substrate.
+
+**Commitment-not-occurrence bound.** A fingerprint commitment asserts
+that the committed metrics were canonically encoded under this
+schema. It does NOT assert that the underlying behavior occurred,
+that the metrics describe a specific physical body, or that the
+session was unobserved. Body-binding is out of scope.
+
+**Pure-stats verifier.** Consistency tests (K-S / Mahalanobis) over
+fingerprint distributions are deterministic functions of their
+sample inputs. No model is in the proof path.
+"""
+
+from arborist.attest.chain import (
+    blast_radius,
+    commit_fingerprint,
+    entity_chain_hash,
+    entity_chain_iter,
+    verify_entity_chain,
+)
+from arborist.attest.consistency import (
+    KSResult,
+    estimate_cov,
+    estimate_mean,
+    invert_matrix,
+    ks_two_sample,
+    mahalanobis,
+)
+from arborist.attest.fingerprint import (
+    Fingerprint,
+    canonical_bytes,
+    from_samples,
+    leaf_hash,
+)
+
+__all__ = [
+    "Fingerprint",
+    "KSResult",
+    "blast_radius",
+    "canonical_bytes",
+    "commit_fingerprint",
+    "entity_chain_hash",
+    "entity_chain_iter",
+    "estimate_cov",
+    "estimate_mean",
+    "from_samples",
+    "invert_matrix",
+    "ks_two_sample",
+    "leaf_hash",
+    "mahalanobis",
+    "verify_entity_chain",
+]

arborist/attest/chain.py

@@ -0,0 +1,201 @@
+"""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()},
+            cites=tuple(body["cites"]),
+            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
+
+
+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/consistency.py

@@ -0,0 +1,196 @@
+"""Pure-stats consistency checks for Fingerprint sequences.
+
+Two deterministic, stdlib-only statistics:
+
+- **Two-sample Kolmogorov-Smirnov** on raw latency samples. Returns
+  the K-S statistic ``D`` and an asymptotic ``p`` from the
+  Kolmogorov distribution. ``D=0, p=1`` for identical CDFs; ``D=1,
+  p≈0`` for disjoint supports.
+- **Mahalanobis squared distance** of a moment vector from a
+  historical baseline ``(mean, inv_cov)``. Returns the raw
+  squared distance.
+
+Both functions return raw numbers. The caller picks the
+accept/reject threshold (e.g. ``reject if p < 0.05`` for K-S, or
+``reject if mahalanobis > chi2_95(k)`` for k-degree moment
+vectors). Tuning a threshold by training a discriminator against
+the corpus = sneaking a soft classifier in the side door —
+forbidden by the project's verifier-stays-binary rule.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from typing import Sequence
+
+
+@dataclass(frozen=True)
+class KSResult:
+    """Two-sample Kolmogorov-Smirnov result."""
+
+    statistic: float
+    p_value: float
+    n1: int
+    n2: int
+
+
+def _empirical_cdf_step(
+    sorted_samples: Sequence[int | float], x: float
+) -> float:
+    """Fraction of sorted_samples <= x."""
+    # Binary search for rightmost index with value <= x.
+    lo, hi = 0, len(sorted_samples)
+    while lo < hi:
+        mid = (lo + hi) // 2
+        if sorted_samples[mid] <= x:
+            lo = mid + 1
+        else:
+            hi = mid
+    return lo / len(sorted_samples)
+
+
+def _ks_p_value(d: float, n1: int, n2: int, terms: int = 100) -> float:
+    """Asymptotic p-value for two-sample K-S via the Kolmogorov series.
+
+    Q(x) = 2 * sum_{k=1..inf} (-1)^(k-1) * exp(-2 * k^2 * x^2)
+    with x = D * sqrt(n_eff) and n_eff = n1*n2 / (n1+n2). Series
+    converges fast; 100 terms is overkill for any practical D > 0.
+    """
+    if n1 == 0 or n2 == 0:
+        return 1.0
+    if d == 0.0:
+        return 1.0
+    n_eff = (n1 * n2) / (n1 + n2)
+    x = d * math.sqrt(n_eff)
+    s = 0.0
+    sign = 1.0
+    for k in range(1, terms + 1):
+        term = sign * math.exp(-2.0 * k * k * x * x)
+        s += term
+        sign = -sign
+        if abs(term) < 1e-20:
+            break
+    p = 2.0 * s
+    return max(0.0, min(1.0, p))
+
+
+def ks_two_sample(
+    samples_a: Sequence[int | float], samples_b: Sequence[int | float]
+) -> KSResult:
+    """Two-sample Kolmogorov-Smirnov.
+
+    Raises ``ValueError`` if either sample is empty.
+    """
+    n1, n2 = len(samples_a), len(samples_b)
+    if n1 == 0 or n2 == 0:
+        raise ValueError("samples must be non-empty")
+    a = sorted(samples_a)
+    b = sorted(samples_b)
+    # Evaluate D at every distinct value present in either sample.
+    # The maximum |F1 - F2| is always reached at one of these points.
+    grid = sorted(set(a) | set(b))
+    d = 0.0
+    for x in grid:
+        f1 = _empirical_cdf_step(a, x)
+        f2 = _empirical_cdf_step(b, x)
+        gap = abs(f1 - f2)
+        if gap > d:
+            d = gap
+    p = _ks_p_value(d, n1, n2)
+    return KSResult(statistic=d, p_value=p, n1=n1, n2=n2)
+
+
+def estimate_mean(vectors: Sequence[Sequence[float]]) -> list[float]:
+    """Sample mean of N row-vectors of length k."""
+    if not vectors:
+        raise ValueError("vectors is empty")
+    k = len(vectors[0])
+    if any(len(v) != k for v in vectors):
+        raise ValueError("vectors must be same length")
+    n = len(vectors)
+    return [sum(v[i] for v in vectors) / n for i in range(k)]
+
+
+def estimate_cov(
+    vectors: Sequence[Sequence[float]], mean: Sequence[float] | None = None
+) -> list[list[float]]:
+    """Sample covariance (n-1 normalization) of N row-vectors of length k.
+
+    Returns a kxk matrix as a list of lists. Raises ``ValueError``
+    if ``n < 2`` (covariance undefined).
+    """
+    n = len(vectors)
+    if n < 2:
+        raise ValueError("need n>=2 vectors for sample covariance")
+    k = len(vectors[0])
+    mu = list(mean) if mean is not None else estimate_mean(vectors)
+    cov = [[0.0] * k for _ in range(k)]
+    for v in vectors:
+        for i in range(k):
+            di = v[i] - mu[i]
+            for j in range(k):
+                cov[i][j] += di * (v[j] - mu[j])
+    denom = n - 1
+    for i in range(k):
+        for j in range(k):
+            cov[i][j] /= denom
+    return cov
+
+
+def invert_matrix(m: Sequence[Sequence[float]]) -> list[list[float]]:
+    """Gauss-Jordan inverse of a square matrix.
+
+    Raises ``ValueError`` on singular input (pivot below 1e-12).
+    """
+    n = len(m)
+    if any(len(row) != n for row in m):
+        raise ValueError("matrix must be square")
+    # Build augmented [m | I] and reduce.
+    aug = [[float(m[i][j]) for j in range(n)] + [1.0 if i == j else 0.0 for j in range(n)] for i in range(n)]
+    for col in range(n):
+        # Partial pivot.
+        pivot = col
+        for r in range(col + 1, n):
+            if abs(aug[r][col]) > abs(aug[pivot][col]):
+                pivot = r
+        if abs(aug[pivot][col]) < 1e-12:
+            raise ValueError("matrix is singular")
+        if pivot != col:
+            aug[col], aug[pivot] = aug[pivot], aug[col]
+        # Scale pivot row.
+        p = aug[col][col]
+        for j in range(2 * n):
+            aug[col][j] /= p
+        # Eliminate other rows.
+        for r in range(n):
+            if r == col:
+                continue
+            factor = aug[r][col]
+            if factor == 0.0:
+                continue
+            for j in range(2 * n):
+                aug[r][j] -= factor * aug[col][j]
+    return [row[n:] for row in aug]
+
+
+def mahalanobis(
+    point: Sequence[float],
+    mean: Sequence[float],
+    inv_cov: Sequence[Sequence[float]],
+) -> float:
+    """Mahalanobis SQUARED distance from ``point`` to ``mean`` under ``inv_cov``.
+
+    ``return = (x - mu)^T * inv_cov * (x - mu)``. Always >= 0 for a
+    positive-definite ``inv_cov``. Caller compares against
+    chi-squared critical values for the dimensionality.
+    """
+    k = len(point)
+    if len(mean) != k:
+        raise ValueError("point and mean dimensionalities differ")
+    if len(inv_cov) != k or any(len(row) != k for row in inv_cov):
+        raise ValueError("inv_cov shape mismatch")
+    diff = [point[i] - mean[i] for i in range(k)]
+    # (M v)_i = sum_j M[i][j] * v[j]
+    mv = [sum(inv_cov[i][j] * diff[j] for j in range(k)) for i in range(k)]
+    return sum(diff[i] * mv[i] for i in range(k))

arborist/attest/fingerprint.py

@@ -0,0 +1,195 @@
+"""Fingerprint schema + canonical encoder + Merkle leaf hash.
+
+Schema version: ``fingerprint-v1``.
+
+All numeric fields are integers. Latency lives in microseconds.
+Skew is quantized to a 4-decimal-place integer (``skew_q10000``).
+This keeps canonical bytes bit-stable across hosts with different
+libm float repr — float math runs once at fingerprint construction,
+quantizes to int, never re-enters the canonical form.
+
+Cross-host stability has a known limit: if libm float drift crosses
+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.
+
+**``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.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import statistics
+from dataclasses import asdict, dataclass, field
+from typing import Mapping, Sequence
+
+from arborist.merkle import hash_leaf
+
+SCHEMA_VERSION = "fingerprint-v1"
+
+
+@dataclass(frozen=True)
+class Fingerprint:
+    """Canonical, integer-only summary of one observed session.
+
+    Field-by-field:
+
+    - ``entity_id``: opaque caller-supplied identifier. Treated as a
+      bytestring; this module does not interpret it.
+    - ``session_id``: opaque per-session identifier.
+    - ``timestamp_utc``: ISO-8601 UTC string (no timezone offset
+      shortcuts, no microseconds — second precision; canonical form
+      is enforced by the caller).
+    - ``domain``: scope tag (e.g. ``"python-debug"``); fingerprints
+      from different domains are not statistically comparable.
+    - ``host_hash``: hex-encoded SHA-256 of the canonical host
+      identity bytes (caller decides what those are: hostname +
+      machine-id + user, opus-mt-style hash-pinning, etc.).
+      Defends against multi-host stitching attacks where an
+      attacker would otherwise replay sessions captured under
+      different hardware as one continuous profile.
+    - ``count``: number of samples folded into the moments.
+    - ``mean_us`` / ``std_us``: latency mean and population std,
+      microseconds, integer.
+    - ``p50_us`` / ``p90_us`` / ``p99_us``: latency percentiles,
+      microseconds, integer.
+    - ``skew_q10000``: ``round(sample_skew * 10000)`` as int.
+    - ``error_taxonomy``: error-kind → count, integer counts only.
+    - ``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``.
+    """
+
+    entity_id: str
+    session_id: str
+    timestamp_utc: str
+    domain: str
+    host_hash: str
+    count: int
+    mean_us: int
+    std_us: int
+    p50_us: int
+    p90_us: int
+    p99_us: int
+    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
+
+
+def _canonical_json(obj) -> str:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def canonical_bytes(fp: Fingerprint) -> bytes:
+    """Canonical UTF-8 bytes for a fingerprint.
+
+    Deterministic across runs and hosts when every payload field is
+    an integer or a fixed string.
+    """
+    payload = asdict(fp)
+    payload["error_taxonomy"] = dict(payload["error_taxonomy"])
+    payload["stress_params"] = dict(payload["stress_params"])
+    return _canonical_json(payload).encode("utf-8", errors="surrogatepass")
+
+
+def leaf_hash(fp: Fingerprint) -> bytes:
+    """SHA-256 leaf hash with project Merkle leaf prefix."""
+    return hash_leaf(canonical_bytes(fp))
+
+
+def _sample_skew(samples: Sequence[int], mean: float, std: float) -> float:
+    """Population skewness; returns 0.0 when std is 0 or n<2."""
+    n = len(samples)
+    if n < 2 or std == 0.0:
+        return 0.0
+    s = 0.0
+    for x in samples:
+        z = (x - mean) / std
+        s += z * z * z
+    return s / n
+
+
+def _percentile(sorted_samples: Sequence[int], q: float) -> int:
+    """Nearest-rank percentile on a pre-sorted integer sequence."""
+    n = len(sorted_samples)
+    if n == 0:
+        raise ValueError("no samples")
+    if n == 1:
+        return sorted_samples[0]
+    # Nearest-rank: ceil(q * n), 1-indexed.
+    import math
+
+    rank = max(1, min(n, math.ceil(q * n)))
+    return sorted_samples[rank - 1]
+
+
+def from_samples(
+    *,
+    entity_id: str,
+    session_id: str,
+    timestamp_utc: str,
+    domain: str,
+    host_hash: str,
+    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.
+
+    Raises ValueError on empty samples.
+    """
+    n = len(latency_samples_us)
+    if n == 0:
+        raise ValueError("latency_samples_us is empty")
+    for v in latency_samples_us:
+        if not isinstance(v, int):
+            raise TypeError("latency samples must be int (microseconds)")
+    sorted_samples = sorted(latency_samples_us)
+    mean_f = statistics.fmean(latency_samples_us)
+    std_f = statistics.pstdev(latency_samples_us) if n > 1 else 0.0
+    skew_f = _sample_skew(latency_samples_us, mean_f, std_f)
+    return Fingerprint(
+        entity_id=entity_id,
+        session_id=session_id,
+        timestamp_utc=timestamp_utc,
+        domain=domain,
+        host_hash=host_hash,
+        count=n,
+        mean_us=int(round(mean_f)),
+        std_us=int(round(std_f)),
+        p50_us=_percentile(sorted_samples, 0.50),
+        p90_us=_percentile(sorted_samples, 0.90),
+        p99_us=_percentile(sorted_samples, 0.99),
+        skew_q10000=int(round(skew_f * 10000)),
+        error_taxonomy=dict(error_taxonomy or {}),
+        stress_params=dict(stress_params or {}),
+        cites=tuple(cites),
+    )
+
+
+def host_hash_from_parts(*parts: str) -> str:
+    """Helper: hex SHA-256 over canonical bytes of a list of strings.
+
+    Caller decides what parts make up host identity. Example:
+    ``host_hash_from_parts(hostname, machine_id, user)``.
+    """
+    payload = _canonical_json(list(parts)).encode("utf-8", errors="surrogatepass")
+    return hashlib.sha256(payload).hexdigest()

arborist/cli.py

@@ -365,6 +365,155 @@ def _cmd_verify(args: argparse.Namespace) -> int:
     return 0 if result["failed"] == 0 else 1
 
 
+def _read_latency_samples_file(path: Path) -> list[int]:
+    """One integer (microseconds) per line; ``#`` lines and blanks ignored."""
+    samples: list[int] = []
+    with open(path) as f:
+        for line in f:
+            line = line.strip()
+            if not line or line.startswith("#"):
+                continue
+            samples.append(int(line))
+    return samples
+
+
+def _now_utc_iso() -> str:
+    import datetime
+    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _cmd_attest_commit(args: argparse.Namespace) -> int:
+    """Build a Fingerprint from a samples file and commit to the entity chain."""
+    from arborist.attest import commit_fingerprint, from_samples
+
+    samples = _read_latency_samples_file(args.samples_file)
+    fp = from_samples(
+        entity_id=args.entity_id,
+        session_id=args.session_id,
+        timestamp_utc=args.timestamp_utc or _now_utc_iso(),
+        domain=args.domain,
+        host_hash=args.host_hash,
+        latency_samples_us=samples,
+    )
+    conn = connect(args.db)
+    try:
+        result = commit_fingerprint(conn, fp)
+    finally:
+        conn.close()
+    print(json.dumps(result, indent=2))
+    return 0
+
+
+def _cmd_attest_verify(args: argparse.Namespace) -> int:
+    """Verify a claimed per-entity chain hash."""
+    from arborist.attest import verify_entity_chain
+
+    conn = connect(args.db)
+    try:
+        ok = verify_entity_chain(conn, args.entity_id, args.chain_hash)
+    finally:
+        conn.close()
+    print(json.dumps(
+        {"ok": ok, "entity_id": args.entity_id, "claim": args.chain_hash},
+        indent=2,
+    ))
+    return 0 if ok else 1
+
+
+def _cmd_attest_chain(args: argparse.Namespace) -> int:
+    """List the entity's fingerprint chain."""
+    from arborist.attest import entity_chain_hash, entity_chain_iter
+
+    conn = connect(args.db)
+    try:
+        rows = []
+        for seq, event_hash, fp in entity_chain_iter(conn, args.entity_id):
+            rows.append({
+                "seq": seq,
+                "event_hash": event_hash,
+                "session_id": fp.session_id,
+                "domain": fp.domain,
+                "timestamp_utc": fp.timestamp_utc,
+                "count": fp.count,
+                "mean_us": fp.mean_us,
+                "std_us": fp.std_us,
+                "skew_q10000": fp.skew_q10000,
+            })
+        head = entity_chain_hash(conn, args.entity_id)
+    finally:
+        conn.close()
+    print(json.dumps(
+        {"entity_id": args.entity_id, "head": head, "n": len(rows), "events": rows},
+        indent=2,
+    ))
+    return 0
+
+
+def _cmd_attest_check(args: argparse.Namespace) -> int:
+    """Mahalanobis squared distance of new-session moments vs entity history.
+
+    Threshold is NOT applied here. Caller compares against
+    a chi-squared critical value for the 3-d moment vector
+    (e.g. ``chi2.ppf(0.95, df=3) ≈ 7.815`` for α=0.05).
+    """
+    from arborist.attest import (
+        entity_chain_iter,
+        estimate_cov,
+        estimate_mean,
+        from_samples,
+        invert_matrix,
+        mahalanobis,
+    )
+
+    samples = _read_latency_samples_file(args.samples_file)
+    new_fp = from_samples(
+        entity_id=args.entity_id,
+        session_id="_check",
+        timestamp_utc=_now_utc_iso(),
+        domain=args.domain or "_check",
+        host_hash="0" * 64,
+        latency_samples_us=samples,
+    )
+    conn = connect(args.db)
+    try:
+        history = [
+            [float(fp.mean_us), float(fp.std_us), float(fp.skew_q10000)]
+            for _seq, _h, fp in entity_chain_iter(conn, args.entity_id)
+        ]
+    finally:
+        conn.close()
+    if len(history) < 2:
+        print(json.dumps({
+            "error": "need at least 2 prior fingerprints to estimate covariance",
+            "n_history": len(history),
+        }), file=sys.stderr)
+        return 2
+    mu = estimate_mean(history)
+    cov = estimate_cov(history, mean=mu)
+    try:
+        inv = invert_matrix(cov)
+    except ValueError as e:
+        print(json.dumps({
+            "error": "history covariance is singular; need uncorrelated history",
+            "detail": str(e),
+        }), file=sys.stderr)
+        return 2
+    point = [
+        float(new_fp.mean_us),
+        float(new_fp.std_us),
+        float(new_fp.skew_q10000),
+    ]
+    d2 = mahalanobis(point, mu, inv)
+    print(json.dumps({
+        "entity_id": args.entity_id,
+        "mahalanobis_d2": d2,
+        "n_history": len(history),
+        "point": point,
+        "mean": mu,
+    }, indent=2))
+    return 0
+
+
 def _cmd_distill(args: argparse.Namespace) -> int:
     """Distill existing documents into cores (surface→core, or core→core+1)."""
     from arborist.distill import get_distiller
@@ -7506,6 +7655,58 @@ def build_parser() -> argparse.ArgumentParser:
     )
     session_cmd.set_defaults(func=_cmd_session)
 
+    attest_cmd = sub.add_parser(
+        "attest",
+        help=(
+            "consistency attestation: commit / verify / inspect "
+            "fingerprint chains (humans or upstreams)"
+        ),
+    )
+    attest_sub = attest_cmd.add_subparsers(dest="attest_op", required=True)
+
+    attest_commit = attest_sub.add_parser(
+        "commit",
+        help="commit a fingerprint to the entity's chain",
+    )
+    attest_commit.add_argument("--entity-id", required=True)
+    attest_commit.add_argument("--session-id", required=True)
+    attest_commit.add_argument("--domain", required=True)
+    attest_commit.add_argument("--host-hash", required=True)
+    attest_commit.add_argument("--samples-file", required=True, type=Path)
+    attest_commit.add_argument(
+        "--timestamp-utc",
+        default=None,
+        help="ISO-8601 UTC; default = now",
+    )
+    attest_commit.set_defaults(func=_cmd_attest_commit)
+
+    attest_verify = attest_sub.add_parser(
+        "verify",
+        help="verify a claimed per-entity chain hash",
+    )
+    attest_verify.add_argument("--entity-id", required=True)
+    attest_verify.add_argument("--chain-hash", required=True)
+    attest_verify.set_defaults(func=_cmd_attest_verify)
+
+    attest_chain = attest_sub.add_parser(
+        "chain",
+        help="list the entity's fingerprint chain",
+    )
+    attest_chain.add_argument("--entity-id", required=True)
+    attest_chain.set_defaults(func=_cmd_attest_chain)
+
+    attest_check = attest_sub.add_parser(
+        "check",
+        help=(
+            "compute Mahalanobis squared distance of new-session moments "
+            "vs entity history; caller picks the chi-squared threshold"
+        ),
+    )
+    attest_check.add_argument("--entity-id", required=True)
+    attest_check.add_argument("--samples-file", required=True, type=Path)
+    attest_check.add_argument("--domain", default=None)
+    attest_check.set_defaults(func=_cmd_attest_check)
+
     return p
 
 

arborist/cold_pack_metadata.py

@@ -470,7 +470,19 @@ def _restore_generic_table(
     table: str,
     in_path: Path,
 ) -> int:
-    """Read array-per-line JSONL → INSERT OR IGNORE batches into `table`."""
+    """Read array-per-line JSONL → INSERT OR IGNORE batches into `table`.
+
+    Chunks special case (#53): the producer dumps a synthetic
+    ``_content_size`` column carrying each chunk's content byte-length
+    (the actual ``content`` BLOB is dropped from the metadata pack to
+    keep it small). Consumer drops the synthetic column from the INSERT
+    column list — chunks rows land with ``content IS NULL`` per
+    ``pull_metadata_pack``'s documented contract; phase 2
+    (``pull_chunk_pack``) fills the bytes via UPDATE WHERE leaf_hash=?.
+    The size hint is currently unused in the single-shard path; a
+    future zeroblob pre-allocation could read it back if page-split
+    cost becomes measurable.
+    """
     BATCH = 5000
     batch: list[tuple] = []
     cols: list[str] | None = None
@@ -478,7 +490,7 @@ def _restore_generic_table(
     insert_sql: str | None = None
     for row in read_columnar_jsonl(in_path):
         if cols is None:
-            cols = list(row.keys())
+            cols = [c for c in row.keys() if c != "_content_size"]
             placeholders = ", ".join("?" for _ in cols)
             col_list = ", ".join(f'"{c}"' for c in cols)
             insert_sql = (

arborist/store.py

@@ -12,6 +12,48 @@ Schema implements the Merkle-AGI v9.8 admissibility ledger:
 - derivations table binds core docs back to source surface roots
 
 The providence_cache layer is schema-only in Phase 0 — no Q&A inference yet.
+
+falsification_state transitions
+-------------------------------
+
+The four states are the substrate's negative-memory channel. Cache
+lookups filter on ``state='live'``; the rest exist to retain failure
+context without polluting reads. Allowed transitions, each written
+through ``append_audit`` so the chain witnesses every state change:
+
+- ``live → failed``: a deterministic verifier rejected the record
+  (evidence reroot, span mismatch, verifier-rule change exposes a
+  prior STRICT as no longer grounded). Hard rejection; not eligible
+  for re-promotion by the same policy.
+- ``live → stale``: drift was detected against current canonical
+  form, but hard rejection would be over-cautious — source updated
+  upstream, canonicalization bumped, retrieval surface shifted. The
+  record's prior claim may still be re-verifiable under current
+  state.
+- ``live → quarantined``: poisoning or tampering is suspected
+  (audit-chain anomaly, witness divergence, source attribution
+  anomaly). Removed from lookup *and* from re-promotion paths until
+  a governance event clears it.
+- ``stale → live``: re-verification against current canonical form
+  passes. The reopen is itself an audit event; the cache_key stays
+  bound to whatever policy was in effect at re-promotion.
+- ``stale → failed``: re-verification finds a hard contradiction
+  against current evidence. Downgrade is one-way under the same
+  policy.
+- ``stale → quarantined``: while stale, integrity evidence
+  accumulates against the record (e.g. cited evidence_root is now
+  contested). Escalation, not lateral move.
+- ``failed → live`` and ``quarantined → live`` are NOT default
+  paths. They require an explicit governance event (policy bump,
+  cited evidence re-rooted, attribution corrected) and a fresh
+  audit row naming the discharge reason. Same record, same
+  cache_key, but the chain records the rehabilitation, not a
+  silent flip.
+
+The rules are policy-level, not enforced as DB CHECK constraints —
+the CHECK above only constrains the column's domain, not its
+trajectory. Trajectory discipline lives in the helpers that mutate
+state and in ``make chain-check-shards``.
 """
 
 from __future__ import annotations

tests/test_attest_chain.py

@@ -0,0 +1,242 @@
+"""Tests for `arborist.attest.chain`."""
+
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+
+from arborist.attest import (
+    Fingerprint,
+    blast_radius,
+    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, cites=()):
+    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={},
+        cites=tuple(cites),
+        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"
+
+
+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) == []
+

tests/test_attest_cli.py

@@ -0,0 +1,178 @@
+"""End-to-end tests for `arborist attest` CLI."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from arborist.cli import build_parser
+
+
+def _write_samples(path, samples):
+    path.write_text("\n".join(str(s) for s in samples) + "\n")
+
+
+def _run(argv, capsys, expect_exit=None):
+    """Dispatch argv through build_parser → handler. Return parsed stdout."""
+    args = build_parser().parse_args(argv)
+    rc = args.func(args)
+    if expect_exit is not None:
+        assert rc == expect_exit
+    captured = capsys.readouterr()
+    return rc, captured.out, captured.err
+
+
+def test_attest_commit_then_chain_then_verify(tmp_path, capsys):
+    db = tmp_path / "t.db"
+    samples_file = tmp_path / "samples.txt"
+    _write_samples(samples_file, [1000, 1100, 1200, 1300, 1400])
+
+    rc, out, _ = _run([
+        "--db", str(db),
+        "attest", "commit",
+        "--entity-id", "alice",
+        "--session-id", "s-1",
+        "--domain", "python-debug",
+        "--host-hash", "0" * 64,
+        "--samples-file", str(samples_file),
+        "--timestamp-utc", "2026-06-05T15:30:00Z",
+    ], capsys)
+    assert rc == 0
+    commit_result = json.loads(out)
+    assert set(commit_result) == {"event_hash", "entity_chain_hash", "seq"}
+    assert commit_result["seq"] == 1
+
+    rc, out, _ = _run([
+        "--db", str(db),
+        "attest", "chain",
+        "--entity-id", "alice",
+    ], capsys)
+    assert rc == 0
+    chain = json.loads(out)
+    assert chain["n"] == 1
+    assert chain["head"] == commit_result["entity_chain_hash"]
+    assert chain["events"][0]["session_id"] == "s-1"
+    assert chain["events"][0]["mean_us"] == 1200
+
+    rc, _, _ = _run([
+        "--db", str(db),
+        "attest", "verify",
+        "--entity-id", "alice",
+        "--chain-hash", commit_result["entity_chain_hash"],
+    ], capsys)
+    assert rc == 0
+
+
+def test_attest_verify_wrong_hash_exits_nonzero(tmp_path, capsys):
+    db = tmp_path / "t.db"
+    samples_file = tmp_path / "s.txt"
+    _write_samples(samples_file, [1000, 1100, 1200])
+    _run([
+        "--db", str(db), "attest", "commit",
+        "--entity-id", "alice", "--session-id", "s-1",
+        "--domain", "d", "--host-hash", "0" * 64,
+        "--samples-file", str(samples_file),
+    ], capsys)
+    rc, _, _ = _run([
+        "--db", str(db), "attest", "verify",
+        "--entity-id", "alice",
+        "--chain-hash", "f" * 64,
+    ], capsys)
+    assert rc == 1
+
+
+def test_attest_check_needs_at_least_two_history(tmp_path, capsys):
+    db = tmp_path / "t.db"
+    s = tmp_path / "s.txt"
+    _write_samples(s, [1000, 1100, 1200])
+    _run([
+        "--db", str(db), "attest", "commit",
+        "--entity-id", "alice", "--session-id", "s-1",
+        "--domain", "d", "--host-hash", "0" * 64,
+        "--samples-file", str(s),
+    ], capsys)
+    rc, _, err = _run([
+        "--db", str(db), "attest", "check",
+        "--entity-id", "alice",
+        "--samples-file", str(s),
+    ], capsys)
+    assert rc == 2
+    msg = json.loads(err)
+    assert "at least 2 prior fingerprints" in msg["error"]
+
+
+def test_attest_check_emits_mahalanobis_after_uncorrelated_history(tmp_path, capsys):
+    """Three commits with intentionally independent moments → invertible cov."""
+    db = tmp_path / "t.db"
+    sessions = [
+        ("s-1", [950, 1000, 1050, 1100, 1150]),     # mean ~1050, std modest
+        ("s-2", [800, 1000, 1200, 1400, 1600]),     # mean ~1200, std wider
+        ("s-3", [1000, 1050, 1100, 1150, 1200]),    # mean ~1100, std small
+    ]
+    for sid, samples in sessions:
+        s = tmp_path / f"{sid}.txt"
+        _write_samples(s, samples)
+        _run([
+            "--db", str(db), "attest", "commit",
+            "--entity-id", "alice", "--session-id", sid,
+            "--domain", "d", "--host-hash", "0" * 64,
+            "--samples-file", str(s),
+        ], capsys)
+
+    new_samples = tmp_path / "new.txt"
+    _write_samples(new_samples, [1020, 1080, 1140, 1200, 1260])
+    rc, out, _ = _run([
+        "--db", str(db), "attest", "check",
+        "--entity-id", "alice",
+        "--samples-file", str(new_samples),
+    ], capsys)
+    # Either success (rc=0 with mahalanobis_d2) or singular-history rc=2.
+    # Both are valid outcomes; this test verifies the pipe end-to-end.
+    if rc == 0:
+        result = json.loads(out)
+        assert result["entity_id"] == "alice"
+        assert result["n_history"] == 3
+        assert result["mahalanobis_d2"] >= 0.0
+        assert len(result["point"]) == 3
+        assert len(result["mean"]) == 3
+    else:
+        assert rc == 2
+
+
+def test_attest_chain_empty_entity_returns_zero_events(tmp_path, capsys):
+    db = tmp_path / "t.db"
+    # Touch the db so connect() builds schema.
+    samples = tmp_path / "s.txt"
+    _write_samples(samples, [1000])
+    _run([
+        "--db", str(db), "attest", "commit",
+        "--entity-id", "alice", "--session-id", "s",
+        "--domain", "d", "--host-hash", "0" * 64,
+        "--samples-file", str(samples),
+    ], capsys)
+    rc, out, _ = _run([
+        "--db", str(db), "attest", "chain",
+        "--entity-id", "carol",
+    ], capsys)
+    assert rc == 0
+    body = json.loads(out)
+    assert body["entity_id"] == "carol"
+    assert body["n"] == 0
+    assert body["head"] is None
+    assert body["events"] == []
+
+
+def test_attest_commit_default_timestamp_when_omitted(tmp_path, capsys):
+    db = tmp_path / "t.db"
+    s = tmp_path / "s.txt"
+    _write_samples(s, [1000, 1100, 1200])
+    rc, out, _ = _run([
+        "--db", str(db), "attest", "commit",
+        "--entity-id", "alice", "--session-id", "s-1",
+        "--domain", "d", "--host-hash", "0" * 64,
+        "--samples-file", str(s),
+    ], capsys)
+    assert rc == 0
+    result = json.loads(out)
+    assert len(result["event_hash"]) == 64

tests/test_attest_consistency.py

@@ -0,0 +1,224 @@
+"""Tests for `arborist.attest.consistency`."""
+
+from __future__ import annotations
+
+import math
+
+import pytest
+
+from arborist.attest import (
+    KSResult,
+    estimate_cov,
+    estimate_mean,
+    invert_matrix,
+    ks_two_sample,
+    mahalanobis,
+)
+from arborist.attest.consistency import _empirical_cdf_step, _ks_p_value
+
+
+# ---------------------------------------------------------------- K-S
+
+
+def test_ks_identical_samples():
+    r = ks_two_sample([1, 2, 3, 4, 5], [1, 2, 3, 4, 5])
+    assert isinstance(r, KSResult)
+    assert r.statistic == 0.0
+    assert r.p_value == 1.0
+    assert r.n1 == 5 and r.n2 == 5
+
+
+def test_ks_disjoint_supports():
+    """D=1 needs n>=~6 per side for the asymptotic p to clear 0.05.
+
+    With n1=n2=10 and D=1, n_eff=5, x=sqrt(5)≈2.236,
+    Q ≈ 2*exp(-2*5) ≈ 9e-5 — well below 0.01.
+    """
+    r = ks_two_sample(list(range(1, 11)), list(range(100, 110)))
+    assert r.statistic == 1.0
+    assert r.p_value < 0.01
+
+
+def test_ks_monotone_as_samples_diverge():
+    """As B drifts further from A, K-S statistic should not decrease."""
+    a = [1000, 1100, 1200, 1300, 1400]
+    d_near = ks_two_sample(a, [1050, 1150, 1250, 1350, 1450]).statistic
+    d_far = ks_two_sample(a, [2000, 2100, 2200, 2300, 2400]).statistic
+    assert d_far >= d_near
+
+
+def test_ks_rejects_empty():
+    with pytest.raises(ValueError):
+        ks_two_sample([], [1, 2, 3])
+    with pytest.raises(ValueError):
+        ks_two_sample([1, 2, 3], [])
+
+
+def test_ks_p_value_bounded():
+    """p in [0, 1] for any D in [0, 1]."""
+    for d in (0.0, 0.1, 0.5, 0.9, 1.0):
+        p = _ks_p_value(d, 10, 10)
+        assert 0.0 <= p <= 1.0
+
+
+def test_ks_p_value_zero_D_is_one():
+    assert _ks_p_value(0.0, 100, 100) == 1.0
+
+
+def test_ks_handles_overlap_partial():
+    """Partial overlap → 0 < D < 1, 0 < p < 1."""
+    a = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+    b = [6, 7, 8, 9, 10, 11, 12, 13, 14, 15]
+    r = ks_two_sample(a, b)
+    assert 0.0 < r.statistic < 1.0
+    # Half-overlap with n=10 each → D=0.5; p moderate, not 0 or 1.
+    assert 0.0 < r.p_value < 1.0
+
+
+def test_empirical_cdf_step_basic():
+    s = sorted([10, 20, 30, 40, 50])
+    assert _empirical_cdf_step(s, 5) == 0.0
+    assert _empirical_cdf_step(s, 10) == 0.2
+    assert _empirical_cdf_step(s, 30) == 0.6
+    assert _empirical_cdf_step(s, 50) == 1.0
+    assert _empirical_cdf_step(s, 100) == 1.0
+
+
+# ---------------------------------------------------------------- mean / cov
+
+
+def test_estimate_mean_1d():
+    assert estimate_mean([[1.0], [2.0], [3.0]]) == [2.0]
+
+
+def test_estimate_mean_3d():
+    vecs = [[1.0, 10.0, 100.0], [2.0, 20.0, 200.0], [3.0, 30.0, 300.0]]
+    assert estimate_mean(vecs) == [2.0, 20.0, 200.0]
+
+
+def test_estimate_mean_rejects_ragged():
+    with pytest.raises(ValueError):
+        estimate_mean([[1.0, 2.0], [1.0]])
+
+
+def test_estimate_mean_rejects_empty():
+    with pytest.raises(ValueError):
+        estimate_mean([])
+
+
+def test_estimate_cov_independent_dimensions():
+    """Independent dimensions → diagonal covariance (within float epsilon)."""
+    # x ~ {1, 2, 3, 4, 5}; y constant; covariance off-diagonals = 0.
+    vecs = [[i, 0.0] for i in (1.0, 2.0, 3.0, 4.0, 5.0)]
+    cov = estimate_cov(vecs)
+    assert cov[0][1] == 0.0
+    assert cov[1][0] == 0.0
+
+
+def test_estimate_cov_rejects_n_lt_2():
+    with pytest.raises(ValueError):
+        estimate_cov([[1.0, 2.0]])
+
+
+def test_estimate_cov_symmetric():
+    vecs = [[1.0, 2.0], [3.0, 1.0], [5.0, 4.0], [2.0, 7.0]]
+    cov = estimate_cov(vecs)
+    assert math.isclose(cov[0][1], cov[1][0])
+
+
+# ---------------------------------------------------------------- invert
+
+
+def test_invert_identity_is_identity():
+    inv = invert_matrix([[1.0, 0.0, 0.0], [0.0, 1.0, 0.0], [0.0, 0.0, 1.0]])
+    for i in range(3):
+        for j in range(3):
+            expected = 1.0 if i == j else 0.0
+            assert math.isclose(inv[i][j], expected, abs_tol=1e-12)
+
+
+def test_invert_diagonal():
+    inv = invert_matrix([[2.0, 0.0], [0.0, 4.0]])
+    assert math.isclose(inv[0][0], 0.5)
+    assert math.isclose(inv[1][1], 0.25)
+    assert math.isclose(inv[0][1], 0.0)
+    assert math.isclose(inv[1][0], 0.0)
+
+
+def test_invert_round_trip_3x3():
+    m = [[4.0, 7.0, 2.0], [3.0, 5.0, 1.0], [2.0, 1.0, 3.0]]
+    inv = invert_matrix(m)
+    # m * inv ≈ I
+    for i in range(3):
+        for j in range(3):
+            s = sum(m[i][k] * inv[k][j] for k in range(3))
+            expected = 1.0 if i == j else 0.0
+            assert math.isclose(s, expected, abs_tol=1e-10)
+
+
+def test_invert_rejects_singular():
+    with pytest.raises(ValueError):
+        invert_matrix([[1.0, 2.0], [2.0, 4.0]])
+
+
+def test_invert_rejects_non_square():
+    with pytest.raises(ValueError):
+        invert_matrix([[1.0, 2.0], [3.0, 4.0], [5.0, 6.0]])
+
+
+# ---------------------------------------------------------------- mahalanobis
+
+
+def test_mahalanobis_at_mean_is_zero():
+    mean = [10.0, 20.0]
+    inv_cov = [[1.0, 0.0], [0.0, 1.0]]
+    assert mahalanobis(mean, mean, inv_cov) == 0.0
+
+
+def test_mahalanobis_identity_cov_is_squared_euclidean():
+    inv_cov = [[1.0, 0.0], [0.0, 1.0]]
+    d2 = mahalanobis([3.0, 4.0], [0.0, 0.0], inv_cov)
+    assert math.isclose(d2, 25.0)  # 3^2 + 4^2
+
+
+def test_mahalanobis_uses_inverse_covariance():
+    """Stretching inv_cov along an axis should scale that axis's contribution."""
+    inv_cov_iso = [[1.0, 0.0], [0.0, 1.0]]
+    inv_cov_x_heavy = [[4.0, 0.0], [0.0, 1.0]]  # x distance counts 4x more
+    d_iso = mahalanobis([1.0, 0.0], [0.0, 0.0], inv_cov_iso)
+    d_xh = mahalanobis([1.0, 0.0], [0.0, 0.0], inv_cov_x_heavy)
+    assert math.isclose(d_iso, 1.0)
+    assert math.isclose(d_xh, 4.0)
+
+
+def test_mahalanobis_dimension_mismatch():
+    with pytest.raises(ValueError):
+        mahalanobis([1.0, 2.0], [0.0, 0.0, 0.0], [[1.0, 0.0], [0.0, 1.0]])
+
+
+def test_mahalanobis_inv_cov_shape_mismatch():
+    with pytest.raises(ValueError):
+        mahalanobis([1.0, 2.0], [0.0, 0.0], [[1.0, 0.0, 0.0], [0.0, 1.0, 0.0]])
+
+
+def test_mahalanobis_end_to_end_with_estimate():
+    """Estimate mean+cov from history → invert → score a new point.
+
+    Dimensions are chosen to vary independently so the sample
+    covariance is well away from singular. Callers must supply
+    non-singular history; Mahalanobis is undefined otherwise.
+    """
+    history = [
+        [1000.0, 100.0, 0.0],
+        [1100.0, 95.0, 10.0],
+        [950.0, 120.0, -5.0],
+        [1050.0, 85.0, 5.0],
+        [1020.0, 110.0, -2.0],
+    ]
+    mu = estimate_mean(history)
+    cov = estimate_cov(history, mean=mu)
+    inv_cov = invert_matrix(cov)
+    d_near = mahalanobis([1020.0, 100.0, 2.0], mu, inv_cov)
+    d_far = mahalanobis([5000.0, 500.0, 100.0], mu, inv_cov)
+    assert d_far > d_near
+    assert d_near >= 0.0

tests/test_attest_fingerprint.py

@@ -0,0 +1,159 @@
+"""Tests for `arborist.attest.fingerprint`."""
+
+from __future__ import annotations
+
+import hashlib
+
+import pytest
+
+from arborist.attest import (
+    Fingerprint,
+    canonical_bytes,
+    from_samples,
+    leaf_hash,
+)
+from arborist.attest.fingerprint import (
+    SCHEMA_VERSION,
+    _percentile,
+    _sample_skew,
+    host_hash_from_parts,
+)
+from arborist.merkle import LEAF_PREFIX
+
+
+def _fp(**overrides):
+    base = dict(
+        entity_id="alice",
+        session_id="s-1",
+        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=SCHEMA_VERSION,
+    )
+    base.update(overrides)
+    return Fingerprint(**base)
+
+
+def test_canonical_bytes_is_deterministic():
+    fp = _fp()
+    assert canonical_bytes(fp) == canonical_bytes(fp)
+
+
+def test_canonical_bytes_is_key_sorted():
+    """Constructor key order must not affect canonical bytes."""
+    fp_a = _fp(error_taxonomy={"a": 1, "b": 2}, stress_params={"x": 1, "y": 2})
+    fp_b = _fp(error_taxonomy={"b": 2, "a": 1}, stress_params={"y": 2, "x": 1})
+    assert canonical_bytes(fp_a) == canonical_bytes(fp_b)
+
+
+def test_canonical_bytes_distinguishes_payloads():
+    assert canonical_bytes(_fp()) != canonical_bytes(_fp(mean_us=1001))
+    assert canonical_bytes(_fp()) != canonical_bytes(_fp(entity_id="bob"))
+    assert canonical_bytes(_fp()) != canonical_bytes(_fp(domain="js-debug"))
+    assert canonical_bytes(_fp()) != canonical_bytes(_fp(host_hash="f" * 64))
+
+
+def test_leaf_hash_uses_project_merkle_prefix():
+    fp = _fp()
+    expected = hashlib.sha256(LEAF_PREFIX + canonical_bytes(fp)).digest()
+    assert leaf_hash(fp) == expected
+    assert len(leaf_hash(fp)) == 32
+
+
+def test_leaf_hash_distinguishes_distinct_fingerprints():
+    assert leaf_hash(_fp()) != leaf_hash(_fp(session_id="s-2"))
+
+
+def test_from_samples_computes_moments():
+    samples = [1000, 1100, 1200, 1300, 1400]
+    fp = from_samples(
+        entity_id="alice",
+        session_id="s-1",
+        timestamp_utc="2026-06-05T15:30:00Z",
+        domain="python-debug",
+        host_hash="0" * 64,
+        latency_samples_us=samples,
+    )
+    assert fp.count == 5
+    assert fp.mean_us == 1200
+    # Nearest-rank percentile at q=0.50 on n=5 = rank ceil(2.5)=3 → 1200.
+    assert fp.p50_us == 1200
+    # q=0.90 → rank ceil(4.5)=5 → 1400.
+    assert fp.p90_us == 1400
+    # q=0.99 → rank ceil(4.95)=5 → 1400.
+    assert fp.p99_us == 1400
+    # Symmetric arithmetic progression → skew = 0 → q10000 = 0.
+    assert fp.skew_q10000 == 0
+
+
+def test_from_samples_rejects_empty():
+    with pytest.raises(ValueError):
+        from_samples(
+            entity_id="alice",
+            session_id="s-1",
+            timestamp_utc="2026-06-05T15:30:00Z",
+            domain="python-debug",
+            host_hash="0" * 64,
+            latency_samples_us=[],
+        )
+
+
+def test_from_samples_rejects_floats():
+    with pytest.raises(TypeError):
+        from_samples(
+            entity_id="alice",
+            session_id="s-1",
+            timestamp_utc="2026-06-05T15:30:00Z",
+            domain="python-debug",
+            host_hash="0" * 64,
+            latency_samples_us=[1.0, 2.0, 3.0],  # type: ignore[list-item]
+        )
+
+
+def test_sample_skew_zero_on_constant():
+    assert _sample_skew([5, 5, 5], 5.0, 0.0) == 0.0
+
+
+def test_sample_skew_positive_for_right_tail():
+    # Long right tail → positive skew.
+    samples = [1, 1, 1, 1, 1, 100]
+    import statistics
+
+    mean = statistics.fmean(samples)
+    std = statistics.pstdev(samples)
+    assert _sample_skew(samples, mean, std) > 0
+
+
+def test_percentile_nearest_rank():
+    s = [10, 20, 30, 40, 50]
+    assert _percentile(s, 0.0) == 10
+    assert _percentile(s, 0.5) == 30
+    assert _percentile(s, 1.0) == 50
+
+
+def test_host_hash_from_parts_is_deterministic_and_order_sensitive():
+    h_ab = host_hash_from_parts("host", "machine-id")
+    h_ba = host_hash_from_parts("machine-id", "host")
+    assert h_ab == host_hash_from_parts("host", "machine-id")
+    assert h_ab != h_ba  # order matters; caller picks the canonical order
+
+
+def test_schema_version_is_pinned_in_default_construction():
+    fp = from_samples(
+        entity_id="alice",
+        session_id="s-1",
+        timestamp_utc="2026-06-05T15:30:00Z",
+        domain="python-debug",
+        host_hash="0" * 64,
+        latency_samples_us=[1000],
+    )
+    assert fp.schema_version == SCHEMA_VERSION