File size: 3,955 Bytes
f930dac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """Exact binding addresses for E3-BIND.
Computes, for each event of a lifetime, which learned slot (if any) that event
is allowed to write, and for each semantic question which slot holds the
currently authoritative version of the queried binding.
Contract (preregistered):
* the address carries entity id, attribute/relation type, version and
slot id - NEVER the semantic value or the answer;
* one slot per (entity, attribute, version); NO slot is ever reused;
* slot indices are drawn from a per-lifetime random permutation seeded by the
lifetime seed alone, so the index cannot encode the label;
* a revert moves the AUTHORITY POINTER to an earlier version's slot; it does
not move or rewrite any payload.
"""
from __future__ import annotations
import numpy as np
from ..world.schema import ET, Fam
N_BIND_SLOTS = 192 # E3-A2: measured max 143 versions/lifetime, p99 139
def binding_plan(lt: dict, n_slots: int = N_BIND_SLOTS) -> dict:
"""-> per-event write slot, per-question read slot, per-slot exact key."""
events = lt["events"]
rng = np.random.default_rng(int(lt["seed"])) # label-independent
perm = rng.permutation(n_slots)
nxt = 0
write_slot = np.full(len(events), -1, np.int16)
read_slot = np.full(len(events), -1, np.int16)
slot_ent = np.full(n_slots, -1, np.int16)
slot_attr = np.full(n_slots, -1, np.int16)
slot_ver = np.full(n_slots, -1, np.int16)
active: dict[tuple, int] = {} # (ent, attr) -> slot of active version
versions: dict[tuple, list] = {} # (ent, attr) -> [(clock, source, slot)]
clock = 0
overflow = 0
def alloc(key, clock, source):
nonlocal nxt, overflow
if nxt >= n_slots:
overflow += 1
return -1
s = int(perm[nxt]); nxt += 1
slot_ent[s] = key[0] % 4096
slot_attr[s] = ATTR_ID.get(key[1], 0)
slot_ver[s] = len(versions.get(key, []))
versions.setdefault(key, []).append((clock, source, s))
active[key] = s
return s
for i, ev in enumerate(events):
clock += ev.dt
p = ev.payload
if ev.etype == ET.OBSERVATION:
key = (p["ent"], p["attr"])
write_slot[i] = alloc(key, clock, p.get("source", ""))
elif ev.etype == ET.CORRECTION:
key = (p["ent"], p["key"])
if p["mode"] == "replace":
write_slot[i] = alloc(key, clock, p.get("source", ""))
else:
# revert: authority pointer only, no payload written anywhere
hist = versions.get(key, [])
tgt = None
if p.get("ref_kind") == "source":
for c, src, s in hist:
if src == p.get("ref_source"):
tgt = s
else:
for c, src, s in hist:
if c == p.get("ref_clock"):
tgt = s
if tgt is not None:
active[key] = tgt
elif ev.etype == ET.QUESTION:
tk = lt["tickets"][i] if "tickets" in lt else None
d_ = tk.data if tk is not None else {}
if ev.family == int(Fam.SEM_LATEST):
read_slot[i] = active.get((d_.get("ent"), d_.get("attr")), -1)
elif ev.family == int(Fam.SEM_2HOP):
# ONLY the first hop is addressed; hop 2 stays a learned lookup
read_slot[i] = active.get((d_.get("src"), d_.get("rel")), -1)
return dict(write_slot=write_slot, read_slot=read_slot, slot_ent=slot_ent,
slot_attr=slot_attr, slot_ver=slot_ver,
n_used=nxt, overflow=overflow)
ATTR_ID = {a: i + 1 for i, a in enumerate(sorted(
["location", "team", "status", "phase", "priority", "destination",
"owner", "project", "vault", "access code", "firmware tag", "reading",
"quantity", "budget"]))}
|