#!/usr/bin/env python3 """Review producers/consumers of the active group gate used before HUD menus. The stream at 0x0047e688 gates the grounded top-menu wrapper with two VM commands: * 0x6e compares command bytes with globals 0x00574544/0x00574543. * 0xe6 branches on the active descriptor count at 0x004576e8. This pass intentionally stays below "opener proof". It asks whether those globals are a normal-field ESC/X opener signal, or a more general active object/descriptor control surface that the menu wrapper also uses. """ from __future__ import annotations import html import json import struct from pathlib import Path from typing import Any from probe_exe_scene_tables import offset_to_va, read_sections, va_to_offset from summarize_object_payload_442c75_callers import decode_stream ROOT = Path(__file__).resolve().parents[1] EXE = ROOT / "Hwanse2.exe" OUT = ROOT / "out" WEB = ROOT / "web" ACTIVE_GROUP_A_VA = 0x00574544 ACTIVE_GROUP_B_VA = 0x00574543 ACTIVE_DESCRIPTOR_COUNT_VA = 0x004576E8 ACTIVE_DESCRIPTOR_ORDER_VA = 0x004576E9 ACTIVE_DESCRIPTOR_SLOT_TABLE_VA = 0x00457750 GROUP_STORE_HANDLER_VA = 0x0040904A GROUP_BRANCH_HANDLER_VA = 0x00409078 COUNT_BRANCH_HANDLER_VA = 0x00410539 FIELD_ACTOR_CONTROLLER_VA = 0x0043022D FIELD_COLLISION_RESPONSE_VA = 0x004319F8 DESCRIPTOR_REBUILD_VA = 0x00432323 DISPLAY_VM_ROOT_VA = 0x00402321 TEXT_STATUS_HANDLER_VA = 0x0041B66D TOP_MENU_PRE_WRAPPER_GATE_VA = 0x0047E688 TOP_MENU_REJECT_LOOP_VA = 0x0047E67C TOP_MENU_WRAPPER_VA = 0x0047E6A0 TOP_MENU_SEQUENCE_VA = 0x004DDC6C def hx(value: int | None, width: int = 8) -> str: return "-" if value is None else f"0x{value:0{width}x}" def h(value: Any) -> str: return html.escape("" if value is None else str(value), quote=True) def read_at(exe: bytes, sections: list[dict[str, Any]], va: int, size: int) -> bytes: offset = va_to_offset(sections, va) if offset is None: raise ValueError(f"VA {hx(va)} is outside raw sections") return exe[offset: offset + size] def section_name_for(sections: list[dict[str, Any]], va: int) -> str: for section in sections: start = int(section["va"]) end = start + int(section["raw_size"]) if start <= va < end: return str(section["name"]) return "" def all_refs(exe: bytes, sections: list[dict[str, Any]], value: int) -> list[int]: needle = struct.pack(" list[int]: refs: list[int] = [] needle = struct.pack(" list[int]: hits: list[int] = [] for section in sections: if section.get("name") != ".text": continue start = int(section["raw"]) end = start + int(section["raw_size"]) for off in range(start, max(start, end - 5)): if exe[off] != 0xE8: continue call_va = offset_to_va(sections, off) if call_va is None: continue rel = struct.unpack_from(" list[dict[str, Any]]: text = next(section for section in sections if section.get("name") == ".text") raw_start = int(text["raw"]) raw_end = raw_start + int(text["raw_size"]) starts: list[int] = [] pos = raw_start while True: hit = exe.find(b"\x55\x8b\xec", pos, raw_end) if hit < 0: break va = offset_to_va(sections, hit) if va is not None: starts.append(va) pos = hit + 1 starts = sorted(set(starts)) rows: list[dict[str, Any]] = [] for index, start_va in enumerate(starts): start_off = va_to_offset(sections, start_va) if start_off is None: continue next_va = starts[index + 1] if index + 1 < len(starts) else int(text["va"]) + int(text["raw_size"]) next_off = va_to_offset(sections, next_va) if index + 1 < len(starts) else raw_end if next_off is None: next_off = raw_end rows.append( { "startVa": start_va, "endVa": next_va, "startOff": start_off, "endOff": max(start_off, next_off), "size": max(0, next_off - start_off), } ) return rows def owner_function(functions: list[dict[str, Any]], va: int) -> dict[str, Any] | None: for function in functions: if int(function["startVa"]) <= va < int(function["endVa"]): return function return None def classify_function(start_va: int) -> str: if start_va == GROUP_STORE_HANDLER_VA: return "VM opcode 0x6d handler" if start_va == GROUP_BRANCH_HANDLER_VA: return "VM opcode 0x6e handler" if start_va == COUNT_BRANCH_HANDLER_VA: return "VM opcode 0xe6 handler" if start_va == FIELD_ACTOR_CONTROLLER_VA: return "field actor input/collision controller" if start_va == FIELD_COLLISION_RESPONSE_VA: return "field collision response helper" if start_va == DESCRIPTOR_REBUILD_VA: return "active descriptor rebuild/materializer" if start_va == DISPLAY_VM_ROOT_VA: return "generic display VM root" if start_va == TEXT_STATUS_HANDLER_VA: return "text/status display handler" if 0x00430000 <= start_va <= 0x00432FFF: return "field active-object/descriptor subsystem" if 0x0040A000 <= start_va <= 0x00410FFF: return "generic VM handler layer" return "unclassified" def classify_ref_instruction(exe: bytes, sections: list[dict[str, Any]], ref_va: int) -> str: off = va_to_offset(sections, ref_va) if off is None: return "unknown" data = exe checks = [ (2, b"\xc6\x05", "direct byte store imm8"), (2, b"\xc7\x05", "direct dword store imm32"), (1, b"\xa0", "read byte into AL"), (1, b"\xa2", "write AL"), (2, b"\x66\xa1", "read word into AX"), (2, b"\x66\xa3", "write AX"), (2, b"\x8a\x88", "indexed byte read"), (2, b"\x66\x8b", "word read/indexed operand"), (3, b"\x66\x8b\x0c", "indexed word read"), (3, b"\x8b\x04\x8d", "indexed dword read"), ] for back, prefix, label in checks: start = off - back if start >= 0 and data[start:start + len(prefix)] == prefix: return label return "raw dword reference" def context_bytes(exe: bytes, sections: list[dict[str, Any]], va: int, before: int = 8, after: int = 12) -> str: off = va_to_offset(sections, va) if off is None: return "" start = max(0, off - before) end = min(len(exe), off + after) return exe[start:end].hex(" ") def ref_rows(exe: bytes, sections: list[dict[str, Any]], functions: list[dict[str, Any]], target_va: int) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for ref in text_refs(exe, sections, target_va): owner = owner_function(functions, ref) owner_start = int(owner["startVa"]) if owner else None rows.append( { "refVa": ref, "refVaHex": hx(ref), "targetVaHex": hx(target_va), "ownerStartVa": owner_start, "ownerStartVaHex": hx(owner_start), "ownerClass": classify_function(owner_start) if owner_start is not None else "", "instructionClass": classify_ref_instruction(exe, sections, ref), "section": section_name_for(sections, ref), "contextBytes": context_bytes(exe, sections, ref), } ) return rows def command_digest(exe: bytes, sections: list[dict[str, Any]], start_va: int, max_commands: int = 12) -> dict[str, Any]: decoded = decode_stream(exe, sections, start_va, max_commands=max_commands, max_bytes=0x120) return { "startVaHex": hx(start_va), "decodedCommandCount": decoded.get("decodedCommandCount", 0), "commands": [ { "vaHex": row.get("vaHex"), "opcodeHex": row.get("opcodeHex"), "opcodeName": row.get("opcodeName"), "summary": row.get("summary"), "rawHex": row.get("rawHex"), } for row in decoded.get("commands", []) ], } def build_report() -> dict[str, Any]: exe = EXE.read_bytes() sections = read_sections(exe) functions = function_ranges(exe, sections) group_a_rows = ref_rows(exe, sections, functions, ACTIVE_GROUP_A_VA) group_b_rows = ref_rows(exe, sections, functions, ACTIVE_GROUP_B_VA) count_rows = ref_rows(exe, sections, functions, ACTIVE_DESCRIPTOR_COUNT_VA) direct_group_writes = [ row for row in group_a_rows + group_b_rows if "store" in row["instructionClass"] or row["instructionClass"] == "write AL" ] descriptor_count_direct_writes = [ row for row in count_rows if "store" in row["instructionClass"] or row["instructionClass"] == "write AL" ] field_group_consumers = [ row for row in group_a_rows + group_b_rows if row["ownerClass"] in { "field actor input/collision controller", "field active-object/descriptor subsystem", "field collision response helper", } ] descriptor_rebuild_context = command_digest(exe, sections, TOP_MENU_PRE_WRAPPER_GATE_VA, max_commands=8) descriptor_rebuild_calls = call_sites(exe, sections, DESCRIPTOR_REBUILD_VA) display_vm_root_calls = call_sites(exe, sections, DISPLAY_VM_ROOT_VA) summary = { "activeGroupARefCount": len(group_a_rows), "activeGroupBRefCount": len(group_b_rows), "activeDescriptorCountRefCount": len(count_rows), "directGroupWriteCount": len(direct_group_writes), "descriptorCountDirectWriteCount": len(descriptor_count_direct_writes), "fieldGroupConsumerCount": len(field_group_consumers), "descriptorRebuildCallCount": len(descriptor_rebuild_calls), "displayVmRootCallCount": len(display_vm_root_calls), "topMenuGateVaHex": hx(TOP_MENU_PRE_WRAPPER_GATE_VA), "topMenuRejectLoopVaHex": hx(TOP_MENU_REJECT_LOOP_VA), "topMenuWrapperVaHex": hx(TOP_MENU_WRAPPER_VA), "topMenuSequenceVaHex": hx(TOP_MENU_SEQUENCE_VA), "normalFieldEscOpenerPromoted": False, "classification": "active-group/descriptor gate grounded; opener producer still unproven", "decision": ( "0x574544/0x574543 are not a narrow ESC/X opener flag. They are written by the " "0x6d VM handler and by descriptor/display setup code, and consumed heavily by the " "field actor/collision subsystem. 0x4576e8 is the active descriptor count. The " "0x0047e688 gate is therefore a grounded active-object/display gate before the top " "menu wrapper, but it does not by itself identify the normal-field menu opener." ), } report = { "version": 1, "kind": "hwanse-hud-active-group-gate-frontier-review", "summary": summary, "addresses": { "activeGroupA": hx(ACTIVE_GROUP_A_VA), "activeGroupB": hx(ACTIVE_GROUP_B_VA), "activeDescriptorCount": hx(ACTIVE_DESCRIPTOR_COUNT_VA), "activeDescriptorOrder": hx(ACTIVE_DESCRIPTOR_ORDER_VA), "activeDescriptorSlotTable": hx(ACTIVE_DESCRIPTOR_SLOT_TABLE_VA), }, "preWrapperGate": descriptor_rebuild_context, "directGroupWrites": direct_group_writes, "descriptorCountDirectWrites": descriptor_count_direct_writes, "fieldGroupConsumers": field_group_consumers[:60], "activeGroupARefs": group_a_rows, "activeGroupBRefs": group_b_rows, "activeDescriptorCountRefs": count_rows, "callSites": { "descriptorRebuild": [hx(va) for va in descriptor_rebuild_calls], "displayVmRoot": [hx(va) for va in display_vm_root_calls[:120]], }, "negativeEvidence": [ "active group globals are consumed inside the field actor/collision controller, so they are broader than menu-open state.", "descriptor count 0x004576e8 is shared by save/object/battle/menu descriptor loops and is not unique to HUD.", "the top-menu wrapper still has no direct pointer ref from a normal-field ESC/X input root.", "the 0x0047e688 gate proves wrapper admission conditions, not the producer that schedules this wrapper from field mode.", ], "nextFrontier": [ "trace writers that attach or schedule the stream containing 0x0047e688/0x0047e6a0 from field mode.", "separate descriptor/display setup calls that use 0x402321 from normal field vs intro/title contexts.", "do not promote active group key writes alone as the ESC/X opener.", ], } return report def render_ref_table(rows: list[dict[str, Any]], limit: int = 80) -> str: shown = rows[:limit] body = "".join( "" f"{h(row['refVaHex'])}" f"{h(row['ownerStartVaHex'])}
{h(row['ownerClass'])}" f"{h(row['instructionClass'])}" f"{h(row['contextBytes'])}" "" for row in shown ) if len(rows) > limit: body += f"... {len(rows) - limit} more rows in JSON" return body def render_command_table(context: dict[str, Any]) -> str: return "".join( "" f"{h(row.get('vaHex'))}" f"{h(row.get('opcodeHex'))}
{h(row.get('opcodeName'))}" f"{h(row.get('summary'))}" f"{h(row.get('rawHex'))}" "" for row in context.get("commands", []) ) def render_html(report: dict[str, Any]) -> str: summary = report["summary"] payload = json.dumps(report, ensure_ascii=False) cards = "".join( f"
{h(key)}{h(value)}
" for key, value in summary.items() if key != "decision" ) negatives = "".join(f"
  • {h(item)}
    • " for item in report["negativeEvidence"]) next_frontier = "".join(f"
  • {h(item)}
    • " for item in report["nextFrontier"]) html_doc = f""" HUD Active Group Gate Frontier Review

    HUD Active Group Gate Frontier Review

    {h(summary["decision"])}

    {cards}

    Pre-wrapper Gate Decode

    {render_command_table(report["preWrapperGate"])}
    VAopcodesummaryraw

    Direct Group Writes

    {render_ref_table(report["directGroupWrites"])}
    refownerinstructionbytes

    Descriptor Count Direct Writes

    {render_ref_table(report["descriptorCountDirectWrites"])}
    refownerinstructionbytes

    Field Group Consumers

    {render_ref_table(report["fieldGroupConsumers"])}
    refownerinstructionbytes

    Negative Evidence

      {negatives}

    Next Frontier

      {next_frontier}

    Raw JSON

    
  
    

    



"""
    return "\n".join(line.rstrip() for line in html_doc.splitlines()) + "\n"


def main() -> int:
    report = build_report()
    OUT.mkdir(exist_ok=True)
    WEB.mkdir(exist_ok=True)
    (OUT / "hud_active_group_gate_frontier_review.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    (WEB / "hud_active_group_gate_frontier_review.html").write_text(render_html(report), encoding="utf-8")
    print("hud_active_group_gate_frontier_review ok")
    return 0


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