feat(gateway): computers follow the clock through a watch, and clock writes wait for them (FD-5471)

src/agent_env/env/envs/_multi_computers.py

@@ -16,6 +16,7 @@
 from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env
 from agent_env.env.envs.computer.utils import 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:
@@ -93,6 +94,12 @@ 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:
+    """The read token a computer's clock watch carries when its card says it follows a clock (``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
+
+
 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
@@ -213,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)
+            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/gateway/__init__.py

@@ -30,11 +30,13 @@ class InternalMCPServer:
 @dataclass(frozen=True)
 class ExternalChild:
     """A child env on a sandbox of its own, which the gateway fronts as it does an MCP server: its name, and its MCP
-    endpoint and base URL as the process holding it reaches them."""
+    endpoint and base URL as the process holding it reaches them. A computer that follows the gateway's clock has the
+    read token its clock watch carries, and the gateway's clock writes wait for it."""
 
     name: str
     mcp_url: str
     base_url: str
+    clock_read_token: str | None = None
 
 
 __all__ = [

src/agent_env/env/gateway/clock.py

@@ -7,6 +7,7 @@
 
 import math
 import re
+import secrets
 import threading
 import time
 from datetime import datetime, timedelta, timezone
@@ -28,13 +29,14 @@ class ClockError(ValueError):
 def _parse_rate(rate) -> float:
     if isinstance(rate, bool) or not isinstance(rate, (int, float)):
         raise ClockError(f"rate must be a number, got {type(rate).__name__}")
-    # NaN passes both ordered bounds below, then poisons every later read with an uncatchable 500.
-    if not math.isfinite(rate):
+    # NaN passes both ordered bounds below, then poisons every later read with an uncatchable 500; only a float can be
+    # NaN or infinite, and math.isfinite on an int past a float's range raises.
+    if isinstance(rate, float) and not math.isfinite(rate):
         raise ClockError(f"rate must be a finite number, got {rate}")
     if rate < 0:
         raise ClockError("rate must be non-negative")
     if rate > MAX_RATE:
-        raise ClockError(f"rate {rate:g} exceeds max {MAX_RATE:g} (1 real s = 1 virtual day)")
+        raise ClockError(f"rate exceeds max {MAX_RATE:g} (1 real s = 1 virtual day)")
     return float(rate)
 
 
@@ -125,6 +127,8 @@ def _advance(t0: datetime, seconds: float) -> datetime:
 class Clock:
     """Gateway-owned virtual clock — wall-clock-driven, monotonic, threadsafe; `now_fn` injectable for tests."""
 
+    BOOT_ID_BYTES = 6  # a version's per-boot id, random, so a restarted gateway's versions never repeat its last boot's
+
     def __init__(self, now_fn=time.monotonic):
         self._now = now_fn
         self._lock = threading.Lock()
@@ -135,11 +139,26 @@ def __init__(self, now_fn=time.monotonic):
         self._generation = 0
         self._lived = 0.0  # virtual seconds run before the current arm/clear
         self._ahead = 0.0  # virtual seconds the current arm was advanced by
+        self._boot = secrets.token_hex(self.BOOT_ID_BYTES)
+        self._count = 0
 
     @property
     def armed(self) -> bool:
         return self._armed
 
+    @property
+    def version(self) -> str:
+        """Opaque, changed by every write: this clock's own id and a count, so it never repeats across restarts."""
+        return f"{self._boot}.{self._count}"
+
+    def count_of(self, version) -> int | None:
+        """The count of ``version`` when this clock has had it; None for another clock's, a later one, or junk."""
+        boot, _, count = version.partition(".") if isinstance(version, str) else ("", "", "")
+        current = str(self._count)
+        if boot != self._boot or not (count.isascii() and count.isdigit()) or len(count) > len(current):
+            return None
+        return int(count) if int(count) <= self._count else None
+
     @property
     def rate(self) -> float:
         """Virtual seconds per real second; 1.0 while unarmed."""
@@ -168,6 +187,25 @@ def set_time(self, virtual_time: str, rate=1.0) -> dict:
             self._ahead = 0.0
             self._armed = True
             self._generation += 1
+            self._count += 1
+        return self.state()
+
+    def set_rate(self, rate) -> dict:
+        """Change an armed clock's rate from its time at the call, keeping t0 and generation as an advance does, so time
+        triggers keep their marks and a rate change can't step the clock back by a time read a moment earlier."""
+        r = _parse_rate(rate)
+        with self._lock:
+            if not self._armed or self._t0 is None:
+                raise ClockError("clock is not armed")
+            now = self._now()
+            ran = (now - self._anchor) * self._rate
+            if _advance(self._t0, ran + self._ahead) > _MAX_T0 - timedelta(seconds=_OVERFLOW_HORIZON_S * r):
+                raise ClockError(f"rate {r:g} would overflow datetime near the max")
+            self._lived += ran
+            self._ahead += ran
+            self._anchor = now
+            self._rate = r
+            self._count += 1
         return self.state()
 
     def advance(self, virtual_time: str) -> dict:
@@ -184,6 +222,7 @@ def advance(self, virtual_time: str) -> dict:
                 raise ClockError(f"advance only moves forward: {virtual_time!r} is before the current {_iso(current)}")
             self._ahead += jump
             self._lived += jump
+            self._count += 1
         return self.state()
 
     def clear(self) -> dict:
@@ -195,7 +234,8 @@ def clear(self) -> dict:
             self._rate = 1.0
             self._ahead = 0.0
             self._generation += 1
-        return {"ok": True, "armed": False}
+            self._count += 1
+            return {"ok": True, "armed": False, "version": self.version}
 
     def elapsed(self) -> float:
         """Virtual seconds run so far: real time at each arm's rate (1 unarmed), plus each forward re-arm or advance."""
@@ -226,10 +266,11 @@ def read(self) -> dict:
     def state(self) -> dict:
         with self._lock:
             if not self._armed or self._t0 is None:
-                return {"armed": False}
+                return {"armed": False, "version": self.version}
             elapsed = self._now() - self._anchor
             return {"armed": True, "t0": _iso(self._t0), "virtual_seconds_per_real_second": self._rate,
-                    "virtual_time": _iso(_advance(self._t0, elapsed * self._rate + self._ahead))}
+                    "virtual_time": _iso(_advance(self._t0, elapsed * self._rate + self._ahead)),
+                    "version": self.version}
 
     def _bank(self, t0: datetime | None = None) -> None:
         """Fold the stretch since the last arm/clear into `_lived` at its rate, and start the next one now."""

src/agent_env/env/gateway/clock_followers.py

@@ -0,0 +1,91 @@
+"""The computers that follow the gateway's clock: each long-polls ``/clock/watch`` with the version it shows, which is
+its acknowledgement, and a clock write can wait until every one shows the write's version."""
+from __future__ import annotations
+
+import asyncio
+import hmac
+import math
+import time
+from collections.abc import Callable, Mapping
+
+from .clock import Clock, ClockError
+from .constants import CLOCK_MAX_WAIT_FOLLOWERS_S, CLOCK_WATCH_HOLD_S
+
+
+def parse_wait(value) -> float:
+    """A write's ``wait_followers_s``: how long it waits for the followers, 0 when absent."""
+    if value is None:
+        return 0.0
+    # Only a float can be NaN or infinite; math.isfinite on an int past a float's range raises.
+    if isinstance(value, bool) or not isinstance(value, (int, float)) or (isinstance(value, float) and not math.isfinite(value)) \
+            or value < 0:
+        raise ClockError(f"wait_followers_s must be a non-negative number, got {value!r}")
+    if value > CLOCK_MAX_WAIT_FOLLOWERS_S:
+        raise ClockError(f"wait_followers_s exceeds max {CLOCK_MAX_WAIT_FOLLOWERS_S:g}")
+    return float(value)
+
+
+class ClockFollowers:
+    """The declared followers, each by name with the read token its watch carries, and the version each last showed;
+    bound to the gateway's event loop."""
+
+    WATCH_S = CLOCK_WATCH_HOLD_S
+
+    def __init__(self, clock: Callable[[], Clock], tokens: Mapping[str, str]):
+        self._clock = clock
+        self._tokens = dict(tokens)
+        self._shown: dict[str, tuple[int, str] | None] = dict.fromkeys(self._tokens)
+        self._seen: dict[str, float] = {}
+        self._changed = asyncio.Event()
+        self._acks = asyncio.Condition()
+
+    def changed(self) -> None:
+        """Answer every held watch: the clock has a new version."""
+        self._changed.set()
+        self._changed = asyncio.Event()
+
+    def follower_for(self, name: str | None, token: str) -> str | None:
+        """``name`` when it's a declared follower and ``token`` its read token, else None."""
+        expected = self._tokens.get(name or "")
+        return name if expected and token and hmac.compare_digest(token.encode(), expected.encode()) else None
+
+    async def watch(self, after: str, follower: str | None) -> None:
+        """Take ``after`` as ``follower``'s acknowledgement (one ``follower_for`` named) when it's a version this clock
+        has had, later than the one it last showed; then return at once when the clock's version isn't ``after``, else
+        at its next change or after ``WATCH_S``."""
+        changed = self._changed
+        if follower is not None:
+            count = self._clock().count_of(after)
+            async with self._acks:
+                shown = self._shown[follower]
+                if count is not None and (shown is None or count > shown[0]):
+                    self._shown[follower] = (count, after)
+                self._seen[follower] = time.monotonic()
+                self._acks.notify_all()
+        if after == self._clock().version:
+            try:
+                await asyncio.wait_for(changed.wait(), self.WATCH_S)
+            except asyncio.TimeoutError:
+                pass
+
+    async def stale_after(self, version: str, wait_s: float) -> list[str]:
+        """The followers that haven't shown ``version`` (or a later one) once each has or ``wait_s`` has passed."""
+        target = self._clock().count_of(version)
+
+        def stale() -> list[str]:
+            return sorted(name for name, shown in self._shown.items()
+                          if shown is None or target is None or shown[0] < target)
+
+        async with self._acks:
+            try:
+                await asyncio.wait_for(self._acks.wait_for(lambda: not stale()), wait_s)
+            except asyncio.TimeoutError:
+                pass
+            return stale()
+
+    def state(self) -> dict:
+        """Each follower's last shown version and how long ago it last watched, for ``/clock/state``."""
+        now = time.monotonic()
+        return {name: {"version": shown[1] if shown else None,
+                       "seen_s_ago": round(now - self._seen[name], 3) if name in self._seen else None}
+                for name, shown in self._shown.items()}

src/agent_env/env/gateway/constants.py

@@ -69,6 +69,14 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int:
 METHOD_GET = "data/get"
 MCP_TRANSPORT = "mcp"
 CONTROL_TOKEN_ENV = "AGENTENV_CONTROL_TOKEN"
+# The computers that follow the gateway's clock, each as `name=read token`, comma-separated; TOKEN in the name keeps
+# it out of logged compose files.
+CLOCK_FOLLOWERS_ENV = "CLOCK_FOLLOWER_TOKENS"
+# How long /clock/watch holds a watch naming the current version before answering it unchanged; a follower's request
+# timeout must be longer. The card's watch method advertises it as hold_s.
+CLOCK_WATCH_HOLD_S = 25.0
+# The longest a clock write's wait_followers_s may ask it to wait for the followers.
+CLOCK_MAX_WAIT_FOLLOWERS_S = 30.0
 
 # Env/gateway-level extensions the gateway itself owns (tool gating + triggers).
 EXT_DISABLE_TOOL_URI = "urn:agentenv:disable-tool/v1"
@@ -201,6 +209,8 @@ def random_mcp_server_name() -> str:
     "if": {"properties": {"action": {"const": "call_tool"}}},
     "then": {"required": ["tool_name"]},
 }
+_WAIT_FOLLOWERS = {"wait_followers_s": {"type": "number", "minimum": 0, "maximum": CLOCK_MAX_WAIT_FOLLOWERS_S}}
+
 GATEWAY_EXTENSIONS = [
     {
         "uri": EXT_DISABLE_TOOL_URI,
@@ -226,13 +236,15 @@ def random_mcp_server_name() -> str:
     },
     {
         "uri": EXT_CLOCK_URI,
-        "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set-time/advance/clear are the harness control path.",
+        "description": "Virtual clock (urn:agentenv:clock/v1): read the gateway's virtual time (wall-clock-derived, rate-scaled) via get_time; set_time/set_rate/advance/clear are the harness control path, each changing the clock's opaque version and, given wait_followers_s, answering 504 naming the computers following the clock that don't show it in time (set_rate keeps t0, as advance does); a computer follows it through watch, a long poll naming the version it shows.",
         "params": {"endpoint": "/clock/time", "methods": {
             "get_time": {"method": "GET", "endpoint": "/clock/time"},
-            "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}}, "required": ["virtual_time"]}},
-            "advance": {"method": "PUT", "endpoint": "/clock/advance", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}}, "required": ["virtual_time"]}},
-            "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {}}},
-            "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}}}},
+            "set_time": {"method": "PUT", "endpoint": "/clock/set-time", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, "virtual_seconds_per_real_second": {"type": "number"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}},
+            "set_rate": {"method": "PUT", "endpoint": "/clock/set-rate", "request": {"type": "object", "properties": {"virtual_seconds_per_real_second": {"type": "number"}, **_WAIT_FOLLOWERS}, "required": ["virtual_seconds_per_real_second"]}},
+            "advance": {"method": "PUT", "endpoint": "/clock/advance", "request": {"type": "object", "properties": {"virtual_time": {"type": "string"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}},
+            "clear": {"method": "POST", "endpoint": "/clock/clear", "request": {"type": "object", "properties": {**_WAIT_FOLLOWERS}}},
+            "state": {"method": "GET", "endpoint": "/clock/state", "request": {"type": "object", "properties": {}}},
+            "watch": {"method": "GET", "endpoint": "/clock/watch", "hold_s": CLOCK_WATCH_HOLD_S, "request": {"type": "object", "properties": {"after": {"type": "string"}, "follower": {"type": "string"}}}}}},
         "required": False,
     },
     {

src/agent_env/env/gateway/entrypoint.py

@@ -8,7 +8,7 @@
 import httpx
 
 from . import AGENT_ENV_GATEWAY_MCP_PORT, GatewayMode, InternalMCPServer
-from .constants import (CONTROL_TOKEN_ENV, DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV, MCP_TRANSPORT,
+from .constants import (CLOCK_FOLLOWERS_ENV, CONTROL_TOKEN_ENV, DEFAULT_MCP_SERVER_NAME, GATEWAY_PROVIDER_HEADERS_ENV, MCP_TRANSPORT,
                         WELL_KNOWN_PATH)
 from .gateway import Gateway, card_url
 
@@ -188,6 +188,19 @@ def parse_name_url_map(env_str: str) -> dict[str, str]:
     return urls
 
 
+def parse_clock_followers(spec: str) -> dict[str, str]:
+    """``CLOCK_FOLLOWER_TOKENS``: each computer that follows the clock as ``name=read token``, comma-separated."""
+    followers: dict[str, str] = {}
+    for entry in filter(None, (part.strip() for part in spec.split(","))):
+        name, _, token = (part.strip() for part in entry.partition("="))
+        if not name or not token:
+            raise ValueError(f"invalid {CLOCK_FOLLOWERS_ENV} entry for {name or '?'!r}: expected 'name=read token'")
+        if name in followers:
+            raise ValueError(f"{CLOCK_FOLLOWERS_ENV} names {name!r} twice")
+        followers[name] = token
+    return followers
+
+
 if __name__ == "__main__":
     logging.basicConfig(
         level=logging.INFO,
@@ -218,5 +231,6 @@ def parse_name_url_map(env_str: str) -> dict[str, str]:
         service_db_url=service_db_url,
         control_token=os.environ.get(CONTROL_TOKEN_ENV),
         provider_headers=os.environ.get(GATEWAY_PROVIDER_HEADERS_ENV, "").split(","),
+        clock_followers=parse_clock_followers(os.environ.get(CLOCK_FOLLOWERS_ENV, "")),
     )
     gateway.run()