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):