feat: agents on the local sandbox provider move objects through the local store's grants

src/agent_env/a2a_agent/a2a_agent.py

@@ -20,6 +20,13 @@
     invoke_transfer,
     skill_add_call,
 )
+from agent_env.providers.sandbox_providers.local_sandbox import (
+    LOCAL_TRUST_ENV,
+    LocalSandbox,
+    LocalSandboxProvider,
+    local_grant_trust,
+    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
@@ -198,6 +205,7 @@ async def register_skill(
             description=description,
             skill_md=skill_md,
             object_url=object_url,
+            sandbox_type=deployed.sandbox_type,
         )
         return await invoke_transfer(
             deployed.a2a_url + add_path,
@@ -518,13 +526,17 @@ async def close(self) -> None:
         self._sandbox = None
 
     async def _run_container(self, image_name: str, a2a_port: int, merged_env: dict[str, str], enable_docker: bool = False) -> None:
-        agent_env = dict(merged_env)
+        # On a local VM sandbox the container runs on this host, so it is given the local CA as the provider's are.
+        trust_dir = await asyncio.to_thread(local_grant_trust) if isinstance(self._sandbox, LocalSandbox) else None
+        agent_env = dict(merged_env) if trust_dir is None else {**LOCAL_TRUST_ENV, **merged_env}
         setup_script = ""
-        network_flag = ""
+        network_flag = "" if trust_dir is None or not LocalSandboxProvider.EXTRA_CONTAINER_RUN_ARGS else (
+            f"{LocalSandboxProvider.EXTRA_CONTAINER_RUN_ARGS} \\\n    "
+        )
         if enable_docker:
             logger.info("enable_docker: starting rootless Docker-in-Docker sidecar for the agent (no host socket)")
             setup_script = self._dind_setup_script()
-            network_flag = f"--network {_DIND_NETWORK} \\\n    "
+            network_flag += f"--network {_DIND_NETWORK} \\\n    "
             agent_env["DOCKER_HOST"] = f"tcp://{_DIND_CONTAINER}:{_DIND_PORT}"
 
         env_flags = []
@@ -536,14 +548,17 @@ async def _run_container(self, image_name: str, a2a_port: int, merged_env: dict[
 
         run_script = f"""#!/bin/bash
 set -e
-{setup_script}docker run -d \\
+{setup_script}docker {"run -d" if trust_dir is None else "create"} \\
     --name {self._sandbox.container_name} \\
     {publish} \\
     {network_flag}{env_str} \\
     {image_name} > /dev/null
-sleep 2
+{"sleep 2" if trust_dir is None else ""}
 """
         await self._sandbox.exec_script(run_script)
+        if trust_dir is not None:
+            await start_trusting(self._sandbox, self._sandbox.container_name, trust_dir)
+            await asyncio.sleep(2)
 
     def _dind_setup_script(self) -> str:
         # Constants become shell vars so the body stays a raw string (no f-string

src/agent_env/a2a_agent/object_transfer.py

@@ -38,6 +38,7 @@
 from agent_env.a2a_agent.protocol import raise_for_extension_status
 from agent_env.store.base import GrantUnavailableError
 from agent_env.store.object_store import DEFAULT_CONTENT_TYPE, ObjectStore
+from agent_env.store.object_store.local.grant_server import unreachable_hint
 
 logger = logging.getLogger(__name__)
 
@@ -47,6 +48,7 @@
 _Response = TypeVar("_Response", bound=BaseModel)
 _MEDIA_TYPE = re.compile(r"[!#$&^_.+\-|~0-9a-z]+/[!#$&^_.+\-|~0-9a-z]+")
 _INCREMENT_NAME = re.compile(r"^(?P<sequence>[0-9]{6})(?:\.[A-Za-z0-9][A-Za-z0-9._-]*)?$")
+_TRANSFER_UNAVAILABLE = "transfer_unavailable"  # the SDK's code for a store it could not reach
 
 # The time budget of one transfer, outermost first: a grant outlives agent-env's wait for the
 # agent's answer, which outlasts the SDK's retries of stalled connections
@@ -97,13 +99,20 @@ def choose_transfer(
     objects: Collection[str] | None = None,
     legacy: Collection[str] | None = None,
     store: ObjectStore,
+    sandbox_type: str | None,
 ) -> TransferMode | None:
     """How one extension call moves its objects, given the fields each form it can take sends.
 
-    Objects need a store that issues grants. A method without a declared request predates
-    variant negotiation and takes the legacy form. None: the agent takes neither form.
+    Objects need a store that issues grants reaching the agent's sandbox, of ``sandbox_type``
+    (None: unknown). A method without a declared request predates variant negotiation and takes
+    the legacy form. None: the agent takes neither form.
     """
-    if objects is not None and _accepts(method, objects) and store.supports_transfer_grants:
+    if (
+        objects is not None
+        and _accepts(method, objects)
+        and store.supports_transfer_grants
+        and store.grants_reach(sandbox_type)
+    ):
         return "objects"
     if legacy is not None and (
         method is None or "request" not in method or _accepts(method, legacy)
@@ -123,13 +132,19 @@ def _require_object_form(
     store: ObjectStore,
     *,
     operation: str,
+    sandbox_type: str | None,
 ) -> None:
-    """Refuse ``operation`` unless the agent takes its object form and the store issues the
-    grants that form carries: agent-env moves objects no other way."""
+    """Refuse ``operation`` unless the agent takes its object form and the store issues grants
+    that reach the agent's sandbox, of ``sandbox_type``: agent-env moves objects no other way."""
     if not _accepts(method, _fields(model)):
         raise RuntimeError(f"{operation}: the agent does not advertise the object form")
     if not store.supports_transfer_grants:
         raise RuntimeError(f"{operation}: the object store does not issue transfer grants")
+    if not store.grants_reach(sandbox_type):
+        raise RuntimeError(
+            f"{operation}: the object store's grants do not reach agents on the "
+            f"{sandbox_type or 'unknown'!r} sandbox provider"
+        )
 
 
 @dataclass(frozen=True)
@@ -155,7 +170,13 @@ async def invoke_transfer(
     async with httpx.AsyncClient() as client:
         send = client.post if verb == "POST" else client.put
         resp = await send(url, json=call.payload, timeout=timeout)
-    raise_for_extension_status(resp, operation=operation, include_body=call.mode == "legacy")
+    try:
+        raise_for_extension_status(resp, operation=operation, include_body=call.mode == "legacy")
+    except httpx.HTTPStatusError as exc:
+        hint = unreachable_hint(call.payload) if call.mode == "objects" else None
+        if hint is None or not str(exc).endswith(f": {_TRANSFER_UNAVAILABLE}"):
+            raise
+        raise httpx.HTTPStatusError(f"{exc}. {hint}", request=exc.request, response=exc.response) from exc
     body = resp.json()
     if call.mode == "objects" and response_model is not None:
         return parse_response(response_model, body, operation=operation)
@@ -283,18 +304,23 @@ def skill_add_call(
     description: str,
     skill_md: str | None = None,
     object_url: str | None = None,
+    sandbox_type: str | None,
 ) -> TransferCall:
     """The skill ``add`` call for a skill given as SKILL.md text, sent inline, or as the objects
     under ``object_url``, sent as a bundle of read grants."""
     base = {"name": name, "description": description}
     if object_url is not None:
-        _require_object_form(method, BundleSkillRequest, store, operation="skill add")
+        _require_object_form(
+            method, BundleSkillRequest, store, operation="skill add", sandbox_type=sandbox_type
+        )
         request = skill_bundle_request(
             store, name=name, description=description, object_url=object_url
         )
         return TransferCall("objects", request.model_dump(mode="json"))
     if skill_md is not None:
-        if choose_transfer(method, legacy=_fields(InlineSkillRequest), store=store) is None:
+        if choose_transfer(
+            method, legacy=_fields(InlineSkillRequest), store=store, sandbox_type=sandbox_type
+        ) is None:
             raise RuntimeError("Agent does not advertise the inline skill variant")
         return TransferCall("legacy", {**base, "skill_md": skill_md})
     return TransferCall("legacy", base)
@@ -307,13 +333,15 @@ def snapshot_save_call(
     agent_name: str,
     context_id: str,
     capture_prefix: str,
+    sandbox_type: str | None,
 ) -> TransferCall:
     """The snapshot ``save`` call that writes one capture below ``capture_prefix``."""
     _require_object_form(
         method,
         ObjectSnapshotSaveRequest,
         store,
         operation=f"snapshot save on agent '{agent_name}'",
+        sandbox_type=sandbox_type,
     )
     objects = {
         name: write_object(store, url, media_type=_OPAQUE, max_bytes=max_bytes)
@@ -333,6 +361,7 @@ def snapshot_load_call(
     bundle_url: str,
     file_names: Collection[str],
     target_context_id: str | None,
+    sandbox_type: str | None,
 ) -> TransferCall:
     """The snapshot ``load`` call that restores the capture at ``bundle_url``. Only a portable
     capture can be restored; an older one holds the runtime's own files."""
@@ -341,7 +370,9 @@ def snapshot_load_call(
         raise RuntimeError(
             f"{operation}: {bundle_url} is not a portable snapshot, so it cannot be restored"
         )
-    _require_object_form(method, ObjectSnapshotLoadRequest, store, operation=operation)
+    _require_object_form(
+        method, ObjectSnapshotLoadRequest, store, operation=operation, sandbox_type=sandbox_type
+    )
     objects = {
         name: read_object(store, url, media_type=_OPAQUE, max_bytes=max_bytes)
         for name, url, max_bytes in _snapshot_objects(store, bundle_url)
@@ -368,11 +399,18 @@ def changelog_enable_call(
     agent_name: str,
     namespace_url: str,
     expires_in: int,
+    sandbox_type: str | None,
 ) -> TransferCall:
     """The ``enable-changelog`` call that captures below ``namespace_url``. Raises
     GrantUnavailableError when the store cannot sign a namespace grant that lasts ``expires_in``."""
     operation = f"changelog enable on agent '{agent_name}'"
-    _require_object_form(method, NamespaceChangelogEnableRequest, store, operation=operation)
+    _require_object_form(
+        method,
+        NamespaceChangelogEnableRequest,
+        store,
+        operation=operation,
+        sandbox_type=sandbox_type,
+    )
     try:
         grant = namespace_grant(
             store, namespace_url, limits=CHANGELOG_LIMITS, expires_in=expires_in
@@ -394,6 +432,7 @@ def changelog_apply_call(
     up_to_tool_call_exclusive: int | None = None,
     resume_conversation: bool = False,
     target_context_id: str | None = None,
+    sandbox_type: str | None,
 ) -> TransferCall:
     """The ``apply-changelog`` call that replays the capture at ``source_url``: read grants for
     its increments before the cutoff, in sequence order, and none when the cutoff precedes the
@@ -403,6 +442,7 @@ def changelog_apply_call(
         ObjectChangelogApplyRequest,
         store,
         operation=f"changelog apply on agent '{agent_name}'",
+        sandbox_type=sandbox_type,
     )
     cutoff = math.inf if up_to_tool_call_exclusive is None else up_to_tool_call_exclusive
     described = read_objects_under(
@@ -453,12 +493,17 @@ def trajectory_mode(
     store: ObjectStore,
     *,
     by: Literal["task_id", "context_id"],
+    sandbox_type: str | None,
 ) -> TransferMode | None:
     """How a trajectory ``get`` selecting by ``by`` moves the trajectory: uploaded by the agent
     through a grant, or returned inline."""
     objects_model, inline_model = _TRAJECTORY_FORMS[by]
     return choose_transfer(
-        method, objects=_fields(objects_model), legacy=_fields(inline_model), store=store
+        method,
+        objects=_fields(objects_model),
+        legacy=_fields(inline_model),
+        store=store,
+        sandbox_type=sandbox_type,
     )
 
 

src/agent_env/a2a_agent/validator.py

@@ -29,6 +29,9 @@
     invoke_transfer,
 )
 from agent_env.config import get_config
+from agent_env.providers.sandbox_providers.chained_sandbox_provider import ChainedSandboxProvider
+from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox, LocalSandboxProvider
+from agent_env.providers.sandbox_providers.sandbox_provider import get_agent_sandbox_provider
 from agent_env.store.ids import derive_id, key_segment
 from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_modalities import (
     AUDIO_M4A_EXPECTED,
@@ -60,6 +63,15 @@
 logger = logging.getLogger(__name__)
 
 
+def _probe_agent_sandbox_type() -> str | None:
+    """The sandbox type the validation agents will deploy on, as far as it is known before they do: the local
+    provider's when it is the agent provider, or first in its chain, which runs the agent unless it fails."""
+    provider = get_agent_sandbox_provider()
+    if isinstance(provider, ChainedSandboxProvider):
+        provider = provider.providers[0]
+    return LocalSandbox.type if isinstance(provider, LocalSandboxProvider) else None
+
+
 @dataclass
 class _UploadedFixtures:
     """Object-store-hosted probe fixtures created at validation runtime."""
@@ -656,6 +668,7 @@ def record(*, supported, advertised, save_ok, apply_ok, roundtrip_ok, note=""):
                 get_config().get_object_store(),
                 agent_name=apply_agent_name,
                 source_url=capture_source,
+                sandbox_type=apply_agent.sandbox_type,
             )
         except RuntimeError as e:
             record(supported=False, advertised=advertised, save_ok=save_ok,
@@ -801,8 +814,10 @@ def _upload_probe_fixtures(agent: "A2AAgent", skill_object_url: str) -> _Uploade
 
         png_object_uri = store.put(f"{prefix}red.png", base64.b64decode(IMAGE_PROBE_PNG_B64), content_type="image/png", allow_overwrite=True)
         png_signed_url = store.signed_get_url(png_object_uri)
+        if png_signed_url is None and store.supports_transfer_grants and store.grants_reach(_probe_agent_sandbox_type()):
+            png_signed_url = store.issue_read_grant(png_object_uri).url
         if png_signed_url is None:
-            raise RuntimeError("A2A validation requires a signable object store for the presigned-URI probe.")
+            raise RuntimeError("A2A validation requires an object store that signs URLs or issues grants, for the presigned-URI probe.")
         logger.info(f"Uploaded probe fixture to {png_object_uri}")
 
         mp4_object_uri = store.put(f"{prefix}clip.mp4", base64.b64decode(VIDEO_PROBE_MP4_B64), content_type="video/mp4", allow_overwrite=True)

src/agent_env/providers/sandbox_providers/chained_sandbox_provider.py

@@ -18,6 +18,11 @@ def __init__(self, providers: list[SandboxProvider]):
             raise ValueError("ChainedSandboxProvider requires at least one provider")
         self._providers = providers
 
+    @property
+    def providers(self) -> tuple[SandboxProvider, ...]:
+        """The providers tried, in order."""
+        return tuple(self._providers)
+
     @staticmethod
     def filter_sandbox_providers(
         providers: list[SandboxProvider], policy: NetworkPolicy | None