arborist/bench/scripts/selfmodel_chain_snapshot.py

"""Append one SelfModel snapshot to a persistent chain shard, with
capability claims drawn from a fresh 5S/5T/5F bench run (#000025 §10.11).

Why this exists: Phase 1a's Finetuning sub-battery scored *synthetic*
parent→child SelfModel pairs; Phase 1b.2's "live" path round-trips a
parent + child through a *fresh temp* shard each run. §10.11 wants the
real thing — a SelfModel lineage that *persists across runs*, so the
Finetuning runner can measure improvement between two genuine,
chained snapshots instead of a pair fabricated for the occasion.

Each invocation:

  1. Runs the canonical 5S/5T/5F sub-batteries (embedded packs — the
     ones `fork_score._BATTERY_RATE_KEYS` reads). Deterministic, ~1s,
     no LLM call.
  2. Builds one :class:`arborist.selfmodel.CapabilityClaim` per
     sub-battery: ``metric = "5S-syntax"`` etc., ``measured_value`` =
     that pack's rate, ``eval_digest`` = that pack's fixture digest,
     ``threshold`` = ``fork_score.SIGNAL_FLOOR`` (the rate floor below
     which a fork-score Δ on this capability is a hard regression).
  3. Calls :func:`arborist.selfmodel.snapshot` (which auto-parents to
     the latest root in the shard) → :func:`with_claims` →
     :func:`store_snapshot`. The new snapshot's distinct parent makes
     it a distinct root, so the chain grows by exactly one per run.

Output shard defaults to ``~/.arborist/shards/selfmodel-chain.db``
(override with ``--shard`` or ``ARBORIST_SELFMODEL_CHAIN_DB``). The
shard carries an ``audit_events`` table (every ``store_snapshot``
write chains through it), so ``make chain-check-shards`` covers it.

Idempotent on content: re-running with no chain growth in between is
a no-op only if the parent root is unchanged — which it isn't after
the first append, so successive runs always extend the lineage. The
bench rates are at ceiling (1.0) today, so the chained Δ is 0.0; the
point is that the *mechanism* is real and the lineage outlives any
single process.
"""

from __future__ import annotations

import argparse
import os
import sys
import time
from dataclasses import asdict
from pathlib import Path

_REPO_ROOT = Path(__file__).resolve().parents[2]
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from bench.batteries import b_5f, b_5s, b_5t  # noqa: E402
from arborist.selfmodel import CapabilityClaim, snapshot, store_snapshot  # noqa: E402
from arborist.selfmodel.canonical import with_claims  # noqa: E402
from arborist.selfmodel.store import claims_for, latest  # noqa: E402
from arborist.store import connect, transaction  # noqa: E402
from arborist.substrate.fork_score import SIGNAL_FLOOR, _BATTERY_RATE_KEYS  # noqa: E402

_FIX = _REPO_ROOT / "bench" / "fixtures"

# Canonical sub-battery → embedded fixture pack (the keys fork_score
# reads). Mirrors bench.batteries.runner._DEFAULT_FIXTURES minus the
# legacy 5t "transfer" the scorer ignores.
_PACKS: list[tuple[str, str, Path]] = [
    ("5s", "syntax", _FIX / "5s" / "syntax-v1.jsonl"),
    ("5s", "semantics", _FIX / "5s" / "semantics-v1.jsonl"),
    ("5s", "syllogism", _FIX / "5s" / "syllogism-v1.jsonl"),
    ("5s", "synthesis", _FIX / "5s" / "synthesis-v1.jsonl"),
    ("5s", "semiotics", _FIX / "5s" / "semiotics-v1.jsonl"),
    ("5t", "transfer-learning", _FIX / "5t" / "transfer-learning-v2.jsonl"),
    ("5t", "triangulation", _FIX / "5t" / "triangulation-v1.jsonl"),
    ("5t", "truthtables", _FIX / "5t" / "truthtables-v1.jsonl"),
    ("5t", "transitivity", _FIX / "5t" / "transitivity-v1.jsonl"),
    ("5t", "time", _FIX / "5t" / "time-v1.jsonl"),
    ("5f", "function", _FIX / "5f" / "function-v1.jsonl"),
    ("5f", "finetuning", _FIX / "5f" / "finetuning-v1.jsonl"),
    ("5f", "falsification", _FIX / "5f" / "falsification-v1.jsonl"),
    ("5f", "formulate", _FIX / "5f" / "formulate-v1.jsonl"),
    ("5f", "feedback-loop", _FIX / "5f" / "feedback-loop-v1.jsonl"),
]

_SUB_RUNNERS = {"5s": b_5s.SUB_BATTERIES, "5t": b_5t.SUB_BATTERIES, "5f": b_5f.SUB_BATTERIES}


def default_shard() -> Path:
    env = os.environ.get("ARBORIST_SELFMODEL_CHAIN_DB")
    if env:
        return Path(env).expanduser()
    return Path.home() / ".arborist" / "shards" / "selfmodel-chain.db"


def _battery_claims(measured_at: int) -> list[CapabilityClaim]:
    claims: list[CapabilityClaim] = []
    for battery, sub, path in _PACKS:
        res = _SUB_RUNNERS[battery][sub](path)
        metric_key = _BATTERY_RATE_KEYS[battery][sub]
        rate = float(res.metrics.get(metric_key, 0.0))
        claims.append(
            CapabilityClaim(
                metric=f"{battery.upper()}-{sub}",
                threshold=float(SIGNAL_FLOOR),
                eval_digest=res.fixture_digest,
                measured_value=rate,
                measured_at=measured_at,
                validity_horizon="next-checkpoint",
                claim_text=f"{battery.upper()} {sub} {metric_key} on the embedded pack",
            )
        )
    return claims


def append_snapshot(shard: Path, *, ts: int | None = None) -> dict:
    """Run the suite, append one chained SelfModel snapshot, return a
    summary dict ``{root, parent_root, depth, claims: [...]}``."""
    if ts is None:
        ts = int(time.time())
    shard.parent.mkdir(parents=True, exist_ok=True)
    conn = connect(shard)
    try:
        prev = latest(conn)
        claims = _battery_claims(ts)
        with transaction(conn):
            sm = with_claims(snapshot(conn), claims)
            root = store_snapshot(conn, sm, claims=claims, ts=ts)
        # Walk the parent chain to report depth.
        depth = 1
        cur = load_parent(conn, root)
        while cur is not None:
            depth += 1
            cur = load_parent(conn, cur)
        rows = claims_for(conn, root)
        return {
            "shard": str(shard),
            "root": root,
            "parent_root": prev["selfmodel_root"] if prev else None,
            "depth": depth,
            "claims": {r["metric"]: r["measured_value"] for r in rows},
        }
    finally:
        conn.close()


def load_parent(conn, root: str) -> str | None:
    row = conn.execute(
        "SELECT parent_selfmodel_root FROM selfmodel_records WHERE selfmodel_root = ?",
        (root,),
    ).fetchone()
    if row is None:
        return None
    return row["parent_selfmodel_root"]


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument(
        "--shard",
        type=Path,
        default=None,
        help="Chain shard path (default: $ARBORIST_SELFMODEL_CHAIN_DB or ~/.arborist/shards/selfmodel-chain.db)",
    )
    args = p.parse_args(argv)
    shard = args.shard.expanduser() if args.shard else default_shard()
    summary = append_snapshot(shard)
    import json as _json

    print(_json.dumps(summary, indent=2, default=str))
    return 0


if __name__ == "__main__":
    sys.exit(main())