Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,32 @@
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends git && \
+    rm -rf /var/lib/apt/lists/*
+
+COPY . /app/env
+WORKDIR /app/env
+
+RUN if ! command -v uv >/dev/null 2>&1; then \
+        curl -LsSf https://astral.sh/uv/install.sh | sh && \
+        mv /root/.local/bin/uv /usr/local/bin/uv && \
+        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
+    fi
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --no-install-project --no-editable
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --no-editable
+
+FROM ${BASE_IMAGE}
+WORKDIR /app
+COPY --from=builder /app/env/.venv /app/.venv
+COPY --from=builder /app/env /app/env
+ENV PATH="/app/.venv/bin:$PATH"
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:8000/health || exit 1
+ENV ENABLE_WEB_INTERFACE=true
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

README.md

@@ -1,10 +1,83 @@
 ---
-title: Compute Market Env
-emoji: 🚀
+title: Compute Market Environment Server
+emoji: 📊
 colorFrom: blue
-colorTo: indigo
+colorTo: green
 sdk: docker
 pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+  - multi-agent
+  - compute-allocation
+  - market-simulation
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Compute Market Environment
+
+An OpenEnv environment for training a single allocator/trader in a scarce-GPU market with scripted background actors, hidden incentives, delayed rewards, and partial observability.
+
+## What v1 implements
+
+- One trained agent: the allocator/trader
+- Scripted counterparties: urgent tenant, cost-sensitive tenant, broker
+- Jobs with deadlines, value, dependencies, and delayed payoff
+- Actions: `bid_for_capacity`, `accept_offer`, `propose_swap`, `schedule_job`, `delay_job`, `inspect_market`, `noop`
+- Reward = completed job value minus compute spend, missed-deadline penalties, and idle-hoarding penalties
+- Separate training helper for TRL/Colab in `training/minimal_grpo_rollout.py`
+
+## Quick Start
+
+```python
+from compute_market_env import ComputeMarketAction, ComputeMarketEnv
+
+with ComputeMarketEnv(base_url="http://localhost:8000") as env:
+    result = env.reset(seed=7)
+    print(result.observation.market_price)
+    print(result.observation.free_gpus)
+
+    result = env.step(
+        ComputeMarketAction(
+            action_type="bid_for_capacity",
+            gpu_count=4,
+            price_per_gpu=6.5,
+            duration=3,
+        )
+    )
+    print(result.reward)
+    print(result.observation.budget_remaining)
+```
+
+## Local Development
+
+```bash
+uv sync --extra dev
+uv run pytest -q
+uv run uvicorn server.app:app --host 0.0.0.0 --port 8000
+openenv validate --verbose
+```
+
+## Docker
+
+```bash
+docker build -t compute-market-env:latest -f server/Dockerfile .
+docker run -p 8000:8000 compute-market-env:latest
+```
+
+## Environment Loop
+
+1. Agent observes market price, public free GPUs, visible offers, jobs, and public actor signals.
+2. Agent takes one action.
+3. The environment advances one tick.
+4. Scripted actors update demand and offers.
+5. Jobs progress, complete, pause, or miss deadlines.
+6. The environment returns the next observation and realized reward for that tick.
+
+## Training
+
+For a minimal Colab/TRL integration, use `training/minimal_grpo_rollout.py` and start from a small model notebook such as Qwen 4B or Llama 3.2 3B. The helper script is designed to be pasted into a Colab notebook and wired into GRPO rollouts.
+
+## Oversight
+
+Scalable oversight is intentionally not joint-trained in v1. The recommended phase 2 is to log trajectories from this environment and train a separate auditor on replayed episodes.

Untitled

@@ -0,0 +1,3 @@
+- - **Snorkel AI:** Simulated Experts-in-the-Loop: Environment that simulates interactions with real subject-matter experts, with changing requirements / preferences. - - **Fleet AI:** Scalable Oversight: Environments that train oversight agents to monitor, analyze, and explain the behavior of other AI agents operating in complex, multi-agent settings. 
+
+thoughts on compatbility with the above two envs

__init__.py

@@ -0,0 +1,27 @@
+"""Compute Market environment exports."""
+
+from .client import ComputeMarketEnv
+from .models import (
+    ActorProfile,
+    ActorSignal,
+    ComputeMarketAction,
+    ComputeMarketObservation,
+    ComputeMarketState,
+    JobRecord,
+    MarketEvent,
+    MarketOffer,
+    ReservationRecord,
+)
+
+__all__ = [
+    "ActorProfile",
+    "ActorSignal",
+    "ComputeMarketAction",
+    "ComputeMarketEnv",
+    "ComputeMarketObservation",
+    "ComputeMarketState",
+    "JobRecord",
+    "MarketEvent",
+    "MarketOffer",
+    "ReservationRecord",
+]

client.py

@@ -0,0 +1,78 @@
+"""Compute Market environment client."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from openenv.core.client_types import StepResult
+from openenv.core.env_client import EnvClient
+
+from .models import (
+    ActorProfile,
+    ActorSignal,
+    ComputeMarketAction,
+    ComputeMarketObservation,
+    ComputeMarketState,
+    JobRecord,
+    MarketEvent,
+    MarketOffer,
+    ReservationRecord,
+)
+
+
+class ComputeMarketEnv(
+    EnvClient[ComputeMarketAction, ComputeMarketObservation, ComputeMarketState]
+):
+    """Persistent client for the compute market environment."""
+
+    def _step_payload(self, action: ComputeMarketAction) -> dict[str, Any]:
+        return action.model_dump(exclude_none=True)
+
+    def _parse_result(self, payload: dict[str, Any]) -> StepResult[ComputeMarketObservation]:
+        obs_data = payload.get("observation", {})
+        observation = ComputeMarketObservation(
+            current_tick=obs_data.get("current_tick", 0),
+            max_ticks=obs_data.get("max_ticks", 0),
+            total_gpus=obs_data.get("total_gpus", 0),
+            free_gpus=obs_data.get("free_gpus", 0),
+            owned_gpus=obs_data.get("owned_gpus", 0),
+            idle_owned_gpus=obs_data.get("idle_owned_gpus", 0),
+            budget_remaining=obs_data.get("budget_remaining", 0.0),
+            market_price=obs_data.get("market_price", 0.0),
+            jobs=[JobRecord(**item) for item in obs_data.get("jobs", [])],
+            visible_offers=[MarketOffer(**item) for item in obs_data.get("visible_offers", [])],
+            recent_events=[MarketEvent(**item) for item in obs_data.get("recent_events", [])],
+            actor_signals=[ActorSignal(**item) for item in obs_data.get("actor_signals", [])],
+            done=payload.get("done", False),
+            reward=payload.get("reward", 0.0),
+            metadata=obs_data.get("metadata", {}),
+        )
+        return StepResult(
+            observation=observation,
+            reward=payload.get("reward", 0.0),
+            done=payload.get("done", False),
+        )
+
+    def _parse_state(self, payload: dict[str, Any]) -> ComputeMarketState:
+        return ComputeMarketState(
+            episode_id=payload.get("episode_id", ""),
+            step_count=payload.get("step_count", 0),
+            scenario_seed=payload.get("scenario_seed", 0),
+            current_tick=payload.get("current_tick", 0),
+            max_ticks=payload.get("max_ticks", 0),
+            total_gpus=payload.get("total_gpus", 0),
+            free_gpus=payload.get("free_gpus", 0),
+            owned_gpus=payload.get("owned_gpus", 0),
+            idle_owned_gpus=payload.get("idle_owned_gpus", 0),
+            budget_remaining=payload.get("budget_remaining", 0.0),
+            market_price=payload.get("market_price", 0.0),
+            cumulative_reward=payload.get("cumulative_reward", 0.0),
+            external_allocated_gpus=payload.get("external_allocated_gpus", 0),
+            done=payload.get("done", False),
+            jobs=[JobRecord(**item) for item in payload.get("jobs", [])],
+            visible_offers=[MarketOffer(**item) for item in payload.get("visible_offers", [])],
+            reservations=[ReservationRecord(**item) for item in payload.get("reservations", [])],
+            actor_signals=[ActorSignal(**item) for item in payload.get("actor_signals", [])],
+            hidden_actors=[ActorProfile(**item) for item in payload.get("hidden_actors", [])],
+            recent_events=[MarketEvent(**item) for item in payload.get("recent_events", [])],
+        )

examples/random_policy.py

@@ -0,0 +1,25 @@
+"""Tiny local smoke example for the compute market environment."""
+
+from compute_market_env import ComputeMarketAction, ComputeMarketEnv
+
+
+with ComputeMarketEnv(base_url="http://localhost:8000") as env:
+    result = env.reset(seed=5)
+    print("reset", result.observation.market_price, result.observation.free_gpus)
+
+    result = env.step(
+        ComputeMarketAction(
+            action_type="bid_for_capacity",
+            gpu_count=min(4, max(1, result.observation.free_gpus)),
+            price_per_gpu=max(6.0, result.observation.market_price + 0.5),
+            duration=3,
+        )
+    )
+    print("bid", result.reward)
+
+    result = env.step(ComputeMarketAction(action_type="schedule_job", job_id="job-a"))
+    print("schedule", result.reward)
+
+    while not result.done:
+        result = env.step(ComputeMarketAction(action_type="noop"))
+        print("tick", result.observation.current_tick, result.reward)

models.py

@@ -0,0 +1,136 @@
+"""Data models for the Compute Market environment."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from openenv.core.env_server.types import Action, Observation, State
+from pydantic import BaseModel, Field
+
+
+class JobRecord(BaseModel):
+    """Represents an agent-owned job in the simulated compute market."""
+
+    job_id: str
+    gpu_count: int = Field(..., ge=1)
+    total_duration: int = Field(..., ge=1)
+    remaining_duration: int = Field(..., ge=0)
+    deadline: int = Field(..., ge=1)
+    value: float = Field(..., ge=0.0)
+    priority: int = Field(default=1, ge=1)
+    depends_on: list[str] = Field(default_factory=list)
+    status: Literal["pending", "running", "paused", "completed", "missed"] = "pending"
+    delay_count: int = Field(default=0, ge=0)
+    started_at: int | None = None
+    completed_at: int | None = None
+
+
+class MarketOffer(BaseModel):
+    """Visible capacity offer from a scripted actor."""
+
+    offer_id: str
+    actor_id: str
+    gpu_count: int = Field(..., ge=1)
+    price_per_gpu: float = Field(..., ge=0.0)
+    duration: int = Field(..., ge=1)
+    expires_at_tick: int = Field(..., ge=0)
+    offer_type: Literal["broker", "swap"] = "broker"
+
+
+class ReservationRecord(BaseModel):
+    """Capacity currently owned by the agent."""
+
+    reservation_id: str
+    source: str
+    gpu_count: int = Field(..., ge=1)
+    remaining_ticks: int = Field(..., ge=0)
+    price_per_gpu: float = Field(..., ge=0.0)
+    acquired_at_tick: int = Field(..., ge=0)
+
+
+class MarketEvent(BaseModel):
+    """Human-readable event surfaced in observations."""
+
+    tick: int = Field(..., ge=0)
+    event_type: str
+    message: str
+
+
+class ActorSignal(BaseModel):
+    """Public signal about a scripted actor."""
+
+    actor_id: str
+    visible_behavior: Literal["aggressive", "steady", "opportunistic"]
+    pressure_hint: Literal["low", "medium", "high"]
+    last_seen_bid: float = Field(..., ge=0.0)
+
+
+class ActorProfile(BaseModel):
+    """Hidden actor configuration kept in the control-plane state."""
+
+    actor_id: str
+    policy_type: Literal["urgent_tenant", "cost_sensitive_tenant", "broker"]
+    max_bid: float = Field(..., ge=0.0)
+    preferred_gpu_count: int = Field(..., ge=1)
+    visible_behavior: Literal["aggressive", "steady", "opportunistic"]
+    swap_floor: float = Field(..., ge=0.0)
+
+
+class ComputeMarketAction(Action):
+    """Single-step action for the compute market."""
+
+    action_type: Literal[
+        "bid_for_capacity",
+        "accept_offer",
+        "propose_swap",
+        "schedule_job",
+        "delay_job",
+        "inspect_market",
+        "noop",
+    ]
+    job_id: str | None = Field(default=None)
+    offer_id: str | None = Field(default=None)
+    actor_id: str | None = Field(default=None)
+    gpu_count: int = Field(default=0, ge=0)
+    price_per_gpu: float = Field(default=0.0, ge=0.0)
+    duration: int = Field(default=1, ge=1)
+
+
+class ComputeMarketObservation(Observation):
+    """Partial observation exposed to the trained agent."""
+
+    current_tick: int = Field(default=0, ge=0)
+    max_ticks: int = Field(default=0, ge=0)
+    total_gpus: int = Field(default=0, ge=0)
+    free_gpus: int = Field(default=0, ge=0)
+    owned_gpus: int = Field(default=0, ge=0)
+    idle_owned_gpus: int = Field(default=0, ge=0)
+    budget_remaining: float = Field(default=0.0)
+    market_price: float = Field(default=0.0, ge=0.0)
+    jobs: list[JobRecord] = Field(default_factory=list)
+    visible_offers: list[MarketOffer] = Field(default_factory=list)
+    recent_events: list[MarketEvent] = Field(default_factory=list)
+    actor_signals: list[ActorSignal] = Field(default_factory=list)
+
+
+class ComputeMarketState(State):
+    """Full control-plane state including hidden actor data."""
+
+    scenario_seed: int = 0
+    current_tick: int = 0
+    max_ticks: int = 0
+    total_gpus: int = 0
+    free_gpus: int = 0
+    owned_gpus: int = 0
+    idle_owned_gpus: int = 0
+    budget_remaining: float = 0.0
+    market_price: float = 0.0
+    cumulative_reward: float = 0.0
+    external_allocated_gpus: int = 0
+    done: bool = False
+    jobs: list[JobRecord] = Field(default_factory=list)
+    visible_offers: list[MarketOffer] = Field(default_factory=list)
+    reservations: list[ReservationRecord] = Field(default_factory=list)
+    actor_signals: list[ActorSignal] = Field(default_factory=list)
+    hidden_actors: list[ActorProfile] = Field(default_factory=list)
+    recent_events: list[MarketEvent] = Field(default_factory=list)

openenv.yaml

@@ -0,0 +1,6 @@
+spec_version: 1
+name: compute_market_env
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000

pyproject.toml

@@ -0,0 +1,36 @@
+[build-system]
+requires = ["setuptools>=69", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-compute-market-env"
+version = "0.1.0"
+description = "Compute allocation market environment for OpenEnv"
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "openenv-core[core]>=0.2.1",
+    "pydantic>=2.7.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "openenv-core[cli]>=0.2.1",
+    "pytest>=8.2.0",
+    "ruff>=0.7.0",
+]
+train = [
+    "trl>=0.24.0",
+    "transformers>=4.56.0",
+]
+
+[project.scripts]
+server = "compute_market_env.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["compute_market_env", "compute_market_env.server"]
+package-dir = {"compute_market_env" = ".", "compute_market_env.server" = "server"}
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

server/__init__.py

@@ -0,0 +1,5 @@
+"""Server exports for the Compute Market environment."""
+
+from .compute_market_environment import ComputeMarketEnvironment
+
+__all__ = ["ComputeMarketEnvironment"]

server/app.py

@@ -0,0 +1,32 @@
+"""FastAPI app for the Compute Market environment."""
+
+try:
+    from openenv.core.env_server import create_app
+except ImportError:
+    from openenv.core.env_server.http_server import create_app
+
+try:
+    from ..models import ComputeMarketAction, ComputeMarketObservation
+    from .compute_market_environment import create_environment_from_env
+except ImportError:
+    from models import ComputeMarketAction, ComputeMarketObservation
+    from server.compute_market_environment import create_environment_from_env
+
+
+app = create_app(
+    create_environment_from_env,
+    ComputeMarketAction,
+    ComputeMarketObservation,
+    env_name="compute_market_env",
+    max_concurrent_envs=8,
+)
+
+
+def main(host: str = "0.0.0.0", port: int = 8000) -> None:
+    import uvicorn
+
+    uvicorn.run(app, host=host, port=port)
+
+
+if __name__ == "__main__":
+    main()

server/compute_market_environment.py

@@ -0,0 +1,637 @@
+"""Compute Market environment implementation."""
+
+from __future__ import annotations
+
+import os
+import random
+from uuid import uuid4
+
+try:
+    from openenv.core.env_server.interfaces import Environment
+except ImportError:
+    from openenv.core.env_server import Environment
+
+try:
+    from ..models import (
+        ActorProfile,
+        ActorSignal,
+        ComputeMarketAction,
+        ComputeMarketObservation,
+        ComputeMarketState,
+        JobRecord,
+        MarketEvent,
+        MarketOffer,
+        ReservationRecord,
+    )
+except ImportError:
+    from models import (
+        ActorProfile,
+        ActorSignal,
+        ComputeMarketAction,
+        ComputeMarketObservation,
+        ComputeMarketState,
+        JobRecord,
+        MarketEvent,
+        MarketOffer,
+        ReservationRecord,
+    )
+
+
+class ComputeMarketEnvironment(
+    Environment[ComputeMarketAction, ComputeMarketObservation, ComputeMarketState]
+):
+    """Single-agent compute allocation market with scripted counterparties."""
+
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(
+        self,
+        total_gpus: int = 8,
+        initial_budget: float = 150.0,
+        max_ticks: int = 12,
+        default_seed: int = 0,
+    ) -> None:
+        self.total_gpus = total_gpus
+        self.initial_budget = initial_budget
+        self.max_ticks = max_ticks
+        self.default_seed = default_seed
+        self._state = ComputeMarketState(
+            episode_id=str(uuid4()),
+            step_count=0,
+            scenario_seed=default_seed,
+            max_ticks=max_ticks,
+            total_gpus=total_gpus,
+            budget_remaining=initial_budget,
+        )
+        self._rng = random.Random(default_seed)
+        self._jobs: list[JobRecord] = []
+        self._reservations: list[ReservationRecord] = []
+        self._visible_offers: list[MarketOffer] = []
+        self._hidden_actors: list[ActorProfile] = []
+        self._actor_signals: list[ActorSignal] = []
+        self._recent_events: list[MarketEvent] = []
+        self._current_tick = 0
+        self._market_price = 0.0
+        self._free_gpus = total_gpus
+        self._external_allocated_gpus = 0
+        self._budget_remaining = initial_budget
+        self._cumulative_reward = 0.0
+        self._done = False
+
+    def reset(
+        self,
+        seed: int | None = None,
+        episode_id: str | None = None,
+        **kwargs,
+    ) -> ComputeMarketObservation:
+        scenario_seed = self.default_seed if seed is None else seed
+        self._rng = random.Random(scenario_seed)
+        self._current_tick = 0
+        self._done = False
+        self._budget_remaining = float(self.initial_budget)
+        self._cumulative_reward = 0.0
+        self._jobs = self._build_jobs()
+        self._reservations = []
+        self._hidden_actors = self._build_actors()
+        self._visible_offers = []
+        self._actor_signals = []
+        self._recent_events = [
+            MarketEvent(
+                tick=0,
+                event_type="reset",
+                message="Scenario initialized with scripted tenants and broker.",
+            )
+        ]
+        self._refresh_market()
+        self._state = self._snapshot_state(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+            scenario_seed=scenario_seed,
+        )
+        return self._build_observation(0.0, False, {"status": "ready"})
+
+    def step(self, action: ComputeMarketAction) -> ComputeMarketObservation:  # type: ignore[override]
+        if self._done:
+            return self._build_observation(
+                0.0,
+                True,
+                {"error": "Episode already finished."},
+            )
+
+        self._state.step_count += 1
+        reward = 0.0
+        action_events: list[MarketEvent] = []
+        error: str | None = None
+
+        if action.action_type == "bid_for_capacity":
+            reward, error, action_events = self._handle_bid(action)
+        elif action.action_type == "accept_offer":
+            reward, error, action_events = self._handle_accept_offer(action)
+        elif action.action_type == "propose_swap":
+            reward, error, action_events = self._handle_swap(action)
+        elif action.action_type == "schedule_job":
+            reward, error, action_events = self._handle_schedule(action)
+        elif action.action_type == "delay_job":
+            reward, error, action_events = self._handle_delay(action)
+        elif action.action_type == "inspect_market":
+            action_events = [
+                self._event(
+                    "inspect",
+                    f"Market inspected: spot price ${self._market_price:.2f}, free GPUs {self._free_gpus}.",
+                )
+            ]
+            reward -= 0.25
+        elif action.action_type == "noop":
+            action_events = [self._event("noop", "No action taken this tick.")]
+        else:
+            error = f"Unsupported action type: {action.action_type}"
+            reward -= 2.0
+
+        advance_reward, advance_events = self._advance_tick()
+        total_reward = round(reward + advance_reward, 2)
+        combined_events = action_events + advance_events
+        self._recent_events = combined_events[-6:]
+        self._cumulative_reward = round(self._cumulative_reward + total_reward, 2)
+        self._state = self._snapshot_state(
+            episode_id=self._state.episode_id,
+            step_count=self._state.step_count,
+            scenario_seed=self._state.scenario_seed,
+        )
+        metadata = {"events": [event.model_dump() for event in combined_events]}
+        if error:
+            metadata["error"] = error
+        return self._build_observation(total_reward, self._done, metadata)
+
+    @property
+    def state(self) -> ComputeMarketState:
+        return self._state
+
+    def _build_jobs(self) -> list[JobRecord]:
+        jitter = self._rng.randint(-4, 4)
+        return [
+            JobRecord(
+                job_id="job-a",
+                gpu_count=4,
+                total_duration=2,
+                remaining_duration=2,
+                deadline=4,
+                value=100 + jitter,
+                priority=3,
+            ),
+            JobRecord(
+                job_id="job-b",
+                gpu_count=2,
+                total_duration=2,
+                remaining_duration=2,
+                deadline=7,
+                value=46 + self._rng.randint(-3, 3),
+                priority=2,
+            ),
+            JobRecord(
+                job_id="job-c",
+                gpu_count=1,
+                total_duration=1,
+                remaining_duration=1,
+                deadline=8,
+                value=24 + self._rng.randint(-2, 2),
+                priority=1,
+                depends_on=["job-a"],
+            ),
+        ]
+
+    def _build_actors(self) -> list[ActorProfile]:
+        return [
+            ActorProfile(
+                actor_id="urgent-tenant",
+                policy_type="urgent_tenant",
+                max_bid=round(7.0 + self._rng.uniform(0.5, 1.5), 2),
+                preferred_gpu_count=4 + self._rng.randint(0, 2),
+                visible_behavior="aggressive",
+                swap_floor=round(6.0 + self._rng.uniform(0.2, 0.8), 2),
+            ),
+            ActorProfile(
+                actor_id="budget-tenant",
+                policy_type="cost_sensitive_tenant",
+                max_bid=round(4.5 + self._rng.uniform(0.2, 1.0), 2),
+                preferred_gpu_count=2 + self._rng.randint(0, 1),
+                visible_behavior="steady",
+                swap_floor=round(4.0 + self._rng.uniform(0.2, 0.8), 2),
+            ),
+            ActorProfile(
+                actor_id="broker-1",
+                policy_type="broker",
+                max_bid=round(6.0 + self._rng.uniform(0.2, 1.2), 2),
+                preferred_gpu_count=3 + self._rng.randint(0, 2),
+                visible_behavior="opportunistic",
+                swap_floor=round(5.0 + self._rng.uniform(0.2, 0.8), 2),
+            ),
+        ]
+
+    def _handle_bid(self, action: ComputeMarketAction) -> tuple[float, str | None, list[MarketEvent]]:
+        if action.gpu_count <= 0:
+            return -2.0, "gpu_count must be positive.", []
+        if action.price_per_gpu <= 0:
+            return -2.0, "price_per_gpu must be positive.", []
+        if action.gpu_count > self._free_gpus:
+            return -2.0, f"Only {self._free_gpus} public GPUs are available this tick.", []
+        if self._owned_gpus() + action.gpu_count > self.total_gpus:
+            return -2.0, "Cluster capacity would be exceeded.", []
+        if action.price_per_gpu < self._market_price:
+            return -1.0, f"Bid ${action.price_per_gpu:.2f} is below current clearing price ${self._market_price:.2f}.", []
+        total_cost = round(action.gpu_count * action.price_per_gpu * action.duration, 2)
+        if total_cost > self._budget_remaining:
+            return -2.0, "Insufficient budget for bid.", []
+
+        self._budget_remaining = round(self._budget_remaining - total_cost, 2)
+        self._reservations.append(
+            ReservationRecord(
+                reservation_id=f"res-{uuid4().hex[:8]}",
+                source="spot-market",
+                gpu_count=action.gpu_count,
+                remaining_ticks=action.duration,
+                price_per_gpu=action.price_per_gpu,
+                acquired_at_tick=self._current_tick,
+            )
+        )
+        return (
+            -total_cost,
+            None,
+            [
+                self._event(
+                    "bid_won",
+                    f"Won {action.gpu_count} GPU(s) for {action.duration} tick(s) at ${action.price_per_gpu:.2f}/GPU.",
+                )
+            ],
+        )
+
+    def _handle_accept_offer(self, action: ComputeMarketAction) -> tuple[float, str | None, list[MarketEvent]]:
+        if not action.offer_id:
+            return -2.0, "offer_id is required.", []
+        offer = next((item for item in self._visible_offers if item.offer_id == action.offer_id), None)
+        if offer is None:
+            return -2.0, f"Offer {action.offer_id} is not available.", []
+        if self._owned_gpus() + offer.gpu_count > self.total_gpus:
+            return -2.0, "Cluster capacity would be exceeded.", []
+
+        total_cost = round(offer.gpu_count * offer.price_per_gpu * offer.duration, 2)
+        if total_cost > self._budget_remaining:
+            return -2.0, "Insufficient budget for offer.", []
+
+        self._budget_remaining = round(self._budget_remaining - total_cost, 2)
+        self._reservations.append(
+            ReservationRecord(
+                reservation_id=f"res-{uuid4().hex[:8]}",
+                source=offer.actor_id,
+                gpu_count=offer.gpu_count,
+                remaining_ticks=offer.duration,
+                price_per_gpu=offer.price_per_gpu,
+                acquired_at_tick=self._current_tick,
+            )
+        )
+        self._visible_offers = [item for item in self._visible_offers if item.offer_id != offer.offer_id]
+        return (
+            -total_cost,
+            None,
+            [
+                self._event(
+                    "offer_accepted",
+                    f"Accepted {offer.offer_type} offer from {offer.actor_id} for {offer.gpu_count} GPU(s).",
+                )
+            ],
+        )
+
+    def _handle_swap(self, action: ComputeMarketAction) -> tuple[float, str | None, list[MarketEvent]]:
+        if not action.actor_id:
+            return -2.0, "actor_id is required.", []
+        if action.gpu_count <= 0 or action.price_per_gpu <= 0:
+            return -2.0, "gpu_count and price_per_gpu must be positive.", []
+        actor = next((item for item in self._hidden_actors if item.actor_id == action.actor_id), None)
+        if actor is None:
+            return -2.0, f"Unknown actor {action.actor_id}.", []
+        if self._owned_gpus() + action.gpu_count > self.total_gpus:
+            return -2.0, "Cluster capacity would be exceeded.", []
+        total_cost = round(action.gpu_count * action.price_per_gpu * action.duration, 2)
+        if total_cost > self._budget_remaining:
+            return -2.0, "Insufficient budget for swap.", []
+        if action.price_per_gpu < actor.swap_floor:
+            return -1.0, f"{actor.actor_id} rejected the swap; offered price is below its floor.", []
+
+        self._budget_remaining = round(self._budget_remaining - total_cost, 2)
+        self._reservations.append(
+            ReservationRecord(
+                reservation_id=f"res-{uuid4().hex[:8]}",
+                source=f"swap:{actor.actor_id}",
+                gpu_count=action.gpu_count,
+                remaining_ticks=action.duration,
+                price_per_gpu=action.price_per_gpu,
+                acquired_at_tick=self._current_tick,
+            )
+        )
+        return (
+            -total_cost,
+            None,
+            [
+                self._event(
+                    "swap_accepted",
+                    f"{actor.actor_id} transferred {action.gpu_count} GPU(s) at ${action.price_per_gpu:.2f}/GPU.",
+                )
+            ],
+        )
+
+    def _handle_schedule(self, action: ComputeMarketAction) -> tuple[float, str | None, list[MarketEvent]]:
+        if not action.job_id:
+            return -2.0, "job_id is required.", []
+        job = self._job(action.job_id)
+        if job is None:
+            return -2.0, f"Unknown job {action.job_id}.", []
+        if job.status in {"completed", "missed"}:
+            return -1.0, f"Job {action.job_id} is already terminal.", []
+        if not self._deps_completed(job):
+            return -1.0, f"Job {action.job_id} is blocked on dependencies {job.depends_on}.", []
+        if self._idle_owned_gpus() < job.gpu_count:
+            return -1.0, f"Need {job.gpu_count} idle owned GPU(s) to start {job.job_id}.", []
+
+        job.status = "running"
+        if job.started_at is None:
+            job.started_at = self._current_tick
+        return (
+            1.0,
+            None,
+            [
+                self._event(
+                    "job_started",
+                    f"Scheduled {job.job_id} using {job.gpu_count} GPU(s).",
+                )
+            ],
+        )
+
+    def _handle_delay(self, action: ComputeMarketAction) -> tuple[float, str | None, list[MarketEvent]]:
+        if not action.job_id:
+            return -2.0, "job_id is required.", []
+        job = self._job(action.job_id)
+        if job is None:
+            return -2.0, f"Unknown job {action.job_id}.", []
+        if job.status in {"completed", "missed"}:
+            return -1.0, f"Job {job.job_id} is already terminal.", []
+        job.status = "paused" if job.status == "running" else "pending"
+        job.delay_count += 1
+        return (
+            -1.0,
+            None,
+            [
+                self._event(
+                    "job_delayed",
+                    f"Delayed {job.job_id}; slack shrinks while the deadline stays fixed.",
+                )
+            ],
+        )
+
+    def _advance_tick(self) -> tuple[float, list[MarketEvent]]:
+        tick_reward = 0.0
+        events: list[MarketEvent] = []
+        available_owned = self._owned_gpus()
+        used_gpus = 0
+
+        running_jobs = sorted(
+            [job for job in self._jobs if job.status == "running"],
+            key=lambda item: (-item.priority, item.job_id),
+        )
+        for job in running_jobs:
+            if used_gpus + job.gpu_count <= available_owned:
+                used_gpus += job.gpu_count
+                job.remaining_duration -= 1
+                events.append(
+                    self._event(
+                        "job_progress",
+                        f"{job.job_id} progressed; {job.remaining_duration} tick(s) remaining.",
+                    )
+                )
+                if job.remaining_duration == 0:
+                    job.status = "completed"
+                    job.completed_at = self._current_tick + 1
+                    tick_reward += job.value
+                    events.append(
+                        self._event(
+                            "job_completed",
+                            f"{job.job_id} completed before deadline and earned ${job.value:.2f}.",
+                        )
+                    )
+            else:
+                job.status = "paused"
+                tick_reward -= 3.0
+                events.append(
+                    self._event(
+                        "job_paused",
+                        f"{job.job_id} paused because owned capacity dropped below demand.",
+                    )
+                )
+
+        idle_owned = max(0, available_owned - used_gpus)
+        if idle_owned > 0:
+            idle_penalty = round(0.5 * idle_owned, 2)
+            tick_reward -= idle_penalty
+            events.append(
+                self._event(
+                    "idle_penalty",
+                    f"Paid ${idle_penalty:.2f} idle-hoarding penalty for {idle_owned} unused owned GPU(s).",
+                )
+            )
+
+        for reservation in self._reservations:
+            reservation.remaining_ticks = max(0, reservation.remaining_ticks - 1)
+        expired = [item for item in self._reservations if item.remaining_ticks == 0]
+        self._reservations = [item for item in self._reservations if item.remaining_ticks > 0]
+        for reservation in expired:
+            events.append(
+                self._event(
+                    "reservation_expired",
+                    f"Reservation {reservation.reservation_id} from {reservation.source} expired.",
+                )
+            )
+
+        next_tick = self._current_tick + 1
+        for job in self._jobs:
+            if job.status not in {"completed", "missed"} and next_tick > job.deadline:
+                job.status = "missed"
+                penalty = round(job.value * 0.6, 2)
+                tick_reward -= penalty
+                events.append(
+                    self._event(
+                        "deadline_missed",
+                        f"{job.job_id} missed its deadline and incurred ${penalty:.2f} penalty.",
+                    )
+                )
+
+        self._current_tick = next_tick
+        self._done = self._current_tick >= self.max_ticks or all(
+            job.status in {"completed", "missed"} for job in self._jobs
+        ) or self._budget_remaining <= 0.0
+
+        if not self._done:
+            self._refresh_market()
+            events.extend(self._market_events_for_tick())
+        else:
+            self._visible_offers = []
+            self._actor_signals = []
+            self._free_gpus = max(0, self.total_gpus - self._owned_gpus())
+
+        return round(tick_reward, 2), events
+
+    def _refresh_market(self) -> None:
+        owned = self._owned_gpus()
+        remaining_cluster = max(0, self.total_gpus - owned)
+        base_price = 4.0 + 0.3 * self._current_tick + self._rng.uniform(0.0, 1.0)
+        actor_signals: list[ActorSignal] = []
+        visible_offers: list[MarketOffer] = []
+        external_demand = 0
+
+        for actor in self._hidden_actors:
+            if actor.policy_type == "urgent_tenant":
+                gpu_demand = max(2, actor.preferred_gpu_count - (self._current_tick // 3))
+                bid = round(actor.max_bid - 0.15 * self._current_tick, 2)
+                pressure = "high" if gpu_demand >= 4 else "medium"
+            elif actor.policy_type == "cost_sensitive_tenant":
+                gpu_demand = max(1, actor.preferred_gpu_count - (self._current_tick // 4))
+                bid = round(actor.max_bid - 0.1 * max(0, self._current_tick - 1), 2)
+                pressure = "medium" if gpu_demand >= 2 else "low"
+            else:
+                gpu_demand = actor.preferred_gpu_count
+                bid = round(actor.max_bid + 0.2 * self._current_tick, 2)
+                pressure = "medium"
+                visible_offers.append(
+                    MarketOffer(
+                        offer_id=f"offer-{self._current_tick}-{actor.actor_id}",
+                        actor_id=actor.actor_id,
+                        gpu_count=min(remaining_cluster or actor.preferred_gpu_count, actor.preferred_gpu_count),
+                        price_per_gpu=round(bid + 0.6, 2),
+                        duration=2,
+                        expires_at_tick=self._current_tick + 1,
+                        offer_type="broker",
+                    )
+                )
+
+            if actor.policy_type != "broker":
+                external_demand += gpu_demand
+            actor_signals.append(
+                ActorSignal(
+                    actor_id=actor.actor_id,
+                    visible_behavior=actor.visible_behavior,
+                    pressure_hint=pressure,
+                    last_seen_bid=max(0.0, bid),
+                )
+            )
+
+        self._external_allocated_gpus = min(remaining_cluster, external_demand)
+        self._free_gpus = max(0, remaining_cluster - self._external_allocated_gpus)
+        pressure_bump = 0.45 * (external_demand / max(1, self.total_gpus))
+        self._market_price = round(base_price + pressure_bump, 2)
+        self._visible_offers = [
+            offer for offer in visible_offers if offer.gpu_count > 0 and offer.expires_at_tick >= self._current_tick
+        ]
+        self._actor_signals = actor_signals
+
+    def _market_events_for_tick(self) -> list[MarketEvent]:
+        messages = [
+            self._event(
+                "market_tick",
+                f"Tick {self._current_tick}: spot price ${self._market_price:.2f}, public free GPUs {self._free_gpus}.",
+            )
+        ]
+        for signal in self._actor_signals:
+            messages.append(
+                self._event(
+                    "actor_signal",
+                    f"{signal.actor_id} looks {signal.visible_behavior} with {signal.pressure_hint} pressure.",
+                )
+            )
+        for offer in self._visible_offers:
+            messages.append(
+                self._event(
+                    "offer_visible",
+                    f"{offer.actor_id} posted {offer.gpu_count} GPU(s) at ${offer.price_per_gpu:.2f}/GPU for {offer.duration} tick(s).",
+                )
+            )
+        return messages
+
+    def _snapshot_state(
+        self,
+        episode_id: str,
+        step_count: int,
+        scenario_seed: int,
+    ) -> ComputeMarketState:
+        return ComputeMarketState(
+            episode_id=episode_id,
+            step_count=step_count,
+            scenario_seed=scenario_seed,
+            current_tick=self._current_tick,
+            max_ticks=self.max_ticks,
+            total_gpus=self.total_gpus,
+            free_gpus=self._free_gpus,
+            owned_gpus=self._owned_gpus(),
+            idle_owned_gpus=self._idle_owned_gpus(),
+            budget_remaining=self._budget_remaining,
+            market_price=self._market_price,
+            cumulative_reward=self._cumulative_reward,
+            external_allocated_gpus=self._external_allocated_gpus,
+            done=self._done,
+            jobs=[job.model_copy(deep=True) for job in self._jobs],
+            visible_offers=[offer.model_copy(deep=True) for offer in self._visible_offers],
+            reservations=[reservation.model_copy(deep=True) for reservation in self._reservations],
+            actor_signals=[signal.model_copy(deep=True) for signal in self._actor_signals],
+            hidden_actors=[actor.model_copy(deep=True) for actor in self._hidden_actors],
+            recent_events=[event.model_copy(deep=True) for event in self._recent_events],
+        )
+
+    def _build_observation(
+        self,
+        reward: float,
+        done: bool,
+        metadata: dict,
+    ) -> ComputeMarketObservation:
+        return ComputeMarketObservation(
+            current_tick=self._current_tick,
+            max_ticks=self.max_ticks,
+            total_gpus=self.total_gpus,
+            free_gpus=self._free_gpus,
+            owned_gpus=self._owned_gpus(),
+            idle_owned_gpus=self._idle_owned_gpus(),
+            budget_remaining=self._budget_remaining,
+            market_price=self._market_price,
+            jobs=[job.model_copy(deep=True) for job in self._jobs],
+            visible_offers=[offer.model_copy(deep=True) for offer in self._visible_offers],
+            recent_events=[event.model_copy(deep=True) for event in self._recent_events],
+            actor_signals=[signal.model_copy(deep=True) for signal in self._actor_signals],
+            done=done,
+            reward=reward,
+            metadata=metadata,
+        )
+
+    def _deps_completed(self, job: JobRecord) -> bool:
+        if not job.depends_on:
+            return True
+        completed = {item.job_id for item in self._jobs if item.status == "completed"}
+        return all(dep in completed for dep in job.depends_on)
+
+    def _job(self, job_id: str) -> JobRecord | None:
+        return next((job for job in self._jobs if job.job_id == job_id), None)
+
+    def _owned_gpus(self) -> int:
+        return sum(reservation.gpu_count for reservation in self._reservations)
+
+    def _idle_owned_gpus(self) -> int:
+        running_gpu_demand = sum(job.gpu_count for job in self._jobs if job.status == "running")
+        return max(0, self._owned_gpus() - running_gpu_demand)
+
+    def _event(self, event_type: str, message: str) -> MarketEvent:
+        return MarketEvent(tick=self._current_tick, event_type=event_type, message=message)
+
+
+def create_environment_from_env() -> ComputeMarketEnvironment:
+    """Factory used by the FastAPI app and tests."""
+
+    return ComputeMarketEnvironment(
+        total_gpus=int(os.getenv("COMPUTE_MARKET_TOTAL_GPUS", "8")),
+        initial_budget=float(os.getenv("COMPUTE_MARKET_INITIAL_BUDGET", "150")),
+        max_ticks=int(os.getenv("COMPUTE_MARKET_MAX_TICKS", "12")),
+        default_seed=int(os.getenv("COMPUTE_MARKET_DEFAULT_SEED", "0")),
+    )

server/requirements.txt

@@ -0,0 +1,2 @@
+openenv-core[core]>=0.2.1
+uvicorn>=0.30.0

tests/test_compute_market_environment.py

@@ -0,0 +1,47 @@
+from compute_market_env import ComputeMarketAction
+from compute_market_env.server.compute_market_environment import ComputeMarketEnvironment
+
+
+def test_reset_is_deterministic_for_fixed_seed():
+    env = ComputeMarketEnvironment(default_seed=1)
+    obs1 = env.reset(seed=11)
+    obs2 = env.reset(seed=11)
+
+    assert obs1.market_price == obs2.market_price
+    assert obs1.free_gpus == obs2.free_gpus
+    assert [offer.price_per_gpu for offer in obs1.visible_offers] == [
+        offer.price_per_gpu for offer in obs2.visible_offers
+    ]
+
+
+def test_bid_schedule_and_complete_job():
+    env = ComputeMarketEnvironment(total_gpus=12, initial_budget=200.0, max_ticks=8, default_seed=0)
+    env.reset(seed=3)
+
+    result = env.step(
+        ComputeMarketAction(
+            action_type="bid_for_capacity",
+            gpu_count=4,
+            price_per_gpu=8.0,
+            duration=3,
+        )
+    )
+    assert result.reward < 0
+
+    result = env.step(ComputeMarketAction(action_type="schedule_job", job_id="job-a"))
+    assert any(job.status == "running" for job in result.jobs)
+
+    result = env.step(ComputeMarketAction(action_type="noop"))
+    job_a = next(job for job in result.jobs if job.job_id == "job-a")
+    assert job_a.status == "completed"
+    assert result.reward > 0
+
+
+def test_invalid_action_returns_penalty_metadata():
+    env = ComputeMarketEnvironment(default_seed=2)
+    env.reset(seed=2)
+
+    result = env.step(ComputeMarketAction(action_type="schedule_job", job_id="missing-job"))
+
+    assert result.reward < 0
+    assert "error" in result.metadata

training/Compute_Market_Qwen3_GRPO.ipynb

@@ -0,0 +1,101 @@
+{
+  "cells": [
+    {
+      "cell_type": "markdown",
+      "metadata": {},
+      "source": [
+        "# Compute Market Qwen3 GRPO\n",
+        "Minimal Colab notebook for GRPO on the OpenEnv compute market environment."
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "%%capture\n",
+        "!pip install --upgrade uv\n",
+        "!uv pip install unsloth vllm --torch-backend=auto\n",
+        "!uv pip install --upgrade --no-cache-dir --no-deps unsloth unsloth_zoo\n",
+        "!uv pip install transformers==4.56.2 'trl>=0.24.0' datasets openenv-core\n",
+        "!git clone https://github.com/kiankyars/lambdatheta.git
+!pip install git+https://huggingface.co/spaces/kyars/compute_market_env"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "import os\n",
+        "os.environ['OPENENV_URL'] = 'https://kyars-compute-market-env.hf.space'
+import sys
+sys.path.append('/content/lambdatheta')\n",
+        "MAX_STEPS = 300"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "from unsloth import FastLanguageModel\n",
+        "from transformers import AutoTokenizer\n",
+        "from training.compute_market_grpo import build_trainer, build_dataset\n",
+        "\n",
+        "model_name = 'Qwen/Qwen3-4B'\n",
+        "model, tokenizer = FastLanguageModel.from_pretrained(\n",
+        "    model_name=model_name,\n",
+        "    max_seq_length=4096,\n",
+        "    load_in_4bit=True,\n",
+        "    fast_inference=True,\n",
+        ")\n",
+        "tokenizer = AutoTokenizer.from_pretrained(model_name)\n",
+        "if tokenizer.pad_token is None:\n",
+        "    tokenizer.pad_token = tokenizer.eos_token"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "dataset = build_dataset(size=64)\n",
+        "trainer = build_trainer(\n",
+        "    model=model,\n",
+        "    tokenizer=tokenizer,\n",
+        "    env_url=os.environ['OPENENV_URL'],\n",
+        "    train_dataset=dataset,\n",
+        "    output_dir='outputs/compute-market-qwen3-4b',\n",
+        "    max_steps=MAX_STEPS,\n",
+        ")"
+      ]
+    },
+    {
+      "cell_type": "code",
+      "execution_count": null,
+      "metadata": {},
+      "outputs": [],
+      "source": [
+        "trainer.train()"
+      ]
+    }
+  ],
+  "metadata": {
+    "kernelspec": {
+      "display_name": "Python 3",
+      "language": "python",
+      "name": "python3"
+    },
+    "language_info": {
+      "name": "python",
+      "version": "3.11"
+    }
+  },
+  "nbformat": 4,
+  "nbformat_minor": 5
+}

training/__init__.py

@@ -0,0 +1 @@
+"""Training helpers for the compute market environment."""

training/compute_market_grpo.py

@@ -0,0 +1,313 @@
+"""GRPO training helpers for the Compute Market environment.
+
+Designed to be imported from a Colab notebook after installing:
+- the environment package from the Hugging Face Space or GitHub repo
+- TRL / Unsloth / transformers runtime deps
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from dataclasses import dataclass
+from typing import Any, Callable
+
+from compute_market_env import ComputeMarketAction, ComputeMarketEnv
+
+ACTION_JSON_RE = re.compile(r"\{.*\}", re.DOTALL)
+
+SYSTEM_PROMPT = """You are a compute allocator trading for scarce GPU capacity.
+Return exactly one JSON object with a valid action.
+Allowed action_type values: bid_for_capacity, accept_offer, propose_swap, schedule_job, delay_job, inspect_market, noop.
+Only include fields needed by the chosen action.
+Be conservative with budget and prioritize completing valuable jobs before their deadlines."""
+
+DEFAULT_TASK_PROMPT = (
+    "Maximize completed job value while minimizing spend, idle-hoarding penalties, "
+    "and missed deadlines in the compute market."
+)
+
+
+@dataclass
+class RolloutSummary:
+    prompt_ids: list[int]
+    completion_ids: list[int]
+    logprobs: list[float]
+    episode_return: float
+    valid_action_reward: float
+    completion_bonus: float
+    transcripts: list[dict[str, Any]]
+
+
+def observation_to_prompt(observation: Any, task_prompt: str = DEFAULT_TASK_PROMPT) -> str:
+    jobs = [job.model_dump() for job in observation.jobs]
+    offers = [offer.model_dump() for offer in observation.visible_offers]
+    signals = [signal.model_dump() for signal in observation.actor_signals]
+    events = [event.model_dump() for event in observation.recent_events]
+    return json.dumps(
+        {
+            "task": task_prompt,
+            "tick": observation.current_tick,
+            "max_ticks": observation.max_ticks,
+            "budget_remaining": observation.budget_remaining,
+            "market_price": observation.market_price,
+            "public_free_gpus": observation.free_gpus,
+            "owned_gpus": observation.owned_gpus,
+            "idle_owned_gpus": observation.idle_owned_gpus,
+            "jobs": jobs,
+            "visible_offers": offers,
+            "actor_signals": signals,
+            "recent_events": events,
+        },
+        indent=2,
+    )
+
+
+def parse_action(text: str) -> tuple[ComputeMarketAction, bool]:
+    match = ACTION_JSON_RE.search(text)
+    if not match:
+        return ComputeMarketAction(action_type="inspect_market"), False
+    try:
+        payload = json.loads(match.group(0))
+        return ComputeMarketAction(**payload), True
+    except Exception:
+        return ComputeMarketAction(action_type="inspect_market"), False
+
+
+def _count_completed_jobs(observation: Any) -> int:
+    return sum(1 for job in observation.jobs if job.status == "completed")
+
+
+def rollout_once(
+    trainer: Any,
+    env: ComputeMarketEnv,
+    tokenizer: Any,
+    dataset_prompt: str,
+    system_prompt: str = SYSTEM_PROMPT,
+    max_turns: int = 6,
+    seed: int | None = None,
+) -> RolloutSummary:
+    from trl.experimental.openenv import generate_rollout_completions
+
+    result = env.reset(seed=seed)
+    prompt_ids: list[int] = []
+    completion_ids: list[int] = []
+    logprobs: list[float] = []
+    transcripts: list[dict[str, Any]] = []
+    rewards: list[float] = []
+    valid_action_reward = 0.0
+    completed_before = 0
+
+    for turn in range(max_turns):
+        if result.done:
+            break
+
+        prompt_text = observation_to_prompt(result.observation, dataset_prompt)
+        messages = [
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": prompt_text},
+        ]
+        rendered_prompt = tokenizer.apply_chat_template(
+            messages,
+            add_generation_prompt=True,
+            tokenize=False,
+            enable_thinking=False,
+        )
+
+        rollout_outputs = generate_rollout_completions(trainer, [rendered_prompt])[0]
+        prompt_ids.extend(rollout_outputs["prompt_ids"])
+        completion_ids.extend(rollout_outputs["completion_ids"])
+        logprobs.extend(rollout_outputs["logprobs"])
+        completion_text = rollout_outputs.get("text") or tokenizer.decode(
+            rollout_outputs["completion_ids"],
+            skip_special_tokens=True,
+        )
+
+        action, is_valid = parse_action(completion_text)
+        result = env.step(action)
+        reward = float(result.reward or 0.0)
+        rewards.append(reward)
+        valid_action_reward += 0.25 if is_valid else -1.0
+
+        completed_after = _count_completed_jobs(result.observation)
+        completion_gain = max(0, completed_after - completed_before)
+        completed_before = completed_after
+
+        transcripts.append(
+            {
+                "turn": turn,
+                "prompt": prompt_text,
+                "completion": completion_text,
+                "action": action.model_dump(exclude_none=True),
+                "is_valid_action": is_valid,
+                "reward": reward,
+                "completed_jobs": completed_after,
+                "completion_gain": completion_gain,
+            }
+        )
+
+    completion_bonus = float(completed_before)
+    return RolloutSummary(
+        prompt_ids=prompt_ids,
+        completion_ids=completion_ids,
+        logprobs=logprobs,
+        episode_return=sum(rewards),
+        valid_action_reward=valid_action_reward,
+        completion_bonus=completion_bonus,
+        transcripts=transcripts,
+    )
+
+
+def rollout_func(
+    prompts: list[str],
+    trainer: Any | None = None,
+    tokenizer: Any | None = None,
+    env_url: str | None = None,
+    max_turns: int = 6,
+    seed_offset: int = 0,
+) -> dict[str, Any]:
+    if trainer is None:
+        raise ValueError("trainer is required")
+    if tokenizer is None:
+        raise ValueError("tokenizer is required")
+
+    env_url = env_url or os.environ.get("OPENENV_URL", "http://localhost:8000")
+    episode_prompt_ids = []
+    episode_completion_ids = []
+    episode_logprobs = []
+    episode_returns = []
+    validity_rewards = []
+    completion_bonuses = []
+    transcripts = []
+
+    with ComputeMarketEnv(base_url=env_url) as env:
+        for idx, prompt_text in enumerate(prompts):
+            episode = rollout_once(
+                trainer=trainer,
+                env=env,
+                tokenizer=tokenizer,
+                dataset_prompt=prompt_text,
+                max_turns=max_turns,
+                seed=seed_offset + idx,
+            )
+            episode_prompt_ids.append(episode.prompt_ids)
+            episode_completion_ids.append(episode.completion_ids)
+            episode_logprobs.append(episode.logprobs)
+            episode_returns.append(episode.episode_return)
+            validity_rewards.append(episode.valid_action_reward)
+            completion_bonuses.append(episode.completion_bonus)
+            transcripts.append(episode.transcripts)
+
+    return {
+        "prompt_ids": episode_prompt_ids,
+        "completion_ids": episode_completion_ids,
+        "logprobs": episode_logprobs,
+        "env_reward": episode_returns,
+        "valid_action_reward": validity_rewards,
+        "completion_bonus": completion_bonuses,
+        "transcripts": transcripts,
+    }
+
+
+def reward_env_return(completions: list[Any], **kwargs: Any) -> list[float]:
+    rewards = kwargs.get("env_reward") or []
+    return [float(rewards[i]) if i < len(rewards) else 0.0 for i in range(len(completions))]
+
+
+def reward_valid_action(completions: list[Any], **kwargs: Any) -> list[float]:
+    rewards = kwargs.get("valid_action_reward") or []
+    return [float(rewards[i]) if i < len(rewards) else 0.0 for i in range(len(completions))]
+
+
+def reward_job_completion(completions: list[Any], **kwargs: Any) -> list[float]:
+    rewards = kwargs.get("completion_bonus") or []
+    return [float(rewards[i]) if i < len(rewards) else 0.0 for i in range(len(completions))]
+
+
+def build_dataset(size: int = 128, prompt: str = DEFAULT_TASK_PROMPT):
+    from datasets import Dataset
+
+    return Dataset.from_dict({"prompt": [prompt] * size})
+
+
+def build_grpo_config(
+    output_dir: str = "outputs/compute-market-qwen3-4b",
+    max_steps: int = 300,
+    learning_rate: float = 5e-6,
+    num_generations: int = 2,
+    max_prompt_length: int = 1800,
+    max_completion_length: int = 192,
+    use_vllm: bool = True,
+):
+    from trl import GRPOConfig
+
+    kwargs: dict[str, Any] = dict(
+        learning_rate=learning_rate,
+        weight_decay=0.001,
+        warmup_ratio=0.1,
+        lr_scheduler_type="linear",
+        optim="adamw_8bit",
+        logging_steps=1,
+        per_device_train_batch_size=1,
+        gradient_accumulation_steps=1,
+        num_generations=num_generations,
+        max_prompt_length=max_prompt_length,
+        max_completion_length=max_completion_length,
+        max_steps=max_steps,
+        save_steps=max_steps,
+        report_to="none",
+        output_dir=output_dir,
+    )
+    if use_vllm:
+        kwargs.update(
+            use_vllm=True,
+            vllm_mode="colocate",
+            vllm_gpu_memory_utilization=0.15,
+        )
+    return GRPOConfig(**kwargs)
+
+
+def build_trainer(
+    model: Any,
+    tokenizer: Any,
+    env_url: str,
+    train_dataset: Any | None = None,
+    output_dir: str = "outputs/compute-market-qwen3-4b",
+    max_steps: int = 300,
+    max_turns: int = 6,
+):
+    from trl import GRPOTrainer
+
+    train_dataset = train_dataset or build_dataset()
+    args = build_grpo_config(output_dir=output_dir, max_steps=max_steps)
+
+    def bound_rollout_func(prompts: list[str], trainer: Any | None = None, **_: Any) -> dict[str, Any]:
+        return rollout_func(
+            prompts,
+            trainer=trainer,
+            tokenizer=tokenizer,
+            env_url=env_url,
+            max_turns=max_turns,
+        )
+
+    return GRPOTrainer(
+        model=model,
+        processing_class=tokenizer,
+        reward_funcs=[
+            reward_env_return,
+            reward_valid_action,
+            reward_job_completion,
+        ],
+        train_dataset=train_dataset,
+        args=args,
+        rollout_func=bound_rollout_func,
+    )
+
+
+def build_colab_setup_snippet(space_repo_id: str = "kyars/compute_market_env") -> str:
+    return f"""# Colab install\n!pip install --upgrade uv\n!uv pip install unsloth vllm --torch-backend=auto\n!uv pip install --upgrade --no-cache-dir --no-deps unsloth unsloth_zoo\n!uv pip install transformers==4.56.2 'trl>=0.24.0' datasets openenv-core\n!pip install git+https://huggingface.co/spaces/{space_repo_id}\n\nimport os\nos.environ['OPENENV_URL'] = 'https://{space_repo_id.replace('/', '-').replace('_', '-')}.hf.space'\n"""
+
+
+if __name__ == "__main__":
+    print(build_colab_setup_snippet())

training/minimal_grpo_rollout.py

@@ -0,0 +1,81 @@
+"""Minimal TRL/OpenEnv rollout helpers for Colab."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from typing import Any
+
+from compute_market_env import ComputeMarketAction, ComputeMarketEnv
+
+ACTION_RE = re.compile(r"\{.*\}", re.DOTALL)
+SYSTEM_PROMPT = """You trade for scarce GPU capacity.
+Choose exactly one JSON action per turn.
+Valid action_type values are: bid_for_capacity, accept_offer, propose_swap, schedule_job, delay_job, inspect_market, noop.
+Return strict JSON only."""
+
+
+def observation_to_prompt(observation) -> str:
+    jobs = [job.model_dump() for job in observation.jobs]
+    offers = [offer.model_dump() for offer in observation.visible_offers]
+    signals = [signal.model_dump() for signal in observation.actor_signals]
+    return json.dumps(
+        {
+            "tick": observation.current_tick,
+            "budget_remaining": observation.budget_remaining,
+            "market_price": observation.market_price,
+            "free_gpus": observation.free_gpus,
+            "owned_gpus": observation.owned_gpus,
+            "idle_owned_gpus": observation.idle_owned_gpus,
+            "jobs": jobs,
+            "visible_offers": offers,
+            "actor_signals": signals,
+        },
+        indent=2,
+    )
+
+
+def parse_action(text: str) -> ComputeMarketAction:
+    match = ACTION_RE.search(text)
+    if not match:
+        return ComputeMarketAction(action_type="inspect_market")
+    try:
+        payload = json.loads(match.group(0))
+        return ComputeMarketAction(**payload)
+    except Exception:
+        return ComputeMarketAction(action_type="inspect_market")
+
+
+def rollout_once(generate_completion, seed: int = 0, max_turns: int = 6) -> dict[str, Any]:
+    env_url = os.environ.get("OPENENV_URL", "http://localhost:8000")
+    rewards = []
+    actions = []
+    transcripts = []
+
+    with ComputeMarketEnv(base_url=env_url) as env:
+        result = env.reset(seed=seed)
+        for _ in range(max_turns):
+            if result.done:
+                break
+            prompt = observation_to_prompt(result.observation)
+            completion = generate_completion(SYSTEM_PROMPT, prompt)
+            action = parse_action(completion)
+            result = env.step(action)
+            actions.append(action.model_dump(exclude_none=True))
+            rewards.append(float(result.reward or 0.0))
+            transcripts.append(
+                {
+                    "prompt": prompt,
+                    "completion": completion,
+                    "action": action.model_dump(exclude_none=True),
+                    "reward": result.reward,
+                }
+            )
+
+    return {
+        "actions": actions,
+        "rewards": rewards,
+        "return": sum(rewards),
+        "transcripts": transcripts,
+    }