fix(computers): two composes of one Linux disk on a machine take turns, so a boot's pinned disk stays put

src/agent_env/env/envs/computer/utils/composer.py

@@ -677,7 +677,9 @@ def _built_disk(env: ComputerEnv, base: ComputerArtifact, arch: str, container:
     """The status of the disk form of ``arch`` once built from ``container``, its container's result, and pushed, or
     reused, or the stage that failed and why. Its key is of the container's key, the base disk and the disk recipe;
     the base disk is the base's own, the one published for it, or else one made on the builder from the base's
-    container and the boot recipe. Its build runs no program of ``arch``'s but that recipe's."""
+    container and the boot recipe. Its build runs no program of ``arch``'s but that recipe's. Two composes of one disk
+    on this machine take turns, so the second reuses the first's: built again, its tag would move off the image the
+    first pinned, which a boot mounts by that name and Docker doesn't pull for a mount."""
     pair, platform = f"disk/{arch}", f"linux/{arch}"
     with (_attempting(env, "disk", arch) as attempt,
           tempfile.TemporaryDirectory(prefix="agentenv-compose-") as work, contextlib.ExitStack() as opened):
@@ -687,34 +689,36 @@ def _built_disk(env: ComputerEnv, base: ComputerArtifact, arch: str, container:
         base_disk, base_identity, recipe = *_base_disk(base, arch), _digest(context)
         key = attempt.key = _key(container=container["key"], base_disk=base_identity, recipe=recipe, form="disk",
                                  arch=arch, format=FORMAT)
-        store, ref, pinned = _stored(env, key)
-        if pinned is not None:
-            progress.at(pair, "reusing", "reusing the disk of the same key, made earlier")
-            return {"status": "unverified", "key": key, "image": pinned, "reused": True, "built_at": now()}
-        link = builds.link(platform, base.image())
-        progress.at(pair, "starting its builder", f"making the disk {builder.where(link)}")
-        on = opened.enter_context(builds.open(link, platform, Path(work), base.image(), {}))
-        on.login(store.auth(container["image"]))
-        if base_disk is not None:
-            on.login(get_config().get_image_store().auth(base_disk))
-        if (tools := published_pinned(DISK_TOOLS_IMAGE)) is None:
-            tools = f"agentenv-disk-tools:{DISK_TOOLS_IMAGE.tag()}"
-            on.build((context / "tools" / "Dockerfile").read_text(), tools, on.platform, {},
-                     context=f"{builder.CONTEXT}/tools")
-        dockerfile = (context / "Dockerfile").read_text()
-        if base_disk is None:
-            progress.at(pair, "making the base's disk")
-            base_disk = _made_base_disk(on, base, arch, f"{base_identity} {recipe}", tools, dockerfile)
-        progress.at(pair, "making the disk")
-        local = f"agentenv-computer:{key[7:23]}-disk-{arch}"
-        on.build(dockerfile, local, platform, {KEY_LABEL: key}, build_args={
-            "ROOTFS": container["image"], "ROOTFS_CONFIG": _encoded(_config(on, container["image"], arch)),
-            "ESP": base_disk, "TOOLS": tools})
-        attempt.stage = "push"
-        progress.at(pair, "pushing the disk")
-        on.push(local, ref, store)
-        return {"status": "unverified", "key": key, "image": pin_digest(ref, store.auth(ref)), "reused": False,
-                "builder": on.name, "built_at": now()}
+        with holding_locks([f"linux-disk:{key}"], lambda: progress.at(
+                pair, "waiting", "waiting for the compose of the same disk under way on this machine")):
+            store, ref, pinned = _stored(env, key)
+            if pinned is not None:
+                progress.at(pair, "reusing", "reusing the disk of the same key, made earlier")
+                return {"status": "unverified", "key": key, "image": pinned, "reused": True, "built_at": now()}
+            link = builds.link(platform, base.image())
+            progress.at(pair, "starting its builder", f"making the disk {builder.where(link)}")
+            on = opened.enter_context(builds.open(link, platform, Path(work), base.image(), {}))
+            on.login(store.auth(container["image"]))
+            if base_disk is not None:
+                on.login(get_config().get_image_store().auth(base_disk))
+            if (tools := published_pinned(DISK_TOOLS_IMAGE)) is None:
+                tools = f"agentenv-disk-tools:{DISK_TOOLS_IMAGE.tag()}"
+                on.build((context / "tools" / "Dockerfile").read_text(), tools, on.platform, {},
+                         context=f"{builder.CONTEXT}/tools")
+            dockerfile = (context / "Dockerfile").read_text()
+            if base_disk is None:
+                progress.at(pair, "making the base's disk")
+                base_disk = _made_base_disk(on, base, arch, f"{base_identity} {recipe}", tools, dockerfile)
+            progress.at(pair, "making the disk")
+            local = f"agentenv-computer:{key[7:23]}-disk-{arch}"
+            on.build(dockerfile, local, platform, {KEY_LABEL: key}, build_args={
+                "ROOTFS": container["image"], "ROOTFS_CONFIG": _encoded(_config(on, container["image"], arch)),
+                "ESP": base_disk, "TOOLS": tools})
+            attempt.stage = "push"
+            progress.at(pair, "pushing the disk")
+            on.push(local, ref, store)
+            return {"status": "unverified", "key": key, "image": pin_digest(ref, store.auth(ref)), "reused": False,
+                    "builder": on.name, "built_at": now()}
     return attempt.result
 
 

tst/integration/conftest.py

@@ -0,0 +1,32 @@
+"""The integration tier's setup shared by every test."""
+
+import os
+import tempfile
+from pathlib import Path
+
+import pytest
+
+from agent_env.store import local_state
+
+#: The locks that guard this machine's Docker and image registry, by the start of their ids: a run's tests keep stores
+#: and state roots of their own, but all of them, on every pytest worker, use the one Docker and registry.
+MACHINE_LOCKS = ("computer-compose:", "linux-disk:", "windows-disk:", "omarchy-disk:", "disk-export:", "image-build:",
+                 "image-pull:", "image-save:")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def machine_locks_shared_by_the_runs_workers():
+    """Each test's state root, where agent-env keeps its locks, is its own, so two pytest workers composing one
+    computer took no lock in common and both built it under the same tags: the second moved them off the image the
+    first had pinned, or their pushes collided. The locks guarding this machine's Docker and registry are kept in one
+    folder for the run, as one user's processes keep theirs, so such work takes turns and the second reuses the first's.
+    """
+    run = os.environ.get("PYTEST_XDIST_TESTRUNUID") or str(os.getpid())
+    folder = Path(tempfile.gettempdir()) / f"agent-env-machine-locks-{run}"
+    folder.mkdir(mode=0o700, exist_ok=True)
+    per_test = local_state._lock_path
+    patch = pytest.MonkeyPatch()
+    patch.setattr(local_state, "_lock_path",
+                  lambda id: folder / per_test(id).name if id.startswith(MACHINE_LOCKS) else per_test(id))
+    yield
+    patch.undo()

tst/unit/env/envs/test_computer_disk.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import base64
+import contextlib
 import json
 import threading
 from datetime import datetime
@@ -22,7 +23,7 @@
 from agent_env.env import Env
 from agent_env.config.runtime import Config
 from agent_env.env.envs.computer.env import ComputerEnv
-from agent_env.env.envs.computer.utils import boot, composer, probing
+from agent_env.env.envs.computer.utils import boot, builder, composer, probing
 from agent_env.env.envs.computer.utils.composer import Composing
 from agent_env.env.published_images import DISK
 from agent_env import providers
@@ -89,6 +90,46 @@ def test_a_disk_is_reused_by_its_key_of_its_containers_key_the_base_disk_and_the
     assert len(docker.disk_builds) > made
 
 
+def test_a_disk_is_looked_up_made_and_pushed_holding_a_lock_of_its_key_so_composes_of_it_take_turns(
+        disk_base, driver, docker, monkeypatch):
+    held, events = set(), []
+    real_locks, real_stored = composer.holding_locks, composer._stored
+    real_build, real_push = docker.build_image, docker.push
+
+    @contextlib.contextmanager
+    def holding(ids, on_wait=None):
+        with real_locks(ids, on_wait):
+            held.update(ids)
+            try:
+                yield
+            finally:
+                held.difference_update(ids)
+
+    def stored(env, key):
+        events.append(("looked up", key, set(held)))
+        return real_stored(env, key)
+
+    def build(dockerfile, context, tag, **kw):
+        events.append(("made", tag, set(held)))
+        return real_build(dockerfile, context, tag, **kw)
+
+    def push(tag, ref, store):
+        events.append(("pushed", tag, set(held)))
+        return real_push(tag, ref, store)
+
+    monkeypatch.setattr(composer, "holding_locks", holding)
+    monkeypatch.setattr(composer, "_stored", stored)
+    monkeypatch.setattr(builder, "build_image", build)
+    monkeypatch.setattr(builder, "push_local_image", push)
+
+    disks = Env.get(_put(disk_base, driver).id).metadata["composed"]["disk"]
+
+    for arch, result in disks.items():
+        names = (result["key"], f"agentenv-computer:{result['key'][7:23]}-disk-{arch}")
+        assert [(what, f"linux-disk:{result['key']}" in locks) for what, name, locks in events if name in names] == [
+            ("looked up", True), ("made", True), ("pushed", True)]
+
+
 def test_a_bases_disk_size_is_in_the_key_and_the_build_of_the_disk_made_for_it(disk_base, driver, docker):
     bigger = _base(disk_base.image(), id="@local/~/kit/bigger", forms={
         **disk_base.forms.model_dump(exclude_none=True), "disk": {**DISK_FORM, "min_disk_gb": 32}})