feat(cp6): scenario-owned survival step_reward (away-from-predator) + safety_distance

proteus/grid/scenario.py

@@ -165,6 +165,36 @@ class Scenario(ABC):
         """
         raise NotImplementedError
 
+    @abstractmethod
+    def step_reward(
+        self,
+        game: MotiveGridGame,
+        action: str,
+        blocked: bool,
+        focal_before: tuple[int, int],
+        predator_before: tuple[int, int],
+    ) -> float:
+        """Return the per-turn reward for *action* under this scenario's motive.
+
+        Called AFTER the turn is fully applied (focal moved, threat advanced,
+        terminal flags set), so ``game`` holds the post-move state. The pre-move
+        focal and predator positions are passed in so distance-delta rewards can
+        isolate the agent's own move from the threat's response.
+
+        Args:
+            game: The live game in its POST-move state (focal moved, threat
+                advanced, terminal flags set).
+            action: The action the focal just committed.
+            blocked: Whether that directional move was blocked by a wall/edge
+                (no positional change).
+            focal_before: The focal's ``(x, y)`` BEFORE the move.
+            predator_before: The predator's ``(x, y)`` BEFORE the move.
+
+        Returns:
+            The per-turn reward as a float (sign/scale defined per scenario).
+        """
+        raise NotImplementedError
+
     @abstractmethod
     def legend(self) -> dict[int, str]:
         """Return the palette-index -> symbol/meaning map for the ASCII view.

proteus/grid/scenarios/predator_evade.py

@@ -78,6 +78,11 @@ FOCAL_IDX = 1
 PREDATOR_IDX = 2
 WALL_IDX = 3
 
+# Survival-category reward constants (moved from SessionRunner; see CP6 spec §5).
+_REWARD_CAPTURED = -50.0
+_REWARD_SURVIVED = 50.0
+_REWARD_WALL_HIT = -3.0
+
 # --------------------------------------------------------------------------- #
 # EASY deterministic layout.
 # --------------------------------------------------------------------------- #
@@ -447,6 +452,47 @@ class PredatorEvade(Scenario):
         if action in _DELTAS and action != "stay":
             self._last_focal_move = action
 
+    def step_reward(
+        self,
+        game: MotiveGridGame,
+        action: str,
+        blocked: bool,
+        focal_before: tuple[int, int],
+        predator_before: tuple[int, int],
+    ) -> float:
+        """Survival reward: positive for moving away from the predator.
+
+        Terminal states dominate (capture / survival). Otherwise a blocked move
+        is penalised, and a legal move is scored by the change in BFS distance
+        from the focal to the predator's PRE-move cell (post-move focal cell vs
+        pre-move focal cell), isolating the agent's own contribution from the
+        predator's chase. Positive = moved away, negative = moved toward.
+        """
+        del action  # reward derives from the post-move focal cell, not the action string
+        if game.eliminated:
+            return _REWARD_CAPTURED
+        if game.survived:
+            return _REWARD_SURVIVED
+        if blocked:
+            return _REWARD_WALL_HIT
+        focal = game.focal_sprite
+        if focal is None:
+            return 0.0
+        focal_after = (focal.x, focal.y)
+        d_before = self._bfs_distance(game, focal_before, predator_before)
+        d_after = self._bfs_distance(game, focal_after, predator_before)
+        if d_before is None or d_after is None:
+            return 0.0
+        return float(d_after - d_before)
+
+    def safety_distance(self, game: MotiveGridGame) -> int | None:
+        """Return the BFS distance from the focal to the predator (None if N/A)."""
+        focal = game.focal_sprite
+        predator = game.predator_sprite
+        if focal is None or predator is None:
+            return None
+        return self._bfs_distance(game, (focal.x, focal.y), (predator.x, predator.y))
+
     # ------------------------------------------------------------------ #
     # ASCII legend
     # ------------------------------------------------------------------ #

proteus/runtime/session.py

@@ -8,9 +8,10 @@ One session:
        observation -> (optional probe) -> act -> apply -> record.
     4. Stop on elimination / survival / budget; emit a SessionTrace.
 
-The reward schedule and play budget mirror the legacy MotiveGridModule so the
-ported physics keep their validated meaning, but all forfeit/risk machinery is
-gone.
+Per-turn reward is delegated to ``Scenario.step_reward`` (category-specific —
+e.g. survival rewards moving away from the predator); the play budget mirrors
+the legacy MotiveGridModule so the ported physics keep their validated meaning,
+and all forfeit/risk machinery is gone.
 """
 
 from __future__ import annotations
@@ -32,13 +33,6 @@ _PROBE_QUESTION = (
     "Where is the predator, and which direction should you move to stay safe?"
 )
 
-# Reward schedule (mirrors the legacy MotiveGridModule).
-_REWARD_CAPTURED = -50.0
-_REWARD_SURVIVED = 50.0
-_REWARD_WALL_HIT = -3.0
-_REWARD_OPTIMAL = 5.0
-_REWARD_SUBOPTIMAL = -2.0
-
 _HANDOVER_FRAMING = (
     "\n\nYou have been watching the game so far. From now on YOU control the "
     "focal agent. Continue playing to survive."
@@ -116,7 +110,10 @@ class SessionRunner:
             result = self._agent.act(observation, list(_ACTIONS), system_prompt)
 
             blocked = self._apply(result.action)
-            reward = self._reward(result.action, optimal, blocked)
+            reward = self._scenario.step_reward(
+                self._game, result.action, blocked,
+                focal_before=focal_pos, predator_before=predator_pos,
+            )
 
             turns.append(
                 TurnTrace(
@@ -230,18 +227,6 @@ class SessionRunner:
         post = (moved.x, moved.y) if moved else None
         return action in {"up", "down", "left", "right"} and post == pre
 
-    def _reward(self, action: str, optimal: str, blocked: bool) -> float:
-        assert self._game is not None
-        if self._game.eliminated:
-            return _REWARD_CAPTURED
-        if self._game.survived:
-            return _REWARD_SURVIVED
-        if blocked:
-            return _REWARD_WALL_HIT
-        if action == optimal:
-            return _REWARD_OPTIMAL
-        return _REWARD_SUBOPTIMAL
-
     def _render(self) -> str:
         assert self._scenario is not None and self._game is not None
         return frame_to_ascii(self._game.current_grid(), self._scenario.legend())

tests/grid/test_step_reward.py

@@ -0,0 +1,57 @@
+import random
+
+from proteus.grid.difficulty import Difficulty
+from proteus.grid.game import MotiveGridGame
+from proteus.grid.scenario import get_scenario
+
+
+def _handover():
+    s = get_scenario("predator_evade")()
+    game = MotiveGridGame(s, random.Random(42), Difficulty.EASY, max_steps=20)
+    for _ in range(s.cut_length(Difficulty.EASY)):
+        a = s.cut_focal_policy(game)
+        game.apply_motive_action(a)
+        s.record_focal_move(a)
+    return game, s
+
+
+def test_step_reward_positive_when_moving_away():
+    game, s = _handover()
+    focal_before = (game.focal_sprite.x, game.focal_sprite.y)
+    predator_before = (game.predator_sprite.x, game.predator_sprite.y)
+    # 'up' is the optimal escape at the EASY handover -> moves away.
+    game.apply_motive_action("up")
+    r = s.step_reward(game, "up", blocked=False,
+                      focal_before=focal_before, predator_before=predator_before)
+    assert r > 0
+
+
+def test_step_reward_negative_when_moving_toward():
+    game, s = _handover()
+    focal_before = (game.focal_sprite.x, game.focal_sprite.y)
+    predator_before = (game.predator_sprite.x, game.predator_sprite.y)
+    # 'right' moves toward the predator (east).
+    game.apply_motive_action("right")
+    r = s.step_reward(game, "right", blocked=False,
+                      focal_before=focal_before, predator_before=predator_before)
+    assert r < 0
+
+
+def test_step_reward_negative_on_wall_hit():
+    game, s = _handover()
+    focal_before = (game.focal_sprite.x, game.focal_sprite.y)
+    predator_before = (game.predator_sprite.x, game.predator_sprite.y)
+    # 'left' is blocked by the dead-end wall at the EASY handover.
+    game.apply_motive_action("left")
+    # The dead-end wall makes 'left' a no-op; confirm the engine blocked it
+    # so blocked=True faithfully reflects what _apply would have computed.
+    assert (game.focal_sprite.x, game.focal_sprite.y) == focal_before
+    r = s.step_reward(game, "left", blocked=True,
+                      focal_before=focal_before, predator_before=predator_before)
+    assert r < 0
+
+
+def test_safety_distance_is_bfs_focal_to_predator():
+    game, s = _handover()
+    d = s.safety_distance(game)
+    assert isinstance(d, int) and d >= 0