From 0c46e2ab03b164b3900a2e384b9ebf653b61ded4 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Tue, 10 Nov 2026 16:18:10 -0800
Subject: [PATCH 1/4] fix(collect): read only the agent's own regular files,
 capped in size and time

A manifest entry is read only under base_path, and its name is stored as a
key with no separators or dot segments. On a local sandbox's host, which is
this machine, a file is read only where it resolves inside the sandbox's work
dir or base_path; sandboxes declare that folder as app_dir_on_this_machine.

Only regular files are read, links followed, up to the step's max_bytes
(16 GiB by default). The sandbox sends at most the size it reported, the
decoder refuses more, and a read past its deadline is abandoned. Modal exec
output streams instead of being read whole, and the CUA path applies the same
checks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../local_docker/provider.py                  |   4 +
 .../sandbox_providers/modal_sandbox.py        |   3 +
 .../providers/sandbox_providers/sandbox.py    |   3 +
 .../task_step/task_steps/collect_artifacts.py | 157 +++++++++++++-----
 .../sandbox_providers/modal_sandbox_test.py   |  14 ++
 .../provider_capabilities_test.py             |  11 ++
 tst/unit/store/fakes.py                       |   5 +-
 tst/unit/task_step/test_collect_artifacts.py  |  96 +++++++++++
 .../test_collect_artifacts_universe_suffix.py |   3 +-
 .../test_collect_artifacts_vm_host.py         |  69 +++++++-
 tst/unit/task_step/test_collect_integrity.py  |  75 ++++++++-
 tst/util/capabilities.py                      |   2 +-
 12 files changed, 384 insertions(+), 58 deletions(-)

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..96125c338 100644
--- a/src/agent_env/providers/sandbox_providers/local_docker/provider.py
+++ b/src/agent_env/providers/sandbox_providers/local_docker/provider.py
@@ -312,6 +312,10 @@ def owns_container(self) -> bool:
     def work_dir(self) -> Path:
         return self._work_dir
 
+    @property
+    def app_dir_on_this_machine(self) -> Path:
+        return self._work_dir
+
     def _rewrite_app_arg(self, arg: str) -> str:
         if arg == "/app" or arg.startswith(("/app/", "/app:")):
             return str(self._work_dir) + arg[4:]
diff --git a/src/agent_env/providers/sandbox_providers/modal_sandbox.py b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
index ac52f2d32..c7a2f7375 100644
--- a/src/agent_env/providers/sandbox_providers/modal_sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/modal_sandbox.py
@@ -133,6 +133,9 @@ def __init__(self, stream):
     async def read(self) -> bytes:
         return await self._stream.read.aio()
 
+    def __aiter__(self):
+        return self._stream.__aiter__()
+
 
 class _ModalProcessAdapter:
     def __init__(self, process):
diff --git a/src/agent_env/providers/sandbox_providers/sandbox.py b/src/agent_env/providers/sandbox_providers/sandbox.py
index 642bfe4db..73dadc7ce 100644
--- a/src/agent_env/providers/sandbox_providers/sandbox.py
+++ b/src/agent_env/providers/sandbox_providers/sandbox.py
@@ -173,6 +173,9 @@ class Sandbox(ABC):
     platform: str | None = DEFAULT_BUILD_PLATFORM
     # The host other containers on its provider's private network reach it at, ready for a URL; None when it's on none.
     private_host: str | None = None
+    # The folder of this machine its host commands see as /app, when they run on this machine as a local sandbox's do;
+    # None when they run on a host of their own.
+    app_dir_on_this_machine: Path | None = None
 
     def url_from_sandbox(self, url: str) -> str:
         """``url``, of a service this process reaches, as this sandbox reaches it."""
diff --git a/src/agent_env/task_step/task_steps/collect_artifacts.py b/src/agent_env/task_step/task_steps/collect_artifacts.py
index c55923a6a..88833b7ce 100644
--- a/src/agent_env/task_step/task_steps/collect_artifacts.py
+++ b/src/agent_env/task_step/task_steps/collect_artifacts.py
@@ -16,7 +16,8 @@
 Artifact paths come from one of these sources (in priority order):
   1. Manifest (`manifest_step_id`): a prior prompt step's declared artifacts from
      that step's PromptResponse.structured_output["artifacts"] (logical name -> abs path);
-     collected files are keyed by logical name.
+     collected files are keyed by logical name, its separators and dot segments dropped. The
+     agent writes the manifest, so an entry outside base_path is skipped.
   2. Seed data: context.metadata["seed"][artifacts_key] — a comma-separated
      list of filenames (e.g., "Dockerfile,run.sh,codebase.patch")
   3. Static fallback: artifact_paths list in the step definition
@@ -25,7 +26,11 @@
 read verbatim). Absolute entries let a task collect deliverables that aren't
 under a single base_path — outputs aren't guaranteed to live on the desktop /
 in /app — without assuming a common directory. base_path only applies to
-relative entries and to the no-list enumeration fallback.
+relative entries, to manifest entries and to the no-list enumeration fallback.
+
+Only regular files are read, up to `max_bytes` each. A local sandbox's host is
+this machine, so there a file is read only where it resolves inside the
+sandbox's work dir or base_path.
 
 This allows the same step definition to collect different artifacts per seed.
 
@@ -67,10 +72,12 @@
 from __future__ import annotations
 
 import asyncio
+import base64
 import functools
 import logging
 import os
 import json
+import posixpath
 import re
 import shlex
 import tempfile
@@ -98,6 +105,11 @@
 # or under this size take that single-shot path; larger ones stream in fixed byte ranges via
 # cua_bash so no single call has to move the whole file.
 _CUA_SINGLE_SHOT_MAX = 16 * 1024 * 1024  # 16 MB
+# The largest file a step collects unless it sets max_bytes.
+_DEFAULT_MAX_BYTES = 16 << 30
+# A read is abandoned past this floor plus a second for each _READ_RATE_FLOOR bytes of the file.
+_READ_DEADLINE_FLOOR_S = 300
+_READ_RATE_FLOOR = 1 << 20
 _CUA_CHUNK_BYTES = 8 * 1024 * 1024       # 8 MB per ranged read
 
 # artifact_paths entries with a URL scheme are not VM files — they're gold/source
@@ -239,6 +251,7 @@ def __init__(
         sandbox_name: Optional[str] = None,
         container_name: Optional[str] = None,
         universe_id_suffix: Optional[str] = None,
+        max_bytes: int = _DEFAULT_MAX_BYTES,
         depends_on: Optional[list[TaskStepDependency]] = None,
         fail_task_on_error: bool = True,
     ):
@@ -283,6 +296,7 @@ def __init__(
         self.exclude_basenames = set(exclude_basenames or [])
         self.sandbox_name = sandbox_name
         self.container_name = container_name
+        self.max_bytes = max_bytes
 
     def to_dict(self) -> dict:
         base = super().to_dict()
@@ -296,6 +310,7 @@ def to_dict(self) -> dict:
         base["sandbox_name"] = self.sandbox_name
         base["container_name"] = self.container_name
         base["universe_id_suffix"] = self.universe_id_suffix
+        base["max_bytes"] = self.max_bytes
         return base
 
     @classmethod
@@ -312,6 +327,7 @@ def from_dict(cls, data: dict) -> CollectArtifactsTaskStep:
             sandbox_name=data.get("sandbox_name"),
             container_name=data.get("container_name"),
             universe_id_suffix=data.get("universe_id_suffix"),
+            max_bytes=data.get("max_bytes", _DEFAULT_MAX_BYTES),
         )
 
     def _resolve_filenames(self, context: TaskStepContext) -> list[str]:
@@ -358,14 +374,15 @@ def _resolve_items(self, context: TaskStepContext) -> list[tuple[str, str, str]]
                     continue
                 if "*" in path:
                     continue  # unresolved glob (e.g. per-iteration debug logs) — can't fetch a literal *
+                # The agent writes the manifest, so only a file under base_path, once normalized, is read.
+                path = posixpath.normpath(path)
+                if not path.startswith(self.base_path + "/"):
+                    logger.warning(f"collect '{self.id}': manifest entry {name!r} names {path}, outside "
+                                   f"{self.base_path}; not collected")
+                    continue
                 if os.path.basename(path) in self.exclude_basenames:
                     continue
-                rel = (
-                    path[len(self.base_path) + 1:]
-                    if path.startswith(self.base_path + "/")
-                    else os.path.basename(path)
-                )
-                items.append((name, path, rel))
+                items.append((_manifest_key(name, path), path, path[len(self.base_path) + 1:]))
             if skipped_nonfile:
                 logger.debug(
                     "collect(manifest=%s): skipped non-absolute manifest entries "
@@ -445,30 +462,34 @@ async def _resolve_live_sandbox(self, provider, sandbox_id: str):
         ) from err
 
     async def _get_file_size(self, sandbox, container: Optional[str], source_path: str) -> int:
-        """Get file size on the agent's filesystem. Returns -1 only if the file is genuinely
-        absent; a `stat` that fails for any other reason RAISES so a failed size check is a
-        hard failure rather than a silent 'missing' skip that drops an existing deliverable.
+        """Size of the regular file at ``source_path`` on the agent's filesystem, links followed. Returns -1 only if the
+        file is genuinely absent; a `stat` that fails for any other reason RAISES so a failed size check is a hard
+        failure rather than a silent 'missing' skip that drops an existing deliverable, and so does anything but a
+        regular file (a device, a FIFO, a directory, or a link to one), which could stream without end.
         VM-mode wraps in `docker exec <container>`; container-mode runs `stat` directly."""
-        args = _exec_args(sandbox, container, ("stat", "-c", "%s", source_path))
+        args = _exec_args(sandbox, container, ("stat", "-L", "-c", "%F:%s", source_path))
         exit_code, stdout, stderr = await sandbox.exec_with_output(*args)
         if exit_code != 0:
             if _looks_absent(stderr):
                 return -1
             raise RuntimeError(f"could not size {source_path} (stat exit {exit_code}): {stderr[:200]}")
-        return int(stdout.strip())
+        kind, _, size = stdout.strip().rpartition(":")
+        if kind not in ("regular file", "regular empty file"):
+            raise RuntimeError(f"{source_path} is not a regular file ({kind or 'unknown type'}); refusing to read it")
+        return int(size)
 
     async def _collect_file(
-        self, sandbox, container: Optional[str], source_path: str, object_name: str, store, artifact_id: str,
-        version: int, content_type: str,
+        self, sandbox, container: Optional[str], source_path: str, size: int, object_name: str, store,
+        artifact_id: str, version: int, content_type: str,
     ) -> str:
-        """Read a file out of the sandbox and upload it to S3 — one path, any size.
+        """Read a file of ``size`` bytes out of the sandbox and upload it to S3.
 
         base64-encode in the sandbox (raw binary over a websocket exec transport is
         silently dropped; ASCII survives), stream stdout + decode incrementally to
         a temp file (bounds memory), then multipart-upload. stderr is drained in
-        parallel so the shared output pump can't backpressure-stall."""
-        import base64
-        bash_cmd = f"base64 < {shlex.quote(source_path)}"
+        parallel so the shared output pump can't backpressure-stall. The sandbox reads
+        no more than ``size`` bytes, and a read that outlasts its deadline is abandoned."""
+        bash_cmd = f"head -c {size} < {shlex.quote(source_path)} | base64"
         args = _exec_args(sandbox, container, ("bash", "-c", bash_cmd))
 
         local_path = None
@@ -479,24 +500,12 @@ async def _collect_file(
 
             process = await sandbox.exec(*args)
             stderr_task = asyncio.create_task(process.stderr.read())
-            stdout = process.stdout
-            with open(local_path, "wb") as fh:
-                if hasattr(stdout, "__aiter__"):
-                    # Async-iterable stdout (websocket exec): stream, decoding 4-char-aligned base64.
-                    pending = bytearray()
-                    async for chunk in stdout:
-                        pending += chunk.translate(None, b"\r\n")  # strip 76-col wrap
-                        n = (len(pending) // 4) * 4
-                        if n:
-                            fh.write(base64.b64decode(pending[:n]))
-                            del pending[:n]
-                    if pending:
-                        fh.write(base64.b64decode(pending))
-                else:
-                    # Modal stdout has read() but no __aiter__: read all, decode once.
-                    data = (await stdout.read()).translate(None, b"\r\n")
-                    fh.write(base64.b64decode(data))
-            exit_code = await process.wait()
+            deadline = _READ_DEADLINE_FLOOR_S + size / _READ_RATE_FLOOR
+            try:
+                exit_code = await asyncio.wait_for(_decode_to_file(process, local_path, size), timeout=deadline)
+            except TimeoutError:
+                stderr_task.cancel()
+                raise RuntimeError(f"reading {source_path} took more than {deadline:.0f}s; abandoned it") from None
             try:
                 await stderr_task
             except Exception:
@@ -558,9 +567,9 @@ async def _controller_file_size(self, deployed_env, path: str) -> int:
         as a missing file and silently dropping an existing deliverable. ``wc -c`` is
         portable across the Ubuntu and macOS desktop VMs; ``stat``'s flags are not. A
         transport exception from ``_controller_call`` propagates unchanged."""
-        out = await self._controller_call(
-            deployed_env, "cua_bash", {"script": f"wc -c < {shlex.quote(path)}"}
-        )
+        quoted = shlex.quote(path)
+        script = f'[ ! -e {quoted} ] || [ -f {quoted} ] || {{ echo "not a regular file" >&2; exit 1; }}; wc -c < {quoted}'
+        out = await self._controller_call(deployed_env, "cua_bash", {"script": script})
         # A present file (even 0 bytes) prints its size to stdout; empty stdout means wc failed.
         text = _cua_bash_stdout(out).strip()
         if text:
@@ -581,11 +590,11 @@ async def _controller_get_file(self, deployed_env, path: str) -> str:
         lenient skip). Small files come back in one ``cua_get_file`` call; large ones stream
         in byte ranges straight to disk, so neither the transfer nor this process ever holds
         the whole file in memory."""
-        import base64
-
         size = await self._controller_file_size(deployed_env, path)
         if size < 0:
             raise FileNotFoundError(f"file not found on CUA VM: {path}")
+        if size > self.max_bytes:
+            raise RuntimeError(f"{path} is {size} bytes, over the step's max_bytes of {self.max_bytes}")
         with tempfile.NamedTemporaryFile(delete=False, suffix=f"-{os.path.basename(path)}") as tmp:
             local_path = tmp.name
         try:
@@ -607,8 +616,6 @@ async def _controller_stream_to_file(self, deployed_env, path: str, size: int, l
         than scans. Each range is decoded and written straight to the file so memory stays
         bounded to one chunk. base64 may wrap, so whitespace is stripped before decoding. A
         short read fails loudly rather than leaving a truncated artifact."""
-        import base64
-
         logger.info(
             f"Streaming {path} ({size} bytes) off the CUA VM in {_CUA_CHUNK_BYTES}-byte chunks"
         )
@@ -920,6 +927,8 @@ async def _collect_items(self, provider, sandbox, container, items, context, sto
         else:
             logger.info(f"Collecting {len(items)} artifacts from {self.base_path}")
 
+        # A local sandbox's host is this machine: only what resolves inside its work dir or base_path is read.
+        roots = await self._local_roots(sandbox) if container is None and sandbox.app_dir_on_this_machine else None
         collected: dict[str, str] = {}
         file_artifacts: dict[str, FileArtifact] = {}
         hard_failures: list[str] = []
@@ -928,16 +937,19 @@ async def _collect_items(self, provider, sandbox, container, items, context, sto
             # silent skip) and provider.close() below still runs; only a genuinely-absent
             # file (size -1) is the lenient skip.
             try:
-                file_size = await self._get_file_size(sandbox, container, source_path)
+                read_path = source_path if roots is None else await _resolved_inside(sandbox, source_path, roots)
+                file_size = -1 if read_path is None else await self._get_file_size(sandbox, container, read_path)
                 if file_size < 0:
                     logger.warning(f"File not found: {source_path}")
                     continue
+                if file_size > self.max_bytes:
+                    raise RuntimeError(f"{source_path} is {file_size} bytes, over the step's max_bytes of {self.max_bytes}")
 
                 ext = os.path.splitext(object_name)[1]
                 content_type = _CONTENT_TYPES.get(ext, "application/octet-stream")
 
                 object_url = await self._collect_file(
-                    sandbox, container, source_path, object_name, store, artifact_id, version, content_type,
+                    sandbox, container, read_path, file_size, object_name, store, artifact_id, version, content_type,
                 )
 
                 collected[key] = object_url
@@ -961,6 +973,10 @@ async def _collect_items(self, provider, sandbox, container, items, context, sto
         await provider.close()
         return collected, file_artifacts, hard_failures
 
+    async def _local_roots(self, sandbox) -> tuple[str, ...]:
+        roots = (str(sandbox.app_dir_on_this_machine.resolve()), await _resolved(sandbox, self.base_path))
+        return tuple(root.rstrip("/") + "/" for root in roots if root)
+
     def _register_file_artifact(self, store, file_artifacts, artifact_id, key, object_name, content_type, object_url, context):
         """Register a FileArtifact document pointing at the already-uploaded S3
         object. We do NOT re-upload the bytes — FileArtifact.object_url is just a URL
@@ -1050,3 +1066,52 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             )
 
         return context
+
+
+def _manifest_key(name: str, path: str) -> str:
+    """The agent's name for a manifest entry as a key no reader takes for a path: its separators and dot segments
+    dropped, else the file's own name. Consumers write files under these keys, a zip of them among them."""
+    parts = [part for part in re.split(r"[\\/]+", name) if part not in ("", ".", "..")]
+    return "_".join(parts) or posixpath.basename(path)
+
+
+async def _decode_to_file(process, local_path: str, size: int) -> int:
+    """Decode ``process``'s base64 stdout into ``local_path`` and return its exit status, refusing a byte past ``size``."""
+    written = 0
+
+    def write(fh, data: bytes) -> None:
+        nonlocal written
+        written += len(data)
+        if written > size:
+            raise RuntimeError(f"the sandbox sent more than the {size} bytes it reported; refusing the rest")
+        fh.write(data)
+
+    stdout = process.stdout
+    with open(local_path, "wb") as fh:
+        if hasattr(stdout, "__aiter__"):
+            # Async-iterable stdout (websocket exec): stream, decoding 4-char-aligned base64.
+            pending = bytearray()
+            async for chunk in stdout:
+                pending += chunk.translate(None, b"\r\n")  # strip 76-col wrap
+                n = (len(pending) // 4) * 4
+                if n:
+                    write(fh, base64.b64decode(pending[:n]))
+                    del pending[:n]
+            if pending:
+                write(fh, base64.b64decode(pending))
+        else:
+            # A provider whose exec hands back its output whole: decode it once. The sandbox reads at most `size` bytes.
+            write(fh, base64.b64decode((await stdout.read()).translate(None, b"\r\n")))
+    return await process.wait()
+
+
+async def _resolved(sandbox, path: str) -> str | None:
+    exit_code, stdout, _ = await sandbox.exec_with_output("readlink", "-f", path)
+    return (stdout.strip() or None) if exit_code == 0 else None
+
+
+async def _resolved_inside(sandbox, path: str, roots: tuple[str, ...]) -> str | None:
+    resolved = await _resolved(sandbox, path)
+    if resolved is not None and not (resolved + "/").startswith(roots):
+        raise RuntimeError(f"{path} resolves to {resolved}, outside the sandbox's work dir and base_path on this machine")
+    return resolved
diff --git a/tst/unit/providers/sandbox_providers/modal_sandbox_test.py b/tst/unit/providers/sandbox_providers/modal_sandbox_test.py
index 5c745ec58..eb962ec01 100644
--- a/tst/unit/providers/sandbox_providers/modal_sandbox_test.py
+++ b/tst/unit/providers/sandbox_providers/modal_sandbox_test.py
@@ -488,3 +488,17 @@ async def test_get_sandbox_still_raises_for_a_finished_sandbox_so_a_reattach_see
         from_id.aio = AsyncMock(return_value=sb)
         with pytest.raises(modal.exception.ConflictError):
             await _patched_provider.get_sandbox("sb-finished")
+
+
+@pytest.mark.asyncio
+async def test_exec_output_streams_as_modal_yields_it():
+    class _Stream:
+        async def _chunks(self):
+            for chunk in (b"QUJD", b"REVG"):
+                yield chunk
+
+        def __aiter__(self):
+            return self._chunks()
+
+    assert [chunk async for chunk in modal_sandbox._ModalStreamAdapter(_Stream())] == [b"QUJD", b"REVG"]
+
diff --git a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py
index cf738e2fc..d66fa994b 100644
--- a/tst/unit/providers/sandbox_providers/provider_capabilities_test.py
+++ b/tst/unit/providers/sandbox_providers/provider_capabilities_test.py
@@ -299,6 +299,17 @@ def test_a_local_sandbox_runs_any_platform_unchecked_since_its_docker_emulates_o
     assert LocalSandbox.platform is None
 
 
+@pytest.mark.parametrize("sandbox", [Sandbox, ModalSandbox, ModalVmSandbox, E2BSandbox, SailVmSandbox, TensorlakeSandbox,
+                                     TenkiSandbox, BoxdSandbox, CoreWeaveSandbox, DaytonaSandbox, BoatSandbox,
+                                     Agent37Sandbox])
+def test_a_sandbox_runs_its_host_commands_on_a_host_of_its_own(sandbox):
+    assert sandbox.app_dir_on_this_machine is None
+
+
+def test_a_local_sandbox_runs_its_host_commands_on_this_machine_in_its_work_dir(tmp_path):
+    assert LocalSandbox(work_dir=tmp_path).app_dir_on_this_machine == tmp_path
+
+
 def _concrete_classes() -> set[str]:
     """The sandbox and sandbox provider classes the providers define, but the base classes everything else may name."""
     names = set()
diff --git a/tst/unit/store/fakes.py b/tst/unit/store/fakes.py
index a308272e2..6719535ba 100644
--- a/tst/unit/store/fakes.py
+++ b/tst/unit/store/fakes.py
@@ -249,6 +249,7 @@ class CollectingVm:
 
     mode = "vm"
     type = None
+    app_dir_on_this_machine = None
 
     def scoped_name(self, name):
         return name
@@ -258,11 +259,11 @@ async def exec_script(self, script: str) -> str:
 
     async def exec_with_output(self, *args):
         if "stat" in args:
-            return 0, str(len(COLLECTED[args[-1]])), ""
+            return 0, f"regular file:{len(COLLECTED[args[-1]])}", ""
         return 0, "ok", ""
 
     async def exec(self, *args):
-        path = args[-1].split("base64 < ", 1)[1].strip("'")
+        path = args[-1].split(" < ", 1)[1].split(" | ", 1)[0].strip("'")
         return SimpleNamespace(stdout=_Stream(base64.b64encode(COLLECTED[path])), stderr=_Stream(b""),
                                wait=lambda: asyncio.sleep(0, result=0))
 
diff --git a/tst/unit/task_step/test_collect_artifacts.py b/tst/unit/task_step/test_collect_artifacts.py
index 31bc1ad76..0ed1c4790 100644
--- a/tst/unit/task_step/test_collect_artifacts.py
+++ b/tst/unit/task_step/test_collect_artifacts.py
@@ -11,6 +11,7 @@
 import asyncio
 import json
 import os
+import subprocess
 import tempfile
 import threading
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -81,6 +82,11 @@ def test_env_id_defaults_none(self):
         })
         assert step.env_id is None
 
+    def test_max_bytes_round_trips_and_defaults(self):
+        step = CollectArtifactsTaskStep.from_dict({"id": "c", "type": "collect_artifacts", "max_bytes": 4096})
+        assert CollectArtifactsTaskStep.from_dict(step.to_dict()).max_bytes == 4096
+        assert CollectArtifactsTaskStep.from_dict({"id": "c", "type": "collect_artifacts"}).max_bytes == 16 << 30
+
     def test_explicit_agent_name_and_env_id_raises(self):
         try:
             CollectArtifactsTaskStep(
@@ -516,6 +522,34 @@ def test_get_file_size_other_error_raises(self):
         except RuntimeError as e:
             assert "could not size" in str(e)
 
+    @pytest.mark.parametrize("kind", ["character special file", "fifo", "directory", "socket"])
+    def test_get_file_size_refuses_anything_but_a_regular_file(self, kind):
+        step = CollectArtifactsTaskStep(id="c", version=None, agent_name="solver")
+        sandbox = MagicMock()
+        sandbox.mode = "container"
+        sandbox.exec_with_output = AsyncMock(return_value=(0, f"{kind}:0\n", ""))
+        with pytest.raises(RuntimeError, match="not a regular file"):
+            _run(step._get_file_size(sandbox, None, "/app/artifact/report.pdf"))
+        assert sandbox.exec_with_output.await_args.args[:4] == ("stat", "-L", "-c", "%F:%s")
+
+    def test_a_file_over_max_bytes_is_a_hard_failure_and_is_not_read(self):
+        step = CollectArtifactsTaskStep(
+            id="c", version=None, base_path="/app/artifact", artifact_paths=["disk.img"], max_bytes=1024,
+        )
+        sandbox = MagicMock()
+        sandbox.mode = "container"
+        sandbox.app_dir_on_this_machine = None
+        sandbox.exec_with_output = AsyncMock(return_value=(0, f"regular file:{8 << 40}\n", ""))
+        sandbox.exec = AsyncMock()
+        provider = MagicMock()
+        provider.close = AsyncMock()
+        ctx = TaskStepContext()
+        collected, _, hard_failures = _run(step._collect_items(
+            provider, sandbox, None, step._resolve_items(ctx), ctx, MagicMock(), "aid", 1,
+        ))
+        assert (collected, hard_failures) == ({}, ["/app/artifact/disk.img"])
+        sandbox.exec.assert_not_awaited()
+
     def test_unreachable_sandbox_is_clear_error(self):
         """A re-run targeting an expired/unreachable sandbox fails with a clear message."""
         step = CollectArtifactsTaskStep(
@@ -599,6 +633,30 @@ def test_manifest_mode_keys_by_logical_name_and_filters(self):
         assert items["gold_patch"] == ("/app/workspace/patches/gold_patch.diff", "patches/gold_patch.diff")
         assert items["dockerfile"] == ("/app/workspace/Dockerfile", "Dockerfile")
 
+    def test_manifest_entries_outside_base_path_are_not_read(self, caplog):
+        step = CollectArtifactsTaskStep(id="collect", version=None, base_path="/app/workspace", manifest_step_id="m9")
+        ctx = _manifest_ctx({
+            "report": "/home/op/.ssh/id_rsa",
+            "climb": "/app/workspace/../../etc/passwd",
+            "sibling": "/app/workspace-other/notes.md",
+            "kept": "/app/workspace/out/./report.md",
+        })
+        with caplog.at_level("WARNING"):
+            items = step._resolve_items(ctx)
+        assert items == [("kept", "/app/workspace/out/report.md", "out/report.md")]
+        assert "names /home/op/.ssh/id_rsa, outside /app/workspace; not collected" in caplog.text
+        assert "names /etc/passwd, outside /app/workspace" in caplog.text
+
+    def test_manifest_names_become_keys_with_no_separators_or_dot_segments(self):
+        step = CollectArtifactsTaskStep(id="collect", version=None, base_path="/app/workspace", manifest_step_id="m9")
+        ctx = _manifest_ctx({
+            "../../.bashrc": "/app/workspace/payload",
+            "logs/run.json": "/app/workspace/logs/run.json",
+            "..\\..\\evil": "/app/workspace/evil",
+            "..": "/app/workspace/dots.txt",
+        })
+        assert [k for k, _, _ in step._resolve_items(ctx)] == [".bashrc", "logs_run.json", "evil", "dots.txt"]
+
     def test_enumeration_drops_excluded_basenames(self):
         step = CollectArtifactsTaskStep(
             id="collect", version=None, base_path="/app/workspace",
@@ -936,6 +994,44 @@ async def call(deployed_env, tool_name, arguments):
             except RuntimeError as e:
                 assert "refusing to upload a partial artifact" in str(e)
 
+    def test_a_file_over_max_bytes_is_refused_before_any_read(self):
+        step = CollectArtifactsTaskStep(id="c", version=None, env_id="ubuntu-cua", max_bytes=1024)
+        calls = []
+
+        async def call(deployed_env, tool_name, arguments):
+            calls.append(tool_name)
+            return json.dumps({"output": f"{8 << 40}\n", "error": "", "returncode": 0})
+
+        with patch.object(step, "_controller_call", side_effect=call), \
+             pytest.raises(RuntimeError, match="over the step's max_bytes of 1024"):
+            _run(step._controller_get_file(_cua_record(), "/x/sparse.img"))
+        assert calls == ["cua_bash"]
+
+    @pytest.mark.parametrize("make, size", [
+        (lambda p, tmp: p.write_bytes(b"hello"), 5),
+        (lambda p, tmp: None, -1),
+        (lambda p, tmp: p.symlink_to(tmp / "gone"), -1),
+        (lambda p, tmp: p.symlink_to("/dev/zero"), None),
+        (lambda p, tmp: os.mkfifo(p), None),
+        (lambda p, tmp: p.mkdir(), None),
+    ], ids=["regular", "absent", "dangling-link", "link-to-dev-zero", "fifo", "directory"])
+    def test_the_size_probe_sizes_only_a_regular_file(self, tmp_path, make, size):
+        """The probe's own script, run by a real shell as cua_bash runs it on the VM."""
+        target = tmp_path / "deliverable"
+        make(target, tmp_path)
+        step = CollectArtifactsTaskStep(id="c", version=None, env_id="ubuntu-cua")
+
+        async def call(deployed_env, tool_name, arguments):
+            done = subprocess.run(["bash", "-c", arguments["script"]], capture_output=True, text=True, timeout=10)
+            return json.dumps({"output": done.stdout, "error": done.stderr, "returncode": done.returncode})
+
+        with patch.object(step, "_controller_call", side_effect=call):
+            if size is None:
+                with pytest.raises(RuntimeError, match="not a regular file"):
+                    _run(step._controller_file_size(_cua_record(), str(target)))
+            else:
+                assert _run(step._controller_file_size(_cua_record(), str(target))) == size
+
     def test_absent_file_raises_filenotfound(self):
         """A missing file (wc errors with 'No such file') is reported absent so the caller
         skips it leniently."""
diff --git a/tst/unit/task_step/test_collect_artifacts_universe_suffix.py b/tst/unit/task_step/test_collect_artifacts_universe_suffix.py
index f4105ad9a..831f81606 100644
--- a/tst/unit/task_step/test_collect_artifacts_universe_suffix.py
+++ b/tst/unit/task_step/test_collect_artifacts_universe_suffix.py
@@ -38,6 +38,7 @@ async def wait(self):
 
 class _VmSandbox:
     mode = "vm"
+    app_dir_on_this_machine = None
 
     async def exec_with_output(self, *args):
         if "find" in args:
@@ -45,7 +46,7 @@ async def exec_with_output(self, *args):
                 return 0, "bundle\n", ""
             return 0, "".join(f"{p}\n" for p in FILES), ""
         if "stat" in args:
-            return 0, str(len(next(iter(FILES.values())))), ""
+            return 0, f"regular file:{len(next(iter(FILES.values())))}", ""
         return 0, "", ""
 
     async def exec(self, *args):
diff --git a/tst/unit/task_step/test_collect_artifacts_vm_host.py b/tst/unit/task_step/test_collect_artifacts_vm_host.py
index 945b06c1e..f00b1e97b 100644
--- a/tst/unit/task_step/test_collect_artifacts_vm_host.py
+++ b/tst/unit/task_step/test_collect_artifacts_vm_host.py
@@ -3,11 +3,14 @@
 from __future__ import annotations
 
 import base64
+from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
+from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandbox
 from agent_env.task_step.context import DeployedAgent, DeployedSandbox, TaskStepContext
 from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep
+from tst.util.capabilities import skip_without_gnu_stat
 
 FILES = {"caa_10/MODIFICATIONS.md": b"# what changed\n", "caa_10/alignment_x/task.toml": b"schema=1\n"}
 
@@ -33,6 +36,7 @@ class _VmSandbox:
     """A VM-mode sandbox that serves the fake tree, recording every command."""
 
     mode = "vm"
+    app_dir_on_this_machine = None
 
     def __init__(self):
         self.calls: list[tuple] = []
@@ -49,14 +53,14 @@ async def exec_with_output(self, *args):
         if "stat" in args:
             path = args[-1]
             rel = path.split("/app/artifact/", 1)[-1]
-            return (0, str(len(FILES[rel])), "") if rel in FILES else (-1, "", "no such file")
+            return (0, f"regular file:{len(FILES[rel])}", "") if rel in FILES else (-1, "", "no such file")
         return 0, "", ""
 
     async def exec(self, *args):
         self.calls.append(args)
         assert args[0] == "sudo", f"VM-host command must run as root: {args}"
         bash_cmd = args[-1]
-        path = bash_cmd.split("base64 < ", 1)[1].strip().strip("'")
+        path = bash_cmd.split(" < ", 1)[1].split(" | ", 1)[0].strip("'")
         rel = path.split("/app/artifact/", 1)[-1]
         return _Proc(base64.b64encode(FILES[rel]))
 
@@ -139,7 +143,7 @@ async def test_enumeration_and_reads_use_no_container(self, vm):
         # sudo'd but not container-wrapped: host-mode agents write as uid 0
         assert all(c.startswith(("sudo ", "echo ")) for c in flat), flat
         assert any(c.startswith("sudo find /app/artifact") for c in flat), flat
-        assert any(c.startswith("sudo bash -c base64 < ") for c in flat), flat
+        assert any(c.startswith("sudo bash -c head -c ") for c in flat), flat
 
     @pytest.mark.asyncio
     async def test_it_does_not_need_a_deployed_agent(self, vm):
@@ -251,3 +255,62 @@ async def _fake_agent_collect(self, context, store, artifact_id, version):
         step = CollectArtifactsTaskStep(id="collect", version=None, agent_name="solver")
         await step.execute(ctx)
         assert seen.get("called") is True
+
+
+@skip_without_gnu_stat()
+class TestOnThisMachine:
+    """A local sandbox's host is the operator's: a file is read only where it resolves inside the sandbox's work dir
+    or the step's base_path, so a link left there reaches nothing else on this machine."""
+
+    @staticmethod
+    async def _collect(step, sandbox):
+        uploaded = {}
+        store = MagicMock()
+
+        def put(**kw):
+            with open(kw["file_path"], "rb") as fh:
+                uploaded[kw["object_name"]] = fh.read()
+            return f"s3://bucket/{kw['object_name']}"
+
+        store.put_object_file.side_effect = put
+        provider = MagicMock()
+        provider.close = AsyncMock()
+        ctx = TaskStepContext()
+        collected, _, hard_failures = await step._collect_items(
+            provider, sandbox, None, step._resolve_items(ctx), ctx, store, "aid", 1,
+        )
+        return set(collected), hard_failures, uploaded
+
+    @pytest.mark.asyncio
+    async def test_a_link_out_of_the_work_dir_is_refused(self, tmp_path):
+        work = tmp_path / "work"
+        (work / "artifact").mkdir(parents=True)
+        (work / "other").mkdir()
+        (tmp_path / "id_rsa").write_bytes(b"operator secret")
+        (work / "artifact" / "report.md").write_bytes(b"# report")
+        (work / "other" / "notes.md").write_bytes(b"notes")
+        (work / "artifact" / "leak").symlink_to(tmp_path / "id_rsa")
+        (work / "artifact" / "notes").symlink_to(work / "other" / "notes.md")
+        step = CollectArtifactsTaskStep(
+            id="c", version=None, sandbox_name="box", base_path="/app/artifact",
+            artifact_paths=["report.md", "leak", "notes", "/app/../id_rsa"],
+        )
+        collected, hard_failures, uploaded = await self._collect(step, LocalSandbox(work_dir=work))
+        assert collected == {"report.md", "notes"}
+        assert set(hard_failures) == {"/app/artifact/leak", "/app/../id_rsa"}
+        assert b"operator secret" not in uploaded.values()
+
+    @pytest.mark.asyncio
+    async def test_a_base_path_the_task_names_on_this_machine_is_read(self, tmp_path):
+        out = tmp_path / "host" / "out"
+        out.mkdir(parents=True)
+        (tmp_path / "id_rsa").write_bytes(b"operator secret")
+        (out / "bundle.zip").write_bytes(b"PK bundle")
+        (out / "leak").symlink_to(tmp_path / "id_rsa")
+        step = CollectArtifactsTaskStep(
+            id="c", version=None, sandbox_name="box", base_path=str(out), artifact_paths=["bundle.zip", "leak"],
+        )
+        collected, hard_failures, uploaded = await self._collect(step, LocalSandbox(work_dir=tmp_path / "work"))
+        assert (collected, hard_failures) == ({"bundle.zip"}, [f"{out}/leak"])
+        assert uploaded == {"bundle.zip": b"PK bundle"}
+
diff --git a/tst/unit/task_step/test_collect_integrity.py b/tst/unit/task_step/test_collect_integrity.py
index e4266410e..bae358310 100644
--- a/tst/unit/task_step/test_collect_integrity.py
+++ b/tst/unit/task_step/test_collect_integrity.py
@@ -6,6 +6,7 @@
 import threading
 
 import pytest
+from agent_env.task_step.task_steps import collect_artifacts
 from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep
 from tst.unit.event_loop_probe import on_event_loop
 
@@ -39,7 +40,7 @@ def __init__(self, full_b64, keep, src_size):
     async def exec(self, *args):
         return _Proc(_TruncStdout(self._full, self._keep))
     async def exec_with_output(self, *args):
-        return (0, f"{self._src}\n", "")
+        return (0, f"regular file:{self._src}\n", "")
 
 
 @pytest.mark.asyncio
@@ -55,7 +56,7 @@ def put_object_file(self, **kw):
 
     step = CollectArtifactsTaskStep(id="collect", version=None)
     with pytest.raises(RuntimeError, match="integrity check failed"):
-        await step._collect_file(sb, None, "/app/artifact/corpus.warc.gz", "corpus.warc.gz",
+        await step._collect_file(sb, None, "/app/artifact/corpus.warc.gz", len(src), "corpus.warc.gz",
                                  _Store(), "uid", 1, "application/octet-stream")
 
 
@@ -72,7 +73,7 @@ def put_object_file(self, **kw):
             return "s3://bucket/ok"
 
     step = CollectArtifactsTaskStep(id="collect", version=None)
-    out = await step._collect_file(sb, None, "/app/artifact/corpus.warc.gz", "corpus.warc.gz",
+    out = await step._collect_file(sb, None, "/app/artifact/corpus.warc.gz", len(src), "corpus.warc.gz",
                                    _OkStore(), "uid", 1, "application/octet-stream")
     assert out == "s3://bucket/ok"
 
@@ -90,7 +91,7 @@ def put_object_file(self, **kw):
             return "s3://bucket/ok"
 
     step = CollectArtifactsTaskStep(id="collect", version=None)
-    await step._collect_file(sb, None, "/app/artifact/a.bin", "a.bin", _Store(), "uid", 1, "application/octet-stream")
+    await step._collect_file(sb, None, "/app/artifact/a.bin", len(src), "a.bin", _Store(), "uid", 1, "application/octet-stream")
     assert on_loop == [False]
 
 
@@ -111,7 +112,7 @@ def put_object_file(self, **kw):
             return "s3://bucket/ok"
 
     step = CollectArtifactsTaskStep(id="collect", version=None)
-    collect = asyncio.create_task(step._collect_file(sb, None, "/app/artifact/a.bin", "a.bin", _Store(), "uid", 1, "application/octet-stream"))
+    collect = asyncio.create_task(step._collect_file(sb, None, "/app/artifact/a.bin", len(src), "a.bin", _Store(), "uid", 1, "application/octet-stream"))
     await asyncio.to_thread(uploading.wait, 5)
     collect.cancel()
     with pytest.raises(asyncio.CancelledError):
@@ -125,3 +126,67 @@ def put_object_file(self, **kw):
     assert uploaded == [src]
     assert "Uploading collected /app/artifact/a.bin finished after its caller was cancelled" in caplog.text
     assert list(tmp_path.iterdir()) == []
+
+
+class _HangingStdout:
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        await asyncio.Event().wait()
+
+
+class _RecordingSandbox(_Sandbox):
+    def __init__(self, *args, stdout=None):
+        super().__init__(*args)
+        self.commands, self._stdout = [], stdout
+
+    async def exec(self, *args):
+        self.commands.append(args)
+        return _Proc(self._stdout or _TruncStdout(self._full, self._keep))
+
+
+class _UploadingStore:
+    def put_object_file(self, **kw):
+        return "s3://bucket/ok"
+
+
+class _RefusingStore:
+    def put_object_file(self, **kw):
+        raise AssertionError("uploaded an artifact the read should have refused")
+
+
+@pytest.mark.asyncio
+async def test_the_sandbox_reads_no_more_than_the_size_it_reported():
+    src = b"F" * 900
+    sb = _RecordingSandbox(base64.b64encode(src), len(base64.b64encode(src)), len(src))
+    step = CollectArtifactsTaskStep(id="collect", version=None)
+    await step._collect_file(sb, None, "/app/artifact/a b.bin", len(src), "a b.bin", _UploadingStore(), "uid", 1,
+                             "application/octet-stream")
+    assert sb.commands == [("bash", "-c", "head -c 900 < '/app/artifact/a b.bin' | base64")]
+
+
+@pytest.mark.asyncio
+async def test_a_stream_longer_than_the_reported_size_is_refused(tmp_path, monkeypatch):
+    monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
+    src = b"G" * 900
+    more = base64.b64encode(src * 3)
+    sb = _Sandbox(more, len(more), len(src))
+    step = CollectArtifactsTaskStep(id="collect", version=None)
+    with pytest.raises(RuntimeError, match="more than the 900 bytes it reported"):
+        await step._collect_file(sb, None, "/app/artifact/zero", len(src), "zero", _RefusingStore(), "uid", 1,
+                                 "application/octet-stream")
+    assert list(tmp_path.iterdir()) == []
+
+
+@pytest.mark.asyncio
+async def test_a_read_past_its_deadline_is_abandoned(tmp_path, monkeypatch):
+    monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
+    monkeypatch.setattr(collect_artifacts, "_READ_DEADLINE_FLOOR_S", 0.05)
+    sb = _RecordingSandbox(b"", 0, 10, stdout=_HangingStdout())
+    step = CollectArtifactsTaskStep(id="collect", version=None)
+    with pytest.raises(RuntimeError, match="took more than 0s; abandoned it"):
+        await step._collect_file(sb, None, "/app/artifact/slow", 10, "slow", _RefusingStore(), "uid", 1,
+                                 "application/octet-stream")
+    assert list(tmp_path.iterdir()) == []
+
diff --git a/tst/util/capabilities.py b/tst/util/capabilities.py
index 1d4d46d17..3765d5f51 100644
--- a/tst/util/capabilities.py
+++ b/tst/util/capabilities.py
@@ -63,7 +63,7 @@
 #: SGLang and Megatron-LM and CPU torch aren't in the dev extra, and only that tier's CI job installs them.
 SLIME = "slime_installed"
 
-#: ``collect_artifacts`` sizes a file with ``stat -c %s``, which only GNU stat takes (macOS ships BSD stat),
+#: ``collect_artifacts`` sizes a file with ``stat -L -c %F:%s``, which only GNU stat takes (macOS ships BSD stat),
 #: so collecting off a local sandbox needs GNU stat first on PATH.
 GNU_STAT = "gnu_stat"
 

From 1b994582b1e0cce5bfd544e4ed6e10166c5f068c Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Tue, 10 Nov 2026 16:18:10 -0800
Subject: [PATCH 2/4] fix(explorer): keep every Download All member name inside
 the archive

Each collected key becomes a clean relative member name: backslashes
separate, drive prefixes, leading separators and '.' segments are dropped,
and '..' climbs no higher than the archive root.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../ui/src/lib/collected-zip-files.smoke.ts   | 30 +++++++++++++++++++
 .../ui/src/lib/collected-zip-files.ts         | 18 +++++++----
 2 files changed, 43 insertions(+), 5 deletions(-)

diff --git a/src/agent_env/explorer/ui/src/lib/collected-zip-files.smoke.ts b/src/agent_env/explorer/ui/src/lib/collected-zip-files.smoke.ts
index 8d172454b..60f700d9b 100644
--- a/src/agent_env/explorer/ui/src/lib/collected-zip-files.smoke.ts
+++ b/src/agent_env/explorer/ui/src/lib/collected-zip-files.smoke.ts
@@ -52,6 +52,36 @@ function main(): void {
     assert(byUri['fake://home/c'] === 'elsewhere/c.txt', 'an absolute key elsewhere loses its leading slash');
   }
 
+  // No member name climbs out of the archive or names a root or a drive, whatever the key holds.
+  {
+    const files = collectedZipFiles(
+      {
+        '../../x': 'fake://home/x',
+        '/abs/../../y': 'fake://home/y',
+        'a/../../../z': 'fake://home/z',
+        '..\\..\\w.txt': 'fake://home/w',
+        'C:\\Users\\op\\.ssh\\v': 'fake://home/v',
+        'C:evil': 'fake://home/e',
+        './out/./u.txt': 'fake://home/u',
+        '..': 'fake://home/dots',
+      },
+      '/app',
+    );
+    const byUri = Object.fromEntries(files.map(f => [f.objectUrl, f.path]));
+    assert(byUri['fake://home/x'] === 'x', 'leading .. segments are dropped');
+    assert(byUri['fake://home/y'] === 'y', 'an absolute key outside base_path climbs no higher than the root');
+    assert(byUri['fake://home/z'] === 'z', 'a .. past the start is dropped');
+    assert(byUri['fake://home/w'] === 'w.txt', 'backslashes separate like slashes');
+    assert(byUri['fake://home/v'] === 'Users/op/.ssh/v', 'a drive prefix is dropped');
+    assert(byUri['fake://home/e'] === 'evil', 'a drive-relative prefix is dropped');
+    assert(byUri['fake://home/u'] === 'out/u.txt', '. segments are dropped');
+    assert(byUri['fake://home/dots'] === 'artifact', 'a key with nothing left is named artifact');
+    assert(
+      files.every(f => !f.path.split('/').some(part => part === '..' || part === '') && !/^[A-Za-z]:/.test(f.path)),
+      'every member name is a clean relative path',
+    );
+  }
+
   if (failures > 0) {
     console.error(`\n${failures} assertion(s) failed`);
     process.exit(1);
diff --git a/src/agent_env/explorer/ui/src/lib/collected-zip-files.ts b/src/agent_env/explorer/ui/src/lib/collected-zip-files.ts
index 68bebbf3a..adca36447 100644
--- a/src/agent_env/explorer/ui/src/lib/collected-zip-files.ts
+++ b/src/agent_env/explorer/ui/src/lib/collected-zip-files.ts
@@ -18,11 +18,19 @@ export function collectedZipFiles(
   return Object.entries(artifacts)
     .filter(([, uri]) => typeof uri === 'string' && uri !== '')
     .map(([key, uri]) => ({
-      path: !isAbsoluteArtifactPath(key)
-        ? key
-        : key.startsWith(`${base}/`)
-        ? key.slice(base.length + 1)
-        : key.replace(/^[/\\]+/, ''),
+      path: archiveName(
+        !isAbsoluteArtifactPath(key) ? key : key.startsWith(`${base}/`) ? key.slice(base.length + 1) : key,
+      ),
       objectUrl: uri,
     }));
 }
+
+/** A key as a member name that stays inside the archive when extracted: an agent names its manifest's entries. */
+export function archiveName(key: string): string {
+  const parts: string[] = [];
+  for (const part of key.replace(/^[A-Za-z]:/, '').split(/[\\/]+/)) {
+    if (part === '..') parts.pop();
+    else if (part !== '' && part !== '.') parts.push(part);
+  }
+  return parts.join('/') || 'artifact';
+}

From 86a628495d6cbffbb6dd926cec043465b348226f Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Tue, 10 Nov 2026 16:25:04 -0800
Subject: [PATCH 3/4] fix(collect): keep /app to the work dir, give colliding
 keys their own, time the whole read

Under /app only the sandbox's work dir counts as a root: an agent could swap a
folder there for a link out. A base_path outside /app still counts, resolved.
Two manifest names that clean to one key get "key (2)" and on. The deadline
covers exec too, which some providers run to completion. Manifest paths are
matched against base_path normalized.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../task_step/task_steps/collect_artifacts.py | 53 ++++++++++++++-----
 tst/unit/task_step/test_collect_artifacts.py  | 19 +++++++
 .../test_collect_artifacts_vm_host.py         | 13 +++++
 tst/unit/task_step/test_collect_integrity.py  | 16 ++++++
 4 files changed, 88 insertions(+), 13 deletions(-)

diff --git a/src/agent_env/task_step/task_steps/collect_artifacts.py b/src/agent_env/task_step/task_steps/collect_artifacts.py
index 88833b7ce..02f0b2441 100644
--- a/src/agent_env/task_step/task_steps/collect_artifacts.py
+++ b/src/agent_env/task_step/task_steps/collect_artifacts.py
@@ -30,7 +30,7 @@
 
 Only regular files are read, up to `max_bytes` each. A local sandbox's host is
 this machine, so there a file is read only where it resolves inside the
-sandbox's work dir or base_path.
+sandbox's work dir, or inside a base_path the task names outside /app.
 
 This allows the same step definition to collect different artifacts per seed.
 
@@ -73,6 +73,7 @@
 
 import asyncio
 import base64
+import contextlib
 import functools
 import logging
 import os
@@ -368,6 +369,8 @@ def _resolve_items(self, context: TaskStepContext) -> list[tuple[str, str, str]]
             manifest = (structured.get("artifacts") if isinstance(structured, dict) else None) or {}
             items: list[tuple[str, str, str]] = []
             skipped_nonfile: list[str] = []
+            base = posixpath.join(posixpath.normpath(self.base_path or "/"), "")
+            keys: set[str] = set()
             for name, path in manifest.items():
                 if not (isinstance(path, str) and path.startswith("/")):
                     skipped_nonfile.append(name)  # literal value (e.g. coverage_state) or relative path
@@ -376,13 +379,15 @@ def _resolve_items(self, context: TaskStepContext) -> list[tuple[str, str, str]]
                     continue  # unresolved glob (e.g. per-iteration debug logs) — can't fetch a literal *
                 # The agent writes the manifest, so only a file under base_path, once normalized, is read.
                 path = posixpath.normpath(path)
-                if not path.startswith(self.base_path + "/"):
+                if not path.startswith(base):
                     logger.warning(f"collect '{self.id}': manifest entry {name!r} names {path}, outside "
                                    f"{self.base_path}; not collected")
                     continue
                 if os.path.basename(path) in self.exclude_basenames:
                     continue
-                items.append((_manifest_key(name, path), path, path[len(self.base_path) + 1:]))
+                key = _unused(_manifest_key(name, path), keys)
+                keys.add(key)
+                items.append((key, path, path[len(base):]))
             if skipped_nonfile:
                 logger.debug(
                     "collect(manifest=%s): skipped non-absolute manifest entries "
@@ -498,18 +503,11 @@ async def _collect_file(
             with tempfile.NamedTemporaryFile(delete=False, suffix=f"-{object_name.replace('/', '_')}") as tmp:
                 local_path = tmp.name
 
-            process = await sandbox.exec(*args)
-            stderr_task = asyncio.create_task(process.stderr.read())
             deadline = _READ_DEADLINE_FLOOR_S + size / _READ_RATE_FLOOR
             try:
-                exit_code = await asyncio.wait_for(_decode_to_file(process, local_path, size), timeout=deadline)
+                exit_code = await asyncio.wait_for(_read_to_file(sandbox, args, local_path, size), timeout=deadline)
             except TimeoutError:
-                stderr_task.cancel()
                 raise RuntimeError(f"reading {source_path} took more than {deadline:.0f}s; abandoned it") from None
-            try:
-                await stderr_task
-            except Exception:
-                pass
             if exit_code != 0:
                 raise RuntimeError(f"base64 read failed for {source_path} (exit {exit_code})")
 
@@ -927,7 +925,7 @@ async def _collect_items(self, provider, sandbox, container, items, context, sto
         else:
             logger.info(f"Collecting {len(items)} artifacts from {self.base_path}")
 
-        # A local sandbox's host is this machine: only what resolves inside its work dir or base_path is read.
+        # A local sandbox's host is this machine: only what resolves inside its work dir, or a base_path outside /app.
         roots = await self._local_roots(sandbox) if container is None and sandbox.app_dir_on_this_machine else None
         collected: dict[str, str] = {}
         file_artifacts: dict[str, FileArtifact] = {}
@@ -974,7 +972,11 @@ async def _collect_items(self, provider, sandbox, container, items, context, sto
         return collected, file_artifacts, hard_failures
 
     async def _local_roots(self, sandbox) -> tuple[str, ...]:
-        roots = (str(sandbox.app_dir_on_this_machine.resolve()), await _resolved(sandbox, self.base_path))
+        """The sandbox's work dir, and base_path where the task names a folder of this machine outside /app: under
+        /app the agent may have swapped a folder for a link, so only the work dir it stands for counts."""
+        base = posixpath.normpath(self.base_path or "/")
+        outside_app = base != "/app" and not base.startswith("/app/")
+        roots = (str(sandbox.app_dir_on_this_machine.resolve()), await _resolved(sandbox, base) if outside_app else None)
         return tuple(root.rstrip("/") + "/" for root in roots if root)
 
     def _register_file_artifact(self, store, file_artifacts, artifact_id, key, object_name, content_type, object_url, context):
@@ -1075,6 +1077,31 @@ def _manifest_key(name: str, path: str) -> str:
     return "_".join(parts) or posixpath.basename(path)
 
 
+def _unused(key: str, keys: set[str]) -> str:
+    """``key``, or the first ``key (n)`` not in ``keys``: two manifest names can clean to the same key."""
+    n = 1
+    unused = key
+    while unused in keys:
+        n += 1
+        unused = f"{key} ({n})"
+    return unused
+
+
+async def _read_to_file(sandbox, args: tuple, local_path: str, size: int) -> int:
+    """Run ``args`` and decode its base64 stdout into ``local_path``, draining stderr beside it so the shared output
+    pump can't backpressure-stall; return its exit status. Some providers run the whole command inside ``exec``."""
+    process = await sandbox.exec(*args)
+    stderr = asyncio.create_task(process.stderr.read())
+    try:
+        exit_code = await _decode_to_file(process, local_path, size)
+    except BaseException:
+        stderr.cancel()
+        raise
+    with contextlib.suppress(Exception):
+        await stderr
+    return exit_code
+
+
 async def _decode_to_file(process, local_path: str, size: int) -> int:
     """Decode ``process``'s base64 stdout into ``local_path`` and return its exit status, refusing a byte past ``size``."""
     written = 0
diff --git a/tst/unit/task_step/test_collect_artifacts.py b/tst/unit/task_step/test_collect_artifacts.py
index 0ed1c4790..2fd7f65c1 100644
--- a/tst/unit/task_step/test_collect_artifacts.py
+++ b/tst/unit/task_step/test_collect_artifacts.py
@@ -657,6 +657,25 @@ def test_manifest_names_become_keys_with_no_separators_or_dot_segments(self):
         })
         assert [k for k, _, _ in step._resolve_items(ctx)] == [".bashrc", "logs_run.json", "evil", "dots.txt"]
 
+    def test_manifest_names_that_clean_to_one_key_each_keep_a_key_of_their_own(self):
+        step = CollectArtifactsTaskStep(id="collect", version=None, base_path="/app/workspace", manifest_step_id="m9")
+        ctx = _manifest_ctx({
+            "logs/run.json": "/app/workspace/a.json",
+            "logs_run.json": "/app/workspace/b.json",
+            "../logs/run.json": "/app/workspace/c.json",
+        })
+        assert [(k, src) for k, src, _ in step._resolve_items(ctx)] == [
+            ("logs_run.json", "/app/workspace/a.json"),
+            ("logs_run.json (2)", "/app/workspace/b.json"),
+            ("logs_run.json (3)", "/app/workspace/c.json"),
+        ]
+
+    @pytest.mark.parametrize("base_path", ["/app/./workspace", "/app/workspace/", "/app//workspace"])
+    def test_manifest_entries_are_matched_against_base_path_normalized(self, base_path):
+        step = CollectArtifactsTaskStep(id="collect", version=None, base_path=base_path, manifest_step_id="m9")
+        ctx = _manifest_ctx({"report": "/app/workspace/out/report.md"})
+        assert step._resolve_items(ctx) == [("report", "/app/workspace/out/report.md", "out/report.md")]
+
     def test_enumeration_drops_excluded_basenames(self):
         step = CollectArtifactsTaskStep(
             id="collect", version=None, base_path="/app/workspace",
diff --git a/tst/unit/task_step/test_collect_artifacts_vm_host.py b/tst/unit/task_step/test_collect_artifacts_vm_host.py
index f00b1e97b..baf7541e6 100644
--- a/tst/unit/task_step/test_collect_artifacts_vm_host.py
+++ b/tst/unit/task_step/test_collect_artifacts_vm_host.py
@@ -314,3 +314,16 @@ async def test_a_base_path_the_task_names_on_this_machine_is_read(self, tmp_path
         assert (collected, hard_failures) == ({"bundle.zip"}, [f"{out}/leak"])
         assert uploaded == {"bundle.zip": b"PK bundle"}
 
+    @pytest.mark.asyncio
+    async def test_a_base_path_the_agent_swapped_for_a_link_reaches_only_the_work_dir(self, tmp_path):
+        work, home = tmp_path / "work", tmp_path / "home"
+        (home / ".ssh").mkdir(parents=True)
+        (home / ".ssh" / "id_rsa").write_bytes(b"operator secret")
+        work.mkdir()
+        (work / "artifact").symlink_to(home)
+        step = CollectArtifactsTaskStep(
+            id="c", version=None, sandbox_name="box", base_path="/app/artifact", artifact_paths=[".ssh/id_rsa"],
+        )
+        collected, hard_failures, uploaded = await self._collect(step, LocalSandbox(work_dir=work))
+        assert (collected, hard_failures, uploaded) == (set(), ["/app/artifact/.ssh/id_rsa"], {})
+
diff --git a/tst/unit/task_step/test_collect_integrity.py b/tst/unit/task_step/test_collect_integrity.py
index bae358310..911bbeed1 100644
--- a/tst/unit/task_step/test_collect_integrity.py
+++ b/tst/unit/task_step/test_collect_integrity.py
@@ -190,3 +190,19 @@ async def test_a_read_past_its_deadline_is_abandoned(tmp_path, monkeypatch):
                                  "application/octet-stream")
     assert list(tmp_path.iterdir()) == []
 
+
+@pytest.mark.asyncio
+async def test_a_provider_whose_exec_runs_the_whole_read_is_held_to_the_deadline(tmp_path, monkeypatch):
+    monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
+    monkeypatch.setattr(collect_artifacts, "_READ_DEADLINE_FLOOR_S", 0.05)
+
+    class _StallingExec(_Sandbox):
+        async def exec(self, *args):
+            await asyncio.Event().wait()
+
+    step = CollectArtifactsTaskStep(id="collect", version=None)
+    with pytest.raises(RuntimeError, match="took more than 0s; abandoned it"):
+        await step._collect_file(_StallingExec(b"", 0, 10), None, "/app/artifact/slow", 10, "slow", _RefusingStore(),
+                                 "uid", 1, "application/octet-stream")
+    assert list(tmp_path.iterdir()) == []
+

From 6a71e94b781e22b40cc3e461cc58823cc30de572 Mon Sep 17 00:00:00 2001
From: Edgar Arakelyan <edgar.arakelyan@scale.com>
Date: Tue, 10 Nov 2026 16:47:50 -0800
Subject: [PATCH 4/4] fix(collect): tell a regular file by its mode bits, which
 no locale translates

stat's %F names the file type in the system's language, so on a host with a
non-English locale every file would have been refused. %f prints the raw mode,
checked with stat.S_ISREG.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../task_step/task_steps/collect_artifacts.py |  9 ++++----
 tst/unit/store/fakes.py                       |  2 +-
 tst/unit/task_step/test_collect_artifacts.py  | 21 +++++++++++++------
 .../test_collect_artifacts_universe_suffix.py |  2 +-
 .../test_collect_artifacts_vm_host.py         |  2 +-
 tst/unit/task_step/test_collect_integrity.py  |  2 +-
 tst/util/capabilities.py                      |  2 +-
 7 files changed, 25 insertions(+), 15 deletions(-)

diff --git a/src/agent_env/task_step/task_steps/collect_artifacts.py b/src/agent_env/task_step/task_steps/collect_artifacts.py
index 02f0b2441..1c204959c 100644
--- a/src/agent_env/task_step/task_steps/collect_artifacts.py
+++ b/src/agent_env/task_step/task_steps/collect_artifacts.py
@@ -81,6 +81,7 @@
 import posixpath
 import re
 import shlex
+import stat
 import tempfile
 import time
 from typing import ClassVar, Optional
@@ -472,15 +473,15 @@ async def _get_file_size(self, sandbox, container: Optional[str], source_path: s
         failure rather than a silent 'missing' skip that drops an existing deliverable, and so does anything but a
         regular file (a device, a FIFO, a directory, or a link to one), which could stream without end.
         VM-mode wraps in `docker exec <container>`; container-mode runs `stat` directly."""
-        args = _exec_args(sandbox, container, ("stat", "-L", "-c", "%F:%s", source_path))
+        args = _exec_args(sandbox, container, ("stat", "-L", "-c", "%f:%s", source_path))
         exit_code, stdout, stderr = await sandbox.exec_with_output(*args)
         if exit_code != 0:
             if _looks_absent(stderr):
                 return -1
             raise RuntimeError(f"could not size {source_path} (stat exit {exit_code}): {stderr[:200]}")
-        kind, _, size = stdout.strip().rpartition(":")
-        if kind not in ("regular file", "regular empty file"):
-            raise RuntimeError(f"{source_path} is not a regular file ({kind or 'unknown type'}); refusing to read it")
+        mode, _, size = stdout.strip().rpartition(":")
+        if not stat.S_ISREG(int(mode, 16)):
+            raise RuntimeError(f"{source_path} is not a regular file ({stat.filemode(int(mode, 16))}); refusing to read it")
         return int(size)
 
     async def _collect_file(
diff --git a/tst/unit/store/fakes.py b/tst/unit/store/fakes.py
index 6719535ba..ecfa09a6a 100644
--- a/tst/unit/store/fakes.py
+++ b/tst/unit/store/fakes.py
@@ -259,7 +259,7 @@ async def exec_script(self, script: str) -> str:
 
     async def exec_with_output(self, *args):
         if "stat" in args:
-            return 0, f"regular file:{len(COLLECTED[args[-1]])}", ""
+            return 0, f"81a4:{len(COLLECTED[args[-1]])}", ""
         return 0, "ok", ""
 
     async def exec(self, *args):
diff --git a/tst/unit/task_step/test_collect_artifacts.py b/tst/unit/task_step/test_collect_artifacts.py
index 2fd7f65c1..3c2f47b6c 100644
--- a/tst/unit/task_step/test_collect_artifacts.py
+++ b/tst/unit/task_step/test_collect_artifacts.py
@@ -522,15 +522,24 @@ def test_get_file_size_other_error_raises(self):
         except RuntimeError as e:
             assert "could not size" in str(e)
 
-    @pytest.mark.parametrize("kind", ["character special file", "fifo", "directory", "socket"])
-    def test_get_file_size_refuses_anything_but_a_regular_file(self, kind):
+    @pytest.mark.parametrize("mode, shown", [("21b6", "crw-rw-rw-"), ("11a4", "prw-r--r--"), ("41ed", "drwxr-xr-x"),
+                                             ("c1ed", "srwxr-xr-x")], ids=["device", "fifo", "directory", "socket"])
+    def test_get_file_size_refuses_anything_but_a_regular_file(self, mode, shown):
+        """By the raw mode bits stat prints, which no locale translates, as it does a file type's name."""
         step = CollectArtifactsTaskStep(id="c", version=None, agent_name="solver")
         sandbox = MagicMock()
         sandbox.mode = "container"
-        sandbox.exec_with_output = AsyncMock(return_value=(0, f"{kind}:0\n", ""))
-        with pytest.raises(RuntimeError, match="not a regular file"):
+        sandbox.exec_with_output = AsyncMock(return_value=(0, f"{mode}:0\n", ""))
+        with pytest.raises(RuntimeError, match=f"not a regular file \\({shown}\\)"):
             _run(step._get_file_size(sandbox, None, "/app/artifact/report.pdf"))
-        assert sandbox.exec_with_output.await_args.args[:4] == ("stat", "-L", "-c", "%F:%s")
+        assert sandbox.exec_with_output.await_args.args[:4] == ("stat", "-L", "-c", "%f:%s")
+
+    def test_get_file_size_sizes_a_regular_file_even_when_empty(self):
+        step = CollectArtifactsTaskStep(id="c", version=None, agent_name="solver")
+        sandbox = MagicMock()
+        sandbox.mode = "container"
+        sandbox.exec_with_output = AsyncMock(side_effect=[(0, "81a4:0\n", ""), (0, "8180:4096\n", "")])
+        assert [_run(step._get_file_size(sandbox, None, "/x")) for _ in range(2)] == [0, 4096]
 
     def test_a_file_over_max_bytes_is_a_hard_failure_and_is_not_read(self):
         step = CollectArtifactsTaskStep(
@@ -539,7 +548,7 @@ def test_a_file_over_max_bytes_is_a_hard_failure_and_is_not_read(self):
         sandbox = MagicMock()
         sandbox.mode = "container"
         sandbox.app_dir_on_this_machine = None
-        sandbox.exec_with_output = AsyncMock(return_value=(0, f"regular file:{8 << 40}\n", ""))
+        sandbox.exec_with_output = AsyncMock(return_value=(0, f"81a4:{8 << 40}\n", ""))
         sandbox.exec = AsyncMock()
         provider = MagicMock()
         provider.close = AsyncMock()
diff --git a/tst/unit/task_step/test_collect_artifacts_universe_suffix.py b/tst/unit/task_step/test_collect_artifacts_universe_suffix.py
index 831f81606..2227655ae 100644
--- a/tst/unit/task_step/test_collect_artifacts_universe_suffix.py
+++ b/tst/unit/task_step/test_collect_artifacts_universe_suffix.py
@@ -46,7 +46,7 @@ async def exec_with_output(self, *args):
                 return 0, "bundle\n", ""
             return 0, "".join(f"{p}\n" for p in FILES), ""
         if "stat" in args:
-            return 0, f"regular file:{len(next(iter(FILES.values())))}", ""
+            return 0, f"81a4:{len(next(iter(FILES.values())))}", ""
         return 0, "", ""
 
     async def exec(self, *args):
diff --git a/tst/unit/task_step/test_collect_artifacts_vm_host.py b/tst/unit/task_step/test_collect_artifacts_vm_host.py
index baf7541e6..a5ac205bc 100644
--- a/tst/unit/task_step/test_collect_artifacts_vm_host.py
+++ b/tst/unit/task_step/test_collect_artifacts_vm_host.py
@@ -53,7 +53,7 @@ async def exec_with_output(self, *args):
         if "stat" in args:
             path = args[-1]
             rel = path.split("/app/artifact/", 1)[-1]
-            return (0, f"regular file:{len(FILES[rel])}", "") if rel in FILES else (-1, "", "no such file")
+            return (0, f"81a4:{len(FILES[rel])}", "") if rel in FILES else (-1, "", "no such file")
         return 0, "", ""
 
     async def exec(self, *args):
diff --git a/tst/unit/task_step/test_collect_integrity.py b/tst/unit/task_step/test_collect_integrity.py
index 911bbeed1..26ff22c38 100644
--- a/tst/unit/task_step/test_collect_integrity.py
+++ b/tst/unit/task_step/test_collect_integrity.py
@@ -40,7 +40,7 @@ def __init__(self, full_b64, keep, src_size):
     async def exec(self, *args):
         return _Proc(_TruncStdout(self._full, self._keep))
     async def exec_with_output(self, *args):
-        return (0, f"regular file:{self._src}\n", "")
+        return (0, f"81a4:{self._src}\n", "")
 
 
 @pytest.mark.asyncio
diff --git a/tst/util/capabilities.py b/tst/util/capabilities.py
index 3765d5f51..3f7b934fc 100644
--- a/tst/util/capabilities.py
+++ b/tst/util/capabilities.py
@@ -63,7 +63,7 @@
 #: SGLang and Megatron-LM and CPU torch aren't in the dev extra, and only that tier's CI job installs them.
 SLIME = "slime_installed"
 
-#: ``collect_artifacts`` sizes a file with ``stat -L -c %F:%s``, which only GNU stat takes (macOS ships BSD stat),
+#: ``collect_artifacts`` sizes a file with ``stat -L -c %f:%s``, which only GNU stat takes (macOS ships BSD stat),
 #: so collecting off a local sandbox needs GNU stat first on PATH.
 GNU_STAT = "gnu_stat"