fix(task_steps): reach the container of a container-mode deploy_sandbox sandbox

src/agent_env/a2a_agent/a2a_agent.py

@@ -29,7 +29,6 @@
     start_trusting,
 )
 from agent_env.providers.sandbox_providers.sandbox_provider import all_sandbox_container_env, all_sandbox_url_rewrites
-from agent_env.utils.paths import validate_relative_filename
 from agent_env.attribution import Attribution
 
 if TYPE_CHECKING:
@@ -263,7 +262,7 @@ async def load_file_artifact_universe(
         destination_path: str,
     ) -> dict[str, str]:
         """Stage every FileArtifact in `universe` into the agent sandbox under destination_path."""
-        from agent_env.providers.sandbox_providers.sandbox import VmSandbox
+        from agent_env.providers.sandbox_providers.sandbox import stage_files_into_container
         from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_agent_sandbox_provider
 
         destination = destination_path.rstrip("/") or "/"
@@ -277,41 +276,11 @@ async def load_file_artifact_universe(
             )
             return {}
 
-        dirs_to_make: set[str] = {destination}
-        loaded: dict[str, str] = {}
-        for filename in file_artifacts:
-            validate_relative_filename(filename)
-            dest_path = posixpath.join(destination, filename)
-            parent = posixpath.dirname(dest_path)
-            if parent:
-                dirs_to_make.add(parent)
-            loaded[filename] = dest_path
-
-        for d in sorted(dirs_to_make):
-            if isinstance(sandbox, VmSandbox):
-                await sandbox.exec_script(f"docker exec -u 0 {shlex.quote(sandbox.container_name)} mkdir -p {shlex.quote(d)}")
-            else:
-                await sandbox.exec("mkdir", "-p", d)
-
-        total = len(file_artifacts)
         logger.info(
             f"Loading FileArtifactUniverse '{universe.id}' v{universe.version} "
-            f"({total} file(s)) into agent at {destination}"
+            f"({len(file_artifacts)} file(s)) into agent at {destination}"
         )
-        # Stage files in parallel with a bounded semaphore so a universe with
-        # many files doesn't accumulate per-file presign + docker cp latency.
-        # 8 matches the default concurrency used by the FAU `get-many` CLI.
-        sem = asyncio.Semaphore(8)
-
-        async def _stage(filename: str, file_artifact: "Any") -> None:
-            dest_path = loaded[filename]
-            async with sem:
-                logger.info(f"  {file_artifact.object_url} -> {dest_path}")
-                await sandbox.write_file_from_s3(file_artifact.object_url, dest_path)
-
-        await asyncio.gather(*[_stage(fn, fa) for fn, fa in file_artifacts.items()])
-
-        return loaded
+        return await stage_files_into_container(sandbox, file_artifacts, destination)
 
     def __init__(
         self,

src/agent_env/providers/sandbox_providers/sandbox.py

@@ -5,6 +5,7 @@
 import asyncio
 import logging
 import os
+import posixpath
 import shlex
 import time
 import uuid
@@ -15,6 +16,7 @@
 from typing import TYPE_CHECKING, Any, Iterable, Optional
 
 from agent_env.config import get_config
+from agent_env.utils.paths import validate_relative_filename
 
 if TYPE_CHECKING:
     from agent_env.store.object_store import ObjectStore
@@ -406,3 +408,32 @@ async def write_file_from_text(self, content: str, destination_path: str) -> Non
 def port_bindings(host_ips: Iterable[str], host_port: int, container_port: int) -> list[str]:
     """Docker publish specs for one port: one per host IP, or a bare one (every interface) when there are none."""
     return [f"{ip}:{host_port}:{container_port}" for ip in host_ips] or [f"{host_port}:{container_port}"]
+
+
+async def stage_files_into_container(sandbox: Sandbox, file_artifacts: dict[str, Any], destination: str) -> dict[str, str]:
+    """Write each object-store file into the sandbox's container under ``destination``; returns ``{name: path}``.
+    A VM-backed sandbox stages through its host into ``container_name``; a container sandbox is written directly."""
+    loaded: dict[str, str] = {}
+    dirs_to_make: set[str] = {destination}
+    for filename in file_artifacts:
+        validate_relative_filename(filename)
+        dest_path = posixpath.join(destination, filename)
+        dirs_to_make.add(posixpath.dirname(dest_path))
+        loaded[filename] = dest_path
+
+    for d in sorted(dirs_to_make):
+        if isinstance(sandbox, VmSandbox):
+            await sandbox.exec_script(f"docker exec -u 0 {shlex.quote(sandbox.container_name)} mkdir -p {shlex.quote(d)}")
+        else:
+            await sandbox.exec("mkdir", "-p", d)
+
+    # Bounded so a universe with many files doesn't serialize per-file presign + copy latency.
+    sem = asyncio.Semaphore(8)
+
+    async def _stage(filename: str, file_artifact: Any) -> None:
+        async with sem:
+            logger.info(f"  {file_artifact.object_url} -> {loaded[filename]}")
+            await sandbox.write_file_from_s3(file_artifact.object_url, loaded[filename])
+
+    await asyncio.gather(*(_stage(fn, fa) for fn, fa in file_artifacts.items()))
+    return loaded

src/agent_env/task_step/task_steps/collect_artifacts.py

@@ -152,7 +152,8 @@ def _exec_args(sandbox, container: Optional[str], cmd: tuple) -> tuple:
     if sandbox.mode != SANDBOX_MODE_VM:
         return cmd  # container-mode sandboxes are the runtime and already root
     if container is not None:
-        return ("sudo", "docker", "exec", container, *cmd)
+        # As root, like the steps that write into the container: a non-root image user can't read their files.
+        return ("sudo", "docker", "exec", "-u", "0", container, *cmd)
     return ("sudo", *cmd)  # host-mode agents write as uid 0, so read them back as root
 
 
@@ -786,8 +787,13 @@ async def _collect_via_agent_container(self, context, store, artifact_id, versio
         )
 
     async def _collect_via_vm_host(self, context, store, artifact_id, version):
-        """Collect off the VM's own filesystem — host-mode agents leave no container to exec into."""
-        from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_sandbox_provider
+        """Collect off a deploy_sandbox sandbox: a VM's own filesystem, or a container-mode sandbox's container."""
+        from agent_env.providers.sandbox_providers.sandbox import VmSandbox
+        from agent_env.providers.sandbox_providers.sandbox_provider import (
+            SANDBOX_MODE_VM,
+            build_sandbox_provider,
+            get_sandbox_provider,
+        )
 
         ds = next(
             (sb for sb in context.deployed_sandboxes if sb.sandbox_name == self.sandbox_name), None
@@ -806,10 +812,18 @@ async def _collect_via_vm_host(self, context, store, artifact_id, version):
 
         provider = build_sandbox_provider(ds.sandbox_type) if ds.sandbox_type else get_sandbox_provider()
         sandbox = await self._resolve_live_sandbox(provider, ds.sandbox_id)
-        logger.info(f"Collecting from the VM host of sandbox '{self.sandbox_name}' ({ds.sandbox_id})")
+        # A container-mode sandbox on a VM-backed provider runs its image beside the host: read from the container.
+        container = (
+            sandbox.container_name
+            if ds.sandbox_mode != SANDBOX_MODE_VM and isinstance(sandbox, VmSandbox)
+            else None
+        )
+        logger.info(
+            f"Collecting from sandbox '{self.sandbox_name}' ({ds.sandbox_id}, {container or ds.sandbox_mode})"
+        )
 
         return await self._collect_items(
-            provider, sandbox, None, items, context, store, artifact_id, version,
+            provider, sandbox, container, items, context, store, artifact_id, version,
         )
 
     async def _collect_via_sandbox_container(self, context, store, artifact_id, version):

src/agent_env/task_step/task_steps/load_artifact.py

@@ -288,6 +288,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         from agent_env.env.env import Env, require_gateway_url
         from agent_env.providers.sandbox_providers.sandbox import VmSandbox
         from agent_env.providers.sandbox_providers.sandbox_provider import (
+            SANDBOX_MODE_VM,
             build_sandbox_provider,
             get_agent_sandbox_provider,
             get_sandbox_provider,
@@ -329,6 +330,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 build_sandbox_provider(ds.sandbox_type) if ds.sandbox_type else get_sandbox_provider()
             )
             target_sandbox = await provider.get_sandbox(ds.sandbox_id)
+        # A VM-mode sandbox with no container_name is loaded onto its host; a container-mode one, into its container.
+        onto_vm_host = (
+            self.sandbox_name is not None and self.container_name is None and ds.sandbox_mode == SANDBOX_MODE_VM
+        )
 
         env = None
         for ref in resolved_artifacts:
@@ -352,8 +357,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                         files = await _load_universe_into_container(
                             target_sandbox, self.container_name, artifact, destination,
                         )
-                    else:
+                    elif onto_vm_host:
                         files = await _load_universe_onto_vm(target_sandbox, artifact, destination)
+                    else:
+                        files = await _load_universe_into_sandbox_container(target_sandbox, artifact, destination)
                     context.metadata.setdefault("loaded_file_artifact_universes", []).append({
                         "id": artifact.id,
                         "version": artifact.version,
@@ -418,8 +425,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                     # involved.
                     if self.sandbox_name is not None:
                         sandbox = target_sandbox
-                        # None targets the VM host itself.
+                        # None targets the VM host itself, or a container sandbox directly.
                         container = self.container_name
+                        if container is None and not onto_vm_host and isinstance(sandbox, VmSandbox):
+                            container = sandbox.container_name
                     else:
                         agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
                         if agent is None or not agent.sandbox_id:
@@ -523,15 +532,11 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             async def _load_one(url: str, filename: str) -> None:
                 async with semaphore:
                     dest = f"{destination}/{filename}"
-                    if self.sandbox_name is not None and self.container_name is None:
+                    if onto_vm_host:
                         await self._load_url_onto_vm(sandbox, url, dest)
                     else:
                         await sandbox.write_file_from_url(url, dest)
-                    target_desc = (
-                        f"VM sandbox '{self.sandbox_name}'"
-                        if self.sandbox_name is not None and self.container_name is None
-                        else "agent"
-                    )
+                    target_desc = f"VM sandbox '{self.sandbox_name}'" if onto_vm_host else "agent"
                     logger.info(f"Loaded URL into {target_desc}: {url} -> {dest}")
 
             await asyncio.gather(*(_load_one(u, f) for u, f in zip(urls, files)))
@@ -734,6 +739,23 @@ async def _load_universe_onto_vm(sandbox, universe, destination: str) -> list[st
     return loaded
 
 
+async def _load_universe_into_sandbox_container(sandbox, universe, destination: str) -> list[str]:
+    """Stage each file in `universe` into a container-mode sandbox, as an agent's container is loaded."""
+    from agent_env.providers.sandbox_providers.sandbox import stage_files_into_container
+
+    file_artifacts = universe.get_file_artifacts()
+    if not file_artifacts:
+        logger.warning(
+            f"FileArtifactUniverse '{universe.id}' v{universe.version} has no files; nothing to load"
+        )
+        return []
+    logger.info(
+        f"Loading FileArtifactUniverse '{universe.id}' v{universe.version} ({len(file_artifacts)} file(s)) "
+        f"into the container sandbox at {destination}"
+    )
+    return list(await stage_files_into_container(sandbox, file_artifacts, destination))
+
+
 async def _load_universe_into_container(sandbox, container_name: str, universe, destination: str) -> list[str]:
     """For each file in `universe`: pull from S3 to a VM temp path, then `docker cp` into `container_name` at `destination/<rel_path>`."""
     file_artifacts = universe.get_file_artifacts()

src/agent_env/task_step/task_steps/verifiers/verify_sandbox.py

@@ -28,6 +28,7 @@
 import uuid
 from typing import ClassVar, Optional
 
+from agent_env.providers.sandbox_providers.sandbox import VmSandbox
 from agent_env.providers.sandbox_providers.sandbox_provider import (
     SANDBOX_MODE_VM,
     build_sandbox_provider,
@@ -118,13 +119,21 @@ def from_dict(cls, data: dict) -> "VerifySandboxTaskStep":
 
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
         on_agent = self.sandbox_name is None
-        sandbox = await (self._agent_sandbox(context) if on_agent else self._deployed_sandbox(context))
+        if on_agent:
+            sandbox = await self._agent_sandbox(context)
+        else:
+            deployed, sandbox = await self._deployed_sandbox(context)
         logger.info(f"Connected to sandbox {sandbox.sandbox_id} (mode={sandbox.mode})")
-        container = (
-            await find_agent_container(sandbox)
-            if on_agent and sandbox.mode == SANDBOX_MODE_VM
-            else None
-        )
+        if on_agent:
+            container = await find_agent_container(sandbox) if sandbox.mode == SANDBOX_MODE_VM else None
+        else:
+            # A container-mode sandbox on a VM-backed provider runs its image beside the host, and reattaches as a VM:
+            # its files are reached in the container, decided by the mode deploy_sandbox recorded.
+            container = (
+                sandbox.container_name
+                if deployed.sandbox_mode != SANDBOX_MODE_VM and isinstance(sandbox, VmSandbox)
+                else None
+            )
         if container:
             logger.info(f"Using agent container: {container}")
 
@@ -204,7 +213,7 @@ async def _deployed_sandbox(self, context: TaskStepContext):
             if deployed.sandbox_type
             else get_sandbox_provider()
         )
-        return await provider.get_sandbox(deployed.sandbox_id)
+        return deployed, await provider.get_sandbox(deployed.sandbox_id)
 
     async def _eval_criterion(self, sandbox, container: Optional[str], criterion: dict) -> dict:
         rtype = criterion["type"]
@@ -225,7 +234,8 @@ async def _eval_criterion(self, sandbox, container: Optional[str], criterion: di
     @staticmethod
     def _probe_args(sandbox, container: Optional[str], cmd: tuple[str, ...]) -> tuple[str, ...]:
         if container is not None:
-            return ("sudo", "docker", "exec", container, *cmd)
+            # As root, like the steps that write into the container: a non-root image user can't read their files.
+            return ("sudo", "docker", "exec", "-u", "0", container, *cmd)
         if sandbox.mode == SANDBOX_MODE_VM:
             return ("sudo", *cmd)
         return cmd