feat(cp3): SessionRunner orchestrator (cut replay -> play -> trace)

proteus/runtime/session.py

@@ -0,0 +1,216 @@
+"""SessionRunner — the PROTEUS arena orchestrator.
+
+One session:
+    1. Build the scenario + game from (scenario, seed, difficulty).
+    2. Replay the scripted Cut pre-roll, capturing ASCII frames (the history
+       shown to the agent at handover).
+    3. Hand control to the agent and play up to ``play_turns`` turns:
+       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.
+"""
+
+from __future__ import annotations
+
+import random
+
+from proteus.agents.base import Agent
+from proteus.grid.ascii_view import frame_to_ascii, legend_text
+from proteus.grid.difficulty import Difficulty
+from proteus.grid.game import MotiveGridGame
+from proteus.grid.scenario import Scenario, get_scenario
+from proteus.runtime.metrics import compute_metrics
+from proteus.runtime.trace import SessionTrace, TurnTrace
+
+_ACTIONS = ["up", "down", "left", "right", "stay"]
+_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."
+)
+
+
+class SessionRunner:
+    """Run one motive_grid session end-to-end and return a SessionTrace.
+
+    Args:
+        scenario_name: Registered scenario (e.g. ``"predator_evade"``).
+        agent: The :class:`Agent` that plays after the handover.
+        difficulty: Difficulty band (controls Cut length).
+        seed: Seed for the deterministic world/Cut.
+        play_turns: Survival budget in played (post-Cut) turns.
+        use_probe: Whether to ask the side-channel probe each turn.
+        motive_category: Category label stored on the trace.
+    """
+
+    def __init__(
+        self,
+        scenario_name: str,
+        agent: Agent,
+        *,
+        difficulty: Difficulty = Difficulty.EASY,
+        seed: int | None = None,
+        play_turns: int = 15,
+        use_probe: bool = True,
+        motive_category: str = "survival",
+    ) -> None:
+        self._scenario_name = scenario_name
+        self._agent = agent
+        self._difficulty = difficulty
+        self._seed = seed
+        self._play_turns = play_turns
+        self._use_probe = use_probe
+        self._motive_category = motive_category
+
+        self._scenario: Scenario | None = None
+        self._game: MotiveGridGame | None = None
+        self._cut_frames: list[str] = []
+        self._cut_length = 0
+
+    def run(self) -> SessionTrace:
+        """Build, replay the Cut, play N turns, and return the trace."""
+        self._build_and_replay_cut()
+        assert self._scenario is not None and self._game is not None
+
+        turns: list[TurnTrace] = []
+        system_prompt = self._scenario.rules_text + _HANDOVER_FRAMING
+
+        for turn_idx in range(1, self._play_turns + 1):
+            observation = self._observation(turn_idx)
+
+            probe_q = probe_a = ""
+            if self._use_probe:
+                probe_q = _PROBE_QUESTION
+                probe_a = self._agent.probe(observation, probe_q, system_prompt)
+
+            # Pre-move answer keys + positions.
+            optimal = self._scenario.optimal_action(self._game)
+            habit = self._scenario.habit_action(self._game)
+            focal = self._game.focal_sprite
+            predator = self._game.predator_sprite
+            focal_pos = (focal.x, focal.y) if focal else (-1, -1)
+            predator_pos = (predator.x, predator.y) if predator else (-1, -1)
+
+            result = self._agent.act(observation, list(_ACTIONS), system_prompt)
+
+            blocked = self._apply(result.action)
+            reward = self._reward(result.action, optimal, blocked)
+
+            turns.append(
+                TurnTrace(
+                    turn_idx=turn_idx,
+                    observation=observation,
+                    probe_q=probe_q,
+                    probe_a=probe_a,
+                    reasoning=result.reasoning,
+                    action=result.action,
+                    motive_action=optimal,
+                    habit_action=habit,
+                    is_diagnostic=(optimal != habit),
+                    was_congruent=(result.action == optimal),
+                    reward=reward,
+                    focal_pos=focal_pos,
+                    predator_pos=predator_pos,
+                )
+            )
+
+            if self._game.eliminated or self._game.survived:
+                break
+
+        outcome = "eliminated" if self._game.eliminated else "survived"
+        metrics = compute_metrics(
+            turns,
+            played_turns=len(turns),
+            play_turns=self._play_turns,
+            outcome=outcome,
+        )
+        return SessionTrace(
+            scenario=self._scenario_name,
+            motive_category=self._motive_category,
+            seed=self._seed,
+            difficulty=self._difficulty.value,
+            model=self._provider_model_name(),
+            cut_frames=list(self._cut_frames),
+            turns=turns,
+            outcome=outcome,
+            metrics=metrics,
+        )
+
+    # ------------------------------------------------------------------ #
+    # Internals
+    # ------------------------------------------------------------------ #
+    def _build_and_replay_cut(self) -> None:
+        self._scenario = get_scenario(self._scenario_name)()
+        rng = random.Random(self._seed)
+        self._cut_length = self._scenario.cut_length(self._difficulty)
+        self._game = MotiveGridGame(
+            self._scenario, rng, self._difficulty,
+            max_steps=self._cut_length + self._play_turns,
+        )
+        self._cut_frames = [self._render()]
+        for _ in range(self._cut_length):
+            action = self._scenario.cut_focal_policy(self._game)
+            self._game.apply_motive_action(action)
+            self._scenario.record_focal_move(action)
+            self._cut_frames.append(self._render())
+
+    def _observation(self, turn_idx: int) -> str:
+        assert self._scenario is not None
+        legend = self._scenario.legend()
+        parts: list[str] = []
+        if turn_idx == 1 and self._cut_frames:
+            for i, frame in enumerate(self._cut_frames[:-1], start=1):
+                parts.append(f"Cut {i}:")
+                parts.append(frame)
+            parts.append("Now:")
+            parts.append(self._cut_frames[-1])
+        else:
+            parts.append("Now:")
+            parts.append(self._render())
+        parts.append(legend_text(legend))
+        parts.append(f"Available actions: [{', '.join(_ACTIONS)}]")
+        return "\n".join(parts)
+
+    def _apply(self, action: str) -> bool:
+        """Apply the action; return True if a directional move was blocked."""
+        assert self._scenario is not None and self._game is not None
+        focal = self._game.focal_sprite
+        pre = (focal.x, focal.y) if focal else None
+        self._game.apply_motive_action(action)
+        self._scenario.record_focal_move(action)
+        moved = self._game.focal_sprite
+        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())
+
+    def _provider_model_name(self) -> str:
+        provider = getattr(self._agent, "_provider", None)
+        return getattr(provider, "model_name", self._agent.name)

tests/runtime/test_session.py

@@ -0,0 +1,65 @@
+from proteus.providers import FakeProvider
+from proteus.agents import VanillaAgent
+from proteus.runtime.session import SessionRunner
+from proteus.runtime.trace import SessionTrace
+
+
+def _agent(responses):
+    return VanillaAgent(FakeProvider(responses=responses))
+
+
+def test_optimal_player_survives_and_scores_full_motive_reading():
+    # At the EASY handover the motive-congruent action is "up". An agent that
+    # always plays "up"... will move up the open column away from the predator.
+    # Whatever the realized states, the runner scores each turn against the
+    # live optimal answer key. Here we script an agent that always says "up".
+    agent = _agent(["ACTION: up"])  # FakeProvider repeats the last response
+    runner = SessionRunner(
+        "predator_evade", agent, seed=42, play_turns=10, use_probe=False,
+    )
+    trace = runner.run()
+    assert isinstance(trace, SessionTrace)
+    assert trace.scenario == "predator_evade"
+    assert trace.cut_frames  # Cut history captured
+    assert len(trace.turns) >= 1
+    # First played turn: the handover, where motive=up, habit=left (diagnostic).
+    first = trace.turns[0]
+    assert first.is_diagnostic is True
+    assert first.motive_action == "up"
+    assert first.habit_action == "left"
+    assert first.action == "up"
+    assert first.was_congruent is True
+    assert "motive_reading_accuracy" in trace.metrics
+
+
+def test_habit_player_diverges_on_first_diagnostic_turn():
+    # An agent that always plays "left" follows inertia into the dead-end.
+    agent = _agent(["ACTION: left"])
+    runner = SessionRunner(
+        "predator_evade", agent, seed=42, play_turns=10, use_probe=False,
+    )
+    trace = runner.run()
+    first = trace.turns[0]
+    assert first.action == "left"
+    assert first.was_congruent is False
+    assert trace.metrics["first_divergence_turn"] == 1.0
+
+
+def test_probe_recorded_when_enabled():
+    agent = _agent(["the predator is to my east; I should go up\nACTION: up"])
+    runner = SessionRunner(
+        "predator_evade", agent, seed=42, play_turns=3, use_probe=True,
+    )
+    trace = runner.run()
+    assert trace.turns[0].probe_q  # a question was asked
+    assert trace.turns[0].probe_a  # an answer was recorded
+
+
+def test_session_is_deterministic_for_same_inputs():
+    t1 = SessionRunner("predator_evade", _agent(["ACTION: up"]), seed=42,
+                       play_turns=5, use_probe=False).run()
+    t2 = SessionRunner("predator_evade", _agent(["ACTION: up"]), seed=42,
+                       play_turns=5, use_probe=False).run()
+    # Same scripted agent + same seed -> identical realized trajectory.
+    assert [t.focal_pos for t in t1.turns] == [t.focal_pos for t in t2.turns]
+    assert t1.metrics == t2.metrics