File size: 8,271 Bytes
4f70a26 a042d28 4f70a26 a042d28 4f70a26 a042d28 4f70a26 a042d28 4f70a26 a042d28 4f70a26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 | """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
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("```"):
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]]]:
"""Apply ops, then evict lowest-salience occupied slots until the
occupied count fits the budget (== number of slots). Appends EVICT ops
to the 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]
for _ in range(max(0, len(occupied) - len(ledger))):
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}.
valid_only=True keeps only active slots; False also returns tombstoned
slots (last-known value), which backs temporal questions."""
out: Dict[Tuple[str, str], str] = {}
for s in ledger:
if s.occupied and (s.valid or not valid_only):
out[(s.subject, s.predicate)] = 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),
} |