feat(task_step): StepContextError, for a step the run's context can't satisfy

src/agent_env/task_step/__init__.py

@@ -1,7 +1,7 @@
 """TaskStep module for AgentEnv."""
 
 from agent_env.env.env import DeployedEnv
-from .context import DeployedAgent, PromptResponse, TaskStepContext
+from .context import DeployedAgent, PromptResponse, StepContextError, TaskStepContext
 from .task_step import TaskStep
 from .task_steps import AddSkillsTaskStep, AgentPromptResponseVerifierTaskStep, BuildMcpCliTaskStep, CombineUniverseVerdictsStep, DeployAgentTaskStep, DeployEnvTaskStep, EnvOutcomeVerifierTaskStep, LoadArtifactTaskStep, ModifyEnvToolAccessStep, PromptAgentTaskStep, RubricsVerifierTaskStep, ScoreAggregator, ValidationGateAggregatorStep, VerifyCoreA2AProtocolStep, VerifyCoreEnvironmentProtocolStep, VerifyEnvironmentCardStep, VerifyMCPEnvAssessmentStep, VerifyMCPToolSchemaTaskStep, VerifySpecConformanceTaskStep, VerifyUniverseLoadExportRoundtripStep, VerifyA2AAgentCardStep, VerifyA2AAgentMCPStep, VerifyA2ASkillConfigStep, VerifyA2ATrajectoryStep
 from .task_steps.verifiers.judge_utils.judge_output_format import JudgeOutputFormat, get_judge_output_format_spec
@@ -14,6 +14,7 @@
     "get_judge_output_format_spec",
     "PromptResponse",
     "ScoreAggregator",
+    "StepContextError",
     "TaskStepContext",
     "TaskStep",
     "AddSkillsTaskStep",

src/agent_env/task_step/context.py

@@ -14,6 +14,12 @@
 
 HEARTBEAT_KEY = "_heartbeat_fn"
 
+
+class StepContextError(RuntimeError):
+    """The run's context can't satisfy a step: something the step needs from an earlier step is missing, or
+    something it would create is already there. Running the step again on the same context fails the same
+    way, so a runner should not retry it."""
+
 _REDACTED_KEYS = {
     "litellm_api_key", "judge_litellm_api_key", "usersim_api_key", "remote_tokens", "cf_access_client_secret",
     HEARTBEAT_KEY,

src/agent_env/task_step/task_steps/add_skills.py

@@ -15,7 +15,7 @@
 from agentenv_protocol.client import children, find_extension_method
 
 from agent_env.config import get_config
-from agent_env.task_step.context import DeployedAgent, TaskStepContext
+from agent_env.task_step.context import DeployedAgent, StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.entity.scopes import FieldScope
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
@@ -261,7 +261,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             raise ValueError("; ".join(problems))
         agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
         if agent is None:
-            raise RuntimeError(f"Agent '{self.agent_name}' not found in context.deployed_agents")
+            raise StepContextError(f"Agent '{self.agent_name}' not found in context.deployed_agents")
         if agent.instance_id is None:
             raise RuntimeError(f"Agent '{self.agent_name}' is missing instance_id; cannot resolve deployment")
 

src/agent_env/task_step/task_steps/collect_artifacts.py

@@ -80,7 +80,7 @@
 from agent_env.config import get_config
 from agent_env.env.gateway.constants import EXT_STEP_URI
 from agent_env.store.ids import derive_path_id, is_local_id, validate_local_id
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 from agent_env.task_step.task_steps.sandbox_utils.sandbox_utils import find_agent_container
@@ -650,7 +650,7 @@ def _resolve_cua_env(self, context):
         DeployedEnv metadata, so its presence authoritatively marks a CUA env."""
         deployed_env = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if deployed_env is None:
-            raise RuntimeError(
+            raise StepContextError(
                 f"Env '{self.env_id}' not found in context.deployed_envs — cannot collect artifacts via controller"
             )
         if _CUA_ENV_METADATA_MARKER not in (deployed_env.metadata or {}):
@@ -758,7 +758,7 @@ async def _collect_via_agent_container(self, context, store, artifact_id, versio
 
         agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
         if agent is None:
-            raise RuntimeError(f"Agent '{self.agent_name}' not found in context.deployed_agents")
+            raise StepContextError(f"Agent '{self.agent_name}' not found in context.deployed_agents")
         if not agent.sandbox_id:
             raise RuntimeError(f"Agent '{self.agent_name}' has no sandbox_id — cannot collect artifacts")
 
@@ -804,7 +804,7 @@ async def _collect_via_vm_host(self, context, store, artifact_id, version):
             (sb for sb in context.deployed_sandboxes if sb.sandbox_name == self.sandbox_name), None
         )
         if ds is None:
-            raise RuntimeError(
+            raise StepContextError(
                 f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes"
             )
 
@@ -862,7 +862,7 @@ async def _collect_via_sandbox_container(self, context, store, artifact_id, vers
             (s for s in context.deployed_sandboxes if s.sandbox_name == self.sandbox_name), None
         )
         if deployed is None:
-            raise RuntimeError(
+            raise StepContextError(
                 f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes"
             )
 

src/agent_env/task_step/task_steps/configure_scheduled_checks.py

@@ -18,7 +18,7 @@
 from agent_env.env.gateway.constants import EXT_NOTIFICATIONS_URI
 from agent_env.env.gateway.notifications import DEFAULT_MAX_ACTIVE, DEFAULT_MIN_INTERVAL
 from agent_env.entity.refs import EntityRef
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import StepContextError, TaskStepContext
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 
 logger = logging.getLogger(__name__)
@@ -77,7 +77,7 @@ def from_dict(cls, data: dict) -> ConfigureScheduledChecksTaskStep:
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
         deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if deployed is None:
-            raise RuntimeError(f"Env '{self.env_id}' not found in context.deployed_envs")
+            raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
         if not deployed.supports(EXT_NOTIFICATIONS_URI, "config"):
             raise RuntimeError(f"Env '{self.env_id}' runs a gateway without {EXT_NOTIFICATIONS_URI}; redeploy it on a "
                                "gateway image that has it.")

src/agent_env/task_step/task_steps/deploy_agent.py

@@ -40,7 +40,7 @@
     sandbox_request_headers_for_url,
 )
 from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy
-from agent_env.task_step.context import DeployedAgent, TaskStepContext
+from agent_env.task_step.context import DeployedAgent, StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.entity.scopes import FieldScope
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
@@ -356,13 +356,13 @@ async def _execute(self, context: TaskStepContext, deployed_agents: list) -> Tas
 
         existing = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
         if existing is not None:
-            raise RuntimeError(f"Agent with name '{self.agent_name}' is already deployed")
+            raise StepContextError(f"Agent with name '{self.agent_name}' is already deployed")
 
         linked_sandbox = None
         if self.sandbox_name:
             ds = next((s for s in context.deployed_sandboxes if s.sandbox_name == self.sandbox_name), None)
             if ds is None:
-                raise RuntimeError(
+                raise StepContextError(
                     f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes; "
                     f"deploy it with DeploySandboxTaskStep before this step"
                 )

src/agent_env/task_step/task_steps/deploy_env.py

@@ -11,7 +11,7 @@
 from agent_env.config import get_config
 from agent_env.env.gateway import GatewayMode
 from agent_env.env.store import update_env_instance_metadata
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.entity.scopes import FieldScope
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
@@ -155,7 +155,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
 
         existing = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if existing is not None:
-            raise RuntimeError(f"Env '{self.env_id}' is already deployed")
+            raise StepContextError(f"Env '{self.env_id}' is already deployed")
 
         env = Env.get(self.env_id, self.env_version)
         user_overrides = context.metadata.get("user_overrides", {})

src/agent_env/task_step/task_steps/deploy_human_agent.py

@@ -10,7 +10,7 @@
 
 import httpx
 
-from agent_env.task_step.context import DeployedAgent, TaskStepContext
+from agent_env.task_step.context import DeployedAgent, StepContextError, TaskStepContext
 from agent_env.entity.scopes import FieldScope
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 
@@ -60,7 +60,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 "only one human agent is allowed per task"
             )
         if any(a.agent_name == self.agent_name for a in context.deployed_agents):
-            raise RuntimeError(f"Agent with name '{self.agent_name}' is already deployed")
+            raise StepContextError(f"Agent with name '{self.agent_name}' is already deployed")
 
         if self.a2a_url:
             url = self.a2a_url

src/agent_env/task_step/task_steps/deploy_sandbox.py

@@ -8,7 +8,7 @@
 from agent_env.artifact.artifacts.docker_image import pull_problem
 from agent_env.providers.sandbox_providers.chained_sandbox_provider import skip_refusing
 from agent_env.providers.sandbox_providers.sandbox import NetworkPolicy
-from agent_env.task_step.context import DeployedSandbox, TaskStepContext
+from agent_env.task_step.context import DeployedSandbox, StepContextError, TaskStepContext
 from agent_env.entity.scopes import FieldScope
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 from agent_env.attribution import deploy_attribution
@@ -113,7 +113,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         from agent_env.providers.sandbox_providers.sandbox_provider import build_sandbox_provider, get_sandbox_provider
 
         if any(s.sandbox_name == self.sandbox_name for s in context.deployed_sandboxes):
-            raise RuntimeError(f"Sandbox '{self.sandbox_name}' is already deployed")
+            raise StepContextError(f"Sandbox '{self.sandbox_name}' is already deployed")
 
         user_overrides = context.metadata.get("user_overrides") or {}
         # Moves only deploy_sandbox steps: deploy_env and deploy_agent read env_sandbox and agent_sandbox.

src/agent_env/task_step/task_steps/env_card_validator/verify_env_card.py

@@ -5,7 +5,7 @@
 import logging
 from typing import ClassVar, Optional
 
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 
@@ -48,7 +48,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
 
         deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if deployed is None:
-            raise RuntimeError(f"Env '{self.env_id}' not found in context.deployed_envs")
+            raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
 
         card = None
         error = None

src/agent_env/task_step/task_steps/env_card_validator/verify_env_core_protocol.py

@@ -9,7 +9,7 @@
 from agentenv_protocol import METHOD_ADD, METHOD_GET, METHOD_RESET, RPC_PATH
 from agentenv_protocol import client as protocol_v1
 
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 
@@ -54,7 +54,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
 
         deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if deployed is None:
-            raise RuntimeError(f"Env '{self.env_id}' not found in context.deployed_envs")
+            raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
 
         card = None
         try:

src/agent_env/task_step/task_steps/install_agent.py

@@ -33,7 +33,7 @@
     trust_env,
 )
 from agent_env.store.base import NotFoundError
-from agent_env.task_step.context import DeployedAgent, TaskStepContext
+from agent_env.task_step.context import DeployedAgent, StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 
@@ -226,7 +226,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
 
         existing = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
         if existing is not None:
-            raise RuntimeError(f"Agent with name '{self.agent_name}' is already deployed")
+            raise StepContextError(f"Agent with name '{self.agent_name}' is already deployed")
 
         if self.container_name is not None:
             containers = context.metadata.get("deployed_docker_containers", [])
@@ -246,7 +246,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
 
         ds = next((s for s in context.deployed_sandboxes if s.sandbox_name == self.sandbox_name), None)
         if ds is None:
-            raise RuntimeError(f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes")
+            raise StepContextError(f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes")
         if ds.sandbox_mode != SANDBOX_MODE_VM:
             raise RuntimeError(
                 f"InstallAgent requires a VM-mode sandbox; '{self.sandbox_name}' is mode={ds.sandbox_mode!r}"

src/agent_env/task_step/task_steps/load_artifact.py

@@ -18,7 +18,7 @@
 from agent_env.providers.sandbox_providers.local_docker.provider import LocalSandbox
 from agent_env.providers.sandbox_providers.sandbox import into_container
 from agent_env.providers.sandbox_providers.sandbox_provider import reachable_url
-from agent_env.task_step.context import DeployedAgent, TaskStepContext
+from agent_env.task_step.context import DeployedAgent, StepContextError, TaskStepContext
 from agent_env.task_step.preflight import stored_env
 from agent_env.entity.refs import EntityRef
 from agent_env.entity.scopes import FieldScope
@@ -322,7 +322,7 @@ def _resolve_step_artifact(context: TaskStepContext, step_id: str) -> Optional[d
                 continue
             ref = extract(entry)
             return {"id": ref["id"], "version": ref.get("version")} if ref else None
-        raise RuntimeError(
+        raise StepContextError(
             f"No artifact recorded by step '{step_id}'. Ensure a step that produces "
             f"one (e.g. `collect_artifacts`, `snapshot_env`) ran with that id before "
             f"this load step."
@@ -447,7 +447,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
         if self.env_id is not None:
             deployed = self._deployed_env(context)
             if deployed is None:
-                raise RuntimeError(f"Env '{self.env_id}' not found in context.deployed_envs")
+                raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
 
         inputs = self._resolve_inputs(context)
         resolved_artifacts = inputs.artifacts
@@ -474,7 +474,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                     )
             ds = next((s for s in context.deployed_sandboxes if s.sandbox_name == self.sandbox_name), None)
             if ds is None:
-                raise RuntimeError(f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes")
+                raise StepContextError(f"Sandbox '{self.sandbox_name}' not found in context.deployed_sandboxes")
             provider = (
                 build_sandbox_provider(ds.sandbox_type) if ds.sandbox_type else get_sandbox_provider()
             )
@@ -545,7 +545,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                 else:
                     agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
                     if agent is None or not agent.instance_id:
-                        raise RuntimeError(f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing instance_id)")
+                        raise StepContextError(f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing instance_id)")
                     deployed_agent = get_a2a_agent_instance_store().get(agent.instance_id)
                     destination = (destination_path or "/tmp/file_artifacts").rstrip("/") or "/"
                     files = await A2AAgent.load_file_artifact_universe(
@@ -582,7 +582,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                     else:
                         agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
                         if agent is None or not agent.sandbox_id:
-                            raise RuntimeError(
+                            raise StepContextError(
                                 f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing sandbox_id)"
                             )
                         provider = (
@@ -641,7 +641,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
                     raise RuntimeError("env_id is required when loading a CliArtifact (for gateway_url)")
                 agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
                 if agent is None or not agent.instance_id:
-                    raise RuntimeError(f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing instance_id)")
+                    raise StepContextError(f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing instance_id)")
                 deployed_agent = get_a2a_agent_instance_store().get(agent.instance_id)
                 gateway_url = require_gateway_url(deployed, "Installing a CliArtifact")
                 # A local sandbox's host is this machine, which reaches the gateway as it was issued.
@@ -664,7 +664,7 @@ async def execute(self, context: TaskStepContext) -> TaskStepContext:
             else:
                 agent = next((a for a in context.deployed_agents if a.agent_name == self.agent_name), None)
                 if agent is None or not agent.sandbox_id:
-                    raise RuntimeError(f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing sandbox_id)")
+                    raise StepContextError(f"Agent '{self.agent_name}' not found in context.deployed_agents (or missing sandbox_id)")
                 provider = (
                     build_sandbox_provider(agent.sandbox_type)
                     if agent.sandbox_type

src/agent_env/task_step/task_steps/mcp_cli_builder/build_mcp_cli.py

@@ -19,7 +19,7 @@
 from agent_env.env import Env
 from agent_env.env.env import require_gateway_url, website_child
 from agent_env.store.ids import derive_id
-from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.context import StepContextError, TaskStepContext
 from agent_env.entity.refs import EntityRef, RefRole
 from agent_env.task_step.task_step import TaskStep, TaskStepDependency
 from agent_env.task_step.thread_work import finish_on_thread
@@ -160,7 +160,7 @@ def from_dict(cls, data: dict) -> BuildMcpCliTaskStep:
     async def execute(self, context: TaskStepContext) -> TaskStepContext:
         deployed = next((d for d in context.deployed_envs if d.env_id == self.env_id), None)
         if deployed is None:
-            raise RuntimeError(f"Env '{self.env_id}' not found in context.deployed_envs")
+            raise StepContextError(f"Env '{self.env_id}' not found in context.deployed_envs")
         gateway_url = require_gateway_url(deployed, "build_mcp_cli")  # the CLI it generates talks to the gateway
 
         env = Env.get(self.env_id, deployed.env_version)