Spaces:
Runtime error
Runtime error
| """SpectateSession — a threadless, agent-driven driver for watching an LLM play. | |
| Sibling of InteractiveSession, but the server (not a human) supplies each action | |
| by calling the agent once per HTTP request. Built on the same ``_session_core`` | |
| helpers, so its trace matches SessionRunner for the same agent (pinned by | |
| tests/runtime/test_spectate_equivalence.py). | |
| Disclosure: a spectator is NOT the scored subject, so ``state()`` exposes the per | |
| turn answer keys (optimal/habit), reward, and the model's reasoning LIVE. (The | |
| model itself still only ever receives the observation passed to ``agent.act``.) | |
| """ | |
| from __future__ import annotations | |
| from proteus.game.agents.base import Agent | |
| from proteus.game.engine.difficulty import Difficulty | |
| from proteus.game.runtime import _session_core as core | |
| from proteus.game.runtime.trace import SessionTrace, TurnTrace | |
| class SpectateSession: | |
| def __init__( | |
| self, | |
| scenario_name: str, | |
| *, | |
| agent: Agent, | |
| model_name: str, | |
| difficulty: Difficulty = Difficulty.EASY, | |
| seed: int | None = None, | |
| play_turns: int = 15, | |
| use_probe: bool = False, | |
| motive_category: str = "survival", | |
| memory: "MemoryCheckpoint | None" = None, | |
| use_default_memory: bool = True, | |
| ) -> None: | |
| self._scenario_name = scenario_name | |
| self._agent = agent | |
| self._model_name = model_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 | |
| # Explicit memory wins; else the scenario default (template persona), | |
| # unless the caller forced no memory via use_default_memory=False. | |
| self._memory = ( | |
| memory if memory is not None | |
| else (getattr(built, "default_memory", None) if use_default_memory else None) | |
| ) | |
| self._system_prompt = self._scenario.rules_text + core._HANDOVER_FRAMING | |
| 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: | |
| 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, | |
| "model": self._model_name, | |
| "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, | |
| # RICH disclosure (spectator): per-turn answer keys + reward + reasoning. | |
| "turns_so_far": [ | |
| { | |
| "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, | |
| "reasoning": t.reasoning, | |
| } | |
| for t in self._turns | |
| ], | |
| "review": None, | |
| } | |
| if done: | |
| trace = self.finish() | |
| st["outcome"] = trace.outcome | |
| st["review"] = {"outcome": trace.outcome, "metrics": trace.metrics} | |
| return st | |
| def advance(self) -> dict: | |
| if self._is_done(): | |
| raise core.SessionFinishedError("session already finished") | |
| turn_idx = len(self._turns) + 1 | |
| observation = core.build_observation( | |
| self._scenario, self._game, self._cut_frames, turn_idx, | |
| memory=self._memory, | |
| prior_actions=[t.action for t in self._turns], | |
| ) | |
| result = self._agent.act(observation, list(core._ACTIONS), self._system_prompt) | |
| self._turns.append(core.make_turn_trace( | |
| self._scenario, self._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, | |
| )) | |
| 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=self._model_name, | |
| ) | |
| return self._trace | |