feat(clock): core's clock writes wait for the computers on the clock and fail their step naming any behind (FD-5473)

src/agent_env/env/env.py

@@ -8,6 +8,7 @@
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, ClassVar, Optional, Self
 
+import httpx
 from agentenv_protocol import client as protocol_v1
 from agentenv_protocol.website import READY_TIMEOUT_S, WEBSITE_EXTENSION_URI, WEBSITE_READY_OP, WebsiteReadiness, env_readiness
 
@@ -18,13 +19,16 @@
 from agent_env.utils.paths import validate_relative_filename
 
 from .gateway import GatewayMode
-from .gateway.constants import WELL_KNOWN_PATH
+from .gateway.constants import CLOCK_FOLLOWERS_BEHIND, CLOCK_WRITES, EXT_CLOCK_URI, WELL_KNOWN_PATH
 from .store import get_env_store
 
 logger = logging.getLogger(__name__)
 
 # Default timeout for one invoke() call; above the class because a default argument needs it.
 _EXTENSION_CALL_TIMEOUT_S = 30
+#: How long each clock write core makes waits for the computers following the clock to show it, so no step hands an
+#: agent a computer still showing the old time.
+CLOCK_WAIT_FOLLOWERS_S = 2.0
 
 if TYPE_CHECKING:
     from agent_env.bundle.authoring import AuthoringContext
@@ -46,6 +50,30 @@ def __init__(self, capability: str, method: str, env_id: str, instance_id: Optio
         super().__init__(f"env '{env_id}'{child} does not offer '{method}' on {capability}.")
 
 
+class ClockFollowersBehind(RuntimeError):
+    """A clock write the env applied that some of its followers, the computers on its clock, didn't show in time: they
+    still show the old time. ``stale`` names them."""
+
+    def __init__(self, env_id: str, method: str, stale: list[str], detail: str) -> None:
+        self.env_id = env_id
+        self.method = method
+        self.stale = stale
+        super().__init__(f"env '{env_id}': clock {method}: {detail}")
+
+
+def _followers_behind(response: httpx.Response) -> list[str] | None:
+    """The computers a clock write's 504 names as behind (``stale``), or None for any other answer, a proxy's own 504
+    page among them."""
+    if response.status_code != 504:
+        return None
+    try:
+        body = response.json()
+    except ValueError:
+        return None
+    stale = body.get("stale") if isinstance(body, dict) else None
+    return stale if isinstance(stale, list) else None
+
+
 class EnvNeedsGateway(RuntimeError):
     """A caller reaches an env through its gateway, and the env was deployed without one."""
 
@@ -171,6 +199,23 @@ async def invoke(self, uri: str, method: str, params: Optional[dict] = None, *,
         return await protocol_v1.invoke_extension(self.environment_url, card, uri, params, timeout, method=method,
                                                   headers=self.control_headers)
 
+    async def write_clock(self, method: str, params: Optional[dict] = None, *,
+                          timeout: int = _EXTENSION_CALL_TIMEOUT_S) -> Any:
+        """A clock/v1 write (``CLOCK_WRITES``) that waits up to ``CLOCK_WAIT_FOLLOWERS_S`` for every computer following
+        the clock to show it, unless ``params`` sets its own ``wait_followers_s``. Raises ``ClockFollowersBehind``
+        naming those that didn't; the write itself has applied. A clock no computer follows answers at once. A gateway
+        from before the error's code answers the same 504 with its message as ``error``, which counts the same."""
+        if method not in CLOCK_WRITES:
+            raise ValueError(f"{method!r} isn't a clock write; one of {list(CLOCK_WRITES)}")
+        params = {"wait_followers_s": CLOCK_WAIT_FOLLOWERS_S, **(params or {})}
+        try:
+            return await self.invoke(EXT_CLOCK_URI, method, params, timeout=timeout)
+        except protocol_v1.ExtensionCallError as e:
+            if (stale := _followers_behind(e.response)) is None:
+                raise
+            detail = e.detail if e.code == CLOCK_FOLLOWERS_BEHIND else e.response.json()["error"]
+            raise ClockFollowersBehind(self.env_id, method, stale, detail) from e
+
     async def readiness(self, timeout: float = READY_TIMEOUT_S) -> WebsiteReadiness:
         """Whether the env is ready, as the protocol determines a website env's health (``env_readiness``): every site
         its card, the stored one or, for a record without it, the card it serves now, advertises ``website/v1``'s

src/agent_env/env/gateway/constants.py

@@ -77,6 +77,11 @@ def data_plane_load_timeout_s(size_bytes: int | None) -> int:
 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
+# The clock/v1 methods that change the clock, each taking wait_followers_s.
+CLOCK_WRITES = ("set_time", "set_rate", "advance", "clear")
+# The error code a clock write answers 504 with when a declared follower didn't show it within its wait_followers_s:
+# the write applied, and the followers named in its stale are behind.
+CLOCK_FOLLOWERS_BEHIND = "clock_followers_behind"
 
 # Env/gateway-level extensions the gateway itself owns (tool gating + triggers).
 EXT_DISABLE_TOOL_URI = "urn:agentenv:disable-tool/v1"
@@ -244,7 +249,7 @@ 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/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 followers that don't show it in time (set_rate keeps t0, as advance does, so time triggers keep their marks; rate 0 freezes the clock); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.",
+        "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, error code clock_followers_behind, with stale naming the followers that don't show it in time (set_rate keeps t0, as advance does, so time triggers keep their marks; rate 0 freezes the clock); a follower follows it through watch, a long poll naming the version it shows and held up to hold_s; state's followers block is diagnostic, not part of the contract.",
         "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"}, **_WAIT_FOLLOWERS}, "required": ["virtual_time"]}},

src/agent_env/env/gateway/gateway.py

@@ -48,6 +48,7 @@
     CHANGES_PAGE_DEFAULT,
     CHANGES_PAGE_MAX,
     CHILD_DISCOVERY_TIMEOUT_S,
+    CLOCK_FOLLOWERS_BEHIND,
     CONTROL_ROUTES,
     DATA_PLANE_LOAD_TIMEOUT_MAX_S,
     DEFAULT_ROLE,
@@ -692,7 +693,8 @@ async def _await_barriers(self, pending: list[asyncio.Task], timeout_s: float |
     async def _clock_write(self, request: Request, change, *, body_optional: bool = False) -> Response:
         """A clock write route: its JSON object body (none at all when ``body_optional``) applied by ``change`` through
         ``_change_clock``; 400 for a body or value the clock refuses, else 200, or 504 naming the followers behind, with
-        what each follower last showed and when it last watched."""
+        what each follower last showed and when it last watched, its error in the protocol's shape so a client reads its
+        code (``CLOCK_FOLLOWERS_BEHIND``) and message."""
         try:
             body = await request.json() if not body_optional or await request.body() else {}
         except Exception:
@@ -705,10 +707,11 @@ async def _clock_write(self, request: Request, change, *, body_optional: bool =
         except ClockError as e:
             return self._json_response({"ok": False, "error": str(e)}, 400)
         if stale:
-            return self._json_response({**state, "ok": False, "stale": stale, "followers": self._clock_followers.state(),
-                                        "error": (f"the clock is at version {state['version']}, but {', '.join(stale)} "
-                                                  f"didn't show it (waited up to {wait:g} s; one that has stopped "
-                                                  "watching isn't waited for): their clocks are behind")}, 504)
+            message = (f"the clock is at version {state['version']}, but {', '.join(stale)} didn't show it (waited "
+                       f"up to {wait:g} s; one that has stopped watching isn't waited for): their clocks are behind")
+            error = {"code": CLOCK_FOLLOWERS_BEHIND, "message": message, "retryable": False, "applied": "yes"}
+            return self._json_response({**state, "ok": False, "error": error, "stale": stale,
+                                        "followers": self._clock_followers.state()}, 504)
         return self._json_response(state, 200)
 
     async def _change_clock(self, change, wait_s: float = 0.0) -> tuple[dict, list[str]]:

src/agent_env/task_step/task_steps/always_on.py

@@ -95,10 +95,11 @@
 from agent_env.a2a_agent import A2AAgent, conversation_store, protocol
 from agent_env.a2a_agent.object_transfer import send_and_wait
 from agent_env.artifact import FileArtifact
-from agent_env.env.env import EnvCapabilityUnsupported
+from agent_env.env.env import ClockFollowersBehind, EnvCapabilityUnsupported
 from agent_env.env.gateway.notifications import DEFAULT_MAX_ACTIVE
 from agent_env.env.gateway.constants import (
     CHANGES_PAGE_MAX,
+    CLOCK_WRITES,
     DEFAULT_ROLE,
     EXT_CHANGES_URI,
     EXT_CLOCK_URI,
@@ -776,6 +777,8 @@ async def advance(self, target: datetime, reason: str) -> None:
         for env in armed:
             try:
                 await _call(env, EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target)}, attempts=2)
+            except ClockFollowersBehind:
+                raise
             except Exception:
                 state = await _call(env, EXT_CLOCK_URI, "state")
                 if not (state.get("armed") and _utc(state["virtual_time"]) >= target):
@@ -841,6 +844,8 @@ async def settle(self) -> list[dict]:
                 elif due:  # an event already due: an advance to now sets it going
                     try:
                         await _call(env, EXT_CLOCK_URI, "advance", {"virtual_time": _iso(now)}, attempts=1)
+                    except ClockFollowersBehind:
+                        raise
                     except Exception:
                         pass
             if waiting:
@@ -1817,12 +1822,15 @@ async def _call(env, uri: str, method: str, params: Optional[dict] = None, *, at
                 timeout: int = 60) -> Any:
     """A gateway control call, retried on a transient failure (a busy gateway can take seconds to answer).
     Every call the step retries is safe to repeat: reads, config, advance (checked against the clock after) and takes
-    with a take_id."""
+    with a take_id. A clock write waits for the computers following the clock, and one they didn't show isn't retried:
+    it applied, and the step fails naming them."""
     last: Optional[Exception] = None
     for attempt in range(attempts):
         try:
+            if uri == EXT_CLOCK_URI and method in CLOCK_WRITES:
+                return await env.write_clock(method, params, timeout=timeout)
             return await env.invoke(uri, method, params, timeout=timeout)
-        except EnvCapabilityUnsupported:
+        except (EnvCapabilityUnsupported, ClockFollowersBehind):
             raise
         except Exception as e:
             last = e

src/agent_env/task_step/task_steps/post_turn_hooks.py

@@ -20,9 +20,10 @@
 from agentenv_protocol import client as protocol_v1
 
 from agent_env.artifact import FileArtifact
-from agent_env.env.env import DeployedEnv, EnvCapabilityUnsupported
+from agent_env.env.env import ClockFollowersBehind, DeployedEnv, EnvCapabilityUnsupported
 from agent_env.env.gateway.constants import (
     AGENT_ENV_ROLE_HEADER,
+    CLOCK_WRITES,
     EXT_CLOCK_URI,
     EXT_NOTIFICATIONS_URI,
     EXT_STEP_URI,
@@ -120,12 +121,14 @@ def close(self) -> None:
 
 class HookEnv:
     """A deployed env as a hook reaches it: clock, triggers and notifications calls, and tool calls recorded as the
-    hook's."""
+    hook's. A clock write waits for the computers following the clock (``DeployedEnv.write_clock``); one they didn't
+    show is kept in ``behind`` too, so the step fails on it even when the hook catches it."""
 
     def __init__(self, deployed: DeployedEnv, hook_artifact_id: str) -> None:
         self.env_id = deployed.env_id
         self.env_step_id = (deployed.metadata or {}).get("deploy_step_id")
         self.instance_id = deployed.instance_id
+        self.behind: Optional[ClockFollowersBehind] = None
         self._deployed = deployed
         self._hook_artifact_id = hook_artifact_id
 
@@ -135,6 +138,12 @@ def supports(self, extension: str, method: str) -> bool:
     async def invoke(self, extension: str, method: str, params: Optional[dict] = None) -> Any:
         if extension not in _INVOKABLE:
             raise ValueError(f"a post-turn hook invokes {list(_INVOKABLE)} only; call tools with call_tool")
+        if extension == EXT_CLOCK_URI and method in CLOCK_WRITES:
+            try:
+                return await self._deployed.write_clock(method, params)
+            except ClockFollowersBehind as e:
+                self.behind = self.behind or e
+                raise
         return await self._deployed.invoke(extension, method, params)
 
     async def call_tool(self, tool: str, args: Optional[dict], role: str) -> dict:
@@ -265,12 +274,17 @@ async def run_post_turn_hooks(hooks: PostTurnHooks, *, context: TaskStepContext,
                               agent_reply: str, agent_parts: list[dict],
                               ask_user: Callable[[], Awaitable[Answer]],
                               user_triggers: Callable[[], Awaitable[Optional[Answer]]]) -> Optional[Answer]:
-    """Run the hooks in order; the first to return a message or an end decides, and one that asked returns that."""
+    """Run the hooks in order; the first to return a message or an end decides, and one that asked returns that. A
+    clock write a hook made that the computers following the clock didn't show fails the step, caught by the hook or
+    not, so the agent never acts on a computer still showing the old time."""
     answer = _Answer(ask_user, user_triggers)
     for hook in hooks.hooks:
         turn = Turn(context=context, step_id=step_id, hook=hook, answer=answer, number=number, last=last,
                     agent_reply=agent_reply, agent_parts=agent_parts)
         returned = await hook.post_turn(turn, copy.deepcopy(hook.params))
+        behind = next((env.behind for env in turn.envs if env.behind), None)
+        if behind is not None:
+            raise behind
         if answer.value is not None:
             if returned != answer.value:
                 raise RuntimeError(f"post-turn hook '{hook.artifact_id}' asked the user side at turn {number} and "

src/agent_env/task_step/task_steps/sync_env_clock.py

@@ -1,8 +1,10 @@
 """Arm a deployed env gateway's virtual clock, then sync its backing servers to it.
 
 Arms the clock (set_time) and reads the gateway's server-reachable env_get_time_url (state), both as the
-stored env card declares clock/v1, then invokes sync_time on each backing server, MCP server or website, the
-env card lists (``legacy_protocol.backing_servers``) whose own card advertises it, at the route that card gives.
+stored env card declares clock/v1. The arm waits for the computers that follow the clock to show it, and fails the
+step naming any that don't (``DeployedEnv.write_clock``). It then invokes sync_time on each backing server, MCP server
+or website, the env card lists (``legacy_protocol.backing_servers``) whose own card advertises it, at the route that
+card gives.
 Non-advertisers are skipped when tolerate_missing_sync_time (default True), and websites always. Place right
 before prompt_agent so the virtual timeline starts at agent-start.
 """
@@ -126,9 +128,9 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         if deployed is None:
             raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
         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)
+            arm = {"virtual_time": self.armed_virtual_time,
+                   "virtual_seconds_per_real_second": self.virtual_seconds_per_real_second}
+            await deployed.write_clock("set_time", arm, timeout=self.timeout_seconds)
         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)