diff --git a/src/agent_env/a2a_agent/a2a_agent.py b/src/agent_env/a2a_agent/a2a_agent.py index 156b763f..e716c8a2 100644 --- a/src/agent_env/a2a_agent/a2a_agent.py +++ b/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 diff --git a/src/agent_env/a2a_agent/object_transfer.py b/src/agent_env/a2a_agent/object_transfer.py index dd41dbf3..71471c35 100644 --- a/src/agent_env/a2a_agent/object_transfer.py +++ b/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[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,6 +333,7 @@ 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( @@ -314,6 +341,7 @@ def snapshot_save_call( 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, ) diff --git a/src/agent_env/a2a_agent/validator.py b/src/agent_env/a2a_agent/validator.py index 8c952532..84133b62 100644 --- a/src/agent_env/a2a_agent/validator.py +++ b/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) diff --git a/src/agent_env/providers/sandbox_providers/chained_sandbox_provider.py b/src/agent_env/providers/sandbox_providers/chained_sandbox_provider.py index e3aa2a9c..a668e7a1 100644 --- a/src/agent_env/providers/sandbox_providers/chained_sandbox_provider.py +++ b/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 diff --git a/src/agent_env/providers/sandbox_providers/local_sandbox.py b/src/agent_env/providers/sandbox_providers/local_sandbox.py index 9e80339f..51c589e9 100644 --- a/src/agent_env/providers/sandbox_providers/local_sandbox.py +++ b/src/agent_env/providers/sandbox_providers/local_sandbox.py @@ -24,6 +24,7 @@ from typing import TYPE_CHECKING, Any, Optional from uuid import uuid4 +from agent_env import config from agent_env.attribution import Attribution from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy, VmSandbox from agent_env.providers.sandbox_providers.sandbox_provider import ( @@ -32,6 +33,9 @@ SandboxProvider, refuse_unenforceable_policy, ) +from agent_env.store.object_store.local.tls import local_ca +from agent_env.store.object_store.local.store import LocalFilesystemObjectStore +from agent_env.store.routing import LocalRunObjectStore if TYPE_CHECKING: from agent_env.store.object_store import ObjectStore @@ -51,6 +55,15 @@ def _runs_in_container(cmd: list[str]) -> bool: _REAP_SECONDS = 5 +# Where a container finds the local transfer CA's trust files, and the variables that point TLS clients at them: +# SSL_CERT_FILE replaces a client's roots, so it gets the public roots plus the CA; NODE_EXTRA_CA_CERTS adds. +_TRUST_DIR = "/etc/agentenv" +LOCAL_TRUST_ENV = { + "SSL_CERT_FILE": f"{_TRUST_DIR}/ca-bundle.pem", + "REQUESTS_CA_BUNDLE": f"{_TRUST_DIR}/ca-bundle.pem", + "NODE_EXTRA_CA_CERTS": f"{_TRUST_DIR}/ca.pem", +} + # Marker dropped in the work dir when this sandbox runs a container, so a later get_sandbox() # (post-run teardown reconstructs the sandbox from disk) knows it owns that container. _CONTAINER_MODE_MARKER = ".agent-container-mode" @@ -281,6 +294,36 @@ def _kill_tree(root: int) -> None: os.kill(pid, signal.SIGKILL) +def local_grant_trust() -> Path | None: + """The trust files a container on this host needs to use the configured object store's grants: the local + transfer CA's, when the store is a local one that hands out grants; None otherwise.""" + store = config.get_config().get_object_store() + if isinstance(store, LocalRunObjectStore): + store = store.local + if isinstance(store, LocalFilesystemObjectStore) and store.supports_transfer_grants: + return local_ca().trust_dir + return None + + +async def start_trusting(sandbox: VmSandbox, container: str, trust_dir: Path) -> None: + """Copy ``trust_dir`` into the created ``container`` where ``LOCAL_TRUST_ENV`` points, then start it. A container + that cannot be given the files or started is removed, so its name is free for the next attempt.""" + try: + await asyncio.to_thread(_copy_into_container, trust_dir, container, _TRUST_DIR) + await sandbox.exec_script(f"docker start {shlex.quote(container)} > /dev/null") + except Exception: + await sandbox.exec_script(f"docker rm -f {shlex.quote(container)} >/dev/null 2>&1 || true") + raise + + +def _copy_into_container(source: Path, container: str, destination: str) -> None: + """Copy what the host directory ``source`` holds to ``destination`` in ``container``. Run directly, not + through a sandbox's shell, which would rewrite a host path under /app.""" + copied = subprocess.run(["docker", "cp", f"{source}/.", f"{container}:{destination}"], capture_output=True, text=True) + if copied.returncode != 0: + raise RuntimeError(f"Could not copy {source} into container {container}: {copied.stderr.strip()}") + + def remove_local_work_dir(sandbox_id: str) -> Path | None: """Delete a local sandbox's work folder, and return it, when it lies directly under the sandbox root and isn't a symlink; anything else, a legacy folder under the temp dir included, is left alone. Raises OSError @@ -374,6 +417,17 @@ async def create_container( raise return sandbox + async def _start_container(self, sandbox: VmSandbox, *, image_name: str, port: int, env: dict[str, str]) -> None: + """Where the configured store hands out local grants, create the container, copy the local transfer CA's + trust files in, then start it, so its TLS clients trust the grant server. Variables the caller sets win.""" + trust_dir = await asyncio.to_thread(local_grant_trust) + if trust_dir is None: + await super()._start_container(sandbox, image_name=image_name, port=port, env=env) + return + args = self._container_args(sandbox, image_name=image_name, port=port, env={**LOCAL_TRUST_ENV, **env}) + await sandbox.exec_script(f"docker create {args} > /dev/null") + await start_trusting(sandbox, sandbox.container_name, trust_dir) + async def create_sandbox( self, *, diff --git a/src/agent_env/providers/sandbox_providers/sandbox_provider.py b/src/agent_env/providers/sandbox_providers/sandbox_provider.py index 0a784735..52975db5 100644 --- a/src/agent_env/providers/sandbox_providers/sandbox_provider.py +++ b/src/agent_env/providers/sandbox_providers/sandbox_provider.py @@ -107,15 +107,7 @@ async def create_container( f"--username {shlex.quote(auth.username)} --password-stdin {shlex.quote(auth.registry)}" ) await sandbox.exec_script(f"docker pull {shlex.quote(image_name)}") - env_flags = " \\\n ".join( - f"-e {shlex.quote(k)}={shlex.quote(v)}" for k, v in env.items() - ) - extra_args = f"{self.EXTRA_CONTAINER_RUN_ARGS} " if self.EXTRA_CONTAINER_RUN_ARGS else "" - publish = " ".join(f"-p {spec}" for spec in port_bindings(sandbox.host_ips, sandbox.host_port(port), port)) - await sandbox.exec_script( - f"docker run -d --name {shlex.quote(sandbox.container_name)} {publish} {extra_args}\\\n " - f"{env_flags} \\\n {shlex.quote(image_name)} > /dev/null" - ) + await self._start_container(sandbox, image_name=image_name, port=port, env=env) sandbox.mode = SANDBOX_MODE_CONTAINER return sandbox except Exception: @@ -125,6 +117,21 @@ async def create_container( pass raise + async def _start_container(self, sandbox: VmSandbox, *, image_name: str, port: int, env: dict[str, str]) -> None: + """Run the pulled ``image_name`` as the sandbox's container, publishing ``port``.""" + args = self._container_args(sandbox, image_name=image_name, port=port, env=env) + await sandbox.exec_script(f"docker run -d {args} > /dev/null") + + def _container_args(self, sandbox: VmSandbox, *, image_name: str, port: int, env: dict[str, str]) -> str: + """The ``docker run`` / ``docker create`` arguments for the sandbox's container.""" + env_flags = " \\\n ".join(f"-e {shlex.quote(k)}={shlex.quote(v)}" for k, v in env.items()) + extra_args = f"{self.EXTRA_CONTAINER_RUN_ARGS} " if self.EXTRA_CONTAINER_RUN_ARGS else "" + publish = " ".join(f"-p {spec}" for spec in port_bindings(sandbox.host_ips, sandbox.host_port(port), port)) + return ( + f"--name {shlex.quote(sandbox.container_name)} {publish} {extra_args}\\\n " + f"{env_flags} \\\n {shlex.quote(image_name)}" + ) + async def get_sandbox(self, sandbox_id: str) -> Sandbox: raise NotImplementedError(f"{type(self).__name__} does not support get_sandbox") diff --git a/src/agent_env/store/object_store/local/grant_server.py b/src/agent_env/store/object_store/local/grant_server.py index 89725141..729c58ab 100644 --- a/src/agent_env/store/object_store/local/grant_server.py +++ b/src/agent_env/store/object_store/local/grant_server.py @@ -12,6 +12,7 @@ import contextlib import functools import ipaddress +import json import logging import os import platform @@ -89,6 +90,24 @@ def grant_server(bind_host: str | None, advertise_host: str | None) -> GrantServ return server +def unreachable_hint(payload: object) -> str | None: + """What to check when an agent could not use grants carried in ``payload``, if any came from a grant + server in this process; None otherwise. Names the server's address, never a grant.""" + text = json.dumps(payload, default=str) + with _servers_lock: + servers = list(_servers.values()) + for server in servers: + origin = server.origin + if origin is not None and f"{origin}/v1/grants/" in text: + return ( + f"The agent may not have reached this process's local grant server at {origin}, which " + f"listens on {server.bind_host}. If the agent's container cannot reach that address, set " + "grant_bind_host (where the server listens) and grant_advertise_host (the host grant URLs " + "name) in the [stores.object] config." + ) + return None + + @functools.cache def default_bind_host() -> str: """Where the grant server listens unless told otherwise. On macOS and Windows, Docker reaches @@ -142,9 +161,16 @@ def issue( op=op, store=self._register(store), expires=expires, key=key, prefix=prefix, max_bytes=max_bytes, content_type=content_type, ) - port = self._start() + self._start() + return f"{self.origin}/v1/grants/{self._signer.sign(claims)}" + + @property + def origin(self) -> str | None: + """The scheme, host and port grant URLs start with; None until the server has started.""" + if self._port is None: + return None host = f"[{self.advertise_host}]" if ":" in self.advertise_host else self.advertise_host - return f"https://{host}:{port}/v1/grants/{self._signer.sign(claims)}" + return f"https://{host}:{self._port}" def close(self) -> None: """Stop serving; grants already issued stop working.""" diff --git a/src/agent_env/store/object_store/local/store.py b/src/agent_env/store/object_store/local/store.py index acbef743..296a41bb 100644 --- a/src/agent_env/store/object_store/local/store.py +++ b/src/agent_env/store/object_store/local/store.py @@ -46,6 +46,8 @@ _LOCK_STRIPES = 256 # keys share this many lock files, so locking leaves no file behind per key _STAGED_PREFIX = "staged-" _STAGED_GRACE_SECONDS = 60 # a staged file younger than this may not be locked by its writer yet +_GRANT_MODES = ("auto", "off") +_LOCAL_SANDBOX_TYPE = "local" # LocalSandbox.type: containers on this host's Docker, which trust the local CA logger = logging.getLogger(__name__) @@ -65,14 +67,22 @@ class LocalFilesystemObjectStore(ObjectStore): A write lands whole or not at all, and keeps any content type but the default. The root-level ``.gitignore``, ``.agentenv-meta`` and ``.agentenv-tmp`` keys are reserved for the store's own files. - Its HTTPS transfer grants are served by a server in this process, started by the first grant: it listens on - ``grant_bind_host`` (by default loopback, or the Docker bridge gateway on Linux) and grant URLs name - ``grant_advertise_host`` (by default ``host.docker.internal``), which must be a local name or a loopback or - private address. A grant works until it expires or this process exits.""" + With ``grants = "auto"`` (the default) it hands agents on the local sandbox provider HTTPS transfer grants, + served by a server in this process, started by the first grant: it listens on ``grant_bind_host`` (by default + loopback, or the Docker bridge gateway on Linux) and grant URLs name ``grant_advertise_host`` (by default + ``host.docker.internal``), which must be a local name or a loopback or private address. A grant works until it + expires or this process exits. ``grants = "off"`` hands out none, so agents get the forms that carry no grant.""" def __init__( - self, root: str | None = None, *, grant_bind_host: str | None = None, grant_advertise_host: str | None = None + self, + root: str | None = None, + *, + grants: str = "auto", + grant_bind_host: str | None = None, + grant_advertise_host: str | None = None, ) -> None: + if grants not in _GRANT_MODES: + raise ValueError(f"grants must be one of {_GRANT_MODES}, got {grants!r}") if grant_bind_host is not None: try: ipaddress.ip_address(grant_bind_host) @@ -81,6 +91,7 @@ def __init__( if grant_advertise_host is not None: check_local_host(grant_advertise_host) self._root = Path(root) if root is not None else state_root() / "object_store" + self.supports_transfer_grants = grants == "auto" self._grant_bind_host = grant_bind_host self._grant_advertise_host = grant_advertise_host self._swept = False @@ -177,6 +188,9 @@ def get_object_key(self, object_url: str) -> str: raise ValueError(f"{object_url!r} is not an object in {root}.") return path.relative_to(root).as_posix() + def grants_reach(self, sandbox_type: str | None) -> bool: + return sandbox_type == _LOCAL_SANDBOX_TYPE + def issue_read_grant(self, object_url: str, *, expires_in: int = 3600) -> HttpGetGrant: url, expires_at = self._grant("get", expires_in, key=self._grant_key(object_url)) return HttpGetGrant(kind="http-get", url=url, expires_at=expires_at) diff --git a/src/agent_env/store/object_store/object_store.py b/src/agent_env/store/object_store/object_store.py index 26f95dcd..ec0fa2da 100644 --- a/src/agent_env/store/object_store/object_store.py +++ b/src/agent_env/store/object_store/object_store.py @@ -43,7 +43,7 @@ class ObjectStore(ABC): """ # Whether this store can serve the three issue_* grant methods; callers check this, not - # the NotImplementedError the defaults raise. + # the NotImplementedError the defaults raise, and grants_reach before handing a grant on. supports_transfer_grants: bool = False # The largest object one upload through a grant can create, where the provider caps it. max_single_upload_bytes: int | None = None @@ -161,6 +161,11 @@ def shared_credentials_env(self) -> dict[str, str]: it overrides this, and the workloads get whatever scope those credentials carry.""" return {} + def grants_reach(self, sandbox_type: str | None) -> bool: + """Whether an agent in a sandbox of ``sandbox_type`` (None: unknown) can use this store's + grants. True by default, for grants that are public HTTPS URLs.""" + return True + def issue_read_grant( self, object_url: str, *, expires_in: int = 3600 ) -> HttpGetGrant: diff --git a/src/agent_env/store/routing.py b/src/agent_env/store/routing.py index d764b672..4fe2bd20 100644 --- a/src/agent_env/store/routing.py +++ b/src/agent_env/store/routing.py @@ -493,6 +493,10 @@ def __init__(self, configured: ObjectStore, local: ObjectStore) -> None: def supports_transfer_grants(self) -> bool: return self.local.supports_transfer_grants + def grants_reach(self, sandbox_type: str | None) -> bool: + # A run writes to the local store, so its grants must reach the agent too. + return self.local.grants_reach(sandbox_type) + @property def max_single_upload_bytes(self) -> int | None: return self.local.max_single_upload_bytes diff --git a/src/agent_env/task_step/snapshot_utils/agent_state_capture.py b/src/agent_env/task_step/snapshot_utils/agent_state_capture.py index 8e13877a..ff4ce216 100644 --- a/src/agent_env/task_step/snapshot_utils/agent_state_capture.py +++ b/src/agent_env/task_step/snapshot_utils/agent_state_capture.py @@ -62,6 +62,7 @@ async def capture_workspace( a2a_context_id: str, artifact_id: str, timeout_seconds: float, + sandbox_type: str | None, ) -> WorkspaceCapture: """Capture the agent's workspace below a fresh prefix and wrap it as a universe. @@ -91,6 +92,7 @@ async def capture_workspace( agent_name=agent_name, context_id=a2a_context_id, capture_prefix=capture_prefix, + sandbox_type=sandbox_type, ) await invoke_transfer( a2a_url + save_path, @@ -144,6 +146,7 @@ async def read_partial_trajectory( context_id: str, timeout_seconds: float, trajectory_output_prefix: str, + sandbox_type: str | None, ) -> TrajectoryCapture: """Read the trajectory-so-far for an in-progress run. @@ -172,7 +175,7 @@ async def read_partial_trajectory( if "request" not in get_method: return TrajectoryCapture(reason="trajectory_context_unsupported") store = get_config().get_object_store() - mode = trajectory_mode(get_method, store, by="context_id") + mode = trajectory_mode(get_method, store, by="context_id", sandbox_type=sandbox_type) if mode is None: return TrajectoryCapture(reason="trajectory_context_unsupported") diff --git a/src/agent_env/task_step/snapshot_utils/snapshot_series.py b/src/agent_env/task_step/snapshot_utils/snapshot_series.py index 3913a732..fecaee22 100644 --- a/src/agent_env/task_step/snapshot_utils/snapshot_series.py +++ b/src/agent_env/task_step/snapshot_utils/snapshot_series.py @@ -427,6 +427,7 @@ def remaining() -> float: a2a_context_id=self.a2a_context_id, artifact_id=self.workspace_artifact_id, timeout_seconds=remaining(), + sandbox_type=agent.sandbox_type, ) row["id"] = workspace.universe_id row["version"] = workspace.universe_version @@ -443,6 +444,7 @@ def remaining() -> float: context_id=self.a2a_context_id, timeout_seconds=remaining(), trajectory_output_prefix=self.trajectory_output_prefix, + sandbox_type=agent.sandbox_type, ) if traj.reason and is_final: recorded = self._recorded_trajectory_uri(context) diff --git a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py index 78283265..7b9e4dc6 100644 --- a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py +++ b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_skill_config.py @@ -97,6 +97,7 @@ async def probe_bundle(name: str, description: str, object_url: str | None) -> b name=name, description=description, object_url=object_url, + sandbox_type=deployed_agent.sandbox_type, ) except RuntimeError as exc: logger.info("Skill bundle request: not sent (%s)", exc) diff --git a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py index cb9efe1e..ca1d7ee8 100644 --- a/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py +++ b/src/agent_env/task_step/task_steps/a2a_agent_validator/verify_a2a_trajectory.py @@ -89,7 +89,8 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: config = get_config() store = config.get_object_store() probe_prefix = f"{config.get_artifact_key_prefix()}a2a_validator_trajectories/{self.a2a_agent_id}/" - if trajectory_mode(get_method, store, by="task_id") == "objects": + sandbox_type = deployed_agent.sandbox_type + if trajectory_mode(get_method, store, by="task_id", sandbox_type=sandbox_type) == "objects": try: prefix = store.object_url(probe_prefix) upload = await asyncio.to_thread( diff --git a/src/agent_env/task_step/task_steps/deploy_agent.py b/src/agent_env/task_step/task_steps/deploy_agent.py index 40824353..21d7d11e 100644 --- a/src/agent_env/task_step/task_steps/deploy_agent.py +++ b/src/agent_env/task_step/task_steps/deploy_agent.py @@ -389,14 +389,14 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: logger.info(f"Set context.default_agent_model to '{default_model}'") if self.agent_snapshot_files_artifact_id: - await self._load_snapshot(a2a_url, card, a2a_agent_id, context) + await self._load_snapshot(a2a_url, card, a2a_agent_id, context, sandbox_type=deployed.sandbox_type) if self.agent_changelog_object_url: - await self._apply_agent_changelog(a2a_url, card, context) + await self._apply_agent_changelog(a2a_url, card, context, sandbox_type=deployed.sandbox_type) if self.enable_agent_changelog: await self._configure_agent_changelog( - a2a_url, card, context, expires_in=resolved_ttl + a2a_url, card, context, expires_in=resolved_ttl, sandbox_type=deployed.sandbox_type ) context.deployed_agents.append(DeployedAgent( @@ -452,6 +452,7 @@ async def _configure_agent_changelog( context: TaskStepContext, *, expires_in: int, + sandbox_type: str | None, ) -> None: """Enable whole-writable-layer changelog capture via the snapshot enable-changelog method; records the derived prefix in context.metadata['agent_changelog'].""" from agent_env.a2a_agent import A2AAgent @@ -475,6 +476,7 @@ async def _configure_agent_changelog( agent_name=self.agent_name, namespace_url=namespace_url, expires_in=expires_in, + sandbox_type=sandbox_type, ) answer = await invoke_transfer( a2a_url + path, @@ -492,7 +494,9 @@ async def _configure_agent_changelog( }) logger.info(f"agent-changelog capture enabled on '{self.agent_name}': {namespace_url}") - async def _apply_agent_changelog(self, a2a_url: str, card: dict, context: TaskStepContext) -> None: + async def _apply_agent_changelog( + self, a2a_url: str, card: dict, context: TaskStepContext, *, sandbox_type: str | None + ) -> None: """Rewind this fresh agent to a point in a source changelog (fs + conversation) via the snapshot apply-changelog method and resume.""" from agent_env.a2a_agent import A2AAgent @@ -514,6 +518,7 @@ async def _apply_agent_changelog(self, a2a_url: str, card: dict, context: TaskSt up_to_tool_call_exclusive=self.agent_changelog_toolcall_position_exclusive, resume_conversation=True, target_context_id=self.agent_snapshot_target_context_id, + sandbox_type=sandbox_type, ) answer = await invoke_transfer( a2a_url + path, @@ -543,6 +548,8 @@ async def _load_snapshot( card: dict, a2a_agent_id: str, context: TaskStepContext, + *, + sandbox_type: str | None, ) -> None: """Restore a snapshot into the freshly-deployed agent via /ext/snapshot.""" from agent_env.a2a_agent import A2AAgent @@ -572,6 +579,7 @@ async def _load_snapshot( bundle_url=universe.bundle_object_url, file_names=set((universe.file_artifact_refs or universe.file_artifact_ids or {}).keys()), target_context_id=self.agent_snapshot_target_context_id, + sandbox_type=sandbox_type, ) answer = await invoke_transfer( a2a_url + load_path, diff --git a/src/agent_env/task_step/task_steps/prompt_agent.py b/src/agent_env/task_step/task_steps/prompt_agent.py index c63f9e15..f99e82c2 100644 --- a/src/agent_env/task_step/task_steps/prompt_agent.py +++ b/src/agent_env/task_step/task_steps/prompt_agent.py @@ -512,7 +512,7 @@ async def _execute_conversation( try: turn_traj_uri = await self._fetch_trajectory( target_url, traj_ext_cached, sent_task_id, trajectory_output_prefix, - target_a2a_task_id, + target_a2a_task_id, sandbox_type=agent.sandbox_type, ) except Exception as e: logger.warning(f"Turn {turn+1} trajectory fetch failed (continuing): {e}") @@ -700,10 +700,13 @@ async def _execute_conversation( return context - async def _fetch_trajectory(self, a2a_url: str, traj_ext: dict, a2a_server_task_id: str, trajectory_output_prefix: str, target_a2a_task_id: str) -> str | None: + async def _fetch_trajectory( + self, a2a_url: str, traj_ext: dict, a2a_server_task_id: str, trajectory_output_prefix: str, + target_a2a_task_id: str, *, sandbox_type: str | None, + ) -> str | None: get_method, get_path = A2AAgent.operation(traj_ext, "get") store = get_config().get_object_store() - mode = trajectory_mode(get_method, store, by="task_id") + mode = trajectory_mode(get_method, store, by="task_id", sandbox_type=sandbox_type) if mode is None: raise RuntimeError( "Agent advertises no trajectory get form this object store can serve" diff --git a/src/agent_env/task_step/task_steps/snapshot_agent_state.py b/src/agent_env/task_step/task_steps/snapshot_agent_state.py index 67745f34..7bb50f60 100644 --- a/src/agent_env/task_step/task_steps/snapshot_agent_state.py +++ b/src/agent_env/task_step/task_steps/snapshot_agent_state.py @@ -117,6 +117,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: a2a_context_id=a2a_context_id, artifact_id=self.artifact_id, timeout_seconds=self.timeout_seconds, + sandbox_type=deployed.sandbox_type, ) snapshots = context.metadata.setdefault("agent_snapshots", []) diff --git a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py index facd8671..07c1e1ae 100644 --- a/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/rubrics_verifier.py @@ -1054,6 +1054,7 @@ async def _invoke_judge_a2a( judge_a2a_url=judge_a2a_url, judge_agent_card=judge_agent_card, a2a_server_task_id=task_id, + sandbox_type=getattr(judge_agent, "sandbox_type", None), ) return {"response": tr.response_text, "trajectory_s3_uri": judge_trajectory_s3_uri} @@ -1063,6 +1064,7 @@ async def _fetch_judge_trajectory( judge_a2a_url: str, judge_agent_card: dict, a2a_server_task_id: str, + sandbox_type: Optional[str], ) -> Optional[str]: """Best-effort capture of the judge's own reasoning trajectory, via the same EXT_TRAJECTORY extension `prompt_agent` uses, keyed by the A2A task id. Retries @@ -1085,7 +1087,7 @@ async def _fetch_judge_trajectory( config = get_config() store = config.get_object_store() prefix = store.object_url(f"{config.get_artifact_key_prefix()}judge_trajectories/verifier_id={self.verifier_id}/") - mode = trajectory_mode(get_method, store, by="task_id") + mode = trajectory_mode(get_method, store, by="task_id", sandbox_type=sandbox_type) if mode is None: logger.warning( "Verifier '%s': judge advertises no trajectory get form this object store " diff --git a/tst/integration/env/task_step/task_steps_test.py b/tst/integration/env/task_step/task_steps_test.py index ec31ba70..340a9208 100644 --- a/tst/integration/env/task_step/task_steps_test.py +++ b/tst/integration/env/task_step/task_steps_test.py @@ -25,6 +25,7 @@ from agent_env.task_step import AddSkillsTaskStep, BuildMcpCliTaskStep, DeployAgentTaskStep, DeployEnvTaskStep, EnvOutcomeVerifierTaskStep, LoadArtifactTaskStep, PromptAgentTaskStep, RubricsVerifierTaskStep, TaskStep, TaskStepContext, VerifyMCPToolSchemaTaskStep from agent_env.task_step.task_step import TaskStepDependency from agent_env.a2a_agent import A2AAgent, conversation_store +from agent_env.a2a_agent.object_transfer import trajectory_mode from agent_env.providers.sandbox_providers.sandbox import VmSandbox from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_agent_sandbox_provider @@ -755,6 +756,42 @@ def put_file(name): await env.close() +@pytest.mark.asyncio +@pytest.mark.integration +async def test_an_agent_uploads_its_trajectory_through_a_local_grant(mcp_server_env, echo_agent): + """On the local defaults an agent that takes the object form uploads its own trajectory: its container + trusts the local transfer CA and reaches the grant server on this host.""" + suffix = uuid.uuid4().hex[:8] + store = get_config().get_object_store() + context = await DeployEnvTaskStep( + id=f"grant-traj-env-{suffix}", version=None, env_id=mcp_server_env.id, env_version=mcp_server_env.version, + ).execute(TaskStepContext()) + deployed_env = context.deployed_envs[0] + try: + await DeployAgentTaskStep( + id=f"grant-traj-agent-{suffix}", version=None, env_ids=[mcp_server_env.id], + a2a_agent_id=echo_agent.id, a2a_agent_version=echo_agent.version, + ).execute(context) + agent = context.deployed_agents[0] + get_method, _ = A2AAgent.operation(A2AAgent.find_extension(agent.a2a_card, A2AAgent.EXT_TRAJECTORY), "get") + assert trajectory_mode(get_method, store, by="task_id", sandbox_type=agent.sandbox_type) == "objects" + + await PromptAgentTaskStep( + id=f"grant-traj-prompt-{suffix}", version=None, prompt="hello through a grant", timeout_seconds=120, + ).execute(context) + uri = context.prompt_responses[-1].agent_trajectory_s3_uri + assert uri, "the agent did not upload its trajectory through the grant" + assert "hello through a grant" in store.get(uri).decode() + finally: + if context.deployed_agents: + try: + await (await _agent_sandbox(context.deployed_agents[0])).terminate() + except Exception as e: + logger.warning(f"grant trajectory test agent cleanup: {e}") + env = await type(Env.get(deployed_env.env_id, deployed_env.env_version)).from_deployed_env(deployed_env) + await env.close() + + @pytest.mark.asyncio @pytest.mark.integration @skip_without_model_endpoint() diff --git a/tst/unit/a2a_agent/add_skill_test.py b/tst/unit/a2a_agent/add_skill_test.py index dba9977c..d5636253 100644 --- a/tst/unit/a2a_agent/add_skill_test.py +++ b/tst/unit/a2a_agent/add_skill_test.py @@ -53,7 +53,11 @@ def _card( def _deployed( - *, bundle: bool, legacy: bool | None = None, config_key: str = "params" + *, + bundle: bool, + legacy: bool | None = None, + config_key: str = "params", + sandbox_type: str = "local", ) -> DeployedA2AAgent: return DeployedA2AAgent( agent_id="agent", @@ -61,6 +65,7 @@ def _deployed( a2a_url="https://agent.example.test", sandbox_id="sandbox", agent_card=_card(bundle=bundle, legacy=legacy, config_key=config_key), + sandbox_type=sandbox_type, ) @@ -195,6 +200,20 @@ async def test_an_object_skill_on_a_store_without_grants_is_refused_before_any_r assert agent.requests == [] +@pytest.mark.asyncio +async def test_an_object_skill_is_refused_when_the_stores_grants_do_not_reach_the_agent( + store: GrantingObjectStore, prefix: str, agent: _Agent +) -> None: + with pytest.raises(RuntimeError, match="grants do not reach agents on the 'modal' sandbox provider"): + await A2AAgent.add_skill( + _deployed(bundle=True, sandbox_type="modal"), + Skill(name="review", description="Review work", s3_url=prefix), + ) + + assert agent.requests == [] + assert store.granted == [] + + @pytest.mark.asyncio async def test_inline_skill_stays_inline_when_bundle_is_advertised( store: GrantingObjectStore, agent: _Agent diff --git a/tst/unit/a2a_agent/object_transfer_reach_test.py b/tst/unit/a2a_agent/object_transfer_reach_test.py new file mode 100644 index 00000000..635aa184 --- /dev/null +++ b/tst/unit/a2a_agent/object_transfer_reach_test.py @@ -0,0 +1,94 @@ +"""Whether an agent is offered grants depends on whether they reach its sandbox, and an object call +whose grants cannot reach the agent is refused, saying why.""" + +import pytest + +from agent_env.a2a_agent.object_transfer import ( + choose_transfer, + skill_add_call, + snapshot_save_call, + trajectory_mode, +) +from agent_env.store import LocalFilesystemObjectStore +from tst.unit.store.fakes import FakeObjectStore + +SDK_TASK_GET = {"request": {"required": ["task_id"], "oneOf": [{}, {"required": ["objects"]}]}} +OBJECTS_ONLY_GET = {"request": {"required": ["task_id", "objects"]}} +SAVE_EITHER = {"request": {"required": ["context_id"], "oneOf": [{"required": ["objects"]}, {"required": ["s3_prefix"]}]}} +SAVE_S3_ONLY = {"request": {"required": ["context_id", "s3_prefix"]}} +SAVE_OBJECTS_ONLY = {"request": {"required": ["context_id", "objects"]}} +SKILL_ANY = { + "request": { + "required": ["name", "description"], + "oneOf": [{"required": ["skill_bundle"]}, {"required": ["skill_s3_url"]}, {"required": ["skill_md"]}], + } +} + + +@pytest.fixture +def local(tmp_path): + return LocalFilesystemObjectStore(str(tmp_path)) + + +@pytest.mark.parametrize("sandbox_type, expected", [("local", "objects"), ("modal", "legacy"), (None, "legacy")]) +def test_a_local_store_offers_grants_only_to_local_agents(local, sandbox_type, expected): + mode = choose_transfer( + SDK_TASK_GET, objects=("task_id", "objects"), legacy=("task_id",), store=local, sandbox_type=sandbox_type + ) + assert mode == expected + + +def test_a_remote_agent_gets_its_trajectory_inline_from_a_local_run(local): + assert trajectory_mode(SDK_TASK_GET, local, by="task_id", sandbox_type="modal") == "legacy" + assert trajectory_mode(OBJECTS_ONLY_GET, local, by="task_id", sandbox_type="modal") is None + + +def test_a_hosted_store_offers_grants_whatever_the_sandbox(): + store = FakeObjectStore() + store.supports_transfer_grants = True + mode = choose_transfer( + SDK_TASK_GET, objects=("task_id", "objects"), legacy=("task_id",), store=store, sandbox_type="modal" + ) + assert mode == "objects" + + +def _save(store, method, sandbox_type): + return snapshot_save_call( + method, store, agent_name="solver", context_id="ctx", capture_prefix=store.object_url("snap/"), + sandbox_type=sandbox_type, + ) + + +def test_an_agent_that_takes_only_the_s3_form_is_refused(local): + with pytest.raises(RuntimeError, match="snapshot save on agent 'solver': the agent does not advertise the object form"): + _save(local, SAVE_S3_ONLY, "local") + + +def test_a_remote_agent_is_refused_and_told_why(local): + with pytest.raises(RuntimeError, match="grants do not reach agents on the 'modal' sandbox provider"): + _save(local, SAVE_EITHER, "modal") + + +def test_with_grants_off_a_local_agent_is_refused(tmp_path): + store = LocalFilesystemObjectStore(str(tmp_path), grants="off") + with pytest.raises(RuntimeError, match="the object store does not issue transfer grants"): + _save(store, SAVE_EITHER, "local") + + +def test_an_objects_only_agent_out_of_reach_is_told_why(local): + with pytest.raises(RuntimeError, match="snapshot save on agent 'solver': the object store's grants do not reach agents on the 'unknown' sandbox provider"): + _save(local, SAVE_OBJECTS_ONLY, None) + + +def test_a_skill_given_inline_still_reaches_a_remote_agent(local): + call = skill_add_call(SKILL_ANY, local, name="s", description="d", skill_md="# s", sandbox_type="modal") + assert (call.mode, call.payload["skill_md"]) == ("legacy", "# s") + + +def test_a_skill_from_local_objects_is_refused_to_a_remote_agent(local): + local.put("skills/s/SKILL.md", b"# s") + with pytest.raises(RuntimeError, match="skill add: the object store's grants do not reach agents on the 'modal' sandbox provider"): + skill_add_call( + SKILL_ANY, local, name="s", description="d", object_url=local.object_url("skills/s"), + sandbox_type="modal", + ) diff --git a/tst/unit/a2a_agent/object_transfer_test.py b/tst/unit/a2a_agent/object_transfer_test.py index 9f686707..1cdac0ce 100644 --- a/tst/unit/a2a_agent/object_transfer_test.py +++ b/tst/unit/a2a_agent/object_transfer_test.py @@ -73,7 +73,7 @@ def test_choose_transfer(tmp_path, method, fields, grants, expected): store = GrantingObjectStore(str(tmp_path)) store.supports_transfer_grants = grants - assert choose_transfer(method, store=store, **fields) == expected + assert choose_transfer(method, store=store, sandbox_type="local", **fields) == expected def test_parse_response_ignores_fields_it_does_not_know(): diff --git a/tst/unit/a2a_agent/test_validator_changelog_objects.py b/tst/unit/a2a_agent/test_validator_changelog_objects.py index a225f57a..bb715ea1 100644 --- a/tst/unit/a2a_agent/test_validator_changelog_objects.py +++ b/tst/unit/a2a_agent/test_validator_changelog_objects.py @@ -49,7 +49,7 @@ async def get_sandbox(self, sandbox_id): async def close(self): return None - monkeypatch.setattr(provider_mod, "get_agent_sandbox_provider", _Provider) + monkeypatch.setattr(provider_mod, "build_sandbox_provider", lambda spec: _Provider()) persisted = SimpleNamespace(metadata={}) persisted.update_metadata = lambda metadata: setattr(persisted, "metadata", metadata) monkeypatch.setattr(A2AAgent, "get", classmethod(lambda cls, agent_id: persisted)) @@ -77,7 +77,7 @@ def _context(capture: dict, *apply_variants: list[str]) -> TaskStepContext: ] } }, - sandbox_type=None, + sandbox_type="local", sandbox_id="sandbox-1", ) context = TaskStepContext() diff --git a/tst/unit/a2a_agent/test_validator_skill_variants.py b/tst/unit/a2a_agent/test_validator_skill_variants.py index 7239e958..a61c3597 100644 --- a/tst/unit/a2a_agent/test_validator_skill_variants.py +++ b/tst/unit/a2a_agent/test_validator_skill_variants.py @@ -66,6 +66,7 @@ def _context(*, bundle: bool) -> TaskStepContext: agent_name="agent", api_url="https://agent", a2a_url="https://agent", + sandbox_type="local", a2a_card={ "capabilities": { "extensions": [ diff --git a/tst/unit/a2a_agent/validator_probe_grant_test.py b/tst/unit/a2a_agent/validator_probe_grant_test.py new file mode 100644 index 00000000..6b1275a5 --- /dev/null +++ b/tst/unit/a2a_agent/validator_probe_grant_test.py @@ -0,0 +1,52 @@ +"""The validator's image-by-URI probe uses a transfer grant where the store signs no URL, but only one that +reaches the provider the validation agents will deploy on.""" + +from types import SimpleNamespace + +import pytest + +from agent_env.a2a_agent.validator import A2AAgentValidator +from agent_env.config import reset_config, set_object_store +from agent_env.providers.sandbox_providers.chained_sandbox_provider import ChainedSandboxProvider +from agent_env.providers.sandbox_providers.local_sandbox import LocalSandboxProvider +from agent_env.providers.sandbox_providers.sandbox_provider import ( + SandboxProvider, + reset_agent_sandbox_provider, + set_agent_sandbox_provider, +) +from tst.util.granting_object_store import GRANT_ORIGIN, GrantingObjectStore + +_AGENT = SimpleNamespace(id="solver", version=3) + + +class _RemoteProvider(SandboxProvider): + async def create_sandbox(self, **kwargs): + raise NotImplementedError + + +@pytest.fixture +def store(tmp_path): + granting = GrantingObjectStore(str(tmp_path)) + set_object_store(granting) + yield granting + reset_config() + reset_agent_sandbox_provider() + + +@pytest.mark.parametrize("provider", [ + LocalSandboxProvider(), ChainedSandboxProvider([LocalSandboxProvider(), _RemoteProvider()]), +], ids=["local", "local-first-in-a-chain"]) +def test_agents_on_the_local_provider_get_the_image_through_a_grant(store, provider): + set_agent_sandbox_provider(provider) + fixtures = A2AAgentValidator._upload_probe_fixtures(_AGENT, "file:///skill") + assert fixtures.png_signed_url.startswith(GRANT_ORIGIN) + + +@pytest.mark.parametrize("provider", [ + _RemoteProvider(), ChainedSandboxProvider([_RemoteProvider(), LocalSandboxProvider()]), +], ids=["remote", "remote-first-in-a-chain"]) +def test_agents_the_grants_do_not_reach_are_not_sent_one(store, provider): + set_agent_sandbox_provider(provider) + with pytest.raises(RuntimeError, match="signs URLs or issues grants"): + A2AAgentValidator._upload_probe_fixtures(_AGENT, "file:///skill") + assert store.granted == [] diff --git a/tst/unit/a2a_agent/validator_trajectory_dag_test.py b/tst/unit/a2a_agent/validator_trajectory_dag_test.py index 093469be..324bc71c 100644 --- a/tst/unit/a2a_agent/validator_trajectory_dag_test.py +++ b/tst/unit/a2a_agent/validator_trajectory_dag_test.py @@ -176,6 +176,9 @@ def test_trajectory_step_records_failed_object_url_construction(monkeypatch): class _FailingObjectStore: supports_transfer_grants = True + def grants_reach(self, _sandbox_type: str | None) -> bool: + return True + def object_url(self, _key: str) -> str: raise RuntimeError("cannot mint object URL") @@ -218,6 +221,9 @@ def __init__(self) -> None: def object_url(self, key: str) -> str: return f"s3://test-bucket/{key}" + def grants_reach(self, sandbox_type: str | None) -> bool: + return True + def get_object_key(self, object_url: str) -> str: return object_url.removeprefix("s3://test-bucket/") diff --git a/tst/unit/providers/sandbox_providers/local_sandbox_test.py b/tst/unit/providers/sandbox_providers/local_sandbox_test.py index 3e344ebe..534827cc 100644 --- a/tst/unit/providers/sandbox_providers/local_sandbox_test.py +++ b/tst/unit/providers/sandbox_providers/local_sandbox_test.py @@ -13,6 +13,47 @@ from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox, LocalSandboxProvider from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER, SANDBOX_MODE_VM from agent_env.store import LocalFilesystemObjectStore +from agent_env.store.object_store.local.tls import local_ca +from agent_env.store.routing import LocalRunObjectStore +from agent_env.a2a_agent import a2a_agent as a2a_agent_module +from agent_env.a2a_agent.a2a_agent import A2AAgent +from tst.unit.store.fakes import FakeObjectStore + +_real_copy_into_container = ls._copy_into_container # before the fixtures below replace them +_real_local_grant_trust = ls.local_grant_trust + + +@pytest.fixture(autouse=True) +def copies(monkeypatch): + """What create_container would ``docker cp`` into its container, recorded instead of run; the configured + store is taken to hand out local grants.""" + recorded: list[tuple] = [] + monkeypatch.setattr(ls, "_copy_into_container", lambda *args: recorded.append(args)) + monkeypatch.setattr(ls, "local_grant_trust", lambda: local_ca().trust_dir) + return recorded + + +class _StubImageStore: + def auth(self, image_name): + return None + + +class _StubConfig: + def get_image_store(self): + return _StubImageStore() + + +class _ScriptedProvider(LocalSandboxProvider): + """Creates a recording sandbox under ``work_dir`` in place of a VM.""" + + def __init__(self, work_dir: Path) -> None: + self.work_dir = work_dir + self.made: list[_RecordingLocalSandbox] = [] + + async def create_vm(self, **kwargs): + sandbox = _RecordingLocalSandbox(work_dir=self.work_dir) + self.made.append(sandbox) + return sandbox @pytest.fixture(autouse=True) @@ -411,7 +452,7 @@ async def _rec(script, *, max_retries=0): return sb await _RecordingProvider().create_sandbox(image_name="img:v1", port=8000, env={"K": "v"}) - run_cmd = next(s for s in recorded if "docker run" in s) + run_cmd = next(s for s in recorded if "docker create" in s) assert "--add-host host.docker.internal:host-gateway" in run_cmd @@ -476,7 +517,7 @@ async def _rec(script, *, max_retries=0): await _RecordingProvider().create_sandbox(image_name="img:v1", port=8000, env={}) assert made[0].host_port(8000) == 41337 - run_cmd = next(s for s in recorded if "docker run" in s) + run_cmd = next(s for s in recorded if "docker create" in s) assert "-p 127.0.0.1:41337:8000 -p 172.17.0.1:41337:8000 " in run_cmd @@ -488,8 +529,8 @@ async def test_an_agent_placed_on_a_local_vm_publishes_only_on_the_host_ips(monk await agent._run_container("img:1", 8000, {}) - [script] = agent._sandbox.scripts - assert "-p 127.0.0.1:8000:8000 -p 172.17.0.1:8000:8000 " in script + create = agent._sandbox.scripts[0] # then started once it trusts the local CA + assert "-p 127.0.0.1:8000:8000 -p 172.17.0.1:8000:8000 " in create @pytest.fixture @@ -708,4 +749,144 @@ async def slow_spawn(*args, **kwargs): os.kill(int(pid_file.read_text()), 0) +@pytest.mark.asyncio +async def test_a_container_is_created_given_the_local_ca_then_started(tmp_path, monkeypatch, copies): + """The CA's trust files are in place before the container's first process runs, and its TLS clients + are pointed at them, so it can use the local object store's grants.""" + import agent_env.config as cfg + + monkeypatch.setattr(cfg, "get_config", lambda: _StubConfig()) + provider = _ScriptedProvider(tmp_path) + sandbox = await provider.create_container(image_name="img:v1", port=8000, env={"K": "v"}) + + create, start = [s for s in provider.made[0].scripts if s.startswith(("docker create", "docker start"))] + assert "docker run" not in "".join(provider.made[0].scripts) + for flag in ( + "-e SSL_CERT_FILE=/etc/agentenv/ca-bundle.pem", + "-e REQUESTS_CA_BUNDLE=/etc/agentenv/ca-bundle.pem", + "-e NODE_EXTRA_CA_CERTS=/etc/agentenv/ca.pem", + "-e K=v", + ): + assert flag in create + assert copies == [(local_ca().trust_dir, sandbox.container_name, "/etc/agentenv")] + assert start == f"docker start {sandbox.container_name} > /dev/null" + assert sandbox.mode == SANDBOX_MODE_CONTAINER + + +@pytest.mark.asyncio +async def test_a_trust_variable_the_caller_sets_wins(tmp_path, monkeypatch): + import agent_env.config as cfg + + monkeypatch.setattr(cfg, "get_config", lambda: _StubConfig()) + provider = _ScriptedProvider(tmp_path) + await provider.create_container(image_name="img:v1", port=8000, env={"SSL_CERT_FILE": "/own/roots.pem"}) + + create = next(s for s in provider.made[0].scripts if s.startswith("docker create")) + assert "-e SSL_CERT_FILE=/own/roots.pem" in create + assert "SSL_CERT_FILE=/etc/agentenv" not in create + + +@pytest.mark.asyncio +async def test_a_failed_copy_removes_the_created_container(tmp_path, monkeypatch): + import agent_env.config as cfg + + monkeypatch.setattr(cfg, "get_config", lambda: _StubConfig()) + + def fail(*args): + raise RuntimeError("no such container") + + monkeypatch.setattr(ls, "_copy_into_container", fail) + provider = _ScriptedProvider(tmp_path) + with pytest.raises(RuntimeError, match="no such container"): + await provider.create_container(image_name="img:v1", port=8000, env={}) + scripts = provider.made[0].scripts + assert not any(s.startswith("docker start") for s in scripts) + assert scripts[-1].startswith(f"docker rm -f {provider.made[0].container_name}") + + +def test_the_copy_runs_docker_directly_and_reports_its_error(monkeypatch, tmp_path): + calls = [] + + def run(cmd, **kwargs): + calls.append(cmd) + return SimpleNamespace(returncode=1, stderr="Error: No such container: agent-x\n") + + monkeypatch.setattr(ls.subprocess, "run", run) + with pytest.raises(RuntimeError, match="No such container"): + _real_copy_into_container(tmp_path, "agent-x", "/etc/agentenv") + assert calls == [["docker", "cp", f"{tmp_path}/.", "agent-x:/etc/agentenv"]] + + +@pytest.mark.asyncio +async def test_without_local_grants_a_container_is_run_as_before(tmp_path, monkeypatch, copies): + import agent_env.config as cfg + + monkeypatch.setattr(cfg, "get_config", lambda: _StubConfig()) + monkeypatch.setattr(ls, "local_grant_trust", lambda: None) + provider = _ScriptedProvider(tmp_path) + await provider.create_container(image_name="img:v1", port=8000, env={"K": "v"}) + + run = next(s for s in provider.made[0].scripts if s.startswith("docker run -d")) + assert "SSL_CERT_FILE" not in run + assert copies == [] + + +def test_only_a_local_store_that_grants_needs_the_local_ca(tmp_path): + local = LocalFilesystemObjectStore(str(tmp_path / "objects")) + try: + set_object_store(local) + assert _real_local_grant_trust() == local_ca().trust_dir + set_object_store(LocalRunObjectStore(FakeObjectStore(), local)) + assert _real_local_grant_trust() == local_ca().trust_dir + set_object_store(LocalFilesystemObjectStore(str(tmp_path / "objects"), grants="off")) + assert _real_local_grant_trust() is None + set_object_store(FakeObjectStore()) + assert _real_local_grant_trust() is None + finally: + reset_config() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("trusted", [True, False]) +async def test_an_agent_placed_on_a_local_vm_sandbox_gets_the_local_ca(tmp_path, monkeypatch, copies, trusted): + """A linked agent starts through A2AAgent._run_container, not the provider: it is given the CA the same way.""" + monkeypatch.setattr(a2a_agent_module, "local_grant_trust", (lambda: local_ca().trust_dir) if trusted else (lambda: None)) + monkeypatch.setattr(a2a_agent_module.asyncio, "sleep", _no_sleep) + sandbox = _RecordingLocalSandbox(work_dir=tmp_path) + agent = A2AAgent.__new__(A2AAgent) + agent._sandbox = sandbox + + await agent._run_container("img:v1", 8000, {"K": "v"}) + + script = sandbox.scripts[0] + assert ("docker create" in script) is trusted and ("docker run -d" in script) is not trusted + assert ("-e SSL_CERT_FILE='/etc/agentenv/ca-bundle.pem'" in script) is trusted + if trusted: + assert copies == [(local_ca().trust_dir, sandbox.container_name, "/etc/agentenv")] + assert sandbox.scripts[-1] == f"docker start {sandbox.container_name} > /dev/null" + assert (LocalSandboxProvider.EXTRA_CONTAINER_RUN_ARGS in script) if LocalSandboxProvider.EXTRA_CONTAINER_RUN_ARGS else True + else: + assert copies == [] and "sleep 2" in script + + +async def _no_sleep(_seconds): + return None + + +@pytest.mark.asyncio +async def test_a_linked_agent_whose_copy_fails_leaves_no_container(tmp_path, monkeypatch): + def fail(*args): + raise RuntimeError("no space left") + + monkeypatch.setattr(ls, "_copy_into_container", fail) + monkeypatch.setattr(a2a_agent_module, "local_grant_trust", lambda: local_ca().trust_dir) + sandbox = _RecordingLocalSandbox(work_dir=tmp_path) + agent = A2AAgent.__new__(A2AAgent) + agent._sandbox = sandbox + + with pytest.raises(RuntimeError, match="no space left"): + await agent._run_container("img:v1", 8000, {}) + assert sandbox.scripts[-1] == f"docker rm -f {sandbox.container_name} >/dev/null 2>&1 || true" + + _HOST_IPS = ls._host_ips diff --git a/tst/unit/store/gcs_object_store_test.py b/tst/unit/store/gcs_object_store_test.py index a1f1a271..dd0eeb4a 100644 --- a/tst/unit/store/gcs_object_store_test.py +++ b/tst/unit/store/gcs_object_store_test.py @@ -520,7 +520,10 @@ def test_signed_post_is_shaped_as_the_s3_store_s_and_clamped_to_the_signer(store def test_changelog_capture_gets_the_object_form(store): only_objects = {"request": {"oneOf": [{"required": ["write_namespace"]}]}} - call = changelog_enable_call(only_objects, store, agent_name="solver", namespace_url=f"gs://{BUCKET}/ns", expires_in=600) + call = changelog_enable_call( + only_objects, store, agent_name="solver", namespace_url=f"gs://{BUCKET}/ns", expires_in=600, + sandbox_type="modal", + ) assert call.mode == "objects" grant = call.payload["write_namespace"] assert grant["root_path"] == "ns" and grant["write"]["kind"] == "http-post-policy" diff --git a/tst/unit/store/local_grant_server_test.py b/tst/unit/store/local_grant_server_test.py index a804b803..e16b437e 100644 --- a/tst/unit/store/local_grant_server_test.py +++ b/tst/unit/store/local_grant_server_test.py @@ -11,10 +11,12 @@ import pytest from pytest_socket import enable_socket +from agent_env.a2a_agent import object_transfer +from agent_env.a2a_agent.object_transfer import TransferCall, invoke_transfer from agent_env.config.paths import state_root from agent_env.store import LocalFilesystemObjectStore from agent_env.store.object_store.local import grant_server as server_module -from agent_env.store.object_store.local.grant_server import default_bind_host, grant_server +from agent_env.store.object_store.local.grant_server import default_bind_host, grant_server, unreachable_hint from agent_env.store.object_store.local.tls import local_ca from tst.store import object_conformance @@ -309,3 +311,28 @@ def fail(*a, **k): monkeypatch.setattr(server_module.subprocess, "run", fail) assert default_bind_host() == "127.0.0.1" assert "grant_bind_host" in caplog.text + + +def test_the_unreachable_hint_names_the_server_and_never_a_grant(store): + grant = store.issue_read_grant(store.put("k", b"v")) + hint = unreachable_hint({"objects": {"trajectory": {"read": grant.model_dump(mode="json")}}}) + assert grant.url.split("/v1/")[0] in hint + assert "listens on 127.0.0.1" in hint and "grant_bind_host" in hint + assert grant.url.rsplit("/", 1)[1] not in hint + assert unreachable_hint({"objects": {"read": {"url": "https://bucket.example/obj"}}}) is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("code, hinted", [("transfer_unavailable", True), ("transfer_rejected", False)]) +async def test_an_agent_that_could_not_reach_the_grant_server_is_told_where_to_look(store, monkeypatch, code, hinted): + grant = store.issue_read_grant(store.put("k", b"v")) + answer = httpx.Response(502, json={"error": {"code": code, "message": "m", "retryable": True}}) + client = httpx.AsyncClient + monkeypatch.setattr( + object_transfer.httpx, "AsyncClient", lambda **kw: client(transport=httpx.MockTransport(lambda _: answer)) + ) + call = TransferCall("objects", {"objects": {"read": grant.model_dump(mode="json")}}) + with pytest.raises(httpx.HTTPStatusError) as raised: + await invoke_transfer("http://agent.test/ext", call, verb="POST", operation="snapshot load", timeout=5) + assert str(raised.value).startswith(f"snapshot load failed with HTTP 502: {code}") + assert ("local grant server" in str(raised.value)) is hinted diff --git a/tst/unit/store/local_id_acceptance_test.py b/tst/unit/store/local_id_acceptance_test.py index 0dc34fa5..79298b35 100644 --- a/tst/unit/store/local_id_acceptance_test.py +++ b/tst/unit/store/local_id_acceptance_test.py @@ -21,6 +21,11 @@ from agent_env.env.envs.multi_env import MultiEnv from agent_env.env.envs.website import WebsiteEnv from agent_env.env.snapshot_store import EnvSnapshot +from agent_env.providers.sandbox_providers.sandbox_provider import ( + SandboxProvider, + reset_agent_sandbox_provider, + set_agent_sandbox_provider, +) from agent_env.store import Filter, LocalSqliteDocumentStore, VersionedEntityStore from agent_env.store.base import NotFoundError from agent_env.store.ids import fs_safe, key_segment @@ -164,14 +169,26 @@ def test_validating_an_local_entity_writes_its_task_under_it_to_the_local_store( assert not _documents().path.exists() or _documents().count("tasks", Filter()) == 0 -def test_validating_an_local_agent_stops_at_the_local_store_it_cannot_sign_with_before_writing_a_task( - local_stores, cli_routing, tmp_path, +class _RemoteProvider(SandboxProvider): + async def create_sandbox(self, **kwargs): + raise NotImplementedError + + +@pytest.fixture +def remote_agents(): + set_agent_sandbox_provider(_RemoteProvider()) + yield + reset_agent_sandbox_provider() + + +def test_validating_an_local_agent_stops_at_a_local_store_its_sandbox_cannot_reach_before_writing_a_task( + local_stores, cli_routing, remote_agents, tmp_path, ): configured = SigningObjectStore(str(tmp_path / "configured-objects")) set_object_store(configured) agent = A2AAgent(id="@local/~/bundle/agents/a", version=1, docker_image_artifact=MagicMock()) - with pytest.raises(RuntimeError, match="A2A validation requires a signable object store"): + with pytest.raises(RuntimeError, match="signs URLs or issues grants"): asyncio.run(A2AAgentValidator.validate(agent)) fixtures = f"a2a_validator/probe_fixtures/{key_segment(agent.id)}-v1" diff --git a/tst/unit/store/test_object_transfer_grants.py b/tst/unit/store/test_object_transfer_grants.py index dda7ea9d..17cb9cd3 100644 --- a/tst/unit/store/test_object_transfer_grants.py +++ b/tst/unit/store/test_object_transfer_grants.py @@ -21,6 +21,7 @@ ) from agent_env.store.object_store import UploadPolicy from agent_env.store.object_store.s3_object_store import S3ObjectStore +from agent_env.store.routing import LocalRunObjectStore BUCKET = "artifact-bucket" @@ -281,9 +282,29 @@ def test_botocore_still_exposes_the_signing_credentials() -> None: assert S3ObjectStore(session, BUCKET)._signing_credentials() == (None, True) -def test_local_store_does_not_offer_transfer_grants(tmp_path) -> None: - """It can issue them (local_grant_server_test.py), but agents on the local store don't use them.""" +def test_local_store_grants_reach_only_local_sandboxes(tmp_path) -> None: store = LocalFilesystemObjectStore(str(tmp_path)) - assert S3ObjectStore.supports_transfer_grants - assert not store.supports_transfer_grants + assert store.supports_transfer_grants + assert store.grants_reach("local") + assert not store.grants_reach("modal") + assert not store.grants_reach(None) + + +def test_local_store_grants_can_be_turned_off(tmp_path) -> None: + assert not LocalFilesystemObjectStore(str(tmp_path), grants="off").supports_transfer_grants + with pytest.raises(ValueError, match="grants must be one of"): + LocalFilesystemObjectStore(str(tmp_path), grants="on") + + +def test_hosted_store_grants_reach_any_sandbox(store: S3ObjectStore) -> None: + assert store.supports_transfer_grants + assert store.grants_reach("modal") and store.grants_reach("local") and store.grants_reach(None) + + +def test_a_local_run_offers_grants_that_reach_its_local_store(tmp_path, store: S3ObjectStore) -> None: + routed = LocalRunObjectStore(store, LocalFilesystemObjectStore(str(tmp_path))) + + assert routed.supports_transfer_grants + assert routed.grants_reach("local") + assert not routed.grants_reach("modal") diff --git a/tst/unit/task_step/capture_stubs.py b/tst/unit/task_step/capture_stubs.py index 3edfa2ca..554c24d2 100644 --- a/tst/unit/task_step/capture_stubs.py +++ b/tst/unit/task_step/capture_stubs.py @@ -77,6 +77,9 @@ def list_at(self, prefix: str) -> list[str]: def object_url(self, key: str) -> str: return f"s3://{BUCKET}/{key}" + def grants_reach(self, sandbox_type: str | None) -> bool: + return True + def issue_write_grant(self, object_url, *, media_type, max_bytes, expires_in): self.write_grants.append( { @@ -192,6 +195,7 @@ def __init__(self, name="solver", a2a_url="https://agent", card=None): self.api_url = a2a_url self.a2a_card = card if card is not None else agent_card() self.sandbox_id = "sb-agent" + self.sandbox_type = None def context(*, agent: Optional[_StubAgent] = None, env_id: Optional[str] = None) -> TaskStepContext: diff --git a/tst/unit/task_step/test_agent_state_capture.py b/tst/unit/task_step/test_agent_state_capture.py index d55844f2..88e717b7 100644 --- a/tst/unit/task_step/test_agent_state_capture.py +++ b/tst/unit/task_step/test_agent_state_capture.py @@ -47,6 +47,7 @@ async def _capture(**overrides): a2a_context_id="ctx-1", artifact_id="wsp", timeout_seconds=30, + sandbox_type="local", ) kwargs.update(overrides) return await mod.capture_workspace(**kwargs) @@ -210,7 +211,7 @@ async def test_a_save_on_a_store_without_grants_is_refused(monkeypatch): async def _read(card, **overrides): kwargs = dict( a2a_url="https://agent", a2a_card=card, context_id="ctx-1", timeout_seconds=30, - trajectory_output_prefix=PARTIAL_PREFIX, + trajectory_output_prefix=PARTIAL_PREFIX, sandbox_type="local", ) kwargs.update(overrides) return await mod.read_partial_trajectory(**kwargs) diff --git a/tst/unit/task_step/test_deploy_agent_changelog_objects.py b/tst/unit/task_step/test_deploy_agent_changelog_objects.py index e65115ed..8c1ce3f4 100644 --- a/tst/unit/task_step/test_deploy_agent_changelog_objects.py +++ b/tst/unit/task_step/test_deploy_agent_changelog_objects.py @@ -116,7 +116,7 @@ async def test_enable_prefers_a_bounded_namespace_grant(monkeypatch, store): context = _run_context() await _step(enable_agent_changelog=True)._configure_agent_changelog( - "https://agent", _card(), context, expires_in=28_800 + "https://agent", _card(), context, expires_in=28_800, sandbox_type="local" ) sent = requests[0] @@ -147,7 +147,7 @@ async def test_the_changelog_namespace_is_under_the_fixture_prefix(monkeypatch, context = _run_context() await _step(enable_agent_changelog=True)._configure_agent_changelog( - "https://agent", _card(), context, expires_in=600 + "https://agent", _card(), context, expires_in=600, sandbox_type="local" ) assert requests[0]["json"]["write_namespace"]["root_path"] == f"fx/{NAMESPACE_KEY}" @@ -174,6 +174,7 @@ def unavailable(*args, **kwargs): _card(enable_legacy=enable_legacy), _run_context(), expires_in=28_800, + sandbox_type="local", ) assert not requests @@ -187,7 +188,7 @@ async def test_enable_on_a_store_without_grants_is_refused(monkeypatch, store, e with pytest.raises(RuntimeError, match="does not issue transfer grants"): await _step(enable_agent_changelog=True)._configure_agent_changelog( - "https://agent", _card(enable_legacy=enable_legacy), _run_context(), expires_in=7_200 + "https://agent", _card(enable_legacy=enable_legacy), _run_context(), expires_in=7_200, sandbox_type="local" ) assert not requests @@ -200,7 +201,7 @@ async def test_enable_is_refused_for_an_agent_without_the_object_form(monkeypatc with pytest.raises(RuntimeError, match="does not advertise the object form"): await _step(enable_agent_changelog=True)._configure_agent_changelog( - "https://agent", _card(enable_objects=False), _run_context(), expires_in=7_200 + "https://agent", _card(enable_objects=False), _run_context(), expires_in=7_200, sandbox_type="local" ) assert not requests @@ -223,7 +224,7 @@ async def test_apply_lists_validates_and_sends_ordered_read_grants(monkeypatch, context = TaskStepContext() event_loop_thread = threading.get_ident() - await step._apply_agent_changelog("https://agent", _card(), context) + await step._apply_agent_changelog("https://agent", _card(), context, sandbox_type="local") sent = requests[0]["json"] assert [item["sequence"] for item in sent["increments"]] == [0, 1] @@ -246,7 +247,7 @@ async def test_apply_uses_absolute_tool_call_positions_for_the_cutoff(monkeypatc agent_changelog_toolcall_position_exclusive=7, ) - await step._apply_agent_changelog("https://agent", _card(), TaskStepContext()) + await step._apply_agent_changelog("https://agent", _card(), TaskStepContext(), sandbox_type="local") assert [item["sequence"] for item in requests[0]["json"]["increments"]] == [2, 4] assert _granted_names(store) == ["000002.tar", "000004.tar"] @@ -261,7 +262,7 @@ async def test_apply_allows_a_cutoff_beyond_the_last_tool_call(monkeypatch, stor agent_changelog_toolcall_position_exclusive=99, ) - await step._apply_agent_changelog("https://agent", _card(), TaskStepContext()) + await step._apply_agent_changelog("https://agent", _card(), TaskStepContext(), sandbox_type="local") assert [item["sequence"] for item in requests[0]["json"]["increments"]] == [2, 4] @@ -273,7 +274,7 @@ async def test_apply_rejects_duplicate_tool_call_positions(monkeypatch, store): with pytest.raises(ValueError, match="strictly increasing"): await _step(agent_changelog_object_url=namespace)._apply_agent_changelog( - "https://agent", _card(), TaskStepContext() + "https://agent", _card(), TaskStepContext(), sandbox_type="local" ) assert not requests @@ -285,7 +286,7 @@ async def test_apply_rejects_unsequenced_objects_in_a_portable_namespace(monkeyp with pytest.raises(ValueError, match="zero-padded sequence names"): await _step(agent_changelog_object_url=namespace)._apply_agent_changelog( - "https://agent", _card(), TaskStepContext() + "https://agent", _card(), TaskStepContext(), sandbox_type="local" ) assert not requests assert store.granted == [] @@ -308,7 +309,7 @@ async def test_apply_sends_the_empty_baseline_when_no_increment_is_selected( agent_changelog_toolcall_position_exclusive=cutoff, ) - await step._apply_agent_changelog("https://agent", _card(), context) + await step._apply_agent_changelog("https://agent", _card(), context, sandbox_type="local") assert requests[0]["json"]["increments"] == [] assert store.granted == [] @@ -320,7 +321,7 @@ async def test_apply_sends_the_empty_baseline_when_no_increment_is_selected( def _apply_call(store, source_url): return changelog_apply_call( - _APPLY_OBJECTS, store, agent_name="solver", source_url=source_url + _APPLY_OBJECTS, store, agent_name="solver", source_url=source_url, sandbox_type="local" ) @@ -367,7 +368,7 @@ async def test_a_source_is_not_sent_to_an_agent_without_the_object_apply_form(mo with pytest.raises(RuntimeError, match="does not advertise the object form"): await _step(agent_changelog_object_url=namespace)._apply_agent_changelog( - "https://agent", _card(apply_objects=False), TaskStepContext() + "https://agent", _card(apply_objects=False), TaskStepContext(), sandbox_type="local" ) assert not requests diff --git a/tst/unit/task_step/test_deploy_agent_snapshot_objects.py b/tst/unit/task_step/test_deploy_agent_snapshot_objects.py index d58f9ec6..494b3aaf 100644 --- a/tst/unit/task_step/test_deploy_agent_snapshot_objects.py +++ b/tst/unit/task_step/test_deploy_agent_snapshot_objects.py @@ -107,7 +107,7 @@ async def test_load_prefers_objects_for_a_portable_snapshot(monkeypatch, store): ) context = TaskStepContext() - await _step()._load_snapshot("https://agent", _card(), "agent-1", context) + await _step()._load_snapshot("https://agent", _card(), "agent-1", context, sandbox_type="local") sent = requests[0] assert sent["url"] == "https://agent/custom/snapshot" @@ -136,7 +136,7 @@ def recording(url): monkeypatch.setattr(store, "get_object_metadata_at", recording) - await _step()._load_snapshot("https://agent", _card(), "agent-1", TaskStepContext()) + await _step()._load_snapshot("https://agent", _card(), "agent-1", TaskStepContext(), sandbox_type="local") assert loop_threads == [False, False] @@ -155,7 +155,7 @@ async def test_a_legacy_snapshot_cannot_be_restored(monkeypatch, store, card, fi requests = _install(monkeypatch, _Universe(store, files), {"context_id": "wrong"}) with pytest.raises(RuntimeError, match="is not a portable snapshot, so it cannot be restored"): - await _step()._load_snapshot("https://agent", card, "agent-1", TaskStepContext()) + await _step()._load_snapshot("https://agent", card, "agent-1", TaskStepContext(), sandbox_type="local") assert not store.granted assert not requests @@ -167,7 +167,7 @@ async def test_load_validates_the_object_response(monkeypatch, store): with pytest.raises(RuntimeError, match="invalid object response"): await _step()._load_snapshot( - "https://agent", _card(), "agent-1", TaskStepContext() + "https://agent", _card(), "agent-1", TaskStepContext(), sandbox_type="local" ) @@ -180,7 +180,7 @@ async def test_load_ignores_response_fields_it_does_not_know(monkeypatch, store) ) context = TaskStepContext() - await _step()._load_snapshot("https://agent", _card(), "agent-1", context) + await _step()._load_snapshot("https://agent", _card(), "agent-1", context, sandbox_type="local") assert context.metadata["agent_loaded_snapshots"][0]["context_id"] == ( "restored-context" @@ -203,7 +203,7 @@ async def test_a_portable_snapshot_loads_through_objects_or_not_at_all( requests = _install(monkeypatch, _Universe(store, PORTABLE[:1]), {"context_id": "wrong"}) with pytest.raises(RuntimeError, match=refusal): - await _step()._load_snapshot("https://agent", card, "agent-1", TaskStepContext()) + await _step()._load_snapshot("https://agent", card, "agent-1", TaskStepContext(), sandbox_type="local") assert not store.granted assert not requests diff --git a/tst/unit/task_step/test_prompt_agent.py b/tst/unit/task_step/test_prompt_agent.py index 177c8943..2ecf61e3 100644 --- a/tst/unit/task_step/test_prompt_agent.py +++ b/tst/unit/task_step/test_prompt_agent.py @@ -138,7 +138,8 @@ def put(key, *args, **kwargs): # unbound call: the inline branch never touches `self`, so we skip AWS-touching construction. uri = asyncio.run(PromptAgentTaskStep._fetch_trajectory( - object(), "http://a", {"params": {"endpoint": "/t"}}, "SERVER-ID", "s3://b/traj/", "CLIENT-ID")) + object(), "http://a", {"params": {"endpoint": "/t"}}, "SERVER-ID", "s3://b/traj/", + "CLIENT-ID", sandbox_type="local")) assert uri == "s3://b/traj/trajectory-CLIENT-ID.json" assert puts_on_loop == [False] @@ -173,6 +174,9 @@ def get_object_key(self, prefix): def object_url(self, key): return f"s3://b/{key}" + def grants_reach(self, sandbox_type): + return True + def issue_write_grant(self, object_url, *, media_type, max_bytes, expires_in): grants_on_loop.append(_on_loop()) assert object_url == "s3://b/traj/trajectory-CLIENT-ID.json" @@ -210,6 +214,7 @@ def issue_write_grant(self, object_url, *, media_type, max_bytes, expires_in): "SERVER-ID", "s3://b/traj/", "CLIENT-ID", + sandbox_type="local", ) ) diff --git a/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py b/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py index 73d67f53..03546358 100644 --- a/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py +++ b/tst/unit/task_step/verifiers/test_rubrics_verifier_judge_trajectory.py @@ -68,6 +68,9 @@ def __init__(self): def object_url(self, key): return f"s3://bucket/{key}" + def grants_reach(self, sandbox_type): + return True + def get_object_key(self, object_url): return object_url.removeprefix("s3://bucket/") @@ -107,6 +110,7 @@ async def boom(self, method, url, **kwargs): uri = await verifier._fetch_judge_trajectory( judge_a2a_url="http://judge.example", judge_agent_card={}, a2a_server_task_id="t1", + sandbox_type="local", ) assert uri is None @@ -123,6 +127,7 @@ async def boom(self, method, url, **kwargs): card = {"capabilities": {"extensions": [{"uri": A2AAgent.EXT_TRAJECTORY, "params": {}}]}} uri = await verifier._fetch_judge_trajectory( judge_a2a_url="http://judge.example", judge_agent_card=card, a2a_server_task_id="t1", + sandbox_type="local", ) assert uri is None @@ -154,6 +159,7 @@ def fake_upload(trajectory, prefix, *, name=None): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert uri == "s3://bucket/judge_trajectories/verifier_id=verifier-test/trajectory-t1.json" @@ -188,6 +194,7 @@ def list_at(self, prefix): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert uri is None @@ -218,6 +225,7 @@ async def fake_request(self, method, url, **kwargs): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_object_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert store.write_grants == [(uri, "application/json", DEFAULT_TRAJECTORY_MAX_BYTES)] @@ -238,6 +246,7 @@ async def fake_request(self, method, url, **kwargs): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_object_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert uri.startswith("s3://bucket/fx/judge_trajectories/verifier_id=verifier-test/trajectory-") @@ -265,6 +274,7 @@ async def fake_request(self, method, url, **kwargs): uris = [ await verifier._fetch_judge_trajectory( judge_a2a_url="http://judge.example", judge_agent_card=card, a2a_server_task_id="t1", + sandbox_type="local", ) for _ in range(2) ] @@ -293,6 +303,7 @@ async def fake_request(self, method, url, **kwargs): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_object_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert sent == [{"task_id": "t1"}] @@ -316,6 +327,7 @@ async def boom(self, method, url, **kwargs): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_object_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert uri is None @@ -338,6 +350,7 @@ async def boom(self, method, url, **kwargs): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert uri is None assert attempts == RubricsVerifierTaskStep.DEFAULT_MAX_RETRIES @@ -368,6 +381,7 @@ async def flaky(self, method, url, **kwargs): judge_a2a_url="http://judge.example", judge_agent_card=_card_with_trajectory_ext(), a2a_server_task_id="t1", + sandbox_type="local", ) assert uri == "s3://bucket/trajectory.json" assert attempts == 2 @@ -388,9 +402,12 @@ async def fake_poll(*args, **kwargs): monkeypatch.setattr(protocol, "send_a2a_message", fake_send) monkeypatch.setattr(protocol, "poll_a2a_task", fake_poll) - async def fake_fetch(self, *, judge_a2a_url, judge_agent_card, a2a_server_task_id): + async def fake_fetch( + self, *, judge_a2a_url, judge_agent_card, a2a_server_task_id, sandbox_type + ): assert judge_a2a_url == "http://judge.example" assert a2a_server_task_id == "task-1" + assert sandbox_type is None return "s3://bucket/judge_trajectories/verifier_id=verifier-test/trajectory-task-1.json" monkeypatch.setattr(RubricsVerifierTaskStep, "_fetch_judge_trajectory", fake_fetch)