scripts/headless_sim.py

from __future__ import annotations

import argparse
from collections import Counter
from pathlib import Path
import sys
import time

REPO_ROOT = Path(__file__).resolve().parents[1]
SRC_ROOT = REPO_ROOT / "src"
if str(SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(SRC_ROOT))

from world_simulator.config import (  # noqa: E402
    GameConfig,
    NpcConfig,
    ServerConfig,
    SimulationConfig,
    WorldConfig,
)
from world_simulator.simulation.spawning import create_world  # noqa: E402
from world_simulator.simulation.overseer import scripted_overseer_controller  # noqa: E402
from world_simulator.simulation.tick import advance_world  # noqa: E402


def main() -> None:
    parser = argparse.ArgumentParser(description="Run a deterministic headless village simulation.")
    parser.add_argument("--mock-overseer", action="store_true", help="Enable scripted Overseer directives.")
    args = parser.parse_args()

    world = create_world(
        GameConfig(
            world=WorldConfig(
                width=80,
                depth=80,
                terrain="plain_green",
                seed=42,
                survival=True,
            ),
            npcs=NpcConfig(count=6),
            simulation=SimulationConfig(tick_ms=1),
            server=ServerConfig(host="127.0.0.1", port=8000),
        )
    )

    min_population = world.population
    cross_role_actions: list[str] = []
    tick_durations: list[float] = []
    overseer = scripted_overseer_controller() if args.mock_overseer else None
    all_debug: list[dict[str, object]] = []

    mode_label = "deterministic+mock_overseer" if args.mock_overseer else "deterministic"
    print(f"HEADLESS_SIM start seed=42 ticks=300 mode={mode_label}")
    print(
        "HEADLESS_SIM initial "
        f"population={world.population} roles={dict(Counter(npc.role for npc in world.npcs))} "
        f"houses={len(world.houses)} beasts={len(world.beasts)}"
    )

    for _ in range(300):
        before_events = len(world.event_log)
        started = time.perf_counter()
        advance_world(world, overseer=overseer)
        tick_durations.append(time.perf_counter() - started)
        min_population = min(min_population, world.population)
        all_debug.extend(world.last_action_debug)

        for trace in world.last_action_debug:
            action = str(trace.get("action", ""))
            npc_id = str(trace.get("npc_id", ""))
            npc = next((candidate for candidate in world.npcs if candidate.id == npc_id), None)
            if npc is None:
                continue
            if npc.role == "guard" and action == "gather":
                cross_role_actions.append(f"tick {world.tick}: guard {npc.id} gathered")
            if npc.role == "builder" and action == "attack":
                cross_role_actions.append(f"tick {world.tick}: builder {npc.id} attacked")

        new_events = world.event_log[before_events:]
        event_text = "; ".join(
            f"{event.tick}:{event.type}:{event.summary}" for event in new_events
        )
        if not event_text:
            event_text = "no_events"
        print(
            f"tick={world.tick:03d} pop={world.population:02d} "
            f"status={world.game_status} events={event_text}"
        )

    counts = Counter(event.type for event in world.event_log)
    role_by_id = {npc.id: npc.role for npc in world.npcs}
    guard_engagements = [
        event
        for event in world.event_log
        if event.type == "npc_attack"
        and event.target_id is not None
        and event.target_id.startswith("beast")
        and role_by_id.get(event.actor_id or "") == "guard"
    ]

    assert counts["build_completed"] >= 1, "expected at least one completed house"
    assert counts["npc_born"] >= 1, "expected at least one reproduction"
    assert guard_engagements, "expected at least one guard attack against a beast"
    assert min_population >= 0, "population went negative"
    assert cross_role_actions, "expected at least one role-flexible action"
    assert all(duration < 0.25 for duration in tick_durations), "tick exceeded time budget"

    if args.mock_overseer:
        directive_events = [event for event in world.event_log if event.type == "directive_issued"]
        assert directive_events, "expected mock Overseer directives to reach the event log"
        assert any(
            trace.get("npc_id") == "npc-006"
            and trace.get("requested_action") == "attack"
            and trace.get("action") != "attack"
            for trace in all_debug
        ), "expected invalid builder attack directive to be validated/repaired"
        assert world.overseer_last_thoughts, "expected mock Overseer thoughts in world state"
        assert world.overseer_score > 0, "expected scoreboard Overseer points to update"

    houses_by_state = Counter(house.state for house in world.houses)
    print("HEADLESS_SIM summary")
    print(f"  ticks={world.tick}")
    print(f"  game_status={world.game_status}")
    print(f"  final_population={world.population}")
    print(f"  peak_population={world.peak_population}")
    print(f"  min_population={min_population}")
    print(f"  births={counts['npc_born']}")
    print(f"  deaths={counts['npc_died']} causes={dict(world.deaths_by_cause)}")
    print(f"  build_completed={counts['build_completed']}")
    print(f"  houses={dict(houses_by_state)}")
    print(f"  guard_beast_engagements={len(guard_engagements)}")
    print(f"  beasts_killed={counts['beast_killed']}")
    if args.mock_overseer:
        print(f"  directive_issued={counts['directive_issued']}")
        print(f"  overseer_status={world.overseer_status}")
        print(f"  overseer_last_tick={world.overseer_last_tick}")
        print(f"  overseer_score={world.overseer_score}")
        print(f"  chaos_score={world.chaos_score}")
    print(f"  max_tick_ms={max(tick_durations) * 1000:.3f}")
    print("HEADLESS_SIM PASS")


if __name__ == "__main__":
    main()