fix(computers): a disk form's console shows the session's programs again, so releases publish

.github/scripts/infra_build.py

@@ -72,7 +72,7 @@
 _LOG_TAIL_LINES = 30  # of a container that never turned healthy, in the error
 # How long a disk form may take to boot to its desktop: seconds with KVM, minutes under TCG.
 _BOOTED_WITHIN_SECONDS = 900
-# What supervisord prints on a disk form's console as each program starts.
+# What supervisord prints on a disk form's console as each program starts, root's and the session's.
 _RUNNING = re.compile(r"success: (\S+) entered RUNNING")
 _PROGRAM = re.compile(r"^\[program:([^\]]+)\]", re.MULTILINE)
 _PRINTED_CHARS = 200  # of what docker printed in place of a manifest, in the error
@@ -415,9 +415,11 @@ def _smoke(name: str, image: str, launcher: str | None) -> int:
 
 def _boots(name: str, image: str, launcher: str) -> int:
     """Boot ``image``, a build of the disk form ``name``, in ``launcher``, with KVM where it can, and wait for every
-    program its base computer's supervisord.conf declares to run, which its console shows."""
-    programs = set(_PROGRAM.findall((REPO / Path(IMAGES[DISK_ARGS[name]["ROOTFS"]]).with_name("supervisord.conf"))
-                                    .read_text()))
+    program its base computer declares to run, which its console shows: root's supervisord's (root.conf) and the
+    session's it starts (supervisord.conf)."""
+    base = REPO / Path(IMAGES[DISK_ARGS[name]["ROOTFS"]])
+    programs = {program for conf in ("root.conf", "supervisord.conf")
+                for program in _PROGRAM.findall(base.with_name(conf).read_text())}
     arch = subprocess.run(["docker", "image", "inspect", "--format", "{{.Architecture}}", image], capture_output=True,
                           text=True, check=True).stdout.strip()
     machine = f"{image}-machine"

src/agent_env/computers/artifacts/ubuntu-24.04/root.conf

@@ -1,7 +1,8 @@
 ; Root's supervisord, which the base's init starts as root once it has made what the session must not touch: the clock
 ; follower and the egress proxy, each as its own user, on real time, and the session's supervisord (supervisord.conf)
 ; as the session user, which runs every session program. Its socket and logs are root's alone, so the session user can
-; neither stop nor restart any of them. The follower follows nothing until core writes its config, then picks it up
+; neither stop nor restart any of them. The session's supervisord logs to the console, as it did when it was the
+; computer's first process, so a disk form's console shows each session program start. The follower follows nothing until core writes its config, then picks it up
 ; within a quarter second. The egress proxy logs a JSON line per connection.
 [supervisord]
 nodaemon=true
@@ -52,5 +53,6 @@ priority=10
 autorestart=true
 stopwaitsecs=30
 redirect_stderr=true
-stdout_logfile=/var/log/agentenv-root/session.log
+stdout_logfile=/dev/stdout
+stdout_logfile_maxbytes=0
 environment=HOME="/home/user",USER="user",LOGNAME="user"

src/agent_env/computers/disk/linux/boot/init

@@ -64,6 +64,12 @@ mount -t devpts -o gid=5,mode=620,ptmxmode=666 devpts /newroot/dev/pts
 case "$AGENTENV_SHM_MB" in ''|0|*[!0-9]*) shm_mb=512 ;; *) shm_mb=$AGENTENV_SHM_MB ;; esac
 unset AGENTENV_SHM_MB
 mount -t tmpfs -o mode=1777,size="${shm_mb}m" shm /newroot/dev/shm
+# The links udev or a container runtime makes, which programs name their standard streams by: the computer's root
+# supervisord sends the session's output to the console through /dev/stdout, and a shell's <(...) reads /dev/fd.
+ln -sf /proc/self/fd /newroot/dev/fd
+ln -sf /proc/self/fd/0 /newroot/dev/stdin
+ln -sf /proc/self/fd/1 /newroot/dev/stdout
+ln -sf /proc/self/fd/2 /newroot/dev/stderr
 log "starting $* as $user ($uid:$gid) in $workdir"
 # Last, as busybox runs its applets through /proc/self/exe.
 mount --move /proc /newroot/proc

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

tst/unit/test_infra_build_script.py

@@ -5,6 +5,7 @@
 
 import ast
 import base64
+import configparser
 import json
 import subprocess
 import sys
@@ -323,8 +324,9 @@ def test_the_service_db_smoke_waits_for_its_health_check_and_removes_the_contain
                                                (True, "x86_64", "under TCG")], ids=["kvm", "no-kvm", "another-arch"])
 def test_a_disk_forms_smoke_boots_it_in_the_launcher_until_its_desktop_runs(script, monkeypatch, capsys, kvm, machine,
                                                                            how):
-    calls, logs = [], iter(["agentenv-init: starting", "success: xvfb entered RUNNING\nsuccess: dbus entered RUNNING\n"
-                                                      "success: desktop entered RUNNING"])
+    console = "".join(f"success: {program} entered RUNNING\n"
+                      for program in ("clock", "egress", "session", "xvfb", "dbus", "desktop"))
+    calls, logs = [], iter(["agentenv-init: starting", console])
 
     def run(command, **kw):
         calls.append((command, kw.get("input")))
@@ -347,15 +349,19 @@ def run(command, **kw):
     assert f"booted to its desktop {how}" in capsys.readouterr().out
 
 
-def test_a_disk_forms_smoke_waits_for_every_program_its_bases_supervisord_conf_declares(script, monkeypatch, capsys,
-                                                                                      tmp_path):
-    conf = tmp_path / Path(script.IMAGES["agentenv-ubuntu-24.04"]).with_name("supervisord.conf")
-    conf.parent.mkdir(parents=True)
-    conf.write_text("[supervisord]\nnodaemon=true\n\n[program:xvfb]\ncommand=Xvfb\n\n[program:recorder]\ncommand=rec\n")
+@pytest.mark.parametrize("missing", ["recorder", "session"])
+def test_a_disk_forms_smoke_waits_for_every_program_its_bases_root_and_session_supervisords_declare(
+        script, monkeypatch, capsys, tmp_path, missing):
+    base = tmp_path / Path(script.IMAGES["agentenv-ubuntu-24.04"])
+    base.parent.mkdir(parents=True)
+    base.with_name("root.conf").write_text("[supervisord]\nnodaemon=true\n\n[program:session]\ncommand=supervisord\n")
+    base.with_name("supervisord.conf").write_text("[supervisord]\nnodaemon=true\n\n[program:xvfb]\ncommand=Xvfb\n\n"
+                                                  "[program:recorder]\ncommand=rec\n")
     monkeypatch.setattr(script, "REPO", tmp_path)
     monkeypatch.setattr(script, "_BOOTED_WITHIN_SECONDS", 0.01)
     monkeypatch.setattr(script.time, "sleep", lambda seconds: None)
-    console = "success: xvfb entered RUNNING\nsuccess: dbus entered RUNNING\nsuccess: desktop entered RUNNING\n"
+    console = "".join(f"success: {program} entered RUNNING\n" for program in ("session", "xvfb", "recorder")
+                      if program != missing)
     monkeypatch.setattr(script.subprocess, "run", lambda command, **kw: subprocess.CompletedProcess(
         command, 0, {"image": "amd64\n", "run": "c1\n", "logs": console}.get(command[1], ""), ""))
 
@@ -364,6 +370,17 @@ def test_a_disk_forms_smoke_waits_for_every_program_its_bases_supervisord_conf_d
     assert "::error::disk:1 didn't boot to its desktop within 0.01s" in capsys.readouterr().out
 
 
+def test_the_sessions_supervisord_logs_to_the_console_where_a_disk_forms_smoke_reads_its_programs_start(script):
+    root = configparser.ConfigParser(interpolation=None)
+    root.read_string((script.REPO / Path(script.IMAGES["agentenv-ubuntu-24.04"]).with_name("root.conf")).read_text())
+
+    assert (root["program:session"]["stdout_logfile"], root["program:session"]["stdout_logfile_maxbytes"]) == (
+        "/dev/stdout", "0")
+    assert root["supervisord"]["nodaemon"] == "true"
+    init = (script.REPO / Path(script.IMAGES["agentenv-ubuntu-24.04-disk"]).with_name("init")).read_text()
+    assert "ln -sf /proc/self/fd/1 /newroot/dev/stdout\n" in init
+
+
 def test_a_disk_forms_smoke_needs_the_launcher_to_boot_it_in(script, monkeypatch, capsys):
     with pytest.raises(SystemExit) as exited:
         _run(script, monkeypatch, "smoke", "agentenv-ubuntu-24.04-disk", "disk:1")