#!/usr/bin/env python3 """Build the field battle-resource consumer boundary review. Movement and active-object trigger paths are now separated. This report checks the next nearby surface: direct calls into the resource command VM (`0x00423a2f`) and battle-like resource candidates. The goal is deliberately narrow: classify the known resource-runner consumers and keep them from being mistaken for a proven random field-encounter entry path. """ from __future__ import annotations import html import json import struct import sys 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 read_sections, va_to_offset # noqa: E402 RESOURCE_RUNNER = 0x00423A2F PROMOTION_STATUS = "field-battle-resource-consumer-boundary-route-blocked" def esc(value: Any) -> str: return html.escape("" if value is None else str(value), quote=True) def hx(value: int | None) -> str: return "" if value is None else f"0x{value:08x}" def read_json(path: Path, fallback: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: return fallback def write_json(path: Path, payload: Any) -> None: path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def read_bytes(exe: bytes, sections: list[dict[str, Any]], va: int, size: int) -> bytes: offset = va_to_offset(sections, va) if offset is None: return b"" return exe[offset : offset + size] def u32(exe: bytes, sections: list[dict[str, Any]], va: int) -> int | None: data = read_bytes(exe, sections, va, 4) if len(data) != 4: return None return struct.unpack(" list[int]: text = next(section for section in sections if section["name"] == ".text") blob = exe[text["raw"] : text["raw"] + text["raw_size"]] rows: list[int] = [] for offset in range(0, max(0, len(blob) - 4)): if blob[offset] != 0xE8: continue rel = struct.unpack_from(" list[dict[str, Any]]: global_roots = { row.get("globalVaHex"): row for row in resource_bridge.get("globalResourceRootRows") or [] if row.get("globalVaHex") } rows = [] for va in scan_call_refs(exe, sections, RESOURCE_RUNNER): info = dict(CALL_CLASSIFICATION.get(va) or {}) root_global = info.get("rootGlobalVaHex") root_row = global_roots.get(root_global) rows.append( { "callVa": va, "callVaHex": hx(va), "consumer": info.get("consumer") or "unclassified direct resource runner caller", "source": info.get("source") or "unknown", "rootGlobalVaHex": root_global, "rootVaHex": root_row.get("rootVaHex") if root_row else None, "rootFieldMaps": root_row.get("fieldMaps") if root_row else [], "rootLinkedCns": root_row.get("linkedCns") if root_row else [], "evidence": info.get("evidence") or "direct rel32 call to 0x00423a2f", "battleEntryProof": bool(info.get("battleEntryProof")), "codeBytesBefore": read_bytes(exe, sections, va - 12, 12).hex(" "), "codeBytesCall": read_bytes(exe, sections, va, 5).hex(" "), } ) return rows def save_selector_branch_rows(save_traces: list[dict[str, Any]]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for stream in save_traces: for item in stream.get("trace") or []: note = item.get("branchNote") or "" if "encounter-state" not in note and item.get("handlerVaHex") != "0x0040fa6d": continue rows.append( { "streamVaHex": stream.get("streamVaHex"), "source": stream.get("source"), "target": stream.get("target"), "step": item.get("step"), "vaHex": item.get("vaHex"), "opcodeHex": item.get("opcodeHex"), "handlerVaHex": item.get("handlerVaHex"), "branchTargetHex": item.get("branchTargetHex"), "fallthroughVaHex": item.get("fallthroughVaHex"), "note": note, } ) return rows def build_report() -> dict[str, Any]: exe = EXE.read_bytes() sections = read_sections(exe) resource_bridge = read_json(OUT / "resource_command_vm_bridge_review.json", {}) loader_trace = read_json(OUT / "resource_loader_consumer_trace_review.json", {}) field_static = read_json(OUT / "field_encounter_static_review.json", {}) step_boundary = read_json(OUT / "field_step_encounter_boundary_review.json", {}) active_boundary = read_json(OUT / "field_active_object_trigger_boundary_review.json", {}) battle_candidates = read_json(OUT / "battle_event_candidates.json", {}) save_traces = read_json(OUT / "save_selector_stream_traces.json", []) call_rows = direct_call_rows(exe, sections, resource_bridge) unclassified = [row for row in call_rows if row["consumer"].startswith("unclassified")] save_encounter_rows = save_selector_branch_rows(save_traces) battle_candidate_rows = battle_candidates.get("candidates") if isinstance(battle_candidates, dict) else None if battle_candidate_rows is None and isinstance(battle_candidates, list): battle_candidate_rows = battle_candidates battle_candidate_rows = battle_candidate_rows or [] field_summary = field_static.get("summary") or {} resource_summary = resource_bridge.get("summary") or {} loader_summary = loader_trace.get("summary") or {} step_summary = step_boundary.get("summary") or {} active_summary = active_boundary.get("summary") or {} return { "kind": "hwanse-field-battle-resource-boundary-review", "promotionStatus": PROMOTION_STATUS, "source": [ "out/resource_command_vm_bridge_review.json", "out/resource_loader_consumer_trace_review.json", "out/field_encounter_static_review.json", "out/field_step_encounter_boundary_review.json", "out/field_active_object_trigger_boundary_review.json", "out/battle_event_candidates.json", "out/save_selector_stream_traces.json", "tools/build_field_battle_resource_boundary_review.py", ], "summary": { "resourceRunnerHex": hx(RESOURCE_RUNNER), "directResourceRunnerCallCount": len(call_rows), "classifiedDirectResourceRunnerCallCount": len(call_rows) - len(unclassified), "unclassifiedDirectResourceRunnerCallCount": len(unclassified), "directResourceRunnerCallsHex": [row["callVaHex"] for row in call_rows], "directResourceRunnerBattleEntryProofCount": sum(1 for row in call_rows if row["battleEntryProof"]), "resourceOpcode10FieldMapRecordCount": resource_summary.get("fieldMapOpcode10RecordCount"), "resourceOpcode10UniqueMapCount": resource_summary.get("uniqueOpcode10MapCount"), "fieldMovementBattleCnsRefCount": step_summary.get("battleBackgroundCnsRefsInKnownMovementWindows"), "fieldMovementMonsterRefCount": step_summary.get("monsterOrBattleActorCnsRefsInKnownMovementWindows"), "fieldMovementCoreRngCount": step_summary.get("sharedRngCallsInCoreMovementWindows"), "activeTriggerRouteProofCount": active_summary.get("routeProofScriptCount"), "activeTriggerMapCnsOperandCount": active_summary.get("mapCnsOperandCandidateScriptCount"), "loaderRouteBindingProofFound": loader_summary.get("routeResourceLoaderBindingProofFound"), "rngDirectFieldEncounterEvidenceCount": field_summary.get("rngDirectFieldEncounterEvidenceCount"), "battleCandidateAdjacencyCount": len(battle_candidate_rows), "saveSelectorEncounterStateBranchHits": len(save_encounter_rows), "fieldEncounterConsumerPromoted": False, "battleEntryConsumerPromoted": False, "conclusion": ( "resource command VM 호출점 5개는 모두 분류됐다. " "고정 global root, stream immediate resource root, active object +0xec payload 경로로 나뉘며, " "현재 어느 것도 field step/trigger -> encounter RNG -> btl/monster package 선택을 증명하지 않는다." ), }, "decisions": [ { "id": "resource-runner-call-sites", "status": "grounded", "decision": "direct calls to 0x00423a2f are fully enumerated and classified.", "evidence": f"{len(call_rows)} direct calls; {len(unclassified)} unclassified", }, { "id": "opcode10-map-load", "status": "grounded-not-entry", "decision": "resource opcode 0x10 remains a map-load consumer, not the field encounter selector.", "evidence": f"{resource_summary.get('fieldMapOpcode10RecordCount')} opcode10 map records; selector producer still unresolved", }, { "id": "battle-adjacency-candidates", "status": "candidate-only", "decision": "battle-like event/resource candidates are kept as adjacency candidates.", "evidence": f"{len(battle_candidate_rows)} candidate rows, no direct field step entry proof", }, { "id": "save-selector-encounter-state-note", "status": "not-hit-in-current-traces", "decision": "the old encounter-state branch note is not a current proof row.", "evidence": f"{len(save_encounter_rows)} rows with handler 0x0040fa6d/encounter-state note in save selector traces", }, { "id": "field-battle-entry", "status": "blocked", "decision": "do not promote random field battle entry yet.", "evidence": "movement core has no RNG/btl/monster refs, active trigger bridge has no route/map CNS operands, and resource-runner callers are generic resource consumers", }, ], "directResourceRunnerCalls": call_rows, "saveSelectorEncounterStateRows": save_encounter_rows, "battleCandidateSample": battle_candidate_rows[:40], "nextFrontier": [ "Find a producer that selects a battle/formation resource package after successful walking steps.", "Separate event-driven battle resources from random field encounter formation tables.", "Search for monster formation IDs or btl_* selector values near step counter / per-field state, not inside already-closed movement mutation helpers.", ], } def render_html(report: dict[str, Any]) -> str: summary = report["summary"] cards = [ ("runner", summary["resourceRunnerHex"]), ("direct calls", summary["directResourceRunnerCallCount"]), ("entry proof", summary["directResourceRunnerBattleEntryProofCount"]), ("battle candidates", summary["battleCandidateAdjacencyCount"]), ("save branch hits", summary["saveSelectorEncounterStateBranchHits"]), ("status", report["promotionStatus"]), ] card_html = "".join( f"
{esc(label)}{esc(value)}
" for label, value in cards ) decision_rows = "".join( "" f"{esc(row['id'])}" f"{esc(row['status'])}" f"{esc(row['decision'])}" f"{esc(row['evidence'])}" "" for row in report["decisions"] ) call_rows = "".join( "" f"{esc(row['callVaHex'])}" f"{esc(row['consumer'])}" f"{esc(row['source'])}" f"{esc(row.get('rootGlobalVaHex') or '')}
{esc(row.get('rootVaHex') or '')}" f"{esc(row['battleEntryProof'])}" f"{esc(row['evidence'])}" "" for row in report["directResourceRunnerCalls"] ) save_rows = "".join( "" f"{esc(row['streamVaHex'])}" f"{esc(row['vaHex'])}" f"{esc(row['handlerVaHex'])}" f"{esc(row['note'])}" "" for row in report["saveSelectorEncounterStateRows"] ) or "current traces contain no encounter-state branch rows" frontier = "".join(f"
  • {esc(item)}
    • " for item in report["nextFrontier"]) payload = json.dumps(report, ensure_ascii=False) return f""" Field Battle Resource Boundary Review
    index field encounter static step boundary active trigger boundary resource reference

    Field Battle Resource Boundary Review

    {esc(summary["conclusion"])}

    {card_html}

    Decisions

    {decision_rows}
    idstatusdecisionevidence

    Direct Resource Runner Calls

    {call_rows}
    callconsumersourcerootentry proofevidence

    Save Selector “Encounter-state” Note

    {save_rows}
    streamrowhandlernote

    Next Frontier

      {frontier}

    Raw JSON

    
  
    

    



"""


def main() -> int:
    report = build_report()
    write_json(OUT / "field_battle_resource_boundary_review.json", report)
    html_text = render_html(report)
    (WEB / "field_battle_resource_boundary_review.html").write_text(html_text, encoding="utf-8")
    print("field_battle_resource_boundary_review ok")
    print(json.dumps(report["summary"], ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())