87 lines
4 KiB
Python
87 lines
4 KiB
Python
"""Run-scoped policy memory and explicitly limited card evidence. No file or network I/O."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
import facts as F
|
|
from policy.context import Decision, PolicyContext
|
|
|
|
|
|
@dataclass
|
|
class RunContext:
|
|
policy: PolicyContext = field(default_factory=PolicyContext)
|
|
run_id: str | None = None
|
|
seed: str | None = None
|
|
last_known_id: str | None = None
|
|
active: bool = False
|
|
deck: dict | None = None
|
|
|
|
def observe(self, obs: dict, current_run: dict | None) -> None:
|
|
"""Bind evidence to reported identity, never to a seed or a local save path.
|
|
|
|
Missing metadata discards card evidence, but retains action guards.
|
|
Clearing accepted toggles on a transient metadata failure is unsafe.
|
|
A different known run ID or an observed run exit resets those guards.
|
|
"""
|
|
st = obs.get("state_type")
|
|
if st in ("unknown", "overlay"):
|
|
return
|
|
if st in ("menu", "game_over"):
|
|
if self.active:
|
|
self.policy = PolicyContext()
|
|
self.run_id = self.seed = self.last_known_id = None
|
|
self.deck = None
|
|
self.active = False
|
|
return
|
|
self.active = True
|
|
identity = current_run if isinstance(current_run, dict) else {}
|
|
run_id = identity.get("run_id")
|
|
if identity.get("is_in_progress") is not True or not isinstance(run_id, str) or not run_id.strip():
|
|
self.run_id = self.seed = None
|
|
self.deck = None
|
|
return
|
|
if self.last_known_id is not None and run_id != self.last_known_id:
|
|
self.policy = PolicyContext()
|
|
self.deck = None
|
|
self.run_id = self.last_known_id = run_id
|
|
self.seed = identity.get("seed") if isinstance(identity.get("seed"), str) else None
|
|
if self.deck is not None:
|
|
self.deck["provenance"]["freshness"] = "historical_combat_piles"
|
|
# Room changes make this weak evidence even less useful. Do not
|
|
# carry it into a later room as a substitute for the run deck.
|
|
location = {k: (obs.get("run") or {}).get(k) for k in ("act", "floor")}
|
|
if any(value is None for value in location.values()) or location != self.deck["provenance"]["location"]:
|
|
self.deck = None
|
|
|
|
def capture_combat(self, obs: dict, facts: F.CombatFacts, step: int) -> None:
|
|
"""Visible piles include temporary cards and can omit unseen cards."""
|
|
if self.run_id is None or not facts.deck_counts:
|
|
self.deck = None
|
|
return
|
|
self.deck = {
|
|
"counts": facts.deck_counts,
|
|
"summary": facts.deck_summary,
|
|
"provenance": {
|
|
"source": "combat_piles",
|
|
"run_id": self.run_id,
|
|
"observed_step": step,
|
|
"location": {k: (obs.get("run") or {}).get(k) for k in ("act", "floor")},
|
|
"persistent_deck": False,
|
|
"pile_lists_present": [name for name in ("hand", "draw_pile", "discard_pile", "exhaust_pile")
|
|
if isinstance((obs.get("player") or {}).get(name), list)],
|
|
"freshness": "observed_combat_piles",
|
|
},
|
|
}
|
|
|
|
def record_result(self, obs: dict, decision: Decision, *, accepted: bool) -> None:
|
|
self.policy.record_result(obs, decision, accepted=accepted)
|
|
if not accepted or obs.get("state_type") in ("monster", "elite", "boss"):
|
|
return
|
|
# Keep limited evidence while opening/skipping a card reward. Other
|
|
# accepted non-combat actions can change the deck or leave this room.
|
|
card_reward = decision.action == "claim_reward" and any(
|
|
item.get("index") == decision.params.get("index") and item.get("type") == "card"
|
|
for item in (obs.get("rewards") or {}).get("items", [])
|
|
)
|
|
if not card_reward and decision.action != "skip_card_reward":
|
|
self.deck = None
|