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

@@ -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"]

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()

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