From 29bb42e53dbd1db830141f2ba61202544fafd3de Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Fri, 13 Nov 2026 17:48:48 -0800 Subject: [PATCH 1/3] test(computers): the clock CI test: a multi env's computer keeps within 1 s of its gateway's clock through jumps, rates, a kill and a reset (FD-5482) computer_clock_test's one deploy now also takes the gateway's clock a week on and a month back, to a rate of 60, frozen and running again, each checked from inside the computer: the session's time read between two reads of real time, so no driver or exec latency is in it and the bound tightens from 5 s to 1 s. Co-Authored-By: Claude Opus 5.5 --- .../computer/computer_clock_test.py | 102 +++++++++++++----- 1 file changed, 75 insertions(+), 27 deletions(-) diff --git a/tst/integration/computer/computer_clock_test.py b/tst/integration/computer/computer_clock_test.py index e6fe5c1ee..152de389f 100644 --- a/tst/integration/computer/computer_clock_test.py +++ b/tst/integration/computer/computer_clock_test.py @@ -1,9 +1,10 @@ """The ubuntu-24.04 computer on its multi env's gateway clock, on the local backend: the deploy has its follower follow the gateway, as the computer's sandbox reaches it, under the computer's name and control token, which the record -keeps but the token; sync_env_clock arms the clock as a task does, calling the computer's sync_time; the session -reads universe time, and each change is acknowledged in time for a write that waits for it; a follower killed comes -back within 3 s and acknowledges the next advance; and a reset brings the page and the follower back on the clock's -current version. Needs Docker and the local registry on localhost:5000.""" +keeps but the token; sync_env_clock arms the clock as a task does, calling the computer's sync_time; through the +gateway's jumps of a day and a week on and a month back, a rate of 60, a freeze and a restart, a killed follower and a +reset, the session's clock is within 1 s of the gateway's once the follower has acknowledged each change, and runs at +the gateway's rate; a follower killed comes back within 3 s. Each reading is taken inside the computer, between two +of real time, so no driver or exec latency is in it. Needs Docker and the local registry on localhost:5000.""" import asyncio import time @@ -27,6 +28,15 @@ pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] DAY = 86400 +#: How far the session's clock may be from the gateway's: the follower maps the gateway's clock exactly, so this is +#: room for the gateway's state read from here, not for the computer. +MAX_SKEW_S = 1.0 +#: The widest a reading's bracket of real time may be; one wider is taken again. +BRACKET_S = 0.5 +#: Inside the computer: real time, the session's time, real time again, each a fresh ``date``, the first and last +#: started on real time (``AGENTENV_REAL_TIME``). +BRACKET = ('r1=$(AGENTENV_REAL_TIME=1 date +%s.%N); u=$(date +%s.%N); r2=$(AGENTENV_REAL_TIME=1 date +%s.%N); ' + 'echo "$r1 $u $r2"') def _iso(moment: datetime) -> str: @@ -34,7 +44,7 @@ def _iso(moment: datetime) -> str: @pytest.mark.asyncio -async def test_a_multi_envs_computer_follows_its_gateways_clock_through_arming_a_killed_follower_and_a_reset( +async def test_a_multi_envs_computer_keeps_within_a_second_of_its_gateways_clock_through_jumps_rates_a_kill_and_a_reset( desktop, items): with namespace_routing(): office = multi_env(items, Env.get(desktop.id)) @@ -51,20 +61,27 @@ async def test_a_multi_envs_computer_follows_its_gateways_clock_through_arming_a await SyncEnvClockTaskStep(id="clock", version=None, env_id=office.id, virtual_time=_iso(universe)).execute(context) synced = context.metadata["clock_configurations"][-1]["synced"] - armed = await _session_offset(own) - (universe.timestamp() - time.time()) - - day_on = universe + timedelta(days=1) - await _advance(deployed, day_on) - advanced = await _session_offset(own) - (day_on.timestamp() - time.time()) + skews = {"armed": await _skew(deployed, own)} + for name, by in (("a day on", timedelta(days=1)), ("a week on", timedelta(weeks=1))): + await _write(deployed, "advance", virtual_time=_iso(await _now(deployed) + by)) + skews[name] = await _skew(deployed, own) + await _write(deployed, "set_time", virtual_time=_iso(await _now(deployed) - timedelta(days=30)), + virtual_seconds_per_real_second=1) + skews["a month back"] = await _skew(deployed, own) + await _write(deployed, "set_rate", virtual_seconds_per_real_second=60) + rates = {"60": await _rate(own)} + await _write(deployed, "set_rate", virtual_seconds_per_real_second=0) + rates["frozen"], skews["frozen"] = await _rate(own), await _skew(deployed, own) + await _write(deployed, "set_rate", virtual_seconds_per_real_second=1) + skews["running again"] = await _skew(deployed, own) killed, back, came_back = kill_follower(container) - later = day_on + timedelta(hours=8) - await _advance(deployed, later) - after_kill = await _session_offset(own) - (later.timestamp() - time.time()) + await _write(deployed, "advance", virtual_time=_iso(await _now(deployed) + timedelta(hours=8))) + skews["after the follower's kill"] = await _skew(deployed, own) await ResetEnvTaskStep(id="reset", version=None, env_id=office.id, fail_task_on_error=True).execute(context) - state = await _shown(deployed) - after_reset = await _session_offset(own) - (_universe_now(state) - time.time()) + await _shown(deployed) + skews["after a reset"] = await _skew(deployed, own) state_after_reset = await _shown(deployed) finally: report = await teardown_run(context) @@ -72,16 +89,54 @@ async def test_a_multi_envs_computer_follows_its_gateways_clock_through_arming_a assert kept == {"watch_url": kept["watch_url"], "name": "desktop"} assert urlsplit(kept["watch_url"]).hostname == "host.docker.internal" and kept["watch_url"].endswith("/clock/watch") assert [entry["service"] for entry in synced if entry.get("service") == "desktop"] == ["desktop"] - assert abs(armed) < 5 and abs(advanced) < 5 and abs(after_kill) < 5 and abs(after_reset) < 5, ( - armed, advanced, after_kill, after_reset) + assert all(abs(skew) < MAX_SKEW_S for skew in skews.values()), skews + assert abs(rates["60"] / 60 - 1) < 0.05 and abs(rates["frozen"]) < 0.01, rates assert back not in ("-", killed) and came_back < FOLLOWER_BACK_S, came_back assert state_after_reset["followers"]["desktop"]["version"] == state_after_reset["version"], state_after_reset assert not report.failed -async def _advance(deployed, to: datetime) -> None: - """The clock moved to ``to``, waiting up to 15 s for every computer to show it, as a 504 naming any that doesn't.""" - await deployed.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(to), "wait_followers_s": 15}, timeout=30) +async def _write(deployed, method: str, **params) -> None: + """A clock write that waits up to 15 s for every computer to show it, as a 504 naming any that doesn't.""" + await deployed.invoke(EXT_CLOCK_URI, method, {**params, "wait_followers_s": 15}, timeout=30) + + +async def _now(deployed) -> datetime: + state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=30) + return datetime.fromtimestamp(_universe_now(state), timezone.utc) + + +async def _reading(computer) -> tuple[float, float]: + """(real, universe): the session's time and the real time it was read at, both from inside the computer.""" + for _ in range(3): + ran = await inside.run(computer, ["sh", "-c", BRACKET], timeout_s=30) + assert ran["exit_code"] == 0, ran + before, universe, after = map(float, ran["stdout"].split()) + if after - before < BRACKET_S: + return (before + after) / 2, universe + raise AssertionError(f"reading the computer's time three times took over {BRACKET_S} s each: {ran['stdout']}") + + +async def _skew(deployed, computer) -> float: + """How far the session's clock is from the gateway's, in universe seconds: the gateway's state, its real time the + middle of its request from here, carried at its rate to the real time of the computer's reading.""" + for _ in range(3): + sent = time.time() + state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=30) + answered = time.time() + if answered - sent < BRACKET_S: + break + real, universe = await _reading(computer) + expected = _universe_now(state) + state["virtual_seconds_per_real_second"] * (real - (sent + answered) / 2) + return universe - expected + + +async def _rate(computer, over_s: float = 2.0) -> float: + """The session clock's rate: universe seconds per real second between two readings ``over_s`` apart.""" + real, universe = await _reading(computer) + await asyncio.sleep(over_s) + later, then = await _reading(computer) + return (then - universe) / (later - real) async def _shown(deployed, within: float = 15) -> dict: @@ -96,10 +151,3 @@ async def _shown(deployed, within: float = 15) -> dict: def _universe_now(state: dict) -> float: return datetime.fromisoformat(state["virtual_time"].replace("Z", "+00:00")).timestamp() - - -async def _session_offset(computer) -> float: - """How far the session's clock is from real time, read with ``date`` through the driver, as the session user.""" - ran = await inside.run(computer, ["date", "+%s.%N"], timeout_s=30) - assert ran["exit_code"] == 0, ran - return float(ran["stdout"]) - time.time() From b50b77b94dcfc99f9f53b4b9cb586f633783da49 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Fri, 13 Nov 2026 18:06:02 -0800 Subject: [PATCH 2/3] test(computers): the clock test sets a rate of 10, not 60 A rate change reaches the session's clock the follower's reaction late, carried at the old rate until the next jump: about 1 ms on a loaded host, 67 ms at rate 60. At 10 the follower would have to be 100 ms late to reach the 1 s bound. Co-Authored-By: Claude Opus 5.5 --- tst/integration/computer/computer_clock_test.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tst/integration/computer/computer_clock_test.py b/tst/integration/computer/computer_clock_test.py index 152de389f..79c2f3091 100644 --- a/tst/integration/computer/computer_clock_test.py +++ b/tst/integration/computer/computer_clock_test.py @@ -1,7 +1,7 @@ """The ubuntu-24.04 computer on its multi env's gateway clock, on the local backend: the deploy has its follower follow the gateway, as the computer's sandbox reaches it, under the computer's name and control token, which the record keeps but the token; sync_env_clock arms the clock as a task does, calling the computer's sync_time; through the -gateway's jumps of a day and a week on and a month back, a rate of 60, a freeze and a restart, a killed follower and a +gateway's jumps of a day and a week on and a month back, a rate of 10, a freeze and a restart, a killed follower and a reset, the session's clock is within 1 s of the gateway's once the follower has acknowledged each change, and runs at the gateway's rate; a follower killed comes back within 3 s. Each reading is taken inside the computer, between two of real time, so no driver or exec latency is in it. Needs Docker and the local registry on localhost:5000.""" @@ -28,6 +28,9 @@ pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow] DAY = 86400 +#: The rate the clock is set to: a rate change reaches the session the follower's reaction late, carried at the old +#: rate until the next jump, so a modest rate keeps a busy runner's late follower well inside ``MAX_SKEW_S``. +RATE = 10 #: How far the session's clock may be from the gateway's: the follower maps the gateway's clock exactly, so this is #: room for the gateway's state read from here, not for the computer. MAX_SKEW_S = 1.0 @@ -68,8 +71,8 @@ async def test_a_multi_envs_computer_keeps_within_a_second_of_its_gateways_clock await _write(deployed, "set_time", virtual_time=_iso(await _now(deployed) - timedelta(days=30)), virtual_seconds_per_real_second=1) skews["a month back"] = await _skew(deployed, own) - await _write(deployed, "set_rate", virtual_seconds_per_real_second=60) - rates = {"60": await _rate(own)} + await _write(deployed, "set_rate", virtual_seconds_per_real_second=RATE) + rates = {RATE: await _rate(own)} await _write(deployed, "set_rate", virtual_seconds_per_real_second=0) rates["frozen"], skews["frozen"] = await _rate(own), await _skew(deployed, own) await _write(deployed, "set_rate", virtual_seconds_per_real_second=1) @@ -90,7 +93,7 @@ async def test_a_multi_envs_computer_keeps_within_a_second_of_its_gateways_clock assert urlsplit(kept["watch_url"]).hostname == "host.docker.internal" and kept["watch_url"].endswith("/clock/watch") assert [entry["service"] for entry in synced if entry.get("service") == "desktop"] == ["desktop"] assert all(abs(skew) < MAX_SKEW_S for skew in skews.values()), skews - assert abs(rates["60"] / 60 - 1) < 0.05 and abs(rates["frozen"]) < 0.01, rates + assert abs(rates[RATE] / RATE - 1) < 0.05 and abs(rates["frozen"]) < 0.01, rates assert back not in ("-", killed) and came_back < FOLLOWER_BACK_S, came_back assert state_after_reset["followers"]["desktop"]["version"] == state_after_reset["version"], state_after_reset assert not report.failed From a4d4897c4191a9612b33918b4f657855a5e5eb80 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Fri, 13 Nov 2026 18:22:15 -0800 Subject: [PATCH 3/3] test(computers): each clock reading's bracket is narrow enough at the clock's rate, and the skew is checked at rate 10 too A reading is taken again until half its bracket of real time, at the clock's rate, is at most 0.25 universe seconds, the gateway's as the computer's, and fails after five tries; so the rate is known to within a few percent and the skew at rate 10 to within 0.5 s of its truth. Co-Authored-By: Claude Opus 5.5 --- .../computer/computer_clock_test.py | 61 ++++++++++++------- 1 file changed, 39 insertions(+), 22 deletions(-) diff --git a/tst/integration/computer/computer_clock_test.py b/tst/integration/computer/computer_clock_test.py index 79c2f3091..cdda42e51 100644 --- a/tst/integration/computer/computer_clock_test.py +++ b/tst/integration/computer/computer_clock_test.py @@ -32,10 +32,13 @@ #: rate until the next jump, so a modest rate keeps a busy runner's late follower well inside ``MAX_SKEW_S``. RATE = 10 #: How far the session's clock may be from the gateway's: the follower maps the gateway's clock exactly, so this is -#: room for the gateway's state read from here, not for the computer. +#: room for reading the two, not for the computer. MAX_SKEW_S = 1.0 -#: The widest a reading's bracket of real time may be; one wider is taken again. -BRACKET_S = 0.5 +#: How far, in universe seconds, a reading's bracket of real time may leave it: half the bracket at the clock's rate. +#: The gateway's reading and the computer's each get this much, together well inside ``MAX_SKEW_S``, and a reading +#: whose bracket is wider is taken again. +READING_S = 0.25 +ATTEMPTS = 5 #: Inside the computer: real time, the session's time, real time again, each a fresh ``date``, the first and last #: started on real time (``AGENTENV_REAL_TIME``). BRACKET = ('r1=$(AGENTENV_REAL_TIME=1 date +%s.%N); u=$(date +%s.%N); r2=$(AGENTENV_REAL_TIME=1 date +%s.%N); ' @@ -72,9 +75,9 @@ async def test_a_multi_envs_computer_keeps_within_a_second_of_its_gateways_clock virtual_seconds_per_real_second=1) skews["a month back"] = await _skew(deployed, own) await _write(deployed, "set_rate", virtual_seconds_per_real_second=RATE) - rates = {RATE: await _rate(own)} + rates, skews[f"at a rate of {RATE}"] = {RATE: await _rate(own, RATE)}, await _skew(deployed, own) await _write(deployed, "set_rate", virtual_seconds_per_real_second=0) - rates["frozen"], skews["frozen"] = await _rate(own), await _skew(deployed, own) + rates["frozen"], skews["frozen"] = await _rate(own, 0), await _skew(deployed, own) await _write(deployed, "set_rate", virtual_seconds_per_real_second=1) skews["running again"] = await _skew(deployed, own) @@ -109,36 +112,50 @@ async def _now(deployed) -> datetime: return datetime.fromtimestamp(_universe_now(state), timezone.utc) -async def _reading(computer) -> tuple[float, float]: +def _widest(rate: float) -> float: + """The widest a bracket of real time may be for a reading at ``rate`` to be within ``READING_S`` of its truth.""" + return 2 * READING_S / max(rate, 1) + + +async def _reading(computer, rate: float) -> tuple[float, float]: """(real, universe): the session's time and the real time it was read at, both from inside the computer.""" - for _ in range(3): + for _ in range(ATTEMPTS): ran = await inside.run(computer, ["sh", "-c", BRACKET], timeout_s=30) assert ran["exit_code"] == 0, ran before, universe, after = map(float, ran["stdout"].split()) - if after - before < BRACKET_S: + if after - before < _widest(rate): return (before + after) / 2, universe - raise AssertionError(f"reading the computer's time three times took over {BRACKET_S} s each: {ran['stdout']}") + raise AssertionError(f"the computer's time took over {_widest(rate):.3f} s to read each of {ATTEMPTS} times: " + f"{ran['stdout']}") -async def _skew(deployed, computer) -> float: - """How far the session's clock is from the gateway's, in universe seconds: the gateway's state, its real time the - middle of its request from here, carried at its rate to the real time of the computer's reading.""" - for _ in range(3): +async def _gateway(deployed) -> tuple[float, dict]: + """(real, state): the gateway's clock state and the real time it was read at, from here.""" + for _ in range(ATTEMPTS): sent = time.time() state = await deployed.invoke(EXT_CLOCK_URI, "state", timeout=30) answered = time.time() - if answered - sent < BRACKET_S: - break - real, universe = await _reading(computer) - expected = _universe_now(state) + state["virtual_seconds_per_real_second"] * (real - (sent + answered) / 2) - return universe - expected + if answered - sent < _widest(state["virtual_seconds_per_real_second"]): + return (sent + answered) / 2, state + raise AssertionError(f"the gateway's state took over {_widest(state['virtual_seconds_per_real_second']):.3f} s " + f"to read each of {ATTEMPTS} times") -async def _rate(computer, over_s: float = 2.0) -> float: - """The session clock's rate: universe seconds per real second between two readings ``over_s`` apart.""" - real, universe = await _reading(computer) +async def _skew(deployed, computer) -> float: + """How far the session's clock is from the gateway's, in universe seconds: the gateway's state carried at its rate + to the real time of the computer's reading.""" + real_then, state = await _gateway(deployed) + rate = state["virtual_seconds_per_real_second"] + real, universe = await _reading(computer, rate) + return universe - (_universe_now(state) + rate * (real - real_then)) + + +async def _rate(computer, rate: float, over_s: float = 2.0) -> float: + """The session clock's rate, set to ``rate``: universe seconds per real second between two readings ``over_s`` + apart.""" + real, universe = await _reading(computer, rate) await asyncio.sleep(over_s) - later, then = await _reading(computer) + later, then = await _reading(computer, rate) return (then - universe) / (later - real)