CLERK core: slot schema, edit-op schema, deterministic reducer, serializers

clerk/common.py

@@ -0,0 +1,225 @@
+"""CLERK: Consolidated Ledger with Eviction and Rewrite Keys.
+
+Core memory machinery shared by the data generator, the training harness and
+the evaluators:
+
+  * ``Slot``        -- one memory slot (a typed fact plus lifecycle metadata)
+  * edit-op schema  -- ADD / UPDATE / TOMBSTONE / EVICT / NOOP
+  * ``apply_ops``   -- the DETERMINISTIC reducer that applies a model-emitted
+                       edit program to a ledger
+  * serializers     -- compact JSON forms of ledgers and op programs
+
+The reducer is the single source of truth: the generator emits gold op
+programs, and every downstream consumer applies them through ``apply_ops``.
+This guarantees that the training supervision, the simulated model rollouts
+and the evaluation all execute the exact same state transition.
+
+Author: Justin Wolcott (fallnai-research.org)
+License: Apache-2.0
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Tuple
+
+LEDGER_BUDGET_DEFAULT = 24
+VALID_OPS = ("ADD", "UPDATE", "TOMBSTONE", "EVICT", "NOOP")
+
+
+@dataclass
+class Slot:
+    """A single ledger slot."""
+
+    id: int
+    subject: str = ""
+    predicate: str = ""
+    object: str = ""
+    t_last: int = -1          # session index when this slot was last written
+    n_uses: int = 0           # number of times the fact has been queried (internal)
+    valid: bool = True        # False => tombstoned (fact no longer true)
+    occupied: bool = False    # False => free (empty, or freed by EVICT)
+
+
+def empty_ledger(budget: int = LEDGER_BUDGET_DEFAULT) -> List[Slot]:
+    return [Slot(id=i) for i in range(budget)]
+
+
+def serialize_ledger(ledger: List[Slot]) -> str:
+    """Compact JSON view of a ledger, as seen by the model.
+
+    Tombstoned slots stay visible (with v=0) until they are evicted, so
+    temporal questions about superseded values remain answerable.
+    """
+    rows = []
+    for s in ledger:
+        if s.occupied:
+            rows.append(
+                {"i": s.id, "s": s.subject, "p": s.predicate,
+                 "o": s.object, "t": s.t_last, "v": 1 if s.valid else 0}
+            )
+    return json.dumps(rows, separators=(",", ":"), ensure_ascii=False)
+
+
+def parse_ledger(text: str, budget: int = LEDGER_BUDGET_DEFAULT) -> List[Slot]:
+    """Rebuild a ledger from its serialized form."""
+    ledger = empty_ledger(budget)
+    for row in json.loads(text):
+        s = ledger[row["i"]]
+        s.occupied = True
+        s.subject = row["s"]
+        s.predicate = row["p"]
+        s.object = row["o"]
+        s.t_last = row["t"]
+        s.valid = bool(row["v"])
+    return ledger
+
+
+def parse_ops(text: str) -> Tuple[List[Dict[str, Any]], str]:
+    """Parse a model-emitted edit program.
+
+    Returns (ops, error). ``error`` is "" on success, otherwise a short
+    description of the first problem found. Malformed programs are surfaced
+    to the evaluation harness instead of being silently dropped.
+    """
+    text = text.strip()
+    if text.startswith("```"):
+        # strip a markdown fence if the model adds one
+        text = text.strip("`")
+        if text.startswith("json"):
+            text = text[4:]
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError as e:
+        return [], f"json_decode_error: {e}"
+    if not isinstance(payload, dict) or "ops" not in payload:
+        return [], "missing_top_level_ops_key"
+    ops = payload["ops"]
+    if not isinstance(ops, list):
+        return [], "ops_not_a_list"
+    clean: List[Dict[str, Any]] = []
+    for op in ops:
+        if not isinstance(op, dict) or "op" not in op:
+            return clean, "op_missing_op_key"
+        if op["op"] not in VALID_OPS:
+            return clean, f"unknown_op:{op['op']}"
+        clean.append(op)
+    return clean, ""
+
+
+def apply_ops(
+    ledger: List[Slot],
+    ops: List[Dict[str, Any]],
+    session_idx: int,
+) -> Tuple[List[Slot], Dict[str, int]]:
+    """Apply an edit program to a ledger. Deterministic; mutates a copy.
+
+    Rules:
+      ADD       -- write into the lowest free slot; if the ledger has no free
+                   slot the op is DROPPED and counted (a trained policy must
+                   EVICT first; the gold generator always does).
+      UPDATE    -- overwrite s/p/o of slot ``i`` and bump t_last. The slot
+                   keeps its identity, so supersession is explicit.
+      TOMBSTONE -- mark slot ``i`` invalid (v=0); content stays until EVICT.
+      EVICT     -- free slot ``i`` entirely.
+      NOOP      -- nothing.
+    """
+    new_ledger = [Slot(**vars(s)) for s in ledger]
+    stats = {"ADD": 0, "UPDATE": 0, "TOMBSTONE": 0, "EVICT": 0,
+             "NOOP": 0, "dropped": 0}
+    for op in ops:
+        kind = op["op"]
+        stats[kind] += 1
+        if kind == "NOOP":
+            continue
+        if kind == "ADD":
+            target = next((s for s in new_ledger if not s.occupied), None)
+            if target is None:
+                stats["dropped"] += 1
+                continue
+            target.occupied = True
+            target.valid = True
+            target.t_last = session_idx
+            target.n_uses = 0
+            target.subject = str(op.get("s", ""))
+            target.predicate = str(op.get("p", ""))
+            target.object = str(op.get("o", ""))
+        elif kind == "UPDATE":
+            i = int(op.get("i", -1))
+            if not (0 <= i < len(new_ledger)) or not new_ledger[i].occupied:
+                stats["dropped"] += 1
+                continue
+            s = new_ledger[i]
+            if op.get("s") is not None:
+                s.subject = str(op["s"])
+            if op.get("p") is not None:
+                s.predicate = str(op["p"])
+            s.object = str(op.get("o", s.object))
+            s.t_last = session_idx
+            s.valid = True
+        elif kind == "TOMBSTONE":
+            i = int(op.get("i", -1))
+            if not (0 <= i < len(new_ledger)) or not new_ledger[i].occupied:
+                stats["dropped"] += 1
+                continue
+            new_ledger[i].valid = False
+        elif kind == "EVICT":
+            i = int(op.get("i", -1))
+            if not (0 <= i < len(new_ledger)) or not new_ledger[i].occupied:
+                stats["dropped"] += 1
+                continue
+            new_ledger[i] = Slot(id=i)
+    return new_ledger, stats
+
+
+def salience(slot: Slot, now: int) -> float:
+    """Gold salience score used by the generator's eviction oracle.
+
+    Frequently-queried, currently-valid, recently-written facts are kept;
+    old, never-queried, tombstoned facts are evicted first. This is a
+    documented oracle, not a model component: the generator needs a
+    deterministic ground truth for eviction supervision.
+    """
+    age = max(0, now - slot.t_last)
+    return 4.0 * slot.n_uses + (3.0 if slot.valid else 0.0) + max(0.0, 4.0 - age)
+
+
+def enforce_budget(
+    ledger: List[Slot], ops: List[Dict[str, Any]], now: int
+) -> Tuple[List[Slot], List[Dict[str, Any]]]:
+    """After applying gold ADD/UPDATE ops, evict lowest-salience occupied
+    slots until the active-slot count fits the budget. Appends the EVICT ops
+    to the gold program so the policy learns budget-aware behavior."""
+    ledger, _ = apply_ops(ledger, ops, now)
+    ops = list(ops)
+    occupied = [s for s in ledger if s.occupied]
+    overflow = len(occupied) - len(ledger)
+    for _ in range(max(0, overflow)):
+        victim = min(
+            (s for s in ledger if s.occupied), key=lambda s: salience(s, now)
+        )
+        ops.append({"op": "EVICT", "i": victim.id})
+        ledger[victim.id] = Slot(id=victim.id)
+    return ledger, ops
+
+
+def facts_of(ledger: List[Slot], valid_only: bool = True) -> Dict[Tuple[str, str], str]:
+    """Current fact map {(subject, predicate): object}."""
+    out: Dict[Tuple[str, str], str] = {}
+    for s in ledger:
+        if s.occupied and (s.valid or not valid_only):
+            key = (s.subject, s.predicate)
+            if s.valid:
+                out[key] = s.object
+    return out
+
+
+def ledger_stats(ledger: List[Slot]) -> Dict[str, int]:
+    occ = [s for s in ledger if s.occupied]
+    return {
+        "occupied": len(occ),
+        "valid": sum(1 for s in occ if s.valid),
+        "tombstoned": sum(1 for s in occ if not s.valid),
+        "budget": len(ledger),
+    }