feat(computer): a boot stops when its sandbox does and says what it last wrote

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

@@ -8,7 +8,9 @@
 import contextlib
 import json
 import logging
+import re
 import secrets
+import signal
 from collections.abc import Iterable
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Optional
@@ -44,12 +46,28 @@
 _POLL_S = 0.5
 #: How long one read of the driver's card may take.
 _CARD_TIMEOUT_S = 5
+#: How often a boot asks whether the computer's sandbox still runs it.
+_EXIT_POLL_S = 2.0
+#: How many of the last lines the computer's sandbox wrote a boot reads, and how long that may take.
+_OUTPUT_LINES, _OUTPUT_TIMEOUT_S = 200, 20
+#: How many of them a boot's error names: the last that say why, and the last of all.
+_REASON_LINES, _LAST_LINES = 3, 5
+#: How the lines a disk's guest init and its launcher say why with begin (``computers/disk/linux``).
+_REASONS = ("agentenv-init:", "launch:", "launch reset:")
+_ESCAPE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")
 #: How a setting a record keeps shows a secret's or a generated one's value, which it never holds.
 WITHHELD = "<withheld>"
 
 
 class ComputerFailed(RuntimeError):
     """The computer's ``computer-ready`` status says it failed for good, a check that can't pass again."""
+
+
+class ComputerStopped(ComputerFailed):
+    """The computer's sandbox stopped running it before it was ready: its container, or the launcher booting its disk,
+    exited."""
+
+
 #: How many random bytes a computer's control token carries: 256 bits, past guessing.
 CONTROL_TOKEN_BYTES = 32
 #: The size of a disk's VM when a deploy sets none: what the desktop was measured booting and passing the probe in; for
@@ -201,7 +219,7 @@ async def boot(sandbox_provider: SandboxProvider, *, image: str, port: int, envi
             **shm_kwargs(sandbox_provider, shm_mb))
     try:
         url = sandbox.tunnel_urls[port]
-        return Booted(sandbox, url, await ready(url, ready_s, token, answering=held))
+        return Booted(sandbox, url, await ready(url, ready_s, token, answering=held, sandbox=sandbox))
     except BaseException:
         await discard(sandbox)
         raise
@@ -226,16 +244,19 @@ def ready_timeout(apps: list[ComputerAppEnv], platform: str) -> float:
                                    if (launch := app.launch(platform)) and launch.at == "boot" and launch.ready)])
 
 
-async def ready(url: str, timeout: float, token: Optional[str] = None, *, answering: bool = False) -> dict:
+async def ready(url: str, timeout: float, token: Optional[str] = None, *, answering: bool = False,
+                sandbox: Optional[Sandbox] = None) -> dict:
     """The card of the driver at ``url``, read once its ``computer-ready`` status, asked with its control ``token``,
     says it's ready, so each app it fronts is on it as the app's own server serves it, or with ``answering``, once it
     answers at all: a computer whose boot apps' launch is held can't be ready before they launch, as a page driver's
-    screen is its browser's page. Raises ComputerFailed as soon as the status says it failed, and TimeoutError, naming
-    what it last saw, when it isn't ready within ``timeout`` seconds."""
-    last, failed = "nothing answered", None
+    screen is its browser's page. Raises ComputerFailed as soon as the status says it failed, ComputerStopped within
+    seconds of ``sandbox``, the sandbox running the computer, no longer running it, and TimeoutError, naming what it
+    last saw, when it isn't ready within ``timeout`` seconds; each error ends with what ``sandbox`` last wrote."""
+    last, failed, stopped, asked = "nothing answered", None, None, None
+    loop = asyncio.get_running_loop()
     try:
         async with asyncio.timeout(timeout):
-            while failed is None:
+            while failed is None and stopped is None:
                 try:
                     card = await protocol_v1.get_card(url, timeout=_CARD_TIMEOUT_S)
                     status = await protocol_v1.invoke_extension(url, card, COMPUTER_READY, None, _STATUS_TIMEOUT_S,
@@ -248,11 +269,59 @@ async def ready(url: str, timeout: float, token: Optional[str] = None, *, answer
                                                      for check in status.get("checks", []) if not check.get("ok"))
                 except Exception as e:
                     last = f"{type(e).__name__}: {e}"
-                if failed is None:
+                if failed is None and sandbox is not None and (asked is None or loop.time() - asked >= _EXIT_POLL_S):
+                    asked = loop.time()
+                    stopped = await _exit_status(sandbox)
+                if failed is None and stopped is None:
                     await asyncio.sleep(_POLL_S)
     except TimeoutError:
-        raise TimeoutError(f"the computer at {url} wasn't ready within {timeout:.0f} s ({last})") from None
-    raise ComputerFailed(f"the computer at {url} failed: {failed}")
+        raise TimeoutError(f"the computer at {url} wasn't ready within {timeout:.0f} s ({last})"
+                           f"{await _last_written(sandbox)}") from None
+    if stopped is not None:
+        raise ComputerStopped(f"the computer at {url} stopped before it was ready, exiting with {_described(stopped)}"
+                              f"{await _last_written(sandbox)}")
+    raise ComputerFailed(f"the computer at {url} failed: {failed}{await _last_written(sandbox)}")
+
+
+async def _exit_status(sandbox: Sandbox) -> Optional[int]:
+    """``sandbox.exit_status()``, None when asking fails, which the next ask may not."""
+    try:
+        return await sandbox.exit_status()
+    except Exception as e:
+        logger.debug("Couldn't ask whether sandbox %s still runs its computer: %s", sandbox.sandbox_id, e)
+        return None
+
+
+async def _last_written(sandbox: Optional[Sandbox]) -> str:
+    """What an error adds of the lines ``sandbox`` last wrote (``said_why``); nothing when there are none, it can't
+    read them, or reading them takes past ``_OUTPUT_TIMEOUT_S``."""
+    if sandbox is None:
+        return ""
+    try:
+        async with asyncio.timeout(_OUTPUT_TIMEOUT_S):
+            lines = said_why(await sandbox.output_tail(_OUTPUT_LINES))
+    except Exception as e:
+        logger.info("Couldn't read what sandbox %s last wrote: %s", sandbox.sandbox_id, e)
+        return ""
+    return "; its sandbox last wrote:\n" + "\n".join(f"  {line}" for line in lines) if lines else ""
+
+
+def said_why(lines: list[str]) -> list[str]:
+    """Of ``lines``, the last a sandbox wrote, each without its terminal escapes and surrounding space, in their order:
+    the last few a disk's guest init and its launcher say why with, and the last few that aren't blank."""
+    cleaned = [text for line in lines if (text := _ESCAPE.sub("", line).strip())]
+    reasons = [index for index, line in enumerate(cleaned) if line.startswith(_REASONS)][-_REASON_LINES:]
+    return [cleaned[index] for index in sorted({*reasons, *range(max(0, len(cleaned) - _LAST_LINES), len(cleaned))})]
+
+
+def _described(status: int) -> str:
+    """``status``, a process's exit status, naming the signal that ended it when one did, as a shell's 128 plus its
+    number, or Python's minus it, says."""
+    number = status - 128 if status > 128 else -status
+    with contextlib.suppress(ValueError):
+        if number > 0:
+            return f"{status} ({signal.Signals(number).name})"
+    return str(status)
 
 
 async def listed(url: str, apps: list[str], timeout: float) -> dict:

src/agent_env/providers/env_providers/env_computer_provider.py

@@ -236,7 +236,7 @@ async def reboot(self, env: ComputerEnv, deployed: DeployedSandboxEnv) -> None:
         except NotImplementedError as e:  # a reset the env supports, which this provider can't carry out
             raise RuntimeError(f"'{env.id}' can't be reset on {type(sandbox_provider).__name__}: {e}") from None
         card = await ready(deployed.environment_url, ready_timeout(env.composed_apps(), platform_of(env.computer.os)),
-                           deployed.control_token, answering=bool(params.site_routing))
+                           deployed.control_token, answering=bool(params.site_routing), sandbox=sandbox)
         _require_name(env, card)
         valid_for_s = (record.get("sites") or {}).get("valid_for_s", 0)
         if routed := await _finish_boot(env, deployed, sandbox, params, valid_for_s, values.get("resolution")):

src/agent_env/providers/sandbox_providers/AGENTS.md

@@ -102,6 +102,7 @@ sandboxes off this machine, runs containers' images by name, has no private netw
 | `platform` | the image platform it runs; `linux/amd64` by default, set another when the provider creates it |
 | `exec(*argv)` | returns an object with async `stdout.read()` and `stderr.read()` (bytes) and async `wait()` (exit code). If the platform takes one command string, build it with `shlex.join(argv)` |
 | `terminate()` | abstract; idempotent: terminating a sandbox that is already gone is not an error |
+| `exit_status()`, `output_tail(lines)` | optional: the exit status of the process `create_container` or a disk's `create_vm` made the sandbox run, once it has exited, else `None`; and the last `lines` lines it wrote, a disk's console among them. A computer's boot stops waiting within seconds of that process exiting and puts its last lines in its error. The defaults, `None` and `[]`, leave a boot waiting out its readiness timeout. Local Docker reads its container's; `modal` and `modal_vm` their sandbox's process, its output only once it has exited |
 
 Everything else is built on `exec`. `exec_script` runs `sudo bash -c <script>`, so the exec user needs passwordless sudo.
 When the transport drops without an exit status, return exit code `-1`: callers retry idempotent commands on `-1` and on

src/agent_env/providers/sandbox_providers/local_docker/provider.py

@@ -308,6 +308,33 @@ def owns_container(self) -> bool:
         """Whether create_container started ``container_name`` for this sandbox, not an agent placed on it."""
         return (self._work_dir / _CONTAINER_MODE_MARKER).exists()
 
+    async def exit_status(self) -> Optional[int]:
+        """The exit status of the container it was made to run, a computer's or the launcher booting a disk, once it
+        has exited; None while it runs, when it runs none, or when Docker doesn't say."""
+        if not self.owns_container:
+            return None
+        try:
+            state = await asyncio.to_thread(docker, "container", "inspect", "--format",
+                                            "{{.State.Status}} {{.State.ExitCode}}", self.container_name,
+                                            timeout=_DOCKER_QUERY_SECONDS)
+        except (DockerFailed, subprocess.TimeoutExpired):
+            return None
+        status, _, code = state.partition(" ")
+        return int(code) if status in ("exited", "dead") else None
+
+    async def output_tail(self, lines: int) -> list[str]:
+        """The last ``lines`` lines that container wrote, its output's and its errors' together as Docker logged them,
+        with no order kept between the two; none when it runs none, or when Docker doesn't say."""
+        if not self.owns_container:
+            return []
+        try:
+            done = await asyncio.to_thread(subprocess.run, ["docker", "logs", "--tail", str(lines), self.container_name],
+                                           stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                           timeout=_DOCKER_QUERY_SECONDS)
+        except subprocess.TimeoutExpired:
+            return []
+        return done.stdout.decode(errors="replace").splitlines() if done.returncode == 0 else []
+
     @property
     def work_dir(self) -> Path:
         return self._work_dir

src/agent_env/providers/sandbox_providers/modal_sandbox.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import base64
+import collections
 import contextlib
 import functools
 import hashlib
@@ -59,6 +60,10 @@ def _resolve_app_base_name(app_name: Optional[str]) -> str:
 _CANCELLED_CREATE_WAIT_SECONDS = 30
 # Terminations started after their create's cancel had gone on, kept so they run to their end.
 _LATE_TERMINATIONS: set[asyncio.Task] = set()
+# How long reading what an ended sandbox's process wrote may take, and how many of the last lines of each of its
+# streams it keeps.
+_OUTPUT_READ_SECONDS, _OUTPUT_KEPT_LINES = 15, 1000
+_LINE_BREAK = re.compile(r"\r\n|\r|\n")
 
 
 async def _terminate_created(sb: modal.Sandbox) -> None:
@@ -126,6 +131,47 @@ async def _resolve_i6pn_address(sb: modal.Sandbox) -> Optional[str]:
     return None
 
 
+async def _ended_output(sb: modal.Sandbox) -> Optional[tuple[list[str], list[str]]]:
+    """The last ``_OUTPUT_KEPT_LINES`` lines of each of the output and the errors of ``sb``'s process, once it has
+    exited, read a chunk at a time so no more than those are held; None while it runs, as Modal hands a sandbox's
+    stream only to its end. Reading that fails or takes past ``_OUTPUT_READ_SECONDS`` keeps what it read."""
+    if await sb.poll.aio() is None:
+        return None
+    output, errors = collections.deque(maxlen=_OUTPUT_KEPT_LINES), collections.deque(maxlen=_OUTPUT_KEPT_LINES)
+    try:
+        async with asyncio.timeout(_OUTPUT_READ_SECONDS):
+            failures = await asyncio.gather(_read_lines(sb.stdout, output), _read_lines(sb.stderr, errors),
+                                            return_exceptions=True)
+    except TimeoutError:
+        failures = [TimeoutError(f"reading took past {_OUTPUT_READ_SECONDS} s")]
+    for failure in failures:
+        if failure is not None:
+            logger.info(f"Couldn't read all the output of Modal sandbox {sb.object_id}: {fmt_exc(failure)}")
+    return list(output), list(errors)
+
+
+async def _read_lines(stream, kept: collections.deque) -> None:
+    """Each line of ``stream``, a sandbox's, into ``kept``, a chunk at a time; blank ones left out."""
+    partial = ""
+    async for chunk in stream:
+        *whole, partial = _LINE_BREAK.split(partial + chunk)
+        kept.extend(line for line in whole if line)
+    if partial:
+        kept.append(partial)
+
+
+def _last_lines(ended: Optional[tuple[list[str], list[str]]], lines: int) -> list[str]:
+    """The last ``lines`` of an ended sandbox's output and errors, ``ended``, its output's then its errors'. Modal keeps
+    no order between the two, so each keeps at least half of them, or all it has, and neither crowds out the other's
+    last lines."""
+    if ended is None:
+        return []
+    output, errors = ended
+    from_errors = min(len(errors), max(lines // 2, lines - len(output)))
+    from_output = min(len(output), lines - from_errors)
+    return output[len(output) - from_output:] + errors[len(errors) - from_errors:]
+
+
 class _ModalStreamAdapter:
     def __init__(self, stream):
         self._stream = stream
@@ -156,6 +202,7 @@ def __init__(self, sb: modal.Sandbox, tunnel_urls: dict[int, str], i6pn_address:
         self.vnc_url = None
         self.mode = SANDBOX_MODE_CONTAINER
         self.network_policy = network_policy
+        self._ended: Optional[tuple[list[str], list[str]]] = None
 
     @property
     def private_host(self) -> str | None:
@@ -165,6 +212,15 @@ def private_host(self) -> str | None:
     async def terminate(self) -> None:
         await self._sb.terminate.aio()
 
+    async def exit_status(self) -> Optional[int]:
+        return await self._sb.poll.aio()
+
+    async def output_tail(self, lines: int) -> list[str]:
+        """The last ``lines`` lines its container wrote, once it has exited; none while it runs."""
+        if self._ended is None:
+            self._ended = await _ended_output(self._sb)
+        return _last_lines(self._ended, lines)
+
     async def exec(self, *command: str):
         process = await self._sb.exec.aio(*command, text=False)
         return _ModalProcessAdapter(process)

src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py

@@ -36,6 +36,8 @@
 from agent_env.providers.sandbox_providers.modal_sandbox import (
     _ModalProcessAdapter,
     _attribution_tags,
+    _ended_output,
+    _last_lines,
     _log_sandbox_started,
     _resolve_app_base_name,
     _terminate_by_id,
@@ -150,10 +152,21 @@ def __init__(self, sb: modal.Sandbox, tunnel_urls: dict[int, str],
         self.mode = SANDBOX_MODE_VM
         self.network_policy = network_policy
         self._download_slots = asyncio.Semaphore(max(1, _dl_setting("PARALLEL_FILES", _DL_PARALLEL_FILES)))
+        self._ended: Optional[tuple[list[str], list[str]]] = None
 
     async def terminate(self) -> None:
         await self._sb.terminate.aio()
 
+    async def exit_status(self) -> Optional[int]:
+        return await self._sb.poll.aio()
+
+    async def output_tail(self, lines: int) -> list[str]:
+        """The last ``lines`` lines its main process wrote, a disk's launcher with its guest's console, once it has
+        exited; none while it runs."""
+        if self._ended is None:
+            self._ended = await _ended_output(self._sb)
+        return _last_lines(self._ended, lines)
+
     async def exec(self, *command: str):
         # Modal VM sandboxes run as root and have no `sudo`; the inherited VmSandbox
         # helpers prefix commands with "sudo" (sandbox.py). Strip it, exactly like

src/agent_env/providers/sandbox_providers/sandbox.py

@@ -211,6 +211,16 @@ def scoped_name(self, name: str) -> str:
     async def terminate(self) -> None:
         """Terminate the sandbox."""
 
+    async def exit_status(self) -> Optional[int]:
+        """The exit status of the process the sandbox was made to run, a computer's container or the launcher booting
+        its disk, once that has exited; None while it runs, or when its provider can't tell, as by default."""
+        return None
+
+    async def output_tail(self, lines: int) -> list[str]:
+        """The last ``lines`` lines that process wrote, a disk's console among them; none when its provider can't read
+        them, as by default."""
+        return []
+
     async def exec(self, *command: str) -> Any:
         """Execute a command in the sandbox. Returns a ContainerProcess-like object with stdout, stderr, wait()."""
         raise NotImplementedError(f"{self.__class__.__name__} does not support exec")

tst/integration/computer/conftest.py

@@ -4,12 +4,16 @@
 composed once for a module's tests;
 ``state``, local stores for a test that runs a bundle; each of these stores starts with the packages the run has resolved
 (``tst.util.package_cache``); ``docker``, which runs the docker CLI for them; and
-``cdp_clients``, Playwright and Puppeteer installed to connect to a CDP address."""
+``cdp_clients``, Playwright and Puppeteer installed to connect to a CDP address. A compose's build of an image waits for
+another test worker's build of the same tag."""
 
+import fcntl
+import hashlib
 import logging
 import re
 import shutil
 import subprocess
+import tempfile
 import uuid
 from pathlib import Path
 
@@ -24,6 +28,7 @@
 from agent_env.bundle.resolve import BuiltImage, resolve_bundle
 from agent_env.config import configure, reset_config
 from agent_env.env import Env
+from agent_env.env.envs.computer.utils import builder
 from agent_env.providers.sandbox_providers.host_images import host_platform
 from agent_env.store.object_store.local.grant_server import grant_server
 from agent_env.store.routing import namespace_routing
@@ -45,6 +50,23 @@
 CHROMIUM_ALONE = "@local/agentenv-framework/browsers/chromium"
 
 
+@pytest.fixture(scope="session", autouse=True)
+def one_build_of_a_tag_at_a_time():
+    """Test workers share one Docker daemon, and two composes of the same computer in stores of their own build the
+    same tag, which fails with "image ... already exists" when both builds finish at once, so they take turns."""
+    built = builder.build_image
+
+    def build_image(dockerfile, context, tag, **kwargs):
+        lock = Path(tempfile.gettempdir()) / f"agentenv-test-build-{hashlib.sha256(tag.encode()).hexdigest()[:16]}.lock"
+        with open(lock, "w") as held:
+            fcntl.flock(held, fcntl.LOCK_EX)
+            return built(dockerfile, context, tag, **kwargs)
+
+    with pytest.MonkeyPatch.context() as patch:
+        patch.setattr(builder, "build_image", build_image)
+        yield
+
+
 @pytest.fixture(scope="session")
 def base():
     """The ubuntu-24.04 base for this machine's platform: the image agent-env published from these sources, pulled, or