feat(sync_env_clock): a multi env's computers follow its clock through their followers, and the arm waits for them (FD-5477)

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
 

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'} </dev/null >>{LOG} 2>&1 &"])
+    return config_id

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

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]

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