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..43c9907de 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,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)
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..f54d90f1a 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,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
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/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
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..3995a4f72
--- /dev/null
+++ b/tst/unit/providers/sandbox_providers/sandbox_exit_test.py
@@ -0,0 +1,163 @@
+"""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, 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
+
+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
+
+
+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) == []
+
+
+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, sb.stderr = _Stream(*stdout), _Stream(*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, stdout=["never read\n"])
+
+    assert await made(running).exit_status() is None and await made(running).output_tail(5) == []
+    assert running.stdout.reads == 0
+
+    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) == ["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_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_stream_breaks_keeps_what_it_read_and_all_the_other_stream_says(caplog):
+    class Breaks(_Stream):
+        async def _chunks(self):
+            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 = Endless()
+
+    assert await ModalVmSandbox(sb, {}).output_tail(5) == ["agentenv-init: no filesystem labelled agentenv-root"]
+    assert "reading took past 0.2 s" in caplog.text