#!/usr/bin/env python3 """vm0 — headless AGE bytecode interpreter (Phase A0 prototype). Executes one script's bytecode to validate the *execution model* before any C#/Godot work. Reuses tools/sys4load.py for all parsing/decoding. Effectful ops and call-script are STUBBED; show-text is captured. Named-op semantics come from Kelebek's table; the classified markers are treated as no-ops (this run TESTS that assumption). Purpose (see docs/phase-a-slice-plan.md): * `--test` run the RECOVER unit test (pointer / 2D-array / loop / control-flow correctness). * run a script; print captured show-text + an opcode-coverage/stub report. Memory model: * one flat GLOBAL bank G (dict addr->int); globals are raw offsets into one space. * per-call local frame with sparse typed banks (int/float/string/ptr). * a `-ptr` variable holds an ADDRESS into G. lookup-array/2d with a ptr dst stores that address (take-reference); reading a ptr derefs (G[addr]); writing through a ptr writes G[addr]. This is the model RECOVER forces; the unit test is its litmus. * jcc(cond, tA, tB): cond truthy -> goto tA else tB; 0xFFFFFFFF = fall through (from RECOVER+SCJUMP). * call/ret (0x8F/0x05) are intra-script subroutine calls (shared frame); call-script (0x03) is the inter-script one and is stubbed. """ from __future__ import annotations import sys import collections from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import paths import sys4load from age_opcodes import OPCODES NOJUMP = 0xFFFFFFFF # operand type tags T_IMM, T_STR = 0x0, 0x2 T_GINT, T_GFLOAT, T_GSTR, T_GPTR = 0x3, 0x4, 0x5, 0x6 T_LINT, T_LFLOAT, T_LSTR, T_LPTR = 0x9, 0xA, 0xB, 0xC # opcodes treated as no-ops in v1 (classified markers; see age_opcodes_himegari.py) # 0x71 = label-definition pseudo-op (count == T1 table size) — structural, no runtime effect. MARKERS = {0x71, 0x1f4, 0x1f5, 0x1d5, 0x1bc, 0x1bf, 0x21b, 0x1d2, 0x258} # loop-guard: halt a run once any single show-text line has been emitted this many times. # State-gated scenes run with zero initial state can spin a loop that re-emits the same block # forever; capping re-emission terminates them cleanly and flags them LOOPED (vs. blind step limit). EMIT_CAP = 2 class Frame: def __init__(self): self.i = collections.defaultdict(int) # local-int self.f = collections.defaultdict(int) # local-float (stored raw) self.s = collections.defaultdict(str) # local-string self.p = collections.defaultdict(int) # local-ptr (holds a G address) class VM: def __init__(self, scr: sys4load.Sys4Script, verbose=False, emit_cap=EMIT_CAP): self.scr = scr self.verbose = verbose self.code = scr.instructions self.by_off = {ins.offset: idx for idx, ins in enumerate(self.code)} self.G = collections.defaultdict(int) # global-int bank (flat address space) self.Gs = collections.defaultdict(str) # global-string bank self.fr = Frame() self.callstack = [] # return indices for call/ret self.text = [] # captured show-text as (str_offset, text) self.emit_seen = collections.Counter() # per-offset emit count (loop-guard) self.emit_cap = emit_cap self.halt_reason = None # 'exit' | 'LOOP:...' | 'STEP-LIMIT' | 'ret-underflow' self.log = collections.Counter() # stub/unknown opcode hits self.exec_count = collections.Counter() # opcode coverage self.steps = 0 # ---- operand resolution -------------------------------------------------- def _string(self, off): e = self.scr.strings.get(off) if e is not None: return e[0] txt, _ = sys4load._decode_string(self.scr.dwords, off) return txt if txt is not None else "" def read(self, op): t, v = op if t == T_IMM: return v if t == T_STR: return self._string(v) if t == T_GINT: return self.G[v] if t == T_GFLOAT: return self.G[v] if t == T_GSTR: return self.Gs[v] if t == T_GPTR: return self.G[self.G[v]] if t == T_LINT: return self.fr.i[v] if t == T_LFLOAT: return self.fr.f[v] if t == T_LSTR: return self.fr.s[v] if t == T_LPTR: return self.G[self.fr.p[v]] # deref ptr self.log[f"read?t{t:#x}"] += 1 return v def write(self, op, val): t, v = op if t == T_GINT or t == T_GFLOAT: self.G[v] = val elif t == T_GSTR: self.Gs[v] = val elif t == T_GPTR: self.G[self.G[v]] = val elif t == T_LINT: self.fr.i[v] = val elif t == T_LFLOAT: self.fr.f[v] = val elif t == T_LSTR: self.fr.s[v] = val elif t == T_LPTR: self.G[self.fr.p[v]] = val # write through else: self.log[f"write?t{t:#x}"] += 1 def base_addr(self, op): """The base ADDRESS an operand names, for array lookups.""" t, v = op if t in (T_IMM, T_GINT, T_GFLOAT, T_GSTR, T_GPTR): return v # global's own offset if t == T_LINT: return self.fr.i[v] if t == T_LPTR: return self.fr.p[v] return v def lookup_store(self, dst, addr): """lookup result: ptr dst gets the reference (address); non-ptr gets the element value.""" t, v = dst if t == T_LPTR: self.fr.p[v] = addr elif t == T_GPTR: self.G[v] = addr else: self.write(dst, self.G[addr]) # ---- execution ----------------------------------------------------------- def run(self, entry_off=0, max_steps=2_000_000): pc = self.by_off.get(entry_off, 0) while 0 <= pc < len(self.code): if self.steps >= max_steps: self.log["STEP-LIMIT"] += 1 self.halt_reason = self.halt_reason or "STEP-LIMIT" break self.steps += 1 ins = self.code[pc] op = ins.opcode self.exec_count[op] += 1 nxt = self.step(ins, pc) if nxt is None: # halt break pc = nxt else: self.halt_reason = self.halt_reason or "pc-out-of-range" return self def step(self, ins, pc): op, a = ins.opcode, ins.args lbl = OPCODES.get(op, ("?", 0))[0] # arithmetic / bitwise: dst = a1 a2 alu = {"add": lambda x, y: x + y, "sub": lambda x, y: x - y, "mul": lambda x, y: x * y, "div": lambda x, y: (x // y if y else 0), "mod": lambda x, y: (x % y if y else 0), "and": lambda x, y: x & y, "or": lambda x, y: x | y, "sar": lambda x, y: x >> (y & 31), "shl": lambda x, y: x << (y & 31)} cmp = {"eq": lambda x, y: int(x == y), "ne": lambda x, y: int(x != y), "lt": lambda x, y: int(x < y), "lte": lambda x, y: int(x <= y), "gr": lambda x, y: int(x > y), "gre": lambda x, y: int(x >= y)} if lbl in alu: self.write(a[0], alu[lbl](self.read(a[1]), self.read(a[2]))); return pc + 1 if lbl in cmp: self.write(a[0], cmp[lbl](self.read(a[1]), self.read(a[2]))); return pc + 1 if lbl == "mov": self.write(a[0], self.read(a[1])); return pc + 1 if lbl == "set-string": self.write(a[0], self.read(a[1])); return pc + 1 if lbl == "lookup-array": # dst = base[idx] addr = self.base_addr(a[1]) + self.read(a[2]) self.lookup_store(a[0], addr); return pc + 1 if lbl == "lookup-array-2d": # dst = base[i*stride + col] addr = self.base_addr(a[1]) + self.read(a[2]) * self.read(a[3]) + self.read(a[4]) self.lookup_store(a[0], addr); return pc + 1 if lbl == "bit-set": self.write(a[0], self.read(a[0]) | self.read(a[1])); return pc + 1 if lbl == "bit-reset": self.write(a[0], self.read(a[0]) & ~self.read(a[1])); return pc + 1 if lbl == "check-bit": # p1 = (p2 >> p3) & 1 self.write(a[0], (self.read(a[1]) >> (self.read(a[2]) & 31)) & 1); return pc + 1 if lbl == "copy-to-global": # best-effort: p1 = p2 (single cell) self.write(a[0], self.read(a[1])); return pc + 1 # control flow if lbl == "jmp": return self.by_off.get(a[0][1], pc + 1) if lbl == "call": # intra-script subroutine self.callstack.append(pc + 1) return self.by_off.get(a[0][1], pc + 1) if lbl == "ret": if self.callstack: return self.callstack.pop() self.halt_reason = "ret-underflow" return None if lbl == "jcc": # (cond, tA, tB); 0xFFFFFFFF=fallthrough tgt = a[1][1] if self.read(a[0]) else a[2][1] return pc + 1 if tgt == NOJUMP else self.by_off.get(tgt, pc + 1) if lbl in ("exit", "exit-script"): self.halt_reason = "exit" return None if lbl == "call-script": # STUB (inter-script) self.log[f"call-script({self.read(a[0]) if a else '?'})"] += 1; return pc + 1 # ADV text: capture the string operand as (offset, text); loop-guard on re-emission if lbl == "show-text": for t, v in a: if t == T_STR: self.emit_seen[v] += 1 if self.emit_seen[v] > self.emit_cap: self.halt_reason = f"LOOP:line@{v:#x}×{self.emit_seen[v]}" return None self.text.append((v, self._string(v))) return pc + 1 if lbl in ("end-text-line", "wait-for-input", "set-font", "comment", "display-furigana", "dev_ukn"): return pc + 1 if op in MARKERS: # classified no-op markers return pc + 1 # everything else (effectful draw/audio/ui/input, unnamed) -> stub + continue self.log[f"stub:{lbl if not lbl.startswith('u00') else hex(op)}"] += 1 return pc + 1 def run_test(): """RECOVER unit test — pointer/2D-array/loop/control-flow correctness.""" scr = sys4load.load(paths.DATA1 / "RECOVER.BIN") vm = VM(scr) unit = 0 vm.G[0x152616] = unit A, B = 0x4e11b, 0x4e085 # block-1 tables (stride 14 / 3) C, E, F, FL = 0x52383, 0x52f3b, 0x5295f, 0xaacb4 # block-2 tables (stride 30) + flags # block 1 source: A[unit*14 + 11..13] for k in range(3): vm.G[A + unit * 14 + (11 + k)] = 100 + k # block 2: slot 5 active, slot 6 zero-C (skip), slot 7 flag-off (skip) vm.G[C + unit * 30 + 5] = 7; vm.G[FL + 5] = 1; vm.G[E + unit * 30 + 5] = 42 vm.G[C + unit * 30 + 6] = 0; vm.G[FL + 6] = 1; vm.G[E + unit * 30 + 6] = 99 vm.G[C + unit * 30 + 7] = 3; vm.G[FL + 7] = 0; vm.G[E + unit * 30 + 7] = 88 vm.run() checks = [ ("block1 B[0]=A[11]", vm.G[B + unit * 3 + 0], 100), ("block1 B[1]=A[12]", vm.G[B + unit * 3 + 1], 101), ("block1 B[2]=A[13]", vm.G[B + unit * 3 + 2], 102), ("block2 s5 C:=E", vm.G[C + unit * 30 + 5], 42), ("block2 s5 F:=-1", vm.G[F + unit * 30 + 5], -1), ("block2 s6 C skip", vm.G[C + unit * 30 + 6], 0), # C stayed 0 (guard) ("block2 s7 C skip", vm.G[C + unit * 30 + 7], 3), # flag off -> untouched ] ok = True for name, got, want in checks: status = "OK " if got == want else "FAIL" if got != want: ok = False print(f" [{status}] {name}: got {got}, want {want}") print(f" steps={vm.steps} call-script stubs={sum(v for k,v in vm.log.items() if k.startswith('call-script'))}") print("RECOVER unit test:", "PASS" if ok else "FAIL") return 0 if ok else 1 def run_file(path): scr = sys4load.load(path) vm = VM(scr).run() print(f"== {Path(path).name}: {vm.steps} steps, {len(vm.text)} show-text lines captured " f"(halt: {vm.halt_reason}) ==") for i, (off, t) in enumerate(vm.text[:20]): print(f" [{i}] {t}") if len(vm.text) > 20: print(f" ... (+{len(vm.text)-20} more)") stubs = [(k, v) for k, v in vm.log.most_common() if k.startswith("stub:")] if stubs: print(" top stubbed effectful/unknown ops:", ", ".join(f"{k[5:]}×{v}" for k, v in stubs[:10])) ncall = sum(v for k, v in vm.log.items() if k.startswith("call-script")) print(f" call-script stubs: {ncall} distinct opcodes executed: {len(vm.exec_count)}") return 0 # ---- dialogue oracle --------------------------------------------------------- import json import re SCENE_RE = re.compile(r"^S[CP]\d{4}\.BIN$") def load_oracle(path=None): """file (UPPER) -> ordered list of (str_offset, text): the static show-text lines extract_phase2 dumped. This is the independent oracle the VM is validated against.""" path = path or (paths.BUILD / "text" / "dialogue.jsonl") by = collections.defaultdict(list) with open(path, encoding="utf-8") as f: for line in f: d = json.loads(line) by[d["file"].upper()].append((int(d["off"], 16), d["text"])) return by def subsequence_status(emitted_offs, static_offs): """Classify the emitted show-text offset stream against the static ordered set. Emitted may repeat offsets (loop re-emission); we validate the distinct stream in first-seen order as an in-order subsequence of the static lines. Returns (status, detail): OK | STRAY (offset never in static) | ORDER (in static but out of order).""" static_list = list(static_offs) static_set = set(static_list) seen, distinct = set(), [] for o in emitted_offs: if o not in seen: seen.add(o) distinct.append(o) strays = [o for o in distinct if o not in static_set] if strays: return "STRAY", strays j = 0 for o in distinct: while j < len(static_list) and static_list[j] != o: j += 1 if j >= len(static_list): return "ORDER", [o] j += 1 return "OK", [] def verdict(status, halt, n_emit, n_static): """Fold subsequence status + halt reason into one scene verdict.""" if n_static == 0: return "NO-DIALOGUE" # not an ADV scene (no static show-text) — skip in scoring if n_emit == 0: return "EMPTY" # scene has dialogue but VM emitted none — investigate if status != "OK": return status # STRAY / ORDER — a real divergence return "CLEAN" if halt == "exit" else f"OK/{halt}" # OK subsequence; did it exit cleanly? def run_scene(name, path, oracle): """Run one scene, return a result dict comparing emitted vs static show-text.""" scr = sys4load.load(path) vm = VM(scr).run() emitted = [off for off, _ in vm.text] static = oracle.get(name.upper(), []) static_offs = [o for o, _ in static] status, detail = subsequence_status(emitted, static_offs) n_emit = len({o for o in emitted}) return {"name": name, "vm": vm, "emitted": emitted, "static": static, "n_emit_distinct": n_emit, "n_static": len(static_offs), "status": status, "detail": detail, "verdict": verdict(status, vm.halt_reason, n_emit, len(static_offs))} def run_one_scene(name): """Detailed single-scene oracle diff (for investigating one script).""" oracle = load_oracle() scripts = paths.scripts() name = name.upper() if not name.endswith(".BIN"): name += ".BIN" if name not in scripts: print(f"no such script: {name}") return 1 r = run_scene(name, scripts[name], oracle) vm = r["vm"] print(f"== {name}: verdict {r['verdict']} ==") print(f" emitted {len(r['emitted'])} lines ({r['n_emit_distinct']} distinct), " f"static {r['n_static']}, steps {vm.steps}, halt {vm.halt_reason}") if r["status"] == "STRAY": stat_set = {o for o, _ in r["static"]} print(f" STRAY offsets (emitted, not in static): " f"{', '.join(hex(o) for o in r['detail'])}") for off, txt in vm.text: if off in set(r["detail"]): print(f" @ {off:#x}: {txt!r}") elif r["status"] == "ORDER": print(f" first out-of-order offset: {hex(r['detail'][0])}") return 0 def run_sweep(limit=None): """Run every SC####/SP#### scene against the dialogue oracle; print a coverage table.""" oracle = load_oracle() scripts = paths.scripts() names = sorted(n for n in scripts if SCENE_RE.match(n)) if limit: names = names[:limit] buckets = collections.Counter() rows = [] for name in names: r = run_scene(name, scripts[name], oracle) buckets[r["verdict"]] += 1 rows.append(r) # non-clean scenes get listed for follow-up problem = [r for r in rows if r["verdict"] not in ("CLEAN", "NO-DIALOGUE")] if problem: print("non-clean scenes:") for r in sorted(problem, key=lambda r: r["verdict"]): d = "" if r["status"] in ("STRAY", "ORDER"): d = " " + " ".join(hex(o) for o in r["detail"][:4]) print(f" {r['name']:<14} {r['verdict']:<16} " f"emit {r['n_emit_distinct']:>4}/{r['n_static']:<4} " f"steps {r['vm'].steps:>7} halt {r['vm'].halt_reason}{d}") print() scene_total = sum(v for k, v in buckets.items() if k != "NO-DIALOGUE") valid = sum(v for k, v in buckets.items() if k == "CLEAN" or k.startswith("OK/")) print(f"scenes with dialogue: {scene_total} (skipped {buckets['NO-DIALOGUE']} with no static show-text)") print("verdict breakdown:", dict(sorted(buckets.items()))) print(f"DIALOGUE-VALID (clean in-order subsequence, no garbage): " f"{valid}/{scene_total} = {valid/scene_total*100:.1f}%") return 0 def run_trace(out_path): """Dump per-scene emitted show-text offsets + halt + steps for the differential test.""" oracle = load_oracle() scripts = paths.scripts() names = sorted(n for n in scripts if SCENE_RE.match(n)) trace = {} for name in names: r = run_scene(name, scripts[name], oracle) vm = r["vm"] trace[name] = {"offsets": [off for off, _ in vm.text], "halt": vm.halt_reason, "steps": vm.steps} Path(out_path).write_text(json.dumps(trace, ensure_ascii=False), encoding="utf-8") print(f"trace: {len(trace)} scenes -> {out_path}") return 0 def main(argv=None): argv = argv if argv is not None else sys.argv[1:] if not argv or argv[0] == "--test": return run_test() if argv[0] == "--sweep": return run_sweep(limit=int(argv[1]) if len(argv) > 1 else None) if argv[0] == "--scene": return run_one_scene(argv[1]) if argv[0] == "--trace": return run_trace(argv[1]) return run_file(argv[0]) if __name__ == "__main__": sys.exit(main())