tests/test_observability_perception.py

from __future__ import annotations

import json
from http import HTTPStatus
from pathlib import Path
from typing import Any

from world_simulator.api.runtime import GameRuntime
from world_simulator.config import (
    ConnectorConfig,
    GameConfig,
    NpcConfig,
    ServerConfig,
    SimulationConfig,
    WorldConfig,
)
from world_simulator.domain import Beast, House, Npc, ResourceNode, Terrain, Vec3, WorldState
from world_simulator.observability import RunLedger
from world_simulator.simulation.connectors.deterministic import DeterministicWorldSimulator
from world_simulator.simulation.connectors.openai_compatible import OpenAICompatibleWorldSimulator
from world_simulator.simulation.directives import (
    DEFAULT_DIRECTIVE_TTL_TICKS,
    expire_directives,
    set_active_directive,
)
from world_simulator.simulation.spawning import create_world
from world_simulator.simulation.survival import _kill_npc
from world_simulator.simulation.tick import advance_world


def test_dead_npcs_never_appear_in_model_target_fields_over_scripted_deaths() -> None:
    world = _survival_world(npc_count=8)
    world.beasts.clear()
    captured_payloads: list[str] = []

    def fake_chat_completer(request: dict[str, Any]) -> dict[str, Any]:
        # The survival prompt is now a readable text briefing; the connector
        # injects the npc_id onto the tool call for us.
        captured_payloads.append(request["messages"][1]["content"])
        return _tool_response("speak", {"message": "observing"})

    simulator = OpenAICompatibleWorldSimulator(
        ConnectorConfig(
            type="openai_compatible",
            base_url="http://llm.local/v1",
            model="test-model",
            max_parallel_npc_calls=8,
        ),
        fallback=DeterministicWorldSimulator(),
        chat_completer=fake_chat_completer,
    )

    for iteration in range(200):
        if iteration in {3, 17, 41, 73} and len(world.living_npcs()) > 3:
            victim = world.living_npcs()[-1]
            victim.inventory_food = 1
            world.houses[0].occupant_ids.append(victim.id)
            world.beasts.append(Beast(f"beast_scripted_{iteration}", victim.position, target_npc_id=victim.id))
            _kill_npc(world, victim, cause="scripted", actor_id="test")

        captured_payloads.clear()
        pre_tick_dead_ids = {npc.id for npc in world.npcs if npc.health <= 0}
        advance_world(world, simulator)
        dead_ids = {npc.id for npc in world.npcs if npc.health <= 0}
        if not dead_ids:
            continue

        for payload in captured_payloads:
            _assert_survival_payload_has_no_dead_targets(payload, world, pre_tick_dead_ids)
        _assert_world_live_registries_have_no_dead_targets(world, dead_ids)

def test_directive_ttl_expiry_and_replacement() -> None:
    npc = _npc("npc-001", "Ada")
    world = _world([npc])

    set_active_directive(npc, "Gather food.", source="test", tick=0)
    assert npc.directive_ttl_ticks == DEFAULT_DIRECTIVE_TTL_TICKS
    expire_directives(world, tick=DEFAULT_DIRECTIVE_TTL_TICKS - 1)
    assert npc.god_directive == "Gather food."

    expire_directives(world, tick=DEFAULT_DIRECTIVE_TTL_TICKS)
    assert npc.god_directive is None
    assert world.event_log[-1].type == "directive_expired"

    set_active_directive(npc, "First order.", source="test", tick=30, ttl_ticks=24)
    set_active_directive(npc, "Replacement order.", source="test", tick=31, ttl_ticks=7)
    assert npc.god_directive == "Replacement order."
    assert npc.directive_issued_tick == 31
    assert npc.directive_ttl_ticks == 7

    expire_directives(world, tick=37)
    assert npc.god_directive == "Replacement order."
    expire_directives(world, tick=38)
    assert npc.god_directive is None


def test_death_drops_loot_and_updates_witnesses_and_live_registries() -> None:
    killer = _npc("npc-001", "Ada", x=0.0)
    victim = _npc(
        "npc-002",
        "Boris",
        x=1.0,
        inventory_food=2,
        inventory_herbs=1,
        inventory_wood=3,
        inventory_weapon=1,
    )
    witness = _npc("npc-003", "Cora", x=2.0)
    house = House("house_1", Vec3(1.0, 0.0, 0.0), occupant_ids=[victim.id], owner_ids=[victim.id])
    beast = Beast("beast_1", Vec3(0.0, 0.0, 0.0), target_npc_id=victim.id)
    world = _world([killer, victim, witness], houses=[house], beasts=[beast], tick=7)
    killer.focus_target_id = victim.id
    witness.help_target_id = victim.id
    set_active_directive(witness, "Help Boris.", source="test", tick=7)

    _kill_npc(world, victim, cause="violence", actor_id=killer.id)

    assert victim.health <= 0
    assert {node.resource_type: node.amount for node in world.resource_nodes} == {
        "food": 2,
        "herbs": 1,
        "wood": 3,
        "weapon": 1,
    }
    assert witness.relationships[killer.id] == -0.5
    assert witness.structured_memory.episodes[-1].emotional_weight == 0.9
    assert victim.id not in house.occupant_ids
    assert victim.id not in house.owner_ids
    assert killer.focus_target_id is None
    assert witness.help_target_id is None
    assert witness.god_directive is None
    assert beast.target_npc_id is None


def test_beast_kill_gives_witness_fear_and_beast_relationship() -> None:
    victim = _npc("npc-001", "Ada", x=0.0)
    witness = _npc("npc-002", "Boris", x=2.0, fear=10.0)
    world = _world([victim, witness], tick=5)

    _kill_npc(world, victim, cause="beast", actor_id="beast_1")

    assert witness.fear == 40.0
    assert witness.relationships["beast_1"] == -1.0
    assert witness.structured_memory.episodes[-1].subject_kind == "beast"
    assert witness.structured_memory.episodes[-1].emotional_weight == 0.9


def test_runtime_ledger_contains_all_required_phase_types_after_20_ticks() -> None:
    world = create_world(_config(npc_count=4, survival=True))
    ledger_root = Path("tmp") / "ledger_tests"
    ledger_root.mkdir(parents=True, exist_ok=True)
    ledger = RunLedger(root=ledger_root)
    runtime = GameRuntime(
        world=world,
        simulator=_ledger_test_simulator(),
        ledger=ledger,
    )

    for _ in range(20):
        status, _ = runtime.tick()
        assert status == HTTPStatus.OK

    records = [
        json.loads(line)
        for line in ledger.ledger_path.read_text(encoding="utf-8").splitlines()
    ]
    phases = {record["phase"] for record in records}

    assert {"npc_request", "npc_response", "npc_fallback", "engine_events"}.issubset(phases)
    assert "god_command" not in phases
    assert "overseer_request" not in phases
    assert "overseer_response" not in phases
    status, payload = runtime.admin_log_records(run_id="current", limit=5)
    assert status == HTTPStatus.OK
    assert payload["records"]
    assert ledger.pretty_path.read_text(encoding="utf-8").strip()


def _ledger_test_simulator() -> OpenAICompatibleWorldSimulator:
    def fake_chat_completer(request: dict[str, Any]) -> dict[str, Any]:
        # The briefing prints the NPC id in plain text. Force a model outage for
        # npc-001 so the deterministic fallback path is exercised/logged.
        content = request["messages"][1]["content"]
        if "(npc-001)" in content:
            raise RuntimeError("scripted model outage")
        return _tool_response("speak", {"message": "observing"})

    return OpenAICompatibleWorldSimulator(
        ConnectorConfig(
            type="openai_compatible",
            base_url="http://llm.local/v1",
            model="test-model",
            max_parallel_npc_calls=4,
        ),
        fallback=DeterministicWorldSimulator(),
        chat_completer=fake_chat_completer,
    )


def _assert_survival_payload_has_no_dead_targets(
    briefing: str,
    world: WorldState,
    dead_ids: set[str],
) -> None:
    # Allies, threats, resources and shelter are all rendered as "  - ..." bullet
    # lines. A dead NPC must never be offered as a target there.
    bullet_lines = [
        line for line in briefing.splitlines() if line.strip().startswith("- ")
    ]
    for dead_id in dead_ids:
        for line in bullet_lines:
            assert dead_id not in line, f"dead {dead_id} leaked into briefing: {line}"

def _assert_world_live_registries_have_no_dead_targets(
    world: WorldState,
    dead_ids: set[str],
) -> None:
    for house in world.houses:
        assert not (set(house.occupant_ids) & dead_ids)
        assert not (set(house.owner_ids) & dead_ids)
    for beast in world.beasts:
        assert beast.target_npc_id not in dead_ids
    for npc in world.living_npcs():
        assert npc.focus_target_id not in dead_ids
        assert npc.help_target_id not in dead_ids
        if npc.god_directive:
            lowered = npc.god_directive.lower()
            dead_names = [
                dead.name.lower()
                for dead in world.npcs
                if dead.id in dead_ids
            ]
            assert not any(dead_id.lower() in lowered for dead_id in dead_ids)
            assert not any(name in lowered for name in dead_names)


def _survival_world(*, npc_count: int) -> WorldState:
    return create_world(_config(npc_count=npc_count, survival=True))


def _world(
    npcs: list[Npc],
    *,
    houses: list[House] | None = None,
    beasts: list[Beast] | None = None,
    tick: int = 0,
) -> WorldState:
    return WorldState(
        tick=tick,
        seed=42,
        terrain=Terrain(kind="plain_green", width=80, depth=80),
        npcs=npcs,
        houses=houses or [],
        beasts=beasts or [],
    )


def _npc(
    npc_id: str,
    name: str,
    *,
    x: float = 0.0,
    z: float = 0.0,
    **state: Any,
) -> Npc:
    return Npc(
        id=npc_id,
        name=name,
        role=str(state.pop("role", "gatherer")),
        position=Vec3(x=x, y=0.0, z=z),
        **state,
    )


def _config(*, npc_count: int, survival: bool) -> GameConfig:
    return GameConfig(
        world=WorldConfig(width=80, depth=80, terrain="plain_green", seed=42, survival=survival),
        npcs=NpcConfig(count=npc_count),
        simulation=SimulationConfig(tick_ms=500),
        server=ServerConfig(host="127.0.0.1", port=8000),
        connector=ConnectorConfig(type="deterministic"),
    )


def _tool_response(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {
        "choices": [
            {
                "message": {
                    "tool_calls": [
                        {
                            "type": "function",
                            "function": {
                                "name": name,
                                "arguments": json.dumps(arguments),
                            },
                        }
                    ]
                }
            }
        ]
    }