diff --git a/src/agent_env/explorer/ui/src/components/steps-pipeline.tsx b/src/agent_env/explorer/ui/src/components/steps-pipeline.tsx index d8917f12f..d8da119c5 100644 --- a/src/agent_env/explorer/ui/src/components/steps-pipeline.tsx +++ b/src/agent_env/explorer/ui/src/components/steps-pipeline.tsx @@ -31,6 +31,7 @@ const STEP_COLORS: Record = { register_env_triggers: '#0d9488', register_agent_triggers: '#c026d3', sync_env_clock: '#0891b2', + set_env_clock_rate: '#0891b2', }; const DEFAULT_COLOR = '#6b7280'; diff --git a/src/agent_env/task/episode/rules.py b/src/agent_env/task/episode/rules.py index 0bfb1d51d..5bb3f48b2 100644 --- a/src/agent_env/task/episode/rules.py +++ b/src/agent_env/task/episode/rules.py @@ -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. diff --git a/src/agent_env/task_step/registry.py b/src/agent_env/task_step/registry.py index d428aeb41..9c08c425a 100644 --- a/src/agent_env/task_step/registry.py +++ b/src/agent_env/task_step/registry.py @@ -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 @@ -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, diff --git a/src/agent_env/task_step/task_steps/deploy_agent.py b/src/agent_env/task_step/task_steps/deploy_agent.py index c95b198ff..65dea69f0 100644 --- a/src/agent_env/task_step/task_steps/deploy_agent.py +++ b/src/agent_env/task_step/task_steps/deploy_agent.py @@ -114,8 +114,7 @@ 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. @@ -123,15 +122,15 @@ def bound_deployment( 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 diff --git a/src/agent_env/task_step/task_steps/set_env_clock_rate.py b/src/agent_env/task_step/task_steps/set_env_clock_rate.py new file mode 100644 index 000000000..c969ebbfa --- /dev/null +++ b/src/agent_env/task_step/task_steps/set_env_clock_rate.py @@ -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) + 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 diff --git a/src/agent_env/task_step/task_steps/sync_env_clock.py b/src/agent_env/task_step/task_steps/sync_env_clock.py index b4c92793e..a87eeb173 100644 --- a/src/agent_env/task_step/task_steps/sync_env_clock.py +++ b/src/agent_env/task_step/task_steps/sync_env_clock.py @@ -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 diff --git a/tst/unit/env/gateway/triggers_time_test.py b/tst/unit/env/gateway/triggers_time_test.py index b08ebbbbe..1cb26034f 100644 --- a/tst/unit/env/gateway/triggers_time_test.py +++ b/tst/unit/env/gateway/triggers_time_test.py @@ -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.""" diff --git a/tst/unit/task_step/set_env_clock_rate_test.py b/tst/unit/task_step/set_env_clock_rate_test.py new file mode 100644 index 000000000..5fe391504 --- /dev/null +++ b/tst/unit/task_step/set_env_clock_rate_test.py @@ -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="" + "z" * 5000))])) + message = str(raised.value) + assert message.startswith("clock rate change failed (HTTP 502): ") + assert len(message) < 400