alex/sots-re
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
sots-re/tools/aiorders_to_tcb.py
alex 0d0b6dcfae lane CV: Rung B on a rich turn -- the replay runs, 1092 leaves, ranked worklist 
Replays lane BR's deep command block for ad-turn27-two-raiders.sav through
sots_turn --turn-commands and compares with bp-pinB-turn28.sav (724528ff).

Verdict: outcome 3. DIVERGED: 1092 leaf difference(s) against a 1166-leaf
do-nothing baseline; 80 closed, 6 regressed; the stream's whole contribution to
the state is /Sim/ModCount (1430 -> 1500, target 1502, residual 2).

Two blockers upstream of the turn: the typed writer drops one usp item in
Game::SpecialProjectNameGen (12 of 43 corpus saves, exactly 12 bytes each), and
--relabel-new-ids refuses by guard G3 because the engine mints no client fleet
ids. The tail's tscr gate is TRUE on this save (253, not 252).

tools/aiorders_to_tcb.py now reads the deep dump's aivec/aistr rows, only at a
word the field map already types as a vector, with the followed count
cross-checked against the begin/end pair.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ARBgSooAfokKUy6wKUKEyZ
2026-09-09 09:28:50 -04:00

337 lines
17 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Turn a live `aiorders` shim dump into a `.tcb` turn-command capture.
The shim's dump is a memory view: for each submitted block it prints the six prologue gates,
the twenty-seven list lengths, and a fixed 48-byte window of every element. That is the right
thing for an instrument to emit -- it commits to nothing -- but it is not a turn record, because
the fields are in MEMORY order (and one list's writer runs them backwards), the floats are still
integers, and a payload behind a pointer is simply absent.
This is the mechanical step in between. It applies the per-list field mapping, reinterprets the
words the element record says are floats, converts a vector's begin/end pair into a length, and
writes `?` wherever the dump window could not reach. The output is what `sots_turn
--turn-commands` reads.
tools/aiorders_to_tcb.py LOG [-o OUT] [--batch SEQ] [--name PID=TECHNAME]... [--seed ID=HEX]...
Two things this tool deliberately will not do.
* It will not invent a payload. A route whose hops live behind a pointer becomes `vN` -- a
vector of known length and unknown contents -- and never `v1:0`. The replayer counts such a
command and refuses to apply it, which is the whole point: a command that was issued and a
command whose effect we can reproduce are different facts.
* It will not pick a batch for you when the choice is ambiguous. A process applies a command
batch at LOAD as well as at End Turn; replaying the load-time one against a save that was
written after it would double-count. The default is the LAST batch in the log, which is the
End-Turn one, and `--batch` overrides it.
`--name PID=TECHNAME` records the tech NAME an instrument observed a research-target gate resolve
to. The wire carries an integer id and the save carries a name; the client resolves one to the
other off the command and that map is unread, so the name can only be recorded, never computed.
`--seed NETID=VALUE` records an AI client's construction seed.
"""
import argparse
import re
import struct
import sys
# Per list: how the 48-byte memory window maps onto the element's wire fields.
#
# `order` is the sequence of word indices to emit; `kind` says how to read each one. The build
# list is the only one that reverses -- its writer emits +0x14, +0x10, +0x0c, +0x08, descending --
# and that is per-list, not a rule. Every other observed list writes ascending.
#
# i an int word
# f a word to reinterpret as a float
# b a word whose low byte is a bool
# v a std::vector: this word and the next hold begin/end, and the length is their difference
# over four. The CONTENTS are behind the pointer and the dump does not follow it, so the
# field is emitted with a length and no values.
# ? a word the element record says is part of an object this window cannot read
LIST_MAP = {
1: [("?", 0)], # a ShipDesignDef object
3: [("i", 3), ("i", 2), ("i", 1), ("i", 0)], # DESCENDING: ordinal, design, system, w
# List 5 carries a system id and then the seven planetary-budget sliders. The id is certain;
# THE SLIDERS ARE NOT, and this used to map them straight onto the wire order and was wrong.
# A replay built on that mapping wrote the AI's one non-zero slider into the wrong member and
# regressed two leaves on the canonical pair -- the command re-issues the rates the save
# already holds, so a correct applier is a no-op and an incorrect one is immediately visible.
#
# What is known: the only non-zero word in every dumped element of this list is at index 2,
# and the same command on the WIRE (where the frame is named) puts its only non-zero in
# `SRsc`, the THIRD member. So memory member 1 is wire member `SRsc` and the frame's memory
# order is not its wire order -- one correspondence pinned, six unread. The `?` below is
# that ignorance, and the replayer counts such a command and refuses to apply it.
#
# The experiment that settles it is one UI run: push two DIFFERENT sliders to two different
# values, capture the block, and read the permutation straight off. Cheaper still, a save
# taken after issuing rates carries the same command on the wire with every field NAMED, so a
# capture converted from a save's own TurnCommands block needs no memory mapping at all.
5: [("i", 0)] + [("?", k) for k in range(1, 8)],
7: [("i", 0), ("i", 1)],
8: [("i", 0), ("v", 1)],
10: [("i", 0), ("i", 1), ("v", 2)],
14: [("i", 0), ("i", 1), ("b", 2)],
23: [("i", 0), ("?", 1)], # a Population object
}
BLK_RE = re.compile(
r"^aiblk seq=(\d+) blk=(\d+)/(\d+) at=\S+ pid=(-?\d+) "
r"rate=(\d):(\S+) target=(\d):(-?\d+) boost=(\d):(-?\d+),(\S+) g4=(\d):(-?\d+),(-?\d+) "
r"f3=(\d):(\S+),(\S+),(\S+) civ=(\d)")
LISTS_RE = re.compile(r"^ailists seq=(\d+) blk=(\d+) pid=(-?\d+) nonEmpty=\d+ sizes\(1\.\.27\)=\[([^\]]*)\]")
ELEM_RE = re.compile(r"^aielem blk=(\d+) pid=(-?\d+) list=(\d+) idx=(\d+) .*? hex=\[([^\]]*)\]")
# The two records a DEEP dump adds (`aiorders.deep=on`). Each says which WORD of the fixed
# element window it followed, so it lands on exactly the field the map already calls a `v` or a
# string -- it is not a second, parallel field map.
VEC_RE = re.compile(r"^aivec blk=(\d+) pid=(-?\d+) list=(\d+) idx=(\d+) at=w(\d+) "
r"first=\S+ cap=(\d+) count=(\d+) ints=\[([^\]]*)\]")
STR_RE = re.compile(r"^aistr blk=(\d+) pid=(-?\d+) list=(\d+) idx=(\d+) at=w(\d+) "
r"sso=\d+ len=(\d+) text=\"(.*)\"\s*$")
def f32(word):
"""A dumped word, as the float32 it is, printed so it round-trips exactly."""
return "%.9g" % struct.unpack("<f", struct.pack("<I", word & 0xffffffff))[0]
def parse_log(path):
"""-> {seq: {"n": int, "blocks": {idx: block}}}"""
batches = {}
def batch(seq):
return batches.setdefault(seq, {"n": 0, "blocks": {}})
with open(path, encoding="utf-8", errors="replace") as f:
for line in f:
m = BLK_RE.match(line)
if m:
seq, idx, n = int(m.group(1)), int(m.group(2)), int(m.group(3))
b = batch(seq)
b["n"] = max(b["n"], n)
b["blocks"][idx] = {
"pid": int(m.group(4)),
"rate": (m.group(5) == "1", m.group(6)),
"target": (m.group(7) == "1", int(m.group(8))),
"boost": (m.group(9) == "1", int(m.group(10)), m.group(11)),
"g4": (m.group(12) == "1", int(m.group(13)), int(m.group(14))),
"f3": (m.group(15) == "1", m.group(16), m.group(17), m.group(18)),
"civ": m.group(19) == "1",
"sizes": [0] * 27,
"elems": {},
"vecs": {},
"strs": {},
}
continue
m = LISTS_RE.match(line)
if m:
seq, idx = int(m.group(1)), int(m.group(2))
blk = batch(seq)["blocks"].get(idx)
if blk is not None:
blk["sizes"] = [int(x) for x in m.group(4).split()]
continue
m = ELEM_RE.match(line)
if m:
# An `aielem` line carries no seq, so it belongs to the batch whose block header
# it followed -- which is the most recent one seen.
seq = max(batches) if batches else 0
idx, listno, elemidx = int(m.group(1)), int(m.group(3)), int(m.group(4))
words = [int(x, 16) for x in m.group(5).split()]
blk = batch(seq)["blocks"].get(idx)
if blk is not None:
blk["elems"].setdefault(listno, {})[elemidx] = words
continue
m = VEC_RE.match(line)
if m:
seq = max(batches) if batches else 0
idx, listno, elemidx = int(m.group(1)), int(m.group(3)), int(m.group(4))
at, count = int(m.group(5)), int(m.group(7))
ints = [int(x) for x in m.group(8).split()]
blk = batch(seq)["blocks"].get(idx)
if blk is not None:
# `count` is what the dump says it followed; `ints` is what it printed. A
# disagreement is a defect in the record and this tool refuses to guess which
# half is right -- the element keeps its shallow (length-only) form.
if len(ints) == count:
blk["vecs"].setdefault(listno, {}).setdefault(elemidx, {})[at] = ints
continue
m = STR_RE.match(line)
if m:
seq = max(batches) if batches else 0
idx, listno, elemidx = int(m.group(1)), int(m.group(3)), int(m.group(4))
at, text = int(m.group(5)), m.group(7)
blk = batch(seq)["blocks"].get(idx)
if blk is not None:
blk["strs"].setdefault(listno, {}).setdefault(elemidx, {})[at] = text
return batches
def element_fields(listno, words, deep=None, problems=None):
"""The wire fields of one element, as `.tcb` tokens.
`deep` is {word index -> [values]} from a deep dump's `aivec` rows. It is consulted ONLY at
a word the shallow map already types as a vector: the deep rows say which word they followed,
so a row that lands anywhere else is a disagreement between the two records and is reported,
never silently used to invent a field the map does not have.
"""
deep = deep or {}
spec = LIST_MAP.get(listno)
if spec is None:
# An unmapped list: record that the element exists and nothing about it. The command
# still costs whatever its list costs; the replayer will decline to apply it.
return ["?"]
out = []
for kind, k in spec:
if k >= len(words):
out.append("?")
continue
w = words[k]
if kind == "i":
out.append("i%d" % struct.unpack("<i", struct.pack("<I", w))[0])
elif kind == "f":
out.append("f%s" % f32(w))
elif kind == "b":
out.append("b%d" % (1 if (w & 0xff) else 0))
elif kind == "v":
if k + 1 >= len(words):
out.append("?")
continue
n = (words[k + 1] - w) // 4
values = deep.get(k)
if values is None:
out.append("v%d" % n)
elif len(values) != n:
# The begin/end pair and the followed contents disagree. Keep the length-only
# form: a route whose hops we are not sure of must stay unapplied.
if problems is not None:
problems.append("list %d elem %d word %d: begin/end says %d hop(s), the deep "
"row printed %d -- kept as length-only"
% (listno, k, k, n, len(values)))
out.append("v%d" % n)
else:
out.append("v%d:%s" % (n, ",".join(str(v) for v in values)))
else:
out.append("?")
return out
def emit(batch, seq, source, names, seeds, input_name, problems=None):
if problems is None:
problems = []
lines = ["tcb 1",
"meta source %s" % source,
"meta input %s" % input_name,
"meta batch seq=%d n=%d" % (seq, batch["n"]),
"meta note the load-time batch is excluded; this is the End-Turn submission"]
for netid, value in seeds:
lines.append("seed %s %s" % (netid, value))
for idx in sorted(batch["blocks"]):
b = batch["blocks"][idx]
lines.append("block %d %d" % (idx, b["pid"]))
if b["rate"][0]:
lines.append("gate %d rate %s" % (idx, b["rate"][1]))
if b["target"][0]:
name = names.get(b["pid"])
lines.append("gate %d target %d%s" % (idx, b["target"][1],
(" name %s" % name) if name else ""))
if b["boost"][0]:
lines.append("gate %d boost %d %s" % (idx, b["boost"][1], b["boost"][2]))
if b["g4"][0]:
lines.append("gate %d group4 %d %d" % (idx, b["g4"][1], b["g4"][2]))
if b["f3"][0]:
lines.append("gate %d group5 %s %s %s" % (idx, b["f3"][1], b["f3"][2], b["f3"][3]))
if b["civ"]:
lines.append("gate %d civilian" % idx)
for listno in range(1, 28):
n = b["sizes"][listno - 1]
if not n:
continue
lines.append("list %d %d %d" % (idx, listno, n))
have = b["elems"].get(listno, {})
vecs = b.get("vecs", {}).get(listno, {})
strs = b.get("strs", {}).get(listno, {})
spec = LIST_MAP.get(listno) or []
vecwords = {k for kind, k in spec if kind == "v"}
for e in range(n):
words = have.get(e)
deep = vecs.get(e, {})
fields = (element_fields(listno, words, deep, problems)
if words is not None else ["?"])
lines.append("elem %d %d %d %s" % (idx, listno, e, " ".join(fields)))
# A deep row the field map has no vector at. It is a real observation and it is
# NOT a field: recording it as one would be inventing a mapping the dump does not
# justify. It goes in as a comment so the record keeps it and the parser does
# not read it, and it is counted as a followed payload this tool cannot place.
for at in sorted(deep):
if at in vecwords:
continue
problems.append(
"list %d elem %d: a followed vector at word %d, which the field map does "
"not type as a vector -- recorded as a comment, not applied" % (listno, e, at))
lines.append("# observed list %d elem %d word %d vector: [%s]"
% (listno, e, at, ",".join(str(v) for v in deep[at])))
for at in sorted(strs.get(e, {})):
problems.append(
"list %d elem %d: a followed string at word %d, which the field map types "
"as part of an object it cannot read -- recorded as a comment, not applied"
% (listno, e, at))
lines.append("# observed list %d elem %d word %d string: %s"
% (listno, e, at, strs[e][at]))
return "\n".join(lines) + "\n"
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("log")
ap.add_argument("-o", "--out")
ap.add_argument("--batch", type=int, help="which seq to convert (default: the last)")
ap.add_argument("--input", default="", metavar="SAVE",
help="the save this turn was run FROM. A capture belongs to one board and "
"replaying it against another charges the counter happily and is wrong; "
"recording it here lets the replayer say so.")
ap.add_argument("--name", action="append", default=[], metavar="PID=TECHNAME",
help="the tech name a research-target gate was observed to resolve to")
ap.add_argument("--seed", action="append", default=[], metavar="NETID=VALUE",
help="an AI client's construction seed")
a = ap.parse_args(argv)
batches = parse_log(a.log)
if not batches:
print("no aiblk records in %s" % a.log, file=sys.stderr)
return 2
seq = a.batch if a.batch is not None else max(batches)
if seq not in batches:
print("no batch seq=%d; the log holds %s" % (seq, sorted(batches)), file=sys.stderr)
return 2
names = {}
for n in a.name:
pid, _, tech = n.partition("=")
if not tech:
print("--name wants PID=TECHNAME", file=sys.stderr)
return 2
names[int(pid)] = tech
seeds = []
for s in a.seed:
netid, _, value = s.partition("=")
if not value:
print("--seed wants NETID=VALUE", file=sys.stderr)
return 2
seeds.append((netid, value))
problems = []
text = emit(batches[seq], seq, a.log.split("/")[-1], names, seeds, a.input, problems)
for p in problems:
print("deep: %s" % p, file=sys.stderr)
if a.out:
with open(a.out, "w") as f:
f.write(text)
print("wrote %s (batch seq=%d, %d blocks)" % (a.out, seq, len(batches[seq]["blocks"])))
else:
sys.stdout.write(text)
return 0
if __name__ == "__main__":
raise SystemExit(main())
Reference in a new issue View git blame Copy permalink