"""The taskset and the env: the loop, the seat, and what reaches the trace. Everything else in this directory runs with nothing installed. This file needs `verifiers`, so it skips when it is absent — `uv sync --project environments/grand_exchange_live` first, exactly as the README's release gate does. The env is driven against a stub agent rather than a model. What is being tested is the control flow that only exists here: a synchronous engine walked through an asynchronous interaction, a prompted task whose first turn must be BARE, a `terminated` segment arriving mid-window, and the rule that `TurnView.market` never reaches the trace. """ from __future__ import annotations import asyncio import sys import types import unittest from dataclasses import dataclass, field from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import context # noqa: E402 - resolves the package before anything imports it try: import verifiers.v1 as vf # noqa: F401 except ImportError: # pragma: no cover - the no-dependency path vf = None from grand_exchange_live.live import HORIZON, TURNS # noqa: E402 from grand_exchange_live.market import build_market # noqa: E402 from grand_exchange_live.protocol import opening # noqa: E402 PRICE_SENTINEL = 9_000_000 @dataclass class StubTrace: rewards: dict = field(default_factory=dict) metrics: dict = field(default_factory=dict) info: dict = field(default_factory=dict) def record_reward(self, name, value, weight=1.0): self.rewards[name] = (float(value), float(weight)) def record_metrics(self, values): self.metrics.update({k: float(v) for k, v in values.items()}) @dataclass class StubSegment: last_reply: str = "" terminated: bool = False class StubInteraction: """One rollout, scripted. Records what each turn was asked with.""" def __init__(self, replies, terminate_at=None): self.replies = replies self.terminate_at = terminate_at self.asked: list = [] self.trace = StubTrace() async def turn(self, message=None): self.asked.append(message) n = len(self.asked) if self.terminate_at is not None and n >= self.terminate_at: return StubSegment(terminated=True) return StubSegment(last_reply=self.replies[(n - 1) % len(self.replies)]) async def __aenter__(self): return self async def __aexit__(self, *exc): return False class StubAgent: def __init__(self, interaction): self._interaction = interaction def interaction(self, task): return self._interaction class StubAgents: def __init__(self, interaction): self.agent = StubAgent(interaction) HOLD = '```json\n{"orders": []}\n```' @unittest.skipIf(vf is None, "verifiers is not installed") class TasksetTests(unittest.TestCase): def taskset(self, num_tasks=4): from grand_exchange_live.taskset import LiveExchangeConfig, LiveExchangeTaskset return LiveExchangeTaskset( LiveExchangeConfig(id="grand-exchange-live", num_tasks=num_tasks) ) def test_the_seat_is_named_agent(self) -> None: """A wrong name parses fine and silently leaves the seat empty, so it is asserted rather than read off the file.""" from grand_exchange_live.taskset import LiveExchangeEnvConfig self.assertIn("agent", LiveExchangeEnvConfig.model_fields) def test_tasks_carry_the_seed_and_the_opening_board(self) -> None: tasks = self.taskset().load() self.assertEqual(len(tasks), 4) for task in tasks: market = build_market(task.data.seed, task.data.num_items, task.data.visible, HORIZON) self.assertEqual(task.data.prompt, opening(market, TURNS)) self.assertEqual(task.data.turns, TURNS) self.assertIn("Grand Exchange", task.data.system_prompt) def test_two_tasks_never_share_a_basket(self) -> None: """`viable_live_market` skips baskets the reference loses money in; a load that did not advance past the seed it landed on would serve the same basket twice.""" seeds = [t.data.seed for t in self.taskset(24).load()] self.assertEqual(len(set(seeds)), len(seeds)) def test_the_task_stops_at_the_turn_budget(self) -> None: task = self.taskset(1).load()[0] trace = types.SimpleNamespace(num_turns=TURNS - 1) self.assertFalse(asyncio.run(task.budget(trace))) trace.num_turns = TURNS self.assertTrue(asyncio.run(task.budget(trace))) @unittest.skipIf(vf is None, "verifiers is not installed") class EnvRunTests(unittest.TestCase): def env_and_task(self): from grand_exchange_live.taskset import ( LiveExchangeConfig, LiveExchangeEnv, LiveExchangeEnvConfig, LiveExchangeTaskset, ) config = LiveExchangeEnvConfig( taskset=LiveExchangeConfig(id="grand-exchange-live", num_tasks=1) ) # No harness, no model, no runtime: `run()` is the whole subject here, and building # the env through its constructor would stand up an agent to reach it. env = LiveExchangeEnv.__new__(LiveExchangeEnv) env.config = config return env, LiveExchangeTaskset(config.taskset).load()[0] def run_env(self, replies, terminate_at=None, task=None, env=None): from grand_exchange_live.taskset import LiveExchangeEnv if env is None: env, task = self.env_and_task() interaction = StubInteraction(replies, terminate_at) asyncio.run(LiveExchangeEnv.run(env, task, StubAgents(interaction))) return interaction def test_the_first_turn_is_bare_and_every_later_turn_carries_the_view(self) -> None: """The task is prompted, so the model speaks first. Passing the opening text on turn one raises ValueError inside `verifiers` — the prompt already opened the exchange.""" interaction = self.run_env([HOLD]) self.assertEqual(len(interaction.asked), TURNS) self.assertIsNone(interaction.asked[0]) for message in interaction.asked[1:]: self.assertIsInstance(message, str) self.assertIn("Look ", message) def test_a_terminated_segment_mid_window_is_not_an_error(self) -> None: """The naive loop calls turn() again and gets RuntimeError('the exchange is over'), which fails the episode over a model that stopped talking. The window runs on.""" interaction = self.run_env([HOLD], terminate_at=4) self.assertEqual(interaction.trace.metrics["ended_early"], 1.0) self.assertEqual(interaction.trace.metrics["answered_turns"], 3.0) self.assertEqual(set(interaction.trace.rewards), {"profit", "discipline", "gate"}) def test_a_hold_every_turn_scores_exactly_zero(self) -> None: """House rule 3's floor, through the whole stack rather than through the engine — including the degenerate reply that 31 of 32 one-shot rollouts produced.""" interaction = self.run_env([HOLD]) self.assertEqual(interaction.trace.metrics["held_replies"], float(TURNS)) self.assertEqual(interaction.trace.metrics["malformed_replies"], 0.0) for name, (score, _) in interaction.trace.rewards.items(): self.assertEqual(score, 0.0, name) def test_a_malformed_reply_is_a_hold_and_never_fails_the_episode(self) -> None: interaction = self.run_env(["I would rather not."]) self.assertEqual(interaction.trace.metrics["malformed_replies"], float(TURNS)) self.assertEqual(interaction.trace.rewards["profit"][0], 0.0) def test_a_real_sheet_trades(self) -> None: import json from grand_exchange_live.book import reference_orders env, task = self.env_and_task() market = build_market(task.data.seed, task.data.num_items, task.data.visible, HORIZON) sheet = "```json\n" + json.dumps({"orders": [ {"item": o.item, "quantity": o.quantity, "buy": o.buy, "sell": o.sell} for o in reference_orders(market) ]}) + "\n```" interaction = self.run_env([sheet, HOLD], env=env, task=task) self.assertGreater(interaction.trace.metrics["bought"], 0.0) self.assertGreater(interaction.trace.rewards["profit"][0], 0.0) def test_the_trace_carries_no_held_out_tick_and_no_market(self) -> None: """House rule 1's other half. The observation is `protocol.render`'s problem; this is the one that would put the whole series in the training data through `trace.info`.""" interaction = self.run_env([HOLD]) for key, value in interaction.trace.info.items(): self.assertIsInstance(value, (int, float, str), key) for key, value in interaction.trace.metrics.items(): self.assertIsInstance(value, float, key) self.assertLess(abs(value), PRICE_SENTINEL, key) self.assertEqual(set(interaction.trace.info), {"basket", "freeze"}) if __name__ == "__main__": unittest.main()