russell@unturf.com
a9fdcf41d5
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.
333 lines
10 KiB
Python
333 lines
10 KiB
Python
"""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()
|