From 15d24be638ec372a8c108f717b61b864092a6caa Mon Sep 17 00:00:00 2001 From: ctan Date: Tue, 22 Sep 2026 12:21:23 +0200 Subject: [PATCH] feat(phase2): P1c --- src/pf1e_simulator/combat.py | 4 +- src/pf1e_simulator/nl_policy.py | 271 ++++++++++++++++++++++++++ src/pf1e_simulator/prompt_builder.py | 181 +++++++++++++++++ tests/test_phase2_nl_policy.py | 280 +++++++++++++++++++++++++++ tests/test_phase2_offline_battle.py | 87 +++++++++ tests/test_phase2_prompt_builder.py | 223 +++++++++++++++++++++ 6 files changed, 1044 insertions(+), 2 deletions(-) create mode 100644 src/pf1e_simulator/nl_policy.py create mode 100644 src/pf1e_simulator/prompt_builder.py create mode 100644 tests/test_phase2_nl_policy.py create mode 100644 tests/test_phase2_offline_battle.py create mode 100644 tests/test_phase2_prompt_builder.py diff --git a/src/pf1e_simulator/combat.py b/src/pf1e_simulator/combat.py index 000df26..0939b94 100644 --- a/src/pf1e_simulator/combat.py +++ b/src/pf1e_simulator/combat.py @@ -1526,7 +1526,7 @@ _REASON_CANNOT_COMBINE = "cannot_combine" # Action kinds that are declared but have no engine dispatch (U1 surfaces # them as explicit rejections instead of silent no-ops). -_UNSUPPORTED_ACTION_KINDS: frozenset[str] = frozenset({"swift", "free", "full_round"}) +UNSUPPORTED_ACTION_KINDS: frozenset[str] = frozenset({"swift", "free", "full_round"}) # Action kinds whose helper is a silent no-op (or worse) without a resolved # target; _execute rejects them up front with missing_target instead. # move / 5foot_step / run are exempt when an explicit destination is provided @@ -5097,7 +5097,7 @@ class CombatEngine: reports ok. """ kind = action.kind - if kind in _UNSUPPORTED_ACTION_KINDS: + if kind in UNSUPPORTED_ACTION_KINDS: self._log( f"round {self._current_round} {state.combatant.id}: " f"{kind} rejected ({_REASON_UNSUPPORTED_KIND})" diff --git a/src/pf1e_simulator/nl_policy.py b/src/pf1e_simulator/nl_policy.py new file mode 100644 index 0000000..507a40b --- /dev/null +++ b/src/pf1e_simulator/nl_policy.py @@ -0,0 +1,271 @@ +"""Natural-language policy: parses LLM JSON replies into Action sequences (P1.3).""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from dataclasses import dataclass, fields +from typing import TYPE_CHECKING, Literal, TypeGuard, get_args + +from pf1e_simulator.candidates import candidates +from pf1e_simulator.combat import Action, ReadyTrigger, TrickCondition + +if TYPE_CHECKING: + from pf1e_simulator.client import LLMClient + from pf1e_simulator.combat import Observation + from pf1e_simulator.prompt_builder import PromptBuilder + + +# ---- Type aliases for the 27 supported kinds (filtered from Action.kind) ---- +KindValue = Literal[ + "attack", "move", "wait", "immediate", "full_attack", "charge", "withdraw", + "5foot_step", "cast_spell", "maneuver", "channel_energy", "hex", "deed", + "spell_combat", "inspire_courage", "retrieve_weapon", "repair_weapon", + "activate_power_attack", "delay", "ready", "total_defense", + "fight_defensively", "stand_up", "drop_prone", "aid_another", "run", "hide", +] +_KIND_VALUES: frozenset[str] = frozenset(get_args(KindValue)) +_FIELD_NAMES: frozenset[str] = frozenset(f.name for f in fields(Action)) +_PAIR_LEN = 2 +_CHANNEL_MODES: frozenset[str] = frozenset(get_args(Literal["heal", "harm"])) +_AID_MODES: frozenset[str] = frozenset(get_args(Literal["attack", "ac"])) +_TRICK_CONDITIONS: frozenset[str] = frozenset(get_args(TrickCondition)) +_READY_TRIGGERS: frozenset[str] = frozenset(get_args(ReadyTrigger)) + + +class NLPolicyError(ValueError): + """The LLM reply is not a valid JSON action sequence.""" + + +# ---- Type guards ------------------------------------------------------------- + +def _is_kind(value: object) -> TypeGuard[KindValue]: + return value in _KIND_VALUES + + +def _is_channel_mode(value: object) -> TypeGuard[Literal["heal", "harm"]]: + return value in _CHANNEL_MODES + + +def _is_aid_mode(value: object) -> TypeGuard[Literal["attack", "ac"]]: + return value in _AID_MODES + + +def _is_trick_condition(value: object) -> TypeGuard[TrickCondition]: + return value in _TRICK_CONDITIONS + + +def _is_ready_trigger(value: object) -> TypeGuard[ReadyTrigger]: + return value in _READY_TRIGGERS + + +def _is_object_list(value: object) -> TypeGuard[list[object]]: + return isinstance(value, list) + + +def _is_object_seq(value: object) -> TypeGuard[list[object] | tuple[object, ...]]: + return isinstance(value, (list, tuple)) + + +def _is_object_dict(value: object) -> TypeGuard[dict[str, object]]: + return isinstance(value, dict) + + +# ---- Coercion helpers -------------------------------------------------------- + +def _opt_str(value: object, name: str) -> str | None: + if value is None: + return None + if isinstance(value, str): + return value + msg = f"{name!r} must be a string" + raise NLPolicyError(msg) + + +def _str_tuple(value: object, name: str) -> tuple[str, ...]: + if value is None: + return () + if not _is_object_seq(value): + msg = f"{name!r} must be a list of strings" + raise NLPolicyError(msg) + out: list[str] = [] + for item in value: + if not isinstance(item, str): + msg = f"{name!r} must be a list of strings" + raise NLPolicyError(msg) + out.append(item) + return tuple(out) + + +def _int_pair(value: object, name: str) -> tuple[int, int] | None: + if value is None: + return None + if not _is_object_seq(value): + msg = f"{name!r} must be a [row, col] pair" + raise NLPolicyError(msg) + if len(value) != _PAIR_LEN: + msg = f"{name!r} must be a [row, col] pair" + raise NLPolicyError(msg) + first, second = value[0], value[1] + if ( + isinstance(first, bool) + or isinstance(second, bool) + or not isinstance(first, int) + or not isinstance(second, int) + ): + msg = f"{name!r} must be a [row, col] pair" + raise NLPolicyError(msg) + return (first, second) + + +def _opt_bool(value: object, name: str) -> bool: + if value is None: + return False + if isinstance(value, bool): + return value + msg = f"{name!r} must be a boolean" + raise NLPolicyError(msg) + + +def _opt_channel_mode(value: object) -> Literal["heal", "harm"] | None: + if value is None: + return None + if _is_channel_mode(value): + return value + msg = "'channel_mode' must be one of: heal, harm" + raise NLPolicyError(msg) + + +def _opt_aid_mode(value: object) -> Literal["attack", "ac"] | None: + if value is None: + return None + if _is_aid_mode(value): + return value + msg = "'aid_mode' must be one of: attack, ac" + raise NLPolicyError(msg) + + +def _opt_trick_condition(value: object) -> TrickCondition | None: + if value is None: + return None + if _is_trick_condition(value): + return value + msg = "'trick_condition' must be one of: " + ", ".join(get_args(TrickCondition)) + raise NLPolicyError(msg) + + +def _opt_ready_trigger(value: object) -> ReadyTrigger | None: + if value is None: + return None + if _is_ready_trigger(value): + return value + msg = "'ready_trigger' must be one of: " + ", ".join(get_args(ReadyTrigger)) + raise NLPolicyError(msg) + + +# ---- Action construction ----------------------------------------------------- + +def _action_from_item(item: object) -> Action: + if not _is_object_dict(item): + msg = "each action must be a JSON object" + raise NLPolicyError(msg) + fields_dict = item + unknown = sorted(set(fields_dict) - _FIELD_NAMES) + if unknown: + msg = f"unknown action field(s): {', '.join(unknown)}" + raise NLPolicyError(msg) + kind = fields_dict.get("kind") + if not _is_kind(kind): + msg = "missing or unknown 'kind'" + raise NLPolicyError(msg) + return Action( + kind=kind, + target_id=_opt_str(fields_dict.get("target_id"), "target_id"), + spell_name=_opt_str(fields_dict.get("spell_name"), "spell_name"), + maneuver=_opt_str(fields_dict.get("maneuver"), "maneuver"), + hex_name=_opt_str(fields_dict.get("hex_name"), "hex_name"), + deed_name=_opt_str(fields_dict.get("deed_name"), "deed_name"), + class_ability=_opt_str(fields_dict.get("class_ability"), "class_ability"), + channel_mode=_opt_channel_mode(fields_dict.get("channel_mode")), + channel_exclude=_str_tuple(fields_dict.get("channel_exclude"), "channel_exclude"), + direction=_int_pair(fields_dict.get("direction"), "direction"), + trick_condition=_opt_trick_condition(fields_dict.get("trick_condition")), + weapon_id=_opt_str(fields_dict.get("weapon_id"), "weapon_id"), + ready_trigger=_opt_ready_trigger(fields_dict.get("ready_trigger")), + destination=_int_pair(fields_dict.get("destination"), "destination"), + target_pos=_int_pair(fields_dict.get("target_pos"), "target_pos"), + aid_mode=_opt_aid_mode(fields_dict.get("aid_mode")), + metamagic=_str_tuple(fields_dict.get("metamagic"), "metamagic"), + cast_defensively=_opt_bool(fields_dict.get("cast_defensively"), "cast_defensively"), + tumble=_opt_bool(fields_dict.get("tumble"), "tumble"), + ) + + +# ---- Public parse API -------------------------------------------------------- + +def parse_actions(reply: str) -> tuple[Action, ...]: + """Parse an LLM reply into a tuple of Action objects.""" + try: + parsed = json.loads(reply) + except json.JSONDecodeError as exc: + msg = f"reply is not valid JSON: {exc}" + raise NLPolicyError(msg) from exc + if not _is_object_list(parsed): + msg = "reply must be a JSON array of action objects" + raise NLPolicyError(msg) + items = parsed + return tuple(_action_from_item(item) for item in items) + + +# ---- Fallback policy --------------------------------------------------------- + +Fallback = Callable[[tuple[Action, ...]], tuple[Action, ...]] + + +def fallback_default(menu: tuple[Action, ...]) -> tuple[Action, ...]: + """Deterministic fallback: first decisive action, then a defensive hold.""" + for kind in ("full_attack", "attack", "charge", "move", "5foot_step"): + for action in menu: + if action.kind == kind: + return (action,) + if menu: + return (menu[0],) + return (Action(kind="wait"),) + + +# ---- NLPolicy ---------------------------------------------------------------- + +@dataclass(frozen=True) +class NLPolicy: + """Turn LLM completions into Action sequences, with a scripted fallback.""" + + client: LLMClient + builder: PromptBuilder + prompt_name: str = "nl_default" + fallback: Fallback = fallback_default + + def __post_init__(self) -> None: + expected = self.builder.prompt_version(self.prompt_name) + if self.client.prompt_version != expected: + msg = ( + f"client prompt_version {self.client.prompt_version!r} does not " + f"match pack {expected!r} for {self.prompt_name!r}" + ) + raise ValueError(msg) + + def __call__(self, observation: Observation) -> tuple[Action, ...]: + menu = candidates(observation) + prompt = self.builder.build(observation, prompt_name=self.prompt_name) + reply = self.client.complete(prompt) + try: + return parse_actions(reply) + except NLPolicyError as exc: + retry_prompt = ( + f"{prompt}\n\n-- previous reply invalid --\n{exc}\n" + "Return a valid JSON action sequence." + ) + retry_reply = self.client.complete(retry_prompt) + try: + return parse_actions(retry_reply) + except NLPolicyError: + return self.fallback(menu) diff --git a/src/pf1e_simulator/prompt_builder.py b/src/pf1e_simulator/prompt_builder.py new file mode 100644 index 0000000..11ee316 --- /dev/null +++ b/src/pf1e_simulator/prompt_builder.py @@ -0,0 +1,181 @@ +"""Prompt builder for the phase-2 LLM layer (P1.4). + +Builds a deterministic prompt from an ``Observation`` plus the external +prompt pack: template + compact observation + plausible actions + JSON +action schema. The schema is derived from the ``Action`` dataclass via +``typing.get_type_hints`` so it stays in sync with the engine. +""" + +from __future__ import annotations + +import json +from dataclasses import MISSING, asdict, dataclass, fields +from types import UnionType +from typing import TYPE_CHECKING, Literal, Union, get_args, get_origin, get_type_hints + +from pf1e_simulator.candidates import candidates +from pf1e_simulator.combat import UNSUPPORTED_ACTION_KINDS, Action, Observation +from pf1e_simulator.llm_view import compact_observation + +if TYPE_CHECKING: + from pf1e_simulator.prompt import PromptPack + + +def _supported_kinds() -> tuple[str, ...]: + """Return the 27 action kinds supported by the LLM layer (declaration order).""" + kind_literal = get_type_hints(Action)["kind"] + all_kinds = get_args(kind_literal) + return tuple(k for k in all_kinds if k not in UNSUPPORTED_ACTION_KINDS) + + +_EMPTY_LITERAL_MSG = "empty Literal" +_UNSUPPORTED_LITERAL_MSG = "unsupported Literal values: {}" +_MULTI_UNION_MSG = "multi-type union not supported in Action: {}" +_UNSUPPORTED_TYPE_MSG = "unsupported type in Action schema: {}" +_TUPLE_ELLIPSIS_LEN = 2 + + +def _json_schema_base(tp: object) -> dict[str, object] | None: + """Return the JSON Schema fragment for a base type, else None.""" + if tp is type(None): + return {"type": "null"} + if tp is str: + return {"type": "string"} + if tp is int: + return {"type": "integer"} + if tp is bool: + return {"type": "boolean"} + return None + + +def _json_schema_literal(args: tuple[object, ...]) -> dict[str, object]: + """Return the JSON Schema fragment for a Literal type.""" + if not args: + raise ValueError(_EMPTY_LITERAL_MSG) + first = args[0] + if isinstance(first, str): + base_type = "string" + elif isinstance(first, int): + base_type = "integer" + else: + msg = _UNSUPPORTED_LITERAL_MSG.format(args) + raise TypeError(msg) + return {"type": base_type, "enum": list(args)} + + +def _json_schema_tuple(args: tuple[object, ...]) -> dict[str, object]: + """Return the JSON Schema fragment for a tuple type.""" + if not args: + return {"type": "array"} + if len(args) == _TUPLE_ELLIPSIS_LEN and args[1] is Ellipsis: + # tuple[T, ...] → homogeneous array + item_schema = _json_schema_for(args[0]) + return {"type": "array", "items": item_schema} + # Fixed-length tuple → prefixItems + minItems/maxItems + prefix = [_json_schema_for(arg) for arg in args] + return { + "type": "array", + "prefixItems": prefix, + "minItems": len(args), + "maxItems": len(args), + } + + +def _json_schema_union(args: tuple[object, ...]) -> dict[str, object]: + """Return the JSON Schema fragment for a nullable single-member union.""" + non_none = [a for a in args if a is not type(None)] + if len(non_none) == 1: + member_schema = _json_schema_for(non_none[0]) + # Merge "null" into the type array + member_type = member_schema.get("type") + if isinstance(member_type, str): + member_schema["type"] = [member_type, "null"] + # If member already has an enum (e.g. nullable Literal), keep it. + return member_schema + msg = _MULTI_UNION_MSG.format(args) + raise ValueError(msg) + + +def _json_schema_for(tp: object) -> dict[str, object]: + """Convert a Python type annotation to a JSON Schema fragment.""" + base_schema = _json_schema_base(tp) + if base_schema is not None: + return base_schema + + origin = get_origin(tp) + args = get_args(tp) + + if origin is Literal: + return _json_schema_literal(args) + + if origin is tuple: + return _json_schema_tuple(args) + + if origin is Union or origin is UnionType: + return _json_schema_union(args) + + msg = _UNSUPPORTED_TYPE_MSG.format(tp) + raise ValueError(msg) + + +def action_schema() -> dict[str, object]: + """JSON Schema for a single Action object.""" + hints = get_type_hints(Action) + properties = {name: _json_schema_for(tp) for name, tp in hints.items()} + # Override kind with the filtered 27-kind enum (source Literal order) + properties["kind"] = {"type": "string", "enum": list(_supported_kinds())} + # Only 'kind' lacks a default + required = [ + f.name + for f in fields(Action) + if f.default is MISSING and f.default_factory is MISSING + ] + return { + "type": "object", + "properties": properties, + "required": required, + "additionalProperties": False, + } + + +def render_candidates(actions: tuple[Action, ...]) -> tuple[str, ...]: + """Render the plausible-action menu as numbered lines with compact JSON.""" + lines: list[str] = [] + for index, action in enumerate(actions): + payload = { + key: value + for key, value in asdict(action).items() + if key != "kind" and value is not None and value != () and value is not False + } + suffix = f" {json.dumps(payload, sort_keys=True, separators=(',', ':'))}" if payload else "" + lines.append(f"{index}. {action.kind}{suffix}") + return tuple(lines) + + +@dataclass(frozen=True) +class PromptBuilder: + """Builds the complete LLM prompt from a prompt pack and an observation.""" + + pack: PromptPack + + def prompt_version(self, name: str) -> str: + """Pack version plus per-prompt content hash (cache-key component, D6).""" + return self.pack.prompt_version(name) + + def build(self, observation: object, *, prompt_name: str = "nl_default") -> str: + """Assemble the full prompt for the given observation.""" + if not isinstance(observation, Observation): + msg = "observation must be an Observation instance" + raise TypeError(msg) + + template = self.pack.get(prompt_name) + obs_text = compact_observation(observation).rstrip("\n") + menu = candidates(observation) + actions_text = "\n".join(render_candidates(menu)) + schema_text = json.dumps(action_schema(), sort_keys=True, indent=2) + return "\n\n".join([ + template, + "## observation\n" + obs_text, + "## plausible actions\n" + actions_text, + "## action schema\n" + schema_text, + ]) diff --git a/tests/test_phase2_nl_policy.py b/tests/test_phase2_nl_policy.py new file mode 100644 index 0000000..ac77c18 --- /dev/null +++ b/tests/test_phase2_nl_policy.py @@ -0,0 +1,280 @@ +"""Tests for the natural-language policy (P1.3).""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from pf1e_simulator.client import FakeLLMClient, ResponseCache, ScriptExhaustedError +from pf1e_simulator.combat import Action, Observation +from pf1e_simulator.nl_policy import ( + NLPolicy, + NLPolicyError, + fallback_default, + parse_actions, +) +from pf1e_simulator.prompt import PromptPack +from pf1e_simulator.prompt_builder import PromptBuilder + +_FIXTURE_PACK = Path(__file__).resolve().parent / "fixtures" / "prompt_pack" + + +def _obs() -> Observation: + """Minimal valid observation for deterministic tests.""" + return Observation( + round=1, + acting_combatant=0, + combatants=( + {"id": "goblin", "side": "players", "active": True, "pos": (1, 1)}, + ), + pairs=(), + terrain=((None,),), + elevation=((0,),), + ) + + +def _pack() -> PromptPack: + return PromptPack.load(_FIXTURE_PACK) + + +def _builder() -> PromptBuilder: + return PromptBuilder(pack=_pack()) + + +def _make_client(script: list[str], *, builder: PromptBuilder, **kwargs) -> FakeLLMClient: + """Create a FakeLLMClient with the correct prompt_version from the builder.""" + return FakeLLMClient( + script=script, + prompt_version=builder.prompt_version("nl_default"), + **kwargs, + ) + + +def _make_policy(client: FakeLLMClient, builder: PromptBuilder) -> NLPolicy: + return NLPolicy(client=client, builder=builder) + + +@pytest.fixture +def builder() -> PromptBuilder: + """A builder over the fixture pack, shared by the NLPolicy integration tests.""" + return _builder() + + +# ---- parse_actions tests ----------------------------------------------------- + +def test_parse_actions_ok() -> None: + reply = '[{"kind": "move", "destination": [2, 3]}]' + actions = parse_actions(reply) + assert actions == (Action(kind="move", destination=(2, 3)),) + + +def test_parse_actions_empty_array() -> None: + assert parse_actions("[]") == () + + +def test_parse_actions_invalid_json() -> None: + with pytest.raises(NLPolicyError, match=r"reply is not valid JSON"): + parse_actions("not json") + + +def test_parse_actions_not_a_list() -> None: + with pytest.raises(NLPolicyError, match=r"reply must be a JSON array of action objects"): + parse_actions('{"kind": "attack"}') + + +def test_parse_actions_unknown_field() -> None: + with pytest.raises(NLPolicyError, match=r"unknown action field\(s\): bogus"): + parse_actions('[{"kind": "attack", "bogus": 1}]') + + +def test_parse_actions_missing_kind() -> None: + with pytest.raises(NLPolicyError, match=r"missing or unknown 'kind'"): + parse_actions('[{"target_id": "orc"}]') + + +def test_parse_actions_unknown_kind() -> None: + with pytest.raises(NLPolicyError, match=r"missing or unknown 'kind'"): + parse_actions('[{"kind": "dance"}]') + + +def test_parse_actions_channel_mode_invalid() -> None: + with pytest.raises(NLPolicyError, match=r"'channel_mode' must be one of: heal, harm"): + parse_actions('[{"kind": "channel_energy", "channel_mode": "loud"}]') + + +def test_parse_actions_destination_invalid_not_pair() -> None: + with pytest.raises(NLPolicyError, match=r"'destination' must be a \[row, col\] pair"): + parse_actions('[{"kind": "move", "destination": [2]}]') + + +def test_parse_actions_destination_invalid_strings() -> None: + with pytest.raises(NLPolicyError, match=r"'destination' must be a \[row, col\] pair"): + parse_actions('[{"kind": "move", "destination": ["2", "3"]}]') + + +def test_parse_actions_aid_mode_invalid() -> None: + with pytest.raises(NLPolicyError, match=r"'aid_mode' must be one of: attack, ac"): + parse_actions('[{"kind": "aid_another", "aid_mode": "damage"}]') + + +def test_parse_actions_trick_condition_invalid() -> None: + with pytest.raises( + NLPolicyError, + match=( + r"'trick_condition' must be one of: dazzled, blinded, " + r"deafened, entangled, shaken, sickened" + ), + ): + parse_actions('[{"kind": "cast_spell", "trick_condition": "nauseated"}]') + + +def test_parse_actions_ready_trigger_invalid() -> None: + with pytest.raises( + NLPolicyError, + match=( + r"'ready_trigger' must be one of: enemy_enters_reach, " + r"enemy_casts, enemy_attacks, enemy_hits_ally" + ), + ): + parse_actions('[{"kind": "ready", "ready_trigger": "enemy_sneezed"}]') + + +def test_parse_actions_channel_exclude_not_list() -> None: + with pytest.raises(NLPolicyError, match=r"'channel_exclude' must be a list of strings"): + parse_actions('[{"kind": "channel_energy", "channel_exclude": "orc"}]') + + +def test_parse_actions_channel_exclude_ok() -> None: + actions = parse_actions('[{"kind": "channel_energy", "channel_exclude": ["orc"]}]') + assert actions[0].channel_exclude == ("orc",) + + +def test_parse_actions_tumble_not_bool() -> None: + with pytest.raises(NLPolicyError, match=r"'tumble' must be a boolean"): + parse_actions('[{"kind": "attack", "tumble": "yes"}]') + + +def test_parse_actions_cast_defensively_not_bool() -> None: + with pytest.raises(NLPolicyError, match=r"'cast_defensively' must be a boolean"): + parse_actions('[{"kind": "cast_spell", "cast_defensively": 1}]') + + +def test_parse_actions_metamagic_ok() -> None: + actions = parse_actions('[{"kind": "cast_spell", "metamagic": ["Empower", "Maximize"]}]') + assert actions[0].metamagic == ("Empower", "Maximize") + + +# ---- fallback_default tests -------------------------------------------------- + +def test_fallback_default_prefers_full_attack() -> None: + menu = ( + Action(kind="total_defense"), + Action(kind="full_attack", target_id="orc"), + Action(kind="move"), + ) + assert fallback_default(menu) == (Action(kind="full_attack", target_id="orc"),) + + +def test_fallback_default_prefers_attack_over_charge() -> None: + menu = ( + Action(kind="charge", target_id="orc"), + Action(kind="attack", target_id="orc"), + ) + assert fallback_default(menu) == (Action(kind="attack", target_id="orc"),) + + +def test_fallback_default_prefers_charge_over_move() -> None: + menu = ( + Action(kind="move", target_id="orc"), + Action(kind="charge", target_id="orc"), + ) + assert fallback_default(menu) == (Action(kind="charge", target_id="orc"),) + + +def test_fallback_default_falls_back_to_first() -> None: + menu = (Action(kind="total_defense"), Action(kind="delay")) + assert fallback_default(menu) == (Action(kind="total_defense"),) + + +def test_fallback_default_empty_menu_returns_wait() -> None: + assert fallback_default(()) == (Action(kind="wait"),) + + +# ---- NLPolicy integration tests --------------------------------------------- + +def test_nl_policy_parse_ok(builder: PromptBuilder) -> None: + """Parse OK path with valid JSON from FakeLLMClient.""" + script = ['[{"kind": "move", "destination": [2, 3]}]'] + client = _make_client(script, builder=builder) + policy = _make_policy(client, builder) + + actions = policy(_obs()) + assert actions == (Action(kind="move", destination=(2, 3)),) + assert client.stats.calls == 1 + + +def test_nl_policy_retry_then_succeeds() -> None: + """One bad reply, then a good one → parsed successfully.""" + builder = _builder() + script = ["not json", '[{"kind": "move", "destination": [2, 3]}]'] + client = _make_client(script, builder=builder) + policy = _make_policy(client, builder) + + actions = policy(_obs()) + assert actions == (Action(kind="move", destination=(2, 3)),) + assert client.stats.calls == 2 + + +def test_nl_policy_retry_then_fallback() -> None: + """Two bad replies → fallback used.""" + builder = _builder() + script = ["not json", "not json"] + client = _make_client(script, builder=builder) + policy = _make_policy(client, builder) + + # Minimal obs menu is (total_defense, delay) + actions = policy(_obs()) + # fallback_default picks first decisive action, but menu has only defensive + assert actions == (Action(kind="total_defense"),) + assert client.stats.calls == 2 + + +def test_nl_policy_prompt_version_mismatch() -> None: + """Client default prompt_version vs builder pack version → ValueError.""" + builder = _builder() + # Use default prompt_version ("fixture:fake:0") which differs from pack + client = FakeLLMClient(script=['[{"kind": "wait"}]']) + with pytest.raises(ValueError, match=r"prompt_version"): + _make_policy(client, builder) + + +def test_nl_policy_script_exhausted_propagates() -> None: + """ScriptExhaustedError from FakeLLMClient propagates (not caught).""" + builder = _builder() + client = _make_client([], builder=builder) + policy = _make_policy(client, builder) + + with pytest.raises(ScriptExhaustedError): + policy(_obs()) + + +def test_nl_policy_shared_cache_hit() -> None: + """Two clients sharing a ResponseCache: second call hits cache.""" + builder = _builder() + cache = ResponseCache() + # First client has one scripted response + client1 = _make_client(['[{"kind": "wait"}]'], builder=builder, cache=cache) + # Second client has NO script; relies on cache hit + client2 = _make_client([], builder=builder, cache=cache) + policy1 = _make_policy(client1, builder) + policy2 = _make_policy(client2, builder) + + obs = _obs() + actions1 = policy1(obs) + actions2 = policy2(obs) + + assert actions2 == actions1 == (Action(kind="wait"),) + # Second call was a cache hit (no ScriptExhaustedError) + assert client2.stats.cache_hits == 1 + assert client1.stats.cache_misses == 1 diff --git a/tests/test_phase2_offline_battle.py b/tests/test_phase2_offline_battle.py new file mode 100644 index 0000000..ae79cce --- /dev/null +++ b/tests/test_phase2_offline_battle.py @@ -0,0 +1,87 @@ +"""Offline battles driven by FakeLLMClient through the NL policy (P1.3/P1.4).""" + +from __future__ import annotations + +from pathlib import Path + +from pf1e_simulator.client import FakeLLMClient +from pf1e_simulator.loaders import load_combatant +from pf1e_simulator.map import load_map +from pf1e_simulator.nl_policy import NLPolicy +from pf1e_simulator.prompt import PromptPack +from pf1e_simulator.prompt_builder import PromptBuilder +from pf1e_simulator.runner import EncounterSpec, Side, run_encounter + +ROOT = Path(__file__).resolve().parents[1] +MAP = ROOT / "data" / "maps" / "sample_arena.yaml" + + +def _c(path: str): + return load_combatant(ROOT / "data" / path) + + +def _duel_spec() -> EncounterSpec: + """A fast 1v1 (goblin archer vs orc) on the sample arena.""" + return EncounterSpec( + map=load_map(MAP), + round_cap=20, + sides=( + Side(name="players", combatants=(_c("monsters/goblin.json"),), zone="A"), + Side(name="monsters", combatants=(_c("monsters/orc.json"),), zone="B"), + ), + ) + + +def _pack() -> PromptPack: + return PromptPack.load(ROOT / "tests" / "fixtures" / "prompt_pack") + + +def _builder() -> PromptBuilder: + return PromptBuilder(pack=_pack()) + + +def _make_client(script: list[str], *, builder: PromptBuilder) -> FakeLLMClient: + return FakeLLMClient( + script=script, + prompt_version=builder.prompt_version("nl_default"), + ) + + +def test_offline_battle_total_defense_draw() -> None: + """Both sides play total_defense → draw at round_cap (20 rounds, 40 calls).""" + spec = _duel_spec() + builder = _builder() + # 60 responses is plenty for 40 turns (20 rounds x 2 combatants) + script = ['[{"kind": "total_defense"}]'] * 60 + client = _make_client(script, builder=builder) + policy = NLPolicy(client=client, builder=builder) + + report, _outcomes = run_encounter(spec, runs=1, seed=1, llm_policy=policy, collect_runs=True) + + # Draw at round_cap + assert report.draws == 1 + assert report.wins.get("players", 0) == 0 + assert report.wins.get("monsters", 0) == 0 + assert report.avg_rounds == 20 + # 20 rounds x 2 combatants = 40 policy invocations + assert client.stats.calls == 40 + + +def test_offline_battle_fallback() -> None: + """Fallback-driven battle: script of invalid JSON forces fallback on every turn.""" + spec = _duel_spec() + builder = _builder() + # Each turn: 2 completes (first parse fail, retry parse fail → fallback). + # Max 20 rounds x 2 combatants x 2 = 80 completes. Use 100 for margin. + script = ["not json"] * 100 + client = _make_client(script, builder=builder) + policy = NLPolicy(client=client, builder=builder) + + report, _outcomes = run_encounter(spec, runs=1, seed=1, llm_policy=policy, collect_runs=True) + + # The battle should complete (not exhaust script) + assert report.runs == 1 + assert report.draws + sum(report.wins.values()) == 1 + # At least some calls happened; script not exhausted + assert client.stats.calls > 0 + assert client.stats.calls <= 100 diff --git a/tests/test_phase2_prompt_builder.py b/tests/test_phase2_prompt_builder.py new file mode 100644 index 0000000..0b82090 --- /dev/null +++ b/tests/test_phase2_prompt_builder.py @@ -0,0 +1,223 @@ +"""Tests for the phase-2 prompt builder (P1.4).""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import fields +from pathlib import Path + +from pf1e_simulator.combat import Action, Observation +from pf1e_simulator.nl_policy import KindValue +from pf1e_simulator.nl_policy import get_args as _get_args +from pf1e_simulator.prompt import PromptPack +from pf1e_simulator.prompt_builder import ( + PromptBuilder, + _supported_kinds, + action_schema, + render_candidates, +) + +_FIXTURE_PACK = Path(__file__).resolve().parent / "fixtures" / "prompt_pack" +_FIXTURE_VERSION = "0.1.0-test" +_FIXTURE_PROMPT_NAME = "nl_default" + + +def _obs() -> Observation: + """Minimal valid observation for deterministic tests.""" + return Observation( + round=1, + acting_combatant=0, + combatants=( + {"id": "goblin", "side": "players", "active": True, "pos": (1, 1)}, + ), + pairs=(), + terrain=((None,),), + elevation=((0,),), + ) + + +def _pack() -> PromptPack: + return PromptPack.load(_FIXTURE_PACK) + + +def _builder() -> PromptBuilder: + return PromptBuilder(pack=_pack()) + + +def test_supported_kinds_filters_unsupported() -> None: + kinds = _supported_kinds() + assert "swift" not in kinds + assert "free" not in kinds + assert "full_round" not in kinds + # 27 kinds remain + assert len(kinds) == 27 + # Order preserved from source Literal + assert kinds == ( + "attack", "move", "wait", "immediate", "full_attack", "charge", + "withdraw", "5foot_step", "cast_spell", "maneuver", "channel_energy", + "hex", "deed", "spell_combat", "inspire_courage", "retrieve_weapon", + "repair_weapon", "activate_power_attack", "delay", "ready", + "total_defense", "fight_defensively", "stand_up", "drop_prone", + "aid_another", "run", "hide", + ) + + +def test_action_schema_structure() -> None: + schema = action_schema() + assert schema["type"] == "object" + assert schema["required"] == ["kind"] + assert schema["additionalProperties"] is False + + props = schema["properties"] + # Properties match Action fields exactly + action_field_names = {f.name for f in fields(Action)} + assert set(props.keys()) == action_field_names + + # Kind enum is the filtered 27 + kind_prop = props["kind"] + assert kind_prop["type"] == "string" + assert kind_prop["enum"] == list(_supported_kinds()) + + +def test_action_schema_spot_check_fragments() -> None: + schema = action_schema() + props = schema["properties"] + + # target_id accepts either a string or null + assert props["target_id"] == {"type": ["string", "null"]} + + # destination is a nullable two-int pair (prefixItems + min/maxItems) + dest = props["destination"] + assert dest["type"] == ["array", "null"] + assert dest["prefixItems"] == [{"type": "integer"}, {"type": "integer"}] + assert dest["minItems"] == 2 + assert dest["maxItems"] == 2 + + # channel_exclude: string[] (array of strings) + ch_ex = props["channel_exclude"] + assert ch_ex["type"] == "array" + assert ch_ex["items"] == {"type": "string"} + + # cast_defensively is a boolean field + assert props["cast_defensively"] == {"type": "boolean"} + + # tumble is a boolean field + assert props["tumble"] == {"type": "boolean"} + + # channel_mode: nullable enum + ch_mode = props["channel_mode"] + assert ch_mode["type"] == ["string", "null"] + assert ch_mode["enum"] == ["heal", "harm"] + + # aid_mode: nullable enum + aid = props["aid_mode"] + assert aid["type"] == ["string", "null"] + assert aid["enum"] == ["attack", "ac"] + + # trick_condition: nullable enum (6 values) + tc = props["trick_condition"] + assert tc["type"] == ["string", "null"] + assert tc["enum"] == ["dazzled", "blinded", "deafened", "entangled", "shaken", "sickened"] + + # ready_trigger: nullable enum (4 values) + rt = props["ready_trigger"] + assert rt["type"] == ["string", "null"] + assert rt["enum"] == ["enemy_enters_reach", "enemy_casts", "enemy_attacks", "enemy_hits_ally"] + + # metamagic: string[] + meta = props["metamagic"] + assert meta["type"] == "array" + assert meta["items"] == {"type": "string"} + + # class_ability is a nullable string field + assert props["class_ability"] == {"type": ["string", "null"]} + + +def test_render_candidates() -> None: + # Single action with payload + rendered = render_candidates((Action(kind="move", destination=(2, 3)),)) + assert rendered == ('0. move {"destination":[2,3]}',) + + # Action with no payload (all defaults/None) + rendered = render_candidates((Action(kind="total_defense"),)) + assert rendered == ("0. total_defense",) + + # Action with target_id + rendered = render_candidates((Action(kind="attack", target_id="orc"),)) + assert rendered == ('0. attack {"target_id":"orc"}',) + + # Empty menu + assert render_candidates(()) == () + + # Multiple actions + rendered = render_candidates(( + Action(kind="attack", target_id="a"), + Action(kind="move", destination=(1, 1)), + )) + assert rendered == ( + '0. attack {"target_id":"a"}', + '1. move {"destination":[1,1]}', + ) + + +def test_prompt_builder_build_deterministic() -> None: + builder = _builder() + obs = _obs() + prompt1 = builder.build(obs) + prompt2 = builder.build(obs) + assert prompt1 == prompt2 + + +def test_prompt_builder_sections_present() -> None: + builder = _builder() + obs = _obs() + prompt = builder.build(obs) + + # Template content + assert "tactical director" in prompt + assert "battle state" in prompt + + # Sections + assert "## observation\n" in prompt + assert "## plausible actions\n" in prompt + assert "## action schema\n" in prompt + + # Schema is valid JSON and contains kind enum + schema_part = prompt.split("## action schema\n")[1] + parsed = json.loads(schema_part) + assert parsed["properties"]["kind"]["enum"] == list(_supported_kinds()) + + +def test_prompt_version_format() -> None: + builder = _builder() + pv = builder.prompt_version(_FIXTURE_PROMPT_NAME) + assert pv.startswith(f"{_FIXTURE_VERSION}:{_FIXTURE_PROMPT_NAME}:") + digest = pv.split(":")[-1] + assert len(digest) == 12 + assert all(c in "0123456789abcdef" for c in digest) + # Cross-check against actual pack content + pack = _pack() + content = pack.get(_FIXTURE_PROMPT_NAME) + expected_digest = hashlib.sha256(content.encode("utf-8")).hexdigest()[:12] + assert digest == expected_digest + + +def test_prompt_builder_custom_prompt_name() -> None: + builder = _builder() + obs = _obs() + # nl_default is the only fixture, but the API accepts any name + prompt = builder.build(obs, prompt_name=_FIXTURE_PROMPT_NAME) + assert "tactical director" in prompt + + +def test_prompt_version_delegates_to_pack() -> None: + builder = _builder() + pack = _pack() + assert builder.prompt_version(_FIXTURE_PROMPT_NAME) == pack.prompt_version(_FIXTURE_PROMPT_NAME) + + +def test_action_schema_cross_pin_with_nl_policy_kindvalue() -> None: + """The kind enum in the schema must match nl_policy.KindValue exactly.""" + schema = action_schema() + assert schema["properties"]["kind"]["enum"] == list(_get_args(KindValue))