feat(phase2): compact observation view for LLM prompts

Build a decision-sufficient ObservationView from the raw snapshot and
render it as a deterministic line-based prompt string. The raw to_dict()
is ~64 KB (~16k tokens) for a 7-combatant encounter; the compact view
renders at ~15.7 KB (~3.9k tokens, 24% of raw), under the 4k-token
ceiling of the phase-2 architecture doc.

Compaction: sparse terrain/elevation/threatened grids, per-combatant
key selection (occupied_cells, abilities, skills, initiative_mod/total,
temp_hp_rounds_left, features, carried_items, weapon_damage, aware,
alignment, armor_spell_failure, arcane_caster, run_rounds dropped),
attacks compacted to formula+type damage strings, spells to 7
decision-relevant fields, pair target_index replaced by the stable
combatant id, condition_definitions dropped. Attacks and spells render
as positional tuples declared by attack_fields/spell_fields legend
lines, with '-' for missing values.

Tests pin the size bounds (compact < 0.25 x raw and < 16000 bytes), the
sparse-grid contract, the dropped-key contract and the deterministic
id/render contract on a caster-heavy 3v4 representative battle.
This commit is contained in:
2026-09-21 13:21:33 +02:00
parent 25418f164f
commit 9da9537995
2 changed files with 824 additions and 0 deletions
+545
View File
@@ -0,0 +1,545 @@
"""Compact, decision-sufficient observation view for LLM policies (P0.2).
The raw ``Observation.to_dict()`` snapshot is ~64 KB (~16k tokens) for a
7-combatant encounter - too large to feed an LLM every turn (docs
``phase-2-llm-architecture.md`` section 16, target ~2-4k tokens). This module
builds a compact ``ObservationView`` that keeps every tactically relevant
fact while dropping duplicates, derivables and verbose expansions, and
renders it as a deterministic, line-based prompt string.
Compaction rules
----------------
- ``terrain``: sparse list of ``(row, col, type)`` for non-default cells
only. A cell is default when its semantics are ``move_cost == 1``,
``blocks_los == False``, ``cover == False`` and ``cover_bonus == 0``.
- ``elevation``: sparse list of ``(row, col, height)`` for non-zero cells.
- ``threatened`` / ``ranged_threatened``: sparse list of ``(row, col)`` for
True cells (empty when no cell is covered).
- per-combatant: only decision-relevant keys are kept (see
``_COMPACT_COMBATANT_KEYS``). Dropped: ``occupied_cells`` (derivable from
``pos`` + ``size``), ``abilities``/``skills`` (raw scores, not used by the
candidate generator), ``initiative_mod``/``initiative_total`` (the
top-level order suffices), ``temp_hp_rounds_left`` (``temp_hp`` is the
current value), ``features``/``feature_details`` (the resources and
``sneak_dice`` already expose the class abilities), ``carried_items``,
``weapon_damage``, ``aware``, ``alignment``, ``armor_spell_failure``,
``arcane_caster`` (all zero in the data) and ``run_rounds``.
- attacks: compacted to ``{id, name, kind, attack_bonus, damage (strings
like "1d6 slashing"), damage_bonus, crit_range, crit_mult, reach_ft,
range_increment_ft, count, firearm}`` - sunder-only object stats
(``hardness``/``hit_points``) and ``splash`` are dropped.
- spells: compacted to ``{name, level, range_ft, save, dc, cast_time,
area}`` - ``summary`` (the bulk of the raw size), ``school``,
``half_on_save``, ``save_negates``, ``harmless`` and ``components`` are
dropped (no component failure in the engine's data).
- per-pair: all decision helpers are kept unchanged, except ``target_index``
is replaced by the stable combatant ``target`` id.
- ``acting_combatant`` is rendered as the combatant id (not the index).
- ``condition_definitions`` is dropped (condition names suffice).
- every collection is deterministically ordered (fixed key order, row-major
or the raw snapshot's own sorted order).
Render format
-------------
The render is line-based ``key=value`` text. The nested ``attacks`` and
``spells`` expansions are rendered as positional tuples (the field order is
declared once in the ``attack_fields`` / ``spell_fields`` legend lines) with
``-`` standing for a missing value (``None`` save/dc/area, no range
increment); this is what brings the view under the ~4k-token ceiling.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import TypeGuard
from .combat import OBSERVATION_SCHEMA_VERSION, Observation
# Per-combatant keys kept in the compact view, in deterministic order.
# Everything else in the raw snapshot is derivable, verbose, or not
# decision-relevant for the candidate generator and the LLM.
_COMPACT_COMBATANT_KEYS: tuple[str, ...] = (
"id",
"side",
"pos",
"hp",
"hp_max",
"nonlethal",
"active",
"dead",
"dying",
"stabilized",
"conditions",
"resources_left",
"weapon_broken",
"weapon_destroyed",
"weapon_stolen",
"weapon_dropped",
"grappled_with",
"hidden_from",
"moved_this_turn",
"power_attack",
"defensive_ac_bonus",
"fight_defensively",
"aid_attack_bonus",
"aid_ac_bonus",
"running",
"run_rest_rounds",
"bleed_damage",
"recurring_damage",
"aoo_remaining",
"name",
"level",
"size",
"natural_reach_ft",
"creature_type",
"bab",
"casting_ability",
"dr",
"metamagic_feats",
"summon_duration",
"immediate_used",
"deed_up_close_and_deadly",
"active_buffs",
"one_shot_bonuses",
"temp_hp",
"speed_land_ft",
"speed_fly_ft",
"speed_effective_ft",
"ac",
"saves",
"cmb",
"cmd",
"attacks",
"spells",
"spell_resistance",
"resistances",
"immunities",
"sneak_dice",
"engaged_in_melee",
"last_turn_results",
"readied",
"pending_cast",
)
# Per-pair keys kept (all decision helpers), in the raw snapshot's order.
# ``target_index`` is replaced by the stable combatant ``target`` id.
_PAIR_KEYS: tuple[str, ...] = (
"distance_ft",
"weapon_for",
"weapon_reason",
"range_penalty",
"can_charge",
"charge_reason",
"can_step5_to_attack",
"has_line_of_effect",
"flanking",
"movement_cost_ft",
"target_threatens_me",
"enemies_threatening_me",
"cover_bonus",
)
# Attack keys kept in the compact expansion (sunder-only object stats and
# the splash marker are dropped).
_ATTACK_KEYS: tuple[str, ...] = (
"id",
"name",
"kind",
"attack_bonus",
"damage",
"damage_bonus",
"crit_range",
"crit_mult",
"reach_ft",
"range_increment_ft",
"count",
"firearm",
)
# Spell keys kept in the compact expansion (the summary is the bulk of the
# raw size; school/components/save flags are not decision-relevant).
_SPELL_KEYS: tuple[str, ...] = (
"name",
"level",
"range_ft",
"save",
"dc",
"cast_time",
"area",
)
@dataclass(frozen=True)
class ObservationView:
"""Compact, decision-sufficient view of one combat observation.
All collections are deterministically ordered; ``terrain``/``elevation``/
``threatened``/``ranged_threatened`` are sparse (see module docstring).
``acting_combatant`` is the acting combatant's id.
"""
schema_version: int
round: int
acting_combatant: str
dimensions: tuple[int, int]
terrain: tuple[tuple[int, int, str], ...]
elevation: tuple[tuple[int, int, int], ...]
threatened: tuple[tuple[int, int], ...]
ranged_threatened: tuple[tuple[int, int], ...]
terrain_semantics: tuple[dict[str, object], ...]
initiative_order: tuple[str, ...]
combatants: tuple[dict[str, object], ...]
pairs: tuple[dict[str, object], ...]
def build_view(observation: Observation) -> ObservationView:
"""Build a compact ``ObservationView`` from a raw ``Observation``."""
height = len(observation.terrain)
width = len(observation.terrain[0]) if height else 0
semantics_by_type = {
str(sem["type"]): sem for sem in observation.terrain_semantics
}
terrain = tuple(
(r, c, cell_type)
for r, row in enumerate(observation.terrain)
for c, cell_type in enumerate(row)
if cell_type is not None
and not _is_default_terrain(cell_type, semantics_by_type)
)
elevation = tuple(
(r, c, height_at)
for r, row in enumerate(observation.elevation)
for c, height_at in enumerate(row)
if height_at != 0
)
combatants = tuple(
_compact_combatant(combatant) for combatant in observation.combatants
)
return ObservationView(
schema_version=OBSERVATION_SCHEMA_VERSION,
round=observation.round,
acting_combatant=_combatant_id(combatants, observation.acting_combatant),
dimensions=(height, width),
terrain=terrain,
elevation=elevation,
threatened=_sparse_bool_grid(observation.threatened),
ranged_threatened=_sparse_bool_grid(observation.ranged_threatened),
terrain_semantics=observation.terrain_semantics,
initiative_order=observation.initiative_order,
combatants=combatants,
pairs=tuple(_compact_pair(pair, combatants) for pair in observation.pairs),
)
def render(view: ObservationView) -> str:
"""Render a compact view as a deterministic, line-based prompt string."""
lines: list[str] = [
f"schema_version={view.schema_version}",
f"round={view.round}",
f"acting={view.acting_combatant}",
f"map={view.dimensions[0]}x{view.dimensions[1]}",
f"terrain={_fmt_cells(view.terrain)}",
f"elevation={_fmt_cells(view.elevation)}",
f"threatened={_fmt_cells2(view.threatened)}",
f"ranged_threatened={_fmt_cells2(view.ranged_threatened)}",
f"terrain_semantics={_fmt_semantics(view.terrain_semantics)}",
f"initiative={','.join(view.initiative_order)}",
f"attack_fields={','.join(_ATTACK_KEYS)}",
f"spell_fields={','.join(_SPELL_KEYS)}",
"combatants:",
]
for index, combatant in enumerate(view.combatants):
lines.append(f" {index} " + _fmt_combatant(combatant))
lines.append("pairs:")
lines.extend(" " + _fmt_kv(pair) for pair in view.pairs)
return "\n".join(lines) + "\n"
def compact_observation(observation: Observation) -> str:
"""One-shot: build the compact view of an observation and render it."""
return render(build_view(observation))
def _is_default_terrain(
cell_type: str,
semantics_by_type: dict[str, dict[str, object]],
) -> bool:
"""True when a cell carries no tactical information worth listing."""
sem = semantics_by_type.get(cell_type)
if sem is None:
return False
return (
sem.get("move_cost") == 1
and not sem.get("blocks_los")
and not sem.get("cover")
and sem.get("cover_bonus") == 0
)
def _sparse_bool_grid(
grid: tuple[tuple[bool, ...], ...],
) -> tuple[tuple[int, int], ...]:
"""Sparse ``(row, col)`` list of True cells, row-major."""
return tuple(
(r, c)
for r, row in enumerate(grid)
for c, value in enumerate(row)
if value
)
def _compact_combatant(combatant: dict[str, object]) -> dict[str, object]:
"""Select the decision-relevant keys and compact nested expansions."""
compact = {
key: combatant[key] for key in _COMPACT_COMBATANT_KEYS if key in combatant
}
attacks = compact.get("attacks")
if _is_tuple(attacks):
compact["attacks"] = tuple(
_compact_attack(attack) for attack in attacks if _is_str_object_dict(attack)
)
spells = compact.get("spells")
if _is_tuple(spells):
compact["spells"] = tuple(
_compact_spell(spell) for spell in spells if _is_str_object_dict(spell)
)
return compact
def _compact_attack(attack: dict[str, object]) -> dict[str, object]:
"""Compact one attack: damage becomes formula+type strings."""
return {
"id": _as_str(attack["id"]),
"name": _as_str(attack["name"]),
"kind": _as_str(attack["kind"]),
"attack_bonus": _as_int(attack["attack_bonus"]),
"damage": _damage_strings(attack.get("damage")),
"damage_bonus": _as_int(attack["damage_bonus"]),
"crit_range": _as_int(attack["crit_range"]),
"crit_mult": _as_int(attack["crit_mult"]),
"reach_ft": _as_int(attack["reach_ft"]),
"range_increment_ft": _as_optional_int(attack["range_increment_ft"]),
"count": _as_int(attack["count"]),
"firearm": _as_bool(attack["firearm"]),
}
def _compact_spell(spell: dict[str, object]) -> dict[str, object]:
"""Compact one spell: keep only the decision-relevant fields."""
return {
"name": _as_str(spell["name"]),
"level": _as_optional_int(spell["level"]),
"range_ft": _as_int(spell["range_ft"]),
"save": _as_str(spell["save"]),
"dc": _as_optional_int(spell["dc"]),
"cast_time": _as_str(spell["cast_time"]),
"area": spell.get("area"),
}
def _compact_pair(
pair: dict[str, object],
combatants: tuple[dict[str, object], ...],
) -> dict[str, object]:
"""Select the decision-helper keys of one pair, in fixed order."""
compact = {key: pair[key] for key in _PAIR_KEYS if key in pair}
compact["target"] = _combatant_id(combatants, _as_int(pair["target_index"]))
return compact
def _combatant_id(combatants: tuple[dict[str, object], ...], index: int) -> str:
"""Stable id of the combatant at ``index`` (falls back to the index)."""
if 0 <= index < len(combatants):
cid = combatants[index].get("id")
if isinstance(cid, str):
return cid
return str(index)
def _damage_strings(value: object) -> tuple[str, ...]:
"""Render a damage component list as ``"1d6 slashing"`` strings."""
if not _is_tuple(value):
return ()
strings: list[str] = []
for component in value:
if not _is_str_object_dict(component):
continue
formula = component.get("formula")
if not isinstance(formula, str):
continue
types = component.get("types")
if _is_tuple(types):
parts = [formula, *(t for t in types if isinstance(t, str))]
else:
parts = [formula]
strings.append(" ".join(parts))
return tuple(strings)
def _fmt_combatant(combatant: dict[str, object]) -> str:
"""Render one compact combatant, attacks/spells positionally."""
parts: list[str] = []
for key, value in combatant.items():
if key == "attacks":
if _is_tuple(value):
parts.append(
"attacks=("
+ ",".join(
_fmt_attack(attack)
for attack in value
if _is_str_object_dict(attack)
)
+ ")"
)
continue
if key == "spells":
if _is_tuple(value):
parts.append(
"spells=("
+ ",".join(
_fmt_spell(spell)
for spell in value
if _is_str_object_dict(spell)
)
+ ")"
)
continue
parts.append(f"{key}={_fmt_value(value)}")
return " ".join(parts)
def _fmt_attack(attack: dict[str, object]) -> str:
"""Positional attack tuple; ``-`` for a missing range increment."""
damage = attack.get("damage")
damage_str = (
"+".join(d for d in damage if isinstance(d, str)) if _is_tuple(damage) else ""
)
if not damage_str:
damage_str = "-"
range_increment = _as_optional_int(attack["range_increment_ft"])
range_str = "-" if range_increment is None else str(range_increment)
return (
f"({_as_str(attack['id'])},{_as_str(attack['name'])},"
f"{_as_str(attack['kind'])},{_as_int(attack['attack_bonus'])},"
f"{damage_str},{_as_int(attack['damage_bonus'])},"
f"{_as_int(attack['crit_range'])},{_as_int(attack['crit_mult'])},"
f"{_as_int(attack['reach_ft'])},{range_str},"
f"{_as_int(attack['count'])},{1 if _as_bool(attack['firearm']) else 0})"
)
def _fmt_spell(spell: dict[str, object]) -> str:
"""Positional spell tuple; ``-`` for a missing save/dc/area."""
save = _as_str(spell["save"])
save_str = "-" if save == "none" else save
dc = _as_optional_int(spell["dc"])
dc_str = "-" if dc is None else str(dc)
area = spell.get("area")
if area is None:
area_str = "-"
elif _is_str_object_dict(area):
shape = _as_str(area["shape"])
radius = _as_int(area["radius_ft"])
centered = _as_bool(area["centered_on_caster"])
area_str = f"{shape}:{radius}" + (":c" if centered else "")
else:
area_str = "-"
level = _as_optional_int(spell["level"])
level_str = "-" if level is None else str(level)
return (
f"({_as_str(spell['name'])},{level_str},"
f"{_as_int(spell['range_ft'])},{save_str},{dc_str},"
f"{_as_str(spell['cast_time'])},{area_str})"
)
def _fmt_cells(cells: tuple[tuple[int, int, object], ...]) -> str:
return "".join(f"({r},{c},{_fmt_value(v)})" for r, c, v in cells)
def _fmt_cells2(cells: tuple[tuple[int, int], ...]) -> str:
return "".join(f"({r},{c})" for r, c in cells)
def _fmt_semantics(semantics: tuple[dict[str, object], ...]) -> str:
parts: list[str] = []
for sem in semantics:
name = str(sem.get("type", "?"))
move_cost = sem.get("move_cost")
blocks_los = sem.get("blocks_los")
cover = sem.get("cover")
cover_bonus = sem.get("cover_bonus")
parts.append(
f"{name}{{mc={move_cost},los={1 if blocks_los else 0},"
f"cov={1 if cover else 0},cb={cover_bonus}}}"
)
return ",".join(parts)
def _fmt_kv(mapping: dict[str, object]) -> str:
return " ".join(f"{key}={_fmt_value(value)}" for key, value in mapping.items())
def _fmt_value(value: object) -> str:
"""Compact, unambiguous rendering of one snapshot value.
``None`` falls through to ``str(value)`` (already ``"None"``).
"""
if isinstance(value, bool):
return "1" if value else "0"
if isinstance(value, (int, float)):
return str(value)
if isinstance(value, str):
return (
value
if value and all(c.isalnum() or c in "_-" for c in value)
else json.dumps(value)
)
if _is_tuple(value):
return "(" + ",".join(_fmt_value(item) for item in value) + ")"
if _is_str_object_dict(value):
return (
"{"
+ ",".join(f"{key}:{_fmt_value(item)}" for key, item in value.items())
+ "}"
)
return str(value)
def _as_str(value: object) -> str:
if not isinstance(value, str):
msg = f"expected str, got {type(value).__name__}"
raise TypeError(msg)
return value
def _as_int(value: object) -> int:
if not isinstance(value, int):
msg = f"expected int, got {type(value).__name__}"
raise TypeError(msg)
return value
def _as_bool(value: object) -> bool:
if not isinstance(value, bool):
msg = f"expected bool, got {type(value).__name__}"
raise TypeError(msg)
return value
def _as_optional_int(value: object) -> int | None:
if value is None:
return None
return _as_int(value)
def _is_tuple(value: object) -> TypeGuard[tuple[object, ...]]:
"""Narrow ``object`` to a tuple (snapshot collections are tuples)."""
return isinstance(value, tuple)
def _is_str_object_dict(value: object) -> TypeGuard[dict[str, object]]:
"""Narrow ``object`` to the snapshot's ``dict[str, object]`` shape."""
return isinstance(value, dict)