"""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)