Spaces:
Runtime error
Runtime error
| """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 | |