irregular6612 commited on
Commit
81b3758
·
1 Parent(s): 8cccda6

feat(cp3): guard cut-replay termination + cover survived/eliminated/cut-frame contracts

Browse files
proteus/runtime/session.py CHANGED
@@ -26,6 +26,8 @@ from proteus.runtime.metrics import compute_metrics
26
  from proteus.runtime.trace import SessionTrace, TurnTrace
27
 
28
  _ACTIONS = ["up", "down", "left", "right", "stay"]
 
 
29
  _PROBE_QUESTION = (
30
  "Where is the predator, and which direction should you move to stay safe?"
31
  )
@@ -130,6 +132,12 @@ class SessionRunner:
130
  if self._game.eliminated or self._game.survived:
131
  break
132
 
 
 
 
 
 
 
133
  outcome = "eliminated" if self._game.eliminated else "survived"
134
  metrics = compute_metrics(
135
  turns,
@@ -166,8 +174,19 @@ class SessionRunner:
166
  self._game.apply_motive_action(action)
167
  self._scenario.record_focal_move(action)
168
  self._cut_frames.append(self._render())
 
 
 
 
 
 
 
169
 
170
  def _observation(self, turn_idx: int) -> str:
 
 
 
 
171
  assert self._scenario is not None
172
  legend = self._scenario.legend()
173
  parts: list[str] = []
@@ -212,5 +231,7 @@ class SessionRunner:
212
  return frame_to_ascii(self._game.current_grid(), self._scenario.legend())
213
 
214
  def _provider_model_name(self) -> str:
 
 
215
  provider = getattr(self._agent, "_provider", None)
216
  return getattr(provider, "model_name", self._agent.name)
 
26
  from proteus.runtime.trace import SessionTrace, TurnTrace
27
 
28
  _ACTIONS = ["up", "down", "left", "right", "stay"]
29
+ # Predator_evade-scoped for the current slice; spec defers scenario generalization.
30
+ # A future multi-scenario runner should source this from the Scenario.
31
  _PROBE_QUESTION = (
32
  "Where is the predator, and which direction should you move to stay safe?"
33
  )
 
132
  if self._game.eliminated or self._game.survived:
133
  break
134
 
135
+ assert (
136
+ self._game.eliminated
137
+ or self._game.survived
138
+ or len(turns) == self._play_turns
139
+ ), "play loop exited without a terminal state or exhausting the budget"
140
+
141
  outcome = "eliminated" if self._game.eliminated else "survived"
142
  metrics = compute_metrics(
143
  turns,
 
174
  self._game.apply_motive_action(action)
175
  self._scenario.record_focal_move(action)
176
  self._cut_frames.append(self._render())
177
+ # The Cut pre-roll must not end the game; if it does, the scenario's
178
+ # cut_focal_policy is buggy and any resulting trace would be corrupt.
179
+ if self._game.eliminated or self._game.survived:
180
+ raise RuntimeError(
181
+ f"Game terminated during Cut replay of '{self._scenario_name}'. "
182
+ "cut_focal_policy must not trigger elimination or survival."
183
+ )
184
 
185
  def _observation(self, turn_idx: int) -> str:
186
+ # Turn 1 replays the captured cut_frames so the agent sees the scripted
187
+ # history; turn 2+ render the live grid directly. This is correct as
188
+ # long as rendering depends only on sprite state (true for all current
189
+ # scenarios).
190
  assert self._scenario is not None
191
  legend = self._scenario.legend()
192
  parts: list[str] = []
 
231
  return frame_to_ascii(self._game.current_grid(), self._scenario.legend())
232
 
233
  def _provider_model_name(self) -> str:
234
+ # Relies on the agent exposing its provider as `_provider` (VanillaAgent's
235
+ # convention); unknown agent types fall back to agent.name.
236
  provider = getattr(self._agent, "_provider", None)
237
  return getattr(provider, "model_name", self._agent.name)
tests/runtime/test_session.py CHANGED
@@ -63,3 +63,34 @@ def test_session_is_deterministic_for_same_inputs():
63
  # Same scripted agent + same seed -> identical realized trajectory.
64
  assert [t.focal_pos for t in t1.turns] == [t.focal_pos for t in t2.turns]
65
  assert t1.metrics == t2.metrics
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
63
  # Same scripted agent + same seed -> identical realized trajectory.
64
  assert [t.focal_pos for t in t1.turns] == [t.focal_pos for t in t2.turns]
65
  assert t1.metrics == t2.metrics
66
+
67
+
68
+ def test_short_budget_yields_survived_outcome():
69
+ # With a tiny budget the step count is exhausted (without capture) right
70
+ # after the played turns, so the engine fires `survived`.
71
+ agent = _agent(["ACTION: up"])
72
+ trace = SessionRunner(
73
+ "predator_evade", agent, seed=42, play_turns=1, use_probe=False,
74
+ ).run()
75
+ assert trace.outcome == "survived"
76
+ assert trace.turns[-1].reward == 50.0 # _REWARD_SURVIVED
77
+
78
+
79
+ def test_eliminated_outcome_is_explicit_and_terminal():
80
+ # The habit player ("left") walks into the dead-end and is caught.
81
+ agent = _agent(["ACTION: left"])
82
+ trace = SessionRunner(
83
+ "predator_evade", agent, seed=42, play_turns=15, use_probe=False,
84
+ ).run()
85
+ assert trace.outcome == "eliminated"
86
+ assert len(trace.turns) <= 15 # stopped at/under budget
87
+ assert trace.turns[-1].reward == -50.0 # _REWARD_CAPTURED
88
+
89
+
90
+ def test_cut_frames_count_matches_cut_length_plus_one():
91
+ agent = _agent(["ACTION: up"])
92
+ trace = SessionRunner(
93
+ "predator_evade", agent, seed=42, play_turns=5, use_probe=False,
94
+ ).run()
95
+ # EASY cut_length is 2 -> initial frame + 2 step frames = 3.
96
+ assert len(trace.cut_frames) == 3