feat: uncapped computer agents, stream-safe responses and provider volumes

deepgym/src/deepgym/agents/base.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import math
 import time
 from typing import Literal, Protocol, runtime_checkable
 
@@ -32,8 +33,8 @@ class Budget(BaseModel):
     """
 
     model_config = ConfigDict(frozen=True, allow_inf_nan=False)
-    max_turns: int = Field(default=30, ge=1)
-    max_seconds: float = Field(default=300, gt=0)
+    max_turns: int | None = Field(default=30, ge=1)  # None: no turn limit
+    max_seconds: float | None = Field(default=300, gt=0)  # None: no wall-clock limit
     max_cost_usd: float | None = Field(default=None, ge=0)
 
 
@@ -132,9 +133,16 @@ def __init__(self, budget: Budget) -> None:
 
     @property
     def remaining_seconds(self) -> float:
-        """Return remaining wall time, including retries and tool execution."""
+        """Return remaining wall time, including retries and tools; inf without a limit."""
+        if self.budget.max_seconds is None:
+            return math.inf
         return max(0, self.budget.max_seconds - (time.monotonic() - self.started))
 
+    @property
+    def request_timeout(self) -> float | None:
+        """Deadline for one provider call or action; None without a wall-clock limit."""
+        return None if self.budget.max_seconds is None else self.remaining_seconds
+
     @property
     def reason(self) -> TerminationReason | None:
         """Return an exhausted time or spend bound, without checking turn admission."""
@@ -153,7 +161,7 @@ def check(self) -> None:
     def before_turn(self) -> None:
         """Admit one provider call, raising BudgetExceeded at an exhausted bound."""
         self.check()
-        if self.turns >= self.budget.max_turns:
+        if self.budget.max_turns is not None and self.turns >= self.budget.max_turns:
             raise BudgetExceeded('turn_budget')
         self.turns += 1
 

deepgym/src/deepgym/agents/cli.py

@@ -88,6 +88,10 @@ def _run(self, session: Session, task: str, budget: Budget) -> AgentResult:
             raise ValueError(f'{source} or api_key is required')
         if tracker.remaining_seconds <= 0:
             return tracker.result(session, 'time_budget')
+        if budget.max_seconds is None:
+            raise ValueError(
+                'CLI runs need Budget.max_seconds: the supervisor enforces the CLI deadline'
+            )
         events = CliEvents(self.provider, tracker, self.prices)
         credential = source if self.provider == 'anthropic' else 'CODEX_API_KEY'
         request = CliRequest(

deepgym/src/deepgym/agents/cli_events.py

@@ -147,7 +147,11 @@ def consume(self, event: dict[str, JsonValue]) -> str | None:
             and self.tracker.cost.usd >= self.tracker.budget.max_cost_usd
         ):
             self.stop_reason = 'cost_budget'
-        elif not self.completed and self.tracker.turns >= self.tracker.budget.max_turns:
+        elif (
+            not self.completed
+            and self.tracker.budget.max_turns is not None
+            and self.tracker.turns >= self.tracker.budget.max_turns
+        ):
             self.stop_reason = 'turn_budget'
         return self.stop_reason
 

deepgym/src/deepgym/agents/computer.py

@@ -43,11 +43,27 @@ def _step(session: Session, action: Action, tracker: BudgetTracker) -> Observati
         raise BudgetExceeded('session_done')
     if action.action_type == 'wait' and action.amount > tracker.remaining_seconds:
         raise BudgetExceeded('time_budget')
-    observation = session.step(action, timeout=tracker.remaining_seconds).observation
+    observation = session.step(action, timeout=tracker.request_timeout).observation
     tracker.check()
     return observation
 
 
+def _snapshot(stream: Any) -> Any:
+    """The accumulated message, or None when no message_start arrived (the SDK asserts then)."""
+    try:
+        return stream.current_message_snapshot
+    except AssertionError:
+        return None
+
+
+def _incomplete(tracker: BudgetTracker, snapshot: Any, prices: TokenPrices | None) -> None:
+    """Keep the usage an unfinished stream reported, with the total bill marked unresolved."""
+    data = snapshot.model_dump(exclude_none=True).get('usage') if snapshot is not None else None
+    usage = _anthropic_usage(data) if isinstance(data, dict) else Usage()
+    unresolved = 'incomplete streamed response: partial usage, billing unresolved'
+    tracker.add_usage(usage, Cost(provenance=unresolved))
+
+
 def _account(
     tracker: BudgetTracker,
     data: dict[str, Any] | None,
@@ -103,6 +119,9 @@ def _openai_usage(data: dict[str, Any]) -> Usage:
 # The SDK retries connection drops, 429s and 5xx with backoff; one blip shouldn't end an
 # hours-long episode. A retried request may be billed twice; usage counts only answers.
 _SDK_RETRIES = 4
+ANTHROPIC_MAX_TOKENS = (
+    128_000  # Opus 5.5's maximum output (the API states it); streamed, as the SDK requires
+)
 
 
 def _api_error(exc: Exception) -> str:
@@ -127,17 +146,21 @@ class AnthropicComputerAgent:
         model: Provider model supporting computer_toolset_20260801.
         prices: Explicit model-specific rates, required for a USD budget.
         effort: Adaptive-thinking effort; None sends no thinking or effort settings.
+        max_tokens: Output limit per response; defaults to Opus 5.5's maximum so the
+            harness never caps the model below what it can produce.
     """
 
     def __init__(
         self,
         model: str,
         prices: TokenPrices | None = None,
-        effort: Literal['low', 'medium', 'high'] | None = 'medium',
+        effort: Literal['low', 'medium', 'high', 'xhigh', 'max'] | None = 'max',
+        max_tokens: int = ANTHROPIC_MAX_TOKENS,
     ) -> None:
         self.model = model
         self.prices = prices
         self.effort = effort
+        self.max_tokens = max_tokens
 
     def run(
         self, session: Session, initial: Observation, *, task: str, budget: Budget
@@ -181,9 +204,9 @@ def run(
                     if self.effort
                     else {}
                 )
-                response = client.messages.create(
+                with client.messages.stream(
                     model=self.model,
-                    max_tokens=16000,
+                    max_tokens=self.max_tokens,
                     cache_control={'type': 'ephemeral'},
                     **thinking,
                     system=_SYSTEM,
@@ -194,8 +217,26 @@ def run(
                         }
                     ],
                     messages=messages,
-                    timeout=tracker.remaining_seconds,
-                ).model_dump(exclude_none=True)
+                    timeout=tracker.request_timeout,
+                ) as stream:
+                    stopped, failure = False, None
+                    try:  # consume every event; only a received message_stop completes a response
+                        for event in stream:
+                            stopped = stopped or event.type == 'message_stop'
+                    # Anything raised while reading (httpx or httpx2 transport errors, by SDK
+                    # version, or an error event) ends this response.
+                    except Exception as exc:  # noqa: BLE001
+                        failure = type(exc).__name__
+                    snapshot = _snapshot(stream)
+                if failure or not stopped or snapshot is None:
+                    _incomplete(tracker, snapshot, self.prices)
+                    error = (
+                        f'Response stream failed: {failure}'
+                        if failure
+                        else 'Response stream ended before completion'
+                    )
+                    return tracker.result(session, 'error', error=error)
+                response = snapshot.model_dump(exclude_none=True)
                 _account(tracker, response.get('usage'), self.prices, 'anthropic')
                 content = response['content']
                 messages.append({'role': 'assistant', 'content': content})
@@ -274,11 +315,19 @@ class OpenAIComputerAgent:
     Args:
         model: Provider model supporting the Responses computer tool.
         prices: Explicit model-specific rates, required for a USD budget.
+        effort: Reasoning effort; None sends no reasoning setting. No output limit is sent,
+            so the model may use its full output allowance.
     """
 
-    def __init__(self, model: str, prices: TokenPrices | None = None) -> None:
+    def __init__(
+        self,
+        model: str,
+        prices: TokenPrices | None = None,
+        effort: Literal['none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max'] | None = 'max',
+    ) -> None:
         self.model = model
         self.prices = prices
+        self.effort = effort
 
     def run(
         self, session: Session, initial: Observation, *, task: str, budget: Budget
@@ -316,8 +365,8 @@ def run(
                     tools=[{'type': 'computer'}],
                     input=items,
                     previous_response_id=previous,
-                    max_output_tokens=4096,
-                    timeout=tracker.remaining_seconds,
+                    **({'reasoning': {'effort': self.effort}} if self.effort else {}),
+                    timeout=tracker.request_timeout,
                 ).model_dump(exclude_none=True)
                 _account(tracker, response.get('usage'), self.prices, 'openai')
                 previous = response['id']
@@ -339,17 +388,15 @@ def run(
                         'safety_check',
                         error=('Provider safety checks require approval; no actions executed'),
                     )
-                if not calls:
-                    if response.get('status') == 'incomplete':
-                        items = [{'role': 'user', 'content': 'Continue from the output limit.'}]
-                        continue
-                    if response.get('status') != 'completed':
-                        return tracker.result(session, 'error', error='Response did not complete')
-                    return tracker.result(session, 'completed', text)
                 if response.get('status') != 'completed':
+                    # Never act on, or blindly continue, a partial response; keep its reason.
+                    details = response.get('incomplete_details') or {}
+                    reason = details.get('reason') or response.get('status')
                     return tracker.result(
-                        session, 'error', error='Computer response did not complete'
+                        session, 'error', error=f'Response did not complete: {reason}'
                     )
+                if not calls:
+                    return tracker.result(session, 'completed', text)
                 items = []
                 notes = []
                 for call in calls:

deepgym/src/deepgym/api/schemas.py

@@ -219,6 +219,8 @@ def reject_runtime(
             raise ValueError('root_service_start cannot be set over the API')
         if not isinstance(v, str) and v.cli_provider_egress:
             raise ValueError('cli_provider_egress cannot be set over the API')
+        if not isinstance(v, str) and v.volumes:
+            raise ValueError('volumes cannot be set over the API')
         return v
 
 

deepgym/src/deepgym/computer_use.py

@@ -26,5 +26,5 @@ class ComputerUseEnvironment(MultiTurnEnvironment):
     allowed_actions: frozenset[ActionType] = frozenset({*GUI_ACTIONS, 'done'})
     """GUI only, so the agent can't skip the interface with a shell."""
 
-    max_steps: int = Field(default=30, ge=1)
+    max_steps: int | None = Field(default=30, ge=1)  # None: no step limit
     """GUI tasks take more steps than shell tasks."""

deepgym/src/deepgym/models.py

@@ -212,7 +212,7 @@ class EnvMetadata(BaseModel):
     image: str | None = None
     snapshot: str | None = None
     """Multi-turn and computer-use envs: the Daytona snapshot their sessions boot."""
-    max_steps: int = Field(default=10, ge=1)
+    max_steps: int | None = Field(default=10, ge=1)  # None: no step limit
     timeout_per_step: int = Field(default=30, ge=1)
     repo_dir: str | None = None
     """Image-based repo patch envs (test_patch.diff): where the repo lives in the image."""
@@ -491,6 +491,15 @@ class Trajectory(BaseModel):
     without one, so a failed episode has one reward fewer than steps."""
 
 
+class VolumeMount(BaseModel):
+    """One provider volume mounted at an absolute path inside the sandbox."""
+
+    model_config = ConfigDict(frozen=True, extra='forbid')
+    volume_id: str = Field(min_length=1)
+    mount_path: str = Field(pattern=r'^/')
+    subpath: str | None = None
+
+
 class MultiTurnEnvironment(BaseModel):
     """Define a multi-turn environment where agents take multiple steps.
 
@@ -516,6 +525,13 @@ class MultiTurnEnvironment(BaseModel):
     The command must start the service itself and exit non-zero if readiness fails.
     """
 
+    volumes: tuple[VolumeMount, ...] = ()
+    """Provider volumes mounted into the sandbox (Daytona), for evidence too large for the image,
+    such as attachment files. Trusted environment configuration only; never accept from an agent.
+    A volume is shared and outlives each sandbox, so whatever one episode writes, the next reads:
+    mount it where untrusted processes cannot write (e.g. a root-only path re-exposed through a
+    read-only bind mount in root_service_start) and check its contents against a pinned manifest."""
+
     cli_provider_egress: bool = Field(default=False, strict=True)
     """Opt in to sandbox-level egress for the trusted CLI provider proxy (Daytona).
     Agent commands, root service startup and verification remain in the offline namespace.
@@ -531,7 +547,7 @@ class MultiTurnEnvironment(BaseModel):
     final_verifier_code: str = ''
     """Verifier run at the end to compute final score."""
 
-    max_steps: int = Field(default=10, ge=1)
+    max_steps: int | None = Field(default=10, ge=1)  # None: no step limit
     """Maximum number of steps before forced termination."""
 
     timeout_per_step: int = Field(default=30, ge=1)

deepgym/src/deepgym/qa/agents.py

@@ -28,14 +28,22 @@
 def do_everything(env: MultiTurnEnvironment) -> list[Action]:
     """Every allowed action type in turn, with fixed arguments, for the whole step budget."""
     kinds = _kinds(env)
-    return [_action(kinds[step % len(kinds)], step, None) for step in range(env.max_steps)]
+    return [
+        _action(kinds[step % len(kinds)], step, None) for step in range(_control_steps(env, kinds))
+    ]
 
 
 def random_agent(env: MultiTurnEnvironment) -> list[Action]:
     """Seeded random actions of the allowed types, for the whole step budget."""
     rng = random.Random(_SEED)
     kinds = _kinds(env)
-    return [_action(rng.choice(kinds), step, rng) for step in range(env.max_steps)]
+    return [_action(rng.choice(kinds), step, rng) for step in range(_control_steps(env, kinds))]
+
+
+def _control_steps(env: MultiTurnEnvironment, kinds: list[ActionType]) -> int:
+    """Length of a scripted QA control: the env's step budget, or one round of every allowed
+    action type when the env sets no step limit (fixed controls must end; never a model run)."""
+    return env.max_steps if env.max_steps is not None else len(kinds)
 
 
 def _kinds(env: MultiTurnEnvironment) -> list[ActionType]:

deepgym/src/deepgym/sandboxes/daytona/sandbox.py

@@ -88,6 +88,11 @@ def _session_params(
     """
     sdk = _sdk()
     options = {'ephemeral': True, 'network_block_all': not env.cli_provider_egress}
+    if env.volumes:
+        options['volumes'] = [
+            sdk.VolumeMount(volume_id=v.volume_id, mount_path=v.mount_path, subpath=v.subpath)
+            for v in env.volumes
+        ]
     if env.image and not env.snapshot:
         return sdk.CreateSandboxFromImageParams(image=sdk.Image.base(env.image), **options)
     return sdk.CreateSandboxFromSnapshotParams(snapshot=env.snapshot, **options)

deepgym/src/deepgym/session.py

@@ -1086,7 +1086,7 @@ def step(self, action: Action, *, timeout: float | None = None) -> StepResult:
             done = (
                 action.action_type == 'done'
                 or _signals_done(output)
-                or self._step >= self._env.max_steps
+                or (self._env.max_steps is not None and self._step >= self._env.max_steps)
             )
             observation: Observation | None = None
             try: