#!/usr/bin/env python3 """Validate opcode 0x1e formation candidates against command-stream context. The formation stream scanner intentionally accepts byte-level candidates. This review adds the stricter question: does a candidate look like executable VM command stream data, or is it more likely an unrelated byte sequence inside a word table/text/resource payload? This still does not promote encounter maps. It exists to prevent false promotion of random-encounter evidence from actor-row-shaped bytes. """ 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 find_cns_strings, offset_to_va, read_sections, va_to_offset # noqa: E402 HANDLER_TABLE_BASE_VA = 0x00440720 DEFAULT_HANDLER_VA = 0x0040239F 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 read_json(path: Path, default: Any) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: return default def write_json(path: Path, payload: Any) -> None: path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def dword_at_va(exe: bytes, sections: list[dict[str, Any]], va: int) -> int | None: offset = va_to_offset(sections, va) if offset is None or offset + 4 > len(exe): return None return struct.unpack_from(" dict[int, int]: rows: dict[int, int] = {} for opcode in range(count): handler = dword_at_va(exe, sections, HANDLER_TABLE_BASE_VA + opcode * 4) if handler is not None and handler != DEFAULT_HANDLER_VA: rows[opcode] = handler return rows def pointer_refs_to_va(exe: bytes, sections: list[dict[str, Any]], target: int) -> list[dict[str, Any]]: needle = struct.pack(" list[str]: names: set[str] = set() lo = max(0, offset - radius) hi = min(len(exe), offset + radius) for scan in range(lo, max(lo, hi - 3)): value = struct.unpack_from(" list[dict[str, Any]]: command_review = read_json(OUT / "scene_event_vm_command_stream_candidates.json", {}) hits: list[dict[str, Any]] = [] for key in ("commandBlocks", "textSourceCommands", "transitionPayloadRecords"): for row in command_review.get(key) or []: start_hex = row.get("startVaHex") or row.get("vaHex") or row.get("recordVaHex") end_hex = row.get("endVaHex") or start_hex if not isinstance(start_hex, str) or not isinstance(end_hex, str): continue start = int(start_hex, 16) end = int(end_hex, 16) if start - 64 <= candidate_va <= end + 64: hits.append( { "sourceList": key, "id": row.get("id") or row.get("blockId"), "classification": row.get("classification") or row.get("state"), "startVaHex": start_hex, "endVaHex": end_hex, } ) return hits def classify_candidate(row: dict[str, Any], next_opcode_valid: bool, next_opcode: int | None, refs: list[dict[str, Any]], scene_hits: list[dict[str, Any]]) -> str: if refs: return "referenced-command-stream-candidate" if scene_hits: return "scene-context-adjacent-unreferenced" if next_opcode_valid: return "inline-context-plausible-unreferenced" if next_opcode in {0x00, 0x01, 0x52}: return "inline-context-weak-terminator-or-data" if row.get("streamVaHex") == "0x00450bd2": return "sequential-row-id-table-likely-false-positive" return "context-invalid-unreferenced" def build_report() -> dict[str, Any]: exe = EXE.read_bytes() sections = read_sections(exe) cns_strings = find_cns_strings(exe, sections) valid_opcodes = non_default_opcodes(exe, sections) stream_review = read_json(OUT / "field_encounter_formation_stream_review.json", {}) rows: list[dict[str, Any]] = [] for row in stream_review.get("candidates") or []: va = row["streamVa"] count = row["actorCount"] length = 4 + count * 8 offset = va_to_offset(sections, va) if offset is None: continue next_opcode = exe[offset + length] if offset + length < len(exe) else None prev_bytes = exe[max(0, offset - 32) : offset].hex(" ") stream_plus_next = exe[offset : min(len(exe), offset + length + 16)].hex(" ") refs = pointer_refs_to_va(exe, sections, va) scene_hits = scene_context_hits(va) next_handler = valid_opcodes.get(next_opcode) if next_opcode is not None else None classification = classify_candidate(row, next_opcode in valid_opcodes if next_opcode is not None else False, next_opcode, refs, scene_hits) rows.append( { "streamVa": va, "streamVaHex": row["streamVaHex"], "actorCount": count, "entryNames": row.get("entryNames") or [], "hasEnemyActor": row.get("hasEnemyActor"), "streamLength": length, "nextByteHex": f"0x{next_opcode:02x}" if next_opcode is not None else None, "nextByteIsNonDefaultOpcode": next_opcode in valid_opcodes if next_opcode is not None else False, "nextHandlerVaHex": hx(next_handler), "pointerRefCount": len(refs), "pointerRefs": refs, "sceneContextHitCount": len(scene_hits), "sceneContextHits": scene_hits, "nearbyCns": cns_names_near(exe, sections, cns_strings, offset), "previousBytes": prev_bytes, "streamAndFollowingBytes": stream_plus_next, "classification": classification, } ) class_counts: dict[str, int] = {} for row in rows: class_counts[row["classification"]] = class_counts.get(row["classification"], 0) + 1 promoted = [ row for row in rows if row["classification"] in {"referenced-command-stream-candidate", "inline-context-plausible-unreferenced"} ] return { "kind": "hwanse-field-encounter-formation-context-review", "status": "formation-byte-candidates-context-checked-no-producer", "source": [ "Hwanse2.exe", "out/field_encounter_formation_stream_review.json", "out/scene_event_vm_command_stream_candidates.json", "tools/build_field_encounter_formation_context_review.py", ], "handlerTable": { "baseVaHex": hx(HANDLER_TABLE_BASE_VA), "nonDefaultOpcodeCount": len(valid_opcodes), "nonDefaultOpcodesHex": [f"0x{opcode:02x}" for opcode in sorted(valid_opcodes)], }, "summary": { "candidateCount": len(rows), "classificationCounts": dict(sorted(class_counts.items())), "referencedCandidateCount": sum(1 for row in rows if row["pointerRefCount"]), "sceneContextHitCandidateCount": sum(1 for row in rows if row["sceneContextHitCount"]), "nextNonDefaultOpcodeCandidateCount": sum(1 for row in rows if row["nextByteIsNonDefaultOpcode"]), "promotedExecutableContextCandidateCount": len(promoted), "directFieldEncounterProducerFound": False, "mapEncounterClassificationPromoted": False, "decision": ( "No opcode 0x1e candidate currently has a producer/root reference or a strong executable command-stream context. " "The byte-level formation candidates remain search anchors only; they must not classify random-encounter maps." ), }, "candidates": rows, "nextFrontier": [ "A real producer should point ctx+0x40 at an opcode 0x1e stream or reach it through opcode 0x1b/0x1c branch selection.", "The current candidates are not enough to connect field walking, map family, btl background, and monster formation.", "Further progress likely needs a root execution trace or a newly identified command list boundary that contains 0x1e in valid opcode sequence.", ], } def render_html(report: dict[str, Any]) -> str: summary = report["summary"] cards = [ ("status", report["status"]), ("candidates", summary["candidateCount"]), ("referenced", summary["referencedCandidateCount"]), ("scene hits", summary["sceneContextHitCandidateCount"]), ("next opcode", summary["nextNonDefaultOpcodeCandidateCount"]), ("promoted", summary["promotedExecutableContextCandidateCount"]), ] card_html = "".join(f"
{h(k)}{h(v)}
" for k, v in cards) rows_html = [] for row in report["candidates"]: rows_html.append( "" f"{h(row['streamVaHex'])}
{h(row['classification'])}" f"{h(row['actorCount'])}" f"{h(', '.join(row['entryNames']))}" f"{h(row['nextByteHex'])}
{h(row['nextHandlerVaHex'])}" f"{h(row['pointerRefCount'])}" f"{h(row['sceneContextHitCount'])}" f"{h(', '.join(row['nearbyCns'][:8]))}" f"{h(row['streamAndFollowingBytes'])}" "" ) frontier = "".join(f"
  • {h(item)}
    • " for item in report["nextFrontier"]) payload = json.dumps(report, ensure_ascii=False) return f""" Field Encounter Formation Context Review
    index formation stream formation opcode cluster field encounter static

    Field Encounter Formation Context Review

    {h(summary["decision"])}

    {card_html}

    Candidate Context

    {''.join(rows_html)}
    streamcountentriesnext byteptr refsscene hitsnear CNSstream + following bytes

    Next Frontier

      {frontier}

    Raw JSON

    
  
    

    



"""


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


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