Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@ const STEP_COLORS: Record<string, string> = {
register_env_triggers: '#0d9488',
register_agent_triggers: '#c026d3',
sync_env_clock: '#0891b2',
set_env_clock_rate: '#0891b2',
};

const DEFAULT_COLOR = '#6b7280';
Expand Down
3 changes: 2 additions & 1 deletion src/agent_env/task/episode/rules.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,8 @@ def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[s
"browser_set_local_storage", "build_mcp_cli", "combine_universe_verdicts", "configure_scheduled_checks",
"deploy_env", "deploy_sandbox", "end_computer_recording", "start_computer_recording", "env_outcome_verifier",
"modify_env_tool_access", "reset_env", "review",
"run_container_unit_tests_verifier", "run_docker_container", "snapshot_env", "snapshot_computer_env", "sync_env_clock",
"run_container_unit_tests_verifier", "run_docker_container", "set_env_clock_rate", "snapshot_env", "snapshot_computer_env",
"sync_env_clock",
"validation_gate_aggregator", "verify_env_card", "verify_env_core_protocol", "verify_mcp_tool_schema",
"verify_spec_conformance", "verify_universe_load_export_roundtrip",
# They read only what an outside turn leaves: its reply.
Expand Down
2 changes: 2 additions & 0 deletions src/agent_env/task_step/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]:
from agent_env.task_step.task_steps.register_env_triggers import RegisterEnvTriggersStep
from agent_env.task_step.task_steps.register_agent_triggers import RegisterAgentTriggersStep
from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep
from agent_env.task_step.task_steps.set_env_clock_rate import SetEnvClockRateTaskStep
from agent_env.task_step.task_steps.configure_scheduled_checks import ConfigureScheduledChecksTaskStep
from agent_env.task_step.task_steps.install_agent import InstallAgentTaskStep
from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep
Expand Down Expand Up @@ -124,6 +125,7 @@ def _builtin_registry() -> dict[str, type["TaskStep"]]:
_get_type(RegisterEnvTriggersStep): RegisterEnvTriggersStep,
_get_type(RegisterAgentTriggersStep): RegisterAgentTriggersStep,
_get_type(SyncEnvClockTaskStep): SyncEnvClockTaskStep,
_get_type(SetEnvClockRateTaskStep): SetEnvClockRateTaskStep,
_get_type(ConfigureScheduledChecksTaskStep): ConfigureScheduledChecksTaskStep,
_get_type(InstallAgentTaskStep): InstallAgentTaskStep,
_get_type(RubricsVerifierTaskStep): RubricsVerifierTaskStep,
Expand Down
9 changes: 4 additions & 5 deletions src/agent_env/task_step/task_steps/deploy_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,24 +114,23 @@ def _deploy_step_of(deployed_env: DeployedEnv) -> Optional[str]:
def bound_deployment(
context: TaskStepContext, env_id: str, *, env_step_id: Optional[str], step_id: str,
) -> Optional[DeployedEnv]:
"""The deployment of `env_id` that the agent deployed by step `step_id` is bound to, by `env_step_id`, or None if
it isn't deployed here."""
"""The deployment of `env_id` that step `step_id` acts on, by `env_step_id`, or None if it isn't deployed here."""
matches = [d for d in context.deployed_envs if d.env_id == env_id]
if env_step_id is not None:
# A filter, not a hint: an unmatched id is an error, never a substitute.
# Envs with no deployments fall through to the caller's live-endpoint path.
named = [d for d in matches if _deploy_step_of(d) == env_step_id]
if matches and not named:
raise RuntimeError(
f"deploy_agent '{step_id}': env_step_id '{env_step_id}' names no deployment of "
f"step '{step_id}': env_step_id '{env_step_id}' names no deployment of "
f"env '{env_id}' (deployed by: {[_deploy_step_of(d) for d in matches]})"
)
matches = named
if len(matches) > 1:
raise RuntimeError(
f"deploy_agent '{step_id}': {len(matches)} deployments of env '{env_id}' "
f"step '{step_id}': {len(matches)} deployments of env '{env_id}' "
f"(deployed by: {[_deploy_step_of(d) for d in matches]}). Set env_step_id "
"to the deploy_env step this agent should bind to."
"to the deploy_env step this step should act on."
)
return matches[0] if matches else None

Expand Down
95 changes: 95 additions & 0 deletions src/agent_env/task_step/task_steps/set_env_clock_rate.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
"""Change an armed env gateway clock's rate in place (clock/v1 set_rate), keeping its time and its time triggers' marks.

A rate of 0 stops the clock, and with it every time trigger and virtual wait. Arm the clock at rate 0 during setup and
set the task's rate right before prompt_agent, so the agent starts at the armed time; set 0 after the agent's last
prompt_agent, so verifiers read a world that no longer moves on its clock. ``env_step_id`` names the deploy_env step
whose deployment's clock to change when the env is deployed more than once.
"""

from __future__ import annotations

import logging
from typing import ClassVar, Optional

import httpx

from agent_env.entity.refs import EntityRef
from agent_env.env.gateway.constants import EXT_CLOCK_URI
from agent_env.task_step.context import StepContextError, TaskStepContext
from agent_env.task_step.task_step import TaskStep, TaskStepDependency
from agent_env.task_step.task_steps.deploy_agent import bound_deployment

logger = logging.getLogger(__name__)

_MAX_ERROR_CHARS = 300


def _gateway_error(response: httpx.Response) -> str:
"""The error a refused clock write names: the gateway's ``error`` field, else its body, cut to _MAX_ERROR_CHARS."""
try:
error = response.json().get("error")
except (ValueError, AttributeError):
error = None
return str(error if error is not None else response.text)[:_MAX_ERROR_CHARS]


class SetEnvClockRateTaskStep(TaskStep):
type: ClassVar[str] = "set_env_clock_rate"
entity_refs = (EntityRef.env("env_id"),)

def __init__(
self,
id: str,
version: Optional[int],
env_id: str,
virtual_seconds_per_real_second: float,
env_step_id: Optional[str] = None,
depends_on: Optional[list[TaskStepDependency]] = None,
fail_task_on_error: bool = True,
timeout_seconds: int = 30,
):
super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
rate = virtual_seconds_per_real_second
if isinstance(rate, bool) or not isinstance(rate, (int, float)) or rate < 0:
raise ValueError(f"virtual_seconds_per_real_second must be a non-negative number, got {rate!r}")
self.env_id = env_id
self.virtual_seconds_per_real_second = rate
self.env_step_id = env_step_id
self.timeout_seconds = timeout_seconds

def to_dict(self) -> dict:
base = super().to_dict()
base["env_id"] = self.env_id
base["virtual_seconds_per_real_second"] = self.virtual_seconds_per_real_second
base["env_step_id"] = self.env_step_id
base["timeout_seconds"] = self.timeout_seconds
return base

@classmethod
def from_dict(cls, data: dict) -> SetEnvClockRateTaskStep:
return cls(
**cls._base_from_dict(data),
env_id=data["env_id"],
virtual_seconds_per_real_second=data["virtual_seconds_per_real_second"],
env_step_id=data.get("env_step_id"),
timeout_seconds=data.get("timeout_seconds", 30),
)

async def execute(self, context: TaskStepContext) -> TaskStepContext:
deployed = bound_deployment(context, self.env_id, env_step_id=self.env_step_id, step_id=self.id)
if deployed is None:
raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
try:
state = await deployed.invoke(EXT_CLOCK_URI, "set_rate",
{"virtual_seconds_per_real_second": self.virtual_seconds_per_real_second},
timeout=self.timeout_seconds)
Comment thread
andrewpark-scaleai marked this conversation as resolved.
Resolved
except httpx.HTTPStatusError as e:
raise RuntimeError(f"clock rate change failed (HTTP {e.response.status_code}): {_gateway_error(e.response)}") from e
context.metadata.setdefault("clock_configurations", []).append({
"step_id": self.id, "env_id": self.env_id,
"virtual_seconds_per_real_second": self.virtual_seconds_per_real_second,
"virtual_time": state.get("virtual_time"),
})
logger.info(f"set_env_clock_rate env={self.env_id}: rate={self.virtual_seconds_per_real_second} "
f"at {state.get('virtual_time')}")
return context
3 changes: 2 additions & 1 deletion src/agent_env/task_step/task_steps/sync_env_clock.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@
stored env card declares clock/v1, then invokes sync_time on each backing server, MCP server or website, the
env card lists (``legacy_protocol.backing_servers``) whose own card advertises it, at the route that card gives.
Non-advertisers are skipped when tolerate_missing_sync_time (default True), and websites always. Place right
before prompt_agent so the virtual timeline starts at agent-start.
before prompt_agent so the virtual timeline starts at agent-start; when setup needs the clock armed, arm it at rate 0
and start it with set_env_clock_rate right before prompt_agent.
"""

from __future__ import annotations
Expand Down
40 changes: 40 additions & 0 deletions tst/unit/env/gateway/triggers_time_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -422,6 +422,46 @@ async def test_unarmed_clock_defers_then_fires_once_armed():
assert engine._triggers["abs"]["fire_count"] == 1


@pytest.mark.asyncio
async def test_a_stopped_clock_holds_a_relative_trigger_and_a_restart_keeps_its_mark():
engine, gw, clock, now = _make()
clock.set_time(T0, rate=1.0)
engine.register({"watch_roles": ["default"], "triggers": [
{"id": "rel", "when": {"type": "time", "at": "PT10M"}, "actions": [_perm()]}]})
_tick_call(engine, now, 300)
clock.set_rate(0)
_tick_call(engine, now, 86_400)
await _settle(engine)
assert engine._triggers["rel"]["fire_count"] == 0
clock.set_rate(1.0)
_tick_call(engine, now, 299)
await _settle(engine)
assert engine._triggers["rel"]["fire_count"] == 0
_tick_call(engine, now, 2)
await _settle(engine)
assert engine._triggers["rel"]["fire_count"] == 1


@pytest.mark.asyncio
async def test_stopping_the_clock_fires_only_the_marks_it_already_passed():
engine, gw, clock, now = _make()
clock.set_time(T0, rate=1.0)
engine.register({"watch_roles": ["default"], "triggers": [
{"id": "rec", "when": {"type": "time", "every": "PT1M"}, "actions": [_perm()]},
{"id": "late", "when": {"type": "time", "at": "PT10M"}, "actions": [_perm()]}]})
_tick_call(engine, now, 1)
await _settle(engine)
now.tick(210)
clock.set_rate(0)
engine.clock_advanced()
await _settle(engine)
assert engine._triggers["rec"]["fire_count"] == 3
for _ in range(24):
_tick_call(engine, now, 3600)
await _settle(engine)
assert (engine._triggers["rec"]["fire_count"], engine._triggers["late"]["fire_count"]) == (3, 0)


@pytest.mark.asyncio
async def test_time_advance_alone_is_inert_without_driver_or_calls():
"""No hidden timer: with no watched call and no driver, advancing virtual time fires nothing."""
Expand Down
168 changes: 168 additions & 0 deletions tst/unit/task_step/set_env_clock_rate_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,168 @@
"""SetEnvClockRateTaskStep: changes an armed env clock's rate in place, so a task can hold the clock through setup and
start it with the agent, or stop it when the agent stops."""
import httpx
import pytest

from agent_env.env.gateway.clock import Clock, ClockError
from agent_env.env.gateway.constants import EXT_CLOCK_URI
from agent_env.task_step.context import StepContextError, TaskStepContext
from agent_env.task_step.registry import get_task_step_registry
from agent_env.task_step.task_steps.set_env_clock_rate import SetEnvClockRateTaskStep

T0 = "2026-07-01T22:00:00Z"


class FakeNow:
def __init__(self):
self.t = 1000.0

def __call__(self):
return self.t


class _GatewayEnv:
"""A deployed env whose clock/v1 calls reach a real gateway Clock, answering a refused write as the gateway does."""

def __init__(self, clock: Clock, env_id: str = "e1", deploy_step_id: str = "deploy-e1"):
self.env_id = env_id
self.metadata = {"deploy_step_id": deploy_step_id}
self.clock = clock
self.calls: list[tuple[str, dict]] = []

async def invoke(self, uri, method, params=None, timeout=None):
assert uri == EXT_CLOCK_URI
self.calls.append((method, params))
try:
if method == "set_rate":
return self.clock.set_rate(params["virtual_seconds_per_real_second"])
if method == "state":
return self.clock.state()
except ClockError as e:
request = httpx.Request("PUT", "http://gw/clock/set-rate")
raise httpx.HTTPStatusError("400", request=request, response=httpx.Response(400, text=str(e), request=request))
raise AssertionError(method)


class _RefusingEnv:
"""A deployed env whose gateway refuses every clock/v1 write with ``response``."""

env_id = "e1"
metadata = {"deploy_step_id": "deploy-e1"}

def __init__(self, response: httpx.Response):
self.response = response

async def invoke(self, uri, method, params=None, timeout=None):
request = httpx.Request("PUT", "http://gw/clock/set-rate")
self.response.request = request
raise httpx.HTTPStatusError("refused", request=request, response=self.response)


def _step(rate, **kw):
return SetEnvClockRateTaskStep(id="clock-rate", version=1, env_id="e1", virtual_seconds_per_real_second=rate,
depends_on=[], **kw)


def _armed(rate):
now = FakeNow()
clock = Clock(now_fn=now)
clock.set_time(T0, rate=rate)
return clock, now


def test_round_trip_and_registered():
d = _step(0, timeout_seconds=12).to_dict()
s = SetEnvClockRateTaskStep.from_dict(d)
assert d["type"] == "set_env_clock_rate"
assert (s.env_id, s.virtual_seconds_per_real_second, s.timeout_seconds) == ("e1", 0, 12)
assert get_task_step_registry()["set_env_clock_rate"] is SetEnvClockRateTaskStep


@pytest.mark.asyncio
async def test_a_clock_held_through_setup_starts_at_its_armed_time():
clock, now = _armed(rate=0)
now.t += 600
await _step(1.0).execute(TaskStepContext(deployed_envs=[_GatewayEnv(clock)]))
assert clock.read() == {"virtual_time": T0}
now.t += 60
assert clock.read() == {"virtual_time": "2026-07-01T22:01:00Z"}


@pytest.mark.asyncio
async def test_stopping_the_clock_holds_virtual_time_and_records_it():
clock, now = _armed(rate=1.0)
now.t += 90
context = await _step(0).execute(TaskStepContext(deployed_envs=[_GatewayEnv(clock)]))
now.t += 3600
assert clock.read() == {"virtual_time": "2026-07-01T22:01:30Z"}
assert context.metadata["clock_configurations"] == [{
"step_id": "clock-rate", "env_id": "e1", "virtual_seconds_per_real_second": 0,
"virtual_time": "2026-07-01T22:01:30Z"}]


@pytest.mark.asyncio
async def test_an_unarmed_clock_fails_the_step():
with pytest.raises(RuntimeError, match="not armed"):
await _step(0).execute(TaskStepContext(deployed_envs=[_GatewayEnv(Clock(now_fn=FakeNow()))]))


@pytest.mark.asyncio
async def test_an_env_not_deployed_fails_the_step():
with pytest.raises(StepContextError, match="e1"):
await _step(0).execute(TaskStepContext(deployed_envs=[_GatewayEnv(Clock(), env_id="other")]))


def test_a_negative_rate_is_refused_at_load():
with pytest.raises(ValueError, match="non-negative"):
_step(-1)


@pytest.mark.asyncio
async def test_env_step_id_stops_the_clock_of_the_deployment_it_names():
first, first_now = _armed(rate=1.0)
second, second_now = _armed(rate=1.0)
envs = [_GatewayEnv(first, deploy_step_id="deploy-a"), _GatewayEnv(second, deploy_step_id="deploy-b")]
await _step(0, env_step_id="deploy-b").execute(TaskStepContext(deployed_envs=envs))
first_now.t += 60
second_now.t += 60
assert first.read() == {"virtual_time": "2026-07-01T22:01:00Z"}
assert second.read() == {"virtual_time": T0}


@pytest.mark.asyncio
async def test_two_deployments_of_the_env_without_env_step_id_fail_the_step():
envs = [_GatewayEnv(_armed(1.0)[0], deploy_step_id="deploy-a"), _GatewayEnv(_armed(1.0)[0], deploy_step_id="deploy-b")]
with pytest.raises(RuntimeError, match="2 deployments of env 'e1'"):
await _step(0).execute(TaskStepContext(deployed_envs=envs))
assert all(env.calls == [] for env in envs)


@pytest.mark.asyncio
async def test_an_env_step_id_naming_no_deployment_fails_the_step():
env = _GatewayEnv(_armed(1.0)[0], deploy_step_id="deploy-a")
with pytest.raises(RuntimeError, match="names no deployment"):
await _step(0, env_step_id="deploy-z").execute(TaskStepContext(deployed_envs=[env]))
assert env.calls == []


def test_env_step_id_round_trips():
assert SetEnvClockRateTaskStep.from_dict(_step(0, env_step_id="deploy-b").to_dict()).env_step_id == "deploy-b"


@pytest.mark.asyncio
async def test_a_refused_rate_change_reports_the_gateway_error_bounded():
body = {"ok": False, "error": "clock is not armed " + "x" * 5000, "detail": "y" * 5000}
with pytest.raises(RuntimeError) as raised:
await _step(0).execute(TaskStepContext(deployed_envs=[_RefusingEnv(httpx.Response(400, json=body))]))
message = str(raised.value)
assert message.startswith("clock rate change failed (HTTP 400): clock is not armed")
assert "y" not in message and len(message) < 400


@pytest.mark.asyncio
async def test_a_non_json_refusal_is_reported_bounded():
with pytest.raises(RuntimeError) as raised:
await _step(0).execute(TaskStepContext(deployed_envs=[_RefusingEnv(httpx.Response(502, text="<html>" + "z" * 5000))]))
message = str(raised.value)
assert message.startswith("clock rate change failed (HTTP 502): <html>")
assert len(message) < 400
Loading