#!/usr/bin/env python3
"""Capture a bounded Hwanse2.exe process-memory snapshot without WineDbg."""
from __future__ import annotations

import argparse
import json
import os
import struct
import subprocess
import time
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "out"
EXE = ROOT / "Hwanse2.exe"
DEFAULT_PREFIX = Path.home() / ".cache" / "hwanse-wine32"
NULL_ALSA_CONFIG = ROOT / "tools" / "runtime_trace_null_alsa.conf"
IMAGE_BASE = 0x00400000
CURRENT_ROOT_VA = 0x00540714
CURRENT_SECOND_LEVEL_TABLE_VA = 0x005429DC
SELECTED_POINTER_GLOBAL_VA = 0x0059DE30
CURRENT_OBJECT_INDEX_VA = 0x0059E33E
OPCODE24_MODE2_SOURCE_VA = 0x0059E347
OPCODE24_MODE1_SOURCE_VA = 0x0059E348
OPCODE24_RUNTIME_ENABLED_FLAG_VA = 0x0059E34D
SECONDARY_BRANCH_STATE_VA = 0x0059E360


WATCH_VALUES = [
    {
        "name": "selected-pointer-global",
        "staticVa": SELECTED_POINTER_GLOBAL_VA,
        "size": 16,
        "role": "runtime global that should contain the active selector stream pointer",
    },
    {
        "name": "current-selector-root-2:0",
        "staticVa": CURRENT_ROOT_VA,
        "size": 32,
        "role": "selector 2:0 root used by the blocked map1_01a -> map2_02d frontier",
    },
    {
        "name": "current-second-level-table-2:0",
        "staticVa": CURRENT_SECOND_LEVEL_TABLE_VA,
        "size": 32,
        "role": "selector 2:0 second-level table containing current frontier leaves",
    },
    {
        "name": "current-frontier-reader-2:0",
        "staticVa": 0x00542B0C,
        "size": 24,
        "role": "frontier reader/resource gate before map1_01a/map2_02d scene records",
    },
    {
        "name": "opcode24-current-object-index",
        "staticVa": CURRENT_OBJECT_INDEX_VA,
        "size": 1,
        "role": "current runtime object index read by opcode 0x24 before object table lookup",
    },
    {
        "name": "opcode24-mode2-source",
        "staticVa": OPCODE24_MODE2_SOURCE_VA,
        "size": 1,
        "role": "runtime byte consumed by opcode 0x24 mode 2",
    },
    {
        "name": "opcode24-mode1-source",
        "staticVa": OPCODE24_MODE1_SOURCE_VA,
        "size": 8,
        "role": "runtime byte consumed by opcode 0x24 mode 1",
    },
    {
        "name": "opcode24-runtime-enabled-flag",
        "staticVa": OPCODE24_RUNTIME_ENABLED_FLAG_VA,
        "size": 1,
        "role": "runtime flag that must be 1 before opcode 0x24 enters mode dispatch",
    },
    {
        "name": "secondary-branch-state",
        "staticVa": SECONDARY_BRANCH_STATE_VA,
        "size": 16,
        "role": "secondaryBranchState gate table used by the current frontier reader",
    },
]


def hex32(value: int | None) -> str | None:
    if value is None:
        return None
    return f"0x{value & 0xFFFFFFFF:08x}"


def truncate(text: str, limit: int = 6000) -> str:
    if len(text) <= limit:
        return text
    return text[:limit] + f"\n... truncated {len(text) - limit} bytes ..."


def pe_image_size() -> int:
    data = EXE.read_bytes()
    pe_off = struct.unpack_from("<I", data, 0x3C)[0]
    if data[pe_off:pe_off + 4] != b"PE\0\0":
        raise ValueError("Hwanse2.exe is not a PE executable")
    optional_off = pe_off + 24
    return struct.unpack_from("<I", data, optional_off + 56)[0]


def env_for_prefix(prefix: Path) -> dict[str, str]:
    env = os.environ.copy()
    env.update({
        "WINEPREFIX": str(prefix),
        "WINEARCH": "win32",
        "WINEDEBUG": "-all",
    })
    if NULL_ALSA_CONFIG.exists():
        env["ALSA_CONFIG_PATH"] = str(NULL_ALSA_CONFIG)
    return env


def run_text(cmd: list[str], env: dict[str, str], timeout: int = 8) -> tuple[int, str]:
    try:
        result = subprocess.run(
            cmd,
            cwd=ROOT,
            env=env,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            errors="replace",
            timeout=timeout,
        )
        return result.returncode, result.stdout
    except subprocess.TimeoutExpired as exc:
        output = (exc.stdout or "") + (exc.stderr or "")
        return 124, output
    except OSError as exc:
        return 127, str(exc)


def parse_maps(pid: int) -> list[dict[str, Any]]:
    rows = []
    path = Path("/proc") / str(pid) / "maps"
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        parts = line.split(maxsplit=5)
        if len(parts) < 5:
            continue
        start_hex, end_hex = parts[0].split("-", 1)
        rows.append({
            "start": int(start_hex, 16),
            "end": int(end_hex, 16),
            "perms": parts[1],
            "offset": int(parts[2], 16),
            "device": parts[3],
            "inode": parts[4],
            "path": parts[5] if len(parts) > 5 else "",
            "raw": line,
        })
    return rows


def process_name_is_hwanse(pid: int) -> bool:
    proc_dir = Path("/proc") / str(pid)
    try:
        comm = (proc_dir / "comm").read_text(encoding="utf-8", errors="replace").strip()
    except OSError:
        comm = ""
    return comm.lower() == "hwanse2.exe"


def process_has_hwanse_mapping(pid: int) -> bool:
    try:
        return find_loaded_base(parse_maps(pid)) is not None
    except OSError:
        return False


def find_hwanse_pid() -> int | None:
    candidates: list[tuple[int, int]] = []
    status, output = run_text(["pgrep", "-x", "Hwanse2.exe"], os.environ.copy(), timeout=3)
    if status in {0, 1}:
        for line in output.splitlines():
            try:
                pid = int(line.strip())
            except ValueError:
                continue
            candidates.append((0 if process_has_hwanse_mapping(pid) else 1, pid))
    for proc_dir in Path("/proc").iterdir():
        if not proc_dir.name.isdigit():
            continue
        pid = int(proc_dir.name)
        mapped = process_has_hwanse_mapping(pid)
        if mapped or process_name_is_hwanse(pid):
            candidates.append((0 if mapped else 2, pid))
    if not candidates:
        return None
    return min(set(candidates))[1]


def find_loaded_base(maps: list[dict[str, Any]]) -> int | None:
    for row in maps:
        if row["offset"] == 0 and row["path"].endswith("/Hwanse2.exe"):
            return row["start"]
    return None


def map_for_address(maps: list[dict[str, Any]], address: int) -> dict[str, Any] | None:
    for row in maps:
        if row["start"] <= address < row["end"]:
            return row
    return None


def read_memory(pid: int, address: int, size: int) -> tuple[bytes | None, str | None]:
    try:
        with (Path("/proc") / str(pid) / "mem").open("rb", buffering=0) as handle:
            handle.seek(address)
            return handle.read(size), None
    except OSError as exc:
        return None, str(exc)


def dword_from(data: bytes | None, offset: int = 0) -> int | None:
    if data is None or len(data) < offset + 4:
        return None
    return struct.unpack_from("<I", data, offset)[0]


def byte_from(data: bytes | None, offset: int = 0) -> int | None:
    if data is None or len(data) <= offset:
        return None
    return data[offset]


def runtime_to_static(runtime_value: int | None, loaded_base: int, image_size: int) -> int | None:
    if runtime_value is None:
        return None
    if loaded_base <= runtime_value < loaded_base + image_size:
        return IMAGE_BASE + (runtime_value - loaded_base)
    return None


def sample_memory(pid: int, maps: list[dict[str, Any]], loaded_base: int, image_size: int) -> dict[str, Any]:
    rows = []
    for spec in WATCH_VALUES:
        runtime_va = loaded_base + (spec["staticVa"] - IMAGE_BASE)
        data, error = read_memory(pid, runtime_va, spec["size"])
        first_byte = byte_from(data)
        first_dword = dword_from(data)
        rows.append({
            "name": spec["name"],
            "role": spec["role"],
            "staticVaHex": hex32(spec["staticVa"]),
            "runtimeVaHex": hex32(runtime_va),
            "size": spec["size"],
            "mapped": map_for_address(maps, runtime_va) is not None,
            "readOk": data is not None and len(data) == spec["size"],
            "error": error,
            "bytesHex": data.hex(" ") if data is not None else None,
            "firstByte": first_byte,
            "firstByteHex": f"0x{first_byte:02x}" if first_byte is not None else None,
            "firstDwordHex": hex32(first_dword),
            "firstDwordStaticHex": hex32(runtime_to_static(first_dword, loaded_base, image_size)),
        })
    selected = next((row for row in rows if row["name"] == "selected-pointer-global"), {})
    current = next((row for row in rows if row["name"] == "current-selector-root-2:0"), {})
    current_bytes = bytes.fromhex((current.get("bytesHex") or "").replace(" ", "")) if current.get("bytesHex") else None
    current_table_runtime = dword_from(current_bytes, 8)
    expected_table_runtime = loaded_base + (CURRENT_SECOND_LEVEL_TABLE_VA - IMAGE_BASE)
    selected_runtime_value = dword_from(
        bytes.fromhex((selected.get("bytesHex") or "").replace(" ", "")) if selected.get("bytesHex") else None
    )
    current_root_runtime = loaded_base + (CURRENT_ROOT_VA - IMAGE_BASE)
    selected_static = runtime_to_static(selected_runtime_value, loaded_base, image_size)
    return {
        "canReadProcessMemory": all(row["readOk"] for row in rows[:2]),
        "selectedPointerRuntimeValueHex": hex32(selected_runtime_value),
        "selectedPointerStaticValueHex": hex32(selected_static),
        "currentRootRuntimeHex": hex32(current_root_runtime),
        "currentRootStaticHex": hex32(CURRENT_ROOT_VA),
        "selectedPointerEqualsCurrentRoot": selected_runtime_value == current_root_runtime,
        "currentRootSecondLevelRuntimeValueHex": hex32(current_table_runtime),
        "expectedSecondLevelRuntimeHex": hex32(expected_table_runtime),
        "currentRootRelocationLooksValid": current_table_runtime == expected_table_runtime,
        "samples": rows,
    }


def build_summary(wait_seconds: float = 12.0, prefix: Path = DEFAULT_PREFIX) -> dict[str, Any]:
    image_size = pe_image_size()
    env = env_for_prefix(prefix)
    prefix.mkdir(parents=True, exist_ok=True)
    run_text(["wineserver", "-k"], env, timeout=5)
    command = ["xvfb-run", "-a", "wine", "explorer", "/desktop=hwanse,640x480", "Hwanse2.exe"]
    process = subprocess.Popen(
        command,
        cwd=ROOT,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        errors="replace",
    )
    time.sleep(wait_seconds)
    pid = find_hwanse_pid()
    tasklist_status, tasklist_output = run_text(["wine", "tasklist"], env, timeout=8)
    startup_output = ""
    maps: list[dict[str, Any]] = []
    loaded_base = None
    memory = {
        "canReadProcessMemory": False,
        "samples": [],
    }
    snapshot_status = "process-not-found"
    try:
        if pid is not None:
            maps = parse_maps(pid)
            loaded_base = find_loaded_base(maps)
            if loaded_base is None:
                snapshot_status = "loaded-base-not-found"
            else:
                memory = sample_memory(pid, maps, loaded_base, image_size)
                if memory.get("canReadProcessMemory"):
                    snapshot_status = "memory-readable-no-route-proof"
                    if memory.get("selectedPointerEqualsCurrentRoot"):
                        snapshot_status = "memory-readable-current-root-selected"
                else:
                    snapshot_status = "memory-read-failed"
    finally:
        run_text(["wineserver", "-k"], env, timeout=5)
        try:
            startup_output, _ = process.communicate(timeout=5)
        except subprocess.TimeoutExpired:
            process.kill()
            startup_output, _ = process.communicate(timeout=5)
    promotion_status = "blocked"
    return {
        "objective": "non-debugger process-memory snapshot for Hwanse2.exe selected-pointer proof",
        "command": command,
        "waitSeconds": wait_seconds,
        "winePrefix": str(prefix),
        "linuxPid": pid,
        "tasklistStatus": tasklist_status,
        "tasklistOutput": truncate(tasklist_output),
        "startupStatus": process.returncode,
        "startupOutput": truncate(startup_output or ""),
        "imageBaseHex": hex32(IMAGE_BASE),
        "imageSizeHex": hex32(image_size),
        "loadedBaseHex": hex32(loaded_base),
        "hwanseMappings": [
            {
                "startHex": hex32(row["start"]),
                "endHex": hex32(row["end"]),
                "perms": row["perms"],
                "offsetHex": hex32(row["offset"]),
                "path": row["path"],
            }
            for row in maps
            if row["path"].endswith("/Hwanse2.exe")
            or (loaded_base is not None and loaded_base <= row["start"] < loaded_base + image_size)
        ],
        "memory": memory,
        "snapshotStatus": snapshot_status,
        "routePromotionStatus": promotion_status,
        "conclusion": (
            "A live Hwanse2.exe process can be started in Wine virtual desktop mode and sampled through "
            "/proc/<pid>/mem without WineDbg. This gives a fallback way to inspect relocated globals such as "
            "0x0059de30, but this idle snapshot is not a route-path watchpoint or strict hotspot proof, so "
            "map1_01a -> map2_02d remains blocked."
            if memory.get("canReadProcessMemory")
            else "The process-memory snapshot did not reach readable Hwanse2.exe memory in this run."
        ),
    }


def markdown(summary: dict[str, Any]) -> str:
    memory = summary.get("memory") or {}
    lines = [
        "# Runtime Memory Snapshot",
        "",
        f"- objective: {summary.get('objective')}",
        f"- snapshot status: `{summary.get('snapshotStatus')}`",
        f"- route promotion status: `{summary.get('routePromotionStatus')}`",
        f"- linux pid: `{summary.get('linuxPid') or '-'}`",
        f"- loaded base: `{summary.get('loadedBaseHex') or '-'}`",
        f"- selected pointer runtime value: `{memory.get('selectedPointerRuntimeValueHex') or '-'}`",
        f"- selected pointer static value: `{memory.get('selectedPointerStaticValueHex') or '-'}`",
        f"- selected pointer equals current root: {memory.get('selectedPointerEqualsCurrentRoot')}",
        f"- current root relocation looks valid: {memory.get('currentRootRelocationLooksValid')}",
        "",
        summary.get("conclusion") or "",
        "",
        "## Samples",
        "",
        "| name | static VA | runtime VA | first byte | first dword | static dword | read | role |",
        "| --- | --- | --- | --- | --- | --- | --- | --- |",
    ]
    for row in memory.get("samples") or []:
        lines.append(
            f"| {row.get('name')} | `{row.get('staticVaHex')}` | `{row.get('runtimeVaHex')}` | "
            f"`{row.get('firstByteHex') or '-'}` | `{row.get('firstDwordHex') or '-'}` | "
            f"`{row.get('firstDwordStaticHex') or '-'}` | {row.get('readOk')} | {row.get('role')} |"
        )
    lines.append("")
    return "\n".join(lines)


def write_outputs(summary: dict[str, Any], out_dir: Path = OUT) -> None:
    out_dir.mkdir(parents=True, exist_ok=True)
    (out_dir / "runtime_memory_snapshot.json").write_text(
        json.dumps(summary, ensure_ascii=False, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--wait", type=float, default=12.0, help="seconds to wait before reading Hwanse2.exe memory")
    parser.add_argument("--prefix", type=Path, default=DEFAULT_PREFIX)
    parser.add_argument("--out-dir", type=Path, default=OUT)
    args = parser.parse_args()
    summary = build_summary(wait_seconds=args.wait, prefix=args.prefix)
    write_outputs(summary, args.out_dir)
    print(
        "wrote runtime memory snapshot -> "
        f"{args.out_dir / 'runtime_memory_snapshot.json'} "
        f"({summary.get('snapshotStatus')})"
    )


if __name__ == "__main__":
    main()