fix(collect): read only the agent's own regular files, capped in size and time

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

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';
+}

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:]

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

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."""

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"]
+

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

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"81a4:{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))
 

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,43 @@ def test_get_file_size_other_error_raises(self):
         except RuntimeError as e:
             assert "could not size" in str(e)
 
+    @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"{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")
+
+    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(
+            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"81a4:{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 +642,49 @@ 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_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",
@@ -936,6 +1022,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."""

tst/unit/task_step/test_collect_artifacts_universe_suffix.py

@@ -38,14 +38,15 @@ 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:
             if "-maxdepth" in 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"81a4:{len(next(iter(FILES.values())))}", ""
         return 0, "", ""
 
     async def exec(self, *args):