From 6527793d0bf5d5fc8d6d4c11030ba0b5580085f0 Mon Sep 17 00:00:00 2001 From: William Loo Date: Thu, 12 Nov 2026 09:33:46 -0800 Subject: [PATCH] feat(computers): a computer that follows the clock follows it again once a multi env's reset replaces its container A reset boots the computer afresh, and its clock follower went with the old container, so it showed real time until the next arm. While the gateway's clock is armed, the multi env's reset now starts each such computer's follower again as an arm does, as root, and waits until the computer shows the clock; a clock not armed is left to the next sync_env_clock. The follower start is shared with sync_env_clock, and the end-to-end test resets the env and checks the new container agrees with the gateway. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/computers/clock/README.md | 9 +++-- src/agent_env/env/envs/_multi_computers.py | 34 +++++++++++++++- .../env/envs/computer/utils/clock.py | 23 +++++++++++ .../task_step/task_steps/sync_env_clock.py | 15 +++---- .../computer/clock_follower_test.py | 17 ++++++-- tst/unit/env/envs/test_computer_clock_core.py | 22 +++++++++++ tst/unit/env/envs/test_multi_env_computers.py | 39 +++++++++++++++++++ 7 files changed, 141 insertions(+), 18 deletions(-) diff --git a/src/agent_env/computers/clock/README.md b/src/agent_env/computers/clock/README.md index cb16cd664..691359077 100644 --- a/src/agent_env/computers/clock/README.md +++ b/src/agent_env/computers/clock/README.md @@ -69,9 +69,12 @@ records what it showed in `/run/agentenv-clock/status.json`. A failed round is r and a new config starts it over from no version. The session user can't write the page or the clock file, read the follower's config, or stop either process: each is -another user's. Not covered yet: a computer reset (`reset_env`) boots it afresh with no follower until the next arm; -agent-env's own servers run on universe time too, with no real-time marker; and routed sites' certificates, Chromium's -time zone and an egress proxy for live sites at other dates are later work. +another user's. A multi env's reset (`reset_env`) boots a computer afresh, follower and all; while the clock is armed, +core starts the new follower as an arm does and the reset ends once the computer shows the clock. + +Not covered yet: agent-env's own servers run on universe time too, with no real-time marker; and certificates checked +by TLS stacks the shim doesn't reach (Chromium's), Chromium's time zone and an egress proxy for live sites at other dates +are later work. ## The clock page diff --git a/src/agent_env/env/envs/_multi_computers.py b/src/agent_env/env/envs/_multi_computers.py index bebe171ab..773301030 100644 --- a/src/agent_env/env/envs/_multi_computers.py +++ b/src/agent_env/env/envs/_multi_computers.py @@ -13,9 +13,10 @@ from agentenv_protocol.computer import tool_prefix_problem -from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env +from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env, gateway_url_of from agent_env.env.envs.computer.utils import clock, sites from agent_env.env.gateway import ExternalChild +from agent_env.env.gateway.constants import EXT_CLOCK_URI from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_env_sandbox_provider if TYPE_CHECKING: @@ -100,6 +101,30 @@ def clock_read_token(record: DeployedEnv) -> str | None: return record.control_token if clock.follows_clock(record.environment_card) else None +async def start_clock_follower(deployed: DeployedEnv, name: str, record: DeployedEnv, sandbox: Any) -> Optional[str]: + """Start the clock follower of ``deployed``'s computer ``name``, its record ``record`` and its sandbox ``sandbox``, + watching the gateway's clock as the sandbox reaches it (``clock.start_follower``): the config id it was given, or + None for a computer that runs none.""" + if not clock.runs_follower(record.environment_card) or (token := clock_read_token(record)) is None: + return None + watch_url = sandbox.url_from_sandbox(f"{(gateway_url_of(deployed) or '').rstrip('/')}/clock/watch") + return await clock.start_follower(sandbox, watch_url=watch_url, token=token, follower=name) + + +async def follow_clock_again(deployed: DeployedEnv, name: str, record: DeployedEnv, computer: ComputerEnv) -> None: + """After ``deployed``'s computer ``name`` reset, its clock follower, gone with its container, started again while + the gateway's clock is armed, and the reset done once the computer shows it; nothing for a computer that runs + none, or a clock not armed, which the next ``sync_env_clock`` starts it for.""" + if not clock.runs_follower(record.environment_card) or not gateway_url_of(deployed): + return + if not (await deployed.invoke(EXT_CLOCK_URI, "state")).get("armed"): + return + sandbox = (await type(computer).from_deployed_env(record))._sandbox + if (config_id := await start_clock_follower(deployed, name, record, sandbox)) is not None: + await clock.wait_shown(sandbox, config_id) + logger.info(f"computer {name!r} follows the clock again after its reset") + + def refusal(computer: ComputerEnv, entry: dict, sandbox_type: str | None, websites: Mapping[str, str]) -> str | None: """Why ``computer`` wouldn't deploy in a multi env with ``entry``, its child_envs entry: why it wouldn't alone, ``site_routing`` naming none of the multi env's ``websites`` (their environment names by env id), or a sandbox @@ -179,9 +204,14 @@ def resets(self, deployed: DeployedEnv) -> list[tuple[str, Awaitable[None]]]: record; none is awaited yet.""" records = getattr(deployed, "child_envs", None) or {} browser = self.browser or self._recorded_browser(records) - return [(env.environment_name, env.reset(DeployedEnv.from_dict(records[env.environment_name]))) + return [(env.environment_name, self._reset(deployed, env, DeployedEnv.from_dict(records[env.environment_name]))) for env in [*self.envs, *([browser] if browser else [])] if env.environment_name in records] + @staticmethod + async def _reset(deployed: DeployedEnv, env: ComputerEnv, record: DeployedEnv) -> None: + await env.reset(record) + await follow_clock_again(deployed, env.environment_name, record, env) + def _recorded_browser(self, records: Mapping[str, dict]) -> Optional[ComputerEnv]: """The website browser ``records``, a multi env's computers' records, hold beside its own computers: the record of a name none of them has, its env at the version it deployed.""" diff --git a/src/agent_env/env/envs/computer/utils/clock.py b/src/agent_env/env/envs/computer/utils/clock.py index 9c122e835..f709025e0 100644 --- a/src/agent_env/env/envs/computer/utils/clock.py +++ b/src/agent_env/env/envs/computer/utils/clock.py @@ -8,8 +8,10 @@ clock, and sets it waiting for their acknowledgements.""" from __future__ import annotations +import asyncio import json import secrets +import time from pathlib import PurePosixPath from typing import TYPE_CHECKING @@ -27,11 +29,15 @@ CONFIG = CONFIG_FOLDER / "follower.json" #: The clock file's folder, where the follower also keeps its status. RUN_FOLDER = PurePosixPath("/run/agentenv-clock") +#: What the follower last showed, which anyone may read. +STATUS = RUN_FOLDER / "status.json" CLOCK = PurePosixPath("/opt/agentenv/clock") LOG = PurePosixPath("/var/log/agentenv/clock-follower.log") #: The ``sync_time`` param that carries the id of the config core wrote, which the clock app waits for its follower #: to show. CONFIG_PARAM = "follower_config" +#: How long a write of the clock, or a computer's follower started again, waits for the computer to show it. +FOLLOWER_WAIT_S = 10.0 def follows_clock(card: dict | None) -> bool: @@ -74,3 +80,20 @@ async def start_follower(sandbox: Sandbox, *, watch_url: str, token: str, follow # Detached from the exec, which returns at once; ``follow`` replaces a follower an earlier arm started. await channel.run(["sh", "-c", f"setsid {CLOCK / 'follow'} >{LOG} 2>&1 &"]) return config_id + + +async def wait_shown(sandbox: Sandbox, config_id: str, timeout_s: float = FOLLOWER_WAIT_S) -> dict: + """What the computer's follower showed once it has shown a version of the clock for ``config_id``, read as root; + TimeoutError naming its last status when it hasn't within ``timeout_s``.""" + channel, deadline, status = RootChannel(sandbox), time.monotonic() + timeout_s, None + while True: + try: + status = json.loads(await channel.run(["sh", "-c", f"cat {STATUS} 2>/dev/null || true"])) + except json.JSONDecodeError: + status = None + if status and status.get("config_id") == config_id and status.get("version") and not status.get("error"): + return status + if time.monotonic() >= deadline: + raise TimeoutError(f"the clock follower didn't show config {config_id!r} within {timeout_s:g} s; its " + f"status: {status}") + await asyncio.sleep(0.2) 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 db1fe0abb..b27c4861b 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 @@ -18,8 +18,8 @@ import tzdata # noqa: F401 from agentenv_protocol import client as protocol_v1 -from agent_env.env.env import DeployedEnv, gateway_url_of -from agent_env.env.envs._multi_computers import clock_read_token +from agent_env.env.env import DeployedEnv +from agent_env.env.envs._multi_computers import start_clock_follower from agent_env.env.envs.computer.env import ComputerEnv from agent_env.env.envs.computer.utils import clock from agent_env.env.gateway.constants import EXT_CLOCK_URI @@ -33,7 +33,7 @@ logger = logging.getLogger(__name__) #: How long the arm waits for the computers that follow the clock to show it; a computer still behind fails the step. -FOLLOWER_WAIT_S = 10.0 +FOLLOWER_WAIT_S = clock.FOLLOWER_WAIT_S def resolve_virtual_time(virtual_time: str, timezone: Optional[str]) -> str: @@ -173,17 +173,14 @@ async def _start_followers(self, deployed) -> tuple[dict[str, str], dict[str, st (``computer.utils.clock``): the config id each was given, by the computer's address, and why each that couldn't start didn't, which fails the step unless ``tolerate_missing_sync_time``.""" started, unstarted = {}, {} - gateway = (gateway_url_of(deployed) or "").rstrip("/") for name, data in (getattr(deployed, "child_envs", None) or {}).items(): record = DeployedEnv.from_dict(data) - if not clock.runs_follower(record.environment_card) or (token := clock_read_token(record)) is None: + if not clock.runs_follower(record.environment_card) or not record.control_token: continue try: sandbox = (await ComputerEnv.from_deployed_env(record))._sandbox - watch_url = sandbox.url_from_sandbox(f"{gateway}/clock/watch") - started[record.environment_url] = await clock.start_follower(sandbox, watch_url=watch_url, token=token, - follower=name) - logger.info(f"sync_env_clock env={self.env_id}: computer {name!r} follows the clock at {watch_url}") + started[record.environment_url] = await start_clock_follower(deployed, name, record, sandbox) + logger.info(f"sync_env_clock env={self.env_id}: computer {name!r} follows the clock") except Exception as e: if not self.tolerate_missing_sync_time: raise RuntimeError(f"computer {name!r} follows the clock, but its follower couldn't start: {e}") from e diff --git a/tst/integration/computer/clock_follower_test.py b/tst/integration/computer/clock_follower_test.py index 35ba335b4..8b9a7a456 100644 --- a/tst/integration/computer/clock_follower_test.py +++ b/tst/integration/computer/clock_follower_test.py @@ -4,8 +4,9 @@ ``sync_env_clock`` arms the gateway about 6 months back: core starts the computer's follower as root, the arm waits for it, and the session's shell and Python then read the gateway's universe time. An advance of a day and a re-arm at 60x reach them too, each acknowledged before its write returns. The session user can't write the clock page, read the -follower's config or stop the follower, and a follower killed as root is back for the next write. Needs Docker and the -local registry on localhost:5000.""" +follower's config or stop the follower, and a follower killed as root is back for the next write. A reset replaces the +computer's container, and the reset is done once its new follower shows the clock. Needs Docker and the local registry +on localhost:5000.""" import asyncio import subprocess @@ -24,6 +25,7 @@ from agent_env.task.teardown import teardown_run from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep +from agent_env.task_step.task_steps.reset_env import ResetEnvTaskStep from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep from tst.integration.env.envs.server_env_local_test import _put_items_image from tst.integration.store.local_parallel_deploy_test import _seed_bootstrap_envs @@ -98,7 +100,7 @@ def items(clock_follower_desktop): @pytest.mark.asyncio -async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_an_advance_and_a_rate( +async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_an_advance_a_rate_and_a_reset( clock_follower_desktop, items): with namespace_routing(): office = MultiEnv.put(id=f"@local/~/clock-{uuid.uuid4().hex[:8]}/office", mcp_server_envs=[items], @@ -146,6 +148,12 @@ async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_a await deployed.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target + 3 * DAY), "wait_followers_s": 10}) after_kill = await _agreement(computer, deployed) + + # A reset replaces the computer's container, follower and all; it's done once the new one shows the clock. + old_container = container + await ResetEnvTaskStep(id="reset", version=None, env_id=office.id, fail_task_on_error=True).execute(context) + after_reset = await _agreement(computer, deployed) + new_container = _container(computer) finally: report = await teardown_run(context) @@ -153,7 +161,8 @@ async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_a assert abs(before["shell"] - real_before) < 5, before # real time before the arm assert [s["service"] for s in configuration["synced"] if s["service"] == "clock"] == ["clock"], configuration assert arm_s < 30, arm_s - assert [read for read in (armed, advanced, after_kill) if not _agrees(read)] == [] + assert [read for read in (armed, advanced, after_kill, after_reset) if not _agrees(read)] == [] + assert new_container != old_container, "the reset didn't replace the computer's container" assert abs(armed["gateway"][0] - universe) < 60, (armed, universe) assert advanced["gateway"][0] - armed["gateway"][1] >= DAY, (armed, advanced) assert after_kill["gateway"][0] >= target + 3 * DAY, after_kill diff --git a/tst/unit/env/envs/test_computer_clock_core.py b/tst/unit/env/envs/test_computer_clock_core.py index 2ee648a24..9270473ae 100644 --- a/tst/unit/env/envs/test_computer_clock_core.py +++ b/tst/unit/env/envs/test_computer_clock_core.py @@ -43,3 +43,25 @@ async def write_secret(self, path, text): assert runs[0][:4] == ["install", "-d", "-m", "0700"] assert runs[-2] == ["python3", "/opt/agentenv/clock/clockpage.py", "create", "agentenv-clock"] assert "setsid /opt/agentenv/clock/follow" in runs[-1][-1] + + +@pytest.mark.asyncio +async def test_wait_shown_reads_the_followers_status_as_root_until_it_shows_the_config(monkeypatch): + statuses = ["", json.dumps({"config_id": "c0", "version": "b.1"}), json.dumps({"config_id": "c1", "version": "b.2"})] + + class Channel: + def __init__(self, sandbox): + pass + + async def run(self, argv): + return statuses.pop(0) if len(statuses) > 1 else statuses[0] + + monkeypatch.setattr(clock, "RootChannel", Channel) + monkeypatch.setattr(clock.asyncio, "sleep", lambda s: _done()) + assert (await clock.wait_shown(SimpleNamespace(), "c1"))["version"] == "b.2" + with pytest.raises(TimeoutError, match="didn't show config 'c9'"): + await clock.wait_shown(SimpleNamespace(), "c9", timeout_s=0) + + +async def _done(): + return None diff --git a/tst/unit/env/envs/test_multi_env_computers.py b/tst/unit/env/envs/test_multi_env_computers.py index 4216aad99..aa4d2f297 100644 --- a/tst/unit/env/envs/test_multi_env_computers.py +++ b/tst/unit/env/envs/test_multi_env_computers.py @@ -585,6 +585,45 @@ async def _stored(*computers) -> DeployedEnv: return DeployedEnv.from_dict({**dataclasses.asdict(deployed), "instance_id": "inst-1"}) + +@pytest.mark.asyncio +@pytest.mark.parametrize("armed", [True, False]) +async def test_a_computer_composing_the_clock_app_follows_an_armed_clock_again_once_it_resets(monkeypatch, armed): + taking = {"method": "POST", "request": {"properties": {"follower_config": {"type": "string"}}}} + clock_app = {"name": "clock", "url": "/svc/clock", "capabilities": {"extensions": [ + {"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": taking}}}]}} + record = dataclasses.replace(_record("desktop"), environment_card={"name": "desktop", + "children_environments": [clock_app]}) + deployed = MagicMock(spec=DeployedGatewayEnv, gateway_url=GATEWAY, invoke=AsyncMock(return_value={"armed": armed})) + sandbox = SimpleNamespace(url_from_sandbox=lambda url: url.replace("gw.example", "gw.internal")) + started, waited = [], [] + + class Rebooted: + @classmethod + async def from_deployed_env(cls, deployed_env): + assert deployed_env is record + return SimpleNamespace(_sandbox=sandbox) + + async def start_follower(sb, **options): + started.append((sb, options)) + return "c1" + + async def wait_shown(sb, config_id): + waited.append((sb, config_id)) + + monkeypatch.setattr(_multi_computers.clock, "start_follower", start_follower) + monkeypatch.setattr(_multi_computers.clock, "wait_shown", wait_shown) + await _multi_computers.follow_clock_again(deployed, "desktop", record, Rebooted()) + + if armed: + assert started == [(sandbox, {"watch_url": "https://gw.internal/clock/watch", "token": "token-desktop", + "follower": "desktop"})] + assert waited == [(sandbox, "c1")] + else: + assert started == waited == [] + await _multi_computers.follow_clock_again(deployed, "laptop", _record("laptop"), Rebooted()) + assert deployed.invoke.await_count == 1 # a computer running no follower never asks the gateway + @pytest.mark.asyncio async def test_reset_resets_every_child_env_at_once_each_as_it_resets_alone_and_forgets_what_loads_put_in( gateway, providers, server_resets):