#!/usr/bin/env python3 """Consolidate the generic object/display VM stream producer frontier. The generic runner at 0x00402321 is now grounded in several separate reviews: manual movement overlap, object +0xec payloads, active-object delayed scripts, display/battle VM child streams, and save-selector/selected-root helpers. This report keeps those surfaces in one place and makes the boundary explicit: which paths write or feed object +0x40, which paths are only consumers/helpers, and whether any path currently promotes to a scene/map transition, encounter, or selected-root route producer. """ from __future__ import annotations import html import json import struct import sys from collections import Counter from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "out" WEB = ROOT / "web" EXE = ROOT / "Hwanse2.exe" sys.path.insert(0, str(ROOT / "tools")) from probe_exe_scene_tables import offset_to_va, read_sections, va_to_offset # noqa: E402 GENERIC_RUNNER_VA = 0x00402321 GENERIC_HANDLER_TABLE_VA = 0x00440538 NESTED_RUNNER_VA = 0x00402360 OBJECT_EC_PRODUCER_HANDLER_VA = 0x004079FF MANUAL_OVERLAP_BRIDGE_VA = 0x00431FBB ACTIVE_OBJECT_DELAYED_SCRIPT_ROUTE_VA = 0x00432FF0 def h(value: Any) -> str: return html.escape("" if value is None else str(value), quote=True) def hx(value: int | None) -> str | None: return None if value is None else f"0x{value:08x}" def load_json(path: Path, fallback: Any) -> Any: if not path.exists(): return fallback return json.loads(path.read_text(encoding="utf-8")) def summary_of(name: str) -> dict[str, Any]: data = load_json(OUT / name, {}) if not isinstance(data, dict): return {} summary = data.get("summary") return summary if isinstance(summary, dict) else data def section_for_offset(sections: list[dict[str, Any]], offset: int) -> dict[str, Any] | None: for section in sections: start = int(section["raw"]) end = start + int(section["raw_size"]) if start <= offset < end: return section return None def section_for_va(sections: list[dict[str, Any]], va: int) -> dict[str, Any] | None: for section in sections: start = int(section["va"]) end = start + int(section["raw_size"]) if start <= va < end: return section return None def find_direct_calls(exe: bytes, sections: list[dict[str, Any]], target_va: int) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] text = next((section for section in sections if section.get("name") == ".text"), None) if not text: return rows raw_start = int(text["raw"]) raw_end = raw_start + int(text["raw_size"]) text_va = int(text["va"]) data = exe[raw_start:raw_end] for index in range(0, max(0, len(data) - 5)): if data[index] != 0xE8: continue rel = struct.unpack_from(" list[dict[str, Any]]: needle = struct.pack(" str: offset = va_to_offset(sections, va) if offset is None: return "" return exe[offset : offset + size].hex(" ") def classify_call_site(call_va: int) -> str: known = { 0x0040237F: "nested-runner-wrapper-self-call", 0x004058AF: "opcode-0x2a-linked-object-script-fanout-restore", 0x004058D8: "opcode-0x2a-linked-object-script-fanout-persistent", 0x00407A98: "opcode-0x5e-transient-object-create", 0x004077ED: "generic-object-script-child/linked-runner", 0x00407891: "generic-object-script-child/linked-runner", 0x00432F2D: "active-object-script-update-loop", 0x004330A9: "active-object-delayed-script-route", 0x004330B3: "active-object-delayed-script-route", 0x00433185: "active-object-list-script-runner", 0x004331B1: "active-object-list-script-runner", } for va, label in known.items(): if abs(call_va - va) <= 8: return label if 0x00407000 <= call_va <= 0x00408050: return "generic-object-script-handler-family" if 0x00432F00 <= call_va <= 0x00433200: return "field-active-object-script-route" if 0x0040A000 <= call_va <= 0x0040F500: return "save-selector/generic-vm-handler-family" if 0x0041B000 <= call_va <= 0x0041F000: return "display-or-event-object-consumer-family" return "unclassified-direct-runner-call" def classify_pointer_ref(ref_va: int) -> str: if 0x00440538 <= ref_va <= 0x00440938: return "handler-table-entry-or-near-table" if 0x00442D00 <= ref_va <= 0x00442E80: return "descriptor/resource-root-table" if 0x00407000 <= ref_va <= 0x00408050: return "generic-object-script-handler-family" if 0x00430000 <= ref_va <= 0x00436000: return "field/object-runtime-code-ref" if 0x00500000 <= ref_va <= 0x005A0000: return "data-script-or-resource-pointer" return "unclassified-pointer-ref" def verified_snippets(exe: bytes, sections: list[dict[str, Any]]) -> list[dict[str, Any]]: snippets = [ { "id": "generic-runner-dispatch", "va": 0x00402327, "meaning": "reset stop flag, read object+0x40 stream opcode, dispatch through 0x00440538", "hex": "c7 05 b8 a1 55 00 00 00 00 00 83 3d b8 a1 55 00 00 0f 85 1d 00 00 00 8b 45 08 50 8b 45 08 8b 40 40 33 c9 8a 08 ff 14 8d 38 05 44 00", }, { "id": "object-ec-to-cursor-copy", "va": MANUAL_OVERLAP_BRIDGE_VA, "meaning": "manual overlap bridge copies object+0xec payload pointer into object+0x40", "hex": "33 c0 a0 33 45 57 00 85 c0 0f 85 14 00 00 00 8b 45 fc 8b 80 ec 00 00 00 8b 4d fc 89 41 40", }, { "id": "opcode-5e-cursor-from-payload", "va": 0x00407ABC, "meaning": "opcode 0x5e mode1 loads new object+0x40 from dword [payload+8]", "hex": "a1 a8 dd 59 00 89 45 08 8b 45 08 8b 80 ec 00 00 00 8b 40 08 8b 4d 08 89 41 40", }, { "id": "nested-runner-wrapper", "va": NESTED_RUNNER_VA + 0x09, "meaning": "0x00402360 saves current object+0x40, runs a supplied child stream through 0x00402321, then restores the parent cursor", "hex": "8b 45 08 8b 40 40 89 45 fc 8b 45 0c 8b 4d 08 89 41 40 8b 45 08 50 e8 9d ff ff ff 83 c4 04 8b 45 fc 8b 4d 08 89 41 40", }, { "id": "opcode-0x2a-fanout-list-select", "va": 0x004057EB, "meaning": "opcode 0x2a reads stream+1 as active-object list index and stream+2 as optional object+0x14 mask", "hex": "55 8b ec 83 ec 0c 53 56 57 8b 45 08 8b 40 40 33 c9 8a 48 01 83 e1 7f 89 4d f4", }, { "id": "opcode-0x2a-fanout-restore-child-cursor", "va": 0x00405893, "meaning": "opcode 0x2a high-bit branch saves target object+0x40, runs stream+4 through 0x00402321, then restores target object+0x40", "hex": "8b 45 f8 8b 40 40 89 45 fc 8b 45 08 8b 40 40 8b 40 04 8b 4d f8 89 41 40 8b 45 f8 50 e8 6d ca ff ff 83 c4 04 8b 45 fc 8b 4d f8 89 41 40", }, { "id": "opcode-0x2a-fanout-persistent-child-cursor", "va": 0x004058C5, "meaning": "opcode 0x2a normal branch writes stream+4 to target object+0x40 and keeps the advanced target cursor after 0x00402321", "hex": "8b 45 08 8b 40 40 8b 40 04 8b 4d f8 89 41 40 8b 45 f8 50 e8 44 ca ff ff 83 c4 04", }, { "id": "active-object-delayed-script-route", "va": 0x00433090, "meaning": "field active object timer route runs object+0x64 through 0x00402321 and stores the advanced cursor back to object+0x64", "hex": "8b 45 f8 8b 40 40 89 45 fc 8b 45 f8 8b 40 64 8b 4d f8 89 41 40 8b 45 f8 50 e8 73 f2 fc ff 83 c4 04 8b 45 f8 8b 40 40 8b 4d f8 89 41 64 8b 45 fc 8b 4d f8 89 41 40", }, ] rows: list[dict[str, Any]] = [] for row in snippets: expected = bytes.fromhex(str(row["hex"])) offset = va_to_offset(sections, int(row["va"])) actual = exe[offset : offset + len(expected)] if offset is not None else b"" rows.append( { "id": row["id"], "vaHex": hx(int(row["va"])), "meaning": row["meaning"], "matches": actual == expected, "expectedHex": row["hex"], "actualHex": actual.hex(" "), } ) return rows def build_report() -> dict[str, Any]: exe = EXE.read_bytes() sections = read_sections(exe) object_payload = summary_of("object_script_payload_producer.json") manual_bridge = summary_of("manual_movement_trigger_bridge.json") active_object = summary_of("field_active_object_trigger_boundary_review.json") map_tick = summary_of("map_animation_tick_route_review.json") field_tick = summary_of("field_encounter_tick_route_boundary_review.json") selected_writer = summary_of("selected_root_live_writer_frontier_review.json") opcode2a_parent = summary_of("generic_vm_opcode2a_parent_stream_review.json") dispatch_context = load_json(OUT / "save_selector_dispatch_table_context.json", {}) battle_vm = load_json(OUT / "battle_display_vm_static_decode.json", {}) direct_calls = find_direct_calls(exe, sections, GENERIC_RUNNER_VA) pointer_refs = find_dword_refs(exe, sections, GENERIC_RUNNER_VA) nested_calls = find_direct_calls(exe, sections, NESTED_RUNNER_VA) nested_refs = find_dword_refs(exe, sections, NESTED_RUNNER_VA) snippets = verified_snippets(exe, sections) surfaces = [ { "surface": "generic object/display VM runner", "status": "grounded-consumer", "producerPromoted": False, "evidence": "0x00402321 reads object+0x40 and dispatches opcode through 0x00440538.", "routeImpact": "This is a consumer/dispatcher. It does not itself select a map, scene root, or encounter formation.", "source": "Hwanse2.exe snippet generic-runner-dispatch", }, { "surface": "opcode 0x5e object +0xec payload producer", "status": "grounded-non-route-producer", "producerPromoted": False, "evidence": ( f"objectEcProducerFound={object_payload.get('objectEcProducerFound')}; " f"mode1 candidates={object_payload.get('mode1CandidateCount')}; " f"map CNS refs in payload heads={object_payload.get('mapCnsRefCountInPayloadHeads')}" ), "routeImpact": "Creates/transitions transient display/object VM payloads, but current static payload scan does not prove field-map transition payloads.", "source": "out/object_script_payload_producer.json", }, { "surface": "manual movement overlap bridge", "status": "grounded-trigger-consumer", "producerPromoted": False, "evidence": ( f"bridgeFound={manual_bridge.get('manualMovementToObjectScriptBridgeFound')}; " f"routeProof={manual_bridge.get('routeProofFound')}; " f"manual opcodes={manual_bridge.get('manualScriptOpcodeCount')}" ), "routeImpact": "Object overlap can copy object+0xec into object+0x40, but strict inventory found no concrete map transition target.", "source": "out/manual_movement_trigger_bridge.json", }, { "surface": "active object delayed script route", "status": "grounded-indirect-candidate", "producerPromoted": False, "evidence": ( f"tick route={map_tick.get('fieldTickRouteVaHex')}; " f"active route grounded={map_tick.get('activeObjectDelayedScriptRouteGrounded')}; " f"visible motion producer={map_tick.get('visibleMotionProducerFound')}" ), "routeImpact": "Can run object+0x64/+0x40 scripts from the field tick, but no animated-map tile-write root is bound yet.", "source": "out/map_animation_tick_route_review.json", }, { "surface": "opcode 0x2a linked-object script fanout", "status": "grounded-fanout-primitive", "producerPromoted": False, "evidence": ( "handler table 0x004405e0 -> 0x004057eb; stream+1 selects active-object list slot, " "stream+2 optionally filters target object+0x14, stream+4 supplies the child script; " "direct runner calls at 0x004058af and 0x004058d8; " f"active +0xec hits={opcode2a_parent.get('activeObjectEcOpcode2aCount', 0)}, " f"raw candidates={opcode2a_parent.get('rawCandidateCount', 0)}, " f"boundary-shaped={opcode2a_parent.get('boundaryCandidateCount', 0)}, " f"promoted={opcode2a_parent.get('promotedParentProducerCount', 0)}" ), "routeImpact": "Executes a supplied child stream on linked active objects. This narrows object script propagation, but it is not a root producer without a parent stream source.", "source": "Hwanse2.exe opcode 0x2a handler + out/generic_vm_opcode2a_parent_stream_review.json", }, { "surface": "field active object script inventory", "status": "negative-route-inventory", "producerPromoted": False, "evidence": ( f"strict initializers={active_object.get('strictInitializerCount')}; " f"unique +ec scripts={active_object.get('uniqueObjectEcScriptCount')}; " f"route proof scripts={active_object.get('routeProofScriptCount')}; " f"map CNS operand scripts={active_object.get('mapCnsOperandCandidateScriptCount')}" ), "routeImpact": "Large active-object script inventory exists, but no route/map CNS operand proof is promoted.", "source": "out/field_active_object_trigger_boundary_review.json", }, { "surface": "field encounter tick route", "status": "negative-encounter-producer", "producerPromoted": False, "evidence": ( f"known field windows={field_tick.get('knownFieldWindowCount')}; " f"promotable calls={field_tick.get('promotableKeyDirectCallsInsideKnownFieldWindows')}; " f"formation exact refs={field_tick.get('formationCandidateExactPointerRefCount')}" ), "routeImpact": "No field walking/tick route currently selects battle resources or formation streams.", "source": "out/field_encounter_tick_route_boundary_review.json", }, { "surface": "selected-root stream primitives", "status": "grounded-separated-global-stream", "producerPromoted": False, "evidence": ( f"selector writer grounded={selected_writer.get('selectorByteWriterGrounded')}; " f"selectedRoot writer grounded={selected_writer.get('selectedRootWriterGrounded')}; " f"executor found={selected_writer.get('selectedRootExecutorFound')}; " f"save-selector runtime dispatch proof={dispatch_context.get('saveSelectorSliceDirectRuntimeDispatchProofFound')}" ), "routeImpact": "Selected-root VM primitives use global selectedRoot/current stream flow, not a proven object+0x40 route producer.", "source": "out/selected_root_live_writer_frontier_review.json + out/save_selector_dispatch_table_context.json", }, { "surface": "battle/display VM child streams", "status": "grounded-battle-display-domain", "producerPromoted": False, "evidence": ( f"status={battle_vm.get('status')}; " f"display runner={(battle_vm.get('displayVmRunner') or {}).get('runnerVaHex')}" ), "routeImpact": "Useful for battle/effect rendering, but not a field scene/route producer.", "source": "out/battle_display_vm_static_decode.json", }, ] direct_call_class_counts = dict(Counter(row["classification"] for row in direct_calls)) summary = { "genericRunnerVaHex": hx(GENERIC_RUNNER_VA), "genericHandlerTableVaHex": hx(GENERIC_HANDLER_TABLE_VA), "nestedRunnerVaHex": hx(NESTED_RUNNER_VA), "directCallCount": len(direct_calls), "directCallClassificationCounts": direct_call_class_counts, "unclassifiedDirectCallCount": direct_call_class_counts.get("unclassified-direct-runner-call", 0), "linkedObjectFanoutCallCount": sum( 1 for row in direct_calls if str(row["classification"]).startswith("opcode-0x2a-") ), "opcode2aActiveObjectEcHitCount": opcode2a_parent.get("activeObjectEcOpcode2aCount", 0), "opcode2aRawCandidateCount": opcode2a_parent.get("rawCandidateCount", 0), "opcode2aBoundaryCandidateCount": opcode2a_parent.get("boundaryCandidateCount", 0), "opcode2aPromotedParentProducerCount": opcode2a_parent.get("promotedParentProducerCount", 0), "pointerRefCount": len(pointer_refs), "nestedRunnerDirectCallCount": len(nested_calls), "nestedRunnerPointerRefCount": len(nested_refs), "verifiedSnippetCount": sum(1 for row in snippets if row["matches"]), "verifiedSnippetTotal": len(snippets), "surfaceCount": len(surfaces), "promotedProducerSurfaceCount": sum(1 for row in surfaces if row["producerPromoted"]), "manualMovementBridgeFound": manual_bridge.get("manualMovementToObjectScriptBridgeFound"), "objectEcProducerFound": object_payload.get("objectEcProducerFound"), "activeObjectDelayedScriptRouteGrounded": map_tick.get("activeObjectDelayedScriptRouteGrounded"), "selectedRootPrimitiveGrounded": ( selected_writer.get("selectorByteWriterGrounded") is True and selected_writer.get("selectedRootWriterGrounded") is True and selected_writer.get("selectedRootExecutorFound") is True ), "routeProducerPromoted": False, "decision": ( "The generic VM stream frontier is now consolidated. 0x00402321/object+0x40 is a grounded " "dispatcher; all direct calls are now classified, including the nested runner wrapper, opcode " "0x2a linked-object fanout, and active-object delayed scripts. Current surfaces remain support/" "display/object domains or negative route inventories. None promotes to a concrete scene route, " "map transition, field encounter, or selected-root gameplay producer." ), } return { "kind": "hwanse-generic-vm-stream-producer-frontier-review", "source": "tools/build_generic_vm_stream_producer_frontier_review.py", "status": "generic-vm-stream-frontier-consolidated-no-route-producer-promoted", "summary": summary, "verifiedSnippets": snippets, "directCallRows": direct_calls, "pointerRefRows": pointer_refs, "nestedRunnerDirectCallRows": nested_calls, "nestedRunnerPointerRefRows": nested_refs, "surfaces": surfaces, "nextFrontier": [ "Do not re-open 0x00402321 itself as an unknown; treat it as a grounded generic object/display VM dispatcher.", "Do not treat opcode 0x2a as a route producer by itself; it is a linked-object script fanout that still needs a parent stream source.", "Opcode 0x2a is absent from the strict active object +0xec script inventory; raw data-section candidates remain review-only until a gameplay parent/root source is found.", "For scene route, look above selected-root writer/executor primitives for the live root/stream producer.", "For map transition, require an active-object script/root that binds a concrete map target or map resource, not just object+0xec -> +0x40.", "For encounter, require field walking/tick state to bridge into battle resource/formation consumers.", "For map animation, require an animated-map object/root that reaches tile-write handlers from the tick route.", ], "sourceArtifacts": [ "out/object_script_payload_producer.json", "out/manual_movement_trigger_bridge.json", "out/field_active_object_trigger_boundary_review.json", "out/generic_vm_opcode2a_parent_stream_review.json", "out/map_animation_tick_route_review.json", "out/field_encounter_tick_route_boundary_review.json", "out/selected_root_live_writer_frontier_review.json", "out/save_selector_dispatch_table_context.json", "out/battle_display_vm_static_decode.json", ], } def render_html(report: dict[str, Any]) -> str: summary = report["summary"] cards = [ ("status", report["status"]), ("runner", summary["genericRunnerVaHex"]), ("direct calls", summary["directCallCount"]), ("unclassified direct", summary["unclassifiedDirectCallCount"]), ("opcode 0x2a fanout calls", summary["linkedObjectFanoutCallCount"]), ("0x2a active +ec hits", summary["opcode2aActiveObjectEcHitCount"]), ("0x2a raw candidates", summary["opcode2aRawCandidateCount"]), ("pointer refs", summary["pointerRefCount"]), ("snippets", f"{summary['verifiedSnippetCount']}/{summary['verifiedSnippetTotal']}"), ("promoted producers", summary["promotedProducerSurfaceCount"]), ] card_html = "".join(f"
{h(k)}{h(v)}
" for k, v in cards) surface_rows = "".join( "" f"{h(row['surface'])}" f"{h(row['status'])}" f"{h(row['producerPromoted'])}" f"{h(row['evidence'])}" f"{h(row['routeImpact'])}" f"{h(row['source'])}" "" for row in report["surfaces"] ) call_rows = "".join( "" f"{h(row['callVaHex'])}" f"{h(row['classification'])}" f"{h(row['contextHex'])}" "" for row in report["directCallRows"] ) or "none" ref_rows = "".join( "" f"{h(row['refVaHex'])}" f"{h(row['section'])}" f"{h(row['classification'])}" "" for row in report["pointerRefRows"] ) or "none" snippet_rows = "".join( "" f"{h(row['id'])}
{h(row['vaHex'])}" f"{h(row['matches'])}" f"{h(row['meaning'])}" f"{h(row['actualHex'])}" "" for row in report["verifiedSnippets"] ) frontier = "".join(f"
  • {h(item)}
    • " for item in report["nextFrontier"]) artifacts = "".join(f"
  • {h(item)}
    • " for item in report["sourceArtifacts"]) payload = json.dumps(report, ensure_ascii=False) return f""" Generic VM Stream Producer Frontier
    index remaining work dispatch table context mode/state frontier opcode 0x2a parent streams

    Generic VM Stream Producer Frontier

    {h(summary["decision"])}

    {card_html}

    Surfaces

    {surface_rows}
    surfacestatusproducer promotedevidenceroute impactsource

    Verified Snippets

    {snippet_rows}
    snippetmatchesmeaningactual bytes

    Direct Calls to 0x00402321

    {call_rows}
    callclassificationcontext

    Dword Refs to 0x00402321

    {ref_rows}
    refsectionclassification

    Next Frontier

      {frontier}

    Source Artifacts

      {artifacts}

    Raw JSON

    
  
    

    




"""


def main() -> None:
    report = build_report()
    OUT.mkdir(exist_ok=True)
    WEB.mkdir(exist_ok=True)
    (OUT / "generic_vm_stream_producer_frontier_review.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    (WEB / "generic_vm_stream_producer_frontier_review.html").write_text(
        render_html(report),
        encoding="utf-8",
    )
    print(
        "wrote generic VM stream producer frontier "
        f"({report['summary']['promotedProducerSurfaceCount']} promoted producers)"
    )


if __name__ == "__main__":
    main()