From 05f93c0829780fe6654a9184d6bc4d9a2b301778 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 11 Nov 2026 14:41:52 -0800
Subject: [PATCH 1/4] feat(computer): a boot stops when its sandbox does and
 says what it last wrote

A computer's boot asks the sandbox running it whether its container, or the
launcher booting its disk, still runs. Within seconds of it exiting, the boot
fails with ComputerStopped, naming the exit status and the sandbox's last lines:
the guest init's agentenv-init: and the launcher's launch: lines, then the last
few. A boot that times out ends its error with those lines too. Sandbox gains
optional exit_status() and output_tail(), defaulting to nothing, which local
Docker answers from its container and Modal from its sandbox's process.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/agent_env/env/envs/computer/utils/boot.py |  87 ++++++++--
 .../env_providers/env_computer_provider.py    |   2 +-
 .../providers/sandbox_providers/AGENTS.md     |   1 +
 .../local_docker/provider.py                  |  27 ++++
 .../sandbox_providers/modal_sandbox.py        |  27 ++++
 .../sandbox_providers/modal_vm_sandbox.py     |  12 ++
 .../providers/sandbox_providers/sandbox.py    |  10 ++
 .../computer/container_stopped_test.py        |  53 ++++++
 .../disk_boot_failure_modal_vm_test.py        |  40 +++++
 tst/integration/computer/ghost_disk.py        |  46 ++++++
 tst/integration/computer/launcher_test.py     |  46 +++++-
 tst/unit/env/envs/test_computer_boot_stops.py | 151 ++++++++++++++++++
 tst/unit/env/envs/test_computer_disk.py       |   2 +-
 tst/unit/env/envs/test_computer_env.py        |   4 +-
 .../sandbox_providers/sandbox_exit_test.py    | 102 ++++++++++++
 15 files changed, 594 insertions(+), 16 deletions(-)
 create mode 100644 tst/integration/computer/container_stopped_test.py
 create mode 100644 tst/integration/computer/disk_boot_failure_modal_vm_test.py
 create mode 100644 tst/integration/computer/ghost_disk.py
 create mode 100644 tst/unit/env/envs/test_computer_boot_stops.py
 create mode 100644 tst/unit/providers/sandbox_providers/sandbox_exit_test.py

diff --git a/src/agent_env/env/envs/computer/utils/boot.py b/src/agent_env/env/envs/computer/utils/boot.py
index 4192b5f94..37f0188a5 100644
--- a/src/agent_env/env/envs/computer/utils/boot.py
+++ b/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:
diff --git a/src/agent_env/providers/env_providers/env_computer_provider.py b/src/agent_env/providers/env_providers/env_computer_provider.py
index 1322c06ba..929fdff5d 100644
--- a/src/agent_env/providers/env_providers/env_computer_provider.py
+++ b/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")):
diff --git a/src/agent_env/providers/sandbox_providers/AGENTS.md b/src/agent_env/providers/sandbox_providers/AGENTS.md
index 4b45b4fe3..4727eaff5 100644
--- a/src/agent_env/providers/sandbox_providers/AGENTS.md
+++ b/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
diff --git a/src/agent_env/providers/sandbox_providers/local_docker/provider.py b/src/agent_env/providers/sandbox_providers/local_docker/provider.py
index e9007c4a1..f8b20301e 100644
--- a/src/agent_env/providers/sandbox_providers/local_docker/provider.py
+++ b/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
diff --git a/src/agent_env/providers/sandbox_providers/modal_sandbox.py b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
index ac52f2d32..211469963 100644
--- a/src/agent_env/providers/sandbox_providers/modal_sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
@@ -59,6 +59,8 @@ 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.
+_OUTPUT_READ_SECONDS = 15
 
 
 async def _terminate_created(sb: modal.Sandbox) -> None:
@@ -126,6 +128,21 @@ async def _resolve_i6pn_address(sb: modal.Sandbox) -> Optional[str]:
     return None
 
 
+async def _ended_output(sb: modal.Sandbox) -> Optional[list[str]]:
+    """The lines ``sb``'s process wrote, its output's and then its errors', once it has exited; None while it runs, as
+    Modal hands a sandbox's stream only whole, and none when reading them fails or takes past
+    ``_OUTPUT_READ_SECONDS``."""
+    if await sb.poll.aio() is None:
+        return None
+    try:
+        async with asyncio.timeout(_OUTPUT_READ_SECONDS):
+            stdout, stderr = await asyncio.gather(sb.stdout.read.aio(), sb.stderr.read.aio())
+    except Exception as e:
+        logger.info(f"Couldn't read the output of Modal sandbox {sb.object_id}: {fmt_exc(e)}")
+        return []
+    return [*stdout.splitlines(), *stderr.splitlines()]
+
+
 class _ModalStreamAdapter:
     def __init__(self, stream):
         self._stream = stream
@@ -156,6 +173,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[list[str]] = None
 
     @property
     def private_host(self) -> str | None:
@@ -165,6 +183,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 (self._ended or [])[-lines:]
+
     async def exec(self, *command: str):
         process = await self._sb.exec.aio(*command, text=False)
         return _ModalProcessAdapter(process)
diff --git a/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py b/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py
index 99312ab53..61083da07 100644
--- a/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py
@@ -36,6 +36,7 @@
 from agent_env.providers.sandbox_providers.modal_sandbox import (
     _ModalProcessAdapter,
     _attribution_tags,
+    _ended_output,
     _log_sandbox_started,
     _resolve_app_base_name,
     _terminate_by_id,
@@ -150,10 +151,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[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 (self._ended or [])[-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
diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index 642bfe4db..addb4c409 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/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")
diff --git a/tst/integration/computer/container_stopped_test.py b/tst/integration/computer/container_stopped_test.py
new file mode 100644
index 000000000..3faa163ec
--- /dev/null
+++ b/tst/integration/computer/container_stopped_test.py
@@ -0,0 +1,53 @@
+"""A computer's container that exits before its driver answers, booted on this machine's Docker: the boot fails within
+seconds, naming the container's exit status and the last lines it wrote, rather than at the readiness timeout, and
+takes the container down. Needs Docker and the local registry."""
+
+import subprocess
+import time
+import uuid
+
+import pytest
+
+from agent_env.env.envs.computer.utils import boot
+from agent_env.providers.sandbox_providers.host_images import host_platform
+from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandboxProvider
+from agent_env.utils.docker.build import build_image
+from tst.integration.computer.ghost_disk import ROOTFS
+
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def exiting(tmp_path):
+    busybox = f"agentenv-test-disk:{uuid.uuid4().hex[:12]}"
+    image = f"localhost:5000/agentenv-test-exiting:{uuid.uuid4().hex[:12]}"
+    build_image(ROOTFS / "Dockerfile", ROOTFS, busybox, platform=host_platform())
+    (tmp_path / "Dockerfile").write_text(f'FROM {busybox}\nCMD ["sh", "-c", "echo starting the desktop; echo Xvfb: no '
+                                         'screens found >&2; exit 4"]\n')
+    build_image(tmp_path / "Dockerfile", tmp_path, image, platform=host_platform())
+    subprocess.run(["docker", "push", "--quiet", image], capture_output=True, check=True)
+    yield image
+    subprocess.run(["docker", "rmi", "-f", busybox, image], capture_output=True)
+
+
+class Recording(LocalSandboxProvider):
+    async def create_container(self, **kwargs):
+        self.made = await super().create_container(**kwargs)
+        return self.made
+
+
+@pytest.mark.asyncio
+async def test_a_computers_container_that_exits_fails_its_boot_at_once_naming_its_status_and_last_lines(exiting):
+    provider, started = Recording(), time.monotonic()
+
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await boot.boot(provider, image=exiting, port=18765, environment={}, token=boot.control_token(),
+                        network_policy=None, timeout=600, attribution=None)
+
+    assert time.monotonic() - started < 30
+    said, _, written = str(stopped.value).partition("; its sandbox last wrote:\n")
+    assert said.endswith("stopped before it was ready, exiting with 4")
+    assert sorted(written.splitlines()) == ["  Xvfb: no screens found", "  starting the desktop"]
+    left = subprocess.run(["docker", "ps", "-aq", "--filter", f"name=^{provider.made.container_name}$"],
+                          capture_output=True, text=True, check=True).stdout
+    assert left == ""
diff --git a/tst/integration/computer/disk_boot_failure_modal_vm_test.py b/tst/integration/computer/disk_boot_failure_modal_vm_test.py
new file mode 100644
index 000000000..c7b6cf856
--- /dev/null
+++ b/tst/integration/computer/disk_boot_failure_modal_vm_test.py
@@ -0,0 +1,40 @@
+"""A disk its guest's init refuses, booted on a Modal VM sandbox under nested virtualization: the boot fails once the
+launcher booting it exits, rather than at the readiness timeout, naming what the guest's console last said, as on this
+machine's Docker; and a launcher that can't forward the driver's port fails the boot as soon as it starts. Needs Docker,
+the local registry and Modal credentials, with ``[sandbox.attribution]`` tags for Modal's apps."""
+
+import pytest
+
+from agent_env.env.envs.computer.utils import boot
+from agent_env.providers import build_sandbox_provider
+from tst.integration.computer.ghost_disk import REFUSED, ghost_disk
+from tst.util.capabilities import skip_without_remote_sandbox
+
+pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow, skip_without_remote_sandbox("modal_vm")]
+
+
+@pytest.fixture(scope="module")
+def ghost(desktop, tmp_path_factory):
+    yield from ghost_disk(desktop.metadata["composed"]["disk"]["amd64"]["image"], tmp_path_factory.mktemp("ghost"))
+
+
+@pytest.mark.asyncio
+async def test_a_disk_its_guests_init_refuses_fails_its_boot_on_a_modal_vm_once_the_launcher_exits(desktop, ghost):
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await boot.boot(build_sandbox_provider("modal_vm"), image=ghost, port=desktop.driver.port, environment={},
+                        token=boot.control_token(), network_policy=None, timeout=1800, attribution=None, form="disk")
+
+    assert "stopped before it was ready, exiting with 0; its sandbox last wrote:\n" in str(stopped.value)
+    assert f"  {REFUSED}" in str(stopped.value).splitlines()
+
+
+@pytest.mark.asyncio
+async def test_a_launcher_that_cant_forward_the_drivers_port_fails_the_boot_on_a_modal_vm_saying_why(desktop):
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await boot.boot(build_sandbox_provider("modal_vm"), image=desktop.metadata["composed"]["disk"]["amd64"]["image"],
+                        port=60000, environment={}, token=boot.control_token(), network_policy=None, timeout=1800,
+                        attribution=None, form="disk")
+
+    assert str(stopped.value).endswith(
+        "exiting with 2; its sandbox last wrote:\n  launch: port 60000 can't be forwarded: QEMU's forward behind its "
+        "proxy would take port 70000, which is above 65535")
diff --git a/tst/integration/computer/ghost_disk.py b/tst/integration/computer/ghost_disk.py
new file mode 100644
index 000000000..c7925442c
--- /dev/null
+++ b/tst/integration/computer/ghost_disk.py
@@ -0,0 +1,46 @@
+"""A disk its guest's init refuses to start, made as a compose makes a disk: a busybox root filesystem whose image runs
+as a user its /etc/passwd lacks, beside the EFI system partition of a disk the boot recipe made, in the local
+registry."""
+
+import base64
+import subprocess
+import uuid
+from collections.abc import Iterator
+from pathlib import Path
+
+from agent_env.env.published_images import DISK
+from agent_env.providers.sandbox_providers.host_images import host_platform
+from agent_env.utils.docker.build import build_image
+from tst.util.published_image import pull_published_else_build
+
+ROOTFS = Path(__file__).parent / "disk" / "rootfs"
+#: What the guest's init says of it.
+REFUSED = "agentenv-init: the image's user ghost isn't in /etc/passwd"
+
+
+def ghost_disk(esp: str, folder: Path) -> Iterator[str]:
+    """The disk, its EFI system partition ``esp``'s, the image of a disk the boot recipe made, for this machine's
+    platform, pushed to the local registry under a tag of its own; every image it made is removed after."""
+    platform, made = host_platform(), []
+    try:
+        tools = _tag(made, "agentenv-test-disk-tools")
+        pull_published_else_build("disk tools", DISK / "tools" / "Dockerfile", DISK / "tools", tools, platform=platform)
+        busybox = _tag(made, "agentenv-test-disk")
+        build_image(ROOTFS / "Dockerfile", ROOTFS, busybox, platform=platform)
+        (folder / "Dockerfile").write_text(f"FROM {busybox}\nUSER ghost\n")
+        root = _tag(made, "agentenv-test-disk")
+        build_image(folder / "Dockerfile", folder, root, platform=platform)
+        config = subprocess.run(["docker", "image", "inspect", "--format", "{{json .Config}}", root],
+                                capture_output=True, text=True, check=True).stdout.strip()
+        disk = _tag(made, "localhost:5000/agentenv-test-ghost-disk")
+        build_image(DISK / "boot" / "Dockerfile", DISK / "boot", disk, platform=platform, build_args={
+            "ROOTFS": root, "TOOLS": tools, "ROOTFS_CONFIG": base64.b64encode(config.encode()).decode(), "ESP": esp})
+        subprocess.run(["docker", "push", "--quiet", disk], capture_output=True, check=True)
+        yield disk
+    finally:
+        subprocess.run(["docker", "rmi", "-f", *made], capture_output=True)
+
+
+def _tag(made: list[str], repository: str) -> str:
+    made.append(f"{repository}:{uuid.uuid4().hex[:12]}")
+    return made[-1]
diff --git a/tst/integration/computer/launcher_test.py b/tst/integration/computer/launcher_test.py
index c5b7050ae..3bc2796fc 100644
--- a/tst/integration/computer/launcher_test.py
+++ b/tst/integration/computer/launcher_test.py
@@ -1,9 +1,12 @@
 """The QEMU launcher booting the shipped desktop's disk on this machine's Docker with KVM: its driver's port served
 through socat, whose listen backlog takes a burst of new connections that QEMU's own forward, with a backlog of 1,
-would drop, and a reset booting the disk afresh in the same container, ready again at the same URL. Needs Docker, the
-local registry and /dev/kvm."""
+would drop, and a reset booting the disk afresh in the same container, ready again at the same URL. A boot that ends,
+the guest's init refusing its disk or the launcher a port it can't forward, fails within seconds of the launcher
+exiting, naming what its console last said, rather than at the readiness timeout. Needs Docker, the local registry and
+/dev/kvm."""
 
 import asyncio
+import time
 
 import httpx
 import pytest
@@ -12,6 +15,7 @@
 from agent_env.env.gateway.constants import WELL_KNOWN_PATH
 from agent_env.providers.sandbox_providers.host_images import host_platform
 from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandboxProvider
+from tst.integration.computer.ghost_disk import REFUSED, ghost_disk
 from tst.util.capabilities import skip_without_kvm
 
 pytestmark = [pytest.mark.integration, pytest.mark.int_test_slow, skip_without_kvm()]
@@ -23,7 +27,7 @@
 
 @pytest.mark.asyncio
 async def test_the_launcher_takes_a_burst_of_new_connections_and_a_reset_boots_its_disk_afresh_at_its_url(desktop):
-    image = desktop.metadata["composed"]["disk"][host_platform().rsplit("/", 1)[1]]["image"]
+    image = _disk(desktop)
     provider, token = LocalSandboxProvider(), boot.control_token()
     environment = boot.settings_environment(desktop.computer, desktop.computer_values(),
                                             environment_name=desktop.environment_name)
@@ -39,6 +43,42 @@ async def test_the_launcher_takes_a_burst_of_new_connections_and_a_reset_boots_i
     assert statuses == [200] * BURST and again["name"] == desktop.environment_name
 
 
+@pytest.fixture(scope="module")
+def ghost(desktop, tmp_path_factory):
+    yield from ghost_disk(_disk(desktop), tmp_path_factory.mktemp("ghost"))
+
+
+@pytest.mark.asyncio
+async def test_a_disk_its_guests_init_refuses_fails_its_boot_within_seconds_of_powering_off_saying_why(desktop, ghost):
+    started = time.monotonic()
+
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await boot.boot(LocalSandboxProvider(), image=ghost, port=desktop.driver.port, environment={},
+                        token=boot.control_token(), network_policy=None, timeout=1800, attribution=None, form="disk")
+
+    assert time.monotonic() - started < 90
+    assert "stopped before it was ready, exiting with 0; its sandbox last wrote:\n" in str(stopped.value)
+    assert f"  {REFUSED}" in str(stopped.value).splitlines()
+
+
+@pytest.mark.asyncio
+async def test_a_launcher_that_cant_forward_the_drivers_port_fails_the_boot_at_once_saying_why(desktop):
+    started = time.monotonic()
+
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await boot.boot(LocalSandboxProvider(), image=_disk(desktop), port=60000, environment={},
+                        token=boot.control_token(), network_policy=None, timeout=1800, attribution=None, form="disk")
+
+    assert time.monotonic() - started < 30
+    assert str(stopped.value).endswith(
+        "exiting with 2; its sandbox last wrote:\n  launch: port 60000 can't be forwarded: QEMU's forward behind its "
+        "proxy would take port 70000, which is above 65535")
+
+
+def _disk(desktop) -> str:
+    return desktop.metadata["composed"]["disk"][host_platform().rsplit("/", 1)[1]]["image"]
+
+
 async def _status(url: str) -> int:
     async with httpx.AsyncClient(timeout=60) as client:
         return (await client.get(url)).status_code
diff --git a/tst/unit/env/envs/test_computer_boot_stops.py b/tst/unit/env/envs/test_computer_boot_stops.py
new file mode 100644
index 000000000..4ba655fad
--- /dev/null
+++ b/tst/unit/env/envs/test_computer_boot_stops.py
@@ -0,0 +1,151 @@
+"""A computer's boot when its sandbox stops running it: it fails within seconds of the sandbox's container, or the
+launcher booting its disk, exiting, rather than at its readiness timeout, naming the exit status and what the sandbox
+last wrote, its guest init's and launcher's reasons first; one never ready says what it last wrote too; and a sandbox
+whose provider can't tell fails as before."""
+
+import asyncio
+
+import httpx
+import pytest
+
+from agent_env.env.envs.computer.utils import boot
+from agent_env.providers.sandbox_providers.sandbox import Sandbox
+
+URL = "http://computer"
+CONSOLE = ["[    0.912] EXT4-fs (vda2): mounted filesystem", "agentenv-init: no DHCP lease on eth0",
+           "\x1b[0;31magentenv-init: the image's user ghost isn't in /etc/passwd\x1b[0m\r",
+           "[    3.120] reboot: Power down", ""]
+
+
+class Running(Sandbox):
+    """A sandbox that runs its computer for ``runs`` asks, then exits with ``status``, having written ``written``."""
+
+    def __init__(self, status=None, written=(), runs=0):
+        self.sandbox_id, self.tunnel_urls, self.status, self.written, self.runs = "sb-1", {}, status, list(written), runs
+        self.asked = 0
+
+    async def terminate(self):
+        pass
+
+    async def exit_status(self):
+        self.asked += 1
+        return self.status if self.asked > self.runs else None
+
+    async def output_tail(self, lines):
+        return self.written[-lines:]
+
+
+class Unasked(Sandbox):
+    sandbox_id, tunnel_urls = "sb-2", {}
+
+    async def terminate(self):
+        pass
+
+
+@pytest.fixture(autouse=True)
+def unanswered(monkeypatch):
+    async def get_card(url, timeout=5):
+        raise httpx.ConnectError("Connection refused")
+
+    monkeypatch.setattr(boot.protocol_v1, "get_card", get_card)
+    monkeypatch.setattr(boot, "_POLL_S", 0.01)
+    monkeypatch.setattr(boot, "_EXIT_POLL_S", 0.05)
+
+
+@pytest.mark.asyncio
+async def test_a_disk_whose_guest_init_fails_stops_the_boot_within_seconds_with_its_reason_and_last_lines():
+    sandbox = Running(status=0, written=CONSOLE, runs=2)
+
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await asyncio.wait_for(boot.ready(URL, 180, "t0ken", sandbox=sandbox), 5)
+
+    assert str(stopped.value) == (
+        "the computer at http://computer stopped before it was ready, exiting with 0; its sandbox last wrote:\n"
+        "  [    0.912] EXT4-fs (vda2): mounted filesystem\n"
+        "  agentenv-init: no DHCP lease on eth0\n"
+        "  agentenv-init: the image's user ghost isn't in /etc/passwd\n"
+        "  [    3.120] reboot: Power down")
+    assert isinstance(stopped.value, boot.ComputerFailed) and sandbox.asked == 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("status", "said"), [
+    (3, "exiting with 3; its sandbox last wrote:\n  launch: KVM can't run this arm64 disk on this amd64 machine"),
+    (137, "exiting with 137 (SIGKILL); its sandbox last wrote:\n  launch: KVM can't run this arm64 disk"),
+    (-9, "exiting with -9 (SIGKILL)"),
+    (2, "exiting with 2"),
+], ids=["refused", "killed", "killed-as-python-says", "plain"])
+async def test_a_launcher_that_exits_names_its_status_and_a_signal_that_ended_it(status, said):
+    sandbox = Running(status=status, written=["launch: KVM can't run this arm64 disk on this amd64 machine"])
+
+    with pytest.raises(boot.ComputerStopped) as stopped:
+        await boot.ready(URL, 180, sandbox=sandbox)
+
+    assert said in str(stopped.value)
+
+
+@pytest.mark.asyncio
+async def test_a_computer_never_ready_says_what_its_sandbox_last_wrote_at_its_timeout():
+    console = [f"[  {n}.0] systemd[1]: Started unit {n}." for n in range(40)]
+
+    with pytest.raises(TimeoutError) as timed_out:
+        await boot.ready(URL, 0.3, sandbox=Running(written=console))
+
+    assert str(timed_out.value) == (
+        "the computer at http://computer wasn't ready within 0 s (ConnectError: Connection refused); its sandbox last "
+        "wrote:\n" + "\n".join(f"  [  {n}.0] systemd[1]: Started unit {n}." for n in range(35, 40)))
+
+
+@pytest.mark.asyncio
+async def test_a_sandbox_whose_provider_cant_tell_waits_out_the_timeout_and_says_what_it_did_before():
+    with pytest.raises(TimeoutError) as timed_out:
+        await boot.ready(URL, 0.3, sandbox=Unasked())
+
+    assert str(timed_out.value) == ("the computer at http://computer wasn't ready within 0 s (ConnectError: Connection "
+                                    "refused)")
+
+
+@pytest.mark.asyncio
+async def test_a_sandbox_that_cant_be_asked_or_read_keeps_the_boot_waiting_then_times_out_as_before():
+    class Unreachable(Running):
+        async def exit_status(self):
+            self.asked += 1
+            raise RuntimeError("docker: Cannot connect to the Docker daemon")
+
+        async def output_tail(self, lines):
+            raise RuntimeError("docker: Cannot connect to the Docker daemon")
+
+    sandbox = Unreachable()
+
+    with pytest.raises(TimeoutError, match=r"\(ConnectError: Connection refused\)$"):
+        await boot.ready(URL, 0.3, sandbox=sandbox)
+
+    assert sandbox.asked > 1
+
+
+@pytest.mark.asyncio
+async def test_a_computer_that_says_it_failed_adds_what_its_sandbox_last_wrote(monkeypatch):
+    async def get_card(url, timeout=5):
+        return {"name": "desk"}
+
+    async def status(url, card, uri, params, timeout, method, headers):
+        return {"ready": False, "failed": "display: Xvfb exited"}
+
+    monkeypatch.setattr(boot.protocol_v1, "get_card", get_card)
+    monkeypatch.setattr(boot.protocol_v1, "invoke_extension", status)
+
+    with pytest.raises(boot.ComputerFailed) as failed:
+        await boot.ready(URL, 5, sandbox=Running(written=["Xvfb: no screens found"]))
+
+    assert str(failed.value) == ("the computer at http://computer failed: display: Xvfb exited; its sandbox last wrote:\n"
+                                 "  Xvfb: no screens found")
+    assert not isinstance(failed.value, boot.ComputerStopped)
+
+
+def test_what_a_boot_names_of_its_sandboxs_last_lines_is_its_reasons_and_its_last_lines_in_order():
+    lines = ["launch: the proxy in front of a forwarded port exited (1), so the computer is stopped",
+             *(f"agentenv-init: line {n}" for n in range(5)), *(f"kernel {n}" for n in range(10)), "  ", "\x1b[2K"]
+
+    assert boot.said_why(lines) == ["agentenv-init: line 2", "agentenv-init: line 3", "agentenv-init: line 4",
+                                    "kernel 5", "kernel 6", "kernel 7", "kernel 8", "kernel 9"]
+    assert boot.said_why(["only", "a few"]) == ["only", "a few"] and boot.said_why([]) == []
diff --git a/tst/unit/env/envs/test_computer_disk.py b/tst/unit/env/envs/test_computer_disk.py
index ec3fc8936..daff46e01 100644
--- a/tst/unit/env/envs/test_computer_disk.py
+++ b/tst/unit/env/envs/test_computer_disk.py
@@ -245,7 +245,7 @@ async def reset_vm(self, sandbox, **kwargs):
     monkeypatch.setattr(local_sandbox.LocalSandboxProvider, "create_vm", create_vm)
     monkeypatch.setattr(local_sandbox.LocalSandboxProvider, "reset_vm", reset_vm)
     monkeypatch.setattr(local_sandbox, "kvm_platforms", lambda: frozenset({"linux/arm64"}))
-    monkeypatch.setattr(boot, "ready", AsyncMock(side_effect=lambda url, timeout, token, answering=False: {
+    monkeypatch.setattr(boot, "ready", AsyncMock(side_effect=lambda url, timeout, token, answering=False, sandbox=None: {
         **DRIVER_CARD, "name": "desktop"}))
     monkeypatch.setattr(env_computer_provider, "ready", boot.ready)
     monkeypatch.setattr(env_computer_provider, "_require_display", AsyncMock())
diff --git a/tst/unit/env/envs/test_computer_env.py b/tst/unit/env/envs/test_computer_env.py
index 704360b3b..0e60904e6 100644
--- a/tst/unit/env/envs/test_computer_env.py
+++ b/tst/unit/env/envs/test_computer_env.py
@@ -926,7 +926,7 @@ async def reset_container(self, sandbox, **kwargs):
     assert (kwargs["env"]["AGENTENV_SETTING_TIMEZONE"], kwargs["env"]["ENVIRONMENT_NAME"]) == ("Asia/Tokyo", "desktop")
     assert kwargs["env"][CONTROL_TOKEN_ENV] == deployed.control_token
     env_computer_provider.ready.assert_awaited_once_with(deployed.environment_url, boot.READY_TIMEOUT_S,
-                                                         deployed.control_token, answering=False)
+                                                         deployed.control_token, answering=False, sandbox=sandbox)
     assert boots.calls == [(COMPUTER_RECORDING, "end_all", {"reason": "reset"}), "reset_container", (COMPUTER_DISPLAY, "get", None)]
 
 
@@ -1563,7 +1563,7 @@ async def test_a_boot_hands_the_computer_its_token_and_one_never_ready_takes_its
                         attribution=None)
 
     assert create.await_args.kwargs["env"] == {"ENVIRONMENT_NAME": "desk", CONTROL_TOKEN_ENV: "t0ken"}
-    assert boot.ready.await_args.args[2] == "t0ken"
+    assert boot.ready.await_args.args[2] == "t0ken" and boot.ready.await_args.kwargs["sandbox"] is sandbox
     sandbox.terminate.assert_awaited_once()
     assert local_sandbox.LocalSandbox.find_work_dir("local-unready") is None
 
diff --git a/tst/unit/providers/sandbox_providers/sandbox_exit_test.py b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
new file mode 100644
index 000000000..0c85c0233
--- /dev/null
+++ b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
@@ -0,0 +1,102 @@
+"""Whether a sandbox still runs the process it was made to run, and what that process last wrote, as a computer's boot
+asks: a local sandbox's container from Docker, its output and errors as Docker logged them; a Modal sandbox's process
+from Modal, its output once it has exited, read once; and nothing from a sandbox whose provider can't tell."""
+
+import subprocess
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import agent_env.providers.sandbox_providers.local_docker.provider as ls
+from agent_env.providers.sandbox_providers.modal_sandbox import ModalSandbox
+from agent_env.providers.sandbox_providers.modal_vm_sandbox import ModalVmSandbox
+
+
+def _local(tmp_path, owns: bool = True) -> ls.LocalSandbox:
+    sandbox = ls.LocalSandbox(sandbox_id="local-disk", work_dir=tmp_path)
+    if owns:
+        (tmp_path / ls._CONTAINER_MODE_MARKER).write_text(sandbox.container_name)
+    return sandbox
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("state", "status"), [("exited 3", 3), ("dead 137", 137), ("running 0", None),
+                                                ("created 0", None)])
+async def test_a_local_sandbox_says_its_container_exited_with_dockers_exit_code(tmp_path, monkeypatch, state, status):
+    asked = []
+    monkeypatch.setattr(ls, "docker", lambda *args, timeout=None: asked.append(args) or state)
+
+    assert await _local(tmp_path).exit_status() == status
+    assert asked == [("container", "inspect", "--format", "{{.State.Status}} {{.State.ExitCode}}",
+                      "agent-local-disk")]
+
+
+@pytest.mark.asyncio
+async def test_a_local_sandbox_reads_its_containers_last_lines_its_output_and_errors_as_docker_logged_them(
+        tmp_path, monkeypatch):
+    ran = []
+
+    def run(args, **kwargs):
+        ran.append((args, kwargs))
+        return SimpleNamespace(returncode=0, stdout=b"agentenv-init: can't mount /dev/vda2\r\nreboot: Power down\r\n")
+
+    monkeypatch.setattr(ls.subprocess, "run", run)
+
+    assert await _local(tmp_path).output_tail(200) == ["agentenv-init: can't mount /dev/vda2", "reboot: Power down"]
+    ((args, kwargs),) = ran
+    assert args == ["docker", "logs", "--tail", "200", "agent-local-disk"]
+    assert (kwargs["stdout"], kwargs["stderr"]) == (subprocess.PIPE, subprocess.STDOUT)
+
+
+@pytest.mark.asyncio
+async def test_a_local_sandbox_says_nothing_of_a_container_it_doesnt_run_or_docker_wont_say(tmp_path, monkeypatch):
+    def failing(*args, timeout=None):
+        raise ls.DockerFailed("docker container inspect failed: No such container", 1)
+
+    monkeypatch.setattr(ls, "docker", failing)
+    monkeypatch.setattr(ls.subprocess, "run", lambda args, **kwargs: SimpleNamespace(returncode=1, stdout=b"Error"))
+    (tmp_path / "theirs").mkdir()
+
+    assert await _local(tmp_path).exit_status() is None and await _local(tmp_path).output_tail(5) == []
+    vm = _local(tmp_path / "theirs", owns=False)
+    monkeypatch.setattr(ls, "docker", MagicMock(side_effect=AssertionError("asked Docker of no container")))
+    assert await vm.exit_status() is None and await vm.output_tail(5) == []
+
+
+def _modal_sb(status, stdout="", stderr=""):
+    sb = MagicMock()
+    sb.object_id = "sb-1"
+    sb.poll.aio = AsyncMock(return_value=status)
+    sb.stdout.read.aio = AsyncMock(return_value=stdout)
+    sb.stderr.read.aio = AsyncMock(return_value=stderr)
+    return sb
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("made", [lambda sb: ModalVmSandbox(sb, {}), lambda sb: ModalSandbox(sb, {})],
+                         ids=["modal_vm", "modal"])
+async def test_a_modal_sandbox_reads_its_process_output_once_it_has_exited_and_keeps_it(made):
+    running = _modal_sb(None)
+
+    assert await made(running).exit_status() is None and await made(running).output_tail(5) == []
+    running.stdout.read.aio.assert_not_awaited()
+
+    ended = _modal_sb(2, stdout="[    1.0] Linux version 6.8\nagentenv-init: no filesystem labelled agentenv-root\n",
+                      stderr="launch: the guest powered off\n")
+    sandbox = made(ended)
+    assert await sandbox.exit_status() == 2
+    assert await sandbox.output_tail(5) == ["[    1.0] Linux version 6.8",
+                                            "agentenv-init: no filesystem labelled agentenv-root",
+                                            "launch: the guest powered off"]
+    assert await sandbox.output_tail(1) == ["launch: the guest powered off"]
+    ended.stdout.read.aio.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_a_modal_sandbox_whose_output_cant_be_read_gives_none(caplog):
+    sb = _modal_sb(2)
+    sb.stdout.read.aio = AsyncMock(side_effect=RuntimeError("stream gone"))
+
+    assert await ModalVmSandbox(sb, {}).output_tail(5) == []
+    assert "Couldn't read the output of Modal sandbox sb-1" in caplog.text

From 8fb19353731099fe0484f62b31bd164fab9db69e Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 11 Nov 2026 15:46:16 -0800
Subject: [PATCH 2/4] fix(computer): read a Modal sandbox's ended output into a
 bounded tail of each stream

The output a failed boot reads of a Modal sandbox is read a chunk at a time,
keeping the last lines of its output and of its errors apart, and its tail
gives each stream at least half of the lines asked for, so a long log isn't
held whole and the guest's console isn't crowded out by the launcher's errors.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../sandbox_providers/modal_sandbox.py        | 47 +++++++++----
 .../sandbox_providers/modal_vm_sandbox.py     |  5 +-
 .../sandbox_providers/sandbox_exit_test.py    | 66 +++++++++++++++----
 3 files changed, 91 insertions(+), 27 deletions(-)

diff --git a/src/agent_env/providers/sandbox_providers/modal_sandbox.py b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
index 211469963..121394189 100644
--- a/src/agent_env/providers/sandbox_providers/modal_sandbox.py
+++ b/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,8 +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.
-_OUTPUT_READ_SECONDS = 15
+# 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:
@@ -128,19 +131,41 @@ async def _resolve_i6pn_address(sb: modal.Sandbox) -> Optional[str]:
     return None
 
 
-async def _ended_output(sb: modal.Sandbox) -> Optional[list[str]]:
-    """The lines ``sb``'s process wrote, its output's and then its errors', once it has exited; None while it runs, as
-    Modal hands a sandbox's stream only whole, and none when reading them fails or takes past
-    ``_OUTPUT_READ_SECONDS``."""
+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):
-            stdout, stderr = await asyncio.gather(sb.stdout.read.aio(), sb.stderr.read.aio())
+            await asyncio.gather(_read_lines(sb.stdout, output), _read_lines(sb.stderr, errors))
     except Exception as e:
-        logger.info(f"Couldn't read the output of Modal sandbox {sb.object_id}: {fmt_exc(e)}")
+        logger.info(f"Couldn't read all the output of Modal sandbox {sb.object_id}: {fmt_exc(e)}")
+    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 []
-    return [*stdout.splitlines(), *stderr.splitlines()]
+    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:
@@ -173,7 +198,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[list[str]] = None
+        self._ended: Optional[tuple[list[str], list[str]]] = None
 
     @property
     def private_host(self) -> str | None:
@@ -190,7 +215,7 @@ 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 (self._ended or [])[-lines:]
+        return _last_lines(self._ended, lines)
 
     async def exec(self, *command: str):
         process = await self._sb.exec.aio(*command, text=False)
diff --git a/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py b/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py
index 61083da07..f54d90f1a 100644
--- a/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py
@@ -37,6 +37,7 @@
     _ModalProcessAdapter,
     _attribution_tags,
     _ended_output,
+    _last_lines,
     _log_sandbox_started,
     _resolve_app_base_name,
     _terminate_by_id,
@@ -151,7 +152,7 @@ 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[list[str]] = None
+        self._ended: Optional[tuple[list[str], list[str]]] = None
 
     async def terminate(self) -> None:
         await self._sb.terminate.aio()
@@ -164,7 +165,7 @@ async def output_tail(self, lines: int) -> list[str]:
         exited; none while it runs."""
         if self._ended is None:
             self._ended = await _ended_output(self._sb)
-        return (self._ended or [])[-lines:]
+        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
diff --git a/tst/unit/providers/sandbox_providers/sandbox_exit_test.py b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
index 0c85c0233..6032abcbe 100644
--- a/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
+++ b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
@@ -1,6 +1,7 @@
 """Whether a sandbox still runs the process it was made to run, and what that process last wrote, as a computer's boot
 asks: a local sandbox's container from Docker, its output and errors as Docker logged them; a Modal sandbox's process
-from Modal, its output once it has exited, read once; and nothing from a sandbox whose provider can't tell."""
+from Modal, its output once it has exited, read once, a chunk at a time, into a bounded tail of each stream, neither
+crowding out the other's last lines; and nothing from a sandbox whose provider can't tell."""
 
 import subprocess
 from types import SimpleNamespace
@@ -9,6 +10,7 @@
 import pytest
 
 import agent_env.providers.sandbox_providers.local_docker.provider as ls
+from agent_env.providers.sandbox_providers import modal_sandbox
 from agent_env.providers.sandbox_providers.modal_sandbox import ModalSandbox
 from agent_env.providers.sandbox_providers.modal_vm_sandbox import ModalVmSandbox
 
@@ -64,12 +66,26 @@ def failing(*args, timeout=None):
     assert await vm.exit_status() is None and await vm.output_tail(5) == []
 
 
-def _modal_sb(status, stdout="", stderr=""):
+class _Stream:
+    """A Modal sandbox's stream, handed a chunk at a time, which ``reads`` counts."""
+
+    def __init__(self, *chunks):
+        self.chunks, self.reads = chunks, 0
+
+    def __aiter__(self):
+        return self._chunks()
+
+    async def _chunks(self):
+        self.reads += 1
+        for chunk in self.chunks:
+            yield chunk
+
+
+def _modal_sb(status, stdout=(), stderr=()):
     sb = MagicMock()
     sb.object_id = "sb-1"
     sb.poll.aio = AsyncMock(return_value=status)
-    sb.stdout.read.aio = AsyncMock(return_value=stdout)
-    sb.stderr.read.aio = AsyncMock(return_value=stderr)
+    sb.stdout, sb.stderr = _Stream(*stdout), _Stream(*stderr)
     return sb
 
 
@@ -77,26 +93,48 @@ def _modal_sb(status, stdout="", stderr=""):
 @pytest.mark.parametrize("made", [lambda sb: ModalVmSandbox(sb, {}), lambda sb: ModalSandbox(sb, {})],
                          ids=["modal_vm", "modal"])
 async def test_a_modal_sandbox_reads_its_process_output_once_it_has_exited_and_keeps_it(made):
-    running = _modal_sb(None)
+    running = _modal_sb(None, stdout=["never read\n"])
 
     assert await made(running).exit_status() is None and await made(running).output_tail(5) == []
-    running.stdout.read.aio.assert_not_awaited()
+    assert running.stdout.reads == 0
 
-    ended = _modal_sb(2, stdout="[    1.0] Linux version 6.8\nagentenv-init: no filesystem labelled agentenv-root\n",
-                      stderr="launch: the guest powered off\n")
+    ended = _modal_sb(2, stdout=["[    1.0] Linux version 6.8\r\nagentenv-init: no filesystem lab", "elled agentenv-root\n",
+                                 "\n"], stderr=["launch: the guest powered off"])
     sandbox = made(ended)
     assert await sandbox.exit_status() == 2
     assert await sandbox.output_tail(5) == ["[    1.0] Linux version 6.8",
                                             "agentenv-init: no filesystem labelled agentenv-root",
                                             "launch: the guest powered off"]
-    assert await sandbox.output_tail(1) == ["launch: the guest powered off"]
-    ended.stdout.read.aio.assert_awaited_once()
+    assert await sandbox.output_tail(1) == ["agentenv-init: no filesystem labelled agentenv-root"]
+    assert (ended.stdout.reads, ended.stderr.reads) == (1, 1)
 
 
 @pytest.mark.asyncio
-async def test_a_modal_sandbox_whose_output_cant_be_read_gives_none(caplog):
+async def test_a_modal_sandbox_keeps_a_bounded_tail_of_each_stream_and_neither_crowds_out_the_others_last_lines(
+        monkeypatch):
+    monkeypatch.setattr(modal_sandbox, "_OUTPUT_KEPT_LINES", 300)
+    console = [f"console {n}\n" for n in range(5000)] + ["agentenv-init: can't mount /dev/vda2\n"]
+    sandbox = ModalVmSandbox(_modal_sb(0, stdout=console, stderr=[f"socat {n}\n" for n in range(400)]), {})
+
+    tail = await sandbox.output_tail(200)
+
+    assert len(sandbox._ended[0]) == len(sandbox._ended[1]) == 300
+    assert tail == [*(f"console {n}" for n in range(4901, 5000)), "agentenv-init: can't mount /dev/vda2",
+                    *(f"socat {n}" for n in range(300, 400))]
+    few = ModalVmSandbox(_modal_sb(0, stdout=["the guest's last words\n"], stderr=[f"socat {n}\n" for n in range(400)]),
+                         {})
+    assert (await few.output_tail(200))[0] == "the guest's last words" and len(await few.output_tail(200)) == 200
+
+
+@pytest.mark.asyncio
+async def test_a_modal_sandbox_whose_output_cant_be_read_to_its_end_keeps_what_it_read(caplog):
+    class Breaks(_Stream):
+        async def _chunks(self):
+            yield "agentenv-init: the image names no command\n"
+            raise RuntimeError("stream gone")
+
     sb = _modal_sb(2)
-    sb.stdout.read.aio = AsyncMock(side_effect=RuntimeError("stream gone"))
+    sb.stdout = Breaks()
 
-    assert await ModalVmSandbox(sb, {}).output_tail(5) == []
-    assert "Couldn't read the output of Modal sandbox sb-1" in caplog.text
+    assert await ModalVmSandbox(sb, {}).output_tail(5) == ["agentenv-init: the image names no command"]
+    assert "Couldn't read all the output of Modal sandbox sb-1" in caplog.text

From ba4ba2a04215bfff7792b4c0eea4318f7df17702 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 11 Nov 2026 18:56:15 -0800
Subject: [PATCH 3/4] test(computer): a compose's build of a tag waits for
 another worker's build of it

Two test workers composing the same computer, each in stores of its own,
build the same tag on one Docker daemon, and the second to finish fails
with "image ... already exists". The computer tests' builds of a tag now
take turns across workers, as the bootstrap images' builds do.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tst/integration/computer/conftest.py | 24 +++++++++++++++++++++++-
 1 file changed, 23 insertions(+), 1 deletion(-)

diff --git a/tst/integration/computer/conftest.py b/tst/integration/computer/conftest.py
index 6dff7d5ed..3d2fdaf35 100644
--- a/tst/integration/computer/conftest.py
+++ b/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

From c09f4c73cd06ef0165fda47c165c635a1081ce4c Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Wed, 11 Nov 2026 19:21:56 -0800
Subject: [PATCH 4/4] fix(computer): a Modal sandbox's ended output waits for
 both of its streams

When one stream's read failed, gather returned at once and the other's
later lines, the guest's reason among them, never reached the boot's
error. Both reads now finish within the timeout, and each failure is
logged on its own.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../sandbox_providers/modal_sandbox.py        | 10 ++++--
 .../sandbox_providers/sandbox_exit_test.py    | 33 ++++++++++++++++---
 2 files changed, 35 insertions(+), 8 deletions(-)

diff --git a/src/agent_env/providers/sandbox_providers/modal_sandbox.py b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
index 121394189..43c9907de 100644
--- a/src/agent_env/providers/sandbox_providers/modal_sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
@@ -140,9 +140,13 @@ async def _ended_output(sb: modal.Sandbox) -> Optional[tuple[list[str], list[str
     output, errors = collections.deque(maxlen=_OUTPUT_KEPT_LINES), collections.deque(maxlen=_OUTPUT_KEPT_LINES)
     try:
         async with asyncio.timeout(_OUTPUT_READ_SECONDS):
-            await asyncio.gather(_read_lines(sb.stdout, output), _read_lines(sb.stderr, errors))
-    except Exception as e:
-        logger.info(f"Couldn't read all the output of Modal sandbox {sb.object_id}: {fmt_exc(e)}")
+            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)
 
 
diff --git a/tst/unit/providers/sandbox_providers/sandbox_exit_test.py b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
index 6032abcbe..3995a4f72 100644
--- a/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
+++ b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
@@ -3,6 +3,7 @@
 from Modal, its output once it has exited, read once, a chunk at a time, into a bounded tail of each stream, neither
 crowding out the other's last lines; and nothing from a sandbox whose provider can't tell."""
 
+import asyncio
 import subprocess
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
@@ -127,14 +128,36 @@ async def test_a_modal_sandbox_keeps_a_bounded_tail_of_each_stream_and_neither_c
 
 
 @pytest.mark.asyncio
-async def test_a_modal_sandbox_whose_output_cant_be_read_to_its_end_keeps_what_it_read(caplog):
+async def test_a_modal_sandbox_whose_stream_breaks_keeps_what_it_read_and_all_the_other_stream_says(caplog):
     class Breaks(_Stream):
         async def _chunks(self):
-            yield "agentenv-init: the image names no command\n"
+            yield "launch: the computer's QEMU exited\n"
             raise RuntimeError("stream gone")
 
+    class Slow(_Stream):
+        async def _chunks(self):
+            for line in ("[    1.0] Linux version 6.8\n", "agentenv-init: the image names no command\n"):
+                await asyncio.sleep(0.05)
+                yield line
+
+    sb = _modal_sb(2)
+    sb.stdout, sb.stderr = Slow(), Breaks()
+
+    assert await ModalVmSandbox(sb, {}).output_tail(5) == [
+        "[    1.0] Linux version 6.8", "agentenv-init: the image names no command", "launch: the computer's QEMU exited"]
+    assert "Couldn't read all the output of Modal sandbox sb-1: builtins.RuntimeError: stream gone" in caplog.text
+
+
+@pytest.mark.asyncio
+async def test_a_modal_sandbox_whose_output_takes_too_long_to_read_keeps_what_it_read(monkeypatch, caplog):
+    class Endless(_Stream):
+        async def _chunks(self):
+            yield "agentenv-init: no filesystem labelled agentenv-root\n"
+            await asyncio.Event().wait()
+
+    monkeypatch.setattr(modal_sandbox, "_OUTPUT_READ_SECONDS", 0.2)
     sb = _modal_sb(2)
-    sb.stdout = Breaks()
+    sb.stdout = Endless()
 
-    assert await ModalVmSandbox(sb, {}).output_tail(5) == ["agentenv-init: the image names no command"]
-    assert "Couldn't read all the output of Modal sandbox sb-1" in caplog.text
+    assert await ModalVmSandbox(sb, {}).output_tail(5) == ["agentenv-init: no filesystem labelled agentenv-root"]
+    assert "reading took past 0.2 s" in caplog.text