arborist/attest/__init__.py

@@ -1,58 +0,0 @@
-"""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

@@ -1,201 +0,0 @@
-"""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

@@ -1,196 +0,0 @@
-"""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

@@ -1,195 +0,0 @@
-"""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,155 +365,6 @@ 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
@@ -7655,58 +7506,6 @@ 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,19 +470,7 @@ def _restore_generic_table(
     table: str,
     in_path: Path,
 ) -> int:
-    """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.
-    """
+    """Read array-per-line JSONL → INSERT OR IGNORE batches into `table`."""
     BATCH = 5000
     batch: list[tuple] = []
     cols: list[str] | None = None
@@ -490,7 +478,7 @@ def _restore_generic_table(
     insert_sql: str | None = None
     for row in read_columnar_jsonl(in_path):
         if cols is None:
-            cols = [c for c in row.keys() if c != "_content_size"]
+            cols = list(row.keys())
             placeholders = ", ".join("?" for _ in cols)
             col_list = ", ".join(f'"{c}"' for c in cols)
             insert_sql = (

arborist/store.py

@@ -12,48 +12,6 @@ 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

@@ -1,242 +0,0 @@
-"""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

@@ -1,178 +0,0 @@
-"""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

@@ -1,224 +0,0 @@
-"""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

@@ -1,159 +0,0 @@
-"""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