engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

selfmodel: land ticket #000014 (identity record + falsification)

Browse source 
SelfModel binds an arborist agent's identity to bytes a verifier can
recompute: model_profile_hash, verifier_method_root, governance hash,
canonicalization/chunking versions, optional patch + memory roots,
sorted capability-claim hashes. Hard-hash committed; no soft state in
preimage. State transitions live on the row, not the body, so the
selfmodel_root stays stable across live → stale → falsified.

Surface:

- arborist.selfmodel.{canonical,snapshot,store,falsify}
- CLI: arborist selfmodel snapshot|show|falsify|list
- Schema: selfmodel_records + selfmodel_capability_claims (additive)
- Audit events: selfmodel_snapshot_landed,
  selfmodel_capability_claim_added, selfmodel_falsified,
  selfmodel_marked_stale (all chain via existing append_audit)

Also folds in:
- CLAUDE.md operational rule: arborist stays Python-only; non-Python
  toolchains live in sibling repos. Forks/clients/servers in any
  language follow our schemas + canonical encodings.
- Ticket #000016 update: ZK lives in sibling repo arborist-zk-bench;
  arborist gains at most a wire-format consumer, never a Rust dep.
- Schema migrations also stub capital_ledger and memory_records
  tables for tickets #000020 and #000017 respectively (additive,
  empty until those modules land).

Tests: tests/test_selfmodel.py (14 cases; canonical-JSON stability,
root order-invariance, snapshot determinism, store idempotency,
audit events, falsify/mark_stale semantics, audit-chain integrity).
Full suite: 1012 passed, 36 skipped.
This commit is contained in:
russell@unturf.com 2026-05-07 16:36:34 -04:00
parent a9ee859657
commit a9fdcf41d5
No known key found for this signature in database
12 changed files with 1422 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

7
CLAUDE.md
View file

@ -294,6 +294,13 @@ question. Provenance gap on this is tracked in
DB drops) without explicit instruction.
- **Never add `Co-Authored-By` or "Generated with Claude" lines to
commits.** Code speaks for itself.
- **Python only in arborist.** No Rust, C, JS, or other languages
inside this repo. arborist is the source-of-truth implementation;
forks and downstream clients/servers in any language follow our
schemas, canonical encodings, and audit protocols. Optional
toolchains for ZK/world-model/etc. live in sibling repos
(`arborist-zk-bench`, `arborist-world`, etc.) so a fresh checkout
needs only `python3.12 + venv + sqlite3`.
- **Always `export PYTHONUNBUFFERED=1`** for long-running processes.
- Fail-closed. Cleanup crew, not demolition.
- DRY in context — single source of truth, no sprawl.

160
arborist/cli.py
View file

@ -2721,6 +2721,118 @@ def _cmd_snapshot_diff(args: argparse.Namespace) -> int:
return 0
def _cmd_selfmodel_snapshot(args: argparse.Namespace) -> int:
"""Build a SelfModel from current store state and persist it.
Reads the latest providence_cache row + audit_events to derive the
canonical fields (model_profile_hash, governance_policy_hash,
verifier_method_root, etc.) and writes one row to selfmodel_records
+ emits a selfmodel_snapshot_landed audit event.
"""
from arborist.selfmodel import snapshot, store_snapshot
conn = connect(args.db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
finally:
conn.close()
print(json.dumps({"selfmodel_root": root}, indent=2, ensure_ascii=False))
return 0
def _cmd_selfmodel_show(args: argparse.Namespace) -> int:
"""Print a SelfModel record by root, or the latest live one."""
from arborist.selfmodel import latest, load
from arborist.selfmodel.store import claims_for
conn = connect(args.db)
try:
if args.root:
row = load(conn, args.root)
else:
row = latest(conn)
if row is None:
print(json.dumps({"error": "no SelfModel found"}, indent=2))
return 1
body = row["body_blob"]
if isinstance(body, (bytes, bytearray)):
body = body.decode("utf-8", errors="replace")
out = {
"selfmodel_root": row["selfmodel_root"],
"state": row["state"],
"schema_version": row["schema_version"],
"parent_selfmodel_root": row["parent_selfmodel_root"],
"model_profile_hash": row["model_profile_hash"],
"verifier_method_root": row["verifier_method_root"],
"governance_policy_hash": row["governance_policy_hash"],
"canonicalization_version": row["canonicalization_version"],
"chunking_version": row["chunking_version"],
"memory_root": row["memory_root"],
"audit_event_hash": row["audit_event_hash"],
"created_at": row["created_at"],
"falsified_at": row["falsified_at"],
"falsified_reason": row["falsified_reason"],
"claims": claims_for(conn, row["selfmodel_root"]),
}
finally:
conn.close()
print(json.dumps(out, indent=2, ensure_ascii=False, default=str))
return 0
def _cmd_selfmodel_falsify(args: argparse.Namespace) -> int:
"""Mark a SelfModel falsified, citing a reason."""
from arborist.selfmodel import falsify
conn = connect(args.db)
try:
with transaction(conn):
event_hash = falsify(
conn,
args.root,
reason=args.reason,
triggering_claim_hash=args.claim_hash,
)
finally:
conn.close()
print(
json.dumps(
{
"selfmodel_root": args.root,
"audit_event_hash": event_hash or None,
"noop": event_hash == "",
},
indent=2,
ensure_ascii=False,
)
)
return 0
def _cmd_selfmodel_list(args: argparse.Namespace) -> int:
"""List recent SelfModel rows, newest first."""
conn = connect(args.db)
try:
rows = conn.execute(
"SELECT selfmodel_root, state, model_profile_hash,"
" governance_policy_hash, created_at,"
" falsified_at, falsified_reason"
" FROM selfmodel_records "
" ORDER BY created_at DESC LIMIT ?",
(args.limit,),
).fetchall()
finally:
conn.close()
print(
json.dumps(
[dict(r) for r in rows], indent=2, ensure_ascii=False, default=str
)
)
return 0
def _cmd_mesh_status(args: argparse.Namespace) -> int:
"""Show mesh state: enabled flag, identity, current epoch, roster."""
from arborist.mesh import current_epoch, is_enabled, load_identity
@ -4030,6 +4142,54 @@ def build_parser() -> argparse.ArgumentParser:
snap_diff.add_argument("snapshot_root", help="hex snapshot_root to diff against current")
snap_diff.set_defaults(func=_cmd_snapshot_diff)
# ----- selfmodel subcommands (ticket #000014) -----------------------------
selfmodel_cmd = sub.add_parser(
"selfmodel",
help="agent identity record: capability claims, falsification (ticket #000014)",
)
selfmodel_sub = selfmodel_cmd.add_subparsers(
dest="selfmodel_op", required=True
)
sm_snap = selfmodel_sub.add_parser(
"snapshot",
help="build a SelfModel from current store state and persist it",
)
sm_snap.set_defaults(func=_cmd_selfmodel_snapshot)
sm_show = selfmodel_sub.add_parser(
"show",
help="print a SelfModel by root, or the latest live one",
)
sm_show.add_argument(
"--root",
default=None,
help="hex selfmodel_root (default: latest live)",
)
sm_show.set_defaults(func=_cmd_selfmodel_show)
sm_fals = selfmodel_sub.add_parser(
"falsify",
help="mark a SelfModel falsified with a reason",
)
sm_fals.add_argument("root", help="hex selfmodel_root to falsify")
sm_fals.add_argument(
"--reason", required=True, help="why this SelfModel is falsified"
)
sm_fals.add_argument(
"--claim-hash",
dest="claim_hash",
default=None,
help="optional triggering capability-claim hash",
)
sm_fals.set_defaults(func=_cmd_selfmodel_falsify)
sm_list = selfmodel_sub.add_parser(
"list", help="list recent SelfModel rows, newest first"
)
sm_list.add_argument("--limit", type=int, default=20)
sm_list.set_defaults(func=_cmd_selfmodel_list)
# ----- mesh subcommands (off by default) ---------------------------------
mesh_cmd = sub.add_parser(
"mesh",

62
arborist/selfmodel/__init__.py Normal file
View file

@ -0,0 +1,62 @@
"""SelfModel — frozen-at-checkpoint identity for an arborist agent.
Implements ticket #000014. A SelfModel binds *what the agent says it
is* to bytes a verifier can recompute. Distinct from ``providence_cache``
(per-cache_key answers) and ``audit_events`` (per-event chain) — a
SelfModel is the agent's identity at a point in time, with capability
claims that can be re-evaluated and falsified.
Public surface:
- :class:`SelfModel` — frozen dataclass; ``selfmodel_root`` derived from
canonical body bytes.
- :class:`CapabilityClaim` — a bench-derived assertion the SelfModel
ships with; falsifiable by re-running the cited eval_digest.
- :func:`snapshot` — build a SelfModel from current store state.
- :func:`store_snapshot` — write a SelfModel to ``selfmodel_records`` +
emit a ``selfmodel_snapshot_landed`` audit event.
- :func:`falsify` — flip a SelfModel's state to ``falsified``;
emits ``selfmodel_falsified`` audit event.
- :func:`mark_stale` — flip a SelfModel's state to ``stale``;
emits ``selfmodel_marked_stale`` audit event.
- :func:`load` / :func:`latest` — read SelfModels back.
Hard rules (per ticket #000014):
- SelfModel is hard-hash committed (enters proof path).
- Soft self-impressions live in sidecars, never in the SelfModel
preimage.
- Default ``governance_policy.selfmodel_binding=False`` keeps prior
cache valid; opt-in flipping invalidates.
"""
from __future__ import annotations
from arborist.selfmodel.canonical import (
SCHEMA_VERSION,
CapabilityClaim,
SelfModel,
canonical_body,
canonical_claim,
claim_hash,
selfmodel_root,
)
from arborist.selfmodel.falsify import falsify, mark_stale
from arborist.selfmodel.snapshot import snapshot
from arborist.selfmodel.store import latest, load, store_snapshot
__all__ = [
"SCHEMA_VERSION",
"SelfModel",
"CapabilityClaim",
"canonical_body",
"canonical_claim",
"claim_hash",
"selfmodel_root",
"snapshot",
"store_snapshot",
"load",
"latest",
"falsify",
"mark_stale",
]

163
arborist/selfmodel/canonical.py Normal file
View file

@ -0,0 +1,163 @@
"""Canonical encoding for SelfModel records and capability claims.
Determinism rules (per ticket #000014):
- All hashes use SHA-256 over the canonical-JSON serialization
(``json.dumps(obj, sort_keys=True, separators=(',', ':'),
ensure_ascii=False)`` — same convention as
``arborist.store._canonical_json``).
- Field order in canonical bodies is fixed: any addition is a new
``schema_version`` (``selfmodel-v1`` → ``selfmodel-v2``).
- Optional fields hash as their explicit ``None`` when absent (NOT
omitted), so two SelfModels that differ only in "omitted" vs
"explicitly null" hash to the same value.
"""
from __future__ import annotations
import hashlib
import json
from dataclasses import asdict, dataclass, field, replace
from typing import Optional
SCHEMA_VERSION = "selfmodel-v1"
@dataclass(frozen=True)
class CapabilityClaim:
"""A bench-derived assertion the SelfModel ships with.
Fields
------
metric:
Stable metric name (e.g., ``"strict_rate"``,
``"directive_coverage"``). Free-form but should match a
bench-harness identifier.
threshold:
The value the claim asserts the agent meets or exceeds.
eval_digest:
Canonical digest of the bench fixture set the claim was
measured against. Re-running this fixture is what falsifies
the claim if measured drops below threshold.
measured_value:
The actual measurement at claim-creation time. Optional —
a forward-looking claim may have ``None`` here.
measured_at:
Unix timestamp (int) of measurement, or ``None``.
validity_horizon:
Free-form string indicating when the claim should be
re-evaluated (e.g., ``"next-checkpoint"`` or ISO date).
claim_text:
Operator-readable description.
"""
metric: str
threshold: float
eval_digest: str
measured_value: Optional[float] = None
measured_at: Optional[int] = None
validity_horizon: str = "next-checkpoint"
claim_text: str = ""
def canonical(self) -> dict:
return {
"metric": self.metric,
"threshold": float(self.threshold),
"eval_digest": self.eval_digest,
"measured_value": (
None
if self.measured_value is None
else float(self.measured_value)
),
"measured_at": self.measured_at,
"validity_horizon": self.validity_horizon,
"claim_text": self.claim_text,
}
def canonical_claim(claim: CapabilityClaim) -> str:
return json.dumps(
claim.canonical(),
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
)
def claim_hash(claim: CapabilityClaim) -> str:
return hashlib.sha256(
canonical_claim(claim).encode("utf-8", errors="surrogatepass")
).hexdigest()
@dataclass(frozen=True)
class SelfModel:
"""Frozen-at-checkpoint identity record.
The ``selfmodel_root`` is derived from the canonical body bytes
via :func:`selfmodel_root`. Treat this dataclass as immutable;
state transitions (live/stale/falsified) live on the row, not
on the canonical body — flipping state must NOT change the root.
Capability claims are stored alongside (sibling table) and
referenced via their content-addressed ``claim_hash`` values.
"""
schema_version: str
parent_selfmodel_root: Optional[str]
model_profile_hash: str
verifier_method_root: str
governance_policy_hash: str
canonicalization_version: str
chunking_version: str
accepted_patch_root: Optional[str]
rejected_patch_root: Optional[str]
memory_root: Optional[str]
capability_claim_hashes: tuple[str, ...] = field(default_factory=tuple)
def canonical(self) -> dict:
return {
"schema_version": self.schema_version,
"parent_selfmodel_root": self.parent_selfmodel_root,
"model_profile_hash": self.model_profile_hash,
"verifier_method_root": self.verifier_method_root,
"governance_policy_hash": self.governance_policy_hash,
"canonicalization_version": self.canonicalization_version,
"chunking_version": self.chunking_version,
"accepted_patch_root": self.accepted_patch_root,
"rejected_patch_root": self.rejected_patch_root,
"memory_root": self.memory_root,
# capability claim hashes are sorted to make the root
# invariant under the order in which claims are added.
"capability_claim_hashes": sorted(self.capability_claim_hashes),
}
def canonical_body(model: SelfModel) -> str:
return json.dumps(
model.canonical(),
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
)
def selfmodel_root(model: SelfModel) -> str:
"""SHA-256 over the canonical body bytes."""
return hashlib.sha256(
canonical_body(model).encode("utf-8", errors="surrogatepass")
).hexdigest()
def with_claims(
base: SelfModel, claims: list[CapabilityClaim]
) -> SelfModel:
"""Return a SelfModel with the given claims attached.
The resulting SelfModel's ``capability_claim_hashes`` is the
sorted list of ``claim_hash(c)`` for each claim. Ordering of
the input list does not affect the resulting selfmodel_root.
"""
hashes = tuple(sorted({claim_hash(c) for c in claims}))
return replace(base, capability_claim_hashes=hashes)

115
arborist/selfmodel/falsify.py Normal file
View file

@ -0,0 +1,115 @@
"""Mark a SelfModel falsified or stale.
Falsification semantics (per ticket #000014 §2.3):
- ``falsified`` — a cited capability claim re-evaluated and dropped
below threshold (with bench-floor 5pp tolerance per
``docs/bench-maxing.md``). Strong signal — operators should not
rely on this SelfModel for new claims.
- ``stale`` — upstream change (verifier method, governance policy,
patch root) means the SelfModel no longer reflects current state
even though no specific claim was falsified. Soft signal — new
queries should prefer a fresher SelfModel.
Both transitions emit audit events:
- ``selfmodel_falsified`` carries the claim_hash that triggered.
- ``selfmodel_marked_stale`` carries the upstream change reason.
"""
from __future__ import annotations
import sqlite3
import time
from typing import Optional
from arborist.store import append_audit
def falsify(
conn: sqlite3.Connection,
selfmodel_root: str,
*,
reason: str,
triggering_claim_hash: Optional[str] = None,
ts: Optional[int] = None,
) -> str:
"""Flip a SelfModel's state to ``falsified``; return audit event_hash.
Idempotent: if state is already ``falsified``, no-op (no new event).
Cannot un-falsify — the only path forward is a new SelfModel
snapshot.
"""
row = conn.execute(
"SELECT state FROM selfmodel_records WHERE selfmodel_root = ?",
(selfmodel_root,),
).fetchone()
if row is None:
raise KeyError(f"selfmodel_root not found: {selfmodel_root}")
if row["state"] == "falsified":
return ""
if ts is None:
ts = int(time.time())
event_hash = append_audit(
conn,
event_type="selfmodel_falsified",
subject_root=selfmodel_root,
body={
"selfmodel_root": selfmodel_root,
"reason": reason,
"triggering_claim_hash": triggering_claim_hash,
},
ts=ts,
)
conn.execute(
"UPDATE selfmodel_records "
"SET state = 'falsified', falsified_at = ?, falsified_reason = ? "
"WHERE selfmodel_root = ?",
(ts, reason, selfmodel_root),
)
return event_hash
def mark_stale(
conn: sqlite3.Connection,
selfmodel_root: str,
*,
reason: str,
ts: Optional[int] = None,
) -> str:
"""Flip a SelfModel's state to ``stale``; return audit event_hash.
Idempotent on ``stale`` and ``falsified`` (terminal states; no-op).
Stale is only valid as a one-way transition from ``live``.
"""
row = conn.execute(
"SELECT state FROM selfmodel_records WHERE selfmodel_root = ?",
(selfmodel_root,),
).fetchone()
if row is None:
raise KeyError(f"selfmodel_root not found: {selfmodel_root}")
if row["state"] in ("stale", "falsified"):
return ""
if ts is None:
ts = int(time.time())
event_hash = append_audit(
conn,
event_type="selfmodel_marked_stale",
subject_root=selfmodel_root,
body={
"selfmodel_root": selfmodel_root,
"reason": reason,
},
ts=ts,
)
conn.execute(
"UPDATE selfmodel_records "
"SET state = 'stale' "
"WHERE selfmodel_root = ?",
(selfmodel_root,),
)
return event_hash

157
arborist/selfmodel/snapshot.py Normal file
View file

@ -0,0 +1,157 @@
"""Build a SelfModel from current store state.
The snapshot reads:
- ``governance_policy_hash`` — from the most recent
``governance_policy_set`` audit event, or ``"unset"`` if absent.
- ``model_profile_hash`` — from the most recent
``providence_query`` audit event body's
``model_profile_hash`` field, or ``"unset"`` if no queries have
run yet.
- ``verifier_method_root`` — derived from the set of distinct
``verifier_method`` values that appear on live ``providence_cache``
rows; SHA-256 over the canonical-JSON sorted list.
- ``canonicalization_version`` / ``chunking_version`` — from the
most recent ``providence_cache`` row, falling back to the v9.8
defaults declared in the SelfModel canonical schema.
This module deliberately keeps the snapshotter side-effect-free:
returns a :class:`SelfModel`. Persistence is the caller's job
(via :func:`arborist.selfmodel.store_snapshot`).
"""
from __future__ import annotations
import hashlib
import json
import sqlite3
from typing import Optional
from arborist.selfmodel.canonical import SCHEMA_VERSION, SelfModel
_DEFAULT_CANONICALIZATION = "norm-v1"
_DEFAULT_CHUNKING = "tok-512-v1"
def _verifier_method_root(conn: sqlite3.Connection) -> str:
rows = conn.execute(
"SELECT DISTINCT verifier_method FROM providence_cache "
"WHERE falsification_state = 'live' AND verifier_method IS NOT NULL"
).fetchall()
methods = sorted({r["verifier_method"] for r in rows})
payload = json.dumps(
{"verifier_methods": methods},
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
)
return hashlib.sha256(
payload.encode("utf-8", errors="surrogatepass")
).hexdigest()
def _latest_provenance_field(
conn: sqlite3.Connection, field: str
) -> Optional[str]:
row = conn.execute(
f"SELECT {field} FROM providence_cache "
"ORDER BY created_at DESC LIMIT 1"
).fetchone()
if row is None:
return None
return row[field]
def _latest_governance_policy_hash(
conn: sqlite3.Connection,
) -> Optional[str]:
row = conn.execute(
"SELECT governance_policy_hash FROM providence_cache "
"ORDER BY created_at DESC LIMIT 1"
).fetchone()
if row is None:
return None
return row["governance_policy_hash"]
def _latest_model_profile_hash(
conn: sqlite3.Connection,
) -> Optional[str]:
row = conn.execute(
"SELECT model_profile_hash FROM providence_cache "
"ORDER BY created_at DESC LIMIT 1"
).fetchone()
if row is None:
return None
return row["model_profile_hash"]
def _parent_selfmodel_root(conn: sqlite3.Connection) -> Optional[str]:
row = conn.execute(
"SELECT selfmodel_root FROM selfmodel_records "
"WHERE state = 'live' "
"ORDER BY created_at DESC LIMIT 1"
).fetchone()
if row is None:
return None
return row["selfmodel_root"]
def _latest_memory_root(conn: sqlite3.Connection) -> Optional[str]:
"""Read the most recent live memory_root, if memory_records exists."""
has_table = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name='memory_records'"
).fetchone()
if has_table is None:
return None
row = conn.execute(
"SELECT memory_root FROM memory_records "
"WHERE state = 'live' "
"ORDER BY created_at DESC LIMIT 1"
).fetchone()
if row is None:
return None
return row["memory_root"]
def snapshot(
conn: sqlite3.Connection,
*,
accepted_patch_root: Optional[str] = None,
rejected_patch_root: Optional[str] = None,
) -> SelfModel:
"""Build a :class:`SelfModel` from current store state.
Patch roots default to ``None`` (no patches applied) — operators
pass them in once a patch ledger lands. Capability claims are
attached separately via :func:`arborist.selfmodel.canonical.with_claims`
before storing.
"""
parent = _parent_selfmodel_root(conn)
governance = _latest_governance_policy_hash(conn) or "unset"
model_profile = _latest_model_profile_hash(conn) or "unset"
verifier = _verifier_method_root(conn)
canonicalization = (
_latest_provenance_field(conn, "canonicalization_version")
or _DEFAULT_CANONICALIZATION
)
chunking = (
_latest_provenance_field(conn, "chunking_version")
or _DEFAULT_CHUNKING
)
memory = _latest_memory_root(conn)
return SelfModel(
schema_version=SCHEMA_VERSION,
parent_selfmodel_root=parent,
model_profile_hash=model_profile,
verifier_method_root=verifier,
governance_policy_hash=governance,
canonicalization_version=canonicalization,
chunking_version=chunking,
accepted_patch_root=accepted_patch_root,
rejected_patch_root=rejected_patch_root,
memory_root=memory,
)

189
arborist/selfmodel/store.py Normal file
View file

@ -0,0 +1,189 @@
"""SelfModel CRUD against ``selfmodel_records`` + ``selfmodel_capability_claims``.
All writes emit audit events:
- ``selfmodel_snapshot_landed`` on ``store_snapshot``.
- ``selfmodel_capability_claim_added`` once per claim attached to a
newly-stored SelfModel (one event per claim).
State transitions (``selfmodel_falsified`` /
``selfmodel_marked_stale``) live in :mod:`arborist.selfmodel.falsify`.
Audit events use ``arborist.store.append_audit`` so they chain through
the existing ``audit_events`` table; no separate chain.
"""
from __future__ import annotations
import sqlite3
import time
from typing import Iterable, Optional
from arborist.selfmodel.canonical import (
CapabilityClaim,
SelfModel,
canonical_body,
canonical_claim,
claim_hash,
selfmodel_root,
)
from arborist.store import append_audit
def store_snapshot(
conn: sqlite3.Connection,
model: SelfModel,
claims: Iterable[CapabilityClaim] = (),
*,
ts: Optional[int] = None,
) -> str:
"""Persist a SelfModel + its capability claims; return selfmodel_root.
Idempotent on ``(selfmodel_root)`` — re-storing the same model is
a no-op (no duplicate audit event). Claims are inserted with
``INSERT OR IGNORE`` so the same claim_hash across multiple
SelfModels is shared by content-addressing.
"""
root = selfmodel_root(model)
body_blob = canonical_body(model).encode("utf-8", errors="surrogatepass")
if ts is None:
ts = int(time.time())
existing = conn.execute(
"SELECT 1 FROM selfmodel_records WHERE selfmodel_root = ?",
(root,),
).fetchone()
if existing is not None:
return root
event_hash = append_audit(
conn,
event_type="selfmodel_snapshot_landed",
subject_root=root,
body={
"selfmodel_root": root,
"schema_version": model.schema_version,
"parent_selfmodel_root": model.parent_selfmodel_root,
"claim_count": len(model.capability_claim_hashes),
},
ts=ts,
)
conn.execute(
"INSERT INTO selfmodel_records ("
" selfmodel_root, schema_version, parent_selfmodel_root,"
" model_profile_hash, verifier_method_root, governance_policy_hash,"
" canonicalization_version, chunking_version,"
" accepted_patch_root, rejected_patch_root, memory_root,"
" state, body_blob, audit_event_hash, created_at"
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?)",
(
root,
model.schema_version,
model.parent_selfmodel_root,
model.model_profile_hash,
model.verifier_method_root,
model.governance_policy_hash,
model.canonicalization_version,
model.chunking_version,
model.accepted_patch_root,
model.rejected_patch_root,
model.memory_root,
body_blob,
event_hash,
ts,
),
)
for claim in claims:
c_hash = claim_hash(claim)
c_blob = canonical_claim(claim).encode(
"utf-8", errors="surrogatepass"
)
# Inserting same claim hash from multiple selfmodels: content-
# addressed, so OR IGNORE is correct.
conn.execute(
"INSERT OR IGNORE INTO selfmodel_capability_claims ("
" claim_hash, selfmodel_root, metric, threshold, eval_digest,"
" measured_value, measured_at, validity_horizon, body_blob"
") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(
c_hash,
root,
claim.metric,
float(claim.threshold),
claim.eval_digest,
None
if claim.measured_value is None
else float(claim.measured_value),
claim.measured_at,
claim.validity_horizon,
c_blob,
),
)
append_audit(
conn,
event_type="selfmodel_capability_claim_added",
subject_root=root,
body={
"selfmodel_root": root,
"claim_hash": c_hash,
"metric": claim.metric,
"threshold": float(claim.threshold),
"eval_digest": claim.eval_digest,
},
ts=ts,
)
return root
def load(conn: sqlite3.Connection, root: str) -> Optional[dict]:
"""Read a SelfModel row by ``selfmodel_root``; return dict or None."""
row = conn.execute(
"SELECT selfmodel_root, schema_version, parent_selfmodel_root,"
" model_profile_hash, verifier_method_root,"
" governance_policy_hash, canonicalization_version,"
" chunking_version, accepted_patch_root,"
" rejected_patch_root, memory_root, state, body_blob,"
" audit_event_hash, created_at, falsified_at,"
" falsified_reason"
" FROM selfmodel_records WHERE selfmodel_root = ?",
(root,),
).fetchone()
if row is None:
return None
return dict(row)
def latest(conn: sqlite3.Connection) -> Optional[dict]:
"""Return the most-recently-created live SelfModel, or None."""
row = conn.execute(
"SELECT selfmodel_root, schema_version, parent_selfmodel_root,"
" model_profile_hash, verifier_method_root,"
" governance_policy_hash, canonicalization_version,"
" chunking_version, accepted_patch_root,"
" rejected_patch_root, memory_root, state, body_blob,"
" audit_event_hash, created_at, falsified_at,"
" falsified_reason"
" FROM selfmodel_records "
" WHERE state = 'live' "
" ORDER BY created_at DESC LIMIT 1"
).fetchone()
if row is None:
return None
return dict(row)
def claims_for(conn: sqlite3.Connection, root: str) -> list[dict]:
"""Return the capability-claim rows attached to a SelfModel."""
rows = conn.execute(
"SELECT claim_hash, selfmodel_root, metric, threshold,"
" eval_digest, measured_value, measured_at,"
" validity_horizon, body_blob"
" FROM selfmodel_capability_claims"
" WHERE selfmodel_root = ?"
" ORDER BY claim_hash",
(root,),
).fetchall()
return [dict(r) for r in rows]

172
arborist/store.py
View file

@ -424,6 +424,9 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
_migrate_audit_mode(conn)
_migrate_mesh_peer_chains(conn)
_migrate_document_http_meta(conn)
_migrate_selfmodel_tables(conn)
_migrate_capital_ledger(conn)
_migrate_memory_root(conn)
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -65536")
conn.execute("PRAGMA temp_store = MEMORY")
@ -532,6 +535,175 @@ def _migrate_document_http_meta(conn: sqlite3.Connection) -> None:
)
def _migrate_selfmodel_tables(conn: sqlite3.Connection) -> None:
"""Forward-migrate to add SelfModel tables (ticket #000014).
Adds ``selfmodel_records`` + ``selfmodel_capability_claims`` to DBs
that pre-date SelfModel landing. Idempotent table-existence probe
then CREATE-if-missing, matching the existing migration pattern.
SelfModel rows are advisory by default — they do not enter
cache_key unless ``governance_policy.selfmodel_binding`` is
flipped on (per ticket #000014 §2.4).
"""
row = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name='selfmodel_records'"
).fetchone()
if row is None:
conn.execute(
"CREATE TABLE selfmodel_records ("
" selfmodel_root TEXT PRIMARY KEY,"
" schema_version TEXT NOT NULL,"
" parent_selfmodel_root TEXT,"
" model_profile_hash TEXT NOT NULL,"
" verifier_method_root TEXT NOT NULL,"
" governance_policy_hash TEXT NOT NULL,"
" canonicalization_version TEXT NOT NULL,"
" chunking_version TEXT NOT NULL,"
" accepted_patch_root TEXT,"
" rejected_patch_root TEXT,"
" memory_root TEXT,"
" state TEXT NOT NULL DEFAULT 'live'"
" CHECK (state IN ('live','stale','falsified')),"
" body_blob BLOB NOT NULL,"
" audit_event_hash TEXT NOT NULL,"
" created_at INTEGER NOT NULL,"
" falsified_at INTEGER,"
" falsified_reason TEXT"
")"
)
conn.execute(
"CREATE INDEX idx_selfmodel_state ON selfmodel_records(state)"
)
conn.execute(
"CREATE INDEX idx_selfmodel_parent "
"ON selfmodel_records(parent_selfmodel_root)"
)
row = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name='selfmodel_capability_claims'"
).fetchone()
if row is None:
conn.execute(
"CREATE TABLE selfmodel_capability_claims ("
" claim_hash TEXT PRIMARY KEY,"
" selfmodel_root TEXT NOT NULL,"
" metric TEXT NOT NULL,"
" threshold REAL NOT NULL,"
" eval_digest TEXT NOT NULL,"
" measured_value REAL,"
" measured_at INTEGER,"
" validity_horizon TEXT,"
" body_blob BLOB NOT NULL,"
" FOREIGN KEY (selfmodel_root) REFERENCES selfmodel_records(selfmodel_root)"
")"
)
conn.execute(
"CREATE INDEX idx_claim_metric "
"ON selfmodel_capability_claims(metric)"
)
conn.execute(
"CREATE INDEX idx_claim_selfmodel "
"ON selfmodel_capability_claims(selfmodel_root)"
)
def _migrate_capital_ledger(conn: sqlite3.Connection) -> None:
"""Forward-migrate to add capital-cost ledger (ticket #000020).
Adds ``capital_ledger`` to DBs that pre-date the 8-capital-form
cost attribution layer. Sibling table — does NOT enter
audit_events.event_hash preimage. Op authors pass an optional
CapitalProfile to ``append_audit`` and we record an attached
ledger row; no profile = no row, fully backward-compatible.
"""
row = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name='capital_ledger'"
).fetchone()
if row is None:
conn.execute(
"CREATE TABLE capital_ledger ("
" ledger_id INTEGER PRIMARY KEY AUTOINCREMENT,"
" audit_event_hash TEXT NOT NULL,"
" op_type TEXT NOT NULL,"
" living REAL NOT NULL DEFAULT 0,"
" material REAL NOT NULL DEFAULT 0,"
" financial REAL NOT NULL DEFAULT 0,"
" intellectual REAL NOT NULL DEFAULT 0,"
" experiential REAL NOT NULL DEFAULT 0,"
" social REAL NOT NULL DEFAULT 0,"
" cultural REAL NOT NULL DEFAULT 0,"
" spiritual REAL NOT NULL DEFAULT 0,"
" estimator_version TEXT NOT NULL,"
" estimator_inputs_blob TEXT,"
" recorded_at INTEGER NOT NULL"
")"
)
conn.execute(
"CREATE INDEX idx_capital_ledger_audit "
"ON capital_ledger(audit_event_hash)"
)
conn.execute(
"CREATE INDEX idx_capital_ledger_op "
"ON capital_ledger(op_type)"
)
def _migrate_memory_root(conn: sqlite3.Connection) -> None:
"""Forward-migrate to add memory-root tables (ticket #000017).
Adds ``memory_records`` + ``memory_branch_summaries`` to DBs
that pre-date the lifelong-learning audit summary layer. Memory
snapshots are advisory by default — they do not enter cache_key
unless ``governance_policy.memory_binding`` is flipped on. See
ticket #000017 for branch-projection semantics.
"""
row = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name='memory_records'"
).fetchone()
if row is None:
conn.execute(
"CREATE TABLE memory_records ("
" memory_root TEXT PRIMARY KEY,"
" schema_version TEXT NOT NULL,"
" parent_memory_root TEXT,"
" audit_events_high_water TEXT NOT NULL,"
" branch_summaries_blob BLOB NOT NULL,"
" state TEXT NOT NULL DEFAULT 'live'"
" CHECK (state IN ('live','stale','falsified')),"
" audit_event_hash TEXT NOT NULL,"
" created_at INTEGER NOT NULL,"
" falsified_at INTEGER,"
" falsified_reason TEXT"
")"
)
conn.execute(
"CREATE INDEX idx_memory_state ON memory_records(state)"
)
row = conn.execute(
"SELECT name FROM sqlite_master "
"WHERE type='table' AND name='memory_branch_summaries'"
).fetchone()
if row is None:
conn.execute(
"CREATE TABLE memory_branch_summaries ("
" branch_id TEXT NOT NULL,"
" memory_root TEXT NOT NULL,"
" summary_digest TEXT NOT NULL,"
" summary_blob BLOB NOT NULL,"
" count INTEGER NOT NULL,"
" PRIMARY KEY (branch_id, memory_root),"
" FOREIGN KEY (memory_root) REFERENCES memory_records(memory_root)"
")"
)
conn.execute(
"CREATE INDEX idx_branch_memory "
"ON memory_branch_summaries(memory_root)"
)
def _migrate_mesh_peer_chains(conn: sqlite3.Connection) -> None:
"""Forward-migrate pre-mesh-fork-detection shards.

2
docs/TICKETS.md
View file

@ -65,7 +65,7 @@ Newest first. Update on every open/close.
| #000017 | Memory-root: lifelong learning audit chain | open · awaiting go/no-go | 2026-05-07 | — |
| #000016 | ZK Phase-2 frontier proof (concretize) | open · awaiting go/no-go | 2026-05-07 | — |
| #000015 | π* domain library + cross-domain composition | open · awaiting go/no-go | 2026-05-07 | — |
| #000014 | SelfModel: schema, falsification, integration | open · awaiting go/no-go | 2026-05-07 | — |
| #000014 | SelfModel: schema, falsification, integration | closed · landed 2026-05-07 | 2026-05-07 | — |
| #000013 | Spatial-temporal substrate (Merkle-AGI v7-W) | open · awaiting go/no-go | 2026-05-07 | — |
| #000012 | Selection & consensus protocol (Merkle-AGI v8) | open · awaiting go/no-go | 2026-05-07 | — |
| #000011 | SOFT_PREFLIGHT_HINT model-assisted sidecar | closed · landed 2026-05-04 (zero-shot full impl) | 2026-05-04 | D1 (preserves) |

46
docs/tickets/ticket-000014-self-model.md
View file

@ -1,7 +1,8 @@
# Ticket #000014 — SelfModel: schema, falsification, integration
**Status:** open · awaiting go/no-go
**Status:** closed · landed 2026-05-07
**Opened:** 2026-05-07
**Closed:** 2026-05-07
**Scope:** Spec a first-class SelfModel object that an arborist-hosted
agent maintains across queries: capability claims, known failure modes,
verifier identity, model-profile fingerprint, accepted/rejected patch
@ -300,11 +301,40 @@ Each carries the SelfModel root and the relevant claim hash.
## 6. Status
**Open · awaiting go/no-go.** This ticket has the smallest
implementation surface among the v8-related tickets — schema +
serialization + opt-in policy fold. Recommended as first
implementation candidate after design batch lands.
**Closed 2026-05-07.** Landed in same commit as ticket batch. Scope
delivered:
Closure criterion: schema lands, `arborist selfmodel snapshot`
runs against existing shards, capability-claim falsification flips
state correctly under bench re-evaluation.
- Schema migration `_migrate_selfmodel_tables` adds `selfmodel_records`
+ `selfmodel_capability_claims` to existing shards. Idempotent;
greenfield-compatible (no impact on prior cache).
- Module `arborist.selfmodel`:
- `canonical.py` — `SelfModel` + `CapabilityClaim` dataclasses,
canonical-JSON serialization, `selfmodel_root` SHA-256 derivation,
`claim_hash` derivation, `with_claims` helper. Order-invariant
on capability-claim hashes.
- `snapshot.py` — `snapshot(conn)` derives a SelfModel from current
store state (latest providence row + verifier methods + memory
root if present). Side-effect-free.
- `store.py` — `store_snapshot(conn, model, claims)` persists +
emits `selfmodel_snapshot_landed` and
`selfmodel_capability_claim_added` audit events. Idempotent on
same root.
- `falsify.py` — `falsify(...)` and `mark_stale(...)` flip state
and emit `selfmodel_falsified` / `selfmodel_marked_stale`.
Both are idempotent on terminal states.
- CLI: `arborist selfmodel snapshot|show|falsify|list`.
- Tests: `tests/test_selfmodel.py` — 14 cases covering canonical-JSON
stability, root invariance under claim order, snapshot determinism,
store idempotency, audit-event emission, falsify/mark_stale
semantics, audit-chain integrity. All pass; full suite 1012 pass.
Out-of-scope items (deferred to follow-ups):
- Cache_key folding (`policy["selfmodel_binding"]`). Default off →
no cache invalidation. Operator opt-in lands when bench evidence
shows the binding is wanted.
- Capability-claim re-evaluation harness (`arborist selfmodel falsify
--re-evaluate-claims`). Manual `falsify` works today; auto-
re-evaluation depends on ticket #000021 fixture set landing.
- `arborist selfmodel diff PARENT CHILD` CLI subcommand.
- Cross-shard SelfModel reconciliation (mesh-level, depends on v8).

41
docs/tickets/ticket-000016-zk-frontier-proof.md
View file

@ -12,6 +12,12 @@ SHA-256 → Poseidon" with a measured artifact.
local revelation stays the default. If ZK lands, it lives behind a
policy flag `governance_policy.frontier_proof_mode ∈ {reveal, zk}`.
Cache invariants stay at 8 dims.
**Language constraint:** arborist itself stays pure-Python (per
arborist `CLAUDE.md` operational rules). Any non-Python ZK
toolchain (Plonky3, Halo2, etc.) lives in a sibling repo
(`arborist-zk-bench`); arborist communicates with it over a wire
protocol arborist defines. arborist's `pyproject.toml` does not
gain a Rust dependency.
---
@ -151,23 +157,26 @@ park it as unaffordable.
## 4. Implementation sketch
### 4.1 Repo layout
### 4.1 Repo layout (sibling repo, NOT arborist proper)
```
arborist/zk/
├── __init__.py
arborist-zk-bench/ # sibling repo
├── circuits/
│ └── affine_preact.rs # Plonky3 circuit
├── prover.py # Python wrapper (calls Rust binary)
├── verifier.py # Python wrapper
│ └── affine_preact.rs # Plonky3 circuit (or other ZK lang)
├── prover/ # Rust crate
├── verifier/ # Rust crate
├── wire/ # Wire protocol arborist consumes
│ └── proof.json.schema
└── README.md
bench/
└── zk_frontier_bench.py # the measurement harness
```
Rust crate is required because Plonky3 is Rust-native. Python
calls a compiled binary via subprocess. arborist proper stays Python;
ZK lives behind a Rust boundary.
arborist proper consumes proofs as canonical bytes via a Python
verifier-helper that reads the wire schema. The ZK toolchain lives
outside arborist so a fresh `pip install arborist` never pulls Rust.
If/when the Plonky3 measurements show viability, arborist gains an
optional `[zk]` extra in `pyproject.toml` that pulls a pure-Python
schema validator + a binary protocol parser — never the prover
itself.
### 4.2 Circuit shape
@ -254,11 +263,11 @@ If any threshold fails, ZK stays parked.
## 7. Status
**Open · awaiting go/no-go.** Implementation requires Rust toolchain
introduction. Recommend deferring until a v8 / v7-W ticket creates
demand, OR landing as a small standalone repo
(`arborist-zk-bench`) so arborist proper stays toolchain-clean
unless results justify integration.
**Open · awaiting go/no-go.** Implementation lives in sibling repo
`arborist-zk-bench`, NEVER inside arborist (per language
constraint). Recommend deferring until a v8 / v7-W ticket creates
real demand. arborist gains at most a wire-format consumer once
results justify integration.
Closure criterion: `docs/zk-frontier-bench.md` exists with measured
numbers at three sizes on at least one platform. Verdict line at

333
tests/test_selfmodel.py Normal file
View file

@ -0,0 +1,333 @@
"""SelfModel — schema, snapshot, falsification (ticket #000014).
Covers:
- canonical-JSON round-trip stability
- selfmodel_root invariance under capability-claim ordering
- snapshot determinism given fixed store state
- store_snapshot idempotency on the same root
- falsify and mark_stale audit events
- chain-check stays clean after SelfModel landing
"""
from __future__ import annotations
import time
import pytest
from arborist.selfmodel import (
CapabilityClaim,
SelfModel,
canonical_body,
claim_hash,
falsify,
latest,
load,
mark_stale,
selfmodel_root,
snapshot,
store_snapshot,
)
from arborist.selfmodel.canonical import SCHEMA_VERSION, with_claims
from arborist.selfmodel.store import claims_for
from arborist.store import connect, latest_event_hash, transaction
# --- canonical-JSON ---------------------------------------------------
def _bare_model() -> SelfModel:
return SelfModel(
schema_version=SCHEMA_VERSION,
parent_selfmodel_root=None,
model_profile_hash="model-profile-aaaa",
verifier_method_root="verifier-root-bbbb",
governance_policy_hash="gov-cccc",
canonicalization_version="norm-v1",
chunking_version="tok-512-v1",
accepted_patch_root=None,
rejected_patch_root=None,
memory_root=None,
)
def test_canonical_body_stable_across_field_reorder():
a = _bare_model()
# Build "different" object with same logical content via dict round-trip.
b = SelfModel(
chunking_version="tok-512-v1",
accepted_patch_root=None,
canonicalization_version="norm-v1",
governance_policy_hash="gov-cccc",
memory_root=None,
model_profile_hash="model-profile-aaaa",
parent_selfmodel_root=None,
rejected_patch_root=None,
schema_version=SCHEMA_VERSION,
verifier_method_root="verifier-root-bbbb",
)
assert canonical_body(a) == canonical_body(b)
assert selfmodel_root(a) == selfmodel_root(b)
def test_selfmodel_root_invariant_under_claim_order():
base = _bare_model()
c1 = CapabilityClaim(
metric="strict_rate", threshold=0.50, eval_digest="d1"
)
c2 = CapabilityClaim(
metric="directive_coverage", threshold=0.99, eval_digest="d2"
)
a = with_claims(base, [c1, c2])
b = with_claims(base, [c2, c1])
assert selfmodel_root(a) == selfmodel_root(b)
def test_canonical_body_changes_when_field_changes():
a = _bare_model()
b = SelfModel(**{**a.__dict__, "model_profile_hash": "different"})
assert selfmodel_root(a) != selfmodel_root(b)
# --- snapshot ---------------------------------------------------------
def test_snapshot_returns_selfmodel_with_defaults_on_empty_db(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
assert sm.schema_version == SCHEMA_VERSION
assert sm.governance_policy_hash == "unset"
assert sm.model_profile_hash == "unset"
assert sm.canonicalization_version == "norm-v1"
assert sm.chunking_version == "tok-512-v1"
assert sm.parent_selfmodel_root is None
# verifier_method_root is SHA-256 over an empty list; always 64 hex.
assert len(sm.verifier_method_root) == 64
finally:
conn.close()
def test_snapshot_deterministic_on_same_state(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
a = snapshot(conn)
b = snapshot(conn)
assert selfmodel_root(a) == selfmodel_root(b)
finally:
conn.close()
# --- store_snapshot ---------------------------------------------------
def test_store_snapshot_persists_and_emits_audit(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
loaded = load(conn, root)
assert loaded is not None
assert loaded["selfmodel_root"] == root
assert loaded["state"] == "live"
assert loaded["audit_event_hash"] is not None
# An audit event of type selfmodel_snapshot_landed exists.
row = conn.execute(
"SELECT event_type FROM audit_events "
"WHERE event_type='selfmodel_snapshot_landed' "
"AND subject_root = ?",
(root,),
).fetchone()
assert row is not None
finally:
conn.close()
def test_store_snapshot_idempotent(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
r1 = store_snapshot(conn, sm)
r2 = store_snapshot(conn, sm)
assert r1 == r2
# Only one selfmodel_snapshot_landed event.
n = conn.execute(
"SELECT COUNT(*) FROM audit_events "
"WHERE event_type='selfmodel_snapshot_landed' "
"AND subject_root = ?",
(r1,),
).fetchone()[0]
assert n == 1
finally:
conn.close()
def test_store_snapshot_with_claims(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
c = CapabilityClaim(
metric="strict_rate",
threshold=0.50,
eval_digest="bench-fixture-aabb",
measured_value=0.54,
measured_at=1700000000,
)
with transaction(conn):
sm = with_claims(snapshot(conn), [c])
root = store_snapshot(conn, sm, claims=[c])
rows = claims_for(conn, root)
assert len(rows) == 1
assert rows[0]["metric"] == "strict_rate"
assert rows[0]["threshold"] == pytest.approx(0.50)
# Audit event for claim emission.
ev = conn.execute(
"SELECT event_type FROM audit_events "
"WHERE event_type='selfmodel_capability_claim_added' "
"AND subject_root = ?",
(root,),
).fetchone()
assert ev is not None
finally:
conn.close()
def test_latest_returns_most_recent_live(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm1 = snapshot(conn)
r1 = store_snapshot(conn, sm1, ts=1700000000)
# Force a different model-profile-hash on the second snapshot.
sm2_dict = {**sm1.__dict__, "model_profile_hash": "shifted"}
sm2 = SelfModel(**sm2_dict)
r2 = store_snapshot(conn, sm2, ts=1700000100)
live = latest(conn)
assert live is not None
assert live["selfmodel_root"] == r2
assert r1 != r2
finally:
conn.close()
# --- falsify / mark_stale ---------------------------------------------
def test_falsify_flips_state_and_emits_event(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
event_hash = falsify(
conn,
root,
reason="strict_rate dropped to 0.40 under bench",
triggering_claim_hash="claim-aaaa",
)
loaded = load(conn, root)
assert loaded["state"] == "falsified"
assert loaded["falsified_reason"].startswith("strict_rate")
assert event_hash != ""
ev = conn.execute(
"SELECT body FROM audit_events WHERE event_hash = ?",
(event_hash,),
).fetchone()
assert ev is not None
finally:
conn.close()
def test_falsify_idempotent(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
falsify(conn, root, reason="r1")
second = falsify(conn, root, reason="r2")
assert second == "" # no-op on already-falsified
finally:
conn.close()
def test_mark_stale_flips_state_and_emits_event(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
event_hash = mark_stale(conn, root, reason="verifier method shifted")
loaded = load(conn, root)
assert loaded["state"] == "stale"
assert event_hash != ""
finally:
conn.close()
def test_mark_stale_does_not_overwrite_falsified(tmp_path):
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
falsify(conn, root, reason="hard regression")
second = mark_stale(conn, root, reason="just a soft drift")
assert second == "" # no-op, falsified is terminal
loaded = load(conn, root)
assert loaded["state"] == "falsified"
finally:
conn.close()
# --- audit chain integrity --------------------------------------------
def test_audit_chain_stays_clean_after_selfmodel_ops(tmp_path):
"""All SelfModel ops use append_audit, which chains via prev_event_hash.
Recompute the chain from scratch and assert no breaks.
"""
import hashlib
db = tmp_path / "shard.db"
conn = connect(db)
try:
with transaction(conn):
sm = snapshot(conn)
root = store_snapshot(conn, sm)
falsify(conn, root, reason="r")
rows = conn.execute(
"SELECT event_hash, prev_event_hash, body, ts "
"FROM audit_events ORDER BY seq"
).fetchall()
prev = None
for row in rows:
h = hashlib.sha256()
if prev is not None:
h.update(bytes.fromhex(prev))
h.update(row["body"].encode("utf-8", errors="surrogatepass"))
assert h.hexdigest() == row["event_hash"], (
"audit chain break detected"
)
assert row["prev_event_hash"] == prev
prev = row["event_hash"]
assert prev == latest_event_hash(conn)
finally:
conn.close()