diff --git a/arborist/attest/__init__.py b/arborist/attest/__init__.py deleted file mode 100644 index e7da74f..0000000 --- a/arborist/attest/__init__.py +++ /dev/null @@ -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", -] diff --git a/arborist/attest/chain.py b/arborist/attest/chain.py deleted file mode 100644 index cd7736e..0000000 --- a/arborist/attest/chain.py +++ /dev/null @@ -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) diff --git a/arborist/attest/consistency.py b/arborist/attest/consistency.py deleted file mode 100644 index 60adcc3..0000000 --- a/arborist/attest/consistency.py +++ /dev/null @@ -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)) diff --git a/arborist/attest/fingerprint.py b/arborist/attest/fingerprint.py deleted file mode 100644 index e6df796..0000000 --- a/arborist/attest/fingerprint.py +++ /dev/null @@ -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() diff --git a/arborist/cli.py b/arborist/cli.py index 98e8192..461d0dc 100644 --- a/arborist/cli.py +++ b/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 diff --git a/arborist/cold_pack_metadata.py b/arborist/cold_pack_metadata.py index 3954a7a..e8017f6 100644 --- a/arborist/cold_pack_metadata.py +++ b/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 = ( diff --git a/arborist/store.py b/arborist/store.py index 2edb2df..ea832e2 100644 --- a/arborist/store.py +++ b/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 diff --git a/tests/test_attest_chain.py b/tests/test_attest_chain.py deleted file mode 100644 index ea1becc..0000000 --- a/tests/test_attest_chain.py +++ /dev/null @@ -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) == [] - diff --git a/tests/test_attest_cli.py b/tests/test_attest_cli.py deleted file mode 100644 index cc1f63c..0000000 --- a/tests/test_attest_cli.py +++ /dev/null @@ -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 diff --git a/tests/test_attest_consistency.py b/tests/test_attest_consistency.py deleted file mode 100644 index 1672063..0000000 --- a/tests/test_attest_consistency.py +++ /dev/null @@ -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 diff --git a/tests/test_attest_fingerprint.py b/tests/test_attest_fingerprint.py deleted file mode 100644 index 6058035..0000000 --- a/tests/test_attest_fingerprint.py +++ /dev/null @@ -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