diff --git a/AGENTS.md b/AGENTS.md index 22cd0a578..bb6605f84 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -114,7 +114,7 @@ and deserialized through a registry. | Package | Contents | |---|---| | `env/` | `Env` base class: a subclass declares `type`, implements `from_dict` and `async deploy(...) -> DeployedEnv`, and may override `reset`, `close` and `load_file_artifact_universe`; teardown closes each env a run deployed before terminating its sandboxes by id. Built-ins in `env/envs/`: `MCPServerEnv` (`mcp_server`), `MultiEnv` (`multi`, MCP servers plus websites), `ServiceDBEnv` (`service_db`, PostgreSQL), `WebsiteEnv` (`website`), `GatewayEnv` (`gateway_server`), `ComputerDriverEnv` (`computer_driver`, `env/envs/computer/driver_env.py`: a computer driver's files and what its class declares for each platform it supports, read at put in a process of its own, `agentenv_protocol.computer._declare`, where the driver's own dependencies needn't be installed, with the `port` it serves on and its install by OS, `[install.]` (`env/envs/computer/utils/install.py`): files, apt `packages` with extra `sources`, a `python` requirements file for a venv, `downloads` by URL and sha256, `run` scripts, the services the base's init runs, and `ports`; it deploys only as a computer env's driver), `ComputerEnv` (`computer`, in `env/envs/computer/`: a base computer and a driver, matched at put by gate 1 before anything is stored, then composed by `composer.py` into an image for each form and arch on a builder of the put's sandbox provider, by default the env's (`utils/builder.py`; `--sandbox` on `env build`, `bundle register` and `run`): this machine's Docker for `local`, else a VM of the provider, which fetches the put's stored files by signed URL, takes the rest inline and pushes to the image store itself, so it builds only with stores a remote VM reaches; each arch goes to the first link that runs it natively, else one that emulates it (a VM registers QEMU; `--no-emulate` leaves it `skipped` instead, for a later put), and its pair records the provider that built it (`builder`); a run or register that reuses a computer env composes what it has left (`pending`); tagged `agentenv-computers:` in the image store by a key of the image's Dockerfile, build context and lock so any env with the same inputs reuses it; every layer's packages, requirements and downloads are resolved at put in a container of the base for each arch (`utils/packages.py`, `utils/resolve_in_base.py`), stored in the object store by sha256 under `agentenv-packages/`, and listed in a lock stored by a digest of its inputs, which any env with the same inputs reuses, and builds install only the lock's files, every install step `RUN --network=none`; each pair's status goes into its metadata, `composed`, the arches built are written as its computer, a `computer` artifact over one index, and `agent-env env build` resumes what a put left; gate 3 then boots each native build on the put's sandbox provider and probes it at `full`, making it `ready` or `failed`, and a build no provider runs natively stays `unverified`. Its deploy, through the `computer` env provider (`providers/env_providers/env_computer_provider.py`), boots its first `ready` build on a sandbox provider that runs its arch natively (`native_platforms()`), with no gateway in front and the driver's card as the env card, then checks its `display`; `deploy_env` `env_params` take a `form`, `computer_settings` (deploy-phase values of the base's settings; a secret one, here or on the env, only as a bare `secret:`/`env:` reference) and `driver_settings` (the driver's `ComputerSettings`, such as its default coordinate space); a `network` that allows nothing runs it on local Docker's internal network behind a relay (`providers/sandbox_providers/local_docker/relay/`); `reset` and `close` have the driver end its recordings first (`computer-recording/v1` `end_all`, each uploading what's left and its manifest under the grant it started with), and `reset` then boots a fresh container in its place (`reset_container`), keeping the computer's address but not an agent's MCP session to the old driver; its `load_file_artifact_universe` writes files under the session user's home through the driver's `data/add`, `Downloads` by default, each file's bytes through a read grant when the store's grants reach it and its network allows everything, else inline (`env/envs/computer/utils/inside.py`)), `ComputerAppEnv` (`computer_app`, `env/envs/computer/app_env.py`: an app on a computer, a `ComputerAppEnvironment` class its env.toml's `module` names, read at put by the same `_declare`, which declares each OS with its own decorator (`@linux(...)`, or `@install(os_type, spec)` for any OS type) holding its install, its launch and, optionally, a skill and settings and build settings that add to the class's there, and install steps and ready checks for every OS or those their `os=` names; everything that reads its settings, launch or steps asks for the computer's OS type; deployed alone it is a one-app computer env, a `ComputerEnv` of the computer and driver its env.toml names, the composer installing it under `/opt/agentenv/apps//` with its services as supervisord programs beside its own server (`agentenv_protocol.computer.serve_app`, in a venv with the protocol's `app` extra, on `.agentenv/server.sock` in its folder), which adds its launcher, launches a `boot` app once per boot and says through `computer-app/v1` whether it's ready, as the driver's `computer-ready` asks, its card named after it, and `env_params` `app_settings` handing it values of its `Settings` in `AGENTENV_APP_SETTINGS`, a secret one only as a bare reference; its env.toml's `build_settings`, values of its `BuildSettings` checked at put, are written to `.agentenv/build_settings.json` in its folder, so they're part of the image's key, its install steps read them, and a deploy can't change them). An env type whose services take environment data defines `load_environment_artifact` and `load_environment_universe_artifact`; `load_artifact` refuses an environment artifact aimed at a type that doesn't, at save and before it loads. An env built on the protocol SDK can serve agents a skill (`@agent_skill`, `skill/v1`), which `add_skills` `env_ids` installs under its environment name (behind a gateway, each server's under that server's name); a skill with files beside its SKILL.md goes as a bundle stored once per content under `agentenv-env-skills/` in the object store. `env/gateway/gateway.py` is the env-side gateway server, container-only code. While its clock is armed it serves agents `get_time` (`env/gateway/get_time.py`), which answers in UTC and, given an IANA `time_zone`, also in that zone; executors' working `hours` (`env/gateway/triggers.py`) take a fixed `utc_offset` or an IANA `time_zone`. | -| `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Every artifact but an environment universe, whose `metadata` key holds its legacy metadata file ids, carries a `metadata` dict outside its versioned fields, changed by compare-and-swap (`merge_metadata`). Built-ins: `CliArtifact` (`cli`), `ComputerArtifact` (`computer`, a base computer: its architectures, OS, session user, settings and container form, whose image `put` reads from its registry into `metadata.validated_images`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`), `EnvironmentUniverseArtifact` (`environment_universe`), `SkillArtifact` (`skill`); `[artifacts] type_aliases` maps another stored spelling of a type onto one of them. A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or a published image's tag with its `image_digest` and its build context, which sandboxes pull by the digest and build from the context when they can't (the infra images, and a bundle agent built from a shipped agent's sources, `Plan.published`), or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/docker/context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. | +| `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Every artifact but an environment universe, whose `metadata` key holds its legacy metadata file ids, carries a `metadata` dict outside its versioned fields, changed by compare-and-swap (`merge_metadata`). Built-ins: `CliArtifact` (`cli`), `ComputerArtifact` (`computer`, a base computer: its architectures, OS, session user, settings and container form, whose image `put` reads from its registry into `metadata.validated_images`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`: its environment's files by path, stored as the one wrapped `file_artifact_ref` every environment artifact had when it has one file at its filename, else as `file_artifact_refs`; `get_file_artifacts()` reads either, and a load hands each file to the env's `data/add` with its path, which a pre-v1 load, taking one file, refuses), `EnvironmentUniverseArtifact` (`environment_universe`), `SkillArtifact` (`skill`); `[artifacts] type_aliases` maps another stored spelling of a type onto one of them. A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or a published image's tag with its `image_digest` and its build context, which sandboxes pull by the digest and build from the context when they can't (the infra images, and a bundle agent built from a shipped agent's sources, `Plan.published`), or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/docker/context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. | | `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, `start_computer_recording` and `end_computer_recording`, and more). On an env whose card serves computer-exec/v1 (a computer), `run_code` with `env_id` and `run_container_unit_tests_verifier` with `env_id` run inside it as its session user through the driver: files staged with `data/add` and read with `data/get`, commands run with `run`; `env_outcome_verifier` hands a `verify` that takes an `env_url` keyword the env's address. `teardown_run` and the `teardown_sandboxes` step share one core (`task/teardown.py` `tear_down`): the recordings the run left open end with fresh grants, each env still up closes (`Env.close`; a computer env ends any recording still open), what stayed open is registered from its namespace, then every sandbox is terminated by id. `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). | | `store/` | Four store ABCs with local and cloud implementations: `DocumentStore` (SQLite, MongoDB, DynamoDB, Firestore with MongoDB compatibility), `ObjectStore` (filesystem, S3, Cloud Storage), `ImageStore` (local OCI registry, ECR), `SecretStore` (env vars or file, AWS Secrets Manager, Google Cloud Secret Manager). `VersionedEntityStore` implements the shared versioned get/put logic, `QueryBuilder` is the immutable chained query API, `store/base.py` holds the error types. `store/routing.py` is the `@local` namespace: while the CLI or a `build_app` server runs, an `@local/…` id and what an `@local` task's run records go to local stores beside the configured ones, and that run's writes to the configured stores are refused. A new backend must pass the conformance kits in `tst/store/`. | | `config/` | The `Config` singleton (`get_config`, `configure`, `reset_config`) in `config/runtime.py`, file discovery in `config/loader.py`, and `load_impl`, which resolves `module:Class` pointers. `agent_env.store` re-exports the config names for compatibility. | diff --git a/packages/agentenv-protocol/tests/test_computer_queue.py b/packages/agentenv-protocol/tests/test_computer_queue.py index 93ca1f07f..fdf6e1f36 100644 --- a/packages/agentenv-protocol/tests/test_computer_queue.py +++ b/packages/agentenv-protocol/tests/test_computer_queue.py @@ -338,17 +338,21 @@ async def test_keys_held_past_max_hold_ms_are_released(): @pytest.mark.asyncio async def test_a_second_hold_restarts_the_release_timer(): - driver = FakeDriver(settings=ComputerSettings(max_hold_ms=150)) + """Timed by the clock: the keys go up no sooner than max_hold_ms after the second hold was sent, where the first + hold's timer would have released them 0.3 s sooner, and a slow loop has 1.2 s to land the second hold before it.""" + driver = FakeDriver(settings=ComputerSettings(max_hold_ms=1500)) async with mcp_session(driver) as session: await session.call_tool("computer_keyboard_down", {"keys": ["Alt"]}) - await asyncio.sleep(0.1) + await asyncio.sleep(0.3) + second = time.monotonic() await session.call_tool("computer_keyboard_down", {"keys": ["Shift"]}) - await asyncio.sleep(0.1) - still = list(driver.screen.held) - with anyio.fail_after(5): + both = list(driver.screen.held) + with anyio.fail_after(10): while driver.screen.held: await asyncio.sleep(0.01) - assert still == ["Alt", "Shift"] and driver.screen.log[-1] == ("key_up", ("Shift", "Alt")) + released = time.monotonic() + assert both == ["Alt", "Shift"] and driver.screen.log[-1] == ("key_up", ("Shift", "Alt")) + assert released - second >= 1.5 @pytest.mark.asyncio diff --git a/src/agent_env/artifact/artifacts/environment.py b/src/agent_env/artifact/artifacts/environment.py index 957a27875..aaa998456 100644 --- a/src/agent_env/artifact/artifacts/environment.py +++ b/src/agent_env/artifact/artifacts/environment.py @@ -1,9 +1,10 @@ -"""Environment artifact for wrapping a FileArtifact with environment identity.""" +"""Environment artifact: the files of the environment it seeds, by path, with its name.""" from __future__ import annotations import logging from abc import ABC, abstractmethod +from pathlib import PurePosixPath from typing import TYPE_CHECKING, Any, ClassVar, Literal, Optional from pydantic import ConfigDict, Field, model_serializer @@ -11,8 +12,10 @@ from agent_env.artifact.artifact import Artifact from agent_env.artifact.artifacts.file import FileArtifact from agent_env.artifact.ref import ArtifactRef +from agent_env.entity.export import Unwritable from agent_env.entity.refs import EntityRef -from agent_env.store.ids import derive_id +from agent_env.store.ids import derive_id, derive_path_id +from agent_env.utils.paths import validate_file_path logger = logging.getLogger(__name__) @@ -33,12 +36,14 @@ def seeded_environments(cls, ctx: AuthoringContext) -> tuple[tuple[str, str], .. class EnvironmentArtifact(Artifact, SeedsEnvironments): - """A FileArtifact wrapped with the name of the environment it seeds.""" + """The files of the environment it seeds, by path, with its name: one wrapped FileArtifact, as every environment + artifact was before it could hold several, or ``file_artifact_refs`` when it holds several or one inside a folder.""" model_config = ConfigDict(populate_by_name=True) # An artifact.toml names the environment and either the artifact to wrap (file) or nothing, to wrap the folder's - # one file, whose description it may set. The stored document's own names for these aren't taken. + # one file, whose description it may set, or keep its several. The stored document's own names for these aren't + # taken. toml_keys: ClassVar[dict[str, type]] = {"environment_name": str, "file": object, "description": str} toml_refs: ClassVar[tuple[EntityRef, ...]] = (EntityRef.artifact("file", artifact_type="file"),) toml_stored_names: ClassVar[dict[str, str]] = {"service_name": "environment_name", "file_artifact_id": "file", @@ -56,6 +61,11 @@ class EnvironmentArtifact(Artifact, SeedsEnvironments): alias="file_artifact_id", description="Legacy unpinned wrapped-FileArtifact id", ) + file_artifact_refs: Optional[dict[str, ArtifactRef]] = Field( + default=None, + description="Pinned (id, version) refs to the environment's FileArtifacts by path, in place of " + "file_artifact_ref when it has several, or one inside a folder", + ) # No return annotation: pydantic builds the serialization schema from one, and a dict drops the fields. @model_serializer(mode="wrap") @@ -65,8 +75,12 @@ def _serialize(self, handler: Any): data = handler(self) if "legacy_file_artifact_id" in data: data["file_artifact_id"] = data.pop("legacy_file_artifact_id") - if self.file_artifact_ref is not None: # new doc + if self.file_artifact_ref is not None or self.file_artifact_refs is not None: # new doc data.pop("file_artifact_id", None) + if self.file_artifact_refs is None: + data.pop("file_artifact_refs", None) + else: + data.pop("file_artifact_ref", None) # Dual-write. Sourced from the attribute, not from `data`, # because `alias=` emits exactly one spelling and which one depends on the # caller's `by_alias` — the store persists with by_alias=True @@ -81,16 +95,30 @@ def put( id: str, *, environment_name: Optional[str] = None, - file_artifact: FileArtifact, + file_artifact: Optional[FileArtifact] = None, + file_artifacts: Optional[dict[str, FileArtifact]] = None, ) -> EnvironmentArtifact: + """Write the environment of ``environment_name`` over ``file_artifact``, or ``file_artifacts`` by path. One file + whose path is its filename is written as ``file_artifact``, the document every environment artifact had.""" from agent_env.artifact.store import get_artifact_store if not environment_name: raise ValueError("environment_name cannot be empty") + if (file_artifact is None) == (file_artifacts is None): + raise ValueError("an environment artifact holds a file_artifact or file_artifacts, one of the two") + if file_artifacts is not None: + if not file_artifacts: + raise ValueError("file_artifacts can't be empty") + for path in file_artifacts: + validate_file_path(path) + if len(file_artifacts) == 1 and (path := next(iter(file_artifacts))) == file_artifacts[path].filename: + file_artifact, file_artifacts = file_artifacts[path], None instance = cls( id=id, environment_name=environment_name, - file_artifact_ref=file_artifact.as_ref(), + file_artifact_ref=file_artifact.as_ref() if file_artifact is not None else None, + file_artifact_refs=({path: file.as_ref() for path, file in file_artifacts.items()} + if file_artifacts is not None else None), ) return get_artifact_store().put_document(instance) @@ -102,19 +130,34 @@ def derived_file_id(id: str) -> str: @classmethod def from_toml(cls, data: dict, ctx: AuthoringContext) -> EnvironmentArtifact: """Write the environment authored as ``data`` (its artifact.toml, with ``file`` resolved to a file artifact's - id) under ``ctx.id`` and return it: over the artifact ``file`` names, or over the folder's one file, written - as ``__file`` with ``description`` defaulting to its name.""" + id) under ``ctx.id`` and return it: over the artifact ``file`` names; over the folder's one file, written as + ``__file`` with ``description`` defaulting to its name; or over its several files by path, each written + under the id a file universe's is (``derive_path_id``).""" fields = cls.accept_toml(data, ctx) if "file" in fields: file = ctx.artifact(fields["file"], FileArtifact) - else: + elif len(files := ctx.files()) == 1: filename, path = ctx.file() file = FileArtifact.put_attempt(cls.derived_file_id(ctx.id), file_path=str(path), filename=filename, description=fields.get("description", filename)) + else: + return cls.put(ctx.id, environment_name=fields["environment_name"], file_artifacts={ + path: FileArtifact.put_attempt(derive_path_id(ctx.id, path), file_path=str(local), + filename=PurePosixPath(path).name, description=path) + for path, local in files.items()}) return cls.put(ctx.id, environment_name=fields["environment_name"], file_artifact=file) def to_toml(self, out: ExportContext) -> dict: - """The artifact.toml of a folder holding the environment's file, which ``from_toml`` reads back as this.""" + """The artifact.toml of a folder holding the environment's files at their paths, which ``from_toml`` reads back + as this. One file kept at a path other than its filename is refused: a folder's one file reads back at its + filename.""" + if self.file_artifact_refs is not None: + if len(self.file_artifact_refs) == 1: + raise Unwritable(f"its one file is kept at {next(iter(self.file_artifact_refs))!r}, not at its filename, " + "and a folder's one file reads back kept at its filename") + for path, file in self.get_file_artifacts().items(): + out.file(path, file.object_url) + return {"type": self.type, "environment_name": self.environment_name} file = self.get_file_artifact() out.file(file.filename, file.object_url) described = {} if file.description == file.filename else {"description": file.description} @@ -122,7 +165,7 @@ def to_toml(self, out: ExportContext) -> dict: @classmethod def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict: - """The keys of ``data``, an artifact.toml, an environment takes, and its folder: its one file, or only + """The keys of ``data``, an artifact.toml, an environment takes, and its folder: its files, or only artifact.toml when ``file`` names the artifact to wrap. Raises BundleError listing every problem.""" problems = [] if (files := ctx.try_files(empty_ok=True)) is not None: # None when the plan's walk reports why @@ -130,13 +173,13 @@ def accept_toml(cls, data: dict, ctx: AuthoringContext) -> dict: if files: problems.append(f"its file is the artifact file names, so the folder holds only artifact.toml, and " f"it also has {', '.join(repr(name) for name in files)}; give the file an artifact " - "folder of its own, or leave out file to wrap the folder's one file") + "folder of its own, or leave out file to keep the folder's files") elif not files: problems.append(ctx.no_files("file for the environment to wrap", - "put its one file here, or name the artifact to wrap with file")) - elif problem := ctx.not_one_file(files, "name the artifact to wrap with file, or give each other file an " - "artifact folder of its own"): - problems.append(problem) + "put its files here, or name the artifact to wrap with file")) + elif "description" in data and len(files) > 1: + problems.append(ctx.config_problem(f"description describes the folder's one file, and it holds " + f"{', '.join(repr(name) for name in files)}; drop description")) data, renamed = ctx.renamed(data, cls.toml_stored_names) problems += renamed fields, more = ctx.accepted(data, **cls.toml_keys) @@ -158,7 +201,19 @@ def seeded_environments(cls, ctx: AuthoringContext) -> tuple[tuple[str, str], .. name = ctx.entry.config.get("environment_name") if isinstance(ctx.entry.config, dict) else None return (("", name),) if isinstance(name, str) and name else () + def get_file_artifacts(self) -> dict[str, FileArtifact]: + """The environment's files by path: ``{filename: file}`` for the one file it wraps.""" + if self.file_artifact_refs is not None: + return {path: FileArtifact.get(ref.id, version=ref.version) for path, ref in self.file_artifact_refs.items()} + file = self.get_file_artifact() + return {file.filename: file} + def get_file_artifact(self) -> FileArtifact: + """The one file the environment wraps; refused for one that holds several, whose files + ``get_file_artifacts`` reads.""" + if self.file_artifact_refs is not None: + raise ValueError(f"EnvironmentArtifact {self.id} v{self.version} holds {len(self.file_artifact_refs)} files " + f"by path, not one; read them with get_file_artifacts()") if self.file_artifact_ref is not None: return FileArtifact.get( self.file_artifact_ref.id, version=self.file_artifact_ref.version diff --git a/src/agent_env/artifact/artifacts/environment_universe.py b/src/agent_env/artifact/artifacts/environment_universe.py index 458677e0b..ae9794f90 100644 --- a/src/agent_env/artifact/artifacts/environment_universe.py +++ b/src/agent_env/artifact/artifacts/environment_universe.py @@ -184,6 +184,9 @@ def to_toml(self, out: ExportContext) -> dict: name = environment.environment_name if name.split("/", 1)[0] == METADATA: raise Unwritable(f"its environment {name!r} would read back as metadata, from its {METADATA}/ folder") + if environment.file_artifact_refs is not None: + raise Unwritable(f"its environment {name!r} holds several files, and a universe's folder holds one per " + f"environment; write environment artifact {environment.id} to a folder of its own") file = environment.get_file_artifact() out.file(environment_path(Path(), name, file.filename).as_posix(), file.object_url) for key, file in (self.get_metadata() or {}).items(): @@ -255,7 +258,8 @@ def get_environment_artifacts(self) -> list[EnvironmentArtifact]: ) def get_file_artifacts(self) -> dict[str, FileArtifact]: - """``{relative filename: FileArtifact}`` — a plain-file view of this universe. + """``{/: FileArtifact}`` — a plain-file view of this universe, each environment's + files at their paths (one file's at its filename). Matching ``FileArtifactUniverse.get_file_artifacts()`` is what lets the same loaders stage a service universe as a file tree (grading a frozen snapshot) @@ -264,14 +268,14 @@ def get_file_artifacts(self) -> dict[str, FileArtifact]: """ out: dict[str, FileArtifact] = {} for sa in self.get_environment_artifacts(): - fa = sa.get_file_artifact() - name = f"{sa.environment_name}/{fa.filename}" - if name in out: - raise ValueError( - f"EnvironmentUniverseArtifact {self.id} v{self.version} has duplicate " - f"environment '{sa.environment_name}'; cannot stage it as a file tree" - ) - out[name] = fa + for path, fa in sa.get_file_artifacts().items(): + name = f"{sa.environment_name}/{path}" + if name in out: + raise ValueError( + f"EnvironmentUniverseArtifact {self.id} v{self.version} has duplicate " + f"environment '{sa.environment_name}'; cannot stage it as a file tree" + ) + out[name] = fa return out def get_metadata(self) -> dict[str, FileArtifact]: diff --git a/src/agent_env/artifact/artifacts/file_artifact_universe.py b/src/agent_env/artifact/artifacts/file_artifact_universe.py index b2d125ceb..72302eab2 100644 --- a/src/agent_env/artifact/artifacts/file_artifact_universe.py +++ b/src/agent_env/artifact/artifacts/file_artifact_universe.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib import logging from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, Literal, Optional @@ -13,7 +12,7 @@ from agent_env.artifact.ref import ArtifactRef from agent_env.artifact.universe import Universe from agent_env.config import get_config -from agent_env.store.ids import derive_id +from agent_env.store.ids import derive_path_id logger = logging.getLogger(__name__) @@ -157,9 +156,8 @@ def put_bundled( raise ValueError(f"Bundle key {rel_path!r} must be a relative path") if ".." in rel_path.split("/"): raise ValueError(f"Bundle key {rel_path!r} must not contain '..' segments") - path_hash = hashlib.sha256(rel_path.encode("utf-8")).hexdigest()[:16] fa = FileArtifact.put_at( - id=derive_id(id, path_hash), + id=derive_path_id(id, rel_path), description=f"Bundled file '{rel_path}' of FileArtifactUniverse '{id}'", file_path=str(local_path), object_url=prefix_url + rel_path, @@ -218,9 +216,8 @@ def put_existing( rel_path = object_url[len(prefix_url):] if not rel_path: continue - path_hash = hashlib.sha256(rel_path.encode("utf-8")).hexdigest()[:16] fa = FileArtifact.put_existing( - id=derive_id(id, path_hash), + id=derive_path_id(id, rel_path), description=f"Bundled file '{rel_path}' of FileArtifactUniverse '{id}'", object_url=object_url, ) diff --git a/src/agent_env/env/envs/_deployment.py b/src/agent_env/env/envs/_deployment.py index 88b169943..8cd3858d0 100644 --- a/src/agent_env/env/envs/_deployment.py +++ b/src/agent_env/env/envs/_deployment.py @@ -6,18 +6,20 @@ import asyncio import inspect import logging +import uuid from typing import TYPE_CHECKING, Optional -from agentenv_protocol import METHOD_ADD, METHOD_RESET, RPC_PATH, FilePart, client as protocol_v1 +from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI, METHOD_ADD, METHOD_RESET, RPC_PATH, FilePart, client as protocol_v1 from agent_env.env import legacy_protocol from agent_env.env.env import DeployedSandboxEnv, gateway_url_of from agent_env.env.store import register_env_instance from agent_env.store.base import ObjectNotFoundError +from agent_env.utils.paths import validate_file_name if TYPE_CHECKING: from collections.abc import Iterable - from agent_env.artifact import FileArtifact + from agent_env.artifact import EnvironmentArtifact, FileArtifact from agent_env.env.env import DeployedEnv, Env from agent_env.env.envs.mcp_server import MCPServerEnv from agent_env.env.envs.website import WebsiteEnv @@ -100,9 +102,9 @@ async def close_replaced(env: Env) -> None: raise -async def load_by_signed_url(env: MCPServerEnv | WebsiteEnv, file_artifact: FileArtifact) -> None: - """Load into a server a plugin's provider deployed, which has no sandbox of ours to stage into: it fetches the file from a - signed URL. The server may be a MultiEnv's child, whose record is the MultiEnv's.""" +async def load_by_signed_url(env: MCPServerEnv | WebsiteEnv, environment_artifact: EnvironmentArtifact) -> None: + """Load into a server a plugin's provider deployed, which has no sandbox of ours to stage into: it fetches each file from + a signed URL. The server may be a MultiEnv's child, whose record is the MultiEnv's.""" from agent_env.config import get_config from agent_env.env.gateway.constants import data_plane_load_timeout_s @@ -115,22 +117,55 @@ async def load_by_signed_url(env: MCPServerEnv | WebsiteEnv, file_artifact: File raise RuntimeError(f"{deployed_by}, whose card offers '{env.environment_name}' no data plane to load into") if operations is not None and (missing := [op for op in LOAD_OPERATIONS if op not in operations]): raise RuntimeError(f"{deployed_by}, whose card offers '{env.environment_name}' no {', '.join(missing)} operation to load with") - store = get_config().get_object_store_at(file_artifact.object_url) - metadata = await asyncio.to_thread(store.get_object_metadata_at, file_artifact.object_url) - if metadata is None: - raise ObjectNotFoundError(f"No object at {file_artifact.object_url}") - timeout = data_plane_load_timeout_s(metadata.size) + files = environment_artifact.get_file_artifacts() + stores = {path: get_config().get_object_store_at(file.object_url) for path, file in files.items()} + size = 0 + for path, file in files.items(): + metadata = await asyncio.to_thread(stores[path].get_object_metadata_at, file.object_url) + if metadata is None: + raise ObjectNotFoundError(f"No object at {file.object_url}") + size += metadata.size + timeout = data_plane_load_timeout_s(size) # Signed before the reset, so a store that can't sign leaves the env's data as it is; it lasts one timeout per call. - url = await asyncio.to_thread(store.signed_get_url, file_artifact.object_url, len(LOAD_OPERATIONS) * timeout) - if url is None: - raise RuntimeError(f"{deployed_by}, so its server fetches {file_artifact.object_url} itself, and {type(store).__name__} " - "can't sign a URL for it; loading into such an env needs an object store that signs URLs (S3, Cloud Storage)") + urls = {} + for path, file in files.items(): + urls[path] = await asyncio.to_thread(stores[path].signed_get_url, file.object_url, len(LOAD_OPERATIONS) * timeout) + if urls[path] is None: + raise RuntimeError(f"{deployed_by}, so its server fetches {file.object_url} itself, and {type(stores[path]).__name__} " + "can't sign a URL for it; loading into such an env needs an object store that signs URLs (S3, Cloud Storage)") await protocol_v1.reset_data(base_url, timeout=timeout) - await protocol_v1.add_data(base_url, [FilePart(file={ - "uri": url, - "mimeType": file_artifact.content_type, - "name": file_artifact.filename, - })], timeout=timeout) + await protocol_v1.add_data(base_url, file_parts(environment_artifact, files, urls), timeout=timeout) + + +def file_parts(environment_artifact: EnvironmentArtifact, files: dict[str, FileArtifact], uris: dict[str, str]) -> list[FilePart]: + """The parts ``data/add`` takes ``environment_artifact``'s ``files`` as, each fetched from its uri in ``uris``: one file + as the one part it has always been sent as, each of several with its path in the data-objects marker.""" + by_path = environment_artifact.file_artifact_refs is not None + return [FilePart(file={"uri": uris[path], "mimeType": file.content_type, "name": file.filename}, + **({"metadata": {DATA_OBJECTS_EXTENSION_URI: {"path": path}}} if by_path else {})) + for path, file in files.items()] + + +def staged_paths(environment_artifact: EnvironmentArtifact, files: dict[str, FileArtifact]) -> tuple[Optional[str], dict[str, str]]: + """Where a load stages each of ``files`` in the server's data folder, by path, and the folder of the load's own that + holds them, which the load removes after it: one file at its filename, a plain name, as it always has been, with no + folder; several each at its path inside a new folder, so no earlier load's files are in their way.""" + if environment_artifact.file_artifact_refs is None: + filename = next(iter(files)) + validate_file_name(filename) + return None, {filename: filename} + folder = f".agentenv-load-{uuid.uuid4().hex[:8]}" + return folder, {path: f"{folder}/{path}" for path in files} + + +def legacy_load_file(env: MCPServerEnv | WebsiteEnv, environment_artifact: EnvironmentArtifact) -> FileArtifact: + """The file a server's pre-v1 load takes, which is one; refused for an environment artifact holding several.""" + if environment_artifact.file_artifact_refs is not None: + raise RuntimeError(f"'{env.environment_name}' takes one file through its pre-v1 load, and environment artifact " + f"'{environment_artifact.id}' holds {len(environment_artifact.file_artifact_refs)}") + file = environment_artifact.get_file_artifact() + validate_file_name(file.filename) + return file async def close_deployed(deployed: DeployedEnv, env_class: type[Env]) -> None: diff --git a/src/agent_env/env/envs/mcp_server.py b/src/agent_env/env/envs/mcp_server.py index 29104edae..2f950f77a 100644 --- a/src/agent_env/env/envs/mcp_server.py +++ b/src/agent_env/env/envs/mcp_server.py @@ -5,12 +5,13 @@ import asyncio import logging import os +import posixpath import shlex import uuid from importlib.metadata import version as pkg_version from typing import TYPE_CHECKING, Any, Callable, ClassVar, Optional -from agentenv_protocol import FilePart, client as protocol_v1 +from agentenv_protocol import client as protocol_v1 from agent_env.artifact import Artifact, DockerImageArtifact from agent_env.artifact.artifacts.docker_image import ProgressCallback, refuse_local_github_build from agent_env.config import get_config @@ -20,12 +21,13 @@ from agent_env.env import legacy_protocol from agent_env.env.envs._deployment import ( as_builtin, builtin_provider_for, close_deployed, close_replaced, deploy_refusal, deploy_through_provider, host_staging_refusal, - load_by_signed_url, plugin_provider_like_a_builtin, provider_or_class, + file_parts, legacy_load_file, load_by_signed_url, plugin_provider_like_a_builtin, provider_or_class, staged_paths, ) from agent_env.env.gateway import GatewayMode +from agent_env.providers.sandbox_providers.sandbox import VmSandbox from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER from agent_env.attribution import Attribution -from agent_env.utils.paths import validate_file_name +from agent_env.utils.paths import validate_file_path if TYPE_CHECKING: from agent_env.artifact import CliArtifact, EnvironmentArtifact, EnvironmentUniverseArtifact @@ -158,29 +160,32 @@ async def load_environment_artifact(self, environment_artifact: EnvironmentArtif f"EnvironmentArtifact environment_name '{environment_artifact.environment_name}' " f"does not match env environment_name '{self.environment_name}'" ) - file_artifact = environment_artifact.get_file_artifact() if builtin is None: - await load_by_signed_url(self, file_artifact) + await load_by_signed_url(self, environment_artifact) return base_url = await legacy_protocol.v1_base_url(self._deployed, self._gateway_url, self.environment_name, mcp=True) if base_url is not None: from agent_env.env.gateway.constants import data_plane_load_timeout_s - container_path = await self._copy_artifact_into_container(file_artifact) - # Sized from the payload rather than a flat constant: a 3.5GB service and a 30KB one - # were previously given the same 600s, which killed healthy large loads. The size is - # the store's for the object staging copied, not whatever the artifact document claims. - timeout = data_plane_load_timeout_s(await self._staged_artifact_size(file_artifact)) - # reset_data previously took the client default (30s), tighter than the load it - # precedes -- dropping and recreating a large service's schemas can plausibly - # exceed that. Give it the same budget as the load it belongs to. - await protocol_v1.reset_data(base_url, timeout=timeout) - await protocol_v1.add_data(base_url, [FilePart(file={ - "uri": f"file://{container_path}", - "mimeType": file_artifact.content_type, - "name": file_artifact.filename, - })], timeout=timeout) + files = environment_artifact.get_file_artifacts() + folder, paths = staged_paths(environment_artifact, files) + try: + staged = {path: f"file://{await self._copy_artifact_into_container(file, paths[path])}" + for path, file in files.items()} + # Sized from the payload rather than a flat constant: a 3.5GB service and a 30KB one + # were previously given the same 600s, which killed healthy large loads. The size is + # the store's for the objects staging copied, not whatever the artifact document claims. + sizes = [await self._staged_artifact_size(file) for file in files.values()] + timeout = data_plane_load_timeout_s(None if None in sizes else sum(sizes)) + # reset_data previously took the client default (30s), tighter than the load it + # precedes -- dropping and recreating a large service's schemas can plausibly + # exceed that. Give it the same budget as the load it belongs to. + await protocol_v1.reset_data(base_url, timeout=timeout) + await protocol_v1.add_data(base_url, file_parts(environment_artifact, files, staged), timeout=timeout) + finally: + if folder is not None: + await self._remove_from_container(folder) else: - await self._load_environment_artifact_legacy(file_artifact) + await self._load_environment_artifact_legacy(legacy_load_file(self, environment_artifact)) await builtin.install_changelog_triggers(self.environment_name) async def load_file_artifact_universe(self, file_artifact_universe: "Any", destination_path: Optional[str] = None, @@ -233,10 +238,13 @@ async def _staged_artifact_size(self, file_artifact) -> int | None: return None return metadata.size if metadata is not None else None - async def _copy_artifact_into_container(self, file_artifact) -> str: - validate_file_name(file_artifact.filename) - container_path = f"/data/{file_artifact.filename}" + async def _copy_artifact_into_container(self, file_artifact, path: str) -> str: + validate_file_path(path) + container_path = f"/data/{path}" + folder = posixpath.dirname(container_path) if self._sandbox.mode == SANDBOX_MODE_CONTAINER: + if folder != "/data": + await self._exec_in_server("mkdir", "-p", folder) await self._sandbox.write_file_from_object(file_artifact.object_url, container_path) else: # Stage on the disk-backed app dir, not /tmp (tmpfs/RAM): a large @@ -244,16 +252,39 @@ async def _copy_artifact_into_container(self, file_artifact) -> str: # Random suffix avoids collisions between concurrent loads. from agent_env.providers.env_providers.constants import GATEWAY_APP_DIR stage_dir = f"{GATEWAY_APP_DIR}/_artifact_staging" - vm_temp_path = f"{stage_dir}/{self.environment_name}-{uuid.uuid4().hex[:8]}-{file_artifact.filename}" + vm_temp_path = f"{stage_dir}/{self.environment_name}-{uuid.uuid4().hex[:8]}-{posixpath.basename(path)}" await self._sandbox.exec_script(f"mkdir -p {shlex.quote(stage_dir)}") await self._sandbox.load_object_file(file_artifact.object_url, vm_temp_path) container_id = await self._env_provider._get_container_id(self._sandbox, self.environment_name) - await self._sandbox.exec_script(f"docker exec {shlex.quote(container_id)} mkdir -p /data") + await self._sandbox.exec_script(f"docker exec {shlex.quote(container_id)} mkdir -p {shlex.quote(folder)}") await self._sandbox.docker_cp(vm_temp_path, f"{container_id}:{container_path}", remove_source=True) return container_path + async def _remove_from_container(self, path: str) -> None: + """Remove what a load staged at ``/data/``, best-effort: a load that finished stands either way.""" + container_path = f"/data/{path}" + try: + await self._exec_in_server("rm", "-rf", container_path) + except Exception as e: # noqa: BLE001 -- cleanup, not the load + logger.warning(f"[{self.environment_name}] could not remove {container_path} after a load (ignored): {e}") + + async def _exec_in_server(self, *command: str) -> None: + """Run ``command`` in the server's container, as staging writes there: through the host's docker unless the + sandbox is a container of its own; a VM sandbox's exec runs on its host. Raises when it fails.""" + if self._sandbox.mode != SANDBOX_MODE_CONTAINER: + container = await self._env_provider._get_container_id(self._sandbox, self.environment_name) + elif isinstance(self._sandbox, VmSandbox): + container = self._sandbox.container_name + else: + exit_code, _, stderr = await self._sandbox.exec_with_output(*command) + if exit_code != 0: + raise RuntimeError(f"{shlex.join(command)} failed in {self.environment_name}'s server (exit {exit_code}): " + f"{stderr[-500:]}") + return + await self._sandbox.exec_script(f"docker exec -u 0 {shlex.quote(container)} {shlex.join(command)}") + async def _load_environment_artifact_legacy(self, file_artifact) -> None: - container_path = await self._copy_artifact_into_container(file_artifact) + container_path = await self._copy_artifact_into_container(file_artifact, file_artifact.filename) await legacy_protocol.reset( self._gateway_url, self.environment_name, container_path, max_retries=self._MCP_MAX_RETRIES ) diff --git a/src/agent_env/env/envs/website.py b/src/agent_env/env/envs/website.py index 622c07133..dd2770c49 100644 --- a/src/agent_env/env/envs/website.py +++ b/src/agent_env/env/envs/website.py @@ -5,12 +5,13 @@ import asyncio import logging import os +import posixpath import shlex import uuid from importlib.metadata import version as pkg_version from typing import TYPE_CHECKING, Any, Callable, ClassVar, Optional -from agentenv_protocol import FilePart, client as protocol_v1 +from agentenv_protocol import client as protocol_v1 from agent_env.artifact import Artifact, DockerImageArtifact, EnvironmentArtifact from agent_env.artifact.artifacts.docker_image import ProgressCallback, refuse_local_github_build from agent_env.entity.refs import EntityRef @@ -19,11 +20,11 @@ from agent_env.env import legacy_protocol from agent_env.env.envs._deployment import ( as_builtin, builtin_provider_for, close_deployed, close_replaced, deploy_refusal, deploy_through_provider, host_staging_refusal, - load_by_signed_url, plugin_provider_like_a_builtin, provider_or_class, + file_parts, legacy_load_file, load_by_signed_url, plugin_provider_like_a_builtin, provider_or_class, staged_paths, ) from agent_env.env.gateway import GatewayMode from agent_env.attribution import Attribution -from agent_env.utils.paths import validate_file_name +from agent_env.utils.paths import validate_file_path if TYPE_CHECKING: from agent_env.bundle.authoring import AuthoringContext from agent_env.entity.export import ExportContext @@ -163,22 +164,25 @@ async def load_environment_artifact(self, environment_artifact: EnvironmentArtif f"EnvironmentArtifact environment_name '{environment_artifact.environment_name}' " f"does not match env environment_name '{self.environment_name}'" ) - file_artifact = environment_artifact.get_file_artifact() if builtin is None: - await load_by_signed_url(self, file_artifact) + await load_by_signed_url(self, environment_artifact) return base_url = await legacy_protocol.v1_base_url(self._deployed, self._gateway_url, self.environment_name, mcp=False) if base_url is not None: from agent_env.env.gateway.constants import DATA_PLANE_LOAD_TIMEOUT_S - container_path = await self._copy_artifact_into_container(file_artifact) - await protocol_v1.reset_data(base_url) - await protocol_v1.add_data(base_url, [FilePart(file={ - "uri": f"file://{container_path}", - "mimeType": file_artifact.content_type, - "name": file_artifact.filename, - })], timeout=DATA_PLANE_LOAD_TIMEOUT_S) + files = environment_artifact.get_file_artifacts() + folder, paths = staged_paths(environment_artifact, files) + try: + staged = {path: f"file://{await self._copy_artifact_into_container(file, paths[path])}" + for path, file in files.items()} + await protocol_v1.reset_data(base_url) + await protocol_v1.add_data(base_url, file_parts(environment_artifact, files, staged), + timeout=DATA_PLANE_LOAD_TIMEOUT_S) + finally: + if folder is not None: + await self._remove_from_container(folder) else: - await self._load_environment_artifact_legacy(file_artifact) + await self._load_environment_artifact_legacy(legacy_load_file(self, environment_artifact)) await builtin.install_changelog_triggers(self.environment_name) async def load_file_artifact_universe(self, file_artifact_universe: "Any", destination_path: Optional[str] = None, @@ -187,25 +191,36 @@ async def load_file_artifact_universe(self, file_artifact_universe: "Any", desti raise RuntimeError(refusal) return await super().load_file_artifact_universe(file_artifact_universe, destination_path) - async def _copy_artifact_into_container(self, file_artifact) -> str: + async def _copy_artifact_into_container(self, file_artifact, path: str) -> str: from agent_env.providers.env_providers.constants import AGENT_ENV_WEBSITE_BACKEND_SUFFIX, DOCKER_COMPOSE_PATH - filename = file_artifact.filename - validate_file_name(filename) + validate_file_path(path) backend_service = f"{self.environment_name}-{AGENT_ENV_WEBSITE_BACKEND_SUFFIX}" # Random suffix so concurrent loads on a shared sandbox (parallel /load-universe, # MCP+website sharing a service_name, repeated loads) don't race on a shared path. - vm_path = f"/tmp/_artifact_{self.environment_name}-{uuid.uuid4().hex[:8]}-{filename}" - container_path = f"/tmp/data/{filename}" + vm_path = f"/tmp/_artifact_{self.environment_name}-{uuid.uuid4().hex[:8]}-{posixpath.basename(path)}" + container_path = f"/tmp/data/{path}" compose = f"docker compose -f {shlex.quote(DOCKER_COMPOSE_PATH)}" await self._sandbox.load_object_file(file_artifact.object_url, vm_path) - await self._sandbox.exec_script(f"{compose} exec -T {shlex.quote(backend_service)} mkdir -p /tmp/data") + await self._sandbox.exec_script( + f"{compose} exec -T {shlex.quote(backend_service)} mkdir -p {shlex.quote(posixpath.dirname(container_path))}") await self._sandbox.exec_script(f"{compose} cp {shlex.quote(vm_path)} {shlex.quote(f'{backend_service}:{container_path}')}") await self._sandbox.exec_script(f"rm -f {shlex.quote(vm_path)}") return container_path + async def _remove_from_container(self, path: str) -> None: + """Remove what a load staged at ``/tmp/data/``, best-effort: a load that finished stands either way.""" + from agent_env.providers.env_providers.constants import AGENT_ENV_WEBSITE_BACKEND_SUFFIX, DOCKER_COMPOSE_PATH + backend_service = f"{self.environment_name}-{AGENT_ENV_WEBSITE_BACKEND_SUFFIX}" + container_path = f"/tmp/data/{path}" + try: + await self._sandbox.exec_script(f"docker compose -f {shlex.quote(DOCKER_COMPOSE_PATH)} exec -T " + f"{shlex.quote(backend_service)} rm -rf {shlex.quote(container_path)}") + except Exception as e: # noqa: BLE001 -- cleanup, not the load + logger.warning(f"[{self.environment_name}] could not remove {container_path} after a load (ignored): {e}") + async def _load_environment_artifact_legacy(self, file_artifact) -> None: - container_path = await self._copy_artifact_into_container(file_artifact) + container_path = await self._copy_artifact_into_container(file_artifact, file_artifact.filename) base_url = legacy_protocol.environment_base_url(self._gateway_url, self.environment_name, mcp=False) from agent_env.env.gateway.constants import DATA_PLANE_LOAD_TIMEOUT_S await legacy_protocol.reset_via_rest(base_url, timeout=60) diff --git a/src/agent_env/store/ids.py b/src/agent_env/store/ids.py index 5c21c0759..de933b3d5 100644 --- a/src/agent_env/store/ids.py +++ b/src/agent_env/store/ids.py @@ -136,3 +136,9 @@ def fs_safe(entity_id: str) -> str: def derive_id(base: str, suffix: str) -> str: """The id of an entity derived from ``base``: a suffix, so the base's namespace carries over.""" return f"{base}__{suffix}" + + +def derive_path_id(base: str, path: str) -> str: + """The id of the file at ``path`` written with ``base``, as a file universe's files are: ``__<16 hex of the + path's sha256>``, which any path fits in.""" + return derive_id(base, hashlib.sha256(path.encode("utf-8")).hexdigest()[:16]) diff --git a/src/agent_env/task_step/task_steps/collect_artifacts.py b/src/agent_env/task_step/task_steps/collect_artifacts.py index 5cd475444..c55923a6a 100644 --- a/src/agent_env/task_step/task_steps/collect_artifacts.py +++ b/src/agent_env/task_step/task_steps/collect_artifacts.py @@ -68,7 +68,6 @@ import asyncio import functools -import hashlib import logging import os import json @@ -80,7 +79,7 @@ from agent_env.config import get_config from agent_env.env.gateway.constants import EXT_STEP_URI -from agent_env.store.ids import derive_id, is_local_id, validate_local_id +from agent_env.store.ids import derive_path_id, is_local_id, validate_local_id from agent_env.task_step.context import TaskStepContext from agent_env.entity.refs import EntityRef from agent_env.task_step.task_step import TaskStep, TaskStepDependency @@ -971,7 +970,7 @@ def _register_file_artifact(self, store, file_artifacts, artifact_id, key, objec FileArtifact filename is the basename for clean downloads.""" from agent_env.artifact import FileArtifact - fa_id = derive_id(artifact_id, hashlib.sha256(object_name.encode("utf-8")).hexdigest()[:16]) + fa_id = derive_path_id(artifact_id, object_name) fa = FileArtifact( id=fa_id, description=f"Collected artifact '{object_name}' from task instance {context.instance_id or artifact_id}", diff --git a/src/agent_env/task_step/task_steps/load_artifact.py b/src/agent_env/task_step/task_steps/load_artifact.py index b45f135bf..fdecb9e75 100644 --- a/src/agent_env/task_step/task_steps/load_artifact.py +++ b/src/agent_env/task_step/task_steps/load_artifact.py @@ -790,9 +790,14 @@ def _target(rel): async def _stage_environment_payload_into_container( sandbox, container_name: Optional[str], environment_artifact, destination: str ) -> list[str]: - """Expand an EnvironmentArtifact's payload into a file tree at ``destination`` — in the container, or on the VM host when ``container_name`` is None.""" - from agent_env.providers.sandbox_providers.sandbox import VmSandbox - + """Expand an EnvironmentArtifact's payload into a file tree at ``destination`` — in the container, or on the VM host when ``container_name`` is None. + One holding several files has each written at its path instead.""" + from agent_env.providers.sandbox_providers.sandbox import VmSandbox, stage_files_into_container + + if environment_artifact.file_artifact_refs is not None: + return list(await stage_files_into_container( + sandbox, environment_artifact.get_file_artifacts(), destination, container=container_name, + on_host=container_name is None and isinstance(sandbox, VmSandbox))) file_artifact = environment_artifact.get_file_artifact() token = uuid.uuid4().hex[:8] safe_name = file_artifact.filename.replace("/", "_") diff --git a/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_agent_judge.py b/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_agent_judge.py index ec145079b..2b024957c 100644 --- a/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_agent_judge.py +++ b/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_agent_judge.py @@ -36,6 +36,12 @@ - `{UNIVERSES_DIR}/export_2/.json` — the data after re-loading export_1 and exporting \ again (present only when available). +A service seeded with several files has a folder, `{UNIVERSES_DIR}/original//`, holding \ +each file at its path, in place of `original/.json`. It has no single original JSON to \ +diff, and the programmatic verifier only compares its two exports. Read its files to learn what it \ +was seeded with, judge whether `export_1/.json` still holds that data in meaning, and use \ +`export_2` to corroborate stability. + Your job is to decide whether the round-trip is **semantically lossless**: does the exported data \ preserve the *meaning* of the original? Compare `original` against `export_1` (and use `export_2` \ to corroborate stability). Read the files with your tools; do not guess. @@ -80,7 +86,10 @@ def build_user_prompt() -> str: f"Inspect the universe snapshots under `{UNIVERSES_DIR}/`. For each service, compare " f"`original/.json` against `export_1/.json` (corroborate with " f"`export_2/.json` when present) and identify any MAJOR differences — real data " - f"loss or corruption — applying the IGNORE/FLAG rules from your instructions. " + f"loss or corruption — applying the IGNORE/FLAG rules from your instructions. A service " + f"seeded with several files has the folder `original//` in place of " + f"`original/.json`: judge its export against what those files hold, as your " + f"instructions say. " f"Produce a concise per-service report: for each service state whether it is semantically " f"preserved, and list every major difference you found with the service, entity, field, " f"and an example (original value vs exported value). If a service is fully preserved " diff --git a/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_roundtrip.py b/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_roundtrip.py index 3e5b6b3da..9e8274beb 100644 --- a/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_roundtrip.py +++ b/src/agent_env/task_step/task_steps/multienv_validator/verify_universe_roundtrip.py @@ -113,8 +113,12 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: await env.load_environment_universe_artifact(universe) originals = {} for sa in environment_artifacts: + files = await asyncio.to_thread(sa.get_file_artifacts) + if len(files) > 1: + logger.info(f"{sa.environment_name}: {len(files)} files, not one JSON state; will only compare export1 vs export2") + continue try: - data = await asyncio.to_thread(lambda sa=sa: sa.get_file_artifact().load()) + data = await asyncio.to_thread(next(iter(files.values())).load) originals[sa.environment_name] = normalize(json.loads(data)) except (json.JSONDecodeError, UnicodeDecodeError): logger.info(f"{sa.environment_name}: non-JSON artifact, will only compare export1 vs export2") @@ -180,7 +184,8 @@ def _create_file_artifact_universe(self, original_environment_artifacts: list, e Faithfulness (this validator measures lossiness, so the bundle must not transform data): - ``original/.json`` references the universe's EXISTING per-service FileArtifacts - (verbatim bytes — what compare_dicts loaded), no re-serialization. + (verbatim bytes — what compare_dicts loaded), no re-serialization; a service of several + files has each at ``original//``. - ``export_1/.json`` references the FileArtifacts the export1 universe was built from (the exact bytes the programmatic verdict used). - ``export_2/.json`` is materialized fresh (export2 is not otherwise persisted), dumped @@ -193,7 +198,11 @@ def _create_file_artifact_universe(self, original_environment_artifacts: list, e with artifact_write_lock(prefix): file_artifacts: dict[str, Any] = {} for sa in original_environment_artifacts: - file_artifacts[f"original/{sa.environment_name}.json"] = sa.get_file_artifact() + files = sa.get_file_artifacts() + if len(files) == 1: + file_artifacts[f"original/{sa.environment_name}.json"] = next(iter(files.values())) + else: + file_artifacts.update({f"original/{sa.environment_name}/{path}": file for path, file in files.items()}) for sa in exported_universe.get_environment_artifacts(): file_artifacts[f"export_1/{sa.environment_name}.json"] = sa.get_file_artifact() for name, data in export2_raw.items(): diff --git a/src/agent_env/utils/paths.py b/src/agent_env/utils/paths.py index a0d3984ec..620d89522 100644 --- a/src/agent_env/utils/paths.py +++ b/src/agent_env/utils/paths.py @@ -25,6 +25,15 @@ def validate_relative_filename(filename: str) -> None: raise ValueError(f"Filename {filename!r} must not contain '..' segments") +def validate_file_path(path: str) -> None: + """Reject anything but plain file names joined by ``/``: an env stages each file of a loaded environment at host and + container paths built from its path.""" + for name in path.split("/"): + if name in ("", ".", "..") or "\\" in name or any(ord(c) < 32 or ord(c) == 127 for c in name): + raise ValueError(f"Artifact file path {path!r} must be plain file names joined by '/', with no '.', '..', " + "'\\' or control characters") + + def validate_file_name(filename: str) -> None: """Reject anything but one plain file name: an env stages a loaded artifact at host and container paths built from it.""" if filename in ("", ".", "..") or "/" in filename or "\\" in filename or any(ord(c) < 32 or ord(c) == 127 for c in filename): diff --git a/tst/integration/env/envs/server_env_local_test.py b/tst/integration/env/envs/server_env_local_test.py index 52ed515c1..cf63b3673 100644 --- a/tst/integration/env/envs/server_env_local_test.py +++ b/tst/integration/env/envs/server_env_local_test.py @@ -2,7 +2,9 @@ The env deploys, a later process restores its record and loads data into the server, and the teardown step, which rebuilds each sandbox from disk, removes the container. A load is timed by the size of the payload it staged, through -the deploy's own handle and a restored one; and the server's state reads as JSON whether ``data/get`` answers with +the deploy's own handle and a restored one; an environment of several files is staged file by file at their paths, in a +folder of the load's own that an earlier load's file can't be in the way of, and handed to the server's data/add +together, timed by their total size; and the server's state reads as JSON whether ``data/get`` answers with data or with a file bundle. A loop of one's own lists, declares and calls the server's tools through ``ToolSession``, and plays a task's agent as an episode, around the echo agent. Requires a docker daemon; spins up a throwaway ``registry:2`` and skips if it can't start. @@ -153,6 +155,34 @@ async def test_a_load_is_timed_by_its_payload_and_the_servers_state_reads_as_jso assert await legacy_protocol.service_state(deployed, None, "items") == items +@pytest.mark.asyncio +async def test_an_environment_of_several_files_loads_each_from_its_path_in_one_add(local_stack, monkeypatch): + staged_sizes, timeout_for = [], constants.data_plane_load_timeout_s + monkeypatch.setattr(constants, "data_plane_load_timeout_s", lambda size: staged_sizes.append(size) or timeout_for(size)) + uid = uuid.uuid4().hex[:8] + env = MCPServerEnv.put(id=f"server-items-{uid}", docker_image_artifact=_put_items_image(f"server-items-{uid}"), + environment_name="items", env_provider_type="server") + deployed = await env.deploy(sandbox_type="local", ttl_seconds=900) + local_stack.append(f"agent-{deployed.sandbox_id}") + contents = {path: json.dumps({"items": items}).encode() + for path, items in (("inbox/a.json", ["in-a", "in-b"]), ("top.json", ["top"]))} + artifact = EnvironmentArtifact.put(id=f"server-items-files-{uid}", environment_name="items", file_artifacts={ + path: FileArtifact.put_bytes(id=f"server-items-file-{index}-{uid}", description="server env e2e", + filename=path.rsplit("/", 1)[-1], content=content, content_type="application/json") + for index, (path, content) in enumerate(contents.items())}) + # A file an earlier load left at /data/inbox, where this load's inbox/ folder would go. + earlier = EnvironmentArtifact.put(id=f"server-items-inbox-{uid}", environment_name="items", file_artifact=FileArtifact.put_bytes( + id=f"server-items-inbox-file-{uid}", description="server env e2e", filename="inbox", + content=json.dumps({"items": ["earlier"]}).encode(), content_type="application/json")) + restored = await Env.from_instance_id(deployed.instance_id) + + await restored.load_environment_artifact(earlier) + await restored.load_environment_artifact(artifact) + + assert await legacy_protocol.service_state(deployed, None, "items") == {"items": ["in-a", "in-b", "top"]} + assert staged_sizes[-1] == sum(map(len, contents.values())) + + @pytest.mark.asyncio async def test_a_deployed_envs_tools_are_listed_and_called_from_a_loop_of_ones_own(local_stack): """A trainer's view: the env's tools listed with their schemas and called, the env's data plane agreeing on the state diff --git a/tst/integration/task_step/test_container_sandbox_files_local.py b/tst/integration/task_step/test_container_sandbox_files_local.py index e3b141a5b..68b84054d 100644 --- a/tst/integration/task_step/test_container_sandbox_files_local.py +++ b/tst/integration/task_step/test_container_sandbox_files_local.py @@ -1,4 +1,5 @@ -"""Files loaded into a container sandbox are found there by verify_sandbox and collect_artifacts, on the local provider. +"""Files loaded into a container sandbox are found there by verify_sandbox and collect_artifacts, on the local provider: +a file universe's, and an agent's environment of several files, each at its path. A local container sandbox, a deploy_sandbox one or an agent's, runs beside the host machine, so each step has to reach into the container rather than act on the host. Every path here is one only the container has. Needs Docker and a @@ -17,7 +18,7 @@ import httpx import pytest -from agent_env.artifact import DockerImageArtifact, FileArtifact, FileArtifactUniverse +from agent_env.artifact import DockerImageArtifact, EnvironmentArtifact, FileArtifact, FileArtifactUniverse from agent_env.artifact.store import reset_artifact_store from agent_env.config import configure, reset_config, set_image_store from agent_env.store.image_store import LocalRegistryImageStore @@ -80,31 +81,38 @@ def local_registry(monkeypatch, tmp_path): reset_config() -def _probe_universe(tmp_path, suffix): - src = tmp_path / "probe.txt" - src.write_text("hello from the probe\n") - return FileArtifactUniverse.put( - id=f"probe-files-{suffix}", - file_artifacts={"probe.txt": FileArtifact.put(id=f"probe-file-{suffix}", description="probe", file_path=str(src))}, - ) +_PROBE = {"probe.txt": "hello from the probe\n"} +_SEVERAL = {**_PROBE, "deeper/second.txt": "hello from deeper\n"} -def _load_verify_collect(target: dict, universe, probe_dir: str): +def _probe_universe(tmp_path, suffix, texts=_PROBE, environment=False): + """A file universe of ``texts`` by path, or an environment artifact of them.""" + files = {} + for index, (path, text) in enumerate(texts.items()): + src = tmp_path / f"probe-{index}.txt" + src.write_text(text) + files[path] = FileArtifact.put(id=f"probe-file-{index}-{suffix}", description="probe", file_path=str(src)) + if environment: + return EnvironmentArtifact.put(f"probe-files-{suffix}", environment_name="probe", file_artifacts=files) + return FileArtifactUniverse.put(id=f"probe-files-{suffix}", file_artifacts=files) + + +def _load_verify_collect(target: dict, universe, probe_dir: str, texts=_PROBE): return [ LoadArtifactTaskStep(id="load", version=None, artifact_id=universe.id, destination_path=probe_dir, **target), VerifySandboxTaskStep( id="verify", version=None, verifier_id="probe", base_dir=probe_dir, **target, criteria=[{"type": "probe_file_contains", "criterion": "loaded", "paths": ["probe.txt"], "expected": "hello"}], ), - CollectArtifactsTaskStep(id="collect", version=None, base_path=probe_dir, artifact_paths=["probe.txt"], **target), + CollectArtifactsTaskStep(id="collect", version=None, base_path=probe_dir, artifact_paths=list(texts), **target), ] -def _assert_found_in_the_container(ctx, probe_dir): +def _assert_found_in_the_container(ctx, probe_dir, texts=_PROBE): assert not Path(probe_dir).exists() assert ctx.metadata["verifications"]["probe"]["score"] == 1.0 universe = FileArtifactUniverse.get(ctx.metadata["collected_artifacts"]["collect"]["file_artifact_universe"]["id"]) - assert universe.get_file_artifacts()["probe.txt"].load() == b"hello from the probe\n" + assert {path: universe.get_file_artifacts()[path].load().decode() for path in texts} == texts assert Task.get_instance(ctx.instance_id).status == "completed" @@ -136,7 +144,9 @@ async def test_a_container_mode_deploy_sandbox(local_registry): @pytest.mark.asyncio -async def test_a_local_agent(local_registry): +@pytest.mark.parametrize("texts, environment", [(_PROBE, False), (_SEVERAL, True)], + ids=["file-universe", "environment-of-several-files"]) +async def test_a_local_agent(local_registry, texts, environment): tmp_path = local_registry suffix = uuid.uuid4().hex[:8] agent = put_test_agent(f"files-local-agent-{suffix}") @@ -146,8 +156,9 @@ async def test_a_local_agent(local_registry): id="deploy", version=None, env_ids=[], a2a_agent_id=agent.id, a2a_agent_version=agent.version, agent_name="solver", sandbox_type="local", ), - *_load_verify_collect({"agent_name": "solver"}, _probe_universe(tmp_path, suffix), probe_dir), + *_load_verify_collect({"agent_name": "solver"}, _probe_universe(tmp_path, suffix, texts, environment), probe_dir, + texts), ]) ctx = await task.run() - _assert_found_in_the_container(ctx, probe_dir) + _assert_found_in_the_container(ctx, probe_dir, texts) diff --git a/tst/unit/agents/hermes/test_hermes_bridge.py b/tst/unit/agents/hermes/test_hermes_bridge.py index d8c335865..4f3cb6ac1 100644 --- a/tst/unit/agents/hermes/test_hermes_bridge.py +++ b/tst/unit/agents/hermes/test_hermes_bridge.py @@ -150,6 +150,6 @@ async def test_an_output_format_is_asked_for_and_checked(monkeypatch, fake_herme def test_the_event_pipe_is_not_left_open_in_the_server(fake_hermes, folders): - before = len(os.listdir("/proc/self/fd")) + before = len(os.listdir("/dev/fd")) asyncio.run(hermes_bridge.run_hermes(prompt="go", context_id="ctx-f", workspace=str(folders))) - assert len(os.listdir("/proc/self/fd")) == before + assert len(os.listdir("/dev/fd")) == before diff --git a/tst/unit/artifact/environment_files_test.py b/tst/unit/artifact/environment_files_test.py new file mode 100644 index 000000000..4678e7292 --- /dev/null +++ b/tst/unit/artifact/environment_files_test.py @@ -0,0 +1,78 @@ +"""An environment artifact holds its environment's files by path. One file whose path is its filename is stored as the one +wrapped file every environment artifact had, document for document; several, or one inside a folder, as +file_artifact_refs. get_file_artifacts reads either shape, get_file_artifact refuses a collection, and a path that isn't +plain file names joined by '/' is refused. One file kept at a path other than its filename can't be written to an +artifact folder, whose one file reads back kept at its filename.""" + +import pytest + +from agent_env.artifact import EnvironmentArtifact, FileArtifact +from agent_env.entity.export import FolderExport, Unwritable + + +def _file(id: str, filename: str) -> FileArtifact: + return FileArtifact.put_bytes(id, description=id, filename=filename, content=id.encode(), content_type="text/plain") + + +def _document(artifact: EnvironmentArtifact) -> dict: + return {key: value for key, value in artifact.model_dump(mode="json", by_alias=True).items() if key != "id"} + + +def test_one_file_at_its_filename_is_stored_as_the_one_wrapped_file(local_stores): + file = _file("seed", "seed.json") + + by_path = EnvironmentArtifact.put("crm", environment_name="crm", file_artifacts={"seed.json": file}) + wrapped = EnvironmentArtifact.put("crm-wrapped", environment_name="crm", file_artifact=file) + + assert _document(by_path) == _document(wrapped) + assert "file_artifact_refs" not in _document(by_path) + assert {path: f.id for path, f in by_path.get_file_artifacts().items()} == {"seed.json": "seed"} + + +@pytest.mark.parametrize("paths", [["a.json", "b.json"], ["inbox/a.json"]], ids=["several", "one-in-a-folder"]) +def test_files_are_kept_by_path_when_one_wrapped_file_cant_say_where_each_goes(local_stores, paths): + files = {path: _file(f"f{index}", path.rsplit("/", 1)[-1]) for index, path in enumerate(paths)} + + stored = EnvironmentArtifact.get(EnvironmentArtifact.put("crm", environment_name="crm", file_artifacts=files).id) + + assert _document(stored) == {"type": "environment", "version": 1, "metadata": {}, "environment_name": "crm", + "service_name": "crm", "file_artifact_refs": { + path: {"id": file.id, "version": 1} for path, file in files.items()}} + assert {path: file.load() for path, file in stored.get_file_artifacts().items()} == { + path: file.load() for path, file in files.items()} + with pytest.raises(ValueError, match=f"holds {len(paths)} files by path, not one; read them with get_file_artifacts"): + stored.get_file_artifact() + + +def test_a_legacy_unpinned_document_reads_as_its_one_file(local_stores): + _file("seed", "seed.json") + + legacy = EnvironmentArtifact(id="crm", version=1, environment_name="crm", file_artifact_id="seed") + + assert {path: f.id for path, f in legacy.get_file_artifacts().items()} == {"seed.json": "seed"} + + +@pytest.mark.parametrize("path", ["/a.json", "../a.json", "a//b.json", "a/./b.json", "a\\b.json", ""]) +def test_a_path_that_isnt_plain_file_names_is_refused(local_stores, path): + files = {path: _file("f", "a.json"), "b.json": _file("g", "b.json")} + + with pytest.raises(ValueError, match="must be plain file names joined by '/'"): + EnvironmentArtifact.put("crm", environment_name="crm", file_artifacts=files) + + +@pytest.mark.parametrize("given, refused", [ + (lambda file: {"file_artifacts": {}}, "file_artifacts can't be empty"), + (lambda file: {"file_artifact": file, "file_artifacts": {"seed.json": file}}, "holds a file_artifact or file_artifacts"), +], ids=["no-files", "a-file-and-files"]) +def test_an_environment_holds_a_file_or_files(local_stores, given, refused): + with pytest.raises(ValueError, match=refused): + EnvironmentArtifact.put("crm", environment_name="crm", **given(_file("seed", "seed.json"))) + + +@pytest.mark.parametrize("path", ["inbox/a.json", "notes.json"], ids=["inside-a-folder", "under-another-name"]) +def test_one_file_kept_at_a_path_other_than_its_filename_cant_be_written_to_an_artifact_folder(local_stores, tmp_path, path): + environment = EnvironmentArtifact.put("crm", environment_name="crm", file_artifacts={path: _file("f", "a.json")}) + + with pytest.raises(Unwritable, match=f"its one file is kept at '{path}', not at its filename, and a folder's one " + "file reads back kept at its filename"): + environment.to_toml(FolderExport(tmp_path)) diff --git a/tst/unit/artifact/test_service_universe_file_view.py b/tst/unit/artifact/test_service_universe_file_view.py index 6883da8a9..c19be1f29 100644 --- a/tst/unit/artifact/test_service_universe_file_view.py +++ b/tst/unit/artifact/test_service_universe_file_view.py @@ -2,8 +2,10 @@ service universe be staged as a file tree (grading) by the same loaders that handle a FileArtifactUniverse. -Contract: keys are `/`, values are the -universe's EXISTING FileArtifacts (nothing copied or re-registered). +Contract: keys are `/`, or `/` +for each file of an environment holding several; values are the universe's +EXISTING FileArtifacts (nothing copied or re-registered). A universe holding an +environment of several files can't be written to a universe folder. """ from __future__ import annotations @@ -14,6 +16,7 @@ from agent_env.artifact.artifacts.environment import EnvironmentArtifact from agent_env.artifact.artifacts.environment_universe import EnvironmentUniverseArtifact from agent_env.artifact.ref import ArtifactRef +from agent_env.entity.export import FolderExport, Unwritable def _file_artifact(id: str, filename: str) -> FileArtifact: @@ -91,3 +94,17 @@ def test_duplicate_service_names_raise(registry): universe = _universe(registry, [("slack", "a.json"), ("slack", "a.json")]) with pytest.raises(ValueError, match="duplicate environment 'slack'"): universe.get_file_artifacts() + + +def test_an_environment_of_several_files_lists_each_at_its_path_and_cant_be_written_to_a_universe_folder(registry, tmp_path): + files, services = registry + a, b = _file_artifact("fa-a", "a.json"), _file_artifact("fa-b", "b.json") + files.update({a.id: a, b.id: b}) + services["mail"] = EnvironmentArtifact(id="mail", version=1, environment_name="mail", + file_artifact_refs={"inbox/a.json": a.as_ref(), "b.json": b.as_ref()}) + universe = EnvironmentUniverseArtifact(id="u", version=1, service_artifact_refs=[ArtifactRef(id="mail", version=1)]) + + assert universe.get_file_artifacts() == {"mail/inbox/a.json": a, "mail/b.json": b} + with pytest.raises(Unwritable, match="its environment 'mail' holds several files, and a universe's folder holds one " + "per environment; write environment artifact mail to a folder of its own"): + universe.to_toml(FolderExport(tmp_path)) diff --git a/tst/unit/bundle/artifact_toml_test.py b/tst/unit/bundle/artifact_toml_test.py index 054d1ee3f..76bf95196 100644 --- a/tst/unit/bundle/artifact_toml_test.py +++ b/tst/unit/bundle/artifact_toml_test.py @@ -1,5 +1,5 @@ """Environment and environment_universe artifacts written from a bundle's artifact folders: an environment over the -file its folder holds or the file artifact it names, and a universe laid out as ``environment-universe get +file its folder holds, even inside a subfolder, its files by path when it holds several, or the file artifact it names, and a universe laid out as ``environment-universe get --output-dir`` writes one, adding the environments it names. Each is checked before any write, rewritten when what it records changes, and equal, field for field, to one the CLI writes from the same files. Each environment the bundle writes must seed a server of the env a task loads it into.""" @@ -127,6 +127,36 @@ def test_a_description_names_the_folders_file(bundle_dir): assert EnvironmentArtifact.get(f"{ROOT}/crm-data").get_file_artifact().description == "crm rows" +def test_an_environment_keeps_the_several_files_its_folder_holds_by_path_and_is_rewritten_when_one_changes(bundle_dir): + files = {"inbox/a.json": "{}\n", "sent.json": "[]\n"} + _bundle(bundle_dir, {**_environment("mail", "mail"), **{f"artifacts/mail/{path}": text for path, text in files.items()}}, + "mail") + + first = _summary(_run(bundle_dir)) + kept = EnvironmentArtifact.get(f"{ROOT}/mail").get_file_artifacts() + + assert first[f"{ROOT}/mail"] == (1, False, ("new",)) + assert {path: (file.filename, file.load().decode()) for path, file in kept.items()} == { + path: (path.rsplit("/", 1)[-1], text) for path, text in files.items()} + assert {path: file.id for path, file in kept.items()} == { + path: f"{ROOT}/mail__{hashlib.sha256(path.encode()).hexdigest()[:16]}" for path in files} + + assert all(reused for _, reused, _ in _summary(_run(bundle_dir)).values()) + (bundle_dir / "artifacts/mail/sent.json").write_text('["x"]\n') + assert _summary(_run(bundle_dir))[f"{ROOT}/mail"] == (2, False, ("files changed: sent.json",)) + assert EnvironmentArtifact.get(f"{ROOT}/mail").get_file_artifacts()["sent.json"].load() == b'["x"]\n' + + +def test_a_folders_one_file_inside_a_subfolder_is_still_the_one_file_it_wraps(bundle_dir): + _bundle(bundle_dir, {**_environment("mail", "mail"), "artifacts/mail/inbox/a.json": "{}\n"}, "mail") + + _run(bundle_dir) + environment = EnvironmentArtifact.get(f"{ROOT}/mail") + + assert environment.file_artifact_refs is None + assert (environment.get_file_artifact().id, environment.get_file_artifact().filename) == (f"{ROOT}/mail__file", "a.json") + + def test_an_environment_over_a_bundle_file_is_rewritten_when_the_file_is(bundle_dir): _bundle(bundle_dir, {"artifacts/crm-file/data.json": "{}\n", **_environment("crm-data", "crm", file='"crm-file"')}, "crm-data") @@ -414,12 +444,12 @@ def _refused(files): @pytest.mark.parametrize("files, problem", [ - (_environment("x", "crm") | {"artifacts/x/a.json": "{}", "artifacts/x/b.json": "{}"}, - "artifacts/x: an environment artifact holds one file, and this folder has 2 ('a.json', 'b.json'); name the " - "artifact to wrap with file, or give each other file an artifact folder of its own"), + (_environment("x", "crm", description='"d"') | {"artifacts/x/a.json": "{}", "artifacts/x/b.json": "{}"}, + "artifacts/x: artifact.toml: description describes the folder's one file, and it holds 'a.json', 'b.json'; drop " + "description"), (_environment("x", "crm"), "artifacts/x: has no file for the environment to wrap (.DS_Store, Thumbs.db, __pycache__, desktop.ini don't " - "count); put its one file here, or name the artifact to wrap with file"), + "count); put its files here, or name the artifact to wrap with file"), ({f"artifacts/x/{TOML}": 'type = "environment"\n', "artifacts/x/a.json": "{}"}, "artifacts/x: artifact.toml: environment_name is required: the environment the file seeds"), (_environment("x", "") | {"artifacts/x/a.json": "{}"}, @@ -433,7 +463,7 @@ def _refused(files): "drop one"), (_environment("x", "crm", file='"f"') | {"artifacts/f/a.json": "{}", "artifacts/x/stray.json": "{}"}, "artifacts/x: its file is the artifact file names, so the folder holds only artifact.toml, and it also has " - "'stray.json'; give the file an artifact folder of its own, or leave out file to wrap the folder's one file"), + "'stray.json'; give the file an artifact folder of its own, or leave out file to keep the folder's files"), (_environment("x", "crm", file='"many"') | {"artifacts/many/a.json": "{}", "artifacts/many/b.json": "{}"}, "artifacts/x: file: 'many' is this bundle's file_artifact_universe, but this field takes file"), (_universe("x"), @@ -458,7 +488,7 @@ def _refused(files): (_universe("x", service_artifact_refs='["a"]', metadata='{ m = "f" }') | {"artifacts/x/gmail/a.json": "{}"}, "artifacts/x: artifact.toml: service_artifact_refs is what the stored document calls it; write " "environment_artifacts instead"), -], ids=["environment-two-files", "environment-no-file", "environment-no-name", "environment-empty-name", +], ids=["environment-description-of-two-files", "environment-no-file", "environment-no-name", "environment-empty-name", "environment-service-name", "environment-file-artifact-id", "environment-file-and-description", "environment-file-and-a-file", "environment-over-a-universe", "universe-empty", "universe-root-file", "universe-two-files", "universe-flat-metadata", "universe-metadata-in-another-case", "universe-repeated-name", diff --git a/tst/unit/bundle/to_toml_round_trip_test.py b/tst/unit/bundle/to_toml_round_trip_test.py index bd47bb90c..b48ed45ea 100644 --- a/tst/unit/bundle/to_toml_round_trip_test.py +++ b/tst/unit/bundle/to_toml_round_trip_test.py @@ -163,6 +163,19 @@ def test_an_environment_reads_back_wrapping_its_file(home): assert (seeded.filename, seeded.description, _bytes(seeded)) == ("seed.json", "the seed", b'{"rows": 1}') +def test_an_environment_of_several_files_reads_back_with_each_file_at_its_path(home): + _env(home, "crm") + texts = {"rows.json": "rows", "deep/er/notes.txt": "notes"} + files = {path: FileArtifact.put_bytes(f"crm-{index}", description=path, filename=path.rsplit("/", 1)[-1], + content=text.encode(), content_type="text/plain") + for index, (path, text) in enumerate(texts.items())} + EnvironmentArtifact.put("crm-seed", environment_name="crm", file_artifacts=files) + + with _read_back(home, [{**LOAD, "artifacts": [{"id": "crm-seed"}]}], "artifact", "crm-seed") as (read, _): + assert (type(read), read.environment_name) == (EnvironmentArtifact, "crm") + assert {path: _bytes(file).decode() for path, file in read.get_file_artifacts().items()} == texts + + def test_an_environment_universe_reads_back_with_its_environments_and_metadata(home): _env(home, "crm") environments = [] diff --git a/tst/unit/env/envs/test_artifact_staging.py b/tst/unit/env/envs/test_artifact_staging.py index 84dbf61b8..98738aa01 100644 --- a/tst/unit/env/envs/test_artifact_staging.py +++ b/tst/unit/env/envs/test_artifact_staging.py @@ -1,6 +1,8 @@ """An env stages a loaded artifact's file through its sandbox's shell: a file or env name with shell syntax in it is -staged and copied as it is and runs nothing, and a file name that isn't one plain name is refused before anything runs. -A real LocalSandbox runs the scripts; docker is a stub that logs its arguments.""" +staged and copied as it is and runs nothing, a file inside folders is staged under them, and a path that isn't plain +file names joined by '/' is refused before anything runs. A server whose VM sandbox is its container stages and removes +a load's folder through docker, never on the host. A real LocalSandbox runs the scripts; docker is a stub that logs its +arguments.""" from __future__ import annotations @@ -15,6 +17,7 @@ from agent_env.env.envs.mcp_server import MCPServerEnv from agent_env.env.envs.website import WebsiteEnv from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandbox +from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER _SHELL_NAMES = [ ("email", "x;touch pwned;#.json"), @@ -24,7 +27,8 @@ ("email", "seed data.json"), ("e;touch pwned;#", "seed.json"), ] -_NOT_ONE_NAME = ["../seed.json", "a/b.json", "a\\b.json", "..", ".", "", "seed\n.json", "seed\x00.json"] +_NOT_A_PATH = ["../seed.json", "/seed.json", "a//b.json", "a/./b.json", "a\\b.json", "..", ".", "", "seed\n.json", + "seed\x00.json"] @pytest.fixture @@ -55,7 +59,7 @@ def docker_calls() -> list[list[str]]: async def test_an_mcp_server_stages_a_shell_looking_name_as_it_is(host, environment_name, filename): env = _mcp(host, environment_name) - assert await env._copy_artifact_into_container(_artifact(filename)) == f"/data/{filename}" + assert await env._copy_artifact_into_container(_artifact(filename), filename) == f"/data/{filename}" exec_mkdir, cp = host.docker_calls() assert exec_mkdir == ["exec", "cid", "mkdir", "-p", "/data"] @@ -71,7 +75,7 @@ async def test_an_mcp_server_stages_a_shell_looking_name_as_it_is(host, environm async def test_a_website_stages_a_shell_looking_name_as_it_is(host, environment_name, filename): env = _website(host, environment_name) - assert await env._copy_artifact_into_container(_artifact(filename)) == f"/tmp/data/{filename}" + assert await env._copy_artifact_into_container(_artifact(filename), filename) == f"/tmp/data/{filename}" exec_mkdir, cp = host.docker_calls() service = f"{environment_name}-website-backend" @@ -82,12 +86,36 @@ async def test_a_website_stages_a_shell_looking_name_as_it_is(host, environment_ assert cp[4].endswith(filename) and not Path(cp[4]).exists() +@pytest.mark.asyncio +@pytest.mark.parametrize(("make_env", "data"), [(lambda h: _mcp(h, "email"), "/data"), + (lambda h: _website(h, "email"), "/tmp/data")], ids=["mcp", "website"]) +async def test_a_file_inside_folders_is_staged_under_them(host, make_env, data): + assert await make_env(host)._copy_artifact_into_container(_artifact("b.json"), "a/b.json") == f"{data}/a/b.json" + + exec_mkdir, cp = host.docker_calls() + assert exec_mkdir[-3:] == ["mkdir", "-p", f"{data}/a"] and cp[-1].endswith(f":{data}/a/b.json") + + +@pytest.mark.asyncio +async def test_a_server_whose_vm_sandbox_is_its_container_stages_and_removes_a_loads_folder_through_docker(host): + env = _mcp(host, "email") + host.sandbox.mode = SANDBOX_MODE_CONTAINER + server = host.sandbox.container_name + + await env._copy_artifact_into_container(_artifact("b.json"), ".agentenv-load-1/a/b.json") + await env._remove_from_container(".agentenv-load-1") + + calls = host.docker_calls() + assert ["exec", "-u", "0", server, "mkdir", "-p", "/data/.agentenv-load-1/a"] in calls + assert calls[-1] == ["exec", "-u", "0", server, "rm", "-rf", "/data/.agentenv-load-1"] + + @pytest.mark.asyncio @pytest.mark.parametrize("make_env", [lambda h: _mcp(h, "email"), lambda h: _website(h, "email")], ids=["mcp", "website"]) -@pytest.mark.parametrize("filename", _NOT_ONE_NAME) -async def test_a_name_that_is_not_one_plain_file_name_is_refused_before_anything_runs(host, make_env, filename): - with pytest.raises(ValueError, match="must be a single file name"): - await make_env(host)._copy_artifact_into_container(_artifact(filename)) +@pytest.mark.parametrize("path", _NOT_A_PATH) +async def test_a_path_that_is_not_plain_file_names_is_refused_before_anything_runs(host, make_env, path): + with pytest.raises(ValueError, match="must be plain file names joined by '/'"): + await make_env(host)._copy_artifact_into_container(_artifact("seed.json"), path) assert host.docker_calls() == [] and list(host.sandbox.work_dir.iterdir()) == [] diff --git a/tst/unit/env/envs/test_env_card_loads.py b/tst/unit/env/envs/test_env_card_loads.py index fc33f8dff..4e2b4301a 100644 --- a/tst/unit/env/envs/test_env_card_loads.py +++ b/tst/unit/env/envs/test_env_card_loads.py @@ -1,7 +1,9 @@ """Loads find each child env's data plane in the stored env card. A child env the card lists is loaded through its v1 data plane at the card's own `url`, with no live -card read. One the card doesn't list, a server that serves no card, takes the legacy REST path. Without +card read. One the card doesn't list, a server that serves no card, takes the legacy REST path, which takes one file +and refuses an environment of several before sending anything; a website stages several in a folder of the load's own, +which it removes after, and sends each with its path. Without a stored card (no deployed record, or a record with none), the live probe decides. Every env keeps the record it deploys or reattaches to, and MultiEnv hands it to every child env except a name an MCP server and a website share. @@ -15,6 +17,7 @@ import httpx import pytest +from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI from agent_env.env.env import DeployedEnv, DeployedGatewayEnv, DeployedSandboxEnv from agent_env.env.envs.mcp_server import MCPServerEnv @@ -63,6 +66,16 @@ async def test_a_child_env_missing_from_the_card_takes_the_legacy_path(sent): assert json.loads(request.content) == {"mock_data_path": "/data/email.json"} +@pytest.mark.asyncio +async def test_the_legacy_path_refuses_an_environment_of_several_files_before_sending_anything(sent): + several = SimpleNamespace(environment_name="email", id="mail-seed", file_artifact_refs={"a.json": None, "b.json": None}) + + with pytest.raises(RuntimeError, match="'email' takes one file through its pre-v1 load, and environment artifact " + "'mail-seed' holds 2"): + await _mcp(_record(await _card("mcp-slack"))).load_environment_artifact(several) + assert sent == [] + + @pytest.mark.asyncio @pytest.mark.parametrize("has_record", [False, True], ids=["no-record", "no-stored-card"]) async def test_an_env_without_a_stored_card_keeps_the_live_probe(sent, has_record): @@ -87,6 +100,27 @@ async def test_a_website_child_env_loads_at_its_own_card_url(sent): assert [(r.method, str(r.url)) for r in sent] == [("POST", f"{_CARD}/svc/shop/agentenv")] * 2 +@pytest.mark.asyncio +async def test_a_website_stages_several_files_in_a_folder_of_the_loads_own_sends_each_with_its_path_and_removes_it(sent): + web = _website() + web._sandbox, web._gateway_url, web._deployed = MagicMock(exec_script=AsyncMock()), _GW, _record(await _card("shop")) + web._copy_artifact_into_container = AsyncMock(side_effect=lambda file, path: f"/tmp/data/{path}") + web._env_provider = MagicMock(spec=EnvironmentGatewayProvider, install_changelog_triggers=AsyncMock()) + files = {path: SimpleNamespace(content_type="application/json", filename=path.rsplit("/", 1)[-1]) + for path in ("orders/1.json", "cart.json")} + + await web.load_environment_artifact(SimpleNamespace(environment_name="shop", file_artifact_refs=dict.fromkeys(files), + get_file_artifacts=lambda: files)) + + staged = [c.args[1] for c in web._copy_artifact_into_container.await_args_list] + folder = staged[0].removesuffix("/orders/1.json") + assert folder.startswith(".agentenv-load-") and staged == [f"{folder}/{path}" for path in files] + parts = json.loads(sent[1].content)["params"]["parts"] + assert [(p["file"]["uri"], p["metadata"][DATA_OBJECTS_EXTENSION_URI]["path"]) for p in parts] == [ + (f"file:///tmp/data/{folder}/{path}", path) for path in files] + assert web._sandbox.exec_script.await_args.args[0].endswith(f"rm -rf /tmp/data/{folder}") + + @pytest.mark.asyncio @pytest.mark.parametrize("make, module", [(lambda: _mcp_env(), "mcp_server"), (lambda: _website(), "website")], ids=["mcp", "website"]) async def test_a_standalone_env_keeps_the_record_it_deploys(make, module): @@ -220,7 +254,8 @@ def _record(card: dict | None) -> DeployedEnv: def _artifact(name: str) -> SimpleNamespace: file = SimpleNamespace(content_type="application/json", filename=f"{name}.json") - return SimpleNamespace(environment_name=name, get_file_artifact=lambda: file) + return SimpleNamespace(environment_name=name, file_artifact_refs=None, get_file_artifact=lambda: file, + get_file_artifacts=lambda: {file.filename: file}) def _mcp_env(name: str = "email") -> MCPServerEnv: diff --git a/tst/unit/env/envs/test_mcp_server_container_mode.py b/tst/unit/env/envs/test_mcp_server_container_mode.py index 4e2b7574d..b89ff3b37 100644 --- a/tst/unit/env/envs/test_mcp_server_container_mode.py +++ b/tst/unit/env/envs/test_mcp_server_container_mode.py @@ -1,7 +1,9 @@ """A standalone MCPServerEnv on Modal containers stages into its server's own container and records every container it created, so a later process can reattach them: loads stage into the server again, and close() and the reapers reach them all. A dead servicedb or sidecar is skipped on restore; a dead server fails it. A VM deploy's record is unchanged. -A load is timed by the size its object store holds for the payload it stages. +A load is timed by the size its object store holds for the payload it stages. An environment of several files has each +staged at its path in a folder of the load's own, removed after it however it ends, and sent with its path, the load timed +by their total size; one file's name is still one plain name. Deployed without a gateway, the server is the env: its own card and container are the record's, and loads go straight to it. Only the sandboxes, the HTTP boundary and each topology's deploy path are faked.""" @@ -9,13 +11,15 @@ import dataclasses import inspect +import json +import posixpath import re from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest -from agentenv_protocol import WELL_KNOWN_PATH +from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI, WELL_KNOWN_PATH from agent_env.artifact import Artifact from agent_env.env.env import DeployedEnv, DeployedSandboxEnv @@ -81,6 +85,56 @@ async def test_a_load_is_timed_by_the_size_its_store_holds_for_the_payload(sent, assert data_plane_load_timeout_s(size) > data_plane_load_timeout_s(None) +@pytest.mark.asyncio +async def test_an_environment_of_several_files_stages_each_at_its_path_and_sends_it_with_the_path(sent, monkeypatch): + env, sandboxes = _env(), _containers() + store = _store_reporting(ObjectMetadata(size=1_000)) + monkeypatch.setattr(mcp_server, "get_config", lambda: MagicMock(get_object_store_at=MagicMock(return_value=store))) + + await _deploy(env, sandboxes) + await env.load_environment_artifact(_artifact("inbox/a.json", "sent.json")) + + server = sandboxes["srv"] + writes = [c.args for c in server.write_file_from_object.await_args_list] + folder = writes[0][1].removesuffix("/inbox/a.json") + assert re.fullmatch(r"/data/\.agentenv-load-[0-9a-f]{8}", folder) + assert writes == [("s3://bucket/inbox/a.json", f"{folder}/inbox/a.json"), ("s3://bucket/sent.json", f"{folder}/sent.json")] + assert [c.args for c in server.exec_with_output.await_args_list] == [ + ("mkdir", "-p", f"{folder}/inbox"), ("mkdir", "-p", folder), ("rm", "-rf", folder)] + assert [json.loads(r.content)["method"] for r in sent] == ["data/reset", "data/add"] + assert json.loads(sent[1].content)["params"]["parts"] == [ + {"kind": "file", "file": {"uri": f"file://{folder}/{path}", "mimeType": "application/json", "name": name}, + "metadata": {DATA_OBJECTS_EXTENSION_URI: {"path": path}}} + for path, name in (("inbox/a.json", "a.json"), ("sent.json", "sent.json"))] + assert {r.extensions["timeout"]["read"] for r in sent} == {data_plane_load_timeout_s(2_000)} + + +@pytest.mark.asyncio +async def test_a_load_of_several_files_that_fails_still_removes_its_folder(sent): + env, sandboxes = _env(), _containers() + await _deploy(env, sandboxes) + server = sandboxes["srv"] + server.write_file_from_object.side_effect = [None, RuntimeError("store unreachable")] + + with pytest.raises(RuntimeError, match="store unreachable"): + await env.load_environment_artifact(_artifact("inbox/a.json", "sent.json")) + + assert sent == [] + assert server.exec_with_output.await_args_list[-1].args[:2] == ("rm", "-rf") + + +@pytest.mark.asyncio +async def test_one_files_name_is_still_one_plain_name(sent): + env, sandboxes = _env(), _containers() + await _deploy(env, sandboxes) + file = SimpleNamespace(filename="a/b.json", object_url="s3://bucket/b.json", content_type="application/json") + artifact = SimpleNamespace(environment_name="email", file_artifact_refs=None, get_file_artifacts=lambda: {file.filename: file}) + + with pytest.raises(ValueError, match="must be a single file name"): + await env.load_environment_artifact(artifact) + assert sent == [] and not sandboxes["srv"].write_file_from_object.await_count + + @pytest.mark.asyncio @pytest.mark.parametrize("lookup", [None, RuntimeError("store unreachable")], ids=["no-metadata", "lookup-fails"]) async def test_a_store_that_cannot_size_the_payload_leaves_the_load_its_floor_timeout(monkeypatch, lookup): @@ -313,7 +367,8 @@ def _env(env_provider_type: str = "gateway") -> MCPServerEnv: def _sandbox(sandbox_id: str, mode: str = SANDBOX_MODE_CONTAINER, sandbox_type: str = "modal") -> MagicMock: - return MagicMock(sandbox_id=sandbox_id, mode=mode, type=sandbox_type, terminate=AsyncMock(), write_file_from_object=AsyncMock()) + return MagicMock(sandbox_id=sandbox_id, mode=mode, type=sandbox_type, terminate=AsyncMock(), write_file_from_object=AsyncMock(), + exec_with_output=AsyncMock(return_value=(0, "", ""))) def _containers() -> dict[str, MagicMock]: @@ -371,6 +426,9 @@ def get_sandbox(sandbox_id): return await MCPServerEnv.from_deployed_env(record) -def _artifact(): - return MagicMock(environment_name="email", **{"get_file_artifact.return_value": SimpleNamespace( - filename="email.json", object_url="s3://bucket/email.json", content_type="application/json")}) +def _artifact(*paths: str): + """The email environment's artifact: its one file, email.json, or a file at each of ``paths``.""" + files = {path: SimpleNamespace(filename=posixpath.basename(path), object_url=f"s3://bucket/{path}", + content_type="application/json") for path in paths or ("email.json",)} + return SimpleNamespace(environment_name="email", file_artifact_refs=dict.fromkeys(paths) if paths else None, + get_file_artifact=lambda: files["email.json"], get_file_artifacts=lambda: files) diff --git a/tst/unit/env/envs/test_mcp_server_plugin_provider.py b/tst/unit/env/envs/test_mcp_server_plugin_provider.py index 02037b78c..b37528070 100644 --- a/tst/unit/env/envs/test_mcp_server_plugin_provider.py +++ b/tst/unit/env/envs/test_mcp_server_plugin_provider.py @@ -1,5 +1,5 @@ """A stock MCPServerEnv deploys through a plugin's environment provider: the provider gets every option, its record is -registered as it returns it, and the env holds no sandbox of ours. Loads hand the server a signed URL; reattaching a +registered as it returns it, and the env holds no sandbox of ours. Loads hand the server a signed URL for each file; reattaching a plugin's record touches no sandbox and looks up no provider; close() closes the provider that deployed. A MultiEnv child is still the gateway's, whatever type it declares.""" @@ -13,6 +13,7 @@ import httpx import pytest +from agentenv_protocol import DATA_OBJECTS_EXTENSION_URI from agent_env.config import get_config, reset_config from agent_env.env.env import DeployedEnv, DeployedSandboxEnv @@ -92,9 +93,12 @@ async def _deploy(env, **kwargs): return await env.deploy(**kwargs) -def _artifact(): - return SimpleNamespace(environment_name="email", get_file_artifact=lambda: SimpleNamespace( - filename="email.json", object_url="s3://bucket/email.json", content_type="application/json")) +def _artifact(*paths: str): + """The email environment's artifact: its one file, email.json, or a file at each of ``paths``.""" + files = {path: SimpleNamespace(filename=path.rsplit("/", 1)[-1], object_url=f"s3://bucket/{path}", + content_type="application/json") for path in paths or ("email.json",)} + return SimpleNamespace(environment_name="email", file_artifact_refs=dict.fromkeys(paths) if paths else None, + get_file_artifact=lambda: files["email.json"], get_file_artifacts=lambda: files) def test_reading_a_stored_env_looks_up_no_environment_provider(monkeypatch): @@ -151,6 +155,22 @@ async def test_a_plugin_env_loads_by_a_signed_url_with_no_staging(sent): assert json.loads(sent[1].content)["params"]["parts"][0]["file"] == {"uri": "https://signed.example/email.json?sig=1", "mimeType": "application/json", "name": "email.json"} +@pytest.mark.asyncio +async def test_a_plugin_env_loads_each_of_several_files_by_its_own_signed_url_with_its_path(sent): + env = _env() + await _deploy(env) + store = _store(None) + store.signed_get_url.side_effect = lambda url, ttl: url.replace("s3://bucket/", "https://signed.example/") + with _stores(store): + await env.load_environment_artifact(_artifact("inbox/a.json", "sent.json")) + + ttl = len(LOAD_OPERATIONS) * data_plane_load_timeout_s(2 * _SIZE) + assert [c.args for c in store.signed_get_url.call_args_list] == [("s3://bucket/inbox/a.json", ttl), ("s3://bucket/sent.json", ttl)] + assert json.loads(sent[1].content)["params"]["parts"] == [ + {"kind": "file", "file": {"uri": f"https://signed.example/{path}", "mimeType": "application/json", "name": path.rsplit("/", 1)[-1]}, + "metadata": {DATA_OBJECTS_EXTENSION_URI: {"path": path}}} for path in ("inbox/a.json", "sent.json")] + + @pytest.mark.asyncio async def test_a_plugin_env_refuses_a_load_the_store_cannot_sign_before_touching_its_data(sent): env = _env() diff --git a/tst/unit/env/envs/test_multi_env_plugin_provider.py b/tst/unit/env/envs/test_multi_env_plugin_provider.py index 5d4020da6..7e34b524f 100644 --- a/tst/unit/env/envs/test_multi_env_plugin_provider.py +++ b/tst/unit/env/envs/test_multi_env_plugin_provider.py @@ -135,8 +135,9 @@ async def _deploy(env, **kwargs): def _artifact(name: str): - return SimpleNamespace(environment_name=name, get_file_artifact=lambda: SimpleNamespace( - filename=f"{name}.json", object_url=f"s3://bucket/{name}.json", content_type="application/json")) + file = SimpleNamespace(filename=f"{name}.json", object_url=f"s3://bucket/{name}.json", content_type="application/json") + return SimpleNamespace(environment_name=name, file_artifact_refs=None, get_file_artifact=lambda: file, + get_file_artifacts=lambda: {file.filename: file}) def _universe(*names: str, metadata=None): diff --git a/tst/unit/task_step/test_load_artifact_agent_container_target.py b/tst/unit/task_step/test_load_artifact_agent_container_target.py index 153af421a..59c4ad02d 100644 --- a/tst/unit/task_step/test_load_artifact_agent_container_target.py +++ b/tst/unit/task_step/test_load_artifact_agent_container_target.py @@ -1,4 +1,5 @@ -"""`container_name` is VmSandbox-only, so `LoadArtifactTaskStep(agent_name=...)` must resolve the staging container by subtype, not by `sandbox.mode`.""" +"""`container_name` is VmSandbox-only, so `LoadArtifactTaskStep(agent_name=...)` must resolve the staging container by subtype, not by `sandbox.mode`. +An environment of several files is staged file by file at their paths, by the same subtype.""" from __future__ import annotations @@ -7,6 +8,8 @@ from agent_env.artifact.artifact import Artifact from agent_env.artifact.artifacts.file import FileArtifact from agent_env.artifact.artifacts.environment import EnvironmentArtifact +from agent_env.artifact.ref import ArtifactRef +from agent_env.providers.sandbox_providers import sandbox as sandbox_mod from agent_env.providers.sandbox_providers.sandbox import Sandbox, VmSandbox from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER, SANDBOX_MODE_VM from agent_env.task_step.context import DeployedAgent, TaskStepContext @@ -164,3 +167,29 @@ async def terminate(self) -> None: # pragma: no cover - never called here assert staged[0]["container"] == expected assert staged[0]["destination"] == "/app/files" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("sandbox, container, on_host", [ + (_ModalLikeSandbox(), None, False), + (_LocalLikeSandbox(), "agent-local-1", False), + (_LocalLikeSandbox(), None, True), +], ids=["container-sandbox", "vm-container", "vm-host"]) +async def test_an_environment_of_several_files_is_staged_file_by_file_at_their_paths(monkeypatch, sandbox, container, + on_host): + calls = [] + + async def stage(*args, **kwargs): + calls.append((args, kwargs)) + return {"inbox/a.json": "/app/files/inbox/a.json", "sent.json": "/app/files/sent.json"} + + monkeypatch.setattr(sandbox_mod, "stage_files_into_container", stage) + files = {"inbox/a.json": object(), "sent.json": object()} + monkeypatch.setattr(EnvironmentArtifact, "get_file_artifacts", lambda self: files) + mail = EnvironmentArtifact(id="mail", version=1, environment_name="mail", file_artifact_refs={ + "inbox/a.json": ArtifactRef(id="fa-a", version=1), "sent.json": ArtifactRef(id="fa-b", version=1)}) + + staged = await mod._stage_environment_payload_into_container(sandbox, container, mail, "/app/files") + + assert staged == ["inbox/a.json", "sent.json"] + assert calls == [((sandbox, files, "/app/files"), {"container": container, "on_host": on_host})] diff --git a/tst/unit/task_step/test_verify_universe_agent_judge.py b/tst/unit/task_step/test_verify_universe_agent_judge.py index 7bee70feb..d0a48e7bd 100644 --- a/tst/unit/task_step/test_verify_universe_agent_judge.py +++ b/tst/unit/task_step/test_verify_universe_agent_judge.py @@ -2,8 +2,8 @@ Covers the pure prompt/rubric builders and the FileArtifactUniverse-emit helper on the round-trip step. The helper test mocks the artifact store so it runs with no external services, and asserts -the transport is **faithful** (original/export1 reference the existing FileArtifacts verbatim; -export2 is dumped losslessly, unicode preserved). +the transport is **faithful** (original/export1 reference the existing FileArtifacts verbatim, a +service of several files each at its path; export2 is dumped losslessly, unicode preserved). """ import json @@ -55,6 +55,10 @@ def test_user_prompt_points_at_universes_dir(self): assert UNIVERSES_DIR in build_user_prompt() assert "original" in build_user_prompt() and "export_1" in build_user_prompt() + def test_both_prompts_name_the_folder_a_service_of_several_files_has(self): + assert f"{UNIVERSES_DIR}/original//" in JUDGE_SYSTEM_PROMPT + assert "`original//`" in build_user_prompt() + class TestFauId: def test_deterministic(self): @@ -167,32 +171,49 @@ def __init__(self, id): class _FakeServiceArtifact: - def __init__(self, environment_name): + def __init__(self, environment_name, paths=()): self.environment_name = environment_name self._fa = _FakeFileArtifact(f"fa-{environment_name}") + self._files = ({path: _FakeFileArtifact(f"fa-{environment_name}-{path}") for path in paths} if paths + else {f"{environment_name}.json": self._fa}) def get_file_artifact(self): return self._fa + def get_file_artifacts(self): + return self._files -class TestCreateFileArtifactUniverse: - def test_bundles_three_dirs_faithfully(self, monkeypatch): - # Capture what the helper hands to the store, plus the bytes it writes for export2. - put_calls = {} - dumped = {} - def fake_fa_put(id, description, file_path): - with open(file_path, encoding="utf-8") as f: - dumped[id] = f.read() - return _FakeFileArtifact(id) +class _FakeExportedUniverse: + def __init__(self, *names): + self.names = names + + def get_environment_artifacts(self): + return [_FakeServiceArtifact(name) for name in self.names] + - def fake_fau_put(id, *, file_artifacts): - put_calls["id"] = id - put_calls["file_artifacts"] = file_artifacts - return _FakeFileArtifact(id) +def _capture_puts(monkeypatch) -> tuple[dict, dict]: + """What the helper hands the store for the bundle, and the text it writes for each export2 file, by id.""" + put_calls, dumped = {}, {} - monkeypatch.setattr(FileArtifact, "put", staticmethod(fake_fa_put)) - monkeypatch.setattr(FileArtifactUniverse, "put", staticmethod(fake_fau_put)) + def fake_fa_put(id, description, file_path): + with open(file_path, encoding="utf-8") as f: + dumped[id] = f.read() + return _FakeFileArtifact(id) + + def fake_fau_put(id, *, file_artifacts): + put_calls["id"] = id + put_calls["file_artifacts"] = file_artifacts + return _FakeFileArtifact(id) + + monkeypatch.setattr(FileArtifact, "put", staticmethod(fake_fa_put)) + monkeypatch.setattr(FileArtifactUniverse, "put", staticmethod(fake_fau_put)) + return put_calls, dumped + + +class TestCreateFileArtifactUniverse: + def test_bundles_three_dirs_faithfully(self, monkeypatch): + put_calls, dumped = _capture_puts(monkeypatch) step = VerifyUniverseLoadExportRoundtripStep( id="s", version=None, env_id="multi-x", universe_artifact_id="uni", emit_file_artifact_universe=True @@ -200,17 +221,14 @@ def fake_fau_put(id, *, file_artifacts): orig_sas = [_FakeServiceArtifact("slack"), _FakeServiceArtifact("oracle_gl")] - class _FakeExportedUniverse: - def get_environment_artifacts(self): - return [_FakeServiceArtifact("slack"), _FakeServiceArtifact("oracle_gl")] - # unicode + float to prove lossless, ensure_ascii=False dump export2_raw = { "slack": {"messages": [{"body": "héllo", "ts": 1772461920.0}]}, "oracle_gl": {"lines": [{"amount": 1500000001.5, "product_code": "OACCT"}]}, } - fau = step._create_file_artifact_universe(orig_sas, _FakeExportedUniverse(), export2_raw, env_version=3, universe_version=7) + fau = step._create_file_artifact_universe(orig_sas, _FakeExportedUniverse("slack", "oracle_gl"), export2_raw, + env_version=3, universe_version=7) assert fau.id == "multi-x__validate-v3-uni-v7-fau" assert put_calls["id"] == "multi-x__validate-v3-uni-v7-fau" @@ -231,6 +249,21 @@ def get_environment_artifacts(self): assert "héllo" in slack_dumped # ensure_ascii=False assert json.loads(dumped["multi-x__validate-v3-uni-v7-export2-oracle_gl"]) == export2_raw["oracle_gl"] + def test_bundles_each_file_of_a_service_of_several_at_its_path(self, monkeypatch): + put_calls, _ = _capture_puts(monkeypatch) + step = VerifyUniverseLoadExportRoundtripStep( + id="s", version=None, env_id="multi-x", universe_artifact_id="uni", emit_file_artifact_universe=True + ) + desk = _FakeServiceArtifact("desk", ["Desktop/a.txt", "b.txt"]) + + step._create_file_artifact_universe([desk], _FakeExportedUniverse("desk"), {"desk": {"files": 2}}, + env_version=3, universe_version=7) + + fa_map = put_calls["file_artifacts"] + assert set(fa_map) == {"original/desk/Desktop/a.txt", "original/desk/b.txt", "export_1/desk.json", + "export_2/desk.json"} + assert all(fa_map[f"original/desk/{path}"] is file for path, file in desk.get_file_artifacts().items()) + class TestFrontendContract: """Pin the shape of the merged UNIVERSE_COMPATIBILITY doc the hub frontend renders. diff --git a/tst/unit/task_step/test_verify_universe_roundtrip.py b/tst/unit/task_step/test_verify_universe_roundtrip.py index 137dec444..67a8badde 100644 --- a/tst/unit/task_step/test_verify_universe_roundtrip.py +++ b/tst/unit/task_step/test_verify_universe_roundtrip.py @@ -1,5 +1,6 @@ """VerifyUniverseLoadExportRoundtripStep exports each child env at the base the stored env card gives it, MCP server or -website, and a child env the card doesn't list through legacy /export-state. Only the gateway HTTP boundary is mocked.""" +website, and a child env the card doesn't list through legacy /export-state. An environment of several files, not one +JSON state, is compared export to export only, as a non-JSON one is. Only the gateway HTTP boundary is mocked.""" from __future__ import annotations @@ -68,7 +69,7 @@ async def test_both_export_rounds_read_the_deployed_record(monkeypatch): monkeypatch.setattr(VerifyUniverseLoadExportRoundtripStep, "_create_universe_artifact", lambda *a: MagicMock(id="u-export1")) monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a: _FakeEnv()) universe = MagicMock(id="u", version=1) - universe.get_environment_artifacts.return_value = [MagicMock(environment_name="items", **{"get_file_artifact.return_value.load.return_value": b'{"items": []}'})] + universe.get_environment_artifacts.return_value = [_environment("items", {"items.json": b'{"items": []}'})] monkeypatch.setattr("agent_env.artifact.EnvironmentUniverseArtifact.get", lambda *a: universe) step = VerifyUniverseLoadExportRoundtripStep(id="rt", version=None, env_id="env-x", universe_artifact_id="u", persist_result=False) @@ -78,6 +79,25 @@ async def test_both_export_rounds_read_the_deployed_record(monkeypatch): assert ctx.metadata["verifications"]["universe_compatibility"]["compatible"] is True +@pytest.mark.asyncio +async def test_an_environment_of_several_files_is_compared_export_to_export_only(monkeypatch): + record = _record(await _card(["mcp-desk"])) + monkeypatch.setattr(VerifyUniverseLoadExportRoundtripStep, "_export_all", AsyncMock(return_value={"desk": {"files": ["a"]}})) + monkeypatch.setattr(VerifyUniverseLoadExportRoundtripStep, "_create_universe_artifact", lambda *a: MagicMock(id="u-export1")) + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a: _FakeEnv()) + universe = MagicMock(id="u", version=1) + desk = _environment("desk", {"Desktop/a.txt": b"not json", "b.txt": b"{}"}) + universe.get_environment_artifacts.return_value = [desk] + monkeypatch.setattr("agent_env.artifact.EnvironmentUniverseArtifact.get", lambda *a: universe) + + step = VerifyUniverseLoadExportRoundtripStep(id="rt", version=None, env_id="env-x", universe_artifact_id="u", persist_result=False) + ctx = await step.execute(TaskStepContext(deployed_envs=[record])) + + verdict = ctx.metadata["verifications"]["universe_compatibility"] + assert verdict["compatible"] is True and verdict["environments"]["desk"]["issues"] == [] + assert all(not file.load.called for file in desk.get_file_artifacts().values()) + + @pytest.mark.asyncio async def test_the_universe_files_are_read_and_the_exports_uploaded_off_the_event_loop(monkeypatch): reads: list[bool] = [] @@ -98,7 +118,9 @@ def put(**kwargs): for cls in (EnvironmentArtifact, EnvironmentUniverseArtifact, FileArtifactUniverse): monkeypatch.setattr(cls, "put", staticmethod(lambda **kwargs: MagicMock(id=kwargs["id"]))) universe = MagicMock(id="u", version=1) - universe.get_environment_artifacts.return_value = [MagicMock(environment_name="items", **{"get_file_artifact.return_value.load": load})] + environment = MagicMock(environment_name="items") + environment.get_file_artifacts.return_value = {"items.json": MagicMock(load=load)} + universe.get_environment_artifacts.return_value = [environment] monkeypatch.setattr(EnvironmentUniverseArtifact, "get", lambda *a: universe) step = VerifyUniverseLoadExportRoundtripStep( @@ -109,6 +131,14 @@ def put(**kwargs): assert reads == [False] and uploads == [False, False] +def _environment(name: str, contents: dict[str, bytes]) -> MagicMock: + """An environment artifact of ``contents``' files by path.""" + environment = MagicMock(environment_name=name) + environment.get_file_artifacts.return_value = {path: MagicMock(**{"load.return_value": data}) + for path, data in contents.items()} + return environment + + async def _card(keys: list[str]) -> dict: """The real gateway's composed card, with a v1 backing server behind each key.""" from agent_env.env.gateway.gateway import Gateway