fix(computer): compose again a build another composer recorded before deploying it

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_<NAME>`, read as it starts:
 
 | Setting | Default | What it does |

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,

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

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]
 
 

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)

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"},

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):