refactor(web): SessionRunner delegates to _session_core (no behavior change)

proteus/runtime/session.py

@@ -1,44 +1,18 @@
 """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.
-
-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.
+One session: build the scenario+game from (scenario, seed, difficulty), replay
+the scripted Cut pre-roll, hand control to the agent for up to ``play_turns``
+turns (observation -> optional probe -> act -> apply -> record), and emit a
+scored SessionTrace. The per-session mechanics are shared with InteractiveSession
+via ``proteus.runtime._session_core`` (single source of truth).
 """
-
 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.rollout import optimal_rollout
+from proteus.runtime import _session_core as core
 from proteus.runtime.trace import SessionTrace, TurnTrace
 
-_ACTIONS = ["up", "down", "left", "right", "stay"]
-# Predator_evade-scoped for the current slice; spec defers scenario generalization.
-# A future multi-scenario runner should source this from the Scenario.
-_PROBE_QUESTION = (
-    "Where is the predator, and which direction should you move to stay safe?"
-)
-
-_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.
@@ -72,173 +46,60 @@ class SessionRunner:
         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
-
+        built = core.build_session(
+            self._scenario_name, self._seed, self._difficulty, self._play_turns,
+        )
+        scenario, game = built.scenario, built.game
+        system_prompt = scenario.rules_text + core._HANDOVER_FRAMING
         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 = probe_reasoning = probe_raw_text = ""
-            probe_input_tokens = probe_output_tokens = probe_thinking_tokens = 0
-            if self._use_probe:
-                probe_q = _PROBE_QUESTION
-                probe = self._agent.probe(observation, probe_q, system_prompt)
-                probe_a = probe.answer
-                probe_reasoning = probe.reasoning
-                probe_raw_text = probe.raw_text
-                probe_input_tokens = probe.input_tokens
-                probe_output_tokens = probe.output_tokens
-                probe_thinking_tokens = probe.thinking_tokens
-
-            # 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._scenario.step_reward(
-                self._game, result.action, blocked,
-                focal_before=focal_pos, predator_before=predator_pos,
+            observation = core.build_observation(
+                scenario, game, built.cut_frames, turn_idx,
             )
 
-            turns.append(
-                TurnTrace(
-                    turn_idx=turn_idx,
-                    observation=observation,
-                    probe_q=probe_q,
-                    probe_a=probe_a,
-                    probe_reasoning=probe_reasoning,
-                    probe_raw_text=probe_raw_text,
-                    probe_input_tokens=probe_input_tokens,
-                    probe_output_tokens=probe_output_tokens,
-                    probe_thinking_tokens=probe_thinking_tokens,
-                    reasoning=result.reasoning,
-                    raw_text=result.raw_text,
-                    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,
-                    input_tokens=result.input_tokens,
-                    output_tokens=result.output_tokens,
-                    thinking_tokens=result.thinking_tokens,
+            probe_fields: dict[str, object] = {}
+            if self._use_probe:
+                probe = self._agent.probe(
+                    observation, core._PROBE_QUESTION, system_prompt,
+                )
+                probe_fields = dict(
+                    probe_q=core._PROBE_QUESTION,
+                    probe_a=probe.answer,
+                    probe_reasoning=probe.reasoning,
+                    probe_raw_text=probe.raw_text,
+                    probe_input_tokens=probe.input_tokens,
+                    probe_output_tokens=probe.output_tokens,
+                    probe_thinking_tokens=probe.thinking_tokens,
                 )
-            )
 
-            if self._game.eliminated or self._game.survived:
+            result = self._agent.act(observation, list(core._ACTIONS), system_prompt)
+            turns.append(core.make_turn_trace(
+                scenario, game,
+                turn_idx=turn_idx,
+                observation=observation,
+                action=result.action,
+                reasoning=result.reasoning,
+                raw_text=result.raw_text,
+                input_tokens=result.input_tokens,
+                output_tokens=result.output_tokens,
+                thinking_tokens=result.thinking_tokens,
+                **probe_fields,
+            ))
+
+            if game.eliminated or game.survived:
                 break
 
-        assert (
-            self._game.eliminated
-            or self._game.survived
-            or len(turns) == self._play_turns
-        ), "play loop exited without a terminal state or exhausting the budget"
-
-        outcome = "eliminated" if self._game.eliminated else "survived"
-        rollout = optimal_rollout(
-            self._scenario_name, self._seed, self._difficulty, len(turns),
-        )
-        realized_final_safety = self._scenario.safety_distance(self._game)
-        metrics = compute_metrics(
-            turns,
-            played_turns=len(turns),
-            play_turns=self._play_turns,
-            outcome=outcome,
-            optimal_focal_positions=rollout.focal_positions,
-            realized_final_safety=realized_final_safety,
-            optimal_final_safety=rollout.final_safety_distance,
-        )
-        return SessionTrace(
-            scenario=self._scenario_name,
-            motive_category=self._motive_category,
-            seed=self._seed,
-            difficulty=self._difficulty.value,
+        return core.finalize(
+            self._scenario_name, scenario, game,
+            seed=self._seed, difficulty=self._difficulty,
+            play_turns=self._play_turns, turns=turns,
+            cut_frames=built.cut_frames, motive_category=self._motive_category,
             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())
-        # The Cut pre-roll must not end the game; if it does, the scenario's
-        # cut_focal_policy is buggy and any resulting trace would be corrupt.
-        if self._game.eliminated or self._game.survived:
-            raise RuntimeError(
-                f"Game terminated during Cut replay of '{self._scenario_name}'. "
-                "cut_focal_policy must not trigger elimination or survival."
-            )
-
-    def _observation(self, turn_idx: int) -> str:
-        # Turn 1 replays the captured cut_frames so the agent sees the scripted
-        # history; turn 2+ render the live grid directly.  This is correct as
-        # long as rendering depends only on sprite state (true for all current
-        # scenarios).
-        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 _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:
         # Relies on the agent exposing its provider as `_provider` (VanillaAgent's
         # convention); unknown agent types fall back to agent.name.