feat(web): InteractiveSession — threadless stepwise play driver

proteus/runtime/interactive.py

@@ -0,0 +1,131 @@
+"""InteractiveSession — a threadless, stepwise driver for human web play.
+
+Holds the live scenario+game and advances exactly one turn per HTTP request.
+Built on the same ``_session_core`` helpers as SessionRunner, so the trace it
+emits at the end is identical to a SessionRunner(HumanAgent) trace for the same
+actions (pinned by tests/runtime/test_interactive_equivalence.py).
+
+Fairness: ``state()`` exposes only the grid + available actions while playing;
+the per-turn answer keys (optimal/habit) and rewards are disclosed in ``review``
+only once the game is over.
+"""
+from __future__ import annotations
+
+from proteus.grid.difficulty import Difficulty
+from proteus.runtime import _session_core as core
+from proteus.runtime.trace import SessionTrace, TurnTrace
+
+
+class InteractiveSession:
+    def __init__(
+        self,
+        scenario_name: str,
+        *,
+        difficulty: Difficulty = Difficulty.EASY,
+        seed: int | None = None,
+        play_turns: int = 15,
+        use_probe: bool = False,
+        motive_category: str = "survival",
+    ) -> None:
+        self._scenario_name = scenario_name
+        self._difficulty = difficulty
+        self._seed = seed
+        self._play_turns = play_turns
+        self._use_probe = use_probe
+        self._motive_category = motive_category
+
+        built = core.build_session(scenario_name, seed, difficulty, play_turns)
+        self._scenario = built.scenario
+        self._game = built.game
+        self._cut_frames = built.cut_frames
+        self._cut_grids = built.cut_grids
+        self._turns: list[TurnTrace] = []
+        self._trace: SessionTrace | None = None
+
+    # ------------------------------------------------------------------ #
+    def _is_done(self) -> bool:
+        return (
+            self._game.eliminated
+            or self._game.survived
+            or len(self._turns) >= self._play_turns
+        )
+
+    def state(self) -> dict:
+        """The JSON-ready view (live = fair, no answer keys; review only when done).
+
+        When the game is over this finalizes and memoizes the trace (idempotent)
+        so it can populate ``review``; while playing it exposes only grid +
+        actions, never reward/optimal/habit.
+        """
+        done = self._is_done()
+        played = len(self._turns)
+        phase = "done" if done else ("cut_intro" if played == 0 else "play")
+        st: dict = {
+            "phase": phase,
+            "turn_idx": played,
+            "play_turns": self._play_turns,
+            "grid": core.grid_to_list(self._game.current_grid()),
+            "legend": {str(k): v for k, v in self._scenario.legend().items()},
+            "actions": list(core._ACTIONS),
+            "outcome": None,
+            "cut_frames": self._cut_grids if played == 0 else None,
+            "review": None,
+        }
+        if done:
+            trace = self.finish()
+            st["outcome"] = trace.outcome
+            st["review"] = {
+                "outcome": trace.outcome,
+                "metrics": trace.metrics,
+                "turns": [
+                    {
+                        "turn_idx": t.turn_idx,
+                        "action": t.action,
+                        "motive_action": t.motive_action,
+                        "habit_action": t.habit_action,
+                        "reward": t.reward,
+                        "is_diagnostic": t.is_diagnostic,
+                        "was_congruent": t.was_congruent,
+                    }
+                    for t in trace.turns
+                ],
+            }
+        return st
+
+    def step(self, action: str, probe_answer: str = "") -> dict:
+        if self._is_done():
+            raise core.SessionFinishedError("session already finished")
+        if action not in core._ACTIONS:
+            raise ValueError(
+                f"invalid action {action!r}; choose one of {core._ACTIONS}"
+            )
+        turn_idx = len(self._turns) + 1
+        observation = core.build_observation(
+            self._scenario, self._game, self._cut_frames, turn_idx,
+        )
+        probe_fields: dict[str, object] = {}
+        if self._use_probe:
+            probe_fields = dict(
+                probe_q=core._PROBE_QUESTION,
+                probe_a=probe_answer,
+                probe_reasoning="",
+                probe_raw_text=probe_answer,
+            )
+        self._turns.append(core.make_turn_trace(
+            self._scenario, self._game,
+            turn_idx=turn_idx, observation=observation,
+            action=action, raw_text=action, **probe_fields,
+        ))
+        return self.state()
+
+    def finish(self) -> SessionTrace:
+        if self._trace is not None:
+            return self._trace
+        self._trace = core.finalize(
+            self._scenario_name, self._scenario, self._game,
+            seed=self._seed, difficulty=self._difficulty,
+            play_turns=self._play_turns, turns=self._turns,
+            cut_frames=self._cut_frames, motive_category=self._motive_category,
+            model="human",
+        )
+        return self._trace

tests/runtime/test_interactive_session.py

@@ -0,0 +1,73 @@
+"""Unit tests for InteractiveSession (HTTP-driven, stepwise play)."""
+from __future__ import annotations
+
+import pytest
+
+import proteus.grid  # noqa: F401
+from proteus.grid.difficulty import Difficulty
+from proteus.runtime._session_core import SessionFinishedError
+from proteus.runtime.interactive import InteractiveSession
+
+
+def _new(play_turns=10):
+    return InteractiveSession(
+        "predator_evade", difficulty=Difficulty.EASY, seed=42,
+        play_turns=play_turns, use_probe=False,
+    )
+
+
+def test_initial_state_is_cut_intro_with_int_grid_and_no_answer_keys():
+    s = _new()
+    st = s.state()
+    assert st["phase"] == "cut_intro"
+    assert st["turn_idx"] == 0
+    assert st["outcome"] is None
+    assert st["review"] is None
+    # grid is a JSON-ready int matrix.
+    assert isinstance(st["grid"], list) and isinstance(st["grid"][0][0], int)
+    # cut animation frames are present on the first state only.
+    assert st["cut_frames"] is not None and len(st["cut_frames"]) >= 1
+    # fairness: live state leaks no reward / optimal / habit.
+    flat = str(st)
+    assert "reward" not in st and "motive_action" not in st and "habit" not in flat
+
+
+def test_step_advances_turn_and_drops_cut_frames():
+    s = _new()
+    st = s.step("up")
+    assert st["phase"] == "play"
+    assert st["turn_idx"] == 1
+    assert st["cut_frames"] is None
+
+
+def test_invalid_action_rejected():
+    s = _new()
+    with pytest.raises(ValueError):
+        s.step("northwest")
+
+
+def test_play_to_budget_then_review_and_finish():
+    s = _new(play_turns=3)
+    for _ in range(3):
+        if s.state()["outcome"] is not None:
+            break
+        s.step("up")
+    st = s.state()
+    assert st["phase"] == "done"
+    assert st["outcome"] in ("survived", "eliminated")
+    # review is disclosed only when done.
+    assert st["review"] is not None
+    assert "metrics" in st["review"] and "turns" in st["review"]
+    trace = s.finish()
+    assert trace.model == "human"
+    assert trace.scenario == "predator_evade"
+    # finish() is memoized: repeated calls return the same trace object.
+    assert s.finish() is trace
+
+
+def test_step_after_done_raises():
+    s = _new(play_turns=1)
+    s.step("up")
+    # play_turns=1 exhausts the budget -> done.
+    with pytest.raises(SessionFinishedError):
+        s.step("up")