refactor(sandbox): object-store names for sandbox object copies; read handed-in urls from the store that holds them

src/agent_env/a2a_agent/a2a_agent.py

@@ -241,11 +241,11 @@ async def install_cli(deployed: "DeployedA2AAgent", cli_artifact: "CliArtifact",
         entrypoint_path = f"{target_root}/{cli_artifact.entrypoint}"
         env_dir = posixpath.dirname(entrypoint_path)
 
-        # write .env first — write_file_from_text creates the parent dir; write_file_from_s3 does not.
+        # write .env first — write_file_from_text creates the parent dir; write_file_from_object does not.
         await sandbox.write_file_from_text(f"AGENT_ENV_GATEWAY_URL={gateway_url}\n", f"{env_dir}/.env")
 
         for rel_path, file_artifact in cli_artifact.get_cli_files().get_file_artifacts().items():
-            await sandbox.write_file_from_s3(file_artifact.object_url, f"{target_root}/{rel_path}")
+            await sandbox.write_file_from_object(file_artifact.object_url, f"{target_root}/{rel_path}")
 
         if isinstance(sandbox, VmSandbox):
             await sandbox.exec_script(f"docker exec -u 0 {shlex.quote(sandbox.container_name)} chmod +x {entrypoint_path}")

src/agent_env/env/env.py

@@ -326,9 +326,9 @@ async def load_file_artifact_universe(
 
         Files land at ``<destination>/<filename>``. Nested relative paths
         (e.g. ``subdir/file.txt``) are supported — parent dirs are created
-        as needed. Each file is fetched directly from S3 by the sandbox via
-        a presigned URL, so we never round-trip the bytes through this
-        process.
+        as needed. The sandbox fetches each file directly from the object
+        store through a signed URL where the store can sign one, so the bytes
+        don't round-trip through this process.
 
         Default implementation uses ``self._sandbox`` (set by ``deploy()`` /
         ``from_deployed_env()`` on most env subclasses). Subclasses that
@@ -373,7 +373,7 @@ async def load_file_artifact_universe(
         for idx, (filename, file_artifact) in enumerate(file_artifacts.items(), 1):
             dest_path = loaded[filename]
             logger.info(f"  [{idx}/{total}] {file_artifact.object_url} -> {dest_path}")
-            await sandbox.load_s3_file(file_artifact.object_url, dest_path)
+            await sandbox.load_object_file(file_artifact.object_url, dest_path)
 
         logger.info(
             f"Loaded FileArtifactUniverse '{file_artifact_universe.id}' "

src/agent_env/env/envs/mcp_server.py

@@ -226,7 +226,7 @@ async def _staged_artifact_size(self, file_artifact) -> int | None:
     async def _copy_artifact_into_container(self, file_artifact) -> str:
         container_path = f"/data/{file_artifact.filename}"
         if self._sandbox.mode == SANDBOX_MODE_CONTAINER:
-            await self._sandbox.write_file_from_s3(file_artifact.object_url, container_path)
+            await self._sandbox.write_file_from_object(file_artifact.object_url, container_path)
         else:
             # Stage on the disk-backed app dir, not /tmp (tmpfs/RAM): a large
             # artifact (github ~8GB) OOM-kills the copy under concurrent loads.
@@ -235,7 +235,7 @@ async def _copy_artifact_into_container(self, file_artifact) -> str:
             stage_dir = f"{GATEWAY_APP_DIR}/_artifact_staging"
             vm_temp_path = f"{stage_dir}/{self.environment_name}-{uuid.uuid4().hex[:8]}-{file_artifact.filename}"
             await self._sandbox.exec_script(f"mkdir -p {stage_dir}")
-            await self._sandbox.load_s3_file(file_artifact.object_url, vm_temp_path)
+            await self._sandbox.load_object_file(file_artifact.object_url, vm_temp_path)
             container_id = await self._env_provider._get_container_id(self._sandbox, self.environment_name)
             await self._sandbox.exec_script(f"docker exec {container_id} mkdir -p /data")
             await self._sandbox.docker_cp(vm_temp_path, f"{container_id}:{container_path}")

src/agent_env/env/envs/multi_env.py

@@ -353,9 +353,9 @@ async def _load_one(idx: int, artifact):
                 dest_path = f"{dest_dir}/{file_artifact.filename}"
                 logger.info(f"Downloading metadata '{key}' -> {dest_path}")
                 if self._sandbox.mode == SANDBOX_MODE_CONTAINER:
-                    await self._sandbox.write_file_from_s3(file_artifact.object_url, dest_path)
+                    await self._sandbox.write_file_from_object(file_artifact.object_url, dest_path)
                 else:
-                    await self._sandbox.load_s3_file(file_artifact.object_url, dest_path)
+                    await self._sandbox.load_object_file(file_artifact.object_url, dest_path)
                 metadata_filepaths[key] = dest_path
                 logger.info(f"Successfully downloaded metadata '{key}'")
         return LoadEnvironmentUniverseArtifactResult(

src/agent_env/env/envs/website.py

@@ -185,7 +185,7 @@ async def _copy_artifact_into_container(self, file_artifact) -> str:
         container_path = f"/tmp/data/{filename}"
         compose = f"docker compose -f {DOCKER_COMPOSE_PATH}"
 
-        await self._sandbox.load_s3_file(file_artifact.object_url, vm_path)
+        await self._sandbox.load_object_file(file_artifact.object_url, vm_path)
         await self._sandbox.exec_script(f"{compose} exec -T {backend_service} mkdir -p /tmp/data")
         await self._sandbox.exec_script(f"{compose} cp {vm_path} {backend_service}:{container_path}")
         await self._sandbox.exec_script(f"rm -f {vm_path}")

src/agent_env/providers/sandbox_providers/modal_sandbox.py

@@ -152,8 +152,8 @@ async def _write_stream_via_exec(
             stderr = await proc.stderr.read.aio()
             raise RuntimeError(f"write to {destination_path} failed (exit={rc}): {stderr!r}")
 
-    async def write_file_from_s3(self, s3_url: str, destination_path: str) -> None:
-        body = await asyncio.to_thread(get_config().get_object_store().open, s3_url)
+    async def write_file_from_object(self, object_url: str, destination_path: str) -> None:
+        body = await asyncio.to_thread(get_config().get_object_store_at(object_url).open, object_url)
         with contextlib.closing(body):
 
             async def _aiter() -> AsyncIterator[bytes]:

src/agent_env/providers/sandbox_providers/modal_vm_sandbox.py

@@ -108,7 +108,7 @@
 class ModalVmSandbox(VmSandbox):
     """A Modal VM sandbox (vm_runtime) presented through the VmSandbox contract.
 
-    Subclasses VmSandbox so it inherits exec_script / load_docker_images / load_s3_file /
+    Subclasses VmSandbox so it inherits exec_script / load_docker_images / load_object_file /
     write_file_* unchanged — they are built on exec_with_output ->
     exec + an in-VM Docker daemon, which this class provides. Only the Modal-specific
     pieces (raw exec, terminate, dockerd startup) are implemented here.
@@ -212,7 +212,7 @@ async def _ensure_dockerd_running(self) -> None:
 
     async def _download_object_to_vm(self, object_url: str, vm_path: str) -> None:
         """Presigned objects go through aria2c (see _DL_*); one a store cannot presign goes over stdin."""
-        store = get_config().get_object_store()
+        store = get_config().get_object_store_at(object_url)
         signed = await asyncio.to_thread(store.signed_get_url, object_url)
         signed_at = time.monotonic()
         if signed is None:

src/agent_env/providers/sandbox_providers/sandbox.py

@@ -15,6 +15,7 @@
 import tempfile
 import time
 import uuid
+import warnings
 import weakref
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, replace
@@ -173,9 +174,17 @@ async def exec_with_output(self, *args: str) -> tuple[int, str, str]:
         exit_code = await process.wait()
         return exit_code, stdout.decode(), stderr.decode()
 
+    async def write_file_from_object(self, object_url: str, destination_path: str) -> None:
+        """Write the object store's object at ``object_url`` into the agent process's filesystem at
+        ``destination_path``."""
+        raise NotImplementedError(f"{self.__class__.__name__} does not support write_file_from_object")
+
     async def write_file_from_s3(self, s3_url: str, destination_path: str) -> None:
-        """Write a file from S3 into the agent process's filesystem at destination_path."""
-        raise NotImplementedError(f"{self.__class__.__name__} does not support write_file_from_s3")
+        """Deprecated: ``write_file_from_object``."""
+        warnings.warn(
+            "Sandbox.write_file_from_s3 is deprecated; use write_file_from_object", DeprecationWarning, stacklevel=2
+        )
+        await self.write_file_from_object(s3_url, destination_path)
 
     async def write_file_from_url(self, url: str, destination_path: str) -> None:
         """Download an HTTP(S) URL into the agent process's filesystem at destination_path."""
@@ -280,7 +289,7 @@ async def _signed_image_urls(artifacts: list) -> list[str | None]:
         """Each image tarball's signed URL, or None where the store cannot sign one. Signed
         concurrently and off the event loop, since a remote signer is a network round trip."""
         logger.info(f"Loading {len(artifacts)} Docker image(s) into the sandbox...")
-        object_store = get_config().get_object_store()
+        config = get_config()
         slots = _sign_slots.setdefault(asyncio.get_running_loop(), asyncio.Semaphore(_CONCURRENT_SIGNS))
         failed = False
 
@@ -290,7 +299,8 @@ async def sign(artifact) -> str | None:
                 if failed:
                     return None
                 try:
-                    return await asyncio.to_thread(object_store.signed_get_url, artifact.tar_gz_object_url)
+                    store = config.get_object_store_at(artifact.tar_gz_object_url)
+                    return await asyncio.to_thread(store.signed_get_url, artifact.tar_gz_object_url)
                 except BaseException:
                     failed = True
                     raise
@@ -326,13 +336,18 @@ async def _load_docker_images(self, artifacts: list, signed_urls: list[str | Non
                 raise RuntimeError(f"{artifact.image_name} image not found. stdout: {stdout}")
         logger.info("  All images loaded successfully")
 
+    async def load_object_file(self, object_url: str, destination_path: str) -> None:
+        """Download the object store's object at ``object_url`` onto the VM host at ``destination_path``."""
+        await self._download_object_to_vm(object_url, destination_path)
+
     async def load_s3_file(self, s3_url: str, destination_path: str) -> None:
-        """Download an object from the object store into the sandbox."""
-        await self._download_object_to_vm(s3_url, destination_path)
+        """Deprecated: ``load_object_file``."""
+        warnings.warn("VmSandbox.load_s3_file is deprecated; use load_object_file", DeprecationWarning, stacklevel=2)
+        await self.load_object_file(s3_url, destination_path)
 
     async def _download_object_to_vm(self, object_url: str, vm_path: str) -> None:
         """Place object_url onto the VM host at vm_path, backend-agnostically."""
-        object_store = get_config().get_object_store()
+        object_store = get_config().get_object_store_at(object_url)
         signed = await asyncio.to_thread(object_store.signed_get_url, object_url)
         if signed is not None:
             await self.exec_script(f"curl -fsSL {CURL_RETRY_FLAGS} {shlex.quote(signed)} -o {shlex.quote(vm_path)}")
@@ -378,10 +393,10 @@ def _staging_path(kind: str, destination_path: str) -> str:
         host's /tmp, as local ones do."""
         return f"/tmp/_{kind}_{uuid.uuid4().hex[:12]}_{destination_path.replace('/', '_').lstrip('_')}"
 
-    async def write_file_from_s3(self, s3_url: str, destination_path: str) -> None:
-        vm_path = self._staging_path("s3", destination_path)
+    async def write_file_from_object(self, object_url: str, destination_path: str) -> None:
+        vm_path = self._staging_path("obj", destination_path)
         try:
-            await self.load_s3_file(s3_url, vm_path)
+            await self.load_object_file(object_url, vm_path)
             await self._copy_into_container(vm_path, destination_path)
         finally:
             await self._remove_vm_temp_file(vm_path)
@@ -458,7 +473,7 @@ async def stage_files_into_container(sandbox: Sandbox, file_artifacts: dict[str,
     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 sandbox.write_file_from_object(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/install_agent.py

@@ -227,7 +227,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         agent_ctx_tar = f"{work_dir}/agent-ctx.tar.gz"
         await sandbox.exec_script(f"rm -rf {shlex.quote(work_dir)} && mkdir -p {shlex.quote(work_dir)}")
         logger.info(f"Downloading agent build context {agent.docker_image_artifact.build_context_object_url} -> {agent_ctx_tar}")
-        await sandbox.load_s3_file(agent.docker_image_artifact.build_context_object_url, agent_ctx_tar)
+        await sandbox.load_object_file(agent.docker_image_artifact.build_context_object_url, agent_ctx_tar)
 
         # Resolve each required_param via the registry, shlex.quote so str.format
         # substitution produces shell-safe tokens.

src/agent_env/task_step/task_steps/load_artifact.py

@@ -264,7 +264,7 @@ def _resolve_inputs(self, context: TaskStepContext) -> _LoadInputs:
         return _LoadInputs(artifacts=resolved_artifacts, urls=urls, destination_path=destination_path)
 
     async def _load_url_onto_vm(self, sandbox, url: str, destination_path: str) -> None:
-        """Host counterpart of ``sandbox.load_s3_file``; ``write_file_from_url`` targets a container instead."""
+        """Host counterpart of ``sandbox.load_object_file``; ``write_file_from_url`` targets a container instead."""
         from agent_env.providers.sandbox_providers.sandbox import CURL_RETRY_FLAGS
 
         parent = posixpath.dirname(destination_path)
@@ -643,7 +643,7 @@ async def _stage_environment_payload_into_container(
     if isinstance(sandbox, VmSandbox):
         vm_payload = f"/tmp/_svc_{token}_{safe_name}"
         vm_stage = f"/tmp/_svc_stage_{token}"
-        await sandbox.load_s3_file(file_artifact.object_url, vm_payload)
+        await sandbox.load_object_file(file_artifact.object_url, vm_payload)
         try:
             # exec_script raises on a non-zero exit, but only with the raw
             # script output — re-raise with the artifact context so a failed
@@ -679,7 +679,7 @@ async def _stage_environment_payload_into_container(
         # Container-mode sandbox: the sandbox IS the agent container — expand
         # in place at the destination.
         payload_tmp = f"/tmp/_svc_{token}_{safe_name}"
-        await sandbox.write_file_from_s3(file_artifact.object_url, payload_tmp)
+        await sandbox.write_file_from_object(file_artifact.object_url, payload_tmp)
         try:
             exit_code, out, err = await sandbox.exec_with_output(
                 "python3", "-c", _ENVIRONMENT_PAYLOAD_EXPAND_SCRIPT, payload_tmp, destination
@@ -712,7 +712,7 @@ async def _stage_environment_payload_into_container(
 
 
 async def _load_universe_onto_vm(sandbox, universe, destination: str) -> list[str]:
-    """Pull each file in `universe` from S3 straight onto the VM host — no temp file, no copy inward."""
+    """Pull each file in `universe` from the object store straight onto the VM host — no temp file, no copy inward."""
     file_artifacts = universe.get_file_artifacts()
     if not file_artifacts:
         logger.warning(
@@ -734,7 +734,7 @@ async def _load_universe_onto_vm(sandbox, universe, destination: str) -> list[st
         logger.info(f"  [{idx}/{total}] {fa.object_url} -> {dest_path}")
         if parent and parent != destination:
             await sandbox.exec_script(f"mkdir -p {shlex.quote(parent)}")
-        await sandbox.load_s3_file(fa.object_url, dest_path)
+        await sandbox.load_object_file(fa.object_url, dest_path)
         loaded.append(filename)
     return loaded
 
@@ -757,7 +757,7 @@ async def _load_universe_into_sandbox_container(sandbox, universe, 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>`."""
+    """For each file in `universe`: pull from the object store to a VM temp path, then `docker cp` into `container_name` at `destination/<rel_path>`."""
     file_artifacts = universe.get_file_artifacts()
     if not file_artifacts:
         logger.warning(
@@ -780,7 +780,7 @@ async def _load_universe_into_container(sandbox, container_name: str, universe,
         parent = posixpath.dirname(dest_path)
         vm_temp = f"/tmp/_load_{uuid.uuid4().hex[:8]}_{filename.replace('/', '_')}"
         logger.info(f"  [{idx}/{total}] {fa.object_url} -> {container_name}:{dest_path}")
-        await sandbox.load_s3_file(fa.object_url, vm_temp)
+        await sandbox.load_object_file(fa.object_url, vm_temp)
         if parent and parent != destination:
             await sandbox.exec_script(
                 f"docker exec {shlex.quote(container_name)} mkdir -p {shlex.quote(parent)}"

src/agent_env/task_step/task_steps/run_docker_container.py

@@ -341,7 +341,7 @@ async def _stage_from_universe(self, sandbox, sandbox_id: str, work_dir: str) ->
         for idx, (filename, fa) in enumerate(file_artifacts.items(), 1):
             dest_path = loaded[filename]
             logger.info(f"  [{idx}/{total}] {fa.object_url} -> {dest_path}")
-            await sandbox.load_s3_file(fa.object_url, dest_path)
+            await sandbox.load_object_file(fa.object_url, dest_path)
 
     @staticmethod
     async def _stage_zip_from_url(sandbox, url: str, work_dir: str) -> None:
@@ -369,7 +369,7 @@ async def _stage_zip_from_url(sandbox, url: str, work_dir: str) -> None:
         archive_path = posixpath.join(work_dir, "_context.zip")
         logger.info(f"Downloading docker context from {url} into {work_dir}")
         if from_store:
-            await sandbox.load_s3_file(url, archive_path)
+            await sandbox.load_object_file(url, archive_path)
         else:
             await sandbox.exec_script(
                 f"curl -fsSL {shlex.quote(url)} -o {shlex.quote(archive_path)}"