chore: initialize age-reimpl repo
Reverse-engineering + open reimplementation workspace for Eushully's AGE/SYS4 engine (first target: Himegari). The repo root is age-reimpl/; the original game install and the extracted ALF data are siblings outside the repo and are never tracked. build/ (derived corpora) is gitignored and regenerated by the tools. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gamer147
425
tools/vm0.py
Normal file
425
tools/vm0.py
Normal file
@@ -0,0 +1,425 @@
|
||||
#!/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).
|
||||
* <file> 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 <op> 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 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])
|
||||
return run_file(argv[0])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user