From 90e85a05b49d2a143fa01d81c124715365da1276 Mon Sep 17 00:00:00 2001 From: William Loo Date: Thu, 12 Nov 2026 01:48:16 -0800 Subject: [PATCH] feat(sync_env_clock): a multi env's computers follow its clock through their followers, and the arm waits for them (FD-5477) A computer that composes the clock app is a clock follower: its app's card offers clock/v1 sync_time, which now counts for the gateway (clock_read_token looks at the computer's app cards too), so the gateway gives it a read token. sync_env_clock configures and starts each such computer's follower before the arm, as root through the sandbox's root channel (computer.utils.clock.start_follower), never through the driver: the gateway's /clock/watch as the computer's sandbox reaches it, its read token, its follower name and a config id, in a config only the follower's user can read; the clock page made for that user; and root's follow loop started. The arm then waits for the followers (wait_followers_s), so a computer still behind fails the step naming it, and each clock app's sync_time gets its config id and reports what its follower showed. A follower that can't start fails the step, or under tolerate_missing_sync_time is reported (follower_not_started) and not waited for. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/env/envs/_multi_computers.py | 12 +-- .../env/envs/computer/utils/clock.py | 76 ++++++++++++++++++ .../task_step/task_steps/sync_env_clock.py | 53 +++++++++++-- tst/unit/env/envs/test_computer_clock_core.py | 45 +++++++++++ tst/unit/env/envs/test_multi_env_computers.py | 18 ++++- tst/unit/task_step/sync_env_clock_test.py | 79 +++++++++++++++++++ 6 files changed, 270 insertions(+), 13 deletions(-) create mode 100644 src/agent_env/env/envs/computer/utils/clock.py create mode 100644 tst/unit/env/envs/test_computer_clock_core.py diff --git a/src/agent_env/env/envs/_multi_computers.py b/src/agent_env/env/envs/_multi_computers.py index 78dd344b8..bebe171ab 100644 --- a/src/agent_env/env/envs/_multi_computers.py +++ b/src/agent_env/env/envs/_multi_computers.py @@ -14,9 +14,8 @@ from agentenv_protocol.computer import tool_prefix_problem from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env -from agent_env.env.envs.computer.utils import sites +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: @@ -94,10 +93,11 @@ def routes_later(entry: Optional[dict]) -> bool: return isinstance(params, dict) and bool(params.get("site_routing")) and not params.get("sites") -def _clock_read_token(record: DeployedEnv) -> str | None: +def clock_read_token(record: DeployedEnv) -> str | None: """The read token a computer's clock watch carries when its card says it's a clock follower (``clock/v1``'s - ``sync_time``): for now its control token, which a token of its own can replace without the gateway knowing.""" - return record.control_token if record.supports(EXT_CLOCK_URI, "sync_time") else None + ``sync_time``, its own or, composing the ``clock`` app, an app's): for now its control token, which a token of its + own can replace without the gateway knowing.""" + return record.control_token if clock.follows_clock(record.environment_card) else None def refusal(computer: ComputerEnv, entry: dict, sandbox_type: str | None, websites: Mapping[str, str]) -> str | None: @@ -220,7 +220,7 @@ async def deploy(reach: Optional[str]) -> list[ExternalChild]: for error in failed[1:]: logger.warning(f"multi env '{self._owner}': another computer failed to deploy too: {error!r}") raise failed[0] - return [ExternalChild(env.environment_name, record.mcp_url, record.environment_url, _clock_read_token(record)) + return [ExternalChild(env.environment_name, record.mcp_url, record.environment_url, clock_read_token(record)) for env, record in zip(self.every, results)] return deploy diff --git a/src/agent_env/env/envs/computer/utils/clock.py b/src/agent_env/env/envs/computer/utils/clock.py new file mode 100644 index 000000000..9c122e835 --- /dev/null +++ b/src/agent_env/env/envs/computer/utils/clock.py @@ -0,0 +1,76 @@ +"""A multi env's computer on its gateway's clock: the clock follower configured and started as root, through the +sandbox's root channel, never through the driver, which runs as the session user. + +A computer follows when it composes the ``clock`` app (``agent_env/computers/envs/clock``): agent-env's libfaketime in +every process, reading the clock page, and the follower that writes it as ``agentenv-clock``. The app's card advertises +``clock/v1`` ``sync_time``, which makes the gateway count the computer as a follower (``follows_clock``) and gives it a +read token for its watch. ``sync_env_clock`` then calls ``start_follower`` for each such computer before it sets the +clock, and sets it waiting for their acknowledgements.""" +from __future__ import annotations + +import json +import secrets +from pathlib import PurePosixPath +from typing import TYPE_CHECKING + +from agentenv_protocol import client as protocol_v1 + +from agent_env.env.envs.computer.utils.sites import RootChannel +from agent_env.env.gateway.constants import EXT_CLOCK_URI + +if TYPE_CHECKING: + from agent_env.providers.sandbox_providers.sandbox import Sandbox + +#: The follower's user, which owns the clock page and the clock file, and alone reads the follower's config. +FOLLOWER_USER = "agentenv-clock" +CONFIG_FOLDER = PurePosixPath("/etc/agentenv/clock-follower") +CONFIG = CONFIG_FOLDER / "follower.json" +#: The clock file's folder, where the follower also keeps its status. +RUN_FOLDER = PurePosixPath("/run/agentenv-clock") +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" + + +def follows_clock(card: dict | None) -> bool: + """Whether a computer with ``card`` follows the gateway's clock: it, or one of its apps, offers ``clock/v1``'s + ``sync_time``.""" + if not isinstance(card, dict): + return False + return any(protocol_v1.find_extension_method(each, EXT_CLOCK_URI, "sync_time") is not None + for each in [card, *(child.card for child in protocol_v1.children(card) if child.card)]) + + +def takes_config(card: dict | None) -> bool: + """Whether a server with ``card`` is a clock app's, whose ``sync_time`` takes the follower's config id.""" + method = protocol_v1.find_extension_method(card or {}, EXT_CLOCK_URI, "sync_time") or {} + return CONFIG_PARAM in ((method.get("request") or {}).get("properties") or {}) + + +def runs_follower(card: dict | None) -> bool: + """Whether a computer with ``card`` runs the clock follower: one of its apps is the clock app (``takes_config``).""" + return isinstance(card, dict) and any(takes_config(child.card) for child in protocol_v1.children(card)) + + +async def start_follower(sandbox: Sandbox, *, watch_url: str, token: str, follower: str) -> str: + """Configure the computer's clock follower to watch ``watch_url`` as ``follower`` with ``token``, its read token, + and (re)start it, all as root; the id of the config written, which the clock app's ``sync_time`` waits for. + + The config folder is the follower's alone (0700), the config its alone to read (0600), so the session user never + sees the token. The clock page is made as root, the follower's from its first byte, replacing any the session user + made first; the library trusts no page another user owns (``clock.conf``'s ``AGENTENV_CLOCK_PAGE_UID``).""" + config_id = secrets.token_hex(8) + channel = RootChannel(sandbox) + owner = f"{FOLLOWER_USER}:{FOLLOWER_USER}" + await channel.run(["install", "-d", "-m", "0700", "-o", FOLLOWER_USER, "-g", FOLLOWER_USER, str(CONFIG_FOLDER)]) + await channel.write_secret(CONFIG, json.dumps({"watch_url": watch_url, "token": token, "follower": follower, + "config_id": config_id})) + await channel.run(["chown", owner, str(CONFIG)]) + await channel.run(["chmod", "0600", str(CONFIG)]) + await channel.run(["install", "-d", "-m", "0755", "-o", FOLLOWER_USER, "-g", FOLLOWER_USER, str(RUN_FOLDER)]) + await channel.run(["python3", str(CLOCK / "clockpage.py"), "create", FOLLOWER_USER]) + # 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 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..db1fe0abb 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,6 +18,10 @@ 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.envs.computer.env import ComputerEnv +from agent_env.env.envs.computer.utils import clock from agent_env.env.gateway.constants import EXT_CLOCK_URI from agent_env.task_step.context import StepContextError, TaskStepContext from agent_env.entity.refs import EntityRef @@ -28,6 +32,9 @@ 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 + def resolve_virtual_time(virtual_time: str, timezone: Optional[str]) -> str: """The instant to arm the gateway at. Without a zone, `virtual_time` unchanged; with one, a naive wall @@ -125,10 +132,13 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None) if deployed is None: raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs") + followers, unstarted = await self._start_followers(deployed) + arm = {"virtual_time": self.armed_virtual_time, "virtual_seconds_per_real_second": self.virtual_seconds_per_real_second} + if followers and not unstarted: + arm["wait_followers_s"] = FOLLOWER_WAIT_S try: - await deployed.invoke(EXT_CLOCK_URI, "set_time", - {"virtual_time": self.armed_virtual_time, "virtual_seconds_per_real_second": self.virtual_seconds_per_real_second}, - timeout=self.timeout_seconds) + await deployed.invoke(EXT_CLOCK_URI, "set_time", arm, + timeout=self.timeout_seconds + (FOLLOWER_WAIT_S if followers else 0)) except httpx.HTTPStatusError as e: raise RuntimeError(f"clock arm failed (HTTP {e.response.status_code}): {e.response.text}") from e state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=self.timeout_seconds) @@ -139,7 +149,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: service = server.environment_name outcome = await self._sync_one(server.owner(deployed), gateway_url_of(deployed), service, env_get_time_url, mcp=server.mcp, tolerate=self.tolerate_missing_sync_time or not server.mcp, - server=server) + server=server, followers=followers, unstarted=unstarted) entry = {"service": service, "environment": service, **({} if server.mcp else {"website": True}), **outcome} (synced if outcome.get("synced") else skipped).append(entry) @@ -158,8 +168,33 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: f"({[(s['service'], s.get('reason')) for s in skipped]}) — every tool still answers wall time") return context + async def _start_followers(self, deployed) -> tuple[dict[str, str], dict[str, str]]: + """Configure and start, as root, the clock follower of each of ``deployed``'s computers that follows the clock + (``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: + 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}") + 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 + logger.warning(f"sync_env_clock env={self.env_id}: computer {name!r}'s clock follower couldn't start, " + f"so the arm doesn't wait for computers: {e}") + unstarted[record.environment_url] = f"{type(e).__name__}: {e}" + return started, unstarted + async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time_url: Optional[str], mcp: bool = True, - tolerate: Optional[bool] = None, server: Optional[BackingServer] = None) -> dict: + tolerate: Optional[bool] = None, server: Optional[BackingServer] = None, + followers: Optional[dict[str, str]] = None, unstarted: Optional[dict[str, str]] = None) -> dict: """Invoke sync_time on one server, an MCP server or with ``mcp=False`` a website, if its card advertises clock/v1; skip a server with no card or no clock/v1 when ``tolerate`` (tolerate_missing_sync_time by default), else raise. ``execute`` tolerates every website, so one that doesn't follow the clock never fails the step; one @@ -185,6 +220,14 @@ async def _sync_one(self, deployed, gateway_url: str, service: str, env_get_time if not env_get_time_url: raise RuntimeError(f"gateway did not expose env_get_time_url; cannot sync {service}") params = {"env_get_time_url": env_get_time_url} + record = getattr(server, "record", None) + if record is not None and clock.takes_config(card): + if (config_id := (followers or {}).get(record.environment_url)) is None: + why = (unstarted or {}).get(record.environment_url, "its computer isn't one the gateway counts as a follower") + if tolerate: + return {"synced": False, "reason": "follower_not_started", "detail": why} + raise RuntimeError(f"{service} follows the clock, but its computer's follower didn't start: {why}") + params[clock.CONFIG_PARAM] = config_id if self.timezone is not None: method = protocol_v1.find_extension_method(card, EXT_CLOCK_URI, "sync_time") or {} if "timezone" not in ((method.get("request") or {}).get("properties") or {}): diff --git a/tst/unit/env/envs/test_computer_clock_core.py b/tst/unit/env/envs/test_computer_clock_core.py new file mode 100644 index 000000000..2ee648a24 --- /dev/null +++ b/tst/unit/env/envs/test_computer_clock_core.py @@ -0,0 +1,45 @@ +"""Core's side of a computer's clock: which computers run the clock follower, and its config written and the follower +started as root through the sandbox's root channel.""" +import json +from types import SimpleNamespace + +import pytest + +from agent_env.env.envs.computer.utils import clock + + +def test_a_computer_runs_the_follower_when_it_composes_the_clock_app(): + method = {"method": "POST", "request": {"properties": {"follower_config": {"type": "string"}}}} + clock_app = {"name": "clock", "capabilities": {"extensions": [ + {"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": method}}}]}} + driver_clock = {"name": "desktop", "capabilities": {"extensions": [ + {"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": {"method": "POST"}}}}]}} + + assert clock.runs_follower({"name": "desktop", "children_environments": [clock_app]}) + assert not clock.runs_follower(driver_clock) and clock.follows_clock(driver_clock) + assert clock.takes_config(clock_app) and not clock.takes_config(driver_clock) + + +@pytest.mark.asyncio +async def test_start_follower_writes_its_config_as_root_for_its_user_alone_then_starts_it(monkeypatch): + runs, secrets = [], {} + + class Channel: + def __init__(self, sandbox): + pass + + async def run(self, argv): + runs.append(argv) + + async def write_secret(self, path, text): + secrets[str(path)] = json.loads(text) + + monkeypatch.setattr(clock, "RootChannel", Channel) + config_id = await clock.start_follower(SimpleNamespace(), watch_url="http://h/clock/watch", token="t", follower="d") + + assert secrets == {str(clock.CONFIG): {"watch_url": "http://h/clock/watch", "token": "t", "follower": "d", + "config_id": config_id}} + assert ["chmod", "0600", str(clock.CONFIG)] in runs + 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] diff --git a/tst/unit/env/envs/test_multi_env_computers.py b/tst/unit/env/envs/test_multi_env_computers.py index 903c91a88..4216aad99 100644 --- a/tst/unit/env/envs/test_multi_env_computers.py +++ b/tst/unit/env/envs/test_multi_env_computers.py @@ -313,11 +313,25 @@ async def test_deploy_hands_the_gateway_each_computer_deployed_under_its_name_an def test_a_computer_whose_card_follows_a_clock_hands_the_gateway_its_control_token_as_its_read_token(): - from agent_env.env.envs._multi_computers import _clock_read_token + from agent_env.env.envs._multi_computers import clock_read_token following = dataclasses.replace(_record("desktop"), environment_card={"name": "desktop", "capabilities": {"extensions": [ {"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": {"method": "POST", "endpoint": "/clock/sync"}}}}]}}) - assert (_clock_read_token(following), _clock_read_token(_record("laptop"))) == ("token-desktop", None) + assert (clock_read_token(following), clock_read_token(_record("laptop"))) == ("token-desktop", None) + + +def test_a_computer_composing_the_clock_app_follows_the_clock_through_its_apps_card(): + from agent_env.env.envs._multi_computers import clock_read_token + + clock_app = {"name": "clock", "url": "/svc/clock", "capabilities": {"extensions": [ + {"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": {"method": "POST", "endpoint": "/x"}}}}]}} + other_app = {"name": "chromium", "url": "/svc/chromium", "capabilities": {"extensions": []}} + card = {"name": "desktop", "capabilities": {"extensions": []}} + composing = dataclasses.replace(_record("desktop"), environment_card={ + **card, "children_environments": [clock_app]}) + without = dataclasses.replace(_record("desktop"), environment_card={ + **card, "children_environments": [other_app]}) + assert (clock_read_token(composing), clock_read_token(without)) == ("token-desktop", None) @pytest.mark.asyncio diff --git a/tst/unit/task_step/sync_env_clock_test.py b/tst/unit/task_step/sync_env_clock_test.py index e95acf2b3..7930891c5 100644 --- a/tst/unit/task_step/sync_env_clock_test.py +++ b/tst/unit/task_step/sync_env_clock_test.py @@ -469,3 +469,82 @@ async def test_a_multi_envs_computer_is_synced_at_its_own_address_with_its_own_c def test_a_padded_zone_is_stored_and_sent_stripped(): s = _zoned(timezone=" America/New_York ") assert s.timezone == "America/New_York" and s.to_dict()["timezone"] == "America/New_York" + + +_CLOCK_APP_EXT = {"uri": EXT_CLOCK_URI, "params": {"endpoint": "/agentenv/ext/sync_time", "methods": {"sync_time": { + "method": "POST", "request": {"type": "object", "properties": {"env_get_time_url": {"type": "string"}, + "follower_config": {"type": "string"}}}}}}} + + +async def _clock_desktop(monkeypatch, start): + """A multi env whose computer, desktop, composes the clock app; ``start`` stands in for starting its follower.""" + import dataclasses + + from agent_env.env.env import DeployedSandboxEnv + from agent_env.env.envs.computer.utils import clock + from agent_env.env.gateway.constants import WELL_KNOWN_PATH + + computer = DeployedSandboxEnv(env_id="desk", env_version=1, sandbox_id="sb-desk", control_token="desk-token", + environment_card_url=f"https://desk.example{WELL_KNOWN_PATH}", + environment_card={"name": "desktop", "url": "/agentenv", + "capabilities": {"extensions": []}, + "children_environments": [{ + "name": "clock", "url": "/svc/clock/agentenv", + "capabilities": {"extensions": [_CLOCK_APP_EXT]}}]}) + record = dataclasses.replace(_carded(await _composed_card({"mcp-email": [_CLOCK_EXT]})), + control_token="gateway-token", child_envs={"desktop": dataclasses.asdict(computer)}) + sandbox = SimpleNamespace(url_from_sandbox=lambda url: url.replace("http://gw", "http://host.docker.internal:7")) + + async def reattach(deployed): + return SimpleNamespace(_sandbox=sandbox) + + monkeypatch.setattr("agent_env.env.envs.computer.env.ComputerEnv.from_deployed_env", reattach) + monkeypatch.setattr(clock, "start_follower", start) + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a, **k: SimpleNamespace( + mcp_server_envs=[SimpleNamespace(environment_name="email")], website_envs=[])) + return record + + +@pytest.mark.asyncio +async def test_a_computer_with_the_clock_app_follows_before_the_arm_which_waits_for_it(monkeypatch): + events = [] + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + + async def start(sandbox, **config): + events.append(("start", config)) + return "config-1" + + record = await _clock_desktop(monkeypatch, start) + context = await _step().execute(TaskStepContext(deployed_envs=[record])) + + arm = next(r for r in sent if r.url.path.endswith("/clock/set-time")) + assert events == [("start", {"watch_url": "http://host.docker.internal:7/clock/watch", "token": "desk-token", + "follower": "desktop"})] + assert json.loads(arm.content)["wait_followers_s"] == mod.FOLLOWER_WAIT_S + app_sync = next(r for r in sent if r.url.path.endswith("/sync_time") and "desk.example" in str(r.url)) + assert json.loads(app_sync.content) == {"env_get_time_url": CLOCK_URL, "follower_config": "config-1"} + synced = {s["service"] for s in context.metadata["clock_configurations"][0]["synced"]} + assert synced == {"email", "clock"} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tolerate", [True, False]) +async def test_a_follower_that_cannot_start_fails_the_step_or_is_reported_and_not_waited_for(monkeypatch, tolerate): + sent = _mock_http(monkeypatch, state={"env_get_time_url": CLOCK_URL}) + + async def start(sandbox, **config): + raise RuntimeError("site routing installs as root, and the given sandbox runs the computer's commands as uid 1000") + + record = await _clock_desktop(monkeypatch, start) + if not tolerate: + with pytest.raises(RuntimeError, match="computer 'desktop' follows the clock, but its follower couldn't start"): + await _step(tolerate_missing_sync_time=False).execute(TaskStepContext(deployed_envs=[record])) + assert not [r for r in sent if r.url.path.endswith("/clock/set-time")] + return + context = await _step().execute(TaskStepContext(deployed_envs=[record])) + + arm = next(r for r in sent if r.url.path.endswith("/clock/set-time")) + assert "wait_followers_s" not in json.loads(arm.content) + assert not [r for r in sent if r.url.path.endswith("/sync_time") and "desk.example" in str(r.url)] + skipped = {s["service"]: s for s in context.metadata["clock_configurations"][0]["skipped"]} + assert skipped["clock"]["reason"] == "follower_not_started" and "uid 1000" in skipped["clock"]["detail"]