tests/test_local_agent.py

import json
from pathlib import Path
from types import SimpleNamespace

import pytest

from tower_game.ai import (
    LocalOpenAIGateway,
    MockGateway,
    OPERATION_BRIEFS,
    SYSTEM_PROMPT,
    TowerAgent,
)
from tower_game.engine import TowerGame
from tower_game.narration import NarrationQueue
from tower_game.schemas import (
    BossTurnDecision,
    ClassEvolution,
    DialChange,
    GeneratedSkillSpec,
    TurnNarration,
    TurnNarrationRequest,
)
from tower_game.state import make_save, new_game_state, new_meta_state


class FakeCompletions:
    def __init__(self, calls):
        self.calls = list(calls)
        self.requests = []

    def create(self, **kwargs):
        self.requests.append(kwargs)
        tool_calls = self.calls.pop(0)
        message = SimpleNamespace(content=None, tool_calls=tool_calls)
        usage = SimpleNamespace(prompt_tokens=10, completion_tokens=4)
        return SimpleNamespace(choices=[SimpleNamespace(message=message)], usage=usage)


class FakeClient:
    def __init__(self, calls):
        self.chat = SimpleNamespace(completions=FakeCompletions(calls))


def tool_call(name, arguments, call_id="call-1"):
    return SimpleNamespace(
        id=call_id,
        function=SimpleNamespace(name=name, arguments=json.dumps(arguments)),
    )


def make_agent(calls):
    agent = TowerAgent.__new__(TowerAgent)
    agent.config = {
        "local_model": "test-model",
        "timeout": 1,
        "tool_round_limit": 3,
        "operation_limits": {},
    }
    agent.client = FakeClient(calls)
    agent.last_metrics = {}
    return agent


def test_local_client_disables_hidden_transport_retries(monkeypatch, tmp_path):
    captured = {}

    class FakeOpenAI:
        def __init__(self, **kwargs):
            captured.update(kwargs)

    monkeypatch.setattr("openai.OpenAI", FakeOpenAI)
    key = tmp_path / "key.txt"
    key.write_text("test-key", encoding="utf-8")
    TowerAgent(
        {
            "base_url": "http://127.0.0.1:8080/v1",
            "api_key_path": str(key),
            "timeout": 8,
        }
    )
    assert captured["max_retries"] == 0


def test_agent_requires_tools_and_supports_read_then_submit():
    agent = make_agent(
        [
            [tool_call("inspect_recent_turns", {}, "read")],
            [tool_call("submit_turn_narration", {"text": "The Tower records the blow."}, "submit")],
        ]
    )
    result = agent.call(
        "turn_narration",
        TurnNarration,
        {"combat_events": [{"turn": 1, "actor": "Player", "text": "A hit."}]},
        {"max_characters": 180},
    )
    assert result.text == "The Tower records the blow."
    assert agent.last_metrics["tool_rounds"] == 2
    request = agent.client.chat.completions.requests[0]
    assert request["tool_choice"] == "required"
    assert request["parallel_tool_calls"] is False


def test_agent_uses_operation_temperature_and_curated_skill_tools():
    agent = make_agent(
        [[tool_call("submit_turn_narration", {"text": "A measured answer."})]]
    )
    agent.config["temperatures"] = {"turn_narration": 0.33}
    agent.call("turn_narration", TurnNarration, {}, {"max_characters": 180})
    request = agent.client.chat.completions.requests[0]
    assert request["temperature"] == 0.33

    tools = agent._context({}, {})["list_skill_templates"]()
    assert tools["primary_effects"] == [
        "damage",
        "heavy_damage",
        "guard",
        "heal",
        "poison",
        "attack_up",
        "defense_up",
        "attack_down",
        "defense_down",
    ]
    assert "lifesteal" in tools["attack_riders"]
    assert "hp_sacrifice" in tools["attack_riders"]
    assert "generic words" in OPERATION_BRIEFS["run_setup"]
    assert "Python owns all mechanics" in SYSTEM_PROMPT


def test_local_evolution_rejects_signature_unrelated_to_repeated_strikes():
    class MismatchedAgent:
        last_metrics = {}

        def call(self, operation, schema, state, payload, validate=None):
            result = ClassEvolution(
                class_name="Unrelated Magus",
                title="A false answer",
                narrative="This proposal ignores the repeated physical pattern.",
                portrait_id="player_intelligence",
                signature_skill=GeneratedSkillSpec(
                    name="Remote Thought",
                    description="An unrelated intellectual technique.",
                    ap_cost=1,
                    base_power=8,
                    scaling_stat="int_scaling_dial",
                    element="Magical",
                    effect_id="damage",
                    cooldown=1,
                ),
                dial_changes=[DialChange(dial="int_scaling_dial", delta=0.2)],
            )
            if validate:
                validate(result)
            return result

    gateway = LocalOpenAIGateway.__new__(LocalOpenAIGateway)
    gateway.config = {}
    gateway.agent = MismatchedAgent()
    state = new_game_state(phase="combat")
    state["action_history"] = ["used_strike"] * 12
    result = gateway.evolution(state, 1)
    assert result.signature_skill is not None
    assert result.signature_skill.scaling_stat == "str_scaling_dial"
    assert state["agent_status"]["mode"] == "fallback"


def test_agent_rejects_unknown_tools_and_stops_at_round_limit():
    agent = make_agent(
        [[tool_call("mutate_game_state", {})] for _ in range(3)]
    )
    with pytest.raises(RuntimeError, match="unknown tool"):
        agent.call("turn_narration", TurnNarration, {}, {"max_characters": 180})
    assert len(agent.client.chat.completions.requests) == 3


def test_boss_decision_uses_one_submit_only_tool_round():
    agent = make_agent(
        [[tool_call("submit_boss_turn_decision", {
            "move_index": 1,
            "reaction": "I see the pattern.",
            "adjustment_enabled": False,
            "replace_index": 0,
            "replacement_move_id": "counter",
            "replacement_power": 14,
            "replacement_intent": "Tactic",
            "replacement_element": "Magical",
        })]]
    )
    result = agent.call(
        "boss_turn_decision",
        BossTurnDecision,
        {},
        {"legal_moves": [{"index": 0}, {"index": 1}]},
    )
    assert result.move_index == 1
    request = agent.client.chat.completions.requests[0]
    assert request["tool_choice"] == "required"
    assert len(request["tools"]) == 1
    assert request["tools"][0]["function"]["name"] == "submit_boss_turn_decision"
    assert agent.last_metrics["tool_rounds"] == 1


class GeneratedGateway(MockGateway):
    backend_name = "generated-test"


def test_generated_run_registry_is_persisted_and_transient_jobs_are_not():
    game = TowerGame(GeneratedGateway())
    state, meta = game.new_character(seed=17)
    game.confirm_allocation(state, [6, 6, 6, 6])
    assert state["game_phase"] == "run_setup"
    game.prepare_run(state)
    assert state["game_phase"] == "starter_skill"
    assert len(state["run_skill_registry"]) == 16
    assert len(state["pending_starter_skills"]) == 10
    assert all(skill["ap_cost"] <= 2 for skill in state["run_skill_registry"].values())

    state["_narration_job_id"] = "transient"
    saved = make_save(state, meta)
    assert "_narration_job_id" not in saved["state"]
    assert len(saved["state"]["run_skill_registry"]) == 16
    assert saved["state"]["loot_lexicon"]["adjectives"]


def test_narration_queue_returns_request_identity_and_result():
    queue = NarrationQueue()
    state = {
        "run_id": "run-1",
        "encounter_id": 2,
        "turn_number": 3,
        "floor_label": "Floor 1",
        "combat_events": [{"turn": 3, "actor": "Player", "text": "Strike."}],
    }
    job_id = queue.submit(MockGateway(), state)
    assert job_id
    completed = None
    for _ in range(100):
        completed = queue.poll(job_id)
        if completed:
            break
    assert completed
    request, result, error, fallback = completed
    assert request == TurnNarrationRequest(
        run_id="run-1",
        encounter_id=2,
        turn_id=3,
        floor_label="Floor 1",
        events=state["combat_events"],
    )
    assert result and error is None
    assert fallback is False


def test_llama_scripts_enforce_local_runtime_contracts():
    root = Path(__file__).parents[1]
    start = (root / "scripts" / "Start-LlamaServer.ps1").read_text(encoding="utf-8")
    stop = (root / "scripts" / "Stop-LlamaServer.ps1").read_text(encoding="utf-8")
    install = (root / "scripts" / "Install-LlamaCpp.ps1").read_text(encoding="utf-8")
    gitignore = (root / ".gitignore").read_text(encoding="utf-8")

    assert "releases/latest" in install
    assert "win-cuda-12" in install
    assert "Refusing to replace newer local llama.cpp" in install
    assert "127.0.0.1" in start
    assert "Refusing non-loopback" in start
    assert "--api-key-file" in start
    assert "--offline" in start
    assert "--no-ui" in start
    assert "--tools" not in start
    assert "server.pid" in start and "server.pid" in stop
    assert "Get-Process" in stop and "processPath.StartsWith" in stop
    assert "Get-CimInstance" not in stop
    assert "Set-Content" in stop
    assert "Assets/Model/" in gitignore
    assert ".local/" in gitignore