Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions src/agent_env/computers/clock/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -69,9 +69,12 @@ records what it showed in `/run/agentenv-clock/status.json`. A failed round is r
and a new config starts it over from no version.

The session user can't write the page or the clock file, read the follower's config, or stop either process: each is
another user's. Not covered yet: a computer reset (`reset_env`) boots it afresh with no follower until the next arm;
agent-env's own servers run on universe time too, with no real-time marker; and routed sites' certificates, Chromium's
time zone and an egress proxy for live sites at other dates are later work.
another user's. A multi env's reset (`reset_env`) boots a computer afresh, follower and all; while the clock is armed,
core starts the new follower as an arm does and the reset ends once the computer shows the clock.

Not covered yet: agent-env's own servers run on universe time too, with no real-time marker; and certificates checked
by TLS stacks the shim doesn't reach (Chromium's), Chromium's time zone and an egress proxy for live sites at other dates
are later work.

## The clock page

Expand Down
34 changes: 32 additions & 2 deletions src/agent_env/env/envs/_multi_computers.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,9 +13,10 @@

from agentenv_protocol.computer import tool_prefix_problem

from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env
from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, Env, gateway_url_of
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:
Expand Down Expand Up @@ -100,6 +101,30 @@ def clock_read_token(record: DeployedEnv) -> str | None:
return record.control_token if clock.follows_clock(record.environment_card) else None


async def start_clock_follower(deployed: DeployedEnv, name: str, record: DeployedEnv, sandbox: Any) -> Optional[str]:
"""Start the clock follower of ``deployed``'s computer ``name``, its record ``record`` and its sandbox ``sandbox``,
watching the gateway's clock as the sandbox reaches it (``clock.start_follower``): the config id it was given, or
None for a computer that runs none."""
if not clock.runs_follower(record.environment_card) or (token := clock_read_token(record)) is None:
return None
watch_url = sandbox.url_from_sandbox(f"{(gateway_url_of(deployed) or '').rstrip('/')}/clock/watch")
return await clock.start_follower(sandbox, watch_url=watch_url, token=token, follower=name)


async def follow_clock_again(deployed: DeployedEnv, name: str, record: DeployedEnv, computer: ComputerEnv) -> None:
"""After ``deployed``'s computer ``name`` reset, its clock follower, gone with its container, started again while
the gateway's clock is armed, and the reset done once the computer shows it; nothing for a computer that runs
none, or a clock not armed, which the next ``sync_env_clock`` starts it for."""
if not clock.runs_follower(record.environment_card) or not gateway_url_of(deployed):
return
if not (await deployed.invoke(EXT_CLOCK_URI, "state")).get("armed"):
return
sandbox = (await type(computer).from_deployed_env(record))._sandbox
if (config_id := await start_clock_follower(deployed, name, record, sandbox)) is not None:
await clock.wait_shown(sandbox, config_id)
logger.info(f"computer {name!r} follows the clock again after its reset")


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
Expand Down Expand Up @@ -179,9 +204,14 @@ def resets(self, deployed: DeployedEnv) -> list[tuple[str, Awaitable[None]]]:
record; none is awaited yet."""
records = getattr(deployed, "child_envs", None) or {}
browser = self.browser or self._recorded_browser(records)
return [(env.environment_name, env.reset(DeployedEnv.from_dict(records[env.environment_name])))
return [(env.environment_name, self._reset(deployed, env, DeployedEnv.from_dict(records[env.environment_name])))
for env in [*self.envs, *([browser] if browser else [])] if env.environment_name in records]

@staticmethod
async def _reset(deployed: DeployedEnv, env: ComputerEnv, record: DeployedEnv) -> None:
await env.reset(record)
await follow_clock_again(deployed, env.environment_name, record, env)

def _recorded_browser(self, records: Mapping[str, dict]) -> Optional[ComputerEnv]:
"""The website browser ``records``, a multi env's computers' records, hold beside its own computers: the record
of a name none of them has, its env at the version it deployed."""
Expand Down
23 changes: 23 additions & 0 deletions src/agent_env/env/envs/computer/utils/clock.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,10 @@
clock, and sets it waiting for their acknowledgements."""
from __future__ import annotations

import asyncio
import json
import secrets
import time
from pathlib import PurePosixPath
from typing import TYPE_CHECKING

Expand All @@ -27,11 +29,15 @@
CONFIG = CONFIG_FOLDER / "follower.json"
#: The clock file's folder, where the follower also keeps its status.
RUN_FOLDER = PurePosixPath("/run/agentenv-clock")
#: What the follower last showed, which anyone may read.
STATUS = RUN_FOLDER / "status.json"
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"
#: How long a write of the clock, or a computer's follower started again, waits for the computer to show it.
FOLLOWER_WAIT_S = 10.0


def follows_clock(card: dict | None) -> bool:
Expand Down Expand Up @@ -74,3 +80,20 @@ async def start_follower(sandbox: Sandbox, *, watch_url: str, token: str, follow
# 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


async def wait_shown(sandbox: Sandbox, config_id: str, timeout_s: float = FOLLOWER_WAIT_S) -> dict:
"""What the computer's follower showed once it has shown a version of the clock for ``config_id``, read as root;
TimeoutError naming its last status when it hasn't within ``timeout_s``."""
channel, deadline, status = RootChannel(sandbox), time.monotonic() + timeout_s, None
while True:
try:
status = json.loads(await channel.run(["sh", "-c", f"cat {STATUS} 2>/dev/null || true"]))
except json.JSONDecodeError:
status = None
if status and status.get("config_id") == config_id and status.get("version") and not status.get("error"):
return status
if time.monotonic() >= deadline:
raise TimeoutError(f"the clock follower didn't show config {config_id!r} within {timeout_s:g} s; its "
f"status: {status}")
await asyncio.sleep(0.2)
15 changes: 6 additions & 9 deletions src/agent_env/task_step/task_steps/sync_env_clock.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@
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.env import DeployedEnv
from agent_env.env.envs._multi_computers import start_clock_follower
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
Expand All @@ -33,7 +33,7 @@
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
FOLLOWER_WAIT_S = clock.FOLLOWER_WAIT_S


def resolve_virtual_time(virtual_time: str, timezone: Optional[str]) -> str:
Expand Down Expand Up @@ -173,17 +173,14 @@ async def _start_followers(self, deployed) -> tuple[dict[str, str], dict[str, st
(``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:
if not clock.runs_follower(record.environment_card) or not record.control_token:
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}")
started[record.environment_url] = await start_clock_follower(deployed, name, record, sandbox)
logger.info(f"sync_env_clock env={self.env_id}: computer {name!r} follows the clock")
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
Expand Down
17 changes: 13 additions & 4 deletions tst/integration/computer/clock_follower_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,9 @@
``sync_env_clock`` arms the gateway about 6 months back: core starts the computer's follower as root, the arm waits for
it, and the session's shell and Python then read the gateway's universe time. An advance of a day and a re-arm at 60x
reach them too, each acknowledged before its write returns. The session user can't write the clock page, read the
follower's config or stop the follower, and a follower killed as root is back for the next write. Needs Docker and the
local registry on localhost:5000."""
follower's config or stop the follower, and a follower killed as root is back for the next write. A reset replaces the
computer's container, and the reset is done once its new follower shows the clock. Needs Docker and the local registry
on localhost:5000."""

import asyncio
import subprocess
Expand All @@ -24,6 +25,7 @@
from agent_env.task.teardown import teardown_run
from agent_env.task_step.context import TaskStepContext
from agent_env.task_step.task_steps.deploy_env import DeployEnvTaskStep
from agent_env.task_step.task_steps.reset_env import ResetEnvTaskStep
from agent_env.task_step.task_steps.sync_env_clock import SyncEnvClockTaskStep
from tst.integration.env.envs.server_env_local_test import _put_items_image
from tst.integration.store.local_parallel_deploy_test import _seed_bootstrap_envs
Expand Down Expand Up @@ -98,7 +100,7 @@ def items(clock_follower_desktop):


@pytest.mark.asyncio
async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_an_advance_and_a_rate(
async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_an_advance_a_rate_and_a_reset(
clock_follower_desktop, items):
with namespace_routing():
office = MultiEnv.put(id=f"@local/~/clock-{uuid.uuid4().hex[:8]}/office", mcp_server_envs=[items],
Expand Down Expand Up @@ -146,14 +148,21 @@ async def test_a_multi_envs_computer_follows_its_gateways_clock_through_an_arm_a
await deployed.invoke(EXT_CLOCK_URI, "advance", {"virtual_time": _iso(target + 3 * DAY),
"wait_followers_s": 10})
after_kill = await _agreement(computer, deployed)

# A reset replaces the computer's container, follower and all; it's done once the new one shows the clock.
old_container = container
await ResetEnvTaskStep(id="reset", version=None, env_id=office.id, fail_task_on_error=True).execute(context)
after_reset = await _agreement(computer, deployed)
new_container = _container(computer)
finally:
report = await teardown_run(context)

assert follows, "the computer's card doesn't say it runs the clock follower"
assert abs(before["shell"] - real_before) < 5, before # real time before the arm
assert [s["service"] for s in configuration["synced"] if s["service"] == "clock"] == ["clock"], configuration
assert arm_s < 30, arm_s
assert [read for read in (armed, advanced, after_kill) if not _agrees(read)] == []
assert [read for read in (armed, advanced, after_kill, after_reset) if not _agrees(read)] == []
assert new_container != old_container, "the reset didn't replace the computer's container"
assert abs(armed["gateway"][0] - universe) < 60, (armed, universe)
assert advanced["gateway"][0] - armed["gateway"][1] >= DAY, (armed, advanced)
assert after_kill["gateway"][0] >= target + 3 * DAY, after_kill
Expand Down
22 changes: 22 additions & 0 deletions tst/unit/env/envs/test_computer_clock_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,3 +43,25 @@ async def write_secret(self, path, text):
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]


@pytest.mark.asyncio
async def test_wait_shown_reads_the_followers_status_as_root_until_it_shows_the_config(monkeypatch):
statuses = ["", json.dumps({"config_id": "c0", "version": "b.1"}), json.dumps({"config_id": "c1", "version": "b.2"})]

class Channel:
def __init__(self, sandbox):
pass

async def run(self, argv):
return statuses.pop(0) if len(statuses) > 1 else statuses[0]

monkeypatch.setattr(clock, "RootChannel", Channel)
monkeypatch.setattr(clock.asyncio, "sleep", lambda s: _done())
assert (await clock.wait_shown(SimpleNamespace(), "c1"))["version"] == "b.2"
with pytest.raises(TimeoutError, match="didn't show config 'c9'"):
await clock.wait_shown(SimpleNamespace(), "c9", timeout_s=0)


async def _done():
return None
39 changes: 39 additions & 0 deletions tst/unit/env/envs/test_multi_env_computers.py
Original file line number Diff line number Diff line change
Expand Up @@ -585,6 +585,45 @@ async def _stored(*computers) -> DeployedEnv:
return DeployedEnv.from_dict({**dataclasses.asdict(deployed), "instance_id": "inst-1"})



@pytest.mark.asyncio
@pytest.mark.parametrize("armed", [True, False])
async def test_a_computer_composing_the_clock_app_follows_an_armed_clock_again_once_it_resets(monkeypatch, armed):
taking = {"method": "POST", "request": {"properties": {"follower_config": {"type": "string"}}}}
clock_app = {"name": "clock", "url": "/svc/clock", "capabilities": {"extensions": [
{"uri": "urn:agentenv:clock/v1", "params": {"methods": {"sync_time": taking}}}]}}
record = dataclasses.replace(_record("desktop"), environment_card={"name": "desktop",
"children_environments": [clock_app]})
deployed = MagicMock(spec=DeployedGatewayEnv, gateway_url=GATEWAY, invoke=AsyncMock(return_value={"armed": armed}))
sandbox = SimpleNamespace(url_from_sandbox=lambda url: url.replace("gw.example", "gw.internal"))
started, waited = [], []

class Rebooted:
@classmethod
async def from_deployed_env(cls, deployed_env):
assert deployed_env is record
return SimpleNamespace(_sandbox=sandbox)

async def start_follower(sb, **options):
started.append((sb, options))
return "c1"

async def wait_shown(sb, config_id):
waited.append((sb, config_id))

monkeypatch.setattr(_multi_computers.clock, "start_follower", start_follower)
monkeypatch.setattr(_multi_computers.clock, "wait_shown", wait_shown)
await _multi_computers.follow_clock_again(deployed, "desktop", record, Rebooted())

if armed:
assert started == [(sandbox, {"watch_url": "https://gw.internal/clock/watch", "token": "token-desktop",
"follower": "desktop"})]
assert waited == [(sandbox, "c1")]
else:
assert started == waited == []
await _multi_computers.follow_clock_again(deployed, "laptop", _record("laptop"), Rebooted())
assert deployed.invoke.await_count == 1 # a computer running no follower never asks the gateway

@pytest.mark.asyncio
async def test_reset_resets_every_child_env_at_once_each_as_it_resets_alone_and_forgets_what_loads_put_in(
gateway, providers, server_resets):
Expand Down
Loading