VoxelForge/Standalone/Research/WorldAssemblerM3/Linux168/RunL168Movement.py

#!/usr/bin/env python3
"""Five interleaved L168 30 m/s UE pairs, using source-default streaming CVars."""
from __future__ import annotations

import json
import math
import shutil
import statistics
import subprocess
import time
from pathlib import Path


ROOT = Path(__file__).resolve().parents[4]
HOST = Path("/home/jahni/VFHost")
EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd")
OUT = ROOT / "Saved/Linux/l168/ue/movement"
BIN_DIR = ROOT / "Saved/Linux/l168/bin"
HOST_PLUGIN = HOST / "Binaries/Linux/libUnrealEditor-VoxelForge.so"
SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375)
MOVE_START_S = 6.0
MOVE_DURATION_S = 12.0

CONDITIONS = {
    "assembler_world": {
        "binary": BIN_DIR / "assembler-world/libUnrealEditor-VoxelForge.so",
        "incremental": 0,
    },
    "incremental_default": {
        "binary": BIN_DIR / "incremental-default/libUnrealEditor-VoxelForge.so",
        "incremental": 1,
    },
}
ORDER = {
    1: ("assembler_world", "incremental_default"),
    2: ("incremental_default", "assembler_world"),
    3: ("assembler_world", "incremental_default"),
    4: ("incremental_default", "assembler_world"),
    5: ("assembler_world", "incremental_default"),
}


def percentile(values: list[float], fraction: float = 0.99) -> float | None:
    if not values:
        return None
    ordered = sorted(values)
    return ordered[max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1))]


def percent(numerator: int | float, denominator: int | float) -> float | None:
    return 100.0 * numerator / denominator if denominator > 0 else None


def read_events(path: Path) -> list[dict]:
    rows = []
    if path.is_file():
        for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
            if not line.startswith("{"):
                continue
            try:
                rows.append(json.loads(line))
            except json.JSONDecodeError:
                continue
    return rows


def summarize_run(condition: str, run_name: str, trace: Path, log: Path,
                  exit_code: int, wall_s: float, audit: bool = False) -> dict:
    expected_incremental = CONDITIONS[condition]["incremental"]
    events = read_events(trace)
    backend = next((e for e in events if e.get("type") == "worker_backend_config"), {})
    scheduler = [e for e in events if e.get("type") == "scheduler_admission"]
    if not scheduler:
        raise RuntimeError(f"missing scheduler readback: {trace}")
    if any(int(e.get("incremental_desired_set", -1)) != expected_incremental for e in scheduler):
        raise RuntimeError(f"incremental default readback mismatch: {trace}")
    if any(int(e.get("forward_first", -1)) != 0
           or int(e.get("rebuild_on_core_region_completion", -1)) != 1
           or int(e.get("core_desired_cache_by_center", -1)) != 0
           for e in scheduler):
        raise RuntimeError(f"L160 control default readback mismatch: {trace}")
    if any(int(e.get("incremental_desired_audit_every", -1)) != (1 if audit else 0)
           or int(e.get("incremental_desired_materialization_audit_every", -1))
               != (1 if audit else 0)
           or int(e.get("incremental_desired_audit_inject_mismatch", -1)) != 0
           for e in scheduler):
        raise RuntimeError(f"audit CVar readback mismatch: {trace}")
    if (int(backend.get("streaming_fast_travel_wait_for_all_critical_tiles", -1)) != 1
            or abs(float(backend.get("streaming_fast_travel_apply_budget_ms", -1.0))) > 0.01
            or int(backend.get("effective_tile_slots", -1)) != 8):
        raise RuntimeError(f"L157/default worker readback mismatch: {trace}; {backend}")

    move_stop = next((e for e in events if e.get("type") == "streaming_test_move_stopped"), None)
    if move_stop is None:
        raise RuntimeError(f"missing move-stop event: {trace}")
    stop_s = float(move_stop.get("t_s", 0.0))
    samples = [e for e in events if e.get("type") == "streaming_sample"]
    moving = [e for e in samples if MOVE_START_S <= float(e.get("t_s", 0.0)) < stop_s
              and float(e.get("speed_cm_s", 0.0)) >= 50.0]
    dwell = [e for e in samples if float(e.get("t_s", 0.0)) >= stop_s
             and float(e.get("speed_cm_s", 0.0)) < 50.0]
    if exit_code != 0 or len(moving) < 40 or len(dwell) < 16:
        raise RuntimeError(f"incomplete run {run_name}: exit={exit_code}, move={len(moving)}, dwell={len(dwell)}")

    move_start = min(float(e.get("t_s", 0.0)) for e in moving)
    work = [e for e in events if e.get("type") == "game_thread_work_frame"
            and move_start <= float(e.get("t_s", 0.0)) < stop_s]
    ticks = [float(e.get("tick_ms", 0.0)) for e in work]
    rebuilds = [e for e in events if e.get("type") == "desired_rebuild_breakdown"
                and move_start <= float(e.get("t_s", 0.0)) < stop_s]
    core_events = [e for e in events if e.get("type") == "core_desired_cache"
                   and move_start <= float(e.get("t_s", 0.0)) < stop_s]
    audits = [e for e in events if e.get("type") == "incremental_desired_audit"
              and move_start <= float(e.get("t_s", 0.0)) < stop_s]
    order_audits = [e for e in events if e.get("type") == "incremental_desired_order_audit"
                    and move_start <= float(e.get("t_s", 0.0)) < stop_s]
    materialization_audits = [e for e in audits if int(e.get("materialization_audit", 0)) != 0]
    if audit:
        if (not audits or not order_audits or not materialization_audits
                or any(int(e.get("pass", 0)) != 1 for e in audits + order_audits)):
            raise RuntimeError(
                f"equality audit missing/failed: {trace}; set={len(audits)} "
                f"order={len(order_audits)} materialization={len(materialization_audits)}")

    near_values = []
    far_holes = []
    for event in moving:
        desired = event.get("desired_by_lod", [0] * 9)
        resident = event.get("resident_by_lod", [0] * 9)
        covered = event.get("covered_by_lod", resident)
        near_values.append(percent(sum(resident[:3]), sum(desired[:3])) or 0.0)
        far_holes.append(percent(sum(desired[3:]) - sum(covered[3:]), sum(desired[3:])) or 0.0)
    settled = next((e for e in samples if float(e.get("t_s", 0.0)) < move_start
                    and float(e.get("speed_cm_s", 0.0)) < 50.0
                    and sum(e.get("settled_by_lod", [0] * 9)[:3])
                        >= sum(e.get("desired_by_lod", [0] * 9)[:3]) > 0), None)
    row = {
        "condition": condition,
        "run": run_name,
        "audit_run": audit,
        "source_default_incremental": expected_incremental,
        "speed_mps": 30,
        "effective_tile_slots": backend.get("effective_tile_slots"),
        "moving_samples": len(moving),
        "dwell_samples": len(dwell),
        "game_thread_tick_p99_ms": percentile(ticks),
        "game_thread_tick_worst_ms": max(ticks, default=None),
        "near_readiness_moving_mean_pct": statistics.mean(near_values) if near_values else None,
        "far_holes_moving_mean_pct": statistics.mean(far_holes) if far_holes else None,
        "first_settled_lod02_s": float(settled.get("t_s", 0.0)) if settled else None,
        "desired_rebuild_count": len(rebuilds),
        "incremental_materialization_update_count": sum(
            int(e.get("materialization_update", 0)) != 0 for e in rebuilds),
        "priority_order_update_count": sum(
            int(e.get("priority_order_updated", 0)) != 0 for e in rebuilds),
        "priority_order_reuse_count": sum(
            int(e.get("priority_order_reused", 0)) != 0 for e in rebuilds),
        "core_incremental_delta_count": sum(int(e.get("incremental_delta", 0)) != 0
                                             for e in core_events),
        "core_cache_hit_count": sum(int(e.get("hit", 0)) != 0 for e in core_events),
        "set_audits": len(audits),
        "order_audits": len(order_audits),
        "materialization_audits": len(materialization_audits),
        "audit_failures": sum(int(e.get("pass", 1)) != 1 for e in audits + order_audits),
        "trace": str(trace.relative_to(ROOT)),
        "log": str(log.relative_to(ROOT)),
        "wall_s": round(wall_s, 3),
    }
    if expected_incremental and not audit and (
            row["incremental_materialization_update_count"] == 0
            or row["core_incremental_delta_count"] == 0):
        raise RuntimeError(f"default incremental trace counter stayed zero: {trace}; {row}")
    return row


def run(condition: str, run_name: str, audit: bool = False) -> dict:
    binary = CONDITIONS[condition]["binary"]
    if not binary.is_file():
        raise FileNotFoundError(binary)
    shutil.copy2(binary, HOST_PLUGIN)
    out = OUT / condition
    export = HOST / "Plugins/VoxelForge/Saved/Linux/l168" / condition / run_name
    userdir = HOST / "Saved/Linux/l168/user" / condition / run_name
    zendata = HOST / "Saved/Linux/l168/zen" / condition / run_name
    ddc = HOST / "Saved/Linux/l168/DDC"
    trace = out / f"{run_name}.trace.jsonl"
    log = out / f"{run_name}.log"
    stdout = out / f"{run_name}.stdout.log"
    for directory in (out, export, userdir, zendata, ddc):
        directory.mkdir(parents=True, exist_ok=True)
    trace.unlink(missing_ok=True)
    spawn_cm = ",".join(f"{value * 100.0:.3f}" for value in SPAWN_M)
    command = [
        str(EDITOR), str(HOST / "VFHost.uproject"), "-nullrhi", "-unattended",
        "-nop4", "-nosplash", "-NoSound", "-corelimit=12", "-NoZenAutoLaunch",
        "-notraceserver", "-nocrashreports", "-DisablePlugins=AndroidFileServer",
        "-ddc=NoZenLocalFallback", f"-userdir={userdir}", f"-ZenDataPath={zendata}",
        f"-LocalDataCachePath={ddc}", "-run=VoxelForgeExplore", "-seed=0",
        "-archetype=TunnelNetwork", "-slot=0", "-modes=export", "-gameconfig=1",
        "-coreworldtest=1", "-ownerfourstrateplan=1", "-corevoxelsizecm=25",
        "-coreislandtest=0", "-corefallbacktest=0", "-coreofftest=0",
        "-coreplantest=0", "-settingsref=/Game/VoxelForge/DA_Settings.DA_Settings",
        "-corehybridmesher=0", f"-out={export}",
        "-voxel.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=3000.0",
        f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}",
        f"-voxel.TestMoveDurationSeconds={MOVE_DURATION_S:.1f}",
        "-voxel.TestExitSeconds=0.0", "-voxel.TestMoveDirection=1,0,0",
        "-voxel.TestExitOnSteady=0", "-coreworldstreamtrace=1",
        f"-voxel.StartupTraceSpawnCm={spawn_cm}",
        "-voxel.StartupTraceUntilEndPlay=1", "-voxel.StartupTraceThroughCrossing=1",
        "-voxel.StartupTraceMaxTiles=16384", f"-voxel.StartupTraceFile={trace}",
        f"-abslog={log}",
    ]
    if audit:
        command += ["-voxel.StreamingIncrementalDesiredAuditEvery=1",
                    "-voxel.StreamingIncrementalDesiredMaterializationAuditEvery=1"]
    if any("StreamingIncrementalDesiredSet=" in arg for arg in command):
        raise RuntimeError("timing/default proof must not override incremental desired-set CVar")
    started = time.monotonic()
    with stdout.open("w", encoding="utf-8") as stream:
        process = subprocess.run(["taskset", "-c", "0-11", "timeout", "1740s", *command],
                                 cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT,
                                 check=False, timeout=1740)
    return summarize_run(condition, run_name, trace, log, process.returncode,
                         time.monotonic() - started, audit)


def main() -> int:
    missing = [p for p in (EDITOR, HOST / "VFHost.uproject", HOST_PLUGIN)
               if not p.exists()]
    if missing:
        raise SystemExit(f"missing UE runner input: {missing}")
    OUT.mkdir(parents=True, exist_ok=True)
    BIN_DIR.mkdir(parents=True, exist_ok=True)

    # One warm-up for each binary, excluded from the paired summary. The candidate warm-up
    # is also the no-CVar-override proof that the compiled default executes incremental updates.
    warmups = [run("assembler_world", "warmup"),
               run("incremental_default", "warmup-default")]
    rows = []
    partial = OUT / "runs-partial.json"
    for pair, sequence in ORDER.items():
        for condition in sequence:
            rows.append(run(condition, f"pair-{pair}"))
            partial.write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n",
                               encoding="utf-8")

    # One audited candidate run, outside the timing matrix.
    audit_row = run("incremental_default", "equality-audit", audit=True)
    (OUT / "warmups.json").write_text(json.dumps(warmups, indent=2, sort_keys=True) + "\n",
                                       encoding="utf-8")
    (OUT / "equality-audit.json").write_text(
        json.dumps(audit_row, indent=2, sort_keys=True) + "\n", encoding="utf-8")

    metrics = ("game_thread_tick_p99_ms", "game_thread_tick_worst_ms",
               "near_readiness_moving_mean_pct", "far_holes_moving_mean_pct",
               "first_settled_lod02_s", "incremental_materialization_update_count",
               "core_incremental_delta_count")
    by_pair = {pair: {row["condition"]: row for row in rows
                      if row["run"] == f"pair-{pair}"} for pair in ORDER}
    medians = {}
    for condition in CONDITIONS:
        selected = [row for row in rows if row["condition"] == condition]
        medians[condition] = {metric: statistics.median(
            row[metric] for row in selected if row[metric] is not None) for metric in metrics}
    deltas = []
    for pair in ORDER:
        base = by_pair[pair]["assembler_world"]
        candidate = by_pair[pair]["incremental_default"]
        deltas.append({"pair": pair, **{f"{m}_delta": candidate[m] - base[m]
                                        for m in metrics
                                        if candidate[m] is not None and base[m] is not None}})
    summary = {
        "protocol": "Five interleaved 30 m/s UE pairs; 12 s move plus 5 s dwell; 8 effective tile slots; four-strata owner plan. No streaming configuration CVar overrides during warm-up or timing runs. One warm-up per binary excluded. Equality audits run once afterward.",
        "default_readback": {
            "incremental_desired_set": 1,
            "incremental_desired_audit_every": 0,
            "incremental_desired_materialization_audit_every": 0,
            "incremental_desired_audit_inject_mismatch": 0,
            "fast_travel_wait_for_all_critical_tiles": 1,
            "fast_travel_apply_budget_ms": 0,
            "forward_first": 0,
            "rebuild_on_core_region_completion": 1,
            "core_desired_cache_by_center": 0,
        },
        "warmups": warmups,
        "movement_medians": medians,
        "paired_deltas_candidate_minus_baseline": deltas,
        "equality_audit": audit_row,
        "runs": rows,
    }
    (OUT / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n",
                                       encoding="utf-8")
    print("[L168Summary] " + json.dumps({"movement_medians": medians,
                                         "audit": audit_row,
                                         "pairs": len(ORDER)}, sort_keys=True), flush=True)
    return 0


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