PNS-Bind-25M: implementation, configs, eval, results, reproduction
f930dac verified about 2 months ago
48.1 kB
"""PNS-World lifetime generator.
Plan storylines -> walk slots in order, maintaining the ground-truth solver and
the exact reducer side by side -> emit typed events with certified gold labels.
Anti-shortcut measures built in (preregistered in preregistration/PREREGISTRATION_PNS.md):
  * revert-corrections restore an *arbitrary earlier* assertion, referenced only
    in text (by source or clock), so "newest record wins" and any fixed
    age-rank rule fail;
  * gold reachability is certified against the same reducer the models replay
    (a question whose gold record was evicted degrades to filler and is counted);
  * delay and competing-record count vary independently; question targets are
    sampled uniformly from live storylines;
  * yes/no, first/second and enum answers are balance-tracked per lifetime;
  * time is only ever visible as per-event dt; deadlines require integration.
"""
from __future__ import annotations

import hashlib
import os
import random
from dataclasses import dataclass, field

from . import questions as Q
from .reducer import ExactReducer
from .schema import (CITIES, DEVICE_NAMES, ENUM_ATTRS, ENUM_ID, EXACT_2HOP_PATHS, EXACT_KEYS,
                     GOAL_KINDS, K_DEVICE, K_PERSON, K_PROJECT, K_SHIPMENT, K_VAULT, KIND_NAMES,
                     OP_SPECS, OPS, PERSON_NAMES, PRIO, PROJECT_NAMES, REL_ATTRS, SEM_2HOP_PATHS,
                     SHIPMENT_NAMES, SOURCES, VAULT_NAMES, ET, Entity, Event, Fam, Mode)
from .verbalize import render

# Gold-label resolver mode (audit/AUDIT.md 7):
#   "v1"  - last-appended valid assertion (the original, defective corpus)
#   "sem" - chronological golds for SEM_LATEST/SEM_2HOP ONLY. Those paths
#           consume no rng and steer no control flow, so the EVENT STREAM is
#           byte-identical to v1 - this is what makes it possible to
#           re-evaluate existing checkpoints on corrected semantic labels.
#   "v2"  - chronological everywhere (exact + temporal too). Control flow
#           depends on the resolver there, so streams differ from v1.
GOLD_MODE = os.environ.get("PNS_GOLD", "v2")

# ------------------------------------------------------------ value factories

def make_code(rng: random.Random) -> str:
    return (f"{rng.choice('BCDFGHJKLMNPQRSTVWXZ')}{rng.choice('ACEFHKLMNPRTUWXY')}-"
            f"{rng.randint(1000, 9999)}-{rng.choice('BCDFGHJKLMNPQRSTVWXZ')}{rng.choice('AEIOUY')}")


def make_fw(rng: random.Random) -> str:
    return f"fw-{rng.randint(2, 9)}.{rng.randint(0, 9)}.{rng.randint(0, 99)}{rng.choice('abcdex')}"


def make_number(rng: random.Random, key: str) -> str:
    if key == "reading":
        return f"{rng.randint(80, 9999) / 10:.1f}"
    if key == "quantity":
        return f"{rng.randint(40, 98000)} units"
    return f"{rng.randint(12, 990)} k-credits"  # budget


def make_handle(rng: random.Random) -> str:
    return f"TRK-{rng.randint(10000, 99999)}"


def make_ref(rng: random.Random) -> str:
    return f"SR-{rng.randint(1000, 9999)}"


def clockstr(clock: int) -> str:
    return f"day {clock // 1440 + 1} {(clock // 60) % 24:02d}:{clock % 60:02d}"


# ------------------------------------------------------------------- solver

@dataclass
class Assertion:
    value: str            # enum surface / exact bytes / str(eid) for relations
    slot: int
    source: str
    valid: bool = True
    is_corr: bool = False
    rid: int = -1         # exact records only, filled at walk
    clock: int = -1       # filled at walk
    ev_idx: int = -1


@dataclass
class Chain:
    ent: int
    attr: str             # attr or exact key or relation name
    exact: bool
    rel: bool = False
    asserts: list[Assertion] = field(default_factory=list)
    planned_corr: dict = field(default_factory=dict)   # slot -> corr spec

    def current(self) -> Assertion | None:
        """Last-APPENDED valid assertion (v1 semantics).
        Kept because it drives rng-consuming value sampling; changing it would
        change the event stream itself and make it impossible to re-evaluate
        existing checkpoints on corrected labels. Gold labels use
        `current_chrono()` instead - see audit/AUDIT.md 7."""
        for a in reversed(self.asserts):
            if a.valid and a.slot >= 0 and a.ev_idx >= 0:
                return a
        return None

    def current_chrono(self) -> Assertion | None:
        """The chronologically latest valid assertion - the value a correct
        tracker should report. `asserts` is not in chronological order:
        planned assertions are appended during plan(), walk-time
        distractor/keep-warm observations afterwards."""
        live = [a for a in self.asserts if a.valid and a.slot >= 0 and a.ev_idx >= 0]
        return max(live, key=lambda a: a.ev_idx) if live else None

    def emitted(self) -> list[Assertion]:
        return [a for a in self.asserts if a.ev_idx >= 0]


class World:
    def __init__(self, rng: random.Random, n_extra: int = 8):
        self.entities: list[Entity] = []
        self.by_kind: dict[int, list[int]] = {k: [] for k in KIND_NAMES}
        pools = {K_PERSON: rng.sample(PERSON_NAMES, 7), K_PROJECT: rng.sample(PROJECT_NAMES, 5),
                 K_SHIPMENT: rng.sample(SHIPMENT_NAMES, 5), K_VAULT: rng.sample(VAULT_NAMES, 4),
                 K_DEVICE: rng.sample(DEVICE_NAMES, 4)}
        for kind, names in pools.items():
            for nm in names:
                eid = len(self.entities)
                self.entities.append(Entity(eid, kind, nm))
                self.by_kind[kind].append(eid)


# ------------------------------------------------------------------ tickets

@dataclass
class Ticket:
    kind: str             # obs|doc|corr|rel|time|goal_set|goal_upd|op|commit|result|
    #                       deadline|question|imm|noop
    data: dict = field(default_factory=dict)


class LifetimeGen:
    """One lifetime.  Deterministic given (seed, length)."""

    def __init__(self, seed: int, length: int = 256):
        self.seed, self.L = seed, length
        self.rng = random.Random(seed * 1_000_003 + 17)       # structure
        self.trng = random.Random(seed * 7_777_777 + 3)       # text surface
        self.world = World(self.rng)
        self.slots: list[Ticket | None] = [None] * length
        self.chains: dict[tuple[int, str], Chain] = {}
        self.seen_values: set[str] = set()
        self.dropped: dict[str, int] = {}
        # random initial parity so per-lifetime odd counts do not bias the corpus
        self.balance = {k: self.rng.choice([-1, 1]) for k in ("yn", "fs", "imm", "tmp")}
        self._dl_parity = self.rng.random() < 0.5
        self.scale = length / 256.0

    # ------------------------------------------------------------- utilities
    def uniq(self, fn, *a):
        for _ in range(64):
            v = fn(self.rng, *a) if a else fn(self.rng)
            if v not in self.seen_values:
                self.seen_values.add(v)
                return v
        raise RuntimeError("value space exhausted")

    def chain(self, ent: int, attr: str, exact: bool, rel: bool = False) -> Chain:
        key = (ent, attr)
        if key not in self.chains:
            self.chains[key] = Chain(ent, attr, exact, rel)
        return self.chains[key]

    def free_slot(self, lo: int, hi: int) -> int | None:
        lo, hi = max(0, lo), min(self.L - 1, hi)
        cands = [i for i in range(lo, hi + 1) if self.slots[i] is None]
        return self.rng.choice(cands) if cands else None

    def place(self, lo: int, hi: int, ticket: Ticket) -> int | None:
        s = self.free_slot(lo, hi)
        if s is not None:
            self.slots[s] = ticket
        return s

    def sample_delay(self, hi: int) -> int:
        """log-uniform delay in [2, hi]."""
        import math
        hi = max(2, hi)
        return min(hi, max(2, int(math.exp(self.rng.uniform(math.log(2), math.log(hi))))))

    def mixed_delay(self, hi: int) -> int:
        """50/50 log-uniform / uniform: decorrelates candidate age from delay."""
        if self.rng.random() < 0.5:
            return self.sample_delay(hi)
        return self.rng.randint(2, max(2, hi))

    def enum_value(self, ent: int, attr: str, prev: str | None) -> str:
        pool = ENUM_ATTRS[(self.world.entities[ent].kind, attr)]
        v = self.rng.choice(pool)
        while v == prev:
            v = self.rng.choice(pool)
        return v

    def exact_value(self, key: str) -> str:
        if key == "access code":
            return self.uniq(make_code)
        if key == "firmware tag":
            return self.uniq(make_fw)
        return self.uniq(make_number, key)

    # ------------------------------------------------------------------ plan
    def plan(self) -> None:
        rng, L = self.rng, self.L
        sc = self.scale
        qplan: list[tuple[int, Ticket]] = []

        def quota(n):
            return max(1, round(n * sc))

        # ---- relations asserted early (some re-assigned later)
        rel_chains = {}
        for (kind, attr), dst_kind in REL_ATTRS.items():
            for ent in self.world.by_kind[kind]:
                ch = self.chain(ent, attr, exact=False, rel=True)
                rel_chains[(ent, attr)] = ch
                dst = rng.choice(self.world.by_kind[dst_kind])
                s = self.place(0, int(L * 0.12), Ticket("rel", dict(ent=ent, attr=attr, dst=dst)))
                if s is not None:
                    ch.asserts.append(Assertion(str(dst), s, ""))
                if rng.random() < 0.35:  # re-assignment mid-life
                    dst2 = rng.choice(self.world.by_kind[dst_kind])
                    s2 = self.place(int(L * 0.3), int(L * 0.8),
                                    Ticket("rel", dict(ent=ent, attr=attr, dst=dst2)))
                    if s2 is not None:
                        ch.asserts.append(Assertion(str(dst2), s2, ""))

        # ---- exact chains (documents + corrections)
        exact_targets = []
        for (kind, key) in EXACT_KEYS:
            for ent in self.world.by_kind[kind]:
                exact_targets.append((ent, key))
        rng.shuffle(exact_targets)
        n_exact_chains = quota(9)
        for ent, key in exact_targets[:n_exact_chains]:
            ch = self.chain(ent, key, exact=True)
            n_ass = rng.randint(2, 5)
            srcs = rng.sample(SOURCES, min(n_ass + 1, len(SOURCES)))
            lo = rng.randint(0, int(L * 0.3))
            hi = rng.randint(int(L * 0.55), L - 20)
            sl = sorted(rng.sample(range(lo, hi), min(n_ass, hi - lo)))
            for i, s in enumerate(sl):
                if self.slots[s] is None:
                    self.slots[s] = Ticket("doc", dict(ent=ent, key=key))
                    ch.asserts.append(Assertion("", s, srcs[i % len(srcs)]))
            prev_cs = None
            for corr_round in range(2):
                p_corr = 0.8 if corr_round == 0 else 0.35
                if len(ch.asserts) < 2 or rng.random() >= p_corr:
                    break
                last = max(ch.asserts[-1].slot, prev_cs or 0)
                cs = self.free_slot(last + 2, min(last + 40, L - 10))
                if cs is None:
                    break
                if rng.random() < 0.7:        # revert to arbitrary earlier assertion
                    tgt = rng.randrange(0, len(ch.asserts) - 1)
                    spec = dict(mode="revert", target=tgt,
                                ref_kind=rng.choice(["source", "clock"]))
                else:                          # replace with fresh value
                    spec = dict(mode="replace")
                self.slots[cs] = Ticket("corr", dict(ent=ent, key=key, spec=spec))
                ch.planned_corr[cs] = spec
                prev_cs = cs

        # ---- enum drift chains
        enum_targets = [(ent, attr) for (kind, attr) in ENUM_ATTRS
                        for ent in self.world.by_kind[kind]]
        rng.shuffle(enum_targets)
        n_enum_chains = quota(10)
        for ent, attr in enum_targets[:n_enum_chains]:
            ch = self.chain(ent, attr, exact=False)
            n_ass = rng.randint(2, 6)
            srcs = rng.sample(SOURCES, min(n_ass, len(SOURCES)))
            lo = rng.randint(0, int(L * 0.25))
            hi = rng.randint(int(L * 0.6), L - 15)
            sl = sorted(rng.sample(range(lo, hi), min(n_ass, hi - lo)))
            for i, s in enumerate(sl):
                if self.slots[s] is None:
                    self.slots[s] = Ticket("obs", dict(ent=ent, attr=attr))
                    ch.asserts.append(Assertion("", s, srcs[i % len(srcs)]))
            if len(ch.asserts) >= 3 and rng.random() < 0.5:
                last = ch.asserts[-1].slot
                cs = self.free_slot(last + 2, min(last + 30, L - 8))
                if cs is not None:
                    tgt = rng.randrange(0, len(ch.asserts) - 1)
                    spec = dict(mode="revert", target=tgt,
                                ref_kind=rng.choice(["source", "clock"]))
                    self.slots[cs] = Ticket("corr", dict(ent=ent, attr=attr, spec=spec,
                                                         enum=True))
                    ch.planned_corr[cs] = spec

        # ---- deadlines (distinct entities: a re-set deadline would orphan the
        # balance construction of the earlier question)
        dl_cands = self.world.by_kind[K_PROJECT] + self.world.by_kind[K_SHIPMENT]
        for ent in rng.sample(dl_cands, min(quota(3), len(dl_cands))):
            tkt = Ticket("deadline", dict(ent=ent))
            s = self.place(5, L - 30, tkt)
            if s is not None:
                qs = self.free_slot(s + 8, min(s + 8 + self.sample_delay(150), L - 2))
                if qs is not None:
                    self.slots[qs] = Ticket("question", dict(fam=Fam.DEADLINE, ent=ent,
                                                             dl_slot=s))
                    tkt.data["q_slot"] = qs

        # ---- goals
        goals = []
        for gid in range(quota(3)):
            kind = rng.choice(GOAL_KINDS)
            ent = rng.randrange(len(self.world.entities))
            s = self.place(int(L * 0.02), int(L * 0.7), Ticket("goal_set",
                           dict(gid=gid, kind=kind, ent=ent)))
            if s is not None:
                goals.append((gid, kind, ent, s))
        for gid, kind, ent, s in goals:
            if rng.random() < 0.6:
                self.place(s + 3, L - 10, Ticket("goal_upd",
                           dict(gid=gid, kind=kind, ent=ent,
                                drop=rng.random() < 0.45)))
        for _ in range(quota(4)):
            if not goals:
                break
            base = max(g[3] for g in goals)
            qs = self.free_slot(base + 2, min(base + 2 + self.sample_delay(180), L - 2))
            if qs is not None:
                self.slots[qs] = Ticket("question", dict(fam=Fam.GOAL_TOP))

        # ---- ops (commit next slot, delayed result, self/handle questions)
        op_seq = 0
        for _ in range(quota(7)):
            op = rng.choice(OPS)
            kind, keys = OP_SPECS[op]
            cands = [e for e in self.world.by_kind[kind]
                     if any(self.chains.get((e, k)) and
                            any(a.slot >= 0 for a in self.chains[(e, k)].asserts)
                            for k in keys)]
            if not cands:
                continue
            ent = rng.choice(cands)
            key = keys[0]
            ch = self.chain(ent, key, exact=True)
            first = min(a.slot for a in ch.asserts if a.slot >= 0)
            s = self.free_slot(first + 2, L - 50)
            if s is None or self.slots[s] is not None:
                continue
            self.slots[s] = Ticket("op", dict(op=op, ent=ent, keys=[key], seq=op_seq))
            cslot = s + 1 if s + 1 < L and self.slots[s + 1] is None else self.free_slot(s + 1, s + 3)
            if cslot is None:
                self.slots[s] = None
                continue
            self.slots[cslot] = Ticket("commit", dict(seq=op_seq))
            rs = self.free_slot(cslot + 3, min(cslot + 3 + rng.randint(3, 40), L - 6))
            if rs is not None:
                self.slots[rs] = Ticket("result", dict(seq=op_seq))
            qs = self.free_slot(cslot + 2, min(cslot + 2 + self.mixed_delay(180), L - 2))
            if qs is not None and rng.random() < 0.85:
                self.slots[qs] = Ticket("question", dict(fam=Fam.SELF_REF, seq=op_seq,
                                                         ent=ent, key=key, op=op))
            if rs is not None:
                hs = self.free_slot(rs + 2, min(rs + 2 + self.mixed_delay(160), L - 2))
                if hs is not None and rng.random() < 0.8:
                    self.slots[hs] = Ticket("question", dict(fam=Fam.HANDLE_REF, seq=op_seq,
                                                             ent=ent, op=op))
            op_seq += 1

        # ---- chain questions (anchor an assertion, query inside its validity gap)
        def chain_questions(fam, chains, per, exact):
            placed = 0
            attempts = 0
            chs = [c for c in chains if c.asserts]
            while placed < per and attempts < per * 8 and chs:
                attempts += 1
                ch = rng.choice(chs)
                emitted = sorted([a for a in ch.asserts if a.slot >= 0], key=lambda a: a.slot)
                if not emitted or (exact and len(emitted) < 2):
                    continue
                idx = rng.randrange(len(emitted))
                a = emitted[idx]
                nxt = emitted[idx + 1].slot if idx + 1 < len(emitted) else self.L - 2
                corr_slots = sorted(ch.planned_corr)
                nxt_c = min([c for c in corr_slots if c > a.slot], default=self.L - 2)
                hi = min(nxt, nxt_c) - 1
                if hi - a.slot < 3:
                    continue
                qs = self.free_slot(a.slot + 2, min(a.slot + 2 + self.sample_delay(hi - a.slot), hi))
                if qs is None:
                    continue
                self.slots[qs] = Ticket("question", dict(fam=fam, ent=ch.ent, attr=ch.attr))
                placed += 1
            # post-correction questions: after a revert/replace the current value
            # is set by the correction; ask beyond it
            for ch in chs:
                for cs in ch.planned_corr:
                    if rng.random() < 0.75:
                        qs = self.free_slot(cs + 2, min(cs + 2 + self.sample_delay(120), self.L - 2))
                        if qs is not None:
                            self.slots[qs] = Ticket("question", dict(fam=fam, ent=ch.ent,
                                                                     attr=ch.attr))

        exact_chs = [c for c in self.chains.values() if c.exact]
        enum_chs = [c for c in self.chains.values() if not c.exact and not c.rel]
        chain_questions(Fam.EXACT_DELAYED, exact_chs, quota(5), True)
        chain_questions(Fam.SEM_LATEST, enum_chs, quota(5), False)

        # ---- 2hop questions
        for fam, paths, per in ((Fam.SEM_2HOP, SEM_2HOP_PATHS, quota(4)),
                                (Fam.EXACT_2HOP, EXACT_2HOP_PATHS, quota(3))):
            placed = 0
            for _ in range(per * 8):
                if placed >= per:
                    break
                skind, rel, dattr = rng.choice(paths)
                srcs_ = [e for e in self.world.by_kind[skind]
                         if (e, rel) in self.chains and self.chains[(e, rel)].asserts]
                if not srcs_:
                    continue
                src = rng.choice(srcs_)
                relch = self.chains[(src, rel)]
                rel_last = max(a.slot for a in relch.asserts if a.slot >= 0)
                dst = int(relch.asserts[-1].value)
                dch = self.chains.get((dst, dattr))
                if not dch or not any(a.slot >= 0 for a in dch.asserts):
                    continue
                d_last = max(a.slot for a in dch.asserts if a.slot >= 0)
                lo = max(rel_last, min(a.slot for a in dch.asserts if a.slot >= 0)) + 2
                qs = self.free_slot(lo, min(lo + self.sample_delay(160), self.L - 2))
                if qs is None:
                    continue
                self.slots[qs] = Ticket("question", dict(fam=fam, src=src, rel=rel,
                                                         attr=dattr))
                placed += 1

        # ---- temporal order
        for _ in range(quota(4)):
            good = [c for c in enum_chs
                    if c.asserts and not c.planned_corr and any(a.slot >= 0 for a in c.asserts)]
            if len(good) < 2:
                break
            ca, cb = rng.sample(good, 2)
            la = max(a.slot for a in ca.asserts if a.slot >= 0)
            lb = max(a.slot for a in cb.asserts if a.slot >= 0)
            if la == lb:
                continue
            qs = self.free_slot(max(la, lb) + 2,
                                min(max(la, lb) + 2 + self.sample_delay(140), self.L - 2))
            if qs is not None:
                self.slots[qs] = Ticket("question", dict(fam=Fam.TEMPORAL_ORDER,
                                                         a=(ca.ent, ca.attr), b=(cb.ent, cb.attr)))

        # ---- immediate comparisons
        for _ in range(quota(5)):
            s = self.free_slot(2, L - 2)
            if s is not None:
                self.slots[s] = Ticket("imm", dict(kind=rng.choice(["num", "code"])))

        # ---- keep-warm touches for pointer questions with long gaps
        for i, t in enumerate(self.slots):
            if t and t.kind == "question" and t.data.get("fam") in (
                    Fam.EXACT_DELAYED, Fam.EXACT_2HOP, Fam.SELF_REF, Fam.HANDLE_REF):
                ent = t.data.get("ent", t.data.get("src"))
                if ent is None:
                    continue
                for w in range(i - 30, 0, -35):
                    kw = self.free_slot(max(0, w - 6), w)
                    if kw is not None:
                        kind = self.world.entities[ent].kind
                        attrs = [a for (k, a) in ENUM_ATTRS if k == kind]
                        if attrs:
                            self.slots[kw] = Ticket("obs", dict(ent=ent, attr=rng.choice(attrs),
                                                                warm=True))

        # ---- fill the rest: distractor drift, time, noop.
        # Distractor documents form CHAINS with the same group-size distribution
        # as queried chains (Chinese-restaurant reuse) so cache group size does
        # not leak which records will be queried.
        queried_pairs = {(c.ent, c.attr) for c in self.chains.values() if c.exact}
        distract_pool = [p for p in exact_targets if p not in queried_pairs]
        rng.shuffle(distract_pool)
        distract_open: list[tuple[int, str]] = []
        for i in range(L):
            if self.slots[i] is not None:
                continue
            r = rng.random()
            if r < 0.30:
                self.slots[i] = Ticket("time", {})
            elif r < 0.42:
                self.slots[i] = Ticket("noop", {})
            elif r < 0.80:
                ent, attr = rng.choice(enum_targets)
                self.slots[i] = Ticket("obs", dict(ent=ent, attr=attr, warm=True))
            else:
                if distract_open and (rng.random() < 0.72 or not distract_pool):
                    ent, key = rng.choice(distract_open)
                elif distract_pool:
                    ent, key = distract_pool.pop()
                    distract_open.append((ent, key))
                else:
                    ent, key = rng.choice(exact_targets)
                self.slots[i] = Ticket("doc", dict(ent=ent, key=key, distract=True))

        # ---- pre-draw every event's dt so future clocks are exact at walk time
        self.dts = [rng.choice([15, 20, 30, 45, 60, 90]) if (self.slots[i] and
                    self.slots[i].kind == "time") else rng.randint(1, 6)
                    for i in range(L)]
        self.cumdt = [0] * (L + 1)
        for i in range(L):
            self.cumdt[i + 1] = self.cumdt[i] + self.dts[i]

    # ------------------------------------------------------------------ walk
    def walk(self):
        rng, trng = self.rng, self.trng
        red = ExactReducer()
        events: list[Event] = []
        ops: dict[int, dict] = {}
        goals: dict[int, dict] = {}          # gid -> {prio, rid, kind, ent}
        deadlines: dict[int, tuple[int, int]] = {}  # dl slot key: ent -> (clock, rid)
        digests: list[tuple[int, str]] = []

        def drop(fam, tag="gold_unreachable"):
            self.dropped[f"{fam}:{tag}"] = self.dropped.get(f"{fam}:{tag}", 0) + 1
            return Event(ET.NOOP, dt=rng.randint(1, 5))

        def key_text(ent, key):
            e = self.world.entities[ent]
            return f"{key} of {KIND_NAMES[e.kind]} {e.name}"

        def current_exact_rid(ent, key):
            ch = self.chains.get((ent, key))
            if not ch:
                return None
            a = ch.current_chrono() if GOLD_MODE == "v2" else ch.current()
            return a if a is not None and a.rid >= 0 else None

        def qmeta(ev, fam, dslot, gold_rid=None, ent=None, key=None):
            ev.meta["delay"] = red.ev_idx + 1 - dslot if dslot >= 0 else -1
            ev.meta["family"] = int(fam)
            if gold_rid is not None and ent is not None:
                same = [r for r in red.records.values()
                        if r.ent == ent and (key is None or r.key_text == key_text(ent, key))]
                same_sorted = sorted(same, key=lambda r: -r.birth_ev)
                ev.meta["kappa"] = len(same)
                ranks = [i for i, r in enumerate(same_sorted) if r.rid == gold_rid]
                ev.meta["gold_age_rank"] = ranks[0] + 1 if ranks else -1

        for slot in range(self.L):
            t = self.slots[slot]
            assert t is not None
            d = t.data
            dt = self.dts[slot]
            ev: Event | None = None

            if t.kind == "time":
                ev = Event(ET.TIME_ELAPSED, dt=dt)
            elif t.kind == "noop":
                ev = Event(ET.NOOP, dt=dt)
            elif t.kind == "rel":
                ch = self.chains[(d["ent"], d["attr"])]
                ev = Event(ET.OBSERVATION, dt=dt,
                           payload=dict(ent=d["ent"], ent2=d["dst"], attr=d["attr"], rel=True,
                                        source=rng.choice(SOURCES)))
                for a in ch.asserts:
                    if a.slot == slot:
                        a.ev_idx, a.clock = len(events), red.clock + ev.dt
            elif t.kind == "obs":
                ch = self.chain(d["ent"], d["attr"], exact=False)
                mine = [a for a in ch.asserts if a.slot == slot]
                prev = ch.current()
                val = self.enum_value(d["ent"], d["attr"], prev.value if prev else None)
                src = mine[0].source if mine and mine[0].source else rng.choice(SOURCES)
                ev = Event(ET.OBSERVATION, dt=dt,
                           payload=dict(ent=d["ent"], attr=d["attr"], value=val, source=src,
                                        ent2=-1))
                if mine:
                    a = mine[0]
                else:                       # distractor/keep-warm drift joins the chain
                    a = Assertion(val, slot, src)
                    ch.asserts.append(a)
                a.value, a.ev_idx, a.clock = val, len(events), red.clock + ev.dt
            elif t.kind == "doc":
                ch = self.chain(d["ent"], d["key"], exact=True)
                mine = [a for a in ch.asserts if a.slot == slot]
                val = self.exact_value(d["key"])
                src = mine[0].source if mine and mine[0].source else rng.choice(SOURCES)
                ev = Event(ET.DOCUMENT, dt=dt,
                           payload=dict(ent=d["ent"], key=d["key"], value=val, source=src,
                                        key_text=key_text(d["ent"], d["key"])))
                if mine:
                    a = mine[0]
                else:
                    a = Assertion(val, slot, src)
                    ch.asserts.append(a)
                a.value, a.ev_idx, a.clock = val, len(events), red.clock + ev.dt
            elif t.kind == "corr":
                key = d.get("key", d.get("attr"))
                ch = self.chains[(d["ent"], key)]
                spec = d["spec"]
                emitted = [a for a in ch.asserts if a.ev_idx >= 0 and not a.is_corr]
                if len(emitted) < 2:
                    ev = drop("corr", "chain_too_short")
                elif spec["mode"] == "replace":
                    exact = not d.get("enum", False)
                    val = (self.exact_value(key) if exact
                           else self.enum_value(d["ent"], key, ch.current().value if ch.current() else None))
                    ev = Event(ET.CORRECTION, dt=dt,
                               payload=dict(ent=d["ent"], key=key, mode="replace", value=val,
                                            exact=exact, source=rng.choice(SOURCES),
                                            key_text=key_text(d["ent"], key)))
                    a = Assertion(val, slot, "", is_corr=True)
                    a.ev_idx, a.clock = len(events), red.clock + ev.dt
                    ch.asserts.append(a)
                else:  # revert to target assertion
                    tgt = emitted[min(spec["target"], len(emitted) - 2)]
                    ref_kind = spec["ref_kind"]
                    srcs = [a.source for a in emitted]
                    if ref_kind == "source" and srcs.count(tgt.source) != 1:
                        ref_kind = "clock"
                    ev = Event(ET.CORRECTION, dt=dt,
                               payload=dict(ent=d["ent"], key=key, mode="revert",
                                            value=None, exact=False,
                                            source=rng.choice(SOURCES), ref_kind=ref_kind,
                                            ref_source=tgt.source, ref_clock=tgt.clock,
                                            key_text=key_text(d["ent"], key)))
                    for a in ch.asserts:
                        if a.ev_idx >= 0 and a.ev_idx > tgt.ev_idx:
                            a.valid = False
                    tgt.valid = True
            elif t.kind == "deadline":
                # place the deadline exactly on the wanted side of the question's
                # (pre-computed) clock; dt values are fixed at plan time
                q_slot = d.get("q_slot")
                clock_set = self.cumdt[slot + 1]
                if q_slot is None:
                    dl_clock = clock_set + rng.randint(60, 900)
                else:
                    clock_q = self.cumdt[q_slot]     # clock before the question event
                    gap = clock_q - clock_set
                    want_yes = self._dl_parity
                    self._dl_parity = not self._dl_parity
                    if want_yes and gap >= 15:
                        dl_clock = clock_set + max(4, int(rng.uniform(0.15, 0.7) * gap))
                    else:
                        dl_clock = clock_q + int(rng.uniform(0.25, 1.0) * max(gap, 60))
                val = clockstr(dl_clock)
                ev = Event(ET.DEADLINE_SET, dt=dt,
                           payload=dict(ent=d["ent"], value=val, value_str=val,
                                        key_text=key_text(d["ent"], "deadline")))
                deadlines[d["ent"]] = (dl_clock, -1)
            elif t.kind == "goal_set":
                prios = {g["prio"] for g in goals.values() if g.get("active")}
                free = [i for i in range(len(PRIO)) if i not in prios]
                if not free:
                    ev = drop("goal", "no_free_prio")
                else:
                    prio = rng.choice(free)
                    ev = Event(ET.GOAL_SET, dt=dt,
                               payload=dict(gid=d["gid"], kind=d["kind"], ent=d["ent"],
                                            prio=PRIO[prio],
                                            desc=f"{d['kind']} {self.world.entities[d['ent']].name}",
                                            key_text=f"goal: {d['kind']} "
                                                     f"{KIND_NAMES[self.world.entities[d['ent']].kind]} "
                                                     f"{self.world.entities[d['ent']].name}"))
                    goals[d["gid"]] = dict(prio=prio, active=True, kind=d["kind"], ent=d["ent"])
            elif t.kind == "goal_upd":
                g = goals.get(d["gid"])
                if not g or not g.get("active"):
                    ev = drop("goal", "inactive")
                elif d["drop"]:
                    ev = Event(ET.GOAL_UPDATE, dt=dt,
                               payload=dict(gid=d["gid"], kind=g["kind"], ent=g["ent"],
                                            drop=True, rid=g.get("rid", -1)))
                    g["active"] = False
                else:
                    prios = {og["prio"] for gid_, og in goals.items()
                             if og.get("active") and gid_ != d["gid"]}
                    free = [i for i in range(len(PRIO)) if i not in prios]
                    prio = rng.choice(free)
                    ev = Event(ET.GOAL_UPDATE, dt=dt,
                               payload=dict(gid=d["gid"], kind=g["kind"], ent=g["ent"],
                                            drop=False, prio=PRIO[prio], rid=g.get("rid", -1)))
                    g["prio"] = prio
            elif t.kind == "op":
                rid_args = []
                for k in d["keys"]:
                    a = current_exact_rid(d["ent"], k)
                    if a is None or a.rid not in red.records:
                        rid_args = None
                        break
                    rid_args.append((k, a))
                if rid_args is None:
                    ev = drop("op_emit")
                    ops[d["seq"]] = dict(dead=True)
                else:
                    ev = Event(ET.OP_REQUEST, dt=dt,
                               payload=dict(op=d["op"], ent=d["ent"], arg_keys=d["keys"]))
                    ev.mode_gold = Mode.EXTERNAL_OPERATION
                    ev.family = int(Fam.OP_EMIT)
                    ev.op_gold = OPS.index(d["op"])
                    ev.op_arg_rids = [a.rid for _, a in rid_args]
                    ev.gold_value = "|".join(a.value for _, a in rid_args)
                    ev.meta["family"] = int(Fam.OP_EMIT)
                    ev.meta["delay"] = red.ev_idx + 1 - min(a.ev_idx for _, a in rid_args)
                    ops[d["seq"]] = dict(op=d["op"], ent=d["ent"],
                                         args=[(k, a.value) for k, a in rid_args])
            elif t.kind == "commit":
                o = ops.get(d["seq"])
                if not o or o.get("dead"):
                    ev = drop("commit", "op_dead")
                else:
                    ref = self.uniq(make_ref)
                    # deliberately minimal descriptors: binding a committed value
                    # to its op/entity is narrative work, not a cache lookup
                    args = [(k, v, f"{k} you submitted") for k, v in o["args"]]
                    ev = Event(ET.ACTION_COMMITTED, dt=dt,
                               payload=dict(op=o["op"], ent=o["ent"], args=args, ref=ref,
                                            ref_text="ledger ref of your action"))
                    o["commit_ev"] = len(events)
            elif t.kind == "result":
                o = ops.get(d["seq"])
                if not o or o.get("dead") or "commit_ev" not in o:
                    ev = drop("result", "op_dead")
                else:
                    hd = self.uniq(make_handle)
                    ev = Event(ET.EXTERNAL_RESULT, dt=dt,
                               payload=dict(op=o["op"], ent=o["ent"], value=hd,
                                            outcome=rng.choice(["accepted", "confirmed", "queued"]),
                                            key_text="returned tracking handle"))
                    o["handle"] = hd
                    o["result_ev"] = len(events)
            elif t.kind == "imm":
                if d["kind"] == "num":
                    v1 = f"{rng.randint(100, 99999)}"
                    v2 = f"{rng.randint(100, 99999)}"
                    while v2 == v1:
                        v2 = f"{rng.randint(100, 99999)}"
                    gold = "first" if int(v1) > int(v2) else "second"
                else:
                    v1, v2 = make_code(rng), make_code(rng)
                    while v2 == v1:
                        v2 = make_code(rng)
                    gold = "first" if v1 < v2 else "second"
                want_first = (self.balance["imm"] < 0
                              or (self.balance["imm"] == 0 and rng.random() < 0.5))
                if (gold == "first") != want_first:
                    v1, v2 = v2, v1
                    gold = "first" if gold == "second" else "second"
                self.balance["imm"] += 1 if gold == "first" else -1
                ev = Event(ET.QUESTION, dt=dt,
                           payload=dict(qtext=Q.q_immediate_cmp(d["kind"], v1, v2, trng),
                                        ents=[]))
                ev.mode_gold = Mode.ANSWER_ENUM
                ev.family = int(Fam.IMMEDIATE_CMP)
                ev.enum_gold = ENUM_ID[gold]
                ev.enum_legal = [ENUM_ID["first"], ENUM_ID["second"]]
                ev.gold_value = gold
                ev.meta.update(delay=0, family=int(Fam.IMMEDIATE_CMP))
            elif t.kind == "question":
                ev = self._emit_question(d, dt, red, ops, goals, deadlines, qmeta, drop,
                                         key_text, current_exact_rid)
            else:
                raise ValueError(t.kind)

            ev.text = render(self.world, ev, trng)
            red.step(ev)
            # resolve rids for records created this event
            if ev.etype == ET.DOCUMENT or (ev.etype == ET.CORRECTION and ev.payload.get("exact")):
                ch = self.chains[(ev.payload["ent"], ev.payload["key"])]
                for a in ch.asserts:
                    if a.ev_idx == len(events) and a.rid < 0:
                        a.rid = red.last_adds[0]
            elif ev.etype == ET.DEADLINE_SET:
                ent = ev.payload["ent"]
                deadlines[ent] = (deadlines[ent][0], red.last_adds[0])
            elif ev.etype == ET.ACTION_COMMITTED:
                o = [o for o in ops.values() if o.get("commit_ev") == len(events)]
                if o:
                    o[0]["arg_rids"] = red.last_adds[:-1]
                    o[0]["ref_rid"] = red.last_adds[-1]
            elif ev.etype == ET.EXTERNAL_RESULT:
                o = [o for o in ops.values() if o.get("result_ev") == len(events)]
                if o:
                    o[0]["handle_rid"] = red.last_adds[0]
            elif ev.etype == ET.GOAL_SET:
                goals[ev.payload["gid"]]["rid"] = ev.payload["rid"]
            ev.meta["adds"] = list(red.last_adds)
            ev.meta["add_slots"] = list(red.last_add_slots)
            ev.meta["evicts"] = list(red.last_evicts)
            ev.meta["clock"] = red.clock
            events.append(ev)
            if (len(events)) % 32 == 0:
                digests.append((len(events), red.state_digest()))

        return events, red, digests

    # -------------------------------------------------------- question walker
    def _emit_question(self, d, dt, red, ops, goals, deadlines, qmeta, drop, key_text,
                       current_exact_rid) -> Event:
        rng, trng = self.rng, self.trng
        fam = d["fam"]
        W = self.world

        def pointer_event(qtext, ents, rid, dslot_ev, ent=None, key=None, ch=None):
            if rid not in red.records:
                return drop(fam.name)
            ev = Event(ET.QUESTION, dt=dt,
                       payload=dict(qtext=qtext, ents=ents))
            ev.mode_gold = Mode.ANSWER_POINTER
            ev.family = int(fam)
            ev.ptr_gold_rid = rid
            ev.gold_value = red.records[rid].value
            qmeta(ev, fam, dslot_ev, gold_rid=rid, ent=ent, key=key)
            if ch is not None:
                ev.meta["corrected"] = int(any((a.is_corr or not a.valid) and a.ev_idx >= 0
                                               for a in ch.asserts))
                ev.meta["reverted"] = int(any(not a.valid and a.ev_idx >= 0
                                              for a in ch.asserts))
                ev.meta["chain_len"] = len(ch.emitted())
            return ev

        if fam == Fam.SEM_LATEST:
            ch = self.chains.get((d["ent"], d["attr"]))
            a = (ch.current_chrono() if GOLD_MODE in ("sem", "v2")
                 else ch.current()) if ch else None
            if a is None:
                return drop(fam.name, "no_value")
            ev = Event(ET.QUESTION, dt=dt,
                       payload=dict(qtext=Q.q_sem_latest(W, d["ent"], d["attr"], trng),
                                    ents=[d["ent"]]))
            ev.mode_gold = Mode.ANSWER_ENUM
            ev.family = int(fam)
            ev.enum_gold = ENUM_ID[a.value]
            pool = ENUM_ATTRS[(W.entities[d["ent"]].kind, d["attr"])]
            ev.enum_legal = [ENUM_ID[v] for v in pool]
            ev.gold_value = a.value
            qmeta(ev, fam, a.ev_idx)
            ev.meta["chain_len"] = len(ch.emitted())
            ev.meta["corrected"] = int(any((x.is_corr or not x.valid) and x.ev_idx >= 0
                                           for x in ch.asserts))
            ev.meta["reverted"] = int(any(not x.valid and x.ev_idx >= 0
                                          for x in ch.asserts))
            return ev

        if fam == Fam.SEM_2HOP:
            relch = self.chains.get((d["src"], d["rel"]))
            _cc = GOLD_MODE in ("sem", "v2")
            ra = (relch.current_chrono() if _cc else relch.current()) if relch else None
            if ra is None:
                return drop(fam.name, "no_rel")
            dst = int(ra.value)
            dch = self.chains.get((dst, d["attr"]))
            da = (dch.current_chrono() if _cc else dch.current()) if dch else None
            if da is None:
                return drop(fam.name, "no_dst_value")
            ev = Event(ET.QUESTION, dt=dt,
                       payload=dict(qtext=Q.q_sem_2hop(W, d["src"], d["rel"], d["attr"], trng),
                                    ents=[d["src"], dst]))
            ev.mode_gold = Mode.ANSWER_ENUM
            ev.family = int(fam)
            ev.enum_gold = ENUM_ID[da.value]
            pool = ENUM_ATTRS[(W.entities[dst].kind, d["attr"])]
            ev.enum_legal = [ENUM_ID[v] for v in pool]
            ev.gold_value = da.value
            qmeta(ev, fam, max(ra.ev_idx, da.ev_idx))
            return ev

        if fam == Fam.EXACT_DELAYED:
            a = current_exact_rid(d["ent"], d["attr"])
            if a is None:
                return drop(fam.name)
            return pointer_event(Q.q_exact_delayed(W, d["ent"], d["attr"], trng),
                                 [d["ent"]], a.rid, a.ev_idx, ent=d["ent"], key=d["attr"],
                                 ch=self.chains.get((d["ent"], d["attr"])))

        if fam == Fam.EXACT_2HOP:
            relch = self.chains.get((d["src"], d["rel"]))
            ra = (relch.current_chrono() if GOLD_MODE == "v2"
                  else relch.current()) if relch else None
            if ra is None:
                return drop(fam.name, "no_rel")
            dst = int(ra.value)
            a = current_exact_rid(dst, d["attr"])
            if a is None:
                return drop(fam.name)
            return pointer_event(Q.q_exact_2hop(W, d["src"], d["rel"], d["attr"], trng),
                                 [d["src"], dst], a.rid, max(a.ev_idx, ra.ev_idx),
                                 ent=dst, key=d["attr"], ch=self.chains.get((dst, d["attr"])))

        if fam == Fam.SELF_REF:
            o = ops.get(d["seq"])
            if not o or o.get("dead") or "arg_rids" not in o:
                return drop(fam.name, "no_commit")
            return pointer_event(Q.q_self_ref(W, d["ent"], d["key"], d["op"], trng),
                                 [d["ent"]], o["arg_rids"][0], o["commit_ev"], ent=d["ent"])

        if fam == Fam.HANDLE_REF:
            o = ops.get(d["seq"])
            if not o or o.get("dead") or "handle_rid" not in o:
                return drop(fam.name, "no_result")
            return pointer_event(Q.q_handle_ref(W, d["ent"], d["op"], trng),
                                 [d["ent"]], o["handle_rid"], o["result_ev"], ent=d["ent"])

        if fam == Fam.TEMPORAL_ORDER:
            (ea, aa), (eb, ab) = d["a"], d["b"]
            ca, cb = self.chains.get((ea, aa)), self.chains.get((eb, ab))
            _cc = GOLD_MODE == "v2"
            la = (ca.current_chrono() if _cc else ca.current()) if ca else None
            lb = (cb.current_chrono() if _cc else cb.current()) if cb else None
            if la is None or lb is None:
                return drop(fam.name)
            if la.clock == lb.clock:
                return drop(fam.name, "tie")
            first_more_recent = la.clock > lb.clock
            want_first = (self.balance["tmp"] < 0
                          or (self.balance["tmp"] == 0 and rng.random() < 0.5))
            if first_more_recent != want_first:
                (ea, aa, la), (eb, ab, lb) = (eb, ab, lb), (ea, aa, la)
                first_more_recent = not first_more_recent
            gold = "first" if first_more_recent else "second"
            self.balance["tmp"] += 1 if gold == "first" else -1
            ev = Event(ET.QUESTION, dt=dt,
                       payload=dict(qtext=Q.q_temporal_order(W, ea, aa, eb, ab, trng),
                                    ents=[ea, eb]))
            ev.mode_gold = Mode.ANSWER_ENUM
            ev.family = int(fam)
            ev.enum_gold = ENUM_ID[gold]
            ev.enum_legal = [ENUM_ID["first"], ENUM_ID["second"]]
            ev.gold_value = gold
            qmeta(ev, fam, min(la.ev_idx, lb.ev_idx))
            return ev

        if fam == Fam.DEADLINE:
            ent = d["ent"]
            if ent not in deadlines:
                return drop(fam.name, "no_deadline")
            dl_clock, rid = deadlines[ent]
            passed = red.clock > dl_clock
            gold = "yes" if passed else "no"
            self.balance["yn"] += 1 if passed else -1
            ev = Event(ET.QUESTION, dt=dt,
                       payload=dict(qtext=Q.q_deadline(W, ent, trng), ents=[ent]))
            ev.mode_gold = Mode.ANSWER_ENUM
            ev.family = int(fam)
            ev.enum_gold = ENUM_ID[gold]
            ev.enum_legal = [ENUM_ID["yes"], ENUM_ID["no"]]
            ev.gold_value = gold
            ev.meta["family"] = int(fam)
            ev.meta["delay"] = -1
            ev.meta["margin_min"] = red.clock - dl_clock
            return ev

        if fam == Fam.GOAL_TOP:
            active = [(g["prio"], g) for g in goals.values() if g.get("active")
                      and g.get("rid", -1) in red.records]
            if not active:
                return drop(fam.name, "no_active_goal")
            prio, g = max(active, key=lambda x: x[0])
            return pointer_event(Q.q_goal_top(trng), [], g["rid"], -1)

        raise ValueError(fam)


def generate_lifetime(seed: int, length: int = 256):
    g = LifetimeGen(seed, length)
    g.plan()
    events, red, digests = g.walk()
    body = "\n".join(e.text for e in events) + "|".join(e.gold_value for e in events)
    digest = hashlib.sha256(body.encode()).hexdigest()
    return dict(seed=seed, length=length, events=events, reducer=red, digests=digests,
                dropped=g.dropped, digest=digest, tickets=g.slots)