diff --git a/src/agent_env/computers/README.md b/src/agent_env/computers/README.md index 7d8ddd21e..a8fff484e 100644 --- a/src/agent_env/computers/README.md +++ b/src/agent_env/computers/README.md @@ -53,6 +53,12 @@ a deploy waiting for another's compose says what it waits for and since when. A env's form at deploy where a configured store isn't the per-user one, since the run's fence refuses its record: the deploy refuses for want of a ready build and names `agent-env env build`. +Each build's record names the digest of the code that composed it: the composer, these build files, the Lume provider +and the protocol's `computer` package. A build an agent-env with other composing code recorded, or one from before +records named it, isn't deployed: a deploy that boots its form composes it again first, saying so, as `env build` and +a put do, and its record says `by another composer` until then. So upgrading agent-env never boots a computer composed +the old way. + Its settings reach the computer as environment variables, `AGENTENV_SETTING_`, read as it starts: | Setting | Default | What it does | diff --git a/src/agent_env/env/envs/computer/env.py b/src/agent_env/env/envs/computer/env.py index 2ec41575d..d339d150d 100644 --- a/src/agent_env/env/envs/computer/env.py +++ b/src/agent_env/env/envs/computer/env.py @@ -28,6 +28,7 @@ provider_or_class, ) from agent_env.env.envs.computer.utils import boot, composer, gate, inside +from agent_env.env.envs.computer.utils.composed import stale from agent_env.env.envs.computer.utils.composer import Composing from agent_env.env.envs.computer.utils import params as computer_params from agent_env.env.envs.computer.driver_env import ComputerDriverEnv @@ -208,9 +209,13 @@ async def waiting(seconds: float) -> None: if pairs := composer.unstarted(self, sandbox_provider, form): them = "it" if len(pairs) == 1 else "them" on = Composing(on_progress=progress) - say(f"composing {_pairs(pairs)} before this deploy on " - f"{' or '.join(type(link).__name__ for link in on.provider().links)} (no compose built {them} for " - f"{' or '.join(type(link).__name__ for link in sandbox_provider.links)} yet); later deploys reuse {them}") + composed = self.metadata.get("composed", {}) + booting, building = (" or ".join(type(link).__name__ for link in provider.links) + for provider in (sandbox_provider, on.provider())) + why = (f"an agent-env that composes differently built {them}" if any( + stale(composed.get(name, {}).get(arch)) for name, arch in pairs) + else f"no compose built {them} for {booting} yet") + say(f"composing {_pairs(pairs)} before this deploy on {building} ({why}); later deploys reuse {them}") composing = asyncio.ensure_future(asyncio.to_thread(composer.compose, self, pairs, on)) try: await asyncio.shield(composing) @@ -225,10 +230,12 @@ async def waiting(seconds: float) -> None: def deployable_builds(self, form: Optional[str] = None) -> list[tuple[str, str, dict]]: """Each build a deploy may boot, as (form, arch, record), in the base's order of forms and of arches; only ``form``'s when it's given: a ready one, or a disk no compose could probe (unverified), which its first boot on - a provider that boots it natively probes before it's used.""" + a provider that boots it natively probes before it's used; never one an agent-env that composes differently + recorded.""" composed, arches = self.metadata.get("composed", {}), self.computer.arch return [(name, arch, composed[name][arch]) for name in gate.forms(self.computer) if form in (None, name) - for arch in arches if composed.get(name, {}).get(arch, {}).get("status") in _DEPLOYABLE[name]] + for arch in arches if composed.get(name, {}).get(arch, {}).get("status") in _DEPLOYABLE[name] + and not stale(composed[name][arch])] def disk_size_refusal(self, form: str, disk_size_gb: float) -> Optional[str]: """Why a deploy of ``form`` can't have ``disk_size_gb``: a disk is the size its base makes it, which a deploy @@ -238,7 +245,8 @@ def disk_size_refusal(self, form: str, disk_size_gb: float) -> Optional[str]: return None def build_statuses(self) -> dict[str, str]: - return {f"{form}/{arch}": result.get("status") for form, by_arch in self.metadata.get("composed", {}).items() + return {f"{form}/{arch}": f"{result.get('status')}, by another composer" if stale(result) else + result.get("status") for form, by_arch in self.metadata.get("composed", {}).items() for arch, result in by_arch.items()} async def deploy(self, ttl_seconds: int = 10800, disk_size_gb: float = 10, diff --git a/src/agent_env/env/envs/computer/utils/composed.py b/src/agent_env/env/envs/computer/utils/composed.py index 6c1d6539a..06bc039e2 100644 --- a/src/agent_env/env/envs/computer/utils/composed.py +++ b/src/agent_env/env/envs/computer/utils/composed.py @@ -1,12 +1,18 @@ -"""A computer env's ``composed`` metadata: each build's result by form and arch, recorded as it finishes.""" +"""A computer env's ``composed`` metadata: each build's result by form and arch, recorded as it finishes, with the +digest of the code that composed it, so a build an agent-env that composes differently made is composed again.""" from __future__ import annotations import copy +import functools +import hashlib import threading from datetime import datetime, timezone +from pathlib import Path from typing import TYPE_CHECKING +import agentenv_protocol.computer + from agent_env.artifact.artifacts.file import FileArtifact from agent_env.env.env import Env from agent_env.store.base import ConcurrentModificationError @@ -20,6 +26,11 @@ #: The image store repository composed computers are pushed to, tagged by their keys. REPOSITORY = "agentenv-computers" _WRITE_TRIES = 4 +#: The code that decides what a composed computer holds: the composer, the computers' build files, the provider that +#: composes VM bundles, and the protocol's computer package each layer installs. +_PACKAGE = Path(__file__).resolve().parents[4] +COMPOSING = (Path(__file__).parent, _PACKAGE / "computers", _PACKAGE / "providers" / "sandbox_providers" / "lume", + Path(agentenv_protocol.computer.__file__).parent) class Record: @@ -32,15 +43,15 @@ def __init__(self, env: ComputerEnv): def started(self, pairs: list[tuple[str, str]]) -> None: at = now() - self._merge({pair: {"status": "building", "started_at": at} for pair in pairs}) + self._merge({pair: {"status": "building", "started_at": at, "composer": composition()} for pair in pairs}) def finished(self, form: str, arch: str, result: dict) -> None: - self._merge({(form, arch): result}) + self._merge({(form, arch): {**result, "composer": composition()}}) def built(self, form: str = "container") -> dict[str, str]: - """Each arch's image of ``form``, of the pairs built or reused.""" + """Each arch's image of ``form``, of the pairs built or reused by this agent-env's composer.""" return {arch: result["image"] for arch, result in self.env.metadata.get("composed", {}).get(form, {}).items() - if result.get("status") in BUILT} + if result.get("status") in BUILT and not stale(result)} def computer(self, ref: dict | None) -> None: """Record the env's computer, with the images it was written over, the disks' and the VM bundles' apart.""" @@ -85,9 +96,31 @@ def now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") +def stale(result: dict | None) -> bool: + """Whether ``result`` was recorded by an agent-env whose composing code differs from this one's, or by one from + before results named it: what it holds may not be what this agent-env composes.""" + return result is not None and result.get("composer") != composition() + + +@functools.cache +def composition() -> str: + """The digest of ``COMPOSING``'s files but their docs, by their paths under each folder, bytes and whether each is + executable, as a build's key takes its context.""" + digest = hashlib.sha256() + for folder in COMPOSING: + for path in sorted(p for p in folder.rglob("*") if p.is_file() and p.suffix != ".md" + and "__pycache__" not in p.parts): + digest.update(f"{folder.name}/{path.relative_to(folder).as_posix()}\0".encode()) + digest.update(hashlib.sha256(path.read_bytes()).digest() + (b"x" if path.stat().st_mode & 0o100 else b"-")) + return "sha256:" + digest.hexdigest() + + def _rank(result: dict) -> int: """How far a pair got: ready, then built and either unprobed or failed at its probe, then anything short of an - image. A result takes a stored one's place only when it got at least as far.""" + image, and below that one another composer recorded. A result takes a stored one's place only when it got at least + as far.""" + if result and stale(result): + return -1 if result.get("status") == "ready": return 2 return 1 if result.get("status") == "unverified" or result.get("stage") == "probe" else 0 diff --git a/src/agent_env/env/envs/computer/utils/composer.py b/src/agent_env/env/envs/computer/utils/composer.py index 42dd18ff1..e1981b96b 100644 --- a/src/agent_env/env/envs/computer/utils/composer.py +++ b/src/agent_env/env/envs/computer/utils/composer.py @@ -38,7 +38,7 @@ from agent_env.config import get_config from agent_env.entity.export import Unwritable, checked_path from agent_env.env.envs.computer.utils import boot, builder, clashes, gate, macos_vm_bundle, packages, probing -from agent_env.env.envs.computer.utils.composed import BUILT, REPOSITORY, Record, failed, now +from agent_env.env.envs.computer.utils.composed import BUILT, REPOSITORY, Record, failed, now, stale from agent_env.env.envs.computer.utils.install import Install from agent_env.env.published_images import DISK, DISK_TOOLS_IMAGE, published_disk, published_pinned from agent_env.providers.sandbox_providers.host_images import platform_tag @@ -151,16 +151,16 @@ def current(env: ComputerEnv) -> bool: def pending(env: ComputerEnv, sandbox_provider: SandboxProvider | None = None, forms: list[str] | None = None) -> list[tuple[str, str]]: """The pairs of ``forms``, by default every form of the base, left to compose in ``env``: never started, failed, - left building by a put that stopped, built but unprobed though ``sandbox_provider``, by default the env's, runs the - arch natively, or a VM bundle the Mac it would compose on doesn't hold, which this agent-env composes and records - there, as one that composes differently names it.""" + left building by a put that stopped, recorded by an agent-env that composes differently, built but unprobed though + ``sandbox_provider``, by default the env's, runs the arch natively, or a VM bundle the Mac it would compose on + doesn't hold, which this agent-env composes and records there, as one that composes differently names it.""" from agent_env.providers import get_env_sandbox_provider provider = sandbox_provider or get_env_sandbox_provider() def left(form: str, arch: str, record: dict | None) -> bool: status = (record or {}).get("status") - if status not in BUILT: + if status not in BUILT or stale(record): return True if status == "unverified": return bool(boot.native_links(provider, arch, form, env.stored_id)) @@ -174,15 +174,16 @@ def left(form: str, arch: str, record: dict | None) -> bool: def unstarted(env: ComputerEnv, sandbox_provider: SandboxProvider, form: str | None = None) -> list[tuple[str, str]]: """The pairs a deploy on ``sandbox_provider`` boots that no compose has started, as a run for another sandbox - leaves them.""" - return _pairs(env, booted_forms(env, sandbox_provider, form), lambda form, arch, record: record is None) + leaves them, or that an agent-env that composes differently recorded.""" + return _pairs(env, booted_forms(env, sandbox_provider, form), + lambda form, arch, record: record is None or stale(record)) def unstarted_or_building(env: ComputerEnv, sandbox_provider: SandboxProvider, form: str | None = None) -> list[tuple[str, str]]: """Those, and the ones a compose is building now, which a deploy waits for.""" return _pairs(env, booted_forms(env, sandbox_provider, form), - lambda form, arch, record: record is None or record.get("status") == "building") + lambda form, arch, record: record is None or stale(record) or record.get("status") == "building") def booted_forms(env: ComputerEnv, sandbox_provider: SandboxProvider, form: str | None = None) -> list[str]: @@ -202,10 +203,11 @@ def _pairs(env: ComputerEnv, forms: list[str], left: Callable[[str, str, dict | def _with_containers(env: ComputerEnv, pairs: list[tuple[str, str]]) -> list[tuple[str, str]]: - """``pairs`` and the container each disk among them is made from, unless it's built, in the base's order.""" + """``pairs`` and the container each disk among them is made from, unless it's built by this agent-env's composer, in + the base's order.""" containers = env.metadata.get("composed", {}).get("container", {}) wanted = {*pairs, *(("container", arch) for form, arch in pairs if form == "disk" - and containers.get(arch, {}).get("status") not in BUILT)} + and (containers.get(arch, {}).get("status") not in BUILT or stale(containers.get(arch))))} return [(form, arch) for form in gate.forms(env.computer) for arch in env.composed_arches() if (form, arch) in wanted] diff --git a/src/agent_env/env/envs/computer/utils/probing.py b/src/agent_env/env/envs/computer/utils/probing.py index c96d5afe3..5362a72d8 100644 --- a/src/agent_env/env/envs/computer/utils/probing.py +++ b/src/agent_env/env/envs/computer/utils/probing.py @@ -38,11 +38,15 @@ async def probe_first_boot(env: ComputerEnv, form: str, arch: str, sandbox_provider: SandboxProvider) -> dict: """Gate 3 for a build its compose couldn't probe, on its first boot on ``sandbox_provider``, which boots it natively: the result, recorded. Deploys of one build at once, in this process or another, probe it once: each - waits for the one probing, then takes the result it recorded, a failure's too, which keeps the build's key.""" + waits for the one probing, then takes the result it recorded, a failure's too, which keeps the build's key. A record + another compose has taken over since, as an agent-env that composes differently does, stays that compose's: the + build this deploy picked is probed for it alone.""" build = env.metadata["composed"][form][arch] async with awaiting_lock(f"computer-first-boot:{env.stored_id}:{form}:{arch}:{build['image']}"): recorded = (await asyncio.to_thread(Env.get, env.stored_id, env.version)).metadata["composed"][form][arch] - if recorded.get("key") == build["key"] and recorded["status"] != "unverified": + if (recorded.get("key"), recorded.get("composer")) != (build["key"], build.get("composer")): + return await checked([sandbox_provider], env, env.driver, form, arch, build) + if recorded["status"] != "unverified": return recorded result = await checked([sandbox_provider], env, env.driver, form, arch, recorded) await asyncio.to_thread(Record(env).finished, form, arch, result) diff --git a/tst/unit/bundle/computer_env_folder_test.py b/tst/unit/bundle/computer_env_folder_test.py index 452e37e34..af0dcec40 100644 --- a/tst/unit/bundle/computer_env_folder_test.py +++ b/tst/unit/bundle/computer_env_folder_test.py @@ -15,6 +15,7 @@ from agent_env.bundle.run import dry_run_bundle from agent_env.env import Env from agent_env.env.envs.computer.utils import composer +from agent_env.env.envs.computer.utils.composed import composition from agent_env.env.envs.computer.utils.composer import Composing from agent_env.env.envs.computer.env import ComputerEnv from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandboxProvider @@ -107,7 +108,8 @@ def test_a_reused_computer_env_composes_what_it_has_left_on_the_runs_provider_an on_progress=lambda write, line: told.append((write.id, line))) again = materialize(_plan(root), build_here=local_builds(_plan(root)), composing=Composing(provider), on_progress=lambda write, line: told.append((write.id, line))) - built_ = {form: {arch: {"status": "ready", "image": f"{form}-{arch}"} for arch in ("amd64", "arm64")} + built_ = {form: {arch: {"status": "ready", "image": f"{form}-{arch}", "composer": composition()} + for arch in ("amd64", "arm64")} for form in ("container", "disk")} env.update_metadata({**env.metadata, "composed": built_, "computer": { "id": "c", "version": 1, "images": {"amd64": "container-amd64", "arm64": "container-arm64"}, diff --git a/tst/unit/env/envs/test_computer_disk.py b/tst/unit/env/envs/test_computer_disk.py index 101e3e60f..2a5464be3 100644 --- a/tst/unit/env/envs/test_computer_disk.py +++ b/tst/unit/env/envs/test_computer_disk.py @@ -324,6 +324,24 @@ async def counted(*args): assert all(record.metadata["computer"]["form"] == "disk" for record in deployed) +@pytest.mark.asyncio +async def test_a_first_boot_whose_record_another_compose_took_over_probes_the_build_it_picked_and_records_none( + disk_base, driver, docker, vms, monkeypatch): + """As a deploy by an agent-env that composes differently does, starting its own compose of the pair.""" + monkeypatch.setattr(local_sandbox, "kvm_platforms", lambda: frozenset()) + env = _put(disk_base, driver) + monkeypatch.setattr(local_sandbox, "kvm_platforms", lambda: frozenset({"linux/arm64"})) + picked, stored = Env.get(env.id), Env.get(env.id) + taken = {"status": "building", "started_at": "2026-11-05T16:00:00+00:00", "composer": "sha256:another-agent-env"} + composed = stored.metadata["composed"] + stored.update_metadata({**stored.metadata, "composed": {**composed, "disk": {**composed["disk"], "arm64": taken}}}) + + result = await probing.probe_first_boot(picked, "disk", "arm64", local_sandbox.LocalSandboxProvider()) + + assert (result["status"], result["image"]) == ("ready", picked.metadata["composed"]["disk"]["arm64"]["image"]) + assert Env.get(env.id).metadata["composed"]["disk"]["arm64"] == taken + + @pytest.mark.asyncio async def test_deploys_of_a_disk_no_compose_started_compose_it_once_first_reusing_the_containers(disk_base, driver, docker, vms): diff --git a/tst/unit/env/envs/test_computer_env.py b/tst/unit/env/envs/test_computer_env.py index cc66c6d82..e9476bc8a 100644 --- a/tst/unit/env/envs/test_computer_env.py +++ b/tst/unit/env/envs/test_computer_env.py @@ -393,10 +393,29 @@ def test_a_probe_ended_by_a_dropped_connection_runs_again_on_a_fresh_boot_and_on ({"status": "failed", "stage": "probe"}, {"status": "ready"}, False), ({"status": "ready"}, {"status": "failed", "stage": "probe"}, True), ({"status": "building"}, {"status": "failed", "stage": "build"}, True), + ({"status": "building"}, {"status": "ready", "composer": "sha256:another"}, True), + ({"status": "failed", "stage": "build"}, {"status": "ready", "composer": None}, True), ], ids=["unprobed-over-ready", "build-failure-over-built", "building-over-built", "probe-failure-over-unprobed", - "probe-failure-over-ready", "ready-over-probe-failure", "building-over-build-failure"]) + "probe-failure-over-ready", "ready-over-probe-failure", "building-over-build-failure", + "building-over-another-composers-ready", "build-failure-over-a-ready-no-composer-named"]) def test_a_result_takes_a_pairs_place_only_when_it_got_at_least_as_far(result, stored, replaces): - assert (composed._rank(result) >= composed._rank(stored)) is replaces + mine = {"composer": composed.composition()} + assert (composed._rank({**mine, **result}) >= composed._rank({**mine, **stored})) is replaces + + +def test_the_composers_digest_changes_with_a_files_bytes_and_with_its_executable_bit_alone(tmp_path, monkeypatch): + (tmp_path / "prepare.sh").write_text("echo hi\n") + (tmp_path / "README.md").write_text("docs") + monkeypatch.setattr(composed, "COMPOSING", (tmp_path,)) + digests = [composed.composition.__wrapped__()] + (tmp_path / "prepare.sh").chmod(0o755) + digests.append(composed.composition.__wrapped__()) + (tmp_path / "prepare.sh").write_text("echo bye\n") + digests.append(composed.composition.__wrapped__()) + (tmp_path / "README.md").write_text("other docs") + digests.append(composed.composition.__wrapped__()) + + assert len(set(digests[:3])) == 3 and digests[3] == digests[2] def test_env_build_builds_what_a_put_left_and_prints_each_pair(base, driver, docker, caplog): @@ -683,6 +702,25 @@ async def test_a_computer_env_deploys_its_first_native_ready_build_with_its_sett "computer_settings": {"timezone": "Asia/Tokyo"}, "driver_settings": {"default_space": "norm"}} +@pytest.mark.asyncio +async def test_a_deploy_composes_again_what_an_agent_env_that_composes_differently_built_and_boots_that( + base, driver, docker, boots, monkeypatch): + env = _put(base, driver, display={"width": 1600, "height": 900}) + monkeypatch.setattr(composed, "composition", lambda: "sha256:a-later-agent-env") + + stale = Env.get(env.id) + statuses, deployable = stale.build_statuses(), stale.deployable_builds() + await stale.deploy() + + (booted,) = boots.boots + record = Env.get(env.id).metadata["composed"]["container"] + assert statuses == {"container/amd64": "unverified, by another composer", + "container/arm64": "ready, by another composer"} and deployable == [] + assert {arch: (r["status"], r["composer"]) for arch, r in record.items()} == { + "amd64": ("unverified", "sha256:a-later-agent-env"), "arm64": ("ready", "sha256:a-later-agent-env")} + assert booted["image"] == record["arm64"]["image"] + + @pytest.mark.asyncio async def test_a_reset_ends_the_envs_recordings_then_boots_a_fresh_container_where_the_last_one_ran( base, driver, docker, boots, monkeypatch):