feat(store)!: name every derived id <base>__<suffix>, bare or @local

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: <name>` (`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.

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)

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-<env_id>``), so it's named like one written out. False when its type can't read the step."""
+        ``<env_id>__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]:

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)

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))

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)
 

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}")

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}")

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:

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

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}")
 

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"