feat(phase2): observation-only LLMPolicy entry (D1) + pinned schema_version (P0.1)

- LLMPolicy is now Callable[[Observation], tuple[Action, ...]]: the snapshot
  is the policy's sole information source; the live CombatantState is no
  longer passed to new-style policies.
- CombatEngine._llm_adapter keeps a compatibility path for pre-D1
  two-argument callables: the signature is inspected ONCE at wrap time
  (_accepts_state) and legacy policies still receive (observation, state).
- Observation.to_dict() carries schema_version (OBSERVATION_SCHEMA_VERSION=1)
  as its first pinned key; bump only on removal/meaning change.
- Pinned keyset tests updated (test_observation, test_observation_enriched);
  new tests: observation-only e2e, legacy 2-arg e2e through the adapter,
  schema_version pin + JSON-serializability.
- README: documents the new entry signature and the schema_version key.
This commit is contained in:
2026-09-21 11:10:28 +02:00
parent 3fd9860b5c
commit 3c3e30634e
6 changed files with 142 additions and 21 deletions
+1 -1
View File
@@ -229,7 +229,7 @@ def test_results_visible_in_next_turn_observation() -> None:
# the second invocation must already see the first turn's rejection.
seen: list[Observation] = []
def llm_policy(obs: Observation, state: CombatantState) -> tuple[Action, ...]:
def llm_policy(obs: Observation) -> tuple[Action, ...]:
seen.append(obs)
if len(seen) == 1:
return (Action(kind="move"),)
+2 -2
View File
@@ -1,7 +1,7 @@
"""Package U8 — final validation: an observation-only LLM policy drives the
Phase-2 capabilities (U1-U7b) end-to-end through the real engine.
Each test wires a ``llm_policy`` (``LLMPolicy(observation, state)``) that
Each test wires a ``llm_policy`` (``LLMPolicy(observation)``) that
reads ONLY the frozen Observation snapshot — never ``engine.*`` — and emits
one newly-unblocked tactical decision per capability: ``ready`` (U4),
``delay`` (U3), ``total_defense`` (U5a), ``move`` with an explicit
@@ -121,7 +121,7 @@ def _counting(decide: Decide) -> LLMPolicy:
"""LLMPolicy calling ``decide(id, call_no, obs)`` per turn (obs only)."""
calls: dict[str, int] = {}
def policy(obs: Observation, _state: CombatantState) -> tuple[Action, ...]:
def policy(obs: Observation) -> tuple[Action, ...]:
cid = str(_me(obs)["id"])
n = calls.get(cid, 0)
calls[cid] = n + 1
+73 -5
View File
@@ -9,9 +9,12 @@ Covers the pinned schema:
- Terrain/elevation layers.
- Frozen copies: mutating live state after construction does not change the
already-built Observation.
- ``to_dict()`` is JSON-serializable and exposes no ``engine.*`` callables.
- ``to_dict()`` is JSON-serializable, carries the pinned ``schema_version``
key, and exposes no ``engine.*`` callables.
- The ``llm_policy`` adapter builds the Observation at the point of policy
invocation and delegates to ``llm_policy(observation, state)``.
invocation and delegates to ``llm_policy(observation)`` (D1: observation
only); a legacy ``llm_policy(observation, state)`` callable keeps working
through the compatibility adapter.
"""
from __future__ import annotations
@@ -20,7 +23,13 @@ import json
import pytest
from pf1e_simulator.combat import Action, CombatantState, CombatEngine, Observation
from pf1e_simulator.combat import (
OBSERVATION_SCHEMA_VERSION,
Action,
CombatantState,
CombatEngine,
Observation,
)
from pf1e_simulator.conditions import SHAKEN
from pf1e_simulator.dice import parse_dice
from pf1e_simulator.effects import ActiveCondition
@@ -441,11 +450,11 @@ def test_observation_frozen_copies() -> None:
def test_llm_policy_wrapper_e2e() -> None:
"""Given: a CombatEngine with an llm_policy that reads ONLY the observation
"""Given: a CombatEngine with an observation-only llm_policy (D1 signature)
When: the battle runs
Then: the policy's decisions are honored and the transcript is coherent."""
def llm_policy(obs: Observation, state: CombatantState) -> tuple:
def llm_policy(obs: Observation) -> tuple:
# Read only the observation: full_attack if the target is adjacent,
# otherwise move toward it.
target_id = obs.combatants[obs.pairs[0]["target_index"]]["id"]
@@ -470,6 +479,65 @@ def test_llm_policy_wrapper_e2e() -> None:
assert any("move" in line for line in result.transcript)
def test_legacy_two_arg_llm_policy_still_works() -> None:
"""Given: a pre-D1 llm_policy taking (observation, state)
When: the battle runs through the compatibility adapter
Then: the policy receives BOTH arguments and its decisions are honored."""
seen_states: list[CombatantState] = []
def legacy_policy(obs: Observation, state: CombatantState) -> tuple:
seen_states.append(state)
assert state.combatant.id == obs.combatants[obs.acting_combatant]["id"]
target_id = obs.combatants[obs.pairs[0]["target_index"]]["id"]
return (Action(kind="move", target_id=target_id),)
hero = make_combatant("hero")
foe = make_combatant("foe")
hero_state = make_state(hero, "players", (0, 0))
foe_state = make_state(foe, "monsters", (0, 3))
engine = CombatEngine(
ScriptedRng([10, 9, 8, 15, 3, 5]),
make_grid(),
[hero_state, foe_state],
round_cap=1,
llm_policy=legacy_policy,
)
result = engine.run()
assert seen_states, "legacy policy was never invoked"
assert any("move" in line for line in result.transcript)
def test_observation_schema_version_pinned() -> None:
"""Given: a built Observation
When: to_dict() is called
Then: the top-level key set is exactly the pinned keys and
schema_version == OBSERVATION_SCHEMA_VERSION, JSON-serializable."""
hero = make_combatant("hero")
foe = make_combatant("foe")
hero_state = make_state(hero, "players", (0, 0))
foe_state = make_state(foe, "monsters", (0, 3))
engine = make_engine([hero_state, foe_state])
data = engine._build_observation(hero_state).to_dict()
assert data["schema_version"] == OBSERVATION_SCHEMA_VERSION
assert set(data) == {
"schema_version",
"round",
"acting_combatant",
"combatants",
"pairs",
"terrain",
"elevation",
"threatened",
"ranged_threatened",
"terrain_semantics",
"condition_definitions",
"initiative_order",
}
json.dumps(data) # must not raise
# ---------------------------------------------------------------------------
# 12. policy + llm_policy → ValueError
# ---------------------------------------------------------------------------
+4 -2
View File
@@ -1051,7 +1051,8 @@ def test_observation_enriched_terrain_semantics_no_callable_no_share() -> None:
def test_observation_enriched_e5_existing_keys_preserved() -> None:
"""Given: an Observation built for the L5 fields
When: the top-level dict is inspected
Then: all pre-existing top-level keys remain (round, acting_combatant,
Then: all pre-existing top-level keys remain (schema_version, round,
acting_combatant,
combatants, pairs, terrain, elevation) and the new keys
(threatened, ranged_threatened, terrain_semantics,
condition_definitions, initiative_order) are added — exact key set."""
@@ -1063,6 +1064,7 @@ def test_observation_enriched_e5_existing_keys_preserved() -> None:
obs = _snapshot(engine, hero_state)
assert set(obs) == {
"schema_version",
"round",
"acting_combatant",
"combatants",
@@ -1138,7 +1140,7 @@ def _best_usable_attack(obs: Observation) -> dict[str, object] | None:
return best
def _enriched_llm_policy(obs: Observation, state: CombatantState) -> tuple[Action, ...]:
def _enriched_llm_policy(obs: Observation) -> tuple[Action, ...]:
"""Read ONLY the observation — never state.combatant.* nor engine.*."""
cast = _castable_spell_action(obs)
if cast is not None: