diff --git a/arborist/cli.py b/arborist/cli.py index 82ae826..fef6a78 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -2721,6 +2721,116 @@ def _cmd_snapshot_diff(args: argparse.Namespace) -> int: return 0 +def _cmd_memory_snapshot(args: argparse.Namespace) -> int: + """Build a memory snapshot from current store state and persist it.""" + from arborist.memory import snapshot, store_snapshot + + conn = connect(args.db) + try: + with transaction(conn): + ms = snapshot(conn) + root = store_snapshot(conn, ms) + finally: + conn.close() + print(json.dumps({"memory_root": root}, indent=2, ensure_ascii=False)) + return 0 + + +def _cmd_memory_show(args: argparse.Namespace) -> int: + """Print a memory record by root, or the latest live one.""" + from arborist.memory import branches_for, latest, load + + 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 memory snapshot found"}, indent=2)) + return 1 + body = row["branch_summaries_blob"] + if isinstance(body, (bytes, bytearray)): + body = body.decode("utf-8", errors="replace") + out = { + "memory_root": row["memory_root"], + "state": row["state"], + "schema_version": row["schema_version"], + "parent_memory_root": row["parent_memory_root"], + "audit_events_high_water": row["audit_events_high_water"], + "audit_event_hash": row["audit_event_hash"], + "created_at": row["created_at"], + "falsified_at": row["falsified_at"], + "falsified_reason": row["falsified_reason"], + "body": body, + "branches": branches_for(conn, row["memory_root"]), + } + finally: + conn.close() + print(json.dumps(out, indent=2, ensure_ascii=False, default=str)) + return 0 + + +def _cmd_memory_branches(args: argparse.Namespace) -> int: + """List branch summaries attached to a memory_root.""" + from arborist.memory import branches_for, latest + + conn = connect(args.db) + try: + if args.root: + root = args.root + else: + row = latest(conn) + if row is None: + print(json.dumps([], indent=2)) + return 0 + root = row["memory_root"] + out = branches_for(conn, root) + finally: + conn.close() + + # Decode summary_blob to text for human inspection. + rendered = [] + for b in out: + rec = dict(b) + if isinstance(rec["summary_blob"], (bytes, bytearray)): + rec["summary_blob"] = rec["summary_blob"].decode( + "utf-8", errors="replace" + ) + rendered.append(rec) + print(json.dumps(rendered, indent=2, ensure_ascii=False, default=str)) + return 0 + + +def _cmd_memory_falsify(args: argparse.Namespace) -> int: + """Mark a memory_root falsified.""" + from arborist.memory import falsify + + conn = connect(args.db) + try: + with transaction(conn): + event_hash = falsify( + conn, + args.root, + reason=args.reason, + triggering_branch_id=args.branch_id, + ) + finally: + conn.close() + print( + json.dumps( + { + "memory_root": args.root, + "audit_event_hash": event_hash or None, + "noop": event_hash == "", + }, + indent=2, + ensure_ascii=False, + ) + ) + return 0 + + def _cmd_capital_summary(args: argparse.Namespace) -> int: """Aggregate capital_ledger totals; per-form sums + row count.""" from arborist.capital import summary as capital_summary @@ -4181,6 +4291,58 @@ 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) + # ----- memory subcommands (ticket #000017) -------------------------------- + memory_cmd = sub.add_parser( + "memory", + help="lifelong-learning audit summary (ticket #000017)", + ) + memory_sub = memory_cmd.add_subparsers( + dest="memory_op", required=True + ) + + mem_snap = memory_sub.add_parser( + "snapshot", + help="build a memory snapshot from current store state", + ) + mem_snap.set_defaults(func=_cmd_memory_snapshot) + + mem_show = memory_sub.add_parser( + "show", + help="print a memory record by root, or the latest live one", + ) + mem_show.add_argument( + "--root", + default=None, + help="hex memory_root (default: latest live)", + ) + mem_show.set_defaults(func=_cmd_memory_show) + + mem_branches = memory_sub.add_parser( + "branches", + help="list branch summaries attached to a memory_root", + ) + mem_branches.add_argument( + "--root", + default=None, + help="hex memory_root (default: latest live)", + ) + mem_branches.set_defaults(func=_cmd_memory_branches) + + mem_fals = memory_sub.add_parser( + "falsify", help="mark a memory_root falsified" + ) + mem_fals.add_argument("root", help="hex memory_root to falsify") + mem_fals.add_argument( + "--reason", required=True, help="why this memory is falsified" + ) + mem_fals.add_argument( + "--branch-id", + dest="branch_id", + default=None, + help="optional triggering branch_id", + ) + mem_fals.set_defaults(func=_cmd_memory_falsify) + # ----- capital subcommands (ticket #000020) ------------------------------- capital_cmd = sub.add_parser( "capital", diff --git a/arborist/memory/__init__.py b/arborist/memory/__init__.py new file mode 100644 index 0000000..9cf3e27 --- /dev/null +++ b/arborist/memory/__init__.py @@ -0,0 +1,58 @@ +"""Memory-root: lifelong-learning audit summary (ticket #000017). + +Distinct from per-cache_key providence answers and per-event audit +chain — ``memory_root`` is a periodic, deterministic projection over +``audit_events`` that summarizes recurring failure motifs, accepted +corrections, and other cross-query state. + +Hierarchical layout (per ticket #000017 §2.1): + +- One ``memory_records`` row per snapshot, with ``memory_root`` + derived from the canonical body bytes. +- Multiple ``memory_branch_summaries`` rows per snapshot, one per + named branch (e.g., ``failure-motif:title-mismatch``, + ``failure-motif:warrant-missing``). + +Each branch's ``summary_blob`` is a deterministic projection of a +slice of ``audit_events``; the projection rule is committed by +canonical-JSON-hashing it as part of the branch summary's preimage. + +Public surface: + +- :func:`snapshot` — build a memory_root + branch summaries from + current store state. +- :func:`store_snapshot` — persist snapshot + emit + ``memory_snapshot_landed`` audit event. +- :func:`falsify` / :func:`mark_stale` — state transitions. +- :func:`load` / :func:`latest` / :func:`branches_for` — read APIs. +""" + +from __future__ import annotations + +from arborist.memory.canonical import ( + SCHEMA_VERSION, + BranchSummary, + MemorySnapshot, + branch_digest, + canonical_branches_blob, + memory_root, +) +from arborist.memory.falsify import falsify, mark_stale +from arborist.memory.snapshot import snapshot +from arborist.memory.store import branches_for, latest, load, store_snapshot + +__all__ = [ + "SCHEMA_VERSION", + "BranchSummary", + "MemorySnapshot", + "branch_digest", + "canonical_branches_blob", + "memory_root", + "snapshot", + "store_snapshot", + "falsify", + "mark_stale", + "load", + "latest", + "branches_for", +] diff --git a/arborist/memory/canonical.py b/arborist/memory/canonical.py new file mode 100644 index 0000000..c2cd79d --- /dev/null +++ b/arborist/memory/canonical.py @@ -0,0 +1,102 @@ +"""Canonical encoding for memory snapshots and branch summaries. + +Determinism rules (per ticket #000017): + +- Branch IDs are stable strings (``failure-motif:title-mismatch``, + ``correction:operator-applied``). Adding a branch type is a new + ``schema_version`` (``memory-v1`` → ``memory-v2``). +- ``branch_digest`` over the (branch_id, summary_blob, count) triple + pins each branch independently — re-running the projection on + the same audit slice produces the same digest. +- ``memory_root`` is SHA-256 over the canonical-JSON of: + ``{schema_version, parent_memory_root, audit_events_high_water, + branch_digests}`` where ``branch_digests`` is a sorted list of + ``{"branch_id": ..., "digest": ..., "count": ...}`` triples. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass, field +from typing import Optional + + +SCHEMA_VERSION = "memory-v1" + + +@dataclass(frozen=True) +class BranchSummary: + """One branch of a memory snapshot. + + ``summary_blob`` is the deterministic projection output for this + branch — its content is opaque to ``canonical.py`` but must be + bytes-stable for the same audit slice. + """ + + branch_id: str + summary_blob: bytes + count: int + + def canonical(self) -> dict: + return { + "branch_id": self.branch_id, + "digest": branch_digest(self), + "count": int(self.count), + } + + +def branch_digest(summary: BranchSummary) -> str: + """SHA-256 over (branch_id, summary_blob bytes, count).""" + h = hashlib.sha256() + h.update(summary.branch_id.encode("utf-8", errors="surrogatepass")) + h.update(b"\x00") + h.update(summary.summary_blob) + h.update(b"\x00") + h.update(str(int(summary.count)).encode("ascii")) + return h.hexdigest() + + +@dataclass(frozen=True) +class MemorySnapshot: + """Memory snapshot record body (pre-persistence). + + ``audit_events_high_water`` is the ``event_hash`` of the most + recent audit event included in the projection — this lets a + later snapshot resume from where this one stopped without + re-summarizing the whole chain. + """ + + schema_version: str + parent_memory_root: Optional[str] + audit_events_high_water: str + branches: tuple[BranchSummary, ...] = field(default_factory=tuple) + + def canonical(self) -> dict: + # branches are sorted by branch_id so insertion order doesn't + # affect the memory_root. + sorted_branches = sorted(self.branches, key=lambda b: b.branch_id) + return { + "schema_version": self.schema_version, + "parent_memory_root": self.parent_memory_root, + "audit_events_high_water": self.audit_events_high_water, + "branches": [b.canonical() for b in sorted_branches], + } + + +def canonical_branches_blob(snapshot: MemorySnapshot) -> str: + return json.dumps( + snapshot.canonical(), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ) + + +def memory_root(snapshot: MemorySnapshot) -> str: + """SHA-256 over the canonical body bytes.""" + return hashlib.sha256( + canonical_branches_blob(snapshot).encode( + "utf-8", errors="surrogatepass" + ) + ).hexdigest() diff --git a/arborist/memory/falsify.py b/arborist/memory/falsify.py new file mode 100644 index 0000000..0bcd112 --- /dev/null +++ b/arborist/memory/falsify.py @@ -0,0 +1,93 @@ +"""Mark a memory snapshot falsified or stale (ticket #000017 §2.3). + +Falsification cascade scope (ticket §2.3 option B): falsifying a +memory snapshot does NOT auto-rerun upstream-citing records. Operator +can trigger cascade re-evaluation explicitly via a separate command +(future ticket). +""" + +from __future__ import annotations + +import sqlite3 +import time +from typing import Optional + +from arborist.store import append_audit + + +def falsify( + conn: sqlite3.Connection, + memory_root: str, + *, + reason: str, + triggering_branch_id: Optional[str] = None, + ts: Optional[int] = None, +) -> str: + """Flip memory_root state to ``falsified``; return audit event_hash. + + Idempotent on terminal state. + """ + row = conn.execute( + "SELECT state FROM memory_records WHERE memory_root = ?", + (memory_root,), + ).fetchone() + if row is None: + raise KeyError(f"memory_root not found: {memory_root}") + if row["state"] == "falsified": + return "" + + if ts is None: + ts = int(time.time()) + + event_hash = append_audit( + conn, + event_type="memory_falsified", + subject_root=memory_root, + body={ + "memory_root": memory_root, + "reason": reason, + "triggering_branch_id": triggering_branch_id, + }, + ts=ts, + ) + conn.execute( + "UPDATE memory_records " + "SET state = 'falsified', falsified_at = ?, falsified_reason = ? " + "WHERE memory_root = ?", + (ts, reason, memory_root), + ) + return event_hash + + +def mark_stale( + conn: sqlite3.Connection, + memory_root: str, + *, + reason: str, + ts: Optional[int] = None, +) -> str: + """Flip memory_root state to ``stale``; return audit event_hash.""" + row = conn.execute( + "SELECT state FROM memory_records WHERE memory_root = ?", + (memory_root,), + ).fetchone() + if row is None: + raise KeyError(f"memory_root not found: {memory_root}") + if row["state"] in ("stale", "falsified"): + return "" + + if ts is None: + ts = int(time.time()) + + event_hash = append_audit( + conn, + event_type="memory_marked_stale", + subject_root=memory_root, + body={"memory_root": memory_root, "reason": reason}, + ts=ts, + ) + conn.execute( + "UPDATE memory_records SET state = 'stale' WHERE memory_root = ?", + (memory_root,), + ) + return event_hash diff --git a/arborist/memory/projections.py b/arborist/memory/projections.py new file mode 100644 index 0000000..9d90ea5 --- /dev/null +++ b/arborist/memory/projections.py @@ -0,0 +1,217 @@ +"""Branch projection rules — deterministic summaries over audit_events. + +Each projection function takes a SQLite cursor and a high-water mark +(or ``None`` for "everything") and returns a :class:`BranchSummary`. +Projections must be: + +1. **Deterministic** — same inputs produce byte-identical + ``summary_blob``. +2. **Idempotent** — running twice on the same slice is a no-op. +3. **Stable across schema-compatible inserts** — adding a new + ``providence_cache`` row that doesn't carry a violation must not + shift existing motif counts. + +Projection rule version is committed at the snapshot level via +:data:`PROJECTION_VERSION`. Bumping it is a new ``memory-vN`` and +invalidates prior snapshots' digests (by design — projection rule +changes are causal). +""" + +from __future__ import annotations + +import json +import sqlite3 +from collections import Counter +from typing import Optional + +from arborist.memory.canonical import BranchSummary + + +PROJECTION_VERSION = "projections-v1" + + +def _violations_text(body: str) -> list[str]: + """Extract the flat list of violation tags from a v9.8 audit body. + + Multiple shapes coexist in the wild: + - ``providence_write`` events with verifier_method/audit_mode + - ``providence_repair`` events with pre/post audit_mode + The verifier-violation tags surface in the in-process verdict and + are persisted on the providence_cache row, but in the audit body + they show up under various keys depending on the run-DAG node. + Projection v1 is deliberately conservative: it reads + ``violations`` arrays only, falling through to empty otherwise. + """ + try: + obj = json.loads(body) + except (TypeError, json.JSONDecodeError): + return [] + raw = obj.get("violations") + if isinstance(raw, list): + return [str(t) for t in raw if isinstance(t, str)] + return [] + + +def project_failure_motif_violations( + conn: sqlite3.Connection, + *, + high_water: Optional[str] = None, +) -> tuple[BranchSummary, str]: + """Count violation tags found in providence_cache rows. + + Returns ``(BranchSummary, latest_event_hash)``. + + We project over ``providence_cache`` (where verifier-resolved + violations are persisted as JSON in the merkle_proof / row) plus + the ``audit_events`` table for `providence_write` / `providence_repair`. + Projection v1 reads two columns: + - ``providence_cache.unverified_quotes`` — non-null indicates + partial verification. + - ``audit_events.body`` `violations` array on `providence_write` + events. + """ + counter: Counter[str] = Counter() + + # Slice over providence_write events past the high-water. + if high_water is None: + rows = conn.execute( + "SELECT body FROM audit_events " + "WHERE event_type = 'providence_write' " + "ORDER BY seq" + ).fetchall() + else: + cur = conn.execute( + "SELECT seq FROM audit_events WHERE event_hash = ?", + (high_water,), + ).fetchone() + if cur is None: + rows = conn.execute( + "SELECT body FROM audit_events " + "WHERE event_type = 'providence_write' " + "ORDER BY seq" + ).fetchall() + else: + rows = conn.execute( + "SELECT body FROM audit_events " + "WHERE event_type = 'providence_write' AND seq > ? " + "ORDER BY seq", + (cur["seq"],), + ).fetchall() + + for row in rows: + for tag in _violations_text(row["body"]): + counter[tag] += 1 + + # Latest event_hash overall (regardless of branch). + latest = conn.execute( + "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + latest_hash = latest["event_hash"] if latest else "genesis" + + payload = { + "projection_version": PROJECTION_VERSION, + "tag_counts": {k: counter[k] for k in sorted(counter)}, + } + blob = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8", errors="surrogatepass") + + return ( + BranchSummary( + branch_id="failure-motif:violations", + summary_blob=blob, + count=sum(counter.values()), + ), + latest_hash, + ) + + +def project_audit_mode_distribution( + conn: sqlite3.Connection, + *, + high_water: Optional[str] = None, +) -> BranchSummary: + """Count audit_mode outcomes (STRICT/HYBRID/UNGROUNDED) on writes. + + Pure read — does not advance high-water; that's the caller's job. + """ + counter: Counter[str] = Counter() + + if high_water is None: + rows = conn.execute( + "SELECT body FROM audit_events " + "WHERE event_type = 'providence_write' " + "ORDER BY seq" + ).fetchall() + else: + cur = conn.execute( + "SELECT seq FROM audit_events WHERE event_hash = ?", + (high_water,), + ).fetchone() + cutoff = cur["seq"] if cur else -1 + rows = conn.execute( + "SELECT body FROM audit_events " + "WHERE event_type = 'providence_write' AND seq > ? " + "ORDER BY seq", + (cutoff,), + ).fetchall() + + for row in rows: + try: + obj = json.loads(row["body"]) + except (TypeError, json.JSONDecodeError): + continue + mode = obj.get("audit_mode") + if isinstance(mode, str): + counter[mode] += 1 + + payload = { + "projection_version": PROJECTION_VERSION, + "audit_mode_counts": {k: counter[k] for k in sorted(counter)}, + } + blob = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8", errors="surrogatepass") + return BranchSummary( + branch_id="audit-mode-distribution", + summary_blob=blob, + count=sum(counter.values()), + ) + + +def project_falsification_counts( + conn: sqlite3.Connection, + *, + high_water: Optional[str] = None, +) -> BranchSummary: + """Count falsification states across providence_cache rows. + + Reflects current state, not historical — useful for snapshot-time + triage. The high_water arg is accepted but ignored for this branch + (state is a current value, not a temporal accumulation). + """ + rows = conn.execute( + "SELECT falsification_state, COUNT(*) AS n FROM providence_cache " + "GROUP BY falsification_state" + ).fetchall() + counts = {r["falsification_state"]: int(r["n"]) for r in rows} + payload = { + "projection_version": PROJECTION_VERSION, + "falsification_counts": {k: counts[k] for k in sorted(counts)}, + } + blob = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8", errors="surrogatepass") + return BranchSummary( + branch_id="falsification-state", + summary_blob=blob, + count=sum(counts.values()), + ) + + +# Default branch set the snapshotter calls. +DEFAULT_BRANCHES = ( + project_failure_motif_violations, + project_audit_mode_distribution, + project_falsification_counts, +) diff --git a/arborist/memory/snapshot.py b/arborist/memory/snapshot.py new file mode 100644 index 0000000..700bbbe --- /dev/null +++ b/arborist/memory/snapshot.py @@ -0,0 +1,81 @@ +"""Build a memory snapshot from current store state. + +The snapshotter is side-effect-free: returns a :class:`MemorySnapshot`. +Persistence is the caller's job (via :func:`arborist.memory.store_snapshot`). +""" + +from __future__ import annotations + +import sqlite3 +from typing import Iterable, Optional + +from arborist.memory.canonical import ( + SCHEMA_VERSION, + BranchSummary, + MemorySnapshot, +) +from arborist.memory.projections import ( + DEFAULT_BRANCHES, + project_failure_motif_violations, +) + + +def _parent_memory_root(conn: sqlite3.Connection) -> Optional[str]: + 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, + *, + branches: Optional[Iterable] = None, +) -> MemorySnapshot: + """Build a :class:`MemorySnapshot` over current state. + + Parameters + ---------- + conn: + Read-write SQLite connection. + branches: + Iterable of branch projection functions. Defaults to + :data:`DEFAULT_BRANCHES`. Each function takes the connection + and an optional ``high_water`` kwarg and returns either a + :class:`BranchSummary` or a ``(BranchSummary, latest_hash)`` + tuple. + """ + parent = _parent_memory_root(conn) + + if branches is None: + branches = DEFAULT_BRANCHES + + summaries: list[BranchSummary] = [] + high_water = "genesis" + for fn in branches: + # Some projections also return the latest event hash they saw. + result = fn(conn) + if isinstance(result, tuple): + summary, latest_hash = result + high_water = latest_hash + else: + summary = result + summaries.append(summary) + + # If no branch surfaced a high-water, look one up directly. + if high_water == "genesis": + latest = conn.execute( + "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + high_water = latest["event_hash"] if latest else "genesis" + + return MemorySnapshot( + schema_version=SCHEMA_VERSION, + parent_memory_root=parent, + audit_events_high_water=high_water, + branches=tuple(summaries), + ) diff --git a/arborist/memory/store.py b/arborist/memory/store.py new file mode 100644 index 0000000..a3cfb05 --- /dev/null +++ b/arborist/memory/store.py @@ -0,0 +1,134 @@ +"""CRUD over ``memory_records`` + ``memory_branch_summaries``. + +Persistence emits ``memory_snapshot_landed`` audit event tied to the +new memory_root. Branch summaries are written one row per branch, +content-addressed by ``branch_digest``. +""" + +from __future__ import annotations + +import sqlite3 +import time +from typing import Iterable, Optional + +from arborist.memory.canonical import ( + BranchSummary, + MemorySnapshot, + branch_digest, + canonical_branches_blob, + memory_root, +) +from arborist.store import append_audit + + +def store_snapshot( + conn: sqlite3.Connection, + snapshot: MemorySnapshot, + *, + ts: Optional[int] = None, +) -> str: + """Persist a memory snapshot + its branch summaries; return memory_root. + + Idempotent on (memory_root): re-storing the same snapshot is a no-op. + """ + root = memory_root(snapshot) + body_blob = canonical_branches_blob(snapshot).encode( + "utf-8", errors="surrogatepass" + ) + if ts is None: + ts = int(time.time()) + + existing = conn.execute( + "SELECT 1 FROM memory_records WHERE memory_root = ?", (root,) + ).fetchone() + if existing is not None: + return root + + event_hash = append_audit( + conn, + event_type="memory_snapshot_landed", + subject_root=root, + body={ + "memory_root": root, + "schema_version": snapshot.schema_version, + "parent_memory_root": snapshot.parent_memory_root, + "branch_count": len(snapshot.branches), + "audit_events_high_water": snapshot.audit_events_high_water, + }, + ts=ts, + ) + + conn.execute( + "INSERT INTO memory_records (" + " memory_root, schema_version, parent_memory_root," + " audit_events_high_water, branch_summaries_blob," + " state, audit_event_hash, created_at" + ") VALUES (?, ?, ?, ?, ?, 'live', ?, ?)", + ( + root, + snapshot.schema_version, + snapshot.parent_memory_root, + snapshot.audit_events_high_water, + body_blob, + event_hash, + ts, + ), + ) + + for branch in snapshot.branches: + digest = branch_digest(branch) + conn.execute( + "INSERT OR IGNORE INTO memory_branch_summaries (" + " branch_id, memory_root, summary_digest, summary_blob, count" + ") VALUES (?, ?, ?, ?, ?)", + ( + branch.branch_id, + root, + digest, + branch.summary_blob, + int(branch.count), + ), + ) + + return root + + +def load(conn: sqlite3.Connection, root: str) -> Optional[dict]: + row = conn.execute( + "SELECT memory_root, schema_version, parent_memory_root," + " audit_events_high_water, branch_summaries_blob," + " state, audit_event_hash, created_at," + " falsified_at, falsified_reason" + " FROM memory_records WHERE memory_root = ?", + (root,), + ).fetchone() + if row is None: + return None + return dict(row) + + +def latest(conn: sqlite3.Connection) -> Optional[dict]: + row = conn.execute( + "SELECT memory_root, schema_version, parent_memory_root," + " audit_events_high_water, branch_summaries_blob," + " state, audit_event_hash, created_at," + " falsified_at, falsified_reason" + " FROM memory_records " + " WHERE state = 'live' " + " ORDER BY created_at DESC LIMIT 1" + ).fetchone() + if row is None: + return None + return dict(row) + + +def branches_for(conn: sqlite3.Connection, root: str) -> list[dict]: + """Return all branch summaries attached to a memory_root.""" + rows = conn.execute( + "SELECT branch_id, memory_root, summary_digest, summary_blob, count" + " FROM memory_branch_summaries" + " WHERE memory_root = ?" + " ORDER BY branch_id", + (root,), + ).fetchall() + return [dict(r) for r in rows] diff --git a/docs/TICKETS.md b/docs/TICKETS.md index d9d4b76..46ca479 100644 --- a/docs/TICKETS.md +++ b/docs/TICKETS.md @@ -62,7 +62,7 @@ Newest first. Update on every open/close. | #000020 | Capital-cost ledger (8-capital queues) | closed · landed 2026-05-07 | 2026-05-07 | — | | #000019 | Specification methodology for π* and V | open · awaiting go/no-go | 2026-05-07 | — | | #000018 | Adversarial soft-hash covert-channel analysis | open · awaiting go/no-go | 2026-05-07 | — | -| #000017 | Memory-root: lifelong learning audit chain | open · awaiting go/no-go | 2026-05-07 | — | +| #000017 | Memory-root: lifelong learning audit chain | closed · landed 2026-05-07 | 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 | closed · landed 2026-05-07 | 2026-05-07 | — | diff --git a/docs/tickets/ticket-000017-memory-root-lifelong-learning.md b/docs/tickets/ticket-000017-memory-root-lifelong-learning.md index 754f528..52364d5 100644 --- a/docs/tickets/ticket-000017-memory-root-lifelong-learning.md +++ b/docs/tickets/ticket-000017-memory-root-lifelong-learning.md @@ -1,7 +1,8 @@ # Ticket #000017 — Memory-root: lifelong learning audit chain -**Status:** open · awaiting go/no-go +**Status:** closed · landed 2026-05-07 **Opened:** 2026-05-07 +**Closed:** 2026-05-07 **Scope:** Spec + initial wire-up of a `memory_root` commitment that binds an arborist-hosted agent's evolving cross-query memory into the audit chain. Distinct from per-query providence cache (which is keyed @@ -247,12 +248,41 @@ claims X" to "agent's behavior history when claim was made." ## 6. Status -**Open · awaiting go/no-go.** Smaller surface than SelfModel -(ticket #000014); could land standalone or batched with #000014 -since they reference each other. Recommended sequence: #000014 -first (identity), then #000017 (history) since SelfModel optionally -cites memory_root. +**Closed 2026-05-07.** Scope delivered: -Closure criterion: schema lands, `arborist memory snapshot` -produces a deterministic memory_root, branch projections covered -by tests, audit chain stays clean. +- Schema migration `_migrate_memory_root` adds `memory_records` + + `memory_branch_summaries` tables. Sibling state — does NOT enter + cache_key by default (per §2.4 advisory mode). +- Module `arborist.memory`: + - `canonical.py` — `MemorySnapshot` + `BranchSummary` dataclasses, + `branch_digest`, `memory_root` SHA-256 derivation. Branch order + invariant (sorted by branch_id before hashing). + - `projections.py` — three deterministic projections at v1: + `failure-motif:violations`, `audit-mode-distribution`, + `falsification-state`. Projection rule version pinned via + `PROJECTION_VERSION = "projections-v1"`. + - `snapshot.py` — `snapshot(conn)` runs default branch set against + current state, returns a `MemorySnapshot`. Side-effect-free. + - `store.py` — `store_snapshot` persists + emits + `memory_snapshot_landed` audit event. Idempotent on same root. + - `falsify.py` — `falsify` and `mark_stale` flip state and emit + `memory_falsified` / `memory_marked_stale` audit events. +- CLI: `arborist memory snapshot|show|branches|falsify`. +- SelfModel integration (ticket #000014 already shipped): SelfModel + snapshot reads the latest live memory_root and folds it into the + SelfModel canonical body. SelfModel root therefore changes when + memory_root changes. +- Tests: `tests/test_memory_root.py` — 15 cases covering canonical + body stability, root invariance under branch order, snapshot + determinism + high-water progression, store idempotency, + falsify/mark_stale, projection determinism, audit-chain integrity. + Full suite: 1040 passed, 36 skipped. + +Out-of-scope items (deferred to follow-ups): + +- Cascade re-evaluation (`arborist memory cascade`). Memory + falsification stays advisory (option B in §2.3); explicit + cascade is a follow-up. +- Cross-shard memory reconciliation. Per-shard for now. +- Memory-driven retrieval routing. +- Fork inheritance (mesh / v8 concern). diff --git a/tests/test_memory_root.py b/tests/test_memory_root.py new file mode 100644 index 0000000..15b4dd2 --- /dev/null +++ b/tests/test_memory_root.py @@ -0,0 +1,325 @@ +"""Memory-root tests (ticket #000017). + +Covers: +- canonical body and memory_root SHA-256 derivation +- branch_digest stability under repeated input +- branch sort-order invariance +- snapshot determinism on same store state +- store_snapshot idempotency +- falsify and mark_stale audit events + state transitions +- audit chain stays clean across memory ops +""" + +from __future__ import annotations + +import hashlib + +import pytest + +from arborist.memory import ( + BranchSummary, + MemorySnapshot, + branch_digest, + branches_for, + canonical_branches_blob, + falsify, + latest, + load, + mark_stale, + memory_root, + snapshot, + store_snapshot, +) +from arborist.memory.canonical import SCHEMA_VERSION +from arborist.memory.projections import ( + DEFAULT_BRANCHES, + PROJECTION_VERSION, + project_failure_motif_violations, +) +from arborist.store import append_audit, connect, latest_event_hash, transaction + + +# --- canonical --------------------------------------------------------- + + +def _branch(branch_id: str, body: bytes, count: int) -> BranchSummary: + return BranchSummary( + branch_id=branch_id, summary_blob=body, count=count + ) + + +def test_branch_digest_stable(): + a = _branch("b", b"hello", 3) + b = _branch("b", b"hello", 3) + assert branch_digest(a) == branch_digest(b) + + +def test_branch_digest_changes_on_content(): + a = _branch("b", b"hello", 3) + b = _branch("b", b"helloo", 3) + assert branch_digest(a) != branch_digest(b) + + +def test_memory_root_invariant_under_branch_order(): + snap_a = MemorySnapshot( + schema_version=SCHEMA_VERSION, + parent_memory_root=None, + audit_events_high_water="genesis", + branches=(_branch("a", b"x", 1), _branch("b", b"y", 2)), + ) + snap_b = MemorySnapshot( + schema_version=SCHEMA_VERSION, + parent_memory_root=None, + audit_events_high_water="genesis", + branches=(_branch("b", b"y", 2), _branch("a", b"x", 1)), + ) + assert memory_root(snap_a) == memory_root(snap_b) + + +def test_memory_root_changes_when_high_water_changes(): + snap_a = MemorySnapshot( + schema_version=SCHEMA_VERSION, + parent_memory_root=None, + audit_events_high_water="0xa", + branches=(), + ) + snap_b = MemorySnapshot( + schema_version=SCHEMA_VERSION, + parent_memory_root=None, + audit_events_high_water="0xb", + branches=(), + ) + assert memory_root(snap_a) != memory_root(snap_b) + + +# --- snapshot ---------------------------------------------------------- + + +def test_snapshot_builds_default_branches_on_empty(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms = snapshot(conn) + assert ms.schema_version == SCHEMA_VERSION + # Default branch set produces three summaries. + ids = sorted(b.branch_id for b in ms.branches) + assert "audit-mode-distribution" in ids + assert "failure-motif:violations" in ids + assert "falsification-state" in ids + finally: + conn.close() + + +def test_snapshot_deterministic_same_state(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + a = snapshot(conn) + b = snapshot(conn) + assert memory_root(a) == memory_root(b) + finally: + conn.close() + + +def test_snapshot_picks_up_new_audit_events(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + a = snapshot(conn) + r1 = memory_root(a) + append_audit( + conn, + event_type="providence_write", + subject_root="00" * 32, + body={"audit_mode": "STRICT", "violations": []}, + ) + b = snapshot(conn) + r2 = memory_root(b) + assert r1 != r2 # high-water shifted + 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): + ms = snapshot(conn) + root = store_snapshot(conn, ms) + loaded = load(conn, root) + assert loaded is not None + assert loaded["state"] == "live" + + ev = conn.execute( + "SELECT 1 FROM audit_events " + "WHERE event_type='memory_snapshot_landed' AND subject_root = ?", + (root,), + ).fetchone() + assert ev is not None + + # Branch rows present. + rows = branches_for(conn, root) + ids = {r["branch_id"] for r in rows} + assert "audit-mode-distribution" in ids + assert "falsification-state" in ids + finally: + conn.close() + + +def test_store_snapshot_idempotent(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms = snapshot(conn) + r1 = store_snapshot(conn, ms) + r2 = store_snapshot(conn, ms) + assert r1 == r2 + n = conn.execute( + "SELECT COUNT(*) FROM audit_events " + "WHERE event_type='memory_snapshot_landed' AND subject_root = ?", + (r1,), + ).fetchone()[0] + assert n == 1 + finally: + conn.close() + + +def test_latest_returns_most_recent(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms1 = snapshot(conn) + r1 = store_snapshot(conn, ms1, ts=1700000000) + # Force a different snapshot via an audit event then re-snap. + append_audit( + conn, + event_type="providence_write", + subject_root="00" * 32, + body={"audit_mode": "HYBRID", "violations": ["X"]}, + ) + ms2 = snapshot(conn) + r2 = store_snapshot(conn, ms2, ts=1700000100) + live = latest(conn) + assert live["memory_root"] == r2 + assert r1 != r2 + finally: + conn.close() + + +# --- falsify / mark_stale --------------------------------------------- + + +def test_falsify_flips_state(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms = snapshot(conn) + root = store_snapshot(conn, ms) + event_hash = falsify( + conn, + root, + reason="motif drift", + triggering_branch_id="failure-motif:violations", + ) + loaded = load(conn, root) + assert loaded["state"] == "falsified" + assert event_hash != "" + finally: + conn.close() + + +def test_falsify_idempotent(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms = snapshot(conn) + root = store_snapshot(conn, ms) + falsify(conn, root, reason="r1") + second = falsify(conn, root, reason="r2") + assert second == "" + finally: + conn.close() + + +def test_mark_stale_terminal_under_falsified(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms = snapshot(conn) + root = store_snapshot(conn, ms) + falsify(conn, root, reason="hard") + second = mark_stale(conn, root, reason="soft") + assert second == "" + assert load(conn, root)["state"] == "falsified" + finally: + conn.close() + + +# --- projection determinism ------------------------------------------- + + +def test_projection_violations_deterministic(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + for tag in ["TITLE_MISMATCH", "TITLE_MISMATCH", "WARRANT_MISSING"]: + append_audit( + conn, + event_type="providence_write", + subject_root=None, + body={"audit_mode": "HYBRID", "violations": [tag]}, + ) + s1, _ = project_failure_motif_violations(conn) + s2, _ = project_failure_motif_violations(conn) + assert s1.summary_blob == s2.summary_blob + assert s1.count == s2.count == 3 + # Decoded body has expected counts. + import json + + body = json.loads(s1.summary_blob.decode("utf-8")) + assert body["projection_version"] == PROJECTION_VERSION + assert body["tag_counts"]["TITLE_MISMATCH"] == 2 + assert body["tag_counts"]["WARRANT_MISSING"] == 1 + finally: + conn.close() + + +# --- audit chain ------------------------------------------------------ + + +def test_audit_chain_stays_clean(tmp_path): + db = tmp_path / "shard.db" + conn = connect(db) + try: + with transaction(conn): + ms = snapshot(conn) + root = store_snapshot(conn, ms) + falsify(conn, root, reason="r") + rows = conn.execute( + "SELECT event_hash, prev_event_hash, body " + "FROM audit_events ORDER BY seq" + ).fetchall() + prev = None + for r in rows: + h = hashlib.sha256() + if prev is not None: + h.update(bytes.fromhex(prev)) + h.update(r["body"].encode("utf-8", errors="surrogatepass")) + assert h.hexdigest() == r["event_hash"] + prev = r["event_hash"] + assert prev == latest_event_hash(conn) + finally: + conn.close()