From e6c42773b0d42bc3bcb9b849a9a59f63ff3d7ac2 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 5 Jun 2026 11:58:30 -0400 Subject: [PATCH 1/8] attest: Fingerprint schema + canonical encoder + Merkle leaf hash First piece of digital-identity-proof-of-overlearning. arborist/attest/ ships an integer-only Fingerprint (latency in microseconds, skew quantized to 4dp int) so canonical bytes stay bit-stable across hosts. Commitment-not-occurrence bound is a hard invariant in arborist/attest/__init__.py: a fingerprint commits the encoded metrics under this schema; it does NOT assert occurrence or body-binding. Sibling repos close those gaps. 13 tests pass. --- arborist/attest/__init__.py | 27 +++++ arborist/attest/fingerprint.py | 178 +++++++++++++++++++++++++++++++ tests/test_attest_fingerprint.py | 159 +++++++++++++++++++++++++++ 3 files changed, 364 insertions(+) create mode 100644 arborist/attest/__init__.py create mode 100644 arborist/attest/fingerprint.py create mode 100644 tests/test_attest_fingerprint.py diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py new file mode 100644 index 0000000..0adf5da --- /dev/null +++ b/arborist/attest/__init__.py @@ -0,0 +1,27 @@ +"""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.fingerprint import ( + Fingerprint, + canonical_bytes, + from_samples, + leaf_hash, +) + +__all__ = ["Fingerprint", "canonical_bytes", "from_samples", "leaf_hash"] diff --git a/arborist/attest/fingerprint.py b/arborist/attest/fingerprint.py new file mode 100644 index 0000000..44f2eb9 --- /dev/null +++ b/arborist/attest/fingerprint.py @@ -0,0 +1,178 @@ +"""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. +""" + +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. + - ``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) + 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, +) -> 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 {}), + ) + + +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() diff --git a/tests/test_attest_fingerprint.py b/tests/test_attest_fingerprint.py new file mode 100644 index 0000000..6058035 --- /dev/null +++ b/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 From a1dd519ae7ba38f56065f7cea0a64b81b0583044 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 5 Jun 2026 12:11:57 -0400 Subject: [PATCH 2/8] attest: per-entity FingerprintChain on top of audit_events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit commit_fingerprint() wraps append_audit with event_type='attest_fingerprint' and subject_root=entity_id. Per-entity chain hash is derived at query time by folding sha256(prev_chain || event_hash) over the entity's filtered event sequence, with genesis prev_chain = ZERO_HASH. Two layers of tamper-evidence: mutating any audit row breaks the global event_hash chain (existing invariant); mutating, reordering, or deleting a fingerprint breaks the derived per-entity chain hash. No new SQL table — head-pointer optimization deferred until long chains warrant it. 15 chain tests + 13 fingerprint tests = 28 pass. --- arborist/attest/__init__.py | 17 +++- arborist/attest/chain.py | 167 +++++++++++++++++++++++++++++++++ tests/test_attest_chain.py | 179 ++++++++++++++++++++++++++++++++++++ 3 files changed, 362 insertions(+), 1 deletion(-) create mode 100644 arborist/attest/chain.py create mode 100644 tests/test_attest_chain.py diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py index 0adf5da..0d8d82c 100644 --- a/arborist/attest/__init__.py +++ b/arborist/attest/__init__.py @@ -17,6 +17,12 @@ fingerprint distributions are deterministic functions of their sample inputs. No model is in the proof path. """ +from arborist.attest.chain import ( + commit_fingerprint, + entity_chain_hash, + entity_chain_iter, + verify_entity_chain, +) from arborist.attest.fingerprint import ( Fingerprint, canonical_bytes, @@ -24,4 +30,13 @@ from arborist.attest.fingerprint import ( leaf_hash, ) -__all__ = ["Fingerprint", "canonical_bytes", "from_samples", "leaf_hash"] +__all__ = [ + "Fingerprint", + "canonical_bytes", + "commit_fingerprint", + "entity_chain_hash", + "entity_chain_iter", + "from_samples", + "leaf_hash", + "verify_entity_chain", +] diff --git a/arborist/attest/chain.py b/arborist/attest/chain.py new file mode 100644 index 0000000..480b693 --- /dev/null +++ b/arborist/attest/chain.py @@ -0,0 +1,167 @@ +"""Per-entity fingerprint chain on top of `audit_events`. + +A FingerprintChain piggybacks on the existing append-only audit chain +via `arborist.store.append_audit`. Each fingerprint commit becomes +one `audit_events` row with `event_type='attest_fingerprint'` and +`subject_root=fp.entity_id`. + +Per-entity chain hash is derived at query time by folding +``sha256(prev_chain || event_hash)`` over the entity's filtered +event sequence, with the genesis ``prev_chain`` fixed at 32 zero +bytes (``arborist.merkle.ZERO_HASH``). Two layers of +tamper-evidence: + +1. Mutating any audit row breaks the global ``event_hash`` chain + (existing arborist invariant). +2. Mutating, reordering, or deleting a fingerprint breaks the + derived per-entity chain hash. + +No new SQL table at this phase. The per-entity head is computable +from existing columns; adding a head-pointer table is a forward +optimization for entities with very long chains. +""" + +from __future__ import annotations + +import hashlib +import sqlite3 +from dataclasses import asdict +from typing import Iterator + +from arborist.attest.fingerprint import Fingerprint, SCHEMA_VERSION +from arborist.merkle import ZERO_HASH +from arborist.store import append_audit + +EVENT_TYPE = "attest_fingerprint" +GENESIS_PREV_CHAIN = ZERO_HASH.hex() + + +def _fold(prev_chain_hex: str, event_hash_hex: str) -> str: + """One step of the per-entity chain hash.""" + h = hashlib.sha256() + h.update(bytes.fromhex(prev_chain_hex)) + h.update(bytes.fromhex(event_hash_hex)) + return h.hexdigest() + + +def commit_fingerprint( + conn: sqlite3.Connection, + fp: Fingerprint, + *, + ts: int | None = None, +) -> dict: + """Append a Fingerprint to the entity's chain. + + Returns ``{"event_hash", "entity_chain_hash", "seq"}`` where + ``seq`` is the 1-indexed per-entity sequence number after this + insert. + + Raises ``ValueError`` if the fingerprint's ``schema_version`` + does not match this module's pinned ``SCHEMA_VERSION``. This + guards against silently committing fingerprints encoded under + a future schema while the chain math still uses v1 + assumptions. + """ + if fp.schema_version != SCHEMA_VERSION: + raise ValueError( + f"fingerprint schema_version {fp.schema_version!r} != " + f"chain SCHEMA_VERSION {SCHEMA_VERSION!r}" + ) + + body = asdict(fp) + body["error_taxonomy"] = dict(body["error_taxonomy"]) + body["stress_params"] = dict(body["stress_params"]) + + event_hash = append_audit( + conn, + event_type=EVENT_TYPE, + body=body, + subject_root=fp.entity_id, + ts=ts, + ) + + new_chain_hash = entity_chain_hash(conn, fp.entity_id) + assert new_chain_hash is not None, "chain just got an event; must not be None" + seq = _entity_count(conn, fp.entity_id) + return { + "event_hash": event_hash, + "entity_chain_hash": new_chain_hash, + "seq": seq, + } + + +def _entity_count(conn: sqlite3.Connection, entity_id: str) -> int: + row = conn.execute( + "SELECT COUNT(*) FROM audit_events " + "WHERE event_type=? AND subject_root=?", + (EVENT_TYPE, entity_id), + ).fetchone() + return int(row[0]) + + +def _iter_entity_events( + conn: sqlite3.Connection, entity_id: str +) -> Iterator[tuple[int, str, str]]: + """Yield ``(seq, event_hash, body_json)`` for the entity in chain order.""" + cur = conn.execute( + "SELECT seq, event_hash, body FROM audit_events " + "WHERE event_type=? AND subject_root=? ORDER BY seq", + (EVENT_TYPE, entity_id), + ) + for row in cur: + yield int(row[0]), row[1], row[2] + + +def entity_chain_hash(conn: sqlite3.Connection, entity_id: str) -> str | None: + """Per-entity chain hash, derived from the entity's audit-event sequence. + + Returns None if the entity has no fingerprints committed yet. + """ + chain = GENESIS_PREV_CHAIN + any_event = False + for _seq, event_hash, _body in _iter_entity_events(conn, entity_id): + chain = _fold(chain, event_hash) + any_event = True + return chain if any_event else None + + +def entity_chain_iter( + conn: sqlite3.Connection, entity_id: str +) -> Iterator[tuple[int, str, Fingerprint]]: + """Yield ``(entity_seq, event_hash, Fingerprint)`` in chain order. + + ``entity_seq`` is 1-indexed per entity (NOT the global audit + ``seq``). Reconstructs the Fingerprint from canonical body JSON. + """ + import json + + n = 0 + for _global_seq, event_hash, body_json in _iter_entity_events(conn, entity_id): + n += 1 + body = json.loads(body_json) + fp = Fingerprint( + entity_id=body["entity_id"], + session_id=body["session_id"], + timestamp_utc=body["timestamp_utc"], + domain=body["domain"], + host_hash=body["host_hash"], + count=int(body["count"]), + mean_us=int(body["mean_us"]), + std_us=int(body["std_us"]), + p50_us=int(body["p50_us"]), + p90_us=int(body["p90_us"]), + p99_us=int(body["p99_us"]), + skew_q10000=int(body["skew_q10000"]), + error_taxonomy={k: int(v) for k, v in body.get("error_taxonomy", {}).items()}, + stress_params={k: int(v) for k, v in body.get("stress_params", {}).items()}, + schema_version=body["schema_version"], + ) + yield n, event_hash, fp + + +def verify_entity_chain( + conn: sqlite3.Connection, entity_id: str, claimed_chain_hash: str +) -> bool: + """Return True iff the derived chain hash matches the claim.""" + actual = entity_chain_hash(conn, entity_id) + return actual is not None and actual == claimed_chain_hash diff --git a/tests/test_attest_chain.py b/tests/test_attest_chain.py new file mode 100644 index 0000000..b3e7532 --- /dev/null +++ b/tests/test_attest_chain.py @@ -0,0 +1,179 @@ +"""Tests for `arborist.attest.chain`.""" + +from __future__ import annotations + +import hashlib + +import pytest + +from arborist.attest import ( + Fingerprint, + commit_fingerprint, + entity_chain_hash, + entity_chain_iter, + verify_entity_chain, +) +from arborist.attest.chain import ( + EVENT_TYPE, + GENESIS_PREV_CHAIN, + _fold, +) +from arborist.attest.fingerprint import SCHEMA_VERSION +from arborist.store import connect + + +def _fp(entity_id="alice", session_id="s-1", mean_us=1000): + return Fingerprint( + entity_id=entity_id, + session_id=session_id, + timestamp_utc="2026-06-05T15:30:00Z", + domain="python-debug", + host_hash="0" * 64, + count=3, + mean_us=mean_us, + std_us=100, + p50_us=mean_us, + p90_us=mean_us + 100, + p99_us=mean_us + 200, + skew_q10000=0, + error_taxonomy={}, + stress_params={}, + schema_version=SCHEMA_VERSION, + ) + + +def test_commit_returns_event_and_chain_hash(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + assert set(out) == {"event_hash", "entity_chain_hash", "seq"} + assert len(out["event_hash"]) == 64 + assert len(out["entity_chain_hash"]) == 64 + assert out["seq"] == 1 + + +def test_first_chain_hash_folds_genesis_and_event(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + expected = _fold(GENESIS_PREV_CHAIN, out["event_hash"]) + assert out["entity_chain_hash"] == expected + + +def test_second_chain_hash_folds_prev_and_event(tmp_path): + conn = connect(tmp_path / "t.db") + out1 = commit_fingerprint(conn, _fp(session_id="s-1")) + out2 = commit_fingerprint(conn, _fp(session_id="s-2", mean_us=1100)) + assert out2["seq"] == 2 + expected = _fold(out1["entity_chain_hash"], out2["event_hash"]) + assert out2["entity_chain_hash"] == expected + + +def test_entity_chain_hash_independent_across_entities(tmp_path): + conn = connect(tmp_path / "t.db") + a1 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-1")) + b1 = commit_fingerprint(conn, _fp(entity_id="bob", session_id="b-1")) + assert a1["entity_chain_hash"] != b1["entity_chain_hash"] + assert a1["seq"] == 1 and b1["seq"] == 1 + + +def test_entity_chain_hash_independent_of_global_audit_position(tmp_path): + """Per-entity chain only counts that entity's attest events. + + Even though alice and bob's events are interleaved in the global + audit_events table, alice's chain hash should be a function only + of alice's filtered event sequence. + """ + conn = connect(tmp_path / "t.db") + a1 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-1")) + _b1 = commit_fingerprint(conn, _fp(entity_id="bob", session_id="b-1")) + a2 = commit_fingerprint(conn, _fp(entity_id="alice", session_id="a-2", mean_us=1100)) + expected = _fold(a1["entity_chain_hash"], a2["event_hash"]) + assert a2["entity_chain_hash"] == expected + assert a2["seq"] == 2 # second alice event, not third global + + +def test_entity_chain_iter_yields_in_order(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(session_id="s-1", mean_us=1000)) + commit_fingerprint(conn, _fp(session_id="s-2", mean_us=1100)) + commit_fingerprint(conn, _fp(session_id="s-3", mean_us=1200)) + seen = list(entity_chain_iter(conn, "alice")) + assert [n for n, _h, _fp in seen] == [1, 2, 3] + assert [fp.session_id for _n, _h, fp in seen] == ["s-1", "s-2", "s-3"] + assert [fp.mean_us for _n, _h, fp in seen] == [1000, 1100, 1200] + + +def test_entity_chain_iter_yields_nothing_for_unknown_entity(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(entity_id="alice")) + assert list(entity_chain_iter(conn, "carol")) == [] + + +def test_entity_chain_hash_returns_none_for_unknown_entity(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(entity_id="alice")) + assert entity_chain_hash(conn, "carol") is None + + +def test_verify_entity_chain_accepts_correct(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + assert verify_entity_chain(conn, "alice", out["entity_chain_hash"]) + + +def test_verify_entity_chain_rejects_wrong(tmp_path): + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp()) + assert not verify_entity_chain(conn, "alice", "f" * 64) + + +def test_commit_rejects_unknown_schema_version(tmp_path): + conn = connect(tmp_path / "t.db") + base = _fp() + bad = Fingerprint(**{**base.__dict__, "schema_version": "fingerprint-v999"}) + with pytest.raises(ValueError): + commit_fingerprint(conn, bad) + + +def test_fingerprint_round_trips_through_chain(tmp_path): + conn = connect(tmp_path / "t.db") + original = _fp(session_id="s-1", mean_us=4242) + commit_fingerprint(conn, original) + recovered = next(entity_chain_iter(conn, "alice"))[2] + # Frozen dataclass equality covers all fields including the dict ones. + assert recovered == original + + +def test_event_type_is_namespaced(tmp_path): + """Other event types with same subject_root must not pollute the chain.""" + from arborist.store import append_audit + + conn = connect(tmp_path / "t.db") + commit_fingerprint(conn, _fp(entity_id="alice")) + # Inject an unrelated event with the same subject_root. + append_audit(conn, event_type="ingest", body={"x": 1}, subject_root="alice") + commit_fingerprint(conn, _fp(entity_id="alice", session_id="s-2", mean_us=1100)) + # Chain still counts only the two attest_fingerprint events. + seen = list(entity_chain_iter(conn, "alice")) + assert len(seen) == 2 + assert [fp.session_id for _n, _h, fp in seen] == ["s-1", "s-2"] + + +def test_event_hash_matches_global_audit_chain_invariant(tmp_path): + """sha256(prev_global || canonical_body) — the rule in store.py:1665-1675.""" + import json + from dataclasses import asdict + + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp()) + body = asdict(_fp()) + body["error_taxonomy"] = dict(body["error_taxonomy"]) + body["stress_params"] = dict(body["stress_params"]) + body_json = json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + expected = hashlib.sha256( + body_json.encode("utf-8", errors="surrogatepass") + ).hexdigest() + assert out["event_hash"] == expected + + +def test_event_type_constant_is_pinned(): + assert EVENT_TYPE == "attest_fingerprint" From 6f88ade766366cecee438809c2b045d6dec6e21a Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 5 Jun 2026 12:14:37 -0400 Subject: [PATCH 3/8] =?UTF-8?q?attest:=20pure-stats=20consistency=20?= =?UTF-8?q?=E2=80=94=20KS=20two-sample=20+=20Mahalanobis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ks_two_sample(a, b) → KSResult(D, p, n1, n2). Asymptotic Kolmogorov series for p. Caller picks the threshold; no soft classifier in the side door (training a threshold against the corpus = forbidden). mahalanobis(point, mean, inv_cov) → squared distance. estimate_mean / estimate_cov / invert_matrix are stdlib-only helpers. Singular cov fails closed (Gauss-Jordan refuses pivot < 1e-12) — caller must supply non-singular history, regularization is not added (a tuned lambda IS a soft classifier). 26 tests. 54 across attest/ now pass. --- arborist/attest/__init__.py | 14 ++ arborist/attest/consistency.py | 196 +++++++++++++++++++++++++++ tests/test_attest_consistency.py | 224 +++++++++++++++++++++++++++++++ 3 files changed, 434 insertions(+) create mode 100644 arborist/attest/consistency.py create mode 100644 tests/test_attest_consistency.py diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py index 0d8d82c..d2d4d0a 100644 --- a/arborist/attest/__init__.py +++ b/arborist/attest/__init__.py @@ -23,6 +23,14 @@ from arborist.attest.chain import ( 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, @@ -32,11 +40,17 @@ from arborist.attest.fingerprint import ( __all__ = [ "Fingerprint", + "KSResult", "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", ] diff --git a/arborist/attest/consistency.py b/arborist/attest/consistency.py new file mode 100644 index 0000000..60adcc3 --- /dev/null +++ b/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)) diff --git a/tests/test_attest_consistency.py b/tests/test_attest_consistency.py new file mode 100644 index 0000000..1672063 --- /dev/null +++ b/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 From d71035ac7181fbef0a39fd0d85890273c7f93b5f Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Fri, 5 Jun 2026 12:18:35 -0400 Subject: [PATCH 4/8] =?UTF-8?q?attest:=20CLI=20surface=20=E2=80=94=20commi?= =?UTF-8?q?t=20/=20verify=20/=20chain=20/=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit arborist attest commit — read latency samples (one int μs per line), build Fingerprint, append to entity chain. arborist attest verify — confirm a claimed per-entity chain hash. arborist attest chain — list the entity's chain entries + head. arborist attest check — Mahalanobis squared distance of new-session moments vs entity history. Caller picks the chi-squared threshold; no soft classifier. Handlers are thin argparse wrappers — all math lives in arborist.attest.{fingerprint,chain,consistency}. 6 e2e CLI tests + 54 unit = 60 attest. 181 cli-touching tests pass. --- arborist/cli.py | 201 +++++++++++++++++++++++++++++++++++++++ tests/test_attest_cli.py | 178 ++++++++++++++++++++++++++++++++++ 2 files changed, 379 insertions(+) create mode 100644 tests/test_attest_cli.py diff --git a/arborist/cli.py b/arborist/cli.py index 461d0dc..98e8192 100644 --- a/arborist/cli.py +++ b/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 diff --git a/tests/test_attest_cli.py b/tests/test_attest_cli.py new file mode 100644 index 0000000..cc1f63c --- /dev/null +++ b/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 From 30aba0e792a33fca5998168efb5effe7f5fdf9cb Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Thu, 11 Jun 2026 06:51:07 -0400 Subject: [PATCH 5/8] =?UTF-8?q?attest:=20schema=20v2=20=E2=80=94=20Fingerp?= =?UTF-8?q?rint.cites=20+=20blast=5Fradius=20over=20audit=20graph?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Fingerprint may declare an optional ``cites: tuple[str, ...]`` of other entities' chain hashes it vouches for. Citation is a claim of dependency, not a verification — arborist does not re-verify the cited chain at commit time. When a cited chain hash is later falsified (KS / Mahalanobis drift, manual quarantine), the audit graph yields a deterministic blast radius: blast_radius(conn, falsified_chain_hash) -> list[entity_id] Pure read over ``audit_events`` where ``event_type='attest_fingerprint'``; returns distinct, sorted entity_ids whose fingerprints cite the falsified hash. Routing (quarantine, retest, downgrade) belongs to a higher layer — blast_radius produces the graph, not the action. Schema bumps ``fingerprint-v1 → fingerprint-v2``. Legacy v1 bodies in storage (no ``cites`` key) reconstruct with ``cites=()`` via ``body.get("cites", ())`` in ``entity_chain_iter``; the schema-version mismatch guard in ``commit_fingerprint`` still rejects v1 commits going forward, as intended. --- arborist/attest/__init__.py | 2 + arborist/attest/chain.py | 34 ++++++++++++ arborist/attest/fingerprint.py | 23 ++++++++- tests/test_attest_chain.py | 94 +++++++++++++++++++++++++++++++++- 4 files changed, 150 insertions(+), 3 deletions(-) diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py index d2d4d0a..e7da74f 100644 --- a/arborist/attest/__init__.py +++ b/arborist/attest/__init__.py @@ -18,6 +18,7 @@ sample inputs. No model is in the proof path. """ from arborist.attest.chain import ( + blast_radius, commit_fingerprint, entity_chain_hash, entity_chain_iter, @@ -41,6 +42,7 @@ from arborist.attest.fingerprint import ( __all__ = [ "Fingerprint", "KSResult", + "blast_radius", "canonical_bytes", "commit_fingerprint", "entity_chain_hash", diff --git a/arborist/attest/chain.py b/arborist/attest/chain.py index 480b693..e653a43 100644 --- a/arborist/attest/chain.py +++ b/arborist/attest/chain.py @@ -154,6 +154,7 @@ def entity_chain_iter( skew_q10000=int(body["skew_q10000"]), error_taxonomy={k: int(v) for k, v in body.get("error_taxonomy", {}).items()}, stress_params={k: int(v) for k, v in body.get("stress_params", {}).items()}, + cites=tuple(body.get("cites", ())), schema_version=body["schema_version"], ) yield n, event_hash, fp @@ -165,3 +166,36 @@ def verify_entity_chain( """Return True iff the derived chain hash matches the claim.""" actual = entity_chain_hash(conn, entity_id) return actual is not None and actual == claimed_chain_hash + + +def blast_radius(conn: sqlite3.Connection, falsified_chain_hash: str) -> list[str]: + """Distinct entity_ids whose fingerprints cite the falsified chain hash. + + When an entity's chain hash is later falsified (KS / Mahalanobis + drift, manual quarantine, etc.), this returns the set of entities + who had declared a citation on that exact chain state via + ``Fingerprint.cites``. Their trust posture is now contingent on + the falsification and should be re-examined. + + Pure read; does NOT mutate state. Routing decisions (quarantine, + retest, downgrade) belong to a caller higher up the stack — the + blast graph is the input, not the action. + + Linear scan over ``audit_events`` of type ``attest_fingerprint``; + O(N) per call. Acceptable while N is small; if it ever isn't, a + derived ``fingerprint_cites(falsified_chain, citing_entity_id)`` + index is the next step. Don't build that until the pressure is + real. + """ + import json + + affected: set[str] = set() + cur = conn.execute( + "SELECT body FROM audit_events WHERE event_type=?", + (EVENT_TYPE,), + ) + for (body_json,) in cur: + body = json.loads(body_json) + if falsified_chain_hash in body.get("cites", ()): + affected.add(body["entity_id"]) + return sorted(affected) diff --git a/arborist/attest/fingerprint.py b/arborist/attest/fingerprint.py index 44f2eb9..65f0e87 100644 --- a/arborist/attest/fingerprint.py +++ b/arborist/attest/fingerprint.py @@ -1,6 +1,6 @@ """Fingerprint schema + canonical encoder + Merkle leaf hash. -Schema version: ``fingerprint-v1``. +Schema version: ``fingerprint-v2``. All numeric fields are integers. Latency lives in microseconds. Skew is quantized to a 4-decimal-place integer (``skew_q10000``). @@ -13,6 +13,16 @@ the quantization boundary at unlucky moments, two hosts could differ by ±1 in the last digit of ``skew_q10000`` or one of the std/mean fields. Acceptable for v1; a future revision can compute moments in integer arithmetic if required. + +**v2 delta: optional ``cites`` field.** A fingerprint may declare a +``cites`` tuple of other entities' chain hashes that it vouches for +or depends on. The citation is a *claim of dependency*, not a +verification — at commit time arborist does not re-verify the cited +chain. The point is that if a cited chain hash is later falsified +(KS / Mahalanobis drift, manual quarantine, etc.), the audit graph +yields a deterministic blast radius via ``chain.blast_radius()``. +Default empty tuple → equivalent to v1 semantics. Legacy v1 bodies +in storage (no ``cites`` key) are reconstructed with ``cites=()``. """ from __future__ import annotations @@ -25,7 +35,7 @@ from typing import Mapping, Sequence from arborist.merkle import hash_leaf -SCHEMA_VERSION = "fingerprint-v1" +SCHEMA_VERSION = "fingerprint-v2" @dataclass(frozen=True) @@ -58,6 +68,12 @@ class Fingerprint: - ``stress_params``: stress parameters in effect during the session (``concurrent_tasks``, ``time_limit_s``, ``interrupts``, etc.); caller-defined keys, integer values only. + - ``cites``: optional tuple of other entities' chain hashes this + fingerprint declares a dependency on. Order is significant + (caller decides) — citations are NOT sorted at canonicalization, + because the order itself carries meaning ("I observed A then B"). + Default empty tuple. See module docstring for blast-radius + semantics. - ``schema_version``: pinned to ``SCHEMA_VERSION``. """ @@ -75,6 +91,7 @@ class Fingerprint: skew_q10000: int error_taxonomy: Mapping[str, int] = field(default_factory=dict) stress_params: Mapping[str, int] = field(default_factory=dict) + cites: tuple[str, ...] = () schema_version: str = SCHEMA_VERSION @@ -135,6 +152,7 @@ def from_samples( latency_samples_us: Sequence[int], error_taxonomy: Mapping[str, int] | None = None, stress_params: Mapping[str, int] | None = None, + cites: Sequence[str] = (), ) -> Fingerprint: """Compute a Fingerprint from a list of integer microsecond latencies. @@ -165,6 +183,7 @@ def from_samples( skew_q10000=int(round(skew_f * 10000)), error_taxonomy=dict(error_taxonomy or {}), stress_params=dict(stress_params or {}), + cites=tuple(cites), ) diff --git a/tests/test_attest_chain.py b/tests/test_attest_chain.py index b3e7532..a226aac 100644 --- a/tests/test_attest_chain.py +++ b/tests/test_attest_chain.py @@ -8,6 +8,7 @@ import pytest from arborist.attest import ( Fingerprint, + blast_radius, commit_fingerprint, entity_chain_hash, entity_chain_iter, @@ -22,7 +23,7 @@ from arborist.attest.fingerprint import SCHEMA_VERSION from arborist.store import connect -def _fp(entity_id="alice", session_id="s-1", mean_us=1000): +def _fp(entity_id="alice", session_id="s-1", mean_us=1000, cites=()): return Fingerprint( entity_id=entity_id, session_id=session_id, @@ -38,6 +39,7 @@ def _fp(entity_id="alice", session_id="s-1", mean_us=1000): skew_q10000=0, error_taxonomy={}, stress_params={}, + cites=tuple(cites), schema_version=SCHEMA_VERSION, ) @@ -177,3 +179,93 @@ def test_event_hash_matches_global_audit_chain_invariant(tmp_path): def test_event_type_constant_is_pinned(): assert EVENT_TYPE == "attest_fingerprint" + + +def test_blast_radius_empty_when_no_events(tmp_path): + conn = connect(tmp_path / "t.db") + assert blast_radius(conn, "a" * 64) == [] + + +def test_blast_radius_empty_when_no_citations(tmp_path): + conn = connect(tmp_path / "t.db") + out = commit_fingerprint(conn, _fp(entity_id="alice")) + assert blast_radius(conn, out["entity_chain_hash"]) == [] + + +def test_blast_radius_finds_single_citer(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + assert blast_radius(conn, a["entity_chain_hash"]) == ["bob"] + + +def test_blast_radius_finds_multiple_citers_distinct_and_sorted(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="carol", session_id="c-1", cites=(a["entity_chain_hash"],)), + ) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + assert blast_radius(conn, a["entity_chain_hash"]) == ["bob", "carol"] + + +def test_blast_radius_deduplicates_repeated_citers(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-2", mean_us=1100, cites=(a["entity_chain_hash"],)), + ) + # Same entity citing the same chain hash from two fingerprints → one row. + assert blast_radius(conn, a["entity_chain_hash"]) == ["bob"] + + +def test_blast_radius_misses_unrelated_chain_hash(tmp_path): + conn = connect(tmp_path / "t.db") + a = commit_fingerprint(conn, _fp(entity_id="alice")) + commit_fingerprint( + conn, + _fp(entity_id="bob", session_id="b-1", cites=(a["entity_chain_hash"],)), + ) + assert blast_radius(conn, "f" * 64) == [] + + +def test_legacy_v1_body_without_cites_reconstructs_with_empty_tuple(tmp_path): + """v1 events in storage (no `cites` key) must round-trip through entity_chain_iter.""" + from arborist.store import append_audit + + conn = connect(tmp_path / "t.db") + legacy_body = { + "entity_id": "alice", + "session_id": "s-legacy", + "timestamp_utc": "2026-06-05T15:30:00Z", + "domain": "python-debug", + "host_hash": "0" * 64, + "count": 3, + "mean_us": 1000, + "std_us": 100, + "p50_us": 1000, + "p90_us": 1100, + "p99_us": 1200, + "skew_q10000": 0, + "error_taxonomy": {}, + "stress_params": {}, + "schema_version": "fingerprint-v1", + } + append_audit(conn, event_type=EVENT_TYPE, body=legacy_body, subject_root="alice") + seen = list(entity_chain_iter(conn, "alice")) + assert len(seen) == 1 + fp = seen[0][2] + assert fp.cites == () + assert fp.schema_version == "fingerprint-v1" From f550a46627c626a764f94c18d7460288f9b672d1 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Thu, 11 Jun 2026 06:51:13 -0400 Subject: [PATCH 6/8] store: document falsification_state transitions in module docstring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The four states (live / failed / stale / quarantined) and the allowed transitions between them were a CLAUDE.md convention with no written spec at the code surface. Document them inline at the module docstring: when live demotes to failed vs stale vs quarantined, when stale re-promotes to live, when failed and quarantined require an explicit governance event to rehabilitate. Docstring only — the CHECK constraint already pins the column domain; trajectory discipline is policy-level and lives in helpers that mutate state plus the chain-check-shards audit. --- arborist/store.py | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/arborist/store.py b/arborist/store.py index ea832e2..2edb2df 100644 --- a/arborist/store.py +++ b/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 From ab7ae4c0fb2c77bf158be0ea0cb40541db0a8555 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Thu, 11 Jun 2026 06:59:37 -0400 Subject: [PATCH 7/8] =?UTF-8?q?cold=5Fpack:=20filter=20=5Fcontent=5Fsize?= =?UTF-8?q?=20in=20single-shard=20restore=20=E2=80=94=20align=20with=20pul?= =?UTF-8?q?l=5Fmetadata=5Fpack=20NULL=20contract?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _restore_generic_table forwarded every JSONL column to the INSERT statement, including the producer's synthetic ``_content_size`` column (#53 — emitted on chunks to carry the size hint while the actual content BLOB is dropped from the metadata pack to keep it small). The target chunks table has no such column, so the restore hard-errored on ``no column named _content_size`` before any row landed; four cold_object tests had been failing on the branch (push_pack_then_pull_full / just_enough_hydrate / splits_to_fit_dvdr / v2_hydrates_fresh_empty_db). Drop ``_content_size`` from the column list. chunks rows land with ``content IS NULL`` per ``pull_metadata_pack``'s documented contract (``evict.py:825``); phase 2 (``pull_chunk_pack``) fills bytes via UPDATE WHERE leaf_hash=?. The size hint is unused in the single-shard path. The multi-shard ``_restore_routed_table`` still pre-allocates a zero-byte BLOB of that size for in-place phase 2 UPDATE (avoids page splits at corpus scale); that's a latent inconsistency with the doc contract on the just-enough routed path, but no test exercises it and the performance argument is real at scale. Not touching it here. Suite: 2995 passed (up from 2991), no regressions. --- arborist/cold_pack_metadata.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/arborist/cold_pack_metadata.py b/arborist/cold_pack_metadata.py index e8017f6..3954a7a 100644 --- a/arborist/cold_pack_metadata.py +++ b/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 = ( From 0c588e4abd170703acbe97cf626f7f178cf18f3e Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Thu, 11 Jun 2026 07:10:59 -0400 Subject: [PATCH 8/8] =?UTF-8?q?attest:=20drop=20v2=20schema=20bump=20?= =?UTF-8?q?=E2=80=94=20no=20v1=20records=20exist,=20cites=20is=20just=20a?= =?UTF-8?q?=20v1=20field?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/chain.py | 2 +- arborist/attest/fingerprint.py | 20 +++++++++----------- tests/test_attest_chain.py | 29 ----------------------------- 3 files changed, 10 insertions(+), 41 deletions(-) diff --git a/arborist/attest/chain.py b/arborist/attest/chain.py index e653a43..cd7736e 100644 --- a/arborist/attest/chain.py +++ b/arborist/attest/chain.py @@ -154,7 +154,7 @@ def entity_chain_iter( skew_q10000=int(body["skew_q10000"]), error_taxonomy={k: int(v) for k, v in body.get("error_taxonomy", {}).items()}, stress_params={k: int(v) for k, v in body.get("stress_params", {}).items()}, - cites=tuple(body.get("cites", ())), + cites=tuple(body["cites"]), schema_version=body["schema_version"], ) yield n, event_hash, fp diff --git a/arborist/attest/fingerprint.py b/arborist/attest/fingerprint.py index 65f0e87..e6df796 100644 --- a/arborist/attest/fingerprint.py +++ b/arborist/attest/fingerprint.py @@ -1,6 +1,6 @@ """Fingerprint schema + canonical encoder + Merkle leaf hash. -Schema version: ``fingerprint-v2``. +Schema version: ``fingerprint-v1``. All numeric fields are integers. Latency lives in microseconds. Skew is quantized to a 4-decimal-place integer (``skew_q10000``). @@ -14,15 +14,13 @@ by ±1 in the last digit of ``skew_q10000`` or one of the std/mean fields. Acceptable for v1; a future revision can compute moments in integer arithmetic if required. -**v2 delta: optional ``cites`` field.** A fingerprint may declare a -``cites`` tuple of other entities' chain hashes that it vouches for -or depends on. The citation is a *claim of dependency*, not a -verification — at commit time arborist does not re-verify the cited -chain. The point is that if a cited chain hash is later falsified -(KS / Mahalanobis drift, manual quarantine, etc.), the audit graph -yields a deterministic blast radius via ``chain.blast_radius()``. -Default empty tuple → equivalent to v1 semantics. Legacy v1 bodies -in storage (no ``cites`` key) are reconstructed with ``cites=()``. +**``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 @@ -35,7 +33,7 @@ from typing import Mapping, Sequence from arborist.merkle import hash_leaf -SCHEMA_VERSION = "fingerprint-v2" +SCHEMA_VERSION = "fingerprint-v1" @dataclass(frozen=True) diff --git a/tests/test_attest_chain.py b/tests/test_attest_chain.py index a226aac..ea1becc 100644 --- a/tests/test_attest_chain.py +++ b/tests/test_attest_chain.py @@ -240,32 +240,3 @@ def test_blast_radius_misses_unrelated_chain_hash(tmp_path): ) assert blast_radius(conn, "f" * 64) == [] - -def test_legacy_v1_body_without_cites_reconstructs_with_empty_tuple(tmp_path): - """v1 events in storage (no `cites` key) must round-trip through entity_chain_iter.""" - from arborist.store import append_audit - - conn = connect(tmp_path / "t.db") - legacy_body = { - "entity_id": "alice", - "session_id": "s-legacy", - "timestamp_utc": "2026-06-05T15:30:00Z", - "domain": "python-debug", - "host_hash": "0" * 64, - "count": 3, - "mean_us": 1000, - "std_us": 100, - "p50_us": 1000, - "p90_us": 1100, - "p99_us": 1200, - "skew_q10000": 0, - "error_taxonomy": {}, - "stress_params": {}, - "schema_version": "fingerprint-v1", - } - append_audit(conn, event_type=EVENT_TYPE, body=legacy_body, subject_root="alice") - seen = list(entity_chain_iter(conn, "alice")) - assert len(seen) == 1 - fp = seen[0][2] - assert fp.cites == () - assert fp.schema_version == "fingerprint-v1"