diff --git a/AGENTS.md b/AGENTS.md index 16266609..9135ed2b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -152,7 +152,7 @@ orchestrators call it the same way. What callers rely on: - Skips are declared capability gaps, decided at collection time: the gates in `tst/util/capabilities.py` produce `skipif` marks with the reason `agentenv-capability-missing: ` (`model_endpoint_configured`, `remote_sandbox`, - `default_a2a_agent`, `mcp_server_sources`). CI rejects any other skip reason; a broken configuration must fail loudly, not skip. + `default_a2a_agent`, `mcp_server_sources`, `gnu_stat`). CI rejects any other skip reason; a broken configuration must fail loudly, not skip. - The suite is backend-agnostic: a test about one particular backend lives with that backend, not here. A custom store must pass `tst/store/{conformance,object_conformance,image_conformance,secret_conformance}.py`, the same kits the built-ins pass. diff --git a/src/agent_env/a2a_agent/validator.py b/src/agent_env/a2a_agent/validator.py index 0a642d95..8c952532 100644 --- a/src/agent_env/a2a_agent/validator.py +++ b/src/agent_env/a2a_agent/validator.py @@ -29,7 +29,7 @@ invoke_transfer, ) from agent_env.config import get_config -from agent_env.store.routing import refuse_local_derivation +from agent_env.store.ids import derive_id, key_segment from agent_env.task_step.task_steps.a2a_agent_validator.verify_a2a_modalities import ( AUDIO_M4A_EXPECTED, AUDIO_M4A_PROBE_PARTS, @@ -144,7 +144,6 @@ async def validate( Returns the merged `verifications` dict from the run's task context. """ - refuse_local_derivation(agent.id, "agent", "validating") from agent_env.task import Task from agent_env.task_step import ( AddSkillsTaskStep, @@ -173,7 +172,7 @@ async def validate( from agent_env.task_step.task_step import TaskStep, TaskStepDependency from agent_env.a2a_agent.a2a_agent import A2AAgent - task_id = f"validate-a2a-{agent.id}-v{agent.version}" + task_id = derive_id(agent.id, f"validate-a2a-v{agent.version}") # ── Upload object-store fixtures (skill + URI probe bytes) ─────────── skill_object_url = await asyncio.to_thread( @@ -744,9 +743,9 @@ def _upload_install_test_image_fixture(agent: "A2AAgent"): from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse ts = int(time.time()) - universe_id = f"validate-install-image-{agent.id}-v{agent.version}-{ts}" + universe_id = derive_id(agent.id, f"validate-install-image-v{agent.version}-{ts}") config = get_config() - s3_url = config.get_object_store().object_url(f"{config.get_artifact_key_prefix()}a2a_validator/install_test_image/{ts}/") + s3_url = config.get_object_store_for(universe_id).object_url(f"{config.get_artifact_key_prefix()}a2a_validator/install_test_image/{ts}/") with tempfile.NamedTemporaryFile("w", suffix=".Dockerfile", delete=False) as f: f.write("FROM ubuntu:24.04\n") @@ -772,9 +771,9 @@ def _upload_skill_fixture( from agent_env.task_step.task_steps.add_skills import Skill config = get_config() - store = config.get_object_store() + store = config.get_object_store_for(agent.id) prefix = ( - f"{config.get_artifact_key_prefix()}a2a_validator/validator_skill/{agent.id}-v{agent.version}/{name}/" + f"{config.get_artifact_key_prefix()}a2a_validator/validator_skill/{key_segment(agent.id)}-v{agent.version}/{name}/" ) skill = Skill( name=name, @@ -797,8 +796,8 @@ def _upload_probe_fixtures(agent: "A2AAgent", skill_object_url: str) -> _Uploade """ config = get_config() - store = config.get_object_store() - prefix = f"{config.get_artifact_key_prefix()}a2a_validator/probe_fixtures/{agent.id}-v{agent.version}/" + store = config.get_object_store_for(agent.id) + prefix = f"{config.get_artifact_key_prefix()}a2a_validator/probe_fixtures/{key_segment(agent.id)}-v{agent.version}/" png_object_uri = store.put(f"{prefix}red.png", base64.b64decode(IMAGE_PROBE_PNG_B64), content_type="image/png", allow_overwrite=True) png_signed_url = store.signed_get_url(png_object_uri) diff --git a/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py index 2fcc2ac4..db9ffe1d 100644 --- a/src/agent_env/bundle/resolve.py +++ b/src/agent_env/bundle/resolve.py @@ -238,7 +238,8 @@ def _image(self, entry: BundleEntry, config: dict, ref: EntityRef, references: l def _derived_outputs(self, entry: BundleEntry, step: dict, cls: type) -> bool: """Write in any output id the step's type derives when the field is left out (build_mcp_cli's - ``cli-``), so it's named like one written out. False when its type can't read the step.""" + ``__cli``), so it's named like one written out. Only its last segment is kept, so an id derived + from another bundle's entity is still named under this task. False when its type can't read the step.""" absent = [ref for ref in cls.entity_refs or () if ref.role is RefRole.OUTPUT and ref.path == ref.field and step.get(ref.field) is None] if not absent: @@ -250,7 +251,7 @@ def _derived_outputs(self, entry: BundleEntry, step: dict, cls: type) -> bool: return False for ref in absent: if read.get(ref.field) is not None: - step[ref.field] = read[ref.field] + step[ref.field] = read[ref.field].rsplit("/", 1)[-1] return True def _outputs(self, entry: BundleEntry, declared: list[tuple[int, dict, type]]) -> dict[tuple[EntityKind, str], Output]: diff --git a/src/agent_env/cli/a2a_agent/put.py b/src/agent_env/cli/a2a_agent/put.py index bc6777e9..09a60beb 100644 --- a/src/agent_env/cli/a2a_agent/put.py +++ b/src/agent_env/cli/a2a_agent/put.py @@ -5,7 +5,7 @@ from agent_env.a2a_agent import A2AAgent from agent_env.artifact import DockerImageArtifact -from agent_env.cli.utils import build_platform_option, detect_env_metadata, skips_local_validation +from agent_env.cli.utils import build_platform_option, detect_env_metadata from agent_env.utils.docker_build import build_image @@ -72,7 +72,7 @@ def put(agent_id: str, dockerfile: str, context_path: str | None, env_var_pairs: ) click.echo(f"Created A2A agent: id={agent.id} version={agent.version} image={artifact.id}:{artifact.version}") - if not skip_validation and not skips_local_validation(agent_id, "agent"): + if not skip_validation: from agent_env.cli.a2a_agent.validate import run_validation click.echo() run_validation(agent, litellm_api_key=litellm_api_key) diff --git a/src/agent_env/cli/artifact/file_artifact_universe.py b/src/agent_env/cli/artifact/file_artifact_universe.py index 6ac6bb47..9702344e 100644 --- a/src/agent_env/cli/artifact/file_artifact_universe.py +++ b/src/agent_env/cli/artifact/file_artifact_universe.py @@ -6,7 +6,7 @@ import click from agent_env.artifact import FileArtifact, FileArtifactUniverse -from agent_env.store.ids import fs_safe +from agent_env.store.ids import fs_safe, is_local_id, key_segment # --------------------------------------------------------------------------- @@ -241,7 +241,7 @@ def get_many( def _download(job: tuple[str, str, FileArtifact]) -> tuple[str, str, int]: universe_id, filename, fa = job data = fa.load() - dest = out / fs_safe(universe_id) / filename + dest = out / (fs_safe(universe_id) if is_local_id(universe_id) else key_segment(universe_id)) / filename dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(data) return (universe_id, filename, len(data)) diff --git a/src/agent_env/cli/env/mcp_server.py b/src/agent_env/cli/env/mcp_server.py index 8030855c..00c3e74b 100644 --- a/src/agent_env/cli/env/mcp_server.py +++ b/src/agent_env/cli/env/mcp_server.py @@ -13,7 +13,6 @@ env_provider_type_option, environment_name_options, resolve_environment_name, - skips_local_validation, ) from agent_env.utils.card_naming import card_name_from_github, card_name_from_source from agent_env.utils.docker_build import DEFAULT_BUILD_PLATFORM, build_image @@ -175,7 +174,7 @@ def _on_progress(step: str, message: str, percent: int) -> None: on_progress=_on_progress, )) click.echo(f"Created MCPServerEnv: id={env.id} version={env.version} environment_name={env.environment_name} env_provider_type={env.env_provider_type}") - if (run_validation or override) and not skips_local_validation(env.id, "env"): + if run_validation or override: # GitHub-sourced build: validation_report.json lives in the repo, not # locally, so the unit-test verdict is unknown here (doesn't block). _gate_release(env, [], override) @@ -216,7 +215,7 @@ def _on_progress(step: str, message: str, percent: int) -> None: metadata=metadata if metadata else None, ) click.echo(f"Created MCPServerEnv: id={env.id} version={env.version} environment_name={env.environment_name} env_provider_type={env.env_provider_type}") - if (run_validation or override) and not skips_local_validation(env.id, "env"): + if run_validation or override: # Local build: look for the env-build handoff report next to the build context. _gate_release(env, [str(context), str(dockerfile_path.parent)], override) diff --git a/src/agent_env/cli/env/multi.py b/src/agent_env/cli/env/multi.py index 188f10d3..848c035f 100644 --- a/src/agent_env/cli/env/multi.py +++ b/src/agent_env/cli/env/multi.py @@ -3,7 +3,7 @@ import click from agent_env.artifact import EnvironmentArtifact, EnvironmentUniverseArtifact -from agent_env.cli.utils import detect_base_metadata, env_provider_type_option, skips_local_validation +from agent_env.cli.utils import detect_base_metadata, env_provider_type_option from agent_env.env import Env, MultiEnv @@ -78,7 +78,7 @@ def put(env_id: str, mcp_servers: tuple[str, ...], websites: tuple[str, ...], me multi_env = MultiEnv.put(id=env_id, mcp_server_envs=mcp_server_envs, website_envs=website_envs, metadata=metadata if metadata else None, name=name, env_provider_type=env_provider_type) click.echo(f"Created MultiEnv: id={multi_env.id} version={multi_env.version} env_provider_type={multi_env.env_provider_type}") - if run_validation and not skips_local_validation(multi_env.id, "env"): + if run_validation: click.echo("\nValidating environment...") instance_id = asyncio.run(multi_env.validate(on_progress=click.echo)) click.echo(f"Validation task: {instance_id}") diff --git a/src/agent_env/cli/env/website.py b/src/agent_env/cli/env/website.py index 49bb4d1c..e7a165ed 100644 --- a/src/agent_env/cli/env/website.py +++ b/src/agent_env/cli/env/website.py @@ -13,7 +13,6 @@ env_provider_type_option, environment_name_options, resolve_environment_name, - skips_local_validation, ) from agent_env.utils.card_naming import card_name_from_github, card_name_from_source from agent_env.utils.docker_build import DEFAULT_BUILD_PLATFORM, build_image @@ -132,7 +131,7 @@ def _frontend_progress(step: str, message: str, percent: int) -> None: env_provider_type=env_provider_type, )) click.echo(f"Created WebsiteEnv: id={env.id} version={env.version} environment_name={env.environment_name}") - if not skip_validation and not skips_local_validation(env.id, "env"): + if not skip_validation: click.echo("\nValidating environment (use --skip-validation to skip)...") instance_id = asyncio.run(env.validate(on_progress=click.echo)) click.echo(f"Validation task: {instance_id}") @@ -188,7 +187,7 @@ def _frontend_progress(step: str, message: str, percent: int) -> None: env_provider_type=env_provider_type, ) click.echo(f"Created WebsiteEnv: id={env.id} version={env.version} environment_name={env.environment_name} env_provider_type={env.env_provider_type}") - if not skip_validation and not skips_local_validation(env.id, "env"): + if not skip_validation: click.echo("\nValidating environment (use --skip-validation to skip)...") instance_id = asyncio.run(env.validate(on_progress=click.echo)) click.echo(f"Validation task: {instance_id}") diff --git a/src/agent_env/cli/task/run.py b/src/agent_env/cli/task/run.py index 83da292f..057454c0 100644 --- a/src/agent_env/cli/task/run.py +++ b/src/agent_env/cli/task/run.py @@ -12,7 +12,8 @@ from agent_env.cli.banner import print_banner from agent_env.cli.identity import get_agent_env_client_id -from agent_env.store.ids import fs_safe +from agent_env.store.ids import derive_id, fs_safe, is_local_id, validate_local_id +from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep _print_lock = asyncio.Lock() @@ -436,6 +437,13 @@ async def _run_all(): _write_context(context, task_id, output_dir) +def _seed_universe_id(task, seed: dict) -> str | None: + """The universe a seed's runs collect into, so runs of one seed share it: derived from the task version and the + seed's id, else its name. None leaves collect_artifacts to name it after the run.""" + seed_name = seed.get("id") or seed.get("name") + return derive_id(task.id, f"v{task.version}-{seed_name}") if seed_name else None + + @click.command("run-batch") @click.option("--id", "task_id", required=True, help="Task id") @click.option("--version", "task_version", default=None, type=int, help="Task version (defaults to latest)") @@ -480,6 +488,15 @@ def run_batch(task_id: str, task_version: int | None, seeds: str, concurrency: i click.echo(f"Fetching task: id={task_id} version={task_version or 'latest'}...") task = Task.get(task_id, version=task_version) click.echo(f"Found task: id={task.id} version={task.version} steps={len(task.steps)}") + suffixes = {step.universe_id_suffix or "" for step in task.steps if isinstance(step, CollectArtifactsTaskStep)} + if is_local_id(task.id) and suffixes: + for index, seed in enumerate(seed_rows, 1): + universe_id = _seed_universe_id(task, seed) + for suffix in sorted(suffixes) if universe_id else (): + try: + validate_local_id(universe_id + suffix) + except ValueError as e: + raise click.ClickException(f"seed {index} can't name an @local universe: {e}") from e click.echo(click.style(f"Running {len(seed_rows)} seeds with concurrency={concurrency}...", fg="blue")) click.echo() @@ -499,11 +516,7 @@ async def _run_seed(index: int, seed: dict): _stamp_agent_env_client_metadata(ctx.metadata, client_id) ctx.metadata["run_group_id"] = batch_run_group_id ctx.metadata["seed"] = seed - # Populate universe_id so collect_artifacts can name the - # FileArtifactUniverse stably across runs of the same seed. - # Prefer seed.id, fall back to seed.name. If neither is set, - # collect_artifacts falls through to the per-run instance_id. - universe_id = seed.get("id") or seed.get("name") + universe_id = _seed_universe_id(task, seed) if universe_id: ctx.metadata["universe_id"] = universe_id if litellm_api_key: diff --git a/src/agent_env/cli/utils.py b/src/agent_env/cli/utils.py index 47aeffc4..a0a45920 100644 --- a/src/agent_env/cli/utils.py +++ b/src/agent_env/cli/utils.py @@ -9,7 +9,6 @@ from agent_env.providers.env_providers.env_provider import _env_provider_class from agent_env.providers.env_providers.env_server_provider import EnvironmentServerProvider from agent_env.store.base import NotFoundError -from agent_env.store.ids import is_local_id from agent_env.utils.docker_build import DEFAULT_BUILD_PLATFORM @@ -172,12 +171,3 @@ def deployed_env_from_instance(env_id: str | None, instance_id: str) -> Env: raise click.UsageError( f"--id '{env_id}' does not match instance '{instance_id}' (env '{env.id}')") return env - - -def skips_local_validation(entity_id: str, kind: str) -> bool: - """Whether a put leaves ``entity_id`` unvalidated because it is an ``@local`` id: validating one - isn't supported yet, and the entity itself was written fine.""" - if not is_local_id(entity_id): - return False - click.echo(f"Skipped validation: validating an @local {kind} isn't supported yet") - return True diff --git a/src/agent_env/env/envs/mcp_server.py b/src/agent_env/env/envs/mcp_server.py index c9975356..c7db1441 100644 --- a/src/agent_env/env/envs/mcp_server.py +++ b/src/agent_env/env/envs/mcp_server.py @@ -13,7 +13,7 @@ from agent_env.artifact import Artifact, DockerImageArtifact from agent_env.artifact.artifacts.docker_image import GitHubBuildResult, ProgressCallback, refuse_local_github_build from agent_env.env.env import Env, gateway_url_of -from agent_env.store.routing import refuse_local_derivation +from agent_env.store.ids import derive_id 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, @@ -235,14 +235,13 @@ async def _load_environment_artifact_legacy(self, file_artifact) -> None: async def validate(self, on_progress: Optional[Callable[[str], None]] = None) -> str: """Run the env validator task and return the task instance ID. - Creates a new versioned Task with ID ``validate-{env_id}-v{version}`` each time. + Creates a new versioned Task with ID ``{env_id}__validate-v{version}`` each time. """ - refuse_local_derivation(self.id, "env", "validating") from agent_env.task import Task from agent_env.task_step import DeployAgentTaskStep, DeployEnvTaskStep, PromptAgentTaskStep, ValidationGateAggregatorStep, VerifyCoreEnvironmentProtocolStep, VerifyEnvironmentCardStep, VerifyMCPToolSchemaTaskStep, VerifySpecConformanceTaskStep, VerifyMCPEnvAssessmentStep from agent_env.task_step.task_steps.mcp_env_validator import TOOL_CORRECTNESS_PROMPT, TOOL_CORRECTNESS_OUTPUT_FORMAT - task_id = f"validate-{self.id}-v{self.version}" + task_id = derive_id(self.id, f"validate-v{self.version}") prompt_id = f"{task_id}-prompt" # Env metadata can narrow the gate, e.g. {"required_gates": ["mcp_tool_schema"]}. @@ -306,12 +305,11 @@ async def create_cli(self, command_name: Optional[str] = None, on_progress: Opti logger.warning(f"Cached CliArtifact {cached} not found in store; rebuilding") cmd = command_name or self.environment_name - task_id = f"create-cli-{self.id}-v{self.version}" - cli_artifact_id = f"cli-{self.id}" + task_id = derive_id(self.id, f"create-cli-v{self.version}") task = Task.put(id=task_id, steps=[ DeployEnvTaskStep(id=f"{task_id}-deploy", version=None, env_id=self.id, env_version=self.version), - BuildMcpCliTaskStep(id=f"{task_id}-build", version=None, env_id=self.id, command_name=cmd, cli_artifact_id=cli_artifact_id), + BuildMcpCliTaskStep(id=f"{task_id}-build", version=None, env_id=self.id, command_name=cmd), ]) logger.info(f"Created create-cli task: {task.id} version={task.version}") diff --git a/src/agent_env/env/envs/multi_env.py b/src/agent_env/env/envs/multi_env.py index 884c81f9..b927798d 100644 --- a/src/agent_env/env/envs/multi_env.py +++ b/src/agent_env/env/envs/multi_env.py @@ -32,7 +32,7 @@ def _snapshot_after_load_default() -> bool: ) from agent_env.env.envs.mcp_server import MCPServerEnv from agent_env.env.envs.website import WebsiteEnv -from agent_env.store.routing import refuse_local_derivation +from agent_env.store.ids import derive_id from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER from agent_env.attribution import Attribution @@ -633,11 +633,10 @@ async def validate(self, on_progress: Callable[[str], None] | None = None) -> st Deploys the env, fetches + persists its composed EnvironmentCard, then tears down. """ - refuse_local_derivation(self.id, "env", "validating") from agent_env.task import Task from agent_env.task_step import DeployEnvTaskStep, VerifyEnvironmentCardStep - task_id = f"validate-{self.id}-v{self.version}" + task_id = derive_id(self.id, f"validate-v{self.version}") task = Task.put( id=task_id, steps=[ @@ -669,7 +668,6 @@ async def validate_universe_compatibility(self, universe_artifact_id: str, unive merged into the ``UNIVERSE_COMPATIBILITY`` result as per-service critical issues — so they surface (and gate ``compatible``) the same way programmatic issues do. """ - refuse_local_derivation(self.id, "env", "validating") from agent_env.task import Task from agent_env.task_step import ( CombineUniverseVerdictsStep, @@ -689,7 +687,9 @@ async def validate_universe_compatibility(self, universe_artifact_id: str, unive from agent_env.artifact import EnvironmentUniverseArtifact universe = EnvironmentUniverseArtifact.get(universe_artifact_id, universe_artifact_version) - task_id = f"validate-universe-compat-{self.id}-v{self.version}-{universe.id}-v{universe.version}" + task_id = VerifyUniverseLoadExportRoundtripStep.validation_id( + "validate-universe-compat", self.id, self.version, universe.id, universe.version + ) environment_names = [sa.environment_name for sa in universe.get_environment_artifacts()] fau_id = VerifyUniverseLoadExportRoundtripStep.file_artifact_universe_id(self.id, self.version, universe.id, universe.version) judge_name = "universe-judge" diff --git a/src/agent_env/env/envs/website.py b/src/agent_env/env/envs/website.py index f9ed1b02..f3e5430b 100644 --- a/src/agent_env/env/envs/website.py +++ b/src/agent_env/env/envs/website.py @@ -13,7 +13,7 @@ from agent_env.artifact import Artifact, DockerImageArtifact, EnvironmentArtifact from agent_env.artifact.artifacts.docker_image import GitHubBuildResult, ProgressCallback, refuse_local_github_build from agent_env.env.env import Env, gateway_url_of -from agent_env.store.routing import refuse_local_derivation +from agent_env.store.ids import derive_id 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, @@ -202,11 +202,10 @@ async def validate(self, on_progress: Optional[Callable[[str], None]] = None) -> Deploys the env, fetches + persists its composed EnvironmentCard, then tears down. """ - refuse_local_derivation(self.id, "env", "validating") from agent_env.task import Task from agent_env.task_step import DeployEnvTaskStep, VerifyEnvironmentCardStep - task_id = f"validate-{self.id}-v{self.version}" + task_id = derive_id(self.id, f"validate-v{self.version}") task = Task.put( id=task_id, steps=[ diff --git a/src/agent_env/env/snapshot_store.py b/src/agent_env/env/snapshot_store.py index 13a756a5..c7214421 100644 --- a/src/agent_env/env/snapshot_store.py +++ b/src/agent_env/env/snapshot_store.py @@ -5,6 +5,7 @@ import asyncio import hashlib import logging +import re from dataclasses import dataclass from datetime import datetime, timezone from typing import Callable, Optional @@ -12,7 +13,8 @@ from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_CONTAINER from agent_env.config import get_config from agent_env.store.document_store import AbsentOrNull, Filter, Sort -from agent_env.store.routing import refuse_local_derivation +from agent_env.store.ids import derive_id, fs_safe, is_local_id, key_segment +from agent_env.store.object_store import ObjectStore logger = logging.getLogger(__name__) @@ -91,7 +93,6 @@ async def create(cls, instance_id: str, on_progress: ProgressCallback | None = N log("validate", "Fetching env instance...", 5) deployed_env = get_env_instance_store().get(instance_id) - refuse_local_derivation(deployed_env.env_id, "env", "snapshotting") env = Env.get(deployed_env.env_id, deployed_env.env_version) if not isinstance(env, MultiEnv): raise ValueError( @@ -110,6 +111,11 @@ async def create(cls, instance_id: str, on_progress: ProgressCallback | None = N raise ValueError(f"Instance '{instance_id}' has no service_universe loaded") universe_id = environment_universe["id"] universe_version = environment_universe["version"] + if is_local_id(env_id) != is_local_id(universe_id): + raise ValueError( + f"env {env_id!r} and universe {universe_id!r} are in different namespaces; a snapshot is recorded " + "under both, so both have to be @local ids or neither" + ) log("reconnect", f"Reconnecting to sandbox {deployed_env.sandbox_id}...", 10) multi_env = await MultiEnv.from_deployed_env(deployed_env) if multi_env._sandbox is None: @@ -126,6 +132,8 @@ async def create(cls, instance_id: str, on_progress: ProgressCallback | None = N f"(this env's state backend does not support it: {type(state_provider).__name__ if state_provider else None}). " "Remote-backed deploys use the normal per-service load." ) + # Fail on an unsignable store before any command runs on the sandbox; the upload presigns a fresh url. + await asyncio.to_thread(_presigned_put_url, *_tarball_destination(env_id, universe_id)) log("check_changelog", "Checking changelog...", 20) is_clean = await _check_changelog_empty(multi_env._sandbox) @@ -137,9 +145,9 @@ async def create(cls, instance_id: str, on_progress: ProgressCallback | None = N log("store_artifact", "Storing Docker image artifact...", 70) logger.info("Storing snapshot as DockerImageArtifact...") - image_tag = f"env-snapshot-{env_id}-{universe_id}" + image_tag = _image_tag(env_id, universe_id) artifact = DockerImageArtifact.put_tar( - id=f"env-snapshot-{env_id}", + id=derive_id(env_id, "env-snapshot"), description=( f"Snapshot of servicedb for env={env_id} " f"universe={universe_id} v{universe_version}" @@ -201,7 +209,7 @@ async def _snapshot_servicedb(sandbox, env_id: str, environment_universe_id: str logger.info("Exporting pgdata from servicedb container...") await sandbox.exec_script(f"docker cp {container_id}:/var/lib/postgresql/data /tmp/pgdata") - image_tag = f"env-snapshot-{env_id}-{environment_universe_id}" + image_tag = _image_tag(env_id, environment_universe_id) log("snapshot_db", f"Building Docker image on sandbox: {image_tag}...", 50) logger.info(f"Building Docker image on sandbox: {image_tag}...") await sandbox.exec_script( @@ -221,16 +229,8 @@ async def _snapshot_servicedb(sandbox, env_id: str, environment_universe_id: str logger.info("Saving Docker image to tar.gz on sandbox...") await sandbox.exec_script(f"docker save {image_tag} | gzip > /tmp/snapshot-image.tar.gz") - config = get_config() - object_store = config.get_object_store() - object_url = object_store.object_url( - f"{config.get_artifact_key_prefix()}env-snapshots/{env_id}/{environment_universe_id}/{image_tag}.tar.gz" - ) - put_url = await asyncio.to_thread(object_store.signed_put_url, object_url) - if put_url is None: - raise RuntimeError( - f"{type(object_store).__name__} can't presign uploads; env snapshots need a signable object store or local execution." - ) + object_store, object_url = _tarball_destination(env_id, environment_universe_id) + put_url = await asyncio.to_thread(_presigned_put_url, object_store, object_url) log("snapshot_db", "Uploading Docker image...", 60) logger.info(f"Uploading Docker image to {object_url}...") await sandbox.exec_script( @@ -242,6 +242,37 @@ async def _snapshot_servicedb(sandbox, env_id: str, environment_universe_id: str return object_url +_DOCKER_NAME = re.compile(r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*") + + +def _image_tag(env_id: str, universe_id: str) -> str: + """The snapshot image's tag, a docker reference the sandbox builds with: ``env-snapshot--`` + when that is a short, valid docker name, else a slug of the env and a hash of both ids.""" + tag = f"env-snapshot-{fs_safe(env_id)}-{fs_safe(universe_id)}" + if len(tag) <= 200 and _DOCKER_NAME.fullmatch(tag): + return tag + slug = re.sub(r"[^a-z0-9]+", "-", fs_safe(env_id).lower()).strip("-")[:64].rstrip("-") + digest = hashlib.sha256(f"{env_id}\0{universe_id}".encode("utf-8", "surrogatepass")).hexdigest()[:12] + return f"env-snapshot-{slug}-{digest}" if slug else f"env-snapshot-{digest}" + + +def _tarball_destination(env_id: str, universe_id: str) -> tuple[ObjectStore, str]: + """The object store the snapshot image's tarball is uploaded to, and its url there.""" + config = get_config() + store = config.get_object_store_for(env_id) + key = f"env-snapshots/{key_segment(env_id)}/{key_segment(universe_id)}/{_image_tag(env_id, universe_id)}.tar.gz" + return store, store.object_url(f"{config.get_artifact_key_prefix()}{key}") + + +def _presigned_put_url(object_store: ObjectStore, object_url: str) -> str: + put_url = object_store.signed_put_url(object_url) + if put_url is None: + raise RuntimeError( + f"{type(object_store).__name__} can't presign uploads; env snapshots need a signable object store or local execution." + ) + return put_url + + class EnvSnapshotStore: def __init__(self) -> None: self._indexed = None diff --git a/src/agent_env/store/ids.py b/src/agent_env/store/ids.py index a0d66b82..b047b279 100644 --- a/src/agent_env/store/ids.py +++ b/src/agent_env/store/ids.py @@ -2,7 +2,8 @@ An id opening ``@/`` names the store it lives in; ``@local/…`` is the only namespace today. Documents keep an id verbatim. Everything derived from an ``@local`` id uses one bounded -segment, ``local/-``, and every other id passes through byte-identical. +segment, ``local/-``; an object key or image repository takes a bare id over +200 bytes as ``-``; every other id passes through byte-identical. The encoders never raise: they expect ids that ``validate_local_id`` accepts, and still return a safe segment for any other input. An id may contain spaces, ``&`` and parentheses, so quote one before it reaches a shell or a URL, and derive object keys, image repositories and filenames with @@ -19,6 +20,8 @@ LOCAL_NAMESPACE = "local" LOCAL_PREFIX = f"@{LOCAL_NAMESPACE}/" MAX_LOCAL_ID_BYTES = 4096 +# A filesystem path component holds at most 255 bytes, so a bare id over this many gets a bounded segment. +MAX_BARE_SEGMENT_BYTES = 200 _SLUG_LENGTH = 48 _HASH_LENGTH = 12 @@ -99,7 +102,11 @@ def aliases_local_encoding(entity_id: str) -> bool: def key_segment(entity_id: str) -> str: """The object-key segment for ``entity_id`` (``artifacts////…``).""" - return _local_segment(entity_id) if is_local_id(entity_id) else entity_id + if is_local_id(entity_id): + return _local_segment(entity_id) + if len(entity_id.encode("utf-8", "surrogatepass")) > MAX_BARE_SEGMENT_BYTES: + return _local_segment(entity_id).removeprefix(f"{LOCAL_NAMESPACE}/") + return entity_id def image_repository(entity_id: str) -> str: @@ -113,8 +120,10 @@ def fs_safe(entity_id: str) -> str: return _local_segment(entity_id).replace("/", "-", 1) if is_local_id(entity_id) else entity_id -# The longest id derive_id makes is a skill's or CLI's per-file artifact, ``{id}__files__<16 hex>``, -# so an authored id leaves room for those 25 bytes. +# The longest chain derive_id makes on an authored skill or CLI id is its per-file artifact, +# ``{id}__files__<16 hex>``, so an authored id leaves room for those 25 bytes. An id built on a derived id +# (an env's ``{env}__cli``) or on a run's id can pass 4096 bytes; validate_local_id refuses it before anything +# reaches a shared store. MAX_AUTHORED_LOCAL_ID_BYTES = MAX_LOCAL_ID_BYTES - len("__files__") - 16 diff --git a/src/agent_env/store/routing.py b/src/agent_env/store/routing.py index a685edd5..d764b672 100644 --- a/src/agent_env/store/routing.py +++ b/src/agent_env/store/routing.py @@ -131,13 +131,6 @@ def refuse_in_local_run(what: str) -> None: ) -def refuse_local_derivation(entity_id: str, kind: str, doing: str) -> None: - """Refuse ``doing`` to an ``@local`` entity whose tasks, artifacts or objects are named after it - in a form that doesn't keep the ``@local`` namespace yet.""" - if is_local_id(entity_id): - raise ValueError(f"{entity_id!r} is an @local {kind}, and {doing} one isn't supported yet") - - def refuse_local_references(entity_id: object, value: Any) -> None: """Refuse a bare entity naming an ``@local`` id anywhere in ``value``: a shared store must never point into one person's local stores.""" diff --git a/src/agent_env/task_step/snapshot_utils/agent_state_capture.py b/src/agent_env/task_step/snapshot_utils/agent_state_capture.py index d284aba1..8e13877a 100644 --- a/src/agent_env/task_step/snapshot_utils/agent_state_capture.py +++ b/src/agent_env/task_step/snapshot_utils/agent_state_capture.py @@ -28,6 +28,7 @@ trajectory_mode, ) from agent_env.config import get_config +from agent_env.store.ids import key_segment logger = logging.getLogger(__name__) @@ -78,7 +79,7 @@ async def capture_workspace( snapshot_version = await asyncio.to_thread(get_artifact_store().next_version, artifact_id) config = get_config() capture_key_prefix = ( - f"{config.get_artifact_key_prefix()}agent_snapshots/{artifact_id}/{snapshot_version}-{uuid.uuid4().hex[:8]}/" + f"{config.get_artifact_key_prefix()}agent_snapshots/{key_segment(artifact_id)}/{snapshot_version}-{uuid.uuid4().hex[:8]}/" ) store = config.get_object_store() capture_prefix = store.object_url(capture_key_prefix) diff --git a/src/agent_env/task_step/snapshot_utils/snapshot_series.py b/src/agent_env/task_step/snapshot_utils/snapshot_series.py index 5dfe3163..3913a732 100644 --- a/src/agent_env/task_step/snapshot_utils/snapshot_series.py +++ b/src/agent_env/task_step/snapshot_utils/snapshot_series.py @@ -17,6 +17,7 @@ from datetime import datetime, timezone from typing import Optional +from agent_env.store.ids import derive_id, is_local_id, validate_local_id from agent_env.task_step.context import TaskStepContext from agent_env.task_step.snapshot_utils import agent_state_capture as capture @@ -84,18 +85,18 @@ def from_dict(cls, data: dict) -> "SnapshotConfig": ) -def _rollout_discriminator(step_id: str, instance_id: Optional[str]) -> str: - """The per-rollout token every artifact id in one series is scoped by. +def _rollout_base(step_id: str, instance_id: Optional[str]) -> str: + """The id every artifact id in one series is derived from: the rollout's instance id. Instance-scoped, so concurrent rollouts never contend on one id and retries reuse it. Deliberately not derived from the context id: a caller-supplied ``context_id`` is shared by every rollout, so ids built from it would collide. """ if instance_id: - return instance_id.rsplit("-", 1)[-1][:16] - generated = uuid.uuid4().hex[:8] + return instance_id + generated = f"adhoc-{uuid.uuid4().hex[:12]}" logger.warning( - "%s: no instance_id in context; using random artifact discriminator %s " + "%s: no instance_id in context; using random artifact base %s " "(retries will not be idempotent)", step_id, generated, ) return generated @@ -131,8 +132,10 @@ def __init__( self.a2a_context_id = a2a_context_id self.config = config self.trajectory_output_prefix = trajectory_output_prefix - self._discriminator = _rollout_discriminator(step_id, instance_id) - self.workspace_artifact_id = f"{step_id}-workspace-{self._discriminator}" + self._base = _rollout_base(step_id, instance_id) + self.workspace_artifact_id = derive_id(self._base, f"{step_id}-workspace") + if is_local_id(self._base): + validate_local_id(derive_id(self._base, f"snapshot-{step_id}-workspace")) self._lock = asyncio.Lock() self._stop = asyncio.Event() self._ticker: Optional[asyncio.Task] = None @@ -502,7 +505,7 @@ async def _capture_env_state( gateway_url=deployed.gateway_url, # Instance-scoped like the workspace id, never context-scoped: a # pinned `context_id` is shared by every concurrent rollout. - snapshot_id=f"snapshot-{self.config.env_id}-{self._discriminator}", + snapshot_id=derive_id(self._base, f"snapshot-{self.step_id}"), deployed=deployed, export_timeout_seconds=int(timeout_seconds), ) diff --git a/src/agent_env/task_step/task_steps/add_skills.py b/src/agent_env/task_step/task_steps/add_skills.py index aa3c3ac6..403d83b8 100644 --- a/src/agent_env/task_step/task_steps/add_skills.py +++ b/src/agent_env/task_step/task_steps/add_skills.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import logging import re from dataclasses import dataclass @@ -146,10 +147,18 @@ def _build_skill_for_installed_cli(cli_artifact_id: str, entry: dict) -> Skill: ) +def _files_skill_name(universe_id: str) -> str: + """``--files``: a slug of the id's last segment and a hash of the whole id, since an id itself + needn't be a valid skill name.""" + slug = re.sub(r"[^a-z0-9]+", "-", universe_id.rsplit("/", 1)[-1].lower()).strip("-")[:45].rstrip("-") + digest = hashlib.sha256(universe_id.encode("utf-8")).hexdigest()[:12] + return f"{slug}-{digest}-files" if slug else f"{digest}-files" + + def _build_skill_for_loaded_file_artifact_universe(universe_id: str, entry: dict) -> Skill: destination_path = entry["destination_path"] return Skill( - name=f"{universe_id}-files", + name=_files_skill_name(universe_id), description=f"Files possibly relevant to the current task are available at {destination_path}.", body=f"Files possibly relevant to the current task are available at `{destination_path}`.\n", ) 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 36a9c08e..e8fbd3f9 100644 --- a/src/agent_env/task_step/task_steps/collect_artifacts.py +++ b/src/agent_env/task_step/task_steps/collect_artifacts.py @@ -68,6 +68,7 @@ import asyncio import functools +import hashlib import logging import os import json @@ -77,7 +78,9 @@ import time from typing import ClassVar, Optional +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.task_step.context import TaskStepContext from agent_env.entity_refs import EntityRef from agent_env.task_step.task_step import TaskStep, TaskStepDependency @@ -142,11 +145,6 @@ def _is_url_entry(entry: str) -> bool: } -def _sanitize_artifact_id(value: str) -> str: - """Artifact ids become S3 key components, so spaces and separators can't survive.""" - return value.replace(" ", "-").replace("/", "_") - - def _exec_args(sandbox, container: Optional[str], cmd: tuple) -> tuple: """Wrap a command for where it has to run: inside the container, on the VM host as root, or as-is.""" from agent_env.providers.sandbox_providers.sandbox_provider import SANDBOX_MODE_VM @@ -985,7 +983,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 = f"{artifact_id}-{object_name}".replace("/", "_")[:200] + fa_id = derive_id(artifact_id, hashlib.sha256(object_name.encode("utf-8")).hexdigest()[:16]) fa_version = store.next_version(fa_id) fa = FileArtifact( id=fa_id, @@ -1011,11 +1009,11 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: context.metadata.get("universe_id") or context.instance_id or "unknown" - ) - # Reserve room for the suffix inside the 100-char budget rather than appending past - # it, so a long run id can never truncate two collects back onto the same name. - suffix = _sanitize_artifact_id(self.universe_id_suffix or "") - artifact_id = _sanitize_artifact_id(artifact_id)[:100 - len(suffix)] + suffix + ) + (self.universe_id_suffix or "") + # Refused before the sandbox is read: past here a write that fails is only logged. + if is_local_id(artifact_id): + validate_local_id(artifact_id) + get_config().check_local_run_write(artifact_id) version = int(time.time()) if self.env_id: diff --git a/src/agent_env/task_step/task_steps/mcp_cli_builder/build_mcp_cli.py b/src/agent_env/task_step/task_steps/mcp_cli_builder/build_mcp_cli.py index 23f226f9..cfee11c0 100644 --- a/src/agent_env/task_step/task_steps/mcp_cli_builder/build_mcp_cli.py +++ b/src/agent_env/task_step/task_steps/mcp_cli_builder/build_mcp_cli.py @@ -18,6 +18,7 @@ from agent_env.artifact.store import artifact_write_lock from agent_env.env import Env from agent_env.env.env import require_gateway_url +from agent_env.store.ids import derive_id from agent_env.task_step.context import TaskStepContext from agent_env.entity_refs import EntityRef, RefRole from agent_env.task_step.task_step import TaskStep, TaskStepDependency @@ -137,7 +138,7 @@ def __init__( super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error) self.env_id = env_id self.command_name = command_name - self.cli_artifact_id = cli_artifact_id or f"cli-{env_id}" + self.cli_artifact_id = cli_artifact_id or derive_id(env_id, "cli") def to_dict(self) -> dict: base = super().to_dict() 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 fb2be3e1..fc69eb0c 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 @@ -11,6 +11,7 @@ from agent_env.env.env_artifact_store import EnvArtifactType, get_env_artifact_store from agent_env.artifact.store import artifact_write_lock +from agent_env.store.ids import derive_id, fs_safe, is_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 @@ -77,11 +78,21 @@ def from_dict(cls, data: dict) -> VerifyUniverseLoadExportRoundtripStep: persist_result=data.get("persist_result", True), ) + @staticmethod + def validation_id(kind: str, env_id: str, env_version: int, universe_id: str, universe_version: int) -> str: + """The id of a compatibility validation's ``kind`` of task or artifact for ``universe_id`` in ``env_id``. + The env owns it, or the universe when only the universe is ``@local``, and the other id goes into its + suffix, through ``fs_safe`` when it is ``@local``.""" + if is_local_id(universe_id) and not is_local_id(env_id): + return derive_id(universe_id, f"{kind}-v{universe_version}-{env_id}-v{env_version}") + return derive_id(env_id, f"{kind}-v{env_version}-{fs_safe(universe_id)}-v{universe_version}") + @staticmethod def file_artifact_universe_id(env_id: str, env_version: int, universe_id: str, universe_version: int) -> str: """Deterministic id of the emitted FileArtifactUniverse, so the callsite can reference it statically without a runtime context lookup (mirrors the export1 artifact id scheme).""" - return f"validate-{env_id}-v{env_version}-{universe_id}-v{universe_version}-fau" + prefix = VerifyUniverseLoadExportRoundtripStep.validation_id("validate", env_id, env_version, universe_id, universe_version) + return f"{prefix}-fau" async def execute(self, context: TaskStepContext) -> TaskStepContext: from agent_env.artifact import EnvironmentUniverseArtifact @@ -178,7 +189,7 @@ def _create_file_artifact_universe(self, original_environment_artifacts: list, e """ from agent_env.artifact import FileArtifact, FileArtifactUniverse - prefix = f"validate-{self.env_id}-v{env_version}-{self.universe_artifact_id}-v{universe_version}" + prefix = self.validation_id("validate", self.env_id, env_version, self.universe_artifact_id, universe_version) with artifact_write_lock(prefix): file_artifacts: dict[str, Any] = {} for sa in original_environment_artifacts: @@ -221,7 +232,7 @@ async def _export_all(deployed: DeployedEnv, environment_names: list[str]) -> di def _create_universe_artifact(self, original_environment_artifacts: list, export_data: dict[str, dict], env_version: int, universe_version: int) -> Any: """Create FileArtifact + EnvironmentArtifact per service, bundle into EnvironmentUniverseArtifact.""" from agent_env.artifact import FileArtifact, EnvironmentArtifact, EnvironmentUniverseArtifact - prefix = f"validate-{self.env_id}-v{env_version}-{self.universe_artifact_id}-v{universe_version}" + prefix = self.validation_id("validate", self.env_id, env_version, self.universe_artifact_id, universe_version) with artifact_write_lock(prefix): export_environment_artifacts = [] for sa in original_environment_artifacts: diff --git a/src/agent_env/task_step/task_steps/snapshot_env.py b/src/agent_env/task_step/task_steps/snapshot_env.py index 410825eb..f24a3a4c 100644 --- a/src/agent_env/task_step/task_steps/snapshot_env.py +++ b/src/agent_env/task_step/task_steps/snapshot_env.py @@ -20,6 +20,7 @@ from agent_env.env.gateway.constants import EXT_TRAJECTORY_URI from agent_env.store import get_config +from agent_env.store.ids import derive_id, is_local_id, validate_local_id from agent_env.store.object_store import S3ObjectStore from agent_env.store.routing import in_local_run from agent_env.task_step.context import TaskStepContext @@ -166,6 +167,10 @@ def from_dict(cls, data: dict) -> SnapshotEnvTaskStep: async def execute(self, context: TaskStepContext) -> TaskStepContext: from agent_env.env.env import Env, gateway_url_of + snapshot_id = self._derive_snapshot_id(context) + # Refused before any export: past here a write that fails only skips its service. + if is_local_id(snapshot_id): + validate_local_id(snapshot_id) deployed = self._resolve_deployed_env(context) env_id = self.env_id or (deployed.env_id if deployed else None) if not env_id: @@ -191,7 +196,6 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: env_version = deployed.env_version if deployed else None env = await asyncio.to_thread(Env.get, env_id, env_version) - snapshot_id = self._derive_snapshot_id(context, env_id) result = await self.snapshot_env_state( env=env, gateway_url=gateway_url, @@ -307,14 +311,14 @@ async def _capture_env_trajectory(self, entry: dict, context: TaskStepContext, e entry["object_url"] = object_url entry["event_count"] = event_count - def _derive_snapshot_id(self, context: TaskStepContext, env_id: str) -> str: + def _derive_snapshot_id(self, context: TaskStepContext) -> str: if self.snapshot_id: return self.snapshot_id instance_id = context.instance_id or context.metadata.get("instance_id") if instance_id: # Stable per run instance: activity retries re-put the same ids. - return f"snapshot-{env_id}-{instance_id.rsplit('-', 1)[-1][:16]}" - generated = f"snapshot-{env_id}-{uuid.uuid4().hex[:8]}" + return derive_id(instance_id, f"snapshot-{self.id}") + generated = derive_id(f"adhoc-{uuid.uuid4().hex[:12]}", f"snapshot-{self.id}") logger.warning( f"snapshot_env: no instance_id in context; using random snapshot_id " f"{generated} (retries will not be idempotent)" @@ -455,8 +459,7 @@ async def snapshot_env_state( services = cls._enumerate_environments(env) gateway = gateway_url.rstrip("/") - # `gateway` identifies which deployment was exported: with k deployments of - # one env_id, env= and snapshot_id= are identical across branches. + # `gateway` identifies which deployment was exported: env= alone doesn't when one env_id has several. logger.info( f"snapshot_env: env={env.id} v{env.version} snapshot_id={snapshot_id} " f"gateway={gateway} services={services}" @@ -491,7 +494,7 @@ async def export_one(environment_name: str): """Returns EnvironmentArtifact on success, (public_err, log_err) on failure.""" async with sem: # Suffix (.json/.zip) is reported by _export_environment_to_file. - fd, tmp_path = tempfile.mkstemp(prefix=f"{snapshot_id}-{environment_name}-") + fd, tmp_path = tempfile.mkstemp(prefix=f"{environment_name}-") os.close(fd) cleanup_paths = [tmp_path] artifact_path = tmp_path diff --git a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py index 4d867abe..f3764a09 100644 --- a/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py +++ b/src/agent_env/task_step/task_steps/verifiers/run_container_unit_tests_verifier.py @@ -37,6 +37,7 @@ all_sandbox_container_env, registered_sandbox_provider_classes, ) +from agent_env.store.ids import derive_id, is_local_id, key_segment, validate_local_id from agent_env.task_step.context import TaskStepContext from agent_env.task_step.task_step import TaskStep, TaskStepDependency @@ -205,6 +206,10 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: ) from agent_env.config import ConfigError, get_config + # The outputs are named after the run: refuse an id they can't take before the tests run. + if context.instance_id and is_local_id(context.instance_id): + validate_local_id(derive_id(context.instance_id, f"verifier-stdout-{self.id}")) + # 1. Lookup container + sandbox, rehydrate VmSandbox containers = context.metadata.get("deployed_docker_containers", []) container_entry = next( @@ -298,20 +303,22 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext: # 7. Persist stdout + stderr as separate FileArtifacts # Every execution gets its own key: a retried step keeps its run's instance id, and the # outputs are write-once. The instance id and the time only group and order them. - run = "-".join(filter(None, (context.instance_id, str(int(time.time())), uuid.uuid4().hex[:12]))) + stamp = f"{int(time.time())}-{uuid.uuid4().hex[:12]}" + run = "-".join(filter(None, (context.instance_id and key_segment(context.instance_id), stamp))) + base = context.instance_id or f"adhoc-{uuid.uuid4().hex[:12]}" store = config.get_object_store() outputs = f"{config.get_artifact_key_prefix()}verifier-outputs/{self.id}/{run}" stdout_artifact = await asyncio.to_thread( self._upload_text_artifact, text=stdout, - artifact_id=f"verifier-stdout-{self.id}-{run}", + artifact_id=derive_id(base, f"verifier-stdout-{self.id}-{stamp}"), description=f"stdout of {self.verifier_id} from step {self.id}", s3_url=store.object_url(f"{outputs}/stdout.txt"), ) stderr_artifact = await asyncio.to_thread( self._upload_text_artifact, text=stderr, - artifact_id=f"verifier-stderr-{self.id}-{run}", + artifact_id=derive_id(base, f"verifier-stderr-{self.id}-{stamp}"), description=f"stderr of {self.verifier_id} from step {self.id}", s3_url=store.object_url(f"{outputs}/stderr.txt"), ) diff --git a/tst/integration/cli/run_bundle_test.py b/tst/integration/cli/run_bundle_test.py index b2569c89..7d95138c 100644 --- a/tst/integration/cli/run_bundle_test.py +++ b/tst/integration/cli/run_bundle_test.py @@ -19,13 +19,17 @@ from click.testing import CliRunner import agent_env +from agent_env.artifact import FileArtifactUniverse from agent_env.artifact.store import reset_artifact_store from agent_env.bundle.installed import find_bundle from agent_env.cli import cli from agent_env.config import configure, reset_config +from agent_env.config.paths import state_root from agent_env.providers.sandbox_providers.sandbox_provider import reset_sandbox_provider +from agent_env.store import Filter, LocalSqliteDocumentStore from agent_env.store.routing import namespace_routing from agent_env.task import Task +from tst.util.capabilities import skip_without_gnu_stat pytestmark = pytest.mark.integration @@ -122,6 +126,24 @@ def test_a_bundle_runs_from_its_folder_and_a_rerun_writes_only_what_changed(bund assert [(result["criterion"], result["result"]) for result in results] == [("greets", False), ("check passes", False)] +@skip_without_gnu_stat() +def test_a_bundle_task_collects_from_its_sandbox_into_a_universe_named_after_the_run(bundle, quiet_logs): + steps = json.loads((bundle / "tasks/hello.json").read_text()) + steps.append({"id": "collect", "type": "collect_artifacts", "sandbox_name": "box", "base_path": "/app/greeting", + "artifact_paths": ["hello.txt"]}) + (bundle / "tasks/hello.json").write_text(json.dumps(steps)) + + output = _run(bundle) + + assert "tasks/hello.json v1: passed (hello: 1)" in output + (instance_id,) = re.findall(r"instance (@local/~/my-hello/hello-[a-z0-9]{8})\b", output) + with namespace_routing(): + universe = FileArtifactUniverse.get(instance_id) + assert {name: fa.load() for name, fa in universe.get_file_artifacts().items()} == {"hello.txt": b"hello\n"} + documents = LocalSqliteDocumentStore(str(state_root() / "document_store" / "documents.db")) + assert not documents.path.exists() or documents.count("artifacts", Filter()) == 0 + + def test_an_installed_bundle_runs_by_name_with_ids_rooted_at_its_package(bundle, monkeypatch, tmp_path, quiet_logs): """A real .dist-info registers the folder through agent_env.bundles; nothing is imported to find it. agent-env ships a hello too, so this one runs by its qualified name.""" diff --git a/tst/integration/env/task_step/task_steps_test.py b/tst/integration/env/task_step/task_steps_test.py index 3e79465e..ec31ba70 100644 --- a/tst/integration/env/task_step/task_steps_test.py +++ b/tst/integration/env/task_step/task_steps_test.py @@ -793,7 +793,7 @@ async def exec_in(sandbox, wrapper_cmd, *cmd): build_step = BuildMcpCliTaskStep(id=f"cli-e2e-build-{suffix}", version=None, env_id=multi_env.id, command_name=multi_env.id) await build_step.execute(context) cli_ref = context.metadata["cli_artifact"] - assert cli_ref["id"] == f"cli-{multi_env.id}" + assert cli_ref["id"] == f"{multi_env.id}__cli" logger.info(f"CLI e2e: built CliArtifact {cli_ref['id']} v{cli_ref['version']}") load_step = LoadArtifactTaskStep( diff --git a/tst/unit/bundle/resolve_test.py b/tst/unit/bundle/resolve_test.py index eed82765..818ae1be 100644 --- a/tst/unit/bundle/resolve_test.py +++ b/tst/unit/bundle/resolve_test.py @@ -149,10 +149,17 @@ def test_an_output_is_named_under_its_task_and_later_steps_find_it(make): def test_an_output_id_a_step_derives_is_named_under_its_task_too(make): entry = resolved(make(tasks={"t": [ {"id": "cli", "type": "build_mcp_cli", "env_id": "tickets", "command_name": "tickets"}, - {"id": "skills", "type": "add_skills", "agent_name": "solver", "cli_artifact_ids": ["cli-tickets"]}, + {"id": "skills", "type": "add_skills", "agent_name": "solver", "cli_artifact_ids": ["tickets__cli"]}, ]}), "t") - assert entry.config[0]["cli_artifact_id"] == entry.config[1]["cli_artifact_ids"][0] == f"{ROOT}/t/cli-tickets" - assert [output.id for output in entry.outputs] == [f"{ROOT}/t/cli-tickets"] + assert entry.config[0]["cli_artifact_id"] == entry.config[1]["cli_artifact_ids"][0] == f"{ROOT}/t/tickets__cli" + assert [output.id for output in entry.outputs] == [f"{ROOT}/t/tickets__cli"] + + +def test_an_output_id_derived_from_another_bundles_entity_is_named_under_this_task(make): + entry = resolved(make(tasks={"t": [ + {"id": "cli", "type": "build_mcp_cli", "env_id": "@local/~/crm-suite/envs/crm", "command_name": "crm"}, + ]}), "t") + assert [output.id for output in entry.outputs] == [entry.config[0]["cli_artifact_id"]] == [f"{ROOT}/t/crm__cli"] def test_a_resolved_task_still_builds_its_steps(make): diff --git a/tst/unit/config/test_fixture_prefix_keys.py b/tst/unit/config/test_fixture_prefix_keys.py index 0b19a3e8..3c99586b 100644 --- a/tst/unit/config/test_fixture_prefix_keys.py +++ b/tst/unit/config/test_fixture_prefix_keys.py @@ -12,22 +12,14 @@ from agent_env.artifact.store import reset_artifact_store from agent_env.config import configure, get_config, set_object_store from agent_env.env.snapshot_store import _snapshot_servicedb -from agent_env.store.object_store.local.store import LocalFilesystemObjectStore from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep from agent_env.task_step.task_steps.verifiers.judge_utils.trajectory_filter import TrajectoryFilter from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep +from tst.unit.store.fakes import SigningObjectStore _AGENT = SimpleNamespace(id="solver", version=3) -class _SigningStore(LocalFilesystemObjectStore): - def signed_get_url(self, object_url: str, expires_in: int = 3600) -> str | None: - return f"https://objects.example.test/{self.get_object_key(object_url)}" - - def signed_put_url(self, object_url: str, expires_in: int = 3600) -> str | None: - return f"https://objects.example.test/{self.get_object_key(object_url)}" - - @pytest.fixture def prefixed(local_stores, monkeypatch): monkeypatch.setenv("AGENT_ENV_FIXTURE_PREFIX", "fx") @@ -58,7 +50,7 @@ def test_a_cli_bundle_is_written_under_the_prefix(prefixed, tmp_path): def test_the_validator_fixtures_are_written_under_the_prefix(prefixed, tmp_path): - set_object_store(_SigningStore(str(tmp_path / "signing"))) + set_object_store(SigningObjectStore(str(tmp_path / "signing"))) store = get_config().get_object_store() skill_url = A2AAgentValidator._upload_skill_fixture(_AGENT, name="probe", description="d", body="b") assert skill_url == store.object_url("fx/a2a_validator/validator_skill/solver-v3/probe/") @@ -90,7 +82,7 @@ async def exec_script(self, script: str) -> str: def test_an_env_snapshot_is_uploaded_under_the_prefix(prefixed, tmp_path): - set_object_store(_SigningStore(str(tmp_path / "signing"))) + set_object_store(SigningObjectStore(str(tmp_path / "signing"))) sandbox = _Sandbox() url = asyncio.run(_snapshot_servicedb(sandbox, "slack-env", "acme", lambda *args: None)) key = "fx/env-snapshots/slack-env/acme/env-snapshot-slack-env-acme.tar.gz" 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 784ebbee..1308cdd0 100644 --- a/tst/unit/env/envs/test_mcp_server_container_mode.py +++ b/tst/unit/env/envs/test_mcp_server_container_mode.py @@ -251,8 +251,7 @@ async def test_a_restored_env_without_a_gateway_loads_into_the_server_and_closes async def test_validation_deploys_a_server_env_as_declared_on_the_default_sandbox_provider(): env, seen, default = _env("server"), {}, MagicMock() ran = MagicMock(run=AsyncMock(return_value=MagicMock(deployed_envs=[], deployed_agents=[], instance_id="i"))) - with patch("agent_env.task.Task.put", side_effect=lambda **kwargs: seen.update(kwargs) or ran), \ - patch("agent_env.env.envs.mcp_server.refuse_local_derivation"): + with patch("agent_env.task.Task.put", side_effect=lambda **kwargs: seen.update(kwargs) or ran): await env.validate() deploy_step = seen["steps"][0] diff --git a/tst/unit/store/derived_ids_test.py b/tst/unit/store/derived_ids_test.py new file mode 100644 index 00000000..cbd1e118 --- /dev/null +++ b/tst/unit/store/derived_ids_test.py @@ -0,0 +1,372 @@ +"""The ids core names after an entity or a run: ``derive_id(base, suffix)``, for a bare base and an ``@local`` one +alike, so the base's namespace carries over. Consumers rebuild two of these spellings to link a validation to its +task, ``__validate-v`` and ``__validate-universe-compat-v--v``, so every table here +pins its bare row.""" + +import asyncio +import time +import uuid +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from click.testing import CliRunner + +import agent_env.providers.sandbox_providers.sandbox_provider as sandbox_provider +from agent_env.a2a_agent.a2a_agent import A2AAgent +from agent_env.a2a_agent.validator import A2AAgentValidator +from agent_env.artifact import DockerImageArtifact, FileArtifactUniverse +from agent_env.cli import cli +from agent_env.config import get_config, set_object_store +from agent_env.env.env import DeployedGatewayEnv +from agent_env.env.envs.mcp_server import MCPServerEnv +from agent_env.env.envs.multi_env import MultiEnv +from agent_env.env.envs.website import WebsiteEnv +from agent_env.env.snapshot_store import _DOCKER_NAME, EnvSnapshot, _image_tag +from agent_env.task import Task +from agent_env.task_step import VerifyUniverseLoadExportRoundtripStep +from agent_env.task_step.context import DeployedSandbox, TaskStepContext +from agent_env.task_step.snapshot_utils.snapshot_series import SnapshotConfig, SnapshotSeries +from agent_env.task_step.task_steps.add_skills import _build_skill_for_loaded_file_artifact_universe +from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep +from agent_env.task_step.task_steps.snapshot_env import SnapshotEnvTaskStep +from agent_env.task_step.task_steps.verifiers import run_container_unit_tests_verifier as verifier_module +from tst.unit.store.fakes import CollectingVm, SigningObjectStore, reattach_for_snapshot + +LOCAL = "@local/~/work/triage" +LOCAL_ENV = f"{LOCAL}/envs/tickets" +LOCAL_UNIVERSE = f"{LOCAL}/artifacts/seed" +LOCAL_AGENT = f"{LOCAL}/agents/solver" +LOCAL_TASK = f"{LOCAL}/tasks/t" +BARE_RUN = "triage-1-abcd1234" +LOCAL_RUN = f"{LOCAL_TASK}-abcd1234" +# fs_safe and key_segment of LOCAL_ENV and LOCAL_UNIVERSE, and key_segment of LOCAL_RUN. +LOCAL_ENV_FILENAME, LOCAL_ENV_KEY = "local-work-triage-envs-tickets-ca64d3fb6d09", "local/work-triage-envs-tickets-ca64d3fb6d09" +LOCAL_UNIVERSE_FILENAME, LOCAL_UNIVERSE_KEY = ( + "local-work-triage-artifacts-seed-ddf2904a1148", "local/work-triage-artifacts-seed-ddf2904a1148", +) +LOCAL_RUN_KEY = "local/work-triage-tasks-t-abcd1234-d34bacfa44b2" + + +class _Written(Exception): + """Raised by a patched write once it has what the code under test writes.""" + + +def _written_by(run, monkeypatch, owner=Task, method="put") -> dict: + """The keywords of the one ``owner.method`` call ``run`` makes, which stops there.""" + written = [] + + def write(cls, **kwargs): + written.append(kwargs) + raise _Written + + monkeypatch.setattr(owner, method, classmethod(write)) + with pytest.raises(_Written): + asyncio.run(run()) + (kwargs,) = written + return kwargs + + +_VALIDATIONS = { + "mcp_server": lambda id: MCPServerEnv(id=id, version=3, docker_image_artifact=MagicMock(), environment_name="slack").validate(), + "website": lambda id: WebsiteEnv(id=id, version=3, backend_docker_image_artifact=MagicMock(), + frontend_docker_image_artifact=MagicMock(), environment_name="shop").validate(), + "multi": lambda id: MultiEnv(id=id, version=3, mcp_server_envs=[]).validate(), + "a2a_agent": lambda id: A2AAgentValidator.validate(A2AAgent(id=id, version=22, docker_image_artifact=MagicMock())), +} + + +@pytest.mark.parametrize("kind, entity_id, task_id", [ + ("mcp_server", "slack-mcp", "slack-mcp__validate-v3"), + ("mcp_server", LOCAL_ENV, f"{LOCAL_ENV}__validate-v3"), + ("website", "shop", "shop__validate-v3"), + ("website", LOCAL_ENV, f"{LOCAL_ENV}__validate-v3"), + ("multi", "crm-suite", "crm-suite__validate-v3"), + ("multi", LOCAL_ENV, f"{LOCAL_ENV}__validate-v3"), + ("a2a_agent", "claude-code", "claude-code__validate-a2a-v22"), + ("a2a_agent", LOCAL_AGENT, f"{LOCAL_AGENT}__validate-a2a-v22"), +], ids=["mcp_server-bare", "mcp_server-local", "website-bare", "website-local", "multi-bare", "multi-local", + "a2a_agent-bare", "a2a_agent-local"]) +def test_a_validation_task_is_named_after_its_entity(local_stores, tmp_path, monkeypatch, kind, entity_id, task_id): + set_object_store(SigningObjectStore(str(tmp_path / "signing"))) + assert _written_by(lambda: _VALIDATIONS[kind](entity_id), monkeypatch)["id"] == task_id + + +@pytest.mark.parametrize("env_id, universe_id, task_id, fau_id", [ + ("crm-suite", "crm-universe", + "crm-suite__validate-universe-compat-v3-crm-universe-v2", "crm-suite__validate-v3-crm-universe-v2-fau"), + (LOCAL_ENV, "crm-universe", + f"{LOCAL_ENV}__validate-universe-compat-v3-crm-universe-v2", f"{LOCAL_ENV}__validate-v3-crm-universe-v2-fau"), + (LOCAL_ENV, LOCAL_UNIVERSE, + f"{LOCAL_ENV}__validate-universe-compat-v3-{LOCAL_UNIVERSE_FILENAME}-v2", + f"{LOCAL_ENV}__validate-v3-{LOCAL_UNIVERSE_FILENAME}-v2-fau"), + ("crm-suite", LOCAL_UNIVERSE, + f"{LOCAL_UNIVERSE}__validate-universe-compat-v2-crm-suite-v3", f"{LOCAL_UNIVERSE}__validate-v2-crm-suite-v3-fau"), +], ids=["bare", "local-env", "both-local", "local-universe"]) +def test_a_compatibility_validation_is_named_after_its_env_unless_only_the_universe_is_local( + monkeypatch, env_id, universe_id, task_id, fau_id): + universe = SimpleNamespace(id=universe_id, version=2, get_environment_artifacts=lambda: []) + monkeypatch.setattr("agent_env.artifact.EnvironmentUniverseArtifact.get", lambda *args: universe) + + task = _written_by(lambda: MultiEnv(id=env_id, version=3, mcp_server_envs=[]).validate_universe_compatibility(universe_id), + monkeypatch) + + (load,) = [step for step in task["steps"] if step.type == "load_artifact"] + assert (task["id"], [artifact["id"] for artifact in load.artifacts]) == (task_id, [fau_id]) + + +@pytest.mark.parametrize("env_id, export", [ + ("crm-suite", "crm-suite__validate-v3-crm-universe-v7-export1"), + (LOCAL_ENV, f"{LOCAL_ENV}__validate-v3-crm-universe-v7-export1"), +], ids=["bare", "local"]) +def test_roundtrip_exports_are_named_after_the_validation(local_stores, cli_routing, env_id, export): + step = VerifyUniverseLoadExportRoundtripStep(id="rt", version=None, env_id=env_id, universe_artifact_id="crm-universe") + + universe = step._create_universe_artifact([SimpleNamespace(environment_name="slack")], {"slack": {"messages": []}}, + env_version=3, universe_version=7) + + (service,) = universe.get_environment_artifacts() + assert (universe.id, service.id, service.get_file_artifact().id) == (export, f"{export}-svc-slack", f"{export}-slack") + + +@pytest.mark.parametrize("env_id, task_id, cli_id", [ + ("slack-mcp", "slack-mcp__create-cli-v3", "slack-mcp__cli"), + (LOCAL_ENV, f"{LOCAL_ENV}__create-cli-v3", f"{LOCAL_ENV}__cli"), +], ids=["bare", "local"]) +def test_create_cli_names_its_task_and_cli_after_the_env(monkeypatch, env_id, task_id, cli_id): + env = MCPServerEnv(id=env_id, version=3, docker_image_artifact=MagicMock(), environment_name="slack") + + task = _written_by(lambda: env.create_cli(force=True), monkeypatch) + + assert (task["id"], task["steps"][1].cli_artifact_id) == (task_id, cli_id) + + +@pytest.mark.parametrize("agent_id, universe_id", [ + ("claude-code", "claude-code__validate-install-image-v22-1790000000"), + (LOCAL_AGENT, f"{LOCAL_AGENT}__validate-install-image-v22-1790000000"), +], ids=["bare", "local"]) +def test_the_install_image_universe_is_named_after_the_agent(local_stores, cli_routing, monkeypatch, agent_id, universe_id): + monkeypatch.setattr(time, "time", lambda: 1_790_000_000.0) + agent = A2AAgent(id=agent_id, version=22, docker_image_artifact=MagicMock()) + + assert A2AAgentValidator._upload_install_test_image_fixture(agent).id == universe_id + + +@pytest.mark.parametrize("env_id, universe_id, artifact_id, image_tag, key", [ + ("crm-suite", "crm-universe", "crm-suite__env-snapshot", "env-snapshot-crm-suite-crm-universe", + "env-snapshots/crm-suite/crm-universe/env-snapshot-crm-suite-crm-universe.tar.gz"), + (LOCAL_ENV, LOCAL_UNIVERSE, f"{LOCAL_ENV}__env-snapshot", f"env-snapshot-{LOCAL_ENV_FILENAME}-{LOCAL_UNIVERSE_FILENAME}", + f"env-snapshots/{LOCAL_ENV_KEY}/{LOCAL_UNIVERSE_KEY}/env-snapshot-{LOCAL_ENV_FILENAME}-{LOCAL_UNIVERSE_FILENAME}.tar.gz"), +], ids=["bare", "local"]) +def test_an_env_snapshot_is_named_after_its_env(local_stores, tmp_path, monkeypatch, env_id, universe_id, artifact_id, + image_tag, key): + set_object_store(SigningObjectStore(str(tmp_path / "signing"))) + sandbox = reattach_for_snapshot(monkeypatch, env_id, 3, universe_id, 2) + + image = _written_by(lambda: EnvSnapshot.create("inst-1"), monkeypatch, DockerImageArtifact, "put_tar") + + assert (image["id"], image["image_name"]) == (artifact_id, image_tag) + assert image["tar_gz_s3_url"].endswith(f"/{key}") + assert f"docker build --platform linux/amd64 -t {image_tag} /tmp/snapshot-build" in sandbox.scripts + + +@pytest.mark.parametrize("universe_id, image_tag", [ + ("crm-universe", "env-snapshot-crm-suite-crm-universe"), + (f"crm-run-{'a' * 210}__snapshot-snap", "env-snapshot-crm-suite-16d9f6db8bf8"), + ("Crm-Run-abcd1234__snapshot-snap", "env-snapshot-crm-suite-a655c6c500a1"), +], ids=["short", "233-chars", "mixed-case"]) +def test_an_env_snapshot_image_name_is_a_docker_name_of_at_most_200_chars(universe_id, image_tag): + tag = _image_tag("crm-suite", universe_id) + assert tag == image_tag + assert _DOCKER_NAME.fullmatch(tag) + assert len(tag) <= 200 + + +@pytest.mark.parametrize("instance_id, snapshot_id", [ + (BARE_RUN, "triage-1-abcd1234__snapshot-snap"), + (LOCAL_RUN, f"{LOCAL_RUN}__snapshot-snap"), +], ids=["bare", "local"]) +def test_an_env_snapshot_step_snapshots_into_a_universe_named_after_the_run(instance_id, snapshot_id): + step = SnapshotEnvTaskStep(id="snap", version=None, env_id="crm-suite") + assert step._derive_snapshot_id(TaskStepContext(instance_id=instance_id)) == snapshot_id + + +def _series(instance_id: str) -> SnapshotSeries: + return SnapshotSeries(step_id="solve", agent_name="solver", prompt_id="p1", a2a_context_id="c1", + config=SnapshotConfig(env_id="crm-suite"), trajectory_output_prefix="unused", instance_id=instance_id) + + +def _captured_snapshot_id(series: SnapshotSeries, monkeypatch) -> str: + """The universe ``series`` snapshots its env into on a capture.""" + snapshot_ids = [] + + async def snapshot_env_state(**kwargs): + snapshot_ids.append(kwargs["snapshot_id"]) + return SimpleNamespace(environment_universe_artifact_id=kwargs["snapshot_id"], environment_universe_artifact_version=1) + + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *args: SimpleNamespace(id="crm-suite")) + monkeypatch.setattr(SnapshotEnvTaskStep, "snapshot_env_state", staticmethod(snapshot_env_state)) + deployed = DeployedGatewayEnv(env_id="crm-suite", env_version=1, gateway_url="https://gw", mcp_url="https://gw/mcp", + db_web_url=None, sandbox_id="sb-1") + assert asyncio.run(series._capture_env_state(TaskStepContext(deployed_envs=[deployed]), {}, 30)) is None + (snapshot_id,) = snapshot_ids + return snapshot_id + + +@pytest.mark.parametrize("instance_id, workspace_id, snapshot_id", [ + (BARE_RUN, "triage-1-abcd1234__solve-workspace", "triage-1-abcd1234__snapshot-solve"), + (LOCAL_RUN, f"{LOCAL_RUN}__solve-workspace", f"{LOCAL_RUN}__snapshot-solve"), +], ids=["bare", "local"]) +def test_a_capture_series_names_its_artifacts_after_the_run(monkeypatch, instance_id, workspace_id, snapshot_id): + series = _series(instance_id) + assert (series.workspace_artifact_id, _captured_snapshot_id(series, monkeypatch)) == (workspace_id, snapshot_id) + + +def test_an_env_snapshot_step_and_a_capture_series_in_one_run_snapshot_one_env_into_different_universes(monkeypatch): + step = SnapshotEnvTaskStep(id="snap", version=None, env_id="crm-suite") + from_step = step._derive_snapshot_id(TaskStepContext(instance_id=BARE_RUN)) + assert from_step != _captured_snapshot_id(_series(BARE_RUN), monkeypatch) + + +class _VerifierSandbox: + async def exec_script(self, script: str) -> str: + return "" + + async def exec_with_output(self, *args) -> tuple[int, str, str]: + return 0, "ok", "" + + +@pytest.mark.parametrize("instance_id, base, run", [ + (BARE_RUN, "triage-1-abcd1234", "triage-1-abcd1234-1790000000-0123456789ab"), + (LOCAL_RUN, LOCAL_RUN, f"{LOCAL_RUN_KEY}-1790000000-0123456789ab"), + (None, "adhoc-0123456789ab", "1790000000-0123456789ab"), +], ids=["bare", "local", "no-instance"]) +def test_verifier_outputs_are_named_and_keyed_after_the_run(local_stores, monkeypatch, instance_id, base, run): + provider = SimpleNamespace(get_sandbox=lambda sandbox_id: asyncio.sleep(0, result=_VerifierSandbox())) + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: provider) + monkeypatch.setattr(verifier_module.time, "time", lambda: 1_790_000_000.0) + monkeypatch.setattr(verifier_module.uuid, "uuid4", lambda: uuid.UUID(hex="0123456789ab" + "0" * 20)) + uploads: list[tuple[str, str]] = [] + + def record(text, artifact_id, description, s3_url): + uploads.append((artifact_id, s3_url)) + return SimpleNamespace(id=artifact_id, version=1, object_url=s3_url) + + monkeypatch.setattr(verifier_module.RunContainerUnitTestsVerifierTaskStep, "_upload_text_artifact", staticmethod(record)) + ctx = TaskStepContext(instance_id=instance_id) + ctx.metadata["deployed_docker_containers"] = [{"container_name": "c", "sandbox_name": "h"}] + ctx.deployed_sandboxes.append(DeployedSandbox(sandbox_name="h", sandbox_id="sb-1", sandbox_mode="vm")) + step = verifier_module.RunContainerUnitTestsVerifierTaskStep(id="scrape", version=None, sandbox_name="h", container_name="c", command="true") + + asyncio.run(step.execute(ctx)) + + outputs = get_config().get_object_store().object_url(f"verifier-outputs/scrape/{run}") + assert uploads == [(f"{base}__verifier-{stream}-scrape-1790000000-0123456789ab", f"{outputs}/{stream}.txt") + for stream in ("stdout", "stderr")] + + +@pytest.mark.parametrize("instance_id", [BARE_RUN, LOCAL_RUN], ids=["bare", "local"]) +def test_collected_files_are_named_after_the_run(local_stores, cli_routing, monkeypatch, instance_id): + provider = SimpleNamespace(get_sandbox=lambda sandbox_id: asyncio.sleep(0, result=CollectingVm()), + close=lambda: asyncio.sleep(0)) + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: provider) + ctx = TaskStepContext(instance_id=instance_id) + ctx.deployed_sandboxes = [DeployedSandbox(sandbox_name="box", sandbox_id="sb-1", sandbox_mode="vm")] + step = CollectArtifactsTaskStep(id="collect", version=None, sandbox_name="box", base_path="/app/artifact", + artifact_paths=["report.pdf", "sub/notes.md"]) + + asyncio.run(step.execute(ctx)) + + universe = FileArtifactUniverse.get(ctx.metadata["file_artifact_universe"]["id"]) + assert universe.id == instance_id + assert universe.file_artifact_ids == { + "report.pdf": f"{instance_id}__6466e450a16b77b8", "sub/notes.md": f"{instance_id}__d72324ebb0d7e97a", + } + + +def test_a_seed_universe_longer_than_a_filename_collects_on_the_local_store(local_stores, monkeypatch): + provider = SimpleNamespace(get_sandbox=lambda sandbox_id: asyncio.sleep(0, result=CollectingVm()), + close=lambda: asyncio.sleep(0)) + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: provider) + universe_id = "triage__v2-" + ("Escalated tickets from the EU desk, " * 8).rstrip(", ") + ctx = TaskStepContext(instance_id=BARE_RUN) + ctx.metadata["universe_id"] = universe_id + ctx.deployed_sandboxes = [DeployedSandbox(sandbox_name="box", sandbox_id="sb-1", sandbox_mode="vm")] + step = CollectArtifactsTaskStep(id="collect", version=None, sandbox_name="box", base_path="/app/artifact", + artifact_paths=["report.pdf", "sub/notes.md"]) + + asyncio.run(step.execute(ctx)) + + universe = FileArtifactUniverse.get(universe_id) + assert {name: fa.load() for name, fa in universe.get_file_artifacts().items()} == {"report.pdf": b"%PDF", "sub/notes.md": b"# notes"} + + +@pytest.mark.parametrize("universe_id, name", [ + ("crm-universe", "crm-universe-506b24e3579f-files"), + ("Crm_Universe", "crm-universe-4c998975eec1-files"), + ("x" * 60, "x" * 45 + "-42f2d9733566-files"), + (LOCAL_UNIVERSE, "seed-ddf2904a1148-files"), + (f"{LOCAL}/artifacts/" + "Ticket Desk (EU) " * 5 + "x", "ticket-desk-eu-ticket-desk-eu-ticket-desk-eu-e6f6e0683a17-files"), + ("@local/~/work/日本", "e3629cd92c57-files"), +], ids=["bare", "bare-mixed-case", "bare-60-chars", "local", "local-long", "local-no-slug"]) +def test_a_loaded_universes_skill_is_named_after_it(universe_id, name): + skill = _build_skill_for_loaded_file_artifact_universe(universe_id, {"destination_path": "/app/files"}) + assert skill.name == name + skill.validate() + + +@pytest.mark.parametrize("task_id, universe_id", [ + ("triage", "triage__v2-VPC Endpoints"), + (LOCAL_TASK, f"{LOCAL_TASK}__v2-VPC Endpoints"), +], ids=["bare", "local"]) +def test_a_seeded_runs_universe_is_named_after_its_task_and_seed(tmp_path, monkeypatch, task_id, universe_id): + seen: list[str] = [] + + class _Task: + id, version, steps = task_id, 2, [] + + async def run(self, context, **kwargs): + seen.append(context.metadata["universe_id"]) + return context + + monkeypatch.setattr(Task, "get", classmethod(lambda cls, id, version=None: _Task())) + (tmp_path / "seeds.csv").write_text("name\nVPC Endpoints\n") + + result = CliRunner().invoke(cli, ["task", "run-batch", "--id", task_id, "--seeds", str(tmp_path / "seeds.csv"), + "--output-dir", str(tmp_path / "out")]) + + assert result.exit_code == 0, result.output + assert seen == [universe_id] + + +def _collect(suffix=None) -> CollectArtifactsTaskStep: + return CollectArtifactsTaskStep(id="collect", version=None, sandbox_name="box", artifact_paths=["report.pdf"], + universe_id_suffix=suffix) + + +@pytest.mark.parametrize(("task_steps", "seeds", "refused", "runs"), [ + ([_collect()], ["fine", "Q&A: onboarding"], 2, 0), + ([], ["fine", "Q&A: onboarding"], None, 2), + ([_collect("-records")], ["VPC Endpoints "], None, 1), + ([_collect(":records")], ["fine"], 1, 0), +], ids=["collects", "doesnt-collect", "suffix-makes-it-valid", "suffix-makes-it-invalid"]) +def test_a_seed_that_cant_name_an_local_universe_stops_a_collecting_run_batch_before_any_run( + tmp_path, monkeypatch, task_steps, seeds, refused, runs): + ran: list[TaskStepContext] = [] + + class _Task: + id, version, steps = LOCAL_TASK, 2, task_steps + + async def run(self, context, **kwargs): + ran.append(context) + return context + + monkeypatch.setattr(Task, "get", classmethod(lambda cls, id, version=None: _Task())) + (tmp_path / "seeds.csv").write_text("".join(f"{row}\n" for row in ["name", *seeds])) + + result = CliRunner().invoke(cli, ["task", "run-batch", "--id", _Task.id, "--seeds", str(tmp_path / "seeds.csv"), + "--output-dir", str(tmp_path / "out")]) + + assert result.exit_code == (1 if refused else 0), result.output + assert (f"seed {refused} can't name an @local universe" in result.output) is bool(refused) + assert len(ran) == runs diff --git a/tst/unit/store/fakes.py b/tst/unit/store/fakes.py index 573b9f18..6cb6d921 100644 --- a/tst/unit/store/fakes.py +++ b/tst/unit/store/fakes.py @@ -11,13 +11,20 @@ from __future__ import annotations +import asyncio +import base64 from datetime import datetime, timezone from pathlib import Path +from types import SimpleNamespace +from agent_env.env.env import DeployedGatewayEnv +from agent_env.env.envs.multi_env import MultiEnv +from agent_env.providers.env_providers import EnvironmentGatewayProvider +from agent_env.providers.env_state import LocalPostgresStateProvider from agent_env.store.base import ObjectAlreadyExistsError, ObjectNotFoundError from agent_env.store.document_store import DocumentStore, DuplicateKeyError, Eq, Filter from agent_env.store.image_store import ImageStore -from agent_env.store.object_store import DEFAULT_CONTENT_TYPE, ObjectMetadata, ObjectStore +from agent_env.store.object_store import DEFAULT_CONTENT_TYPE, LocalFilesystemObjectStore, ObjectMetadata, ObjectStore class FakeDocumentStore(DocumentStore): @@ -173,6 +180,15 @@ def read(self, key): return super().read(key) +class SigningObjectStore(LocalFilesystemObjectStore): + """A filesystem object store that signs urls, as a remote store does; the urls lead nowhere.""" + + def signed_get_url(self, object_url, expires_in=3600): + return f"https://objects.example.test/{self.get_object_key(object_url)}" + + signed_put_url = signed_get_url + + class FakeImageStore(ImageStore): """In-memory ImageStore fake: records ensure_repository calls, no docker involved. @@ -192,3 +208,64 @@ def ensure_repository(self, repository): def auth(self, ref): return None + + +COLLECTED = {"/app/artifact/report.pdf": b"%PDF", "/app/artifact/sub/notes.md": b"# notes"} + + +class _Stream: + def __init__(self, data: bytes): + self._data = data + + async def read(self): + return self._data + + +class CollectingVm: + """A VM serving ``COLLECTED`` to collect_artifacts' size probe and base64 read, and exiting 0 with ``ok`` + otherwise.""" + + mode = "vm" + + async def exec_script(self, script: str) -> str: + return "" + + async def exec_with_output(self, *args): + if "stat" in args: + return 0, str(len(COLLECTED[args[-1]])), "" + return 0, "ok", "" + + async def exec(self, *args): + path = args[-1].split("base64 < ", 1)[1].strip("'") + return SimpleNamespace(stdout=_Stream(base64.b64encode(COLLECTED[path])), stderr=_Stream(b""), + wait=lambda: asyncio.sleep(0, result=0)) + + +class SnapshotSandbox: + """A VM whose servicedb answers the changelog count and the container lookup; records every script.""" + + mode = "vm" + + def __init__(self) -> None: + self.scripts: list[str] = [] + + async def exec_script(self, script: str) -> str: + self.scripts.append(script) + return "0\n" if "_changelog" in script else "container-1\n" + + +def reattach_for_snapshot(monkeypatch, env_id: str, env_version: int, universe_id: str, + universe_version: int) -> SnapshotSandbox: + """Make ``EnvSnapshot.create`` find ``env_id`` deployed with ``universe_id`` loaded, and reconnect to a + ``SnapshotSandbox``, which it returns.""" + deployed = DeployedGatewayEnv(env_id=env_id, env_version=env_version, sandbox_id="sb-1", gateway_url="https://gw", + mcp_url="https://gw/mcp", db_web_url=None) + instances = SimpleNamespace(get=lambda instance_id: deployed, + get_environment_universe=lambda instance_id: {"id": universe_id, "version": universe_version}) + provider = EnvironmentGatewayProvider() + provider._state_provider = LocalPostgresStateProvider() + reattached = SimpleNamespace(_sandbox=SnapshotSandbox(), _env_provider=provider) + monkeypatch.setattr("agent_env.env.store.get_env_instance_store", lambda: instances) + monkeypatch.setattr("agent_env.env.env.Env.get", lambda *args: MultiEnv(id=env_id, version=env_version, mcp_server_envs=[])) + monkeypatch.setattr(MultiEnv, "from_deployed_env", classmethod(lambda cls, record: asyncio.sleep(0, result=reattached))) + return reattached._sandbox diff --git a/tst/unit/store/ids_test.py b/tst/unit/store/ids_test.py index 9df50cfa..5482a69b 100644 --- a/tst/unit/store/ids_test.py +++ b/tst/unit/store/ids_test.py @@ -1,6 +1,6 @@ """The encoding of namespaced ids into object keys, image repositories and filenames. -The golden table is frozen: persisted ``object_url``s and ``image_name``s record these strings, +The golden tables are frozen: persisted ``object_url``s and ``image_name``s record these strings, so a change here orphans every local cache written before it.""" import hashlib @@ -51,7 +51,8 @@ "with space", "hash#frag?q", "user@example.com", - "x" * 300, + "x" * 200, + "é" * 100, "", " ", "__", @@ -59,6 +60,13 @@ "@LOCAL/upper-namespace", ] +LONG_BARE = [ + ("x" * 201, "x" * 48 + "-84a0678c9093"), + ("é" * 101, "e" * 48 + "-96cbf9775498"), + ("tickets/" + "Ticket Desk (EU) " * 12, "tickets-ticket-desk-eu-ticket-desk-eu-ticket-des-abdff07f35af"), + ("!" * 201, "667fa95bbbea"), +] + @pytest.mark.parametrize("entity_id, segment", GOLDEN) def test_local_ids_encode_to_the_frozen_segment(entity_id, segment): @@ -75,15 +83,25 @@ def test_the_hash_covers_the_full_unnormalized_id(): assert key_segment("@local/~/My Work/Tickets V2") != key_segment("@local/~/my work/tickets v2") long_a, long_b = "@local/~/" + "x" * 60 + "a", "@local/~/" + "x" * 60 + "b" assert key_segment(long_a) != key_segment(long_b) + assert key_segment("x" * 300 + "a") != key_segment("x" * 300 + "b") @pytest.mark.parametrize("entity_id", LEGACY) -def test_every_other_id_passes_through_every_helper_byte_identical(entity_id): +def test_every_other_id_up_to_200_bytes_passes_through_every_helper_byte_identical(entity_id): assert key_segment(entity_id) == entity_id assert image_repository(entity_id) == entity_id assert fs_safe(entity_id) == entity_id +@pytest.mark.parametrize("entity_id, segment", LONG_BARE) +def test_a_bare_id_over_200_bytes_encodes_to_a_frozen_segment_that_is_one_path_component(tmp_path, entity_id, segment): + assert key_segment(entity_id) == image_repository(entity_id) == segment + assert segment.endswith(hashlib.sha256(entity_id.encode("utf-8")).hexdigest()[:12]) + assert _OCI_REPOSITORY.fullmatch(segment) + (tmp_path / segment).write_bytes(b"") + assert [path.name for path in tmp_path.iterdir()] == [segment] + + @pytest.mark.parametrize("entity_id, namespace, local", [ ("tickets", None, False), ("user@example.com", None, False), diff --git a/tst/unit/store/local_backends_composition_test.py b/tst/unit/store/local_backends_composition_test.py index 504d46d2..cecbe86d 100644 --- a/tst/unit/store/local_backends_composition_test.py +++ b/tst/unit/store/local_backends_composition_test.py @@ -331,3 +331,26 @@ def test_get_many_writes_each_local_id_into_its_own_encoded_directory(local_stor assert (out / fs_safe("@local/t/A") / "1").is_file() assert (out / fs_safe("@local/t/A/1") / "f").is_file() assert not (out / "@local").exists() + + +def test_get_many_writes_a_universe_whose_id_is_longer_than_a_filename(local_stores, tmp_path): + universe_id = "crm-suite__v2-" + "Escalated tickets from the EU desk, " * 7 + "and the US desk, escalated by hand" + FileArtifactUniverse.put_bundled(id=universe_id, files={"report.md": _write(tmp_path, "report.md", b"# report")}) + + out = tmp_path / "out" + res = CliRunner().invoke(file_artifact_universe, ["get-many", "--id", universe_id, "--output-dir", str(out)]) + + assert res.exit_code == 0, res.output + assert (out / key_segment(universe_id) / "report.md").read_bytes() == b"# report" + + +def test_get_many_writes_universes_that_differ_by_a_slash_into_different_directories(local_stores, tmp_path): + for universe_id, data in (("crm/eu", b"eu"), ("crm-eu", b"flat")): + FileArtifactUniverse.put_bundled(id=universe_id, files={"report.md": _write(tmp_path, f"{data.decode()}.md", data)}) + + out = tmp_path / "out" + res = CliRunner().invoke(file_artifact_universe, ["get-many", "--id", "crm/eu", "--id", "crm-eu", "--output-dir", str(out)]) + + assert res.exit_code == 0, res.output + assert (out / "crm" / "eu" / "report.md").read_bytes() == b"eu" + assert (out / "crm-eu" / "report.md").read_bytes() == b"flat" diff --git a/tst/unit/store/local_id_acceptance_test.py b/tst/unit/store/local_id_acceptance_test.py index e75e5995..0dc34fa5 100644 --- a/tst/unit/store/local_id_acceptance_test.py +++ b/tst/unit/store/local_id_acceptance_test.py @@ -4,16 +4,18 @@ import asyncio import importlib from types import SimpleNamespace +from unittest.mock import MagicMock import pytest from click.testing import CliRunner +from agent_env.a2a_agent.a2a_agent import A2AAgent from agent_env.a2a_agent.validator import A2AAgentValidator from agent_env.artifact.artifacts.docker_image import DockerImageArtifact from agent_env.artifact.artifacts.file import FileArtifact from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse from agent_env.cli import cli -from agent_env.config import configure, get_config, set_image_store +from agent_env.config import configure, get_config, set_image_store, set_object_store from agent_env.config.paths import state_root from agent_env.env.envs.mcp_server import MCPServerEnv from agent_env.env.envs.multi_env import MultiEnv @@ -23,9 +25,11 @@ from agent_env.store.base import NotFoundError from agent_env.store.ids import fs_safe, key_segment from agent_env.store.routing import LocalNamespaceDocumentStore -from tst.unit.store.fakes import FakeDocumentStore, FakeImageStore +from agent_env.task import Task +from tst.unit.store.fakes import FakeDocumentStore, FakeImageStore, SigningObjectStore, reattach_for_snapshot LOCAL_ID = "@local/t/Chaos" +LOCAL_ENV = "@local/~/bundle/envs/e" UNIVERSE = "@local/~/bundle/artifacts/data" @@ -131,39 +135,91 @@ def test_a_github_build_refuses_an_local_id_before_touching_a_registry_or_a_vm(l assert images.repositories == [] -@pytest.mark.parametrize("validate, kind", [ - (A2AAgentValidator.validate, "agent"), - (MCPServerEnv.validate, "env"), - (WebsiteEnv.validate, "env"), - (MultiEnv.validate, "env"), - (lambda env: MultiEnv.validate_universe_compatibility(env, "registry-universe"), "env"), -]) -def test_validating_an_local_entity_is_refused_before_anything_is_written(validate, kind): - with pytest.raises(ValueError, match=f"is an @local {kind}, and validating one isn't supported yet"): - asyncio.run(validate(SimpleNamespace(id="@local/~/bundle/x", version=1))) +def _validation_runs_nothing(monkeypatch): + universes = {u: SimpleNamespace(id=u, version=2, get_environment_artifacts=lambda: []) for u in ("registry-universe", UNIVERSE)} + monkeypatch.setattr("agent_env.artifact.EnvironmentUniverseArtifact.get", lambda id, version=None: universes[id]) + finished = SimpleNamespace(deployed_envs=[], deployed_agents=[], instance_id="validation-run") + monkeypatch.setattr(Task, "run", lambda self, **kwargs: asyncio.sleep(0, result=finished)) + + +@pytest.mark.parametrize("validate, task_id", [ + (lambda: MCPServerEnv(id=LOCAL_ENV, version=1, docker_image_artifact=MagicMock(), environment_name="svc").validate(), + f"{LOCAL_ENV}__validate-v1"), + (lambda: WebsiteEnv(id=LOCAL_ENV, version=1, backend_docker_image_artifact=MagicMock(), frontend_docker_image_artifact=MagicMock(), + environment_name="svc").validate(), + f"{LOCAL_ENV}__validate-v1"), + (lambda: MultiEnv(id=LOCAL_ENV, version=1, mcp_server_envs=[]).validate(), f"{LOCAL_ENV}__validate-v1"), + (lambda: MultiEnv(id=LOCAL_ENV, version=1, mcp_server_envs=[]).validate_universe_compatibility("registry-universe"), + f"{LOCAL_ENV}__validate-universe-compat-v1-registry-universe-v2"), + (lambda: MultiEnv(id="registry-env", version=1, mcp_server_envs=[]).validate_universe_compatibility(UNIVERSE), + f"{UNIVERSE}__validate-universe-compat-v2-registry-env-v1"), +], ids=["mcp_server", "website", "multi", "compat-local-env", "compat-local-universe"]) +def test_validating_an_local_entity_writes_its_task_under_it_to_the_local_store(local_stores, cli_routing, monkeypatch, + validate, task_id): + _validation_runs_nothing(monkeypatch) + + assert asyncio.run(validate()) == "validation-run" + + assert [d["id"] for d in _local().query("tasks", Filter())] == [task_id] + assert not _documents().path.exists() or _documents().count("tasks", Filter()) == 0 + + +def test_validating_an_local_agent_stops_at_the_local_store_it_cannot_sign_with_before_writing_a_task( + local_stores, cli_routing, tmp_path, +): + configured = SigningObjectStore(str(tmp_path / "configured-objects")) + set_object_store(configured) + agent = A2AAgent(id="@local/~/bundle/agents/a", version=1, docker_image_artifact=MagicMock()) + + with pytest.raises(RuntimeError, match="A2A validation requires a signable object store"): + asyncio.run(A2AAgentValidator.validate(agent)) + fixtures = f"a2a_validator/probe_fixtures/{key_segment(agent.id)}-v1" + skills = f"a2a_validator/validator_skill/{key_segment(agent.id)}-v1" + assert sorted(get_config().get_object_store_for(agent.id).list("a2a_validator/")) == [ + f"{fixtures}/red.png", *(f"{skills}/{name}/SKILL.md" for name in ("validator-probe-bundle", "validator-test-s3")), + ] + assert configured.list("") == [] + assert all(not store.path.exists() or store.count("tasks", Filter()) == 0 for store in (_local(), _documents())) -def test_a_put_skips_validating_an_local_entity_and_succeeds(local_stores): + +def test_a_put_validates_an_local_entity_like_any_other(local_stores, monkeypatch): image_url = local_stores.get_object_store().put("artifacts/docker_image/srv-img/1/x.tar.gz", b"x") image = DockerImageArtifact.put_tar("srv-img", description="d", image_name="reg/srv:v1", tar_gz_s3_url=image_url) MCPServerEnv.put(id="srv", docker_image_artifact=image, environment_name="svc") + _validation_runs_nothing(monkeypatch) result = CliRunner().invoke(cli, ["env", "multi", "put", "--id", "@local/~/bundle/envs/m", "--mcp-server", "srv", "--validate"]) assert result.exit_code == 0, result.output - assert "Skipped validation: validating an @local env isn't supported yet" in result.output assert [d["id"] for d in _local().query("envs", Filter())] == ["@local/~/bundle/envs/m"] + assert [d["id"] for d in _local().query("tasks", Filter())] == ["@local/~/bundle/envs/m__validate-v1"] -def test_snapshotting_an_local_env_is_refused_before_anything_is_uploaded(local_stores, cli_routing, monkeypatch): - deployed = SimpleNamespace(env_id="@local/~/bundle/envs/e", env_version=1) - monkeypatch.setattr("agent_env.env.store.get_env_instance_store", lambda: SimpleNamespace(get=lambda instance_id: deployed)) - monkeypatch.setattr("agent_env.env.env.Env.get", lambda *a, **k: pytest.fail("the env was read")) +def test_snapshotting_an_local_env_stops_at_the_local_store_it_cannot_presign_before_the_sandbox_runs_anything( + local_stores, cli_routing, tmp_path, monkeypatch, +): + configured = SigningObjectStore(str(tmp_path / "configured-objects")) + set_object_store(configured) + sandbox = reattach_for_snapshot(monkeypatch, LOCAL_ENV, 1, UNIVERSE, 1) - with pytest.raises(ValueError, match="is an @local env, and snapshotting one isn't supported yet"): + with pytest.raises(RuntimeError, match="LocalFilesystemObjectStore can't presign uploads"): asyncio.run(EnvSnapshot.create("instance-1")) - assert local_stores.get_object_store().list("") == [] + assert sandbox.scripts == [] + assert configured.list("") == [] and get_config().get_object_store_for(LOCAL_ENV).list("") == [] + + +@pytest.mark.parametrize("env_id, universe_id", [(LOCAL_ENV, "registry-universe"), ("registry-env", UNIVERSE)], + ids=["local-env", "local-universe"]) +def test_a_snapshot_of_an_env_and_a_universe_in_different_namespaces_is_refused_before_the_sandbox_is_reached( + local_stores, cli_routing, monkeypatch, env_id, universe_id, +): + reattach_for_snapshot(monkeypatch, env_id, 1, universe_id, 1) + monkeypatch.setattr(MultiEnv, "from_deployed_env", classmethod(lambda cls, record: pytest.fail("the sandbox was reached"))) + + with pytest.raises(ValueError, match="are in different namespaces"): + asyncio.run(EnvSnapshot.create("instance-1")) @pytest.mark.parametrize("derived", ["env-snapshot-@local/~/bundle/envs/e", "cua-vm-@local/~/bundle/envs/e", "x-@local/y"]) diff --git a/tst/unit/task/local_run_write_fence_test.py b/tst/unit/task/local_run_write_fence_test.py index 182322ec..477d728b 100644 --- a/tst/unit/task/local_run_write_fence_test.py +++ b/tst/unit/task/local_run_write_fence_test.py @@ -2,25 +2,39 @@ and everything it writes lands in the per-user local stores or is refused before it happens.""" import asyncio +import hashlib import json +from pathlib import Path +from types import SimpleNamespace import pytest +import agent_env.providers.sandbox_providers.sandbox_provider as sandbox_provider +from agent_env.artifact import EnvironmentUniverseArtifact, FileArtifact from agent_env.artifact.artifacts.file_artifact_universe import FileArtifactUniverse +from agent_env.artifact.store import get_artifact_store from agent_env.bundle import Outcome, RunInterrupted, run_bundle from agent_env.bundle.installed import find_bundle from agent_env.config import configure, get_config from agent_env.config.paths import state_root +from agent_env.env.env import DeployedGatewayEnv from agent_env.store import Filter, LocalSqliteDocumentStore +from agent_env.store.ids import derive_id from agent_env.store.image_store import OciRegistryImageStore from agent_env.store.object_store import LocalFilesystemObjectStore from agent_env.providers.sandbox_providers.local_sandbox import LocalSandbox -from agent_env.store.routing import LocalRunWriteError +from agent_env.store.routing import LocalRunWriteError, run_scope from agent_env.task.task import Task -from agent_env.task_step.context import TaskStepContext +from agent_env.task_step.context import DeployedSandbox, TaskStepContext +from agent_env.task_step.snapshot_utils.snapshot_series import SnapshotConfig, SnapshotSeries from agent_env.task_step.task_step import TaskStep +from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep +from agent_env.task_step.task_steps.snapshot_env import SnapshotEnvTaskStep +from agent_env.task_step.task_steps.verifiers.run_container_unit_tests_verifier import RunContainerUnitTestsVerifierTaskStep +from tst.unit.store.fakes import CollectingVm LOCAL_TASK = "@local/~/bundle/tasks/t" +LOCAL_ENV = "@local/~/bundle/envs/tickets" _DOCUMENT_WRITES = ("insert", "update", "update_one_and_get", "replace", "delete") _OBJECT_WRITES = ("put", "put_file", "put_file_at", "signed_put_url", "signed_post", "issue_write_grant", "issue_upload_policy") @@ -193,3 +207,154 @@ async def execute(self, context): local = get_config().get_document_store().local assert {doc["owner"] for doc in local.query("runs", Filter())} == {LOCAL_TASK, "@local/~/bundle/tasks/u"} assert {doc["owner"] for doc in documents.query("runs", Filter())} == {"registry-task", "other-registry-task"} + + +@pytest.fixture +def vm(monkeypatch): + provider = SimpleNamespace(get_sandbox=lambda sandbox_id: asyncio.sleep(0, result=CollectingVm()), close=lambda: asyncio.sleep(0)) + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: provider) + return DeployedSandbox(sandbox_name="box", sandbox_id="sb-1", sandbox_mode="vm") + + +@pytest.mark.asyncio +async def test_an_local_run_collects_into_a_universe_named_after_the_run(sentinels, vm): + step = CollectArtifactsTaskStep(id="collect", version=None, sandbox_name="box", base_path="/app/artifact", + artifact_paths=["report.pdf", "sub/notes.md"], universe_id_suffix="-docs") + + context = await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=TaskStepContext(deployed_sandboxes=[vm])) + + universe = FileArtifactUniverse.get(f"{context.instance_id}-docs") + assert universe.file_artifact_ids == { + name: derive_id(universe.id, hashlib.sha256(name.encode()).hexdigest()[:16]) for name in ("report.pdf", "sub/notes.md") + } + assert {name: fa.load() for name, fa in universe.get_file_artifacts().items()} == {"report.pdf": b"%PDF", "sub/notes.md": b"# notes"} + + +@pytest.mark.asyncio +async def test_an_local_run_refuses_a_bare_universe_to_collect_into_before_reading_the_sandbox(sentinels, vm, monkeypatch): + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: pytest.fail("the sandbox was read")) + context = TaskStepContext(deployed_sandboxes=[vm]) + context.metadata["universe_id"] = "VPC Endpoints" + step = CollectArtifactsTaskStep(id="collect", version=None, sandbox_name="box", artifact_paths=["report.pdf"]) + + with pytest.raises(LocalRunWriteError, match="'VPC Endpoints'"): + await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=context) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("seed", ["Q&A: onboarding", "VPC Endpoints "]) +async def test_an_local_run_refuses_an_invalid_local_universe_to_collect_into_before_reading_the_sandbox( + sentinels, vm, monkeypatch, seed): + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: pytest.fail("the sandbox was read")) + context = TaskStepContext(deployed_sandboxes=[vm]) + context.metadata["universe_id"] = f"{LOCAL_TASK}-v1-{seed}" + step = CollectArtifactsTaskStep(id="collect", version=None, sandbox_name="box", artifact_paths=["report.pdf"]) + + with pytest.raises(ValueError, match="@local"): + await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=context) + + +class _Gmail: + """An env with one service, gmail.""" + + version = 1 + + def __init__(self, id, version=None): + self.id = id + self.mcp_server_envs = [SimpleNamespace(environment_name="gmail")] + + +async def _export(gateway_url, name, tmp_path, timeout, deployed=None): + Path(tmp_path).write_text(json.dumps({"service": name})) + return ".json" + + +@pytest.fixture +def exports(monkeypatch): + """Envs resolve to ``_Gmail``, and each service exports ``{"service": }``.""" + monkeypatch.setattr("agent_env.env.env.Env.get", _Gmail) + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(_export)) + + +def _deployed(env_id): + return DeployedGatewayEnv(env_id=env_id, env_version=1, gateway_url="https://gw", mcp_url="https://gw/mcp", db_web_url=None, + sandbox_id="sb-1") + + +def _assert_snapshot(universe_id): + (gmail,) = EnvironmentUniverseArtifact.get(universe_id).get_environment_artifacts() + assert (gmail.id, json.loads(gmail.get_file_artifact().load())) == (f"{universe_id}-gmail", {"service": "gmail"}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("env_id, snapshot_id", [(LOCAL_ENV, None), ("rocket", None), ("rocket", f"{LOCAL_TASK}/greeting")], + ids=["local-env", "registry-env", "named"]) +async def test_an_local_runs_env_snapshot_is_named_after_the_run_unless_the_task_names_it(sentinels, exports, env_id, snapshot_id): + step = SnapshotEnvTaskStep(id="snap", version=None, env_id=env_id, snapshot_id=snapshot_id) + + context = await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=TaskStepContext(deployed_envs=[_deployed(env_id)])) + + universe_id = snapshot_id or derive_id(context.instance_id, "snapshot-snap") + assert context.metadata["env_snapshotted_universes"]["snap"] == {"id": universe_id, "version": 1} + _assert_snapshot(universe_id) + + +@pytest.mark.asyncio +async def test_an_local_run_refuses_a_snapshot_step_id_its_universe_cant_take_before_resolving_the_env(sentinels, monkeypatch): + monkeypatch.setattr(SnapshotEnvTaskStep, "_resolve_deployed_env", lambda self, context: pytest.fail("the env was resolved")) + step = SnapshotEnvTaskStep(id="snap:final", version=None, env_id=LOCAL_ENV) + + with pytest.raises(ValueError, match="contains ':'"): + await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=TaskStepContext(deployed_envs=[_deployed(LOCAL_ENV)])) + + +@pytest.mark.asyncio +async def test_an_local_runs_capture_series_names_its_captures_after_the_run(sentinels, exports): + instance_id = f"{LOCAL_TASK}-ab12cd34" + series = SnapshotSeries(step_id="solve", agent_name="solver", prompt_id="p1", a2a_context_id="c1", + config=SnapshotConfig(env_id="rocket"), trajectory_output_prefix="unused", instance_id=instance_id) + row: dict = {} + + with run_scope(LOCAL_TASK): + assert await series._capture_env_state(TaskStepContext(instance_id=instance_id, deployed_envs=[_deployed("rocket")]), row, 30) is None + assert get_artifact_store().next_version(series.workspace_artifact_id) == 1 + + assert series.workspace_artifact_id == derive_id(instance_id, "solve-workspace") + assert row["env_universe_id"] == derive_id(instance_id, "snapshot-solve") + _assert_snapshot(row["env_universe_id"]) + + +def test_an_local_runs_capture_series_refuses_a_step_id_its_captures_cant_take_when_it_is_built(): + def series(instance_id): + return SnapshotSeries(step_id="solve:1", agent_name="solver", prompt_id="p1", a2a_context_id="c1", + config=SnapshotConfig(env_id="rocket"), trajectory_output_prefix="unused", instance_id=instance_id) + + with pytest.raises(ValueError, match="contains ':'"): + series(f"{LOCAL_TASK}-ab12cd34") + assert series("rocket-run-ab12cd34").workspace_artifact_id == "rocket-run-ab12cd34__solve:1-workspace" + + +@pytest.mark.asyncio +async def test_an_local_runs_verifier_outputs_are_named_after_the_run(sentinels, vm): + context = TaskStepContext(deployed_sandboxes=[vm]) + context.metadata["deployed_docker_containers"] = [{"container_name": "c", "sandbox_name": "box"}] + step = RunContainerUnitTestsVerifierTaskStep(id="verify", version=None, sandbox_name="box", container_name="c", command="true") + + context = await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=context) + + entry = context.metadata["verifications"]["verify"] + for stream, text in (("stdout", b"ok"), ("stderr", b"")): + artifact = FileArtifact.get(entry[f"{stream}_artifact"]["id"]) + assert artifact.id.startswith(derive_id(context.instance_id, f"verifier-{stream}-verify-")) + assert artifact.load() == text + + +@pytest.mark.asyncio +async def test_an_local_run_refuses_a_verifier_step_id_its_outputs_cant_take_before_the_tests_run(sentinels, vm, monkeypatch): + monkeypatch.setattr(sandbox_provider, "get_sandbox_provider", lambda: pytest.fail("the sandbox was read")) + context = TaskStepContext(deployed_sandboxes=[vm]) + context.metadata["deployed_docker_containers"] = [{"container_name": "c", "sandbox_name": "box"}] + step = RunContainerUnitTestsVerifierTaskStep(id="verify:1", version=None, sandbox_name="box", container_name="c", command="true") + + with pytest.raises(ValueError, match="contains ':'"): + await Task(id=LOCAL_TASK, version=1, steps=[step]).run(context=context) diff --git a/tst/unit/task_step/test_agent_state_capture.py b/tst/unit/task_step/test_agent_state_capture.py index 58fb8b6c..d55844f2 100644 --- a/tst/unit/task_step/test_agent_state_capture.py +++ b/tst/unit/task_step/test_agent_state_capture.py @@ -133,6 +133,18 @@ async def test_a_capture_is_issued_under_the_fixture_prefix(monkeypatch): assert all(url.startswith(result.capture_prefix) for url in _granted(rec).values()) +@pytest.mark.asyncio +@pytest.mark.parametrize("artifact_id, segment", [ + ("wsp", "wsp"), + ("@local/~/work/triage/tasks/t-abcd1234__solve-workspace", "local/work-triage-tasks-t-abcd1234-solve-workspace-606593d6050a"), +], ids=["bare", "local"]) +async def test_a_capture_is_issued_under_the_workspaces_key_segment(monkeypatch, artifact_id, segment): + install_capture_stubs(monkeypatch) + result = await _capture(artifact_id=artifact_id) + assert result.capture_prefix.startswith(f"s3://{BUCKET}/agent_snapshots/{segment}/") + assert result.universe_id == artifact_id + + @pytest.mark.asyncio async def test_a_card_offering_both_save_forms_gets_the_object_one(monkeypatch): rec = install_capture_stubs(monkeypatch) diff --git a/tst/unit/task_step/test_collect_artifacts_universe_suffix.py b/tst/unit/task_step/test_collect_artifacts_universe_suffix.py index ed9f6479..f4105ad9 100644 --- a/tst/unit/task_step/test_collect_artifacts_universe_suffix.py +++ b/tst/unit/task_step/test_collect_artifacts_universe_suffix.py @@ -128,28 +128,11 @@ async def test_it_falls_back_to_the_instance_id(self, env): assert env == ["inst-42-records"] @pytest.mark.asyncio - async def test_suffix_is_sanitized_like_the_id(self, env): - await _step("collect", "/rec ords").execute(_ctx(universe_id="caa_105")) - assert env == ["caa_105_rec-ords"] - - -class TestLengthBudget: - """The id is capped at 100 chars; the suffix must fit inside that, not past it.""" - - @pytest.mark.asyncio - async def test_a_long_run_id_still_yields_a_bounded_id(self, env): - await _step("collect", "-records").execute(_ctx(universe_id="c" * 300)) - assert len(env[0]) == 100 - assert env[0].endswith("-records") - - @pytest.mark.asyncio - async def test_a_long_run_id_still_yields_distinct_names(self, env): - long_id = "c" * 300 - ctx = _ctx(universe_id=long_id) - await _step("collect-records", "-records", "/app/records").execute(ctx) - await _step("collect-artifacts").execute(ctx) - assert env[0] != env[1], env - assert all(len(i) <= 100 for i in env), env + async def test_runs_whose_ids_share_a_100_char_prefix_get_distinct_names(self, env): + step, prefix = _step("collect", "-records"), "c" * 100 + await step.execute(_ctx(instance_id=f"{prefix}-run-1")) + await step.execute(_ctx(instance_id=f"{prefix}-run-2")) + assert env == [f"{prefix}-run-1-records", f"{prefix}-run-2-records"] class TestConcurrentSeeds: diff --git a/tst/unit/task_step/test_mcp_cli_builder.py b/tst/unit/task_step/test_mcp_cli_builder.py index d7773229..fb1deae1 100644 --- a/tst/unit/task_step/test_mcp_cli_builder.py +++ b/tst/unit/task_step/test_mcp_cli_builder.py @@ -90,7 +90,7 @@ def test_codegen_loads_sibling_env_file(): def test_build_mcp_cli_step_dict_roundtrip(): step = BuildMcpCliTaskStep(id="s1", version=None, env_id="slack-mcp", command_name="slack") - assert step.cli_artifact_id == "cli-slack-mcp" # eager default + assert step.cli_artifact_id == "slack-mcp__cli" # eager default payload = step.to_dict() cls = get_task_step_registry()["build_mcp_cli"] @@ -99,7 +99,7 @@ def test_build_mcp_cli_step_dict_roundtrip(): assert restored.id == "s1" assert restored.env_id == "slack-mcp" assert restored.command_name == "slack" - assert restored.cli_artifact_id == "cli-slack-mcp" + assert restored.cli_artifact_id == "slack-mcp__cli" assert restored.to_dict() == payload @@ -253,7 +253,7 @@ def put(**kw): if "finished after its caller was cancelled" in caplog.text: break await asyncio.sleep(0.02) - assert "Uploading CLI cli-slack-mcp finished after its caller was cancelled" in caplog.text + assert "Uploading CLI slack-mcp__cli finished after its caller was cancelled" in caplog.text assert not dirs[0].exists() diff --git a/tst/unit/task_step/test_prompt_agent_snapshots.py b/tst/unit/task_step/test_prompt_agent_snapshots.py index ff7559cc..4beef2c6 100644 --- a/tst/unit/task_step/test_prompt_agent_snapshots.py +++ b/tst/unit/task_step/test_prompt_agent_snapshots.py @@ -8,6 +8,7 @@ from __future__ import annotations import asyncio +import re import time import pytest @@ -31,8 +32,8 @@ def _s3_store_backs_prefix_minting(): # they stay s3://BUCKET/... instead of resolving a filesystem path under cwd. set_object_store(S3ObjectStore(client=object(), bucket=BUCKET)) INSTANCE_ID = "solve-run-0123456789abcdef" -# `INSTANCE_ID`'s tail, capped at 16 — what every artifact id is scoped by. -DISCRIMINATOR = "0123456789abcdef" +# The env universe a `_series()` capture versions into. +SNAPSHOT_ID = f"{INSTANCE_ID}__snapshot-solve" def _series( @@ -330,8 +331,12 @@ def test_rollouts_sharing_a_context_id_get_distinct_artifact_ids(): shared = "a-very-long-restored-context-id" # > 16 chars, so a prefix collides a = _series(a2a_context_id=shared, instance_id="solve-run-1111111111111111", env_id="env-1") b = _series(a2a_context_id=shared, instance_id="solve-run-2222222222222222", env_id="env-1") - # The token both the workspace and env universe ids are built from. - assert a._discriminator != b._discriminator + assert a.workspace_artifact_id != b.workspace_artifact_id + + +def test_rollouts_without_an_instance_id_get_distinct_adhoc_artifact_ids(): + a, b = _series(instance_id=None), _series(instance_id=None) + assert re.fullmatch(r"adhoc-[0-9a-f]{12}__solve-workspace", a.workspace_artifact_id) assert a.workspace_artifact_id != b.workspace_artifact_id @@ -684,11 +689,11 @@ async def test_env_state_lands_as_a_versioned_service_universe_on_the_row(monkey assert len(rows) >= 2 # Two scalars per row, not a per-service URL map a long series would grow # without bound. - assert all(r["env_universe_id"] == f"snapshot-env-1-{DISCRIMINATOR}" for r in rows) + assert all(r["env_universe_id"] == SNAPSHOT_ID for r in rows) versions = [r["env_universe_version"] for r in rows] assert versions == sorted(versions) and len(set(versions)) == len(versions) # One artifact id per rollout, scoped to the instance. - assert {c["snapshot_id"] for c in calls} == {f"snapshot-env-1-{DISCRIMINATOR}"} + assert {c["snapshot_id"] for c in calls} == {SNAPSHOT_ID} @pytest.mark.asyncio @@ -702,7 +707,7 @@ async def test_the_final_capture_publishes_the_universe_for_load_artifact(monkey # Keyed on the step id, so a judge step can be wired to this prompt step # without knowing it snapshots. `load_artifact` resolves it. assert ctx.metadata["env_snapshotted_universes"]["solve"] == { - "id": f"snapshot-env-1-{DISCRIMINATOR}", + "id": SNAPSHOT_ID, "version": 4, } @@ -1093,8 +1098,7 @@ async def ok_conversation(self, context, *, a2a_context_id=None, series=None): ctx.instance_id = "solve-0123456789abcdef01" await step.execute(ctx) - # Last dash-segment of the instance id, truncated to 16 — as snapshot_env does. - assert ws.artifact_ids == ["solve-workspace-0123456789abcdef"] + assert ws.artifact_ids == ["solve-0123456789abcdef01__solve-workspace"] # ============================================================== concurrency @@ -1132,7 +1136,7 @@ async def one(instance_id): assert len(_rows(slow)) > len(_rows(fast)) # And each rollout derived its own artifact, so no two captures across the two # rollouts ever allocate a version of the same id. - assert set(ws.artifact_ids) == {"solve-workspace-fast", "solve-workspace-slow"} + assert set(ws.artifact_ids) == {"solve-fast__solve-workspace", "solve-slow__solve-workspace"} # ============================================================== turn boundaries diff --git a/tst/unit/task_step/test_snapshot_env.py b/tst/unit/task_step/test_snapshot_env.py index 14e27776..fa94ae93 100644 --- a/tst/unit/task_step/test_snapshot_env.py +++ b/tst/unit/task_step/test_snapshot_env.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re import pytest from agentenv_protocol import FilePart @@ -121,20 +122,21 @@ def test_resolve_unknown_instance_id_raises(): def test_snapshot_id_derived_from_instance_id_is_deterministic(): ctx = _ctx(_deployed()) step = _step() - first = step._derive_snapshot_id(ctx, "env-1") - second = step._derive_snapshot_id(ctx, "env-1") - assert first == second == "snapshot-env-1-65717eb66eeec083" + first = step._derive_snapshot_id(ctx) + second = step._derive_snapshot_id(ctx) + assert first == second == "task-x-65717eb66eeec083__snapshot-snap-1" def test_snapshot_id_override_wins(): ctx = _ctx(_deployed()) - assert _step(snapshot_id="snap-mine")._derive_snapshot_id(ctx, "env-1") == "snap-mine" + assert _step(snapshot_id="snap-mine")._derive_snapshot_id(ctx) == "snap-mine" -def test_snapshot_id_falls_back_to_random_without_instance_id(): +def test_snapshot_id_falls_back_to_an_adhoc_run_without_instance_id(): ctx = _ctx(_deployed(), task_instance_id=None) - generated = _step()._derive_snapshot_id(ctx, "env-1") - assert generated.startswith("snapshot-env-1-") + first, second = _step()._derive_snapshot_id(ctx), _step()._derive_snapshot_id(ctx) + assert re.fullmatch(r"adhoc-[0-9a-f]{12}__snapshot-snap-1", first) + assert first != second # ── execute wiring ─────────────────────────────────────────────────────────── @@ -177,7 +179,7 @@ class _FakeEnv: assert captured["gateway_url"] == "https://gw-1" assert captured["original_universe_artifact_id"] == "base-universe" assert ctx.metadata["env_snapshotted_universes"]["snap-1"] == { - "id": "snapshot-env-1-65717eb66eeec083", + "id": "task-x-65717eb66eeec083__snapshot-snap-1", "version": 1, } @@ -824,6 +826,26 @@ async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None): ) +@pytest.mark.asyncio +async def test_core_snapshots_under_an_id_longer_than_a_filename_can_be(local_stores, monkeypatch): + """A run's snapshot id can pass the 255-byte filename limit, so neither a temp file nor a key segment of the + local store may be the whole id.""" + snapshot_id = f"task-{'x' * 250}__snapshot-snap-1" + + async def fake_export(gateway_url, name, tmp_path, timeout, deployed=None): + with open(tmp_path, "w") as f: + json.dump({"service": name}, f) + return ".json" + + monkeypatch.setattr(SnapshotEnvTaskStep, "_export_environment_to_file", staticmethod(fake_export)) + + result = await SnapshotEnvTaskStep.snapshot_env_state(env=_FakeMultiEnv(["slack"]), gateway_url="https://gw", snapshot_id=snapshot_id) + + (service,) = EnvironmentUniverseArtifact.get(result.environment_universe_artifact_id).get_environment_artifacts() + assert (result.environment_universe_artifact_id, service.id) == (snapshot_id, f"{snapshot_id}-slack") + assert json.loads(service.get_file_artifact().load()) == {"service": "slack"} + + @pytest.mark.asyncio async def test_export_service_to_file_filepart_without_bytes_or_uri_raises_valueerror( monkeypatch, tmp_path 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 ace8ccbe..7bee70fe 100644 --- a/tst/unit/task_step/test_verify_universe_agent_judge.py +++ b/tst/unit/task_step/test_verify_universe_agent_judge.py @@ -60,7 +60,7 @@ class TestFauId: def test_deterministic(self): a = VerifyUniverseLoadExportRoundtripStep.file_artifact_universe_id("env", 3, "uni", 7) b = VerifyUniverseLoadExportRoundtripStep.file_artifact_universe_id("env", 3, "uni", 7) - assert a == b == "validate-env-v3-uni-v7-fau" + assert a == b == "env__validate-v3-uni-v7-fau" def test_emit_flag_round_trips_through_serialization(self): s = VerifyUniverseLoadExportRoundtripStep( @@ -212,8 +212,8 @@ def get_environment_artifacts(self): fau = step._create_file_artifact_universe(orig_sas, _FakeExportedUniverse(), export2_raw, env_version=3, universe_version=7) - assert fau.id == "validate-multi-x-v3-uni-v7-fau" - assert put_calls["id"] == "validate-multi-x-v3-uni-v7-fau" + assert fau.id == "multi-x__validate-v3-uni-v7-fau" + assert put_calls["id"] == "multi-x__validate-v3-uni-v7-fau" fa_map = put_calls["file_artifacts"] assert set(fa_map) == { @@ -226,10 +226,10 @@ def get_environment_artifacts(self): assert fa_map["export_1/slack.json"].id == "fa-slack" # export_2 dumped losslessly: re-parses to the same object, unicode kept literal - slack_dumped = dumped["validate-multi-x-v3-uni-v7-export2-slack"] + slack_dumped = dumped["multi-x__validate-v3-uni-v7-export2-slack"] assert json.loads(slack_dumped) == export2_raw["slack"] assert "héllo" in slack_dumped # ensure_ascii=False - assert json.loads(dumped["validate-multi-x-v3-uni-v7-export2-oracle_gl"]) == export2_raw["oracle_gl"] + assert json.loads(dumped["multi-x__validate-v3-uni-v7-export2-oracle_gl"]) == export2_raw["oracle_gl"] class TestFrontendContract: diff --git a/tst/util/capabilities.py b/tst/util/capabilities.py index 6129a275..e276b661 100644 --- a/tst/util/capabilities.py +++ b/tst/util/capabilities.py @@ -10,6 +10,8 @@ from __future__ import annotations +import subprocess + import pytest MISSING_CAPABILITY_PREFIX = "agentenv-capability-missing: " @@ -29,6 +31,10 @@ #: found through ``AGENT_ENV_TEST_MCP_SERVERS_DIR`` (one ``/Dockerfile`` per server). MCP_SERVER_SOURCES = "mcp_server_sources" +#: ``collect_artifacts`` sizes a file with ``stat -c %s``, which only GNU stat takes (macOS ships BSD stat), +#: so collecting off a local sandbox needs GNU stat first on PATH. +GNU_STAT = "gnu_stat" + def missing_capability_reason(capability: str) -> str: return f"{MISSING_CAPABILITY_PREFIX}{capability}" @@ -131,3 +137,19 @@ def skip_without_default_a2a_agent() -> pytest.MarkDecorator: not default_a2a_agent_is_registered(), reason=missing_capability_reason(DEFAULT_A2A_AGENT), ) + + +def gnu_stat_is_available() -> bool: + """Whether ``stat -c %s`` prints a file's size on this host.""" + try: + return subprocess.run(["stat", "-c", "%s", __file__], capture_output=True).returncode == 0 + except OSError: + return False + + +def skip_without_gnu_stat() -> pytest.MarkDecorator: + """Collection-time ``skipif`` for tests that collect artifacts off a local sandbox.""" + return pytest.mark.skipif( + not gnu_stat_is_available(), + reason=missing_capability_reason(GNU_STAT), + )