re: diff_optrace.py — engine-vs-VM offset-path divergence oracle
Pure first_divergence + pick_scene_codebase (longest-common-prefix codebase identification), unit-tested (test_diff_optrace.py, 4/4). CLI loads the engine jsonl + VM json, isolates the scene's codebase, and reports the first divergence with the mis-modeled instruction and +/-3 ops of context on each side (opcode + rendered line via sys4load). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gamer147
181
tools/diff_optrace.py
Normal file
181
tools/diff_optrace.py
Normal file
@@ -0,0 +1,181 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Differential offset-path oracle (docs/engine-re.md): diff the real engine's executed script-offset
|
||||
path against our C# VM's, on the deterministic opening, and report the FIRST divergence — the exact
|
||||
branch/opcode we modeled wrong.
|
||||
|
||||
Both run the same bytecode, so the opcode at each script offset is static (from disassembly); we diff the
|
||||
*sequence of executed offsets* (control flow), not opcodes or effects. A path divergence is a jcc that
|
||||
went the other way, an opcode length we got wrong, or a native context (coroutine) the VM doesn't model.
|
||||
|
||||
Inputs (both under build/, disposable):
|
||||
build/engine-optrace.jsonl — {"codebase":int,"offset":int} per executed op, in order (trace_engine_ops.py)
|
||||
build/vm-optrace.json — {"scene":"SC0000","offsets":[int,...]} (Age.Cli trace --trace-json)
|
||||
|
||||
py -3.11 -X utf8 tools/diff_optrace.py SC0000 [--engine PATH] [--vm PATH]
|
||||
|
||||
The pure core (first_divergence / pick_scene_codebase) is unit-tested in tools/test_diff_optrace.py.
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import paths # noqa: E402 (sibling import; tools dir is on sys.path when run directly)
|
||||
|
||||
ENGINE_TRACE = paths.BUILD / "engine-optrace.jsonl"
|
||||
VM_TRACE = paths.BUILD / "vm-optrace.json"
|
||||
|
||||
|
||||
# ---- pure core (unit-tested) -------------------------------------------------
|
||||
|
||||
def first_divergence(a: list, b: list) -> dict:
|
||||
"""First index where offset sequences `a` (engine) and `b` (VM) differ.
|
||||
|
||||
Returns {"agreed": n, "index": i, "a": a[i]|None, "b": b[i]|None}, or
|
||||
{"agreed": len, "index": None} when one is a prefix of the other / they are equal.
|
||||
When they agree over the whole common prefix but differ in length, the shorter ended early: the
|
||||
divergence index is that common length, with the missing side reported as None."""
|
||||
n = min(len(a), len(b))
|
||||
for i in range(n):
|
||||
if a[i] != b[i]:
|
||||
return {"agreed": i, "index": i, "a": a[i], "b": b[i]}
|
||||
if len(a) == len(b):
|
||||
return {"agreed": n, "index": None, "a": None, "b": None}
|
||||
return {"agreed": n, "index": n,
|
||||
"a": a[n] if n < len(a) else None,
|
||||
"b": b[n] if n < len(b) else None}
|
||||
|
||||
|
||||
def pick_scene_codebase(entries: list, vm_offsets: list):
|
||||
"""The codebase whose in-order offset sequence shares the longest common prefix with `vm_offsets`
|
||||
— identifies which loaded-script instance in the engine trace is the scene we ran in the VM. Returns
|
||||
the codebase id, or None if no codebase shares even one leading offset (nothing matches)."""
|
||||
seqs, order = {}, []
|
||||
for e in entries:
|
||||
cb = e["codebase"]
|
||||
if cb not in seqs:
|
||||
seqs[cb] = []
|
||||
order.append(cb)
|
||||
seqs[cb].append(e["offset"])
|
||||
|
||||
def lcp(seq):
|
||||
n = 0
|
||||
for x, y in zip(seq, vm_offsets):
|
||||
if x != y:
|
||||
break
|
||||
n += 1
|
||||
return n
|
||||
|
||||
best, best_len = None, 0
|
||||
for cb in order:
|
||||
length = lcp(seqs[cb])
|
||||
if length > best_len:
|
||||
best, best_len = cb, length
|
||||
return best
|
||||
|
||||
|
||||
# ---- CLI: load traces, decode opcode context, report -------------------------
|
||||
|
||||
def _load_engine(path: Path) -> list:
|
||||
entries = []
|
||||
with open(path, encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if line:
|
||||
entries.append(json.loads(line))
|
||||
return entries
|
||||
|
||||
|
||||
def _load_scene_disasm(scene: str):
|
||||
"""offset -> (opcode:int, rendered_line:str) for the scene, straight from sys4load (authoritative)."""
|
||||
import sys4load
|
||||
key = scene.upper()
|
||||
if not key.endswith(".BIN"):
|
||||
key += ".BIN"
|
||||
scr = sys4load.load(paths.scripts()[key])
|
||||
sys4load.decode_code(scr)
|
||||
op_by_off = {ins.offset: ins.opcode for ins in scr.instructions}
|
||||
line_by_off = {}
|
||||
for line in sys4load.render_listing(scr).splitlines():
|
||||
m = re.match(r"\s*0x([0-9a-fA-F]+):\s*(.*)", line)
|
||||
if m:
|
||||
line_by_off[int(m.group(1), 16)] = m.group(2).rstrip()
|
||||
return op_by_off, line_by_off
|
||||
|
||||
|
||||
def _fmt(off, op_by_off, line_by_off) -> str:
|
||||
if off is None:
|
||||
return "(none — trace ended)"
|
||||
op = op_by_off.get(off)
|
||||
line = line_by_off.get(off, "?")
|
||||
return f"0x{off:05x}: {line}" + (f" [op 0x{op:x}]" if op is not None else " [op ?]")
|
||||
|
||||
|
||||
def report(scene, entries, vm_offsets):
|
||||
cb = pick_scene_codebase(entries, vm_offsets)
|
||||
if cb is None:
|
||||
print(f"[!] could not identify {scene}'s codebase in the engine trace "
|
||||
f"({len(entries)} entries, {len({e['codebase'] for e in entries})} codebases) — "
|
||||
f"no shared leading offset with the VM trace.")
|
||||
return 2
|
||||
engine_seq = [e["offset"] for e in entries if e["codebase"] == cb]
|
||||
op_by_off, line_by_off = _load_scene_disasm(scene)
|
||||
|
||||
d = first_divergence(engine_seq, vm_offsets)
|
||||
print(f"=== differential offset-path oracle: {scene} ===")
|
||||
print(f"engine codebase 0x{cb:x}: {len(engine_seq)} executed offsets "
|
||||
f"(of {len(entries)} total across {len({e['codebase'] for e in entries})} codebases)")
|
||||
print(f"VM: {len(vm_offsets)} executed offsets")
|
||||
|
||||
if d["index"] is None:
|
||||
print(f"\nNO DIVERGENCE over {d['agreed']} steps — the VM path matches the engine exactly. ✓")
|
||||
return 0
|
||||
|
||||
i = d["index"]
|
||||
print(f"\nAGREED {d['agreed']} steps, then DIVERGE at index {i}:")
|
||||
if i > 0:
|
||||
shared = engine_seq[i - 1]
|
||||
print(f" last agreed op {_fmt(shared, op_by_off, line_by_off)}")
|
||||
print(f" ^ this instruction's successor differs — the mis-modeled branch/op")
|
||||
print(f" engine went -> {_fmt(d['a'], op_by_off, line_by_off)}")
|
||||
print(f" VM went -> {_fmt(d['b'], op_by_off, line_by_off)}")
|
||||
|
||||
def ctx(seq, label):
|
||||
lo, hi = max(0, i - 3), min(len(seq), i + 4)
|
||||
print(f"\n {label} context [{lo}..{hi}):")
|
||||
for j in range(lo, hi):
|
||||
mark = " <-- diverge" if j == i else (" (last agreed)" if j == i - 1 else "")
|
||||
print(f" [{j}] {_fmt(seq[j], op_by_off, line_by_off)}{mark}")
|
||||
|
||||
ctx(engine_seq, "engine")
|
||||
ctx(vm_offsets, "VM")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
argv = list(sys.argv[1:] if argv is None else argv)
|
||||
scene = next((a for a in argv if not a.startswith("-")), "SC0000")
|
||||
|
||||
def opt(flag, default):
|
||||
return argv[argv.index(flag) + 1] if flag in argv and argv.index(flag) + 1 < len(argv) else default
|
||||
|
||||
engine_path = Path(opt("--engine", ENGINE_TRACE))
|
||||
vm_path = Path(opt("--vm", VM_TRACE))
|
||||
|
||||
if not engine_path.exists():
|
||||
print(f"[!] engine trace not found: {engine_path}\n Capture it first: "
|
||||
f"py -3.11 -u -X utf8 tools/frida/trace_engine_ops.py --hook tick")
|
||||
return 1
|
||||
if not vm_path.exists():
|
||||
print(f"[!] VM trace not found: {vm_path}\n Produce it first: "
|
||||
f"dotnet run --project engine/Age.Cli -- trace {scene}.BIN --boot --trace-json {vm_path}")
|
||||
return 1
|
||||
|
||||
entries = _load_engine(engine_path)
|
||||
vm_offsets = json.loads(vm_path.read_text(encoding="utf-8"))["offsets"]
|
||||
return report(scene, entries, vm_offsets)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
paths.add_self_to_syspath()
|
||||
sys.exit(main())
|
||||
28
tools/test_diff_optrace.py
Normal file
28
tools/test_diff_optrace.py
Normal file
@@ -0,0 +1,28 @@
|
||||
# tools/test_diff_optrace.py (plain runner)
|
||||
import sys
|
||||
from diff_optrace import first_divergence, pick_scene_codebase
|
||||
FAILS=[]
|
||||
def check(c,m): (FAILS.append(m) or print("FAIL:",m)) if not c else print("ok:",m)
|
||||
|
||||
def test_equal_no_divergence():
|
||||
r = first_divergence([0,1,2,3],[0,1,2,3])
|
||||
check(r["index"] is None and r["agreed"]==4, "equal traces -> no divergence")
|
||||
|
||||
def test_first_divergence_point():
|
||||
r = first_divergence([0,1,2,9],[0,1,2,3])
|
||||
check(r["index"]==3 and r["a"]==9 and r["b"]==3, "divergence at first differing offset")
|
||||
|
||||
def test_prefix_shorter_vm():
|
||||
r = first_divergence([0,1,2,3],[0,1]) # vm ends early
|
||||
check(r["index"]==2 and r["b"] is None and r["agreed"]==2, "shorter VM trace flagged at end")
|
||||
|
||||
def test_pick_codebase_by_longest_common_prefix():
|
||||
entries=[{"codebase":100,"offset":0},{"codebase":100,"offset":5}, # cb100: [0,5,...]
|
||||
{"codebase":200,"offset":0},{"codebase":200,"offset":1},{"codebase":200,"offset":2}]
|
||||
check(pick_scene_codebase(entries,[0,1,2])==200, "codebase matching VM prefix chosen")
|
||||
|
||||
def main():
|
||||
test_equal_no_divergence(); test_first_divergence_point(); test_prefix_shorter_vm()
|
||||
test_pick_codebase_by_longest_common_prefix()
|
||||
print("FAILURES:",len(FAILS)); return 1 if FAILS else 0
|
||||
if __name__=="__main__": sys.exit(main())
|
||||
Reference in New Issue
Block a user