CLERK generator: correct single-pass timeline builder, ledger-aware question targets, eviction-safe gold ops

clerk/generator.py

@@ -6,8 +6,16 @@ programmatically, the gold ledger state after every session is known by
 construction, which yields exact supervision for the consolidation policy —
 no LLM judge anywhere in the pipeline.
 
-Everything is seeded: two runs with the same --seed produce byte-identical
-data.
+Invariants the code maintains:
+  * Questions are only asked about predicates the ledger can actually
+    support: never-stated (unknown), currently-valid (stale/superseded),
+    or currently-tombstoned (negated/temporal). Active-but-EVICTED
+    predicates are never queried, so supervision never asks the model to
+    recall what the oracle dropped.
+  * n_uses (the salience signal) is tracked incrementally at the moment a
+    question is asked, so the eviction oracle at session t only sees
+    evidence available up to t.
+  * Everything is seeded: the same --seed yields byte-identical data.
 
 Author: Justin Wolcott (fallnai-research.org)
 License: Apache-2.0
@@ -26,10 +34,13 @@ from clerk.common import (
     apply_ops,
     empty_ledger,
     enforce_budget,
+    parse_ledger,
     serialize_ledger,
 )
 
 # ---------------------------------------------------------------- predicates
+# p: ledger key | h: human phrase | v: value pool
+# add/upd/tmb: statement templates. {v} = new value. {v_old} = previous value.
 
 PREDICATES: List[Dict[str, Any]] = [
     {"p": "job", "h": "job", "v": ["barista", "chef", "nurse", "high school teacher", "librarian", "electrician", "accountant", "software engineer", "pharmacist", "bus driver"],
@@ -97,10 +108,14 @@ Q_TEMPLATES = [
     "Remind me — what's my current {h}?",
     "What about my {h}? Do you have that on record?",
 ]
+Q_TEMPLATES_TEMPORAL = [
+    "What was my {h} back then, before it changed?",
+    "What did my {h} used to be?",
+]
 
 ANSWER_NEGATED = "Not anymore — you mentioned that's no longer the case."
 ANSWER_UNKNOWN = "I don't know — you haven't told me about that yet."
-ANSWER_TEMPORAL = "Back then it was {v}, though it's changed since."
+ANSWER_TEMPORAL = "Back then it was {v}, though it has changed since."
 
 
 def spec_of(pred: str) -> Dict[str, Any]:
@@ -130,27 +145,27 @@ def make_persona(rng: random.Random, name: str, n_init: int = 8) -> Dict[str, An
 def render_session(
     rng: random.Random,
     persona: Dict[str, Any],
-    session_idx: int,
+    ledger_before: List[Slot],
     n_add: int, n_upd: int, n_tmb: int, n_q: int,
-) -> Tuple[List[Dict[str, str]], List[Dict[str, Any]], List[Dict[str, Any]], List[Dict[str, str]]]:
-    """Render one session of the evolving dialogue.
-
-    Returns (turns, write_events, qas, chatter_turns_before_qa).
-    write_events: [{"kind": "ADD"|"UPDATE"|"TOMBSTONE", "pred":..., "value":...}]
-    qas:          [{"q":..., "a":..., "type":..., "pred":..., "value":...}]
+) -> Tuple[List[Dict[str, str]], List[Dict[str, Any]], List[Dict[str, Any]]]:
+    """Render one session. Returns (turns, write_events, qas).
+
+    Question targets are constrained by the ledger the assistant will read
+    from (ledger_before, i.e. end of the previous session):
+      predicate absent from persona facts ....... unknown
+      valid slot in ledger ....................... stale / superseded
+      tombstoned slot in ledger .................. negated / temporal
+      active in persona but EVICTED from ledger .. never asked
     """
-    name = persona["name"]
     turns: List[Dict[str, str]] = []
-    events: List[Dict[str, Any]] = []
+    qas: List[Dict[str, Any]] = []
 
     def say(user: str, assistant: Optional[str] = None) -> None:
         turns.append({"role": "user", "content": user})
-        if assistant is None:
-            turns.append({"role": "assistant", "content": rng.choice(ACKS)})
-        else:
-            turns.append({"role": "assistant", "content": assistant})
+        turns.append({"role": "assistant",
+                      "content": rng.choice(ACKS) if assistant is None else assistant})
 
-    # ---- write events, in randomized order
+    # ---- events (chosen against persona state at session start)
     events: List[Dict[str, Any]] = []
 
     for _ in range(n_add):
@@ -164,7 +179,7 @@ def render_session(
 
     for _ in range(n_upd):
         updatable = [f for f in persona["facts"]
-                     if f["state"] == "active" and f["born_session"] < session_idx]
+                     if f["state"] == "active" and f["born_session"] < _session_being_rendered]
         if not updatable:
             break
         f = rng.choice(updatable)
@@ -184,178 +199,184 @@ def render_session(
 
     rng.shuffle(events)
 
-    # ---- interleave chatter
-    n_chatter = rng.randint(1, 3)
-
-    # ---- questions (asked BEFORE consolidation, answered from ledger@t-1)
-    qas: List[Dict[str, Any]] = []
-    q_specs: List[str] = rng.sample(PREDICATES, min(n_q + 2, len(PREDICATES)))
-    for spec in q_specs[:n_q]:
+    # ---- questions, asked before consolidation
+    led_state: Dict[str, Tuple[bool, bool]] = {}  # pred -> (in_persona, in_ledger_valid)
+    for s in ledger_before:
+        if s.occupied:
+            led_state[s.predicate] = (True, s.valid)
+    for f in persona["facts"]:
+        if f["pred"] not in led_state:
+            # active but never consolidated (session 0) or evicted
+            led_state[f["pred"]] = (True, f["state"] != "tombstoned")
+
+    askable = [p for p in PREDICATES
+               if p["p"] not in led_state            # unknown questions
+               or led_state[p["p"]][1] is True       # stale/superseded
+               or not led_state[p["p"]][1]]          # negated/temporal
+    rng.shuffle(askable)
+    for spec in askable[:n_q]:
         f = find_fact(persona, spec["p"])
         q = rng.choice(Q_TEMPLATES).replace("{h}", spec["h"])
         if f is None:
             qas.append({"q": q, "a": ANSWER_UNKNOWN, "type": "unknown",
                         "pred": spec["p"], "value": None})
         elif f["state"] == "tombstoned":
-            # temporal question if old value is known, else negated
             old = f["history"][-1]["value"] if f["history"] else None
             if old and rng.random() < 0.5:
-                q = "What was my " + spec["h"] + " back then, before it changed?"
-                qas.append({"q": q, "a": ANSWER_TEMPORAL.replace("{v}", old),
+                q2 = rng.choice(Q_TEMPLATES_TEMPORAL).replace("{h}", spec["h"])
+                qas.append({"q": q2, "a": ANSWER_TEMPORAL.replace("{v}", old),
                             "type": "temporal", "pred": spec["p"], "value": old})
             else:
                 qas.append({"q": q, "a": ANSWER_NEGATED, "type": "negated",
                             "pred": spec["p"], "value": None})
-        elif f["state"] == "updated":
-            qas.append({"q": q, "a": f["value"], "type": "superseded",
-                        "pred": spec["p"], "value": f["value"]})
         else:
-            qas.append({"q": q, "a": f["value"], "type": "stale",
+            qas.append({"q": q, "a": f["value"], "type": "superseded",
                         "pred": spec["p"], "value": f["value"]})
 
-    # ---- render: statements + chatter, then Q&A block at the end
-    statement_turns: List[Dict[str, str]] = []
+    # ---- render turns: statements (+ chatter), then the Q&A block
+    n_chatter = rng.randint(1, 3)
     for ev in events:
-        statement_turns.append({"role": "user", "content": ev["text"]})
-        statement_turns.append({"role": "assistant", "content": rng.choice(ACKS)})
+        say(ev["text"])
         if rng.random() < 0.35 and n_chatter > 0:
-            statement_turns.append({"role": "user", "content": rng.choice(CHATTER)})
-            statement_turns.append({"role": "assistant", "content": rng.choice(ACKS)})
+            say(rng.choice(CHATTER))
             n_chatter -= 1
-
-    qa_turns: List[Dict[str, str]] = []
     for qa in qas:
-        qa_turns.append({"role": "user", "content": qa["q"]})
-        qa_turns.append({"role": "assistant", "content": qa["a"]})
+        say(qa["q"], qa["a"])
 
-    turns = statement_turns + qa_turns
-    return turns, events, qas, statement_turns
+    return turns, events, qas
+
+
+_session_being_rendered: int = 0
 
 
 def apply_event_to_persona(persona: Dict[str, Any], ev: Dict[str, Any],
                            session_idx: int) -> None:
-    f = find_fact(persona, ev["pred"])
     if ev["kind"] == "ADD":
         persona["facts"].append({"pred": ev["pred"], "value": ev["value"],
                                  "born_session": session_idx, "state": "active",
                                  "history": []})
     elif ev["kind"] == "UPDATE":
+        f = find_fact(persona, ev["pred"])
         f["history"].append({"value": f["value"], "until_session": session_idx})
         f["value"] = ev["value"]
         f["state"] = "updated"
     elif ev["kind"] == "TOMBSTONE":
+        f = find_fact(persona, ev["pred"])
         f["state"] = "tombstoned"
 
 
-# ---------------------------------------------------------------- timeline
+# ------------------------------------------------------- gold op programs
+
+def _slot_id_for(ledger: List[Slot], pending: List[Dict[str, Any]],
+                 subject: str, pred: str) -> Optional[int]:
+    """Slot id holding (subject, pred) after `ledger` + `pending` ops."""
+    sim = [Slot(**vars(s)) for s in ledger]
+    sim, _ = apply_ops(sim, pending, -1)
+    for s in sim:
+        if s.occupied and s.subject == subject and s.predicate == pred:
+            return s.id
+    return None
+
 
 def gold_ops_for_events(
-    ledger: List[Slot], events: List[Dict[str, Any]],
-    session_idx: int, budget: int, persona: Dict[str, Any],
-) -> List[Dict[str, Any]]:
-    """Translate the session's write events into a gold edit program, then
-    apply budget pressure with the salience oracle. Slot ids are resolved by
-    looking at the ledger the reducer will actually see."""
+    ledger: List[Slot],
+    persona_name: str,
+    events: List[Dict[str, Any]],
+    session_idx: int,
+) -> Tuple[List[Dict[str, Any]], List[Slot]]:
+    """Build the gold edit program for one session's events, in transcript
+    order, applying budget pressure with the salience oracle afterwards.
+
+    Eviction-safe: if a predicate's slot was already evicted, an UPDATE
+    degrades to an ADD (the fact must be re-remembered) and a TOMBSTONE of
+    an absent slot is skipped."""
     ops: List[Dict[str, Any]] = []
-    ledger, _ = apply_ops(ledger, [], session_idx)  # no-op copy
+    ledger_now = [Slot(**vars(s)) for s in ledger]
     for ev in events:
         if ev["kind"] == "ADD":
-            ops.append({"op": "ADD", "s": persona["name"],
-                        "p": ev["pred"], "o": ev["value"]})
+            op = {"op": "ADD", "s": persona_name, "p": ev["pred"], "o": ev["value"]}
         elif ev["kind"] == "UPDATE":
-            slot_id = _slot_id_for(ledger, persona["name"], ev["pred"],
-                                   ev["value"], ops)
-            ops.append({"op": "UPDATE", "i": slot_id, "s": persona["name"],
-                        "p": ev["pred"], "o": ev["value"]})
-        elif ev["kind"] == "TOMBSTONE":
-            slot_id = _slot_id_for(ledger, persona["name"], ev["pred"], None, ops)
-            ops.append({"op": "TOMBSTONE", "i": slot_id})
-        ledger, _ = apply_ops(ledger, [ops[-1]], session_idx)
-    ledger, evict_ops = enforce_budget(ledger, [], session_idx)
-    ops.extend(evict_ops)
-    return ops
-
-
-def _slot_id_for(ledger: List[Slot], subject: str, pred: str,
-                 new_value: Optional[str], pending_ops: List[Dict[str, Any]]) -> int:
-    """Find the slot id that (after pending ops so far) holds (subject, pred)."""
-    sim = [Slot(**vars(s)) for s in ledger]
-    sim, _ = apply_ops(sim, pending_ops, -1)
-    for s in sim:
-        if s.occupied and s.subject == subject and s.predicate == pred:
-            return s.id
-    raise KeyError(f"no slot for ({subject}, {pred})")
+            sid = _slot_id_for(ledger_now, ops, persona_name, ev["pred"])
+            if sid is None:
+                op = {"op": "ADD", "s": persona_name, "p": ev["pred"], "o": ev["value"]}
+            else:
+                op = {"op": "UPDATE", "i": sid, "s": persona_name,
+                      "p": ev["pred"], "o": ev["value"]}
+        else:  # TOMBSTONE
+            sid = _slot_id_for(ledger_now, ops, persona_name, ev["pred"])
+            if sid is None:
+                continue
+            op = {"op": "TOMBSTONE", "i": sid}
+        ops.append(op)
+        ledger_now, _ = apply_ops(ledger_now, [op], session_idx)
+
+    # budget pressure: evict lowest-salience occupied slots until it fits
+    occupied = [s for s in ledger_now if s.occupied]
+    for _ in range(max(0, len(occupied) - len(ledger_now))):
+        from clerk.common import salience
+        victim = min((s for s in ledger_now if s.occupied),
+                     key=lambda s: salience(s, session_idx))
+        ops.append({"op": "EVICT", "i": victim.id})
+        ledger_now[victim.id] = Slot(id=victim.id)
+    return ops, ledger_now
 
 
+# ---------------------------------------------------------------- timeline
+
 def generate_persona_timeline(
     rng: random.Random, name: str, n_sessions: int, budget: int,
 ) -> Dict[str, Any]:
     persona = make_persona(rng, name)
     ledger = empty_ledger(budget)
-    sessions = []
-
-    # session 0 introduces the initial facts as ADD events
-    intro_events = [{"kind": "ADD", "pred": f["pred"], "value": f["value"],
-                     "text": spec_of(f["pred"])["add"].replace("{v}", f["value"])}
-                    for f in persona["facts"]]
+    sessions: List[Dict[str, Any]] = []
 
     for s_idx in range(n_sessions):
-        if s_idx == 0:
-            events, n_add, n_upd, n_tmb = intro_events, 0, 0, 0
-        else:
-            n_add = rng.randint(1, 2)
-            n_upd = rng.randint(0, 2)
-            n_tmb = rng.randint(0, 1)
-            turns, events, qas, _ = render_session(
-                rng, persona, s_idx, n_add, n_upd, n_tmb, rng.randint(1, 3))
-            for ev in events:
-                apply_event_to_persona(persona, ev, s_idx)
+        global _session_being_rendered
+        _session_being_rendered = s_idx
 
-        # gold consolidation program + resulting ledger
         if s_idx == 0:
-            turns = []
-            for ev in intro_events:
+            # intro session: persona states the initial facts
+            events = [{"kind": "ADD", "pred": f["pred"], "value": f["value"],
+                       "text": spec_of(f["pred"])["add"].replace("{v}", f["value"])}
+                      for f in persona["facts"]]
+            rng.shuffle(events)
+            turns: List[Dict[str, str]] = []
+            for ev in events:
                 turns.append({"role": "user", "content": ev["text"]})
                 turns.append({"role": "assistant", "content": rng.choice(ACKS)})
-        ops = gold_ops_for_events(ledger, events, s_idx, budget, persona)
-        ledger, _ = apply_ops(ledger, ops, s_idx)
-        ledger, _ = enforce_budget(ledger, [], s_idx)
+            qas: List[Dict[str, Any]] = []
+        else:
+            turns, events, qas = render_session(
+                rng, persona, ledger,
+                rng.randint(1, 2), rng.randint(0, 2), rng.randint(0, 1),
+                rng.randint(1, 3))
+            for ev in events:
+                apply_event_to_persona(persona, ev, s_idx)
+            for qa in qas:
+                for sl in ledger:
+                    if sl.occupied and sl.predicate == qa["pred"]:
+                        sl.n_uses += 1
 
-        # increment n_uses on slots that were queried this session
-        for qa in (sessions[-1]["qas"] if s_idx > 0 and sessions else []):
-            pass
+        ledger_before = serialize_ledger(ledger)
+        ops, ledger_after = gold_ops_for_events(ledger, name, events, s_idx)
+        ledger = ledger_after
 
         sessions.append({
             "session_idx": s_idx,
             "turns": turns,
             "events": events,
-            "qas": qas if s_idx > 0 else [],
+            "qas": qas,
             "gold_ops": ops,
-            "ledger_before": serialize_ledger(ledger) if False else None,
+            "ledger_before": ledger_before,
+            "ledger_after": serialize_ledger(ledger),
         })
 
-    # second pass to store ledger states cleanly
-    ledger = empty_ledger(budget)
-    for sess in sessions:
-        sess["ledger_before"] = serialize_ledger(ledger)
-        ledger, _ = apply_ops(ledger, sess["gold_ops"], sess["session_idx"])
-        ledger, _ = enforce_budget(ledger, [], sess["session_idx"])
-        sess["ledger_after"] = serialize_ledger(ledger)
-
-    # count question hits for salience bookkeeping (documented oracle signal)
-    for sess in sessions:
-        for qa in sess["qas"]:
-            pred = qa["pred"]
-            for sl in ledger:
-                if sl.occupied and sl.predicate == pred:
-                    sl.n_uses += 1
-
     return {"name": name, "sessions": sessions,
             "final_ledger": serialize_ledger(ledger)}
 
 
 def main() -> None:
-    ap = argparse.ArgumentParser(description=__doc__)
+    ap = argparse.ArgumentParser(description="CLERK synthetic session generator")
     ap.add_argument("--personas", type=int, default=100)
     ap.add_argument("--sessions", type=int, default=8)
     ap.add_argument("--budget", type=int, default=LEDGER_BUDGET_DEFAULT)
@@ -366,15 +387,15 @@ def main() -> None:
 
     rng = random.Random(args.seed)
     names = TRAIN_NAMES if args.split == "train" else TEST_NAMES
-    written = 0
+    n = 0
     with open(args.out, "w") as f:
         for i in range(args.personas):
-            name = names[i % len(names)] + ("" if i < len(names)
-                                            else f" {i // len(names)}")
+            suffix = "" if i < len(names) else f" {i // len(names)}"
+            name = names[i % len(names)] + suffix
             tl = generate_persona_timeline(rng, name, args.sessions, args.budget)
             f.write(json.dumps(tl, ensure_ascii=False) + "\n")
-            written += 1
-    print(f"wrote {written} personas to {args.out}")
+            n += 1
+    print(f"wrote {n} personas ({args.split}) to {args.out}")
 
 
 if __name__ == "__main__":