Tasks and steps

A task is a named, versioned list of TaskSteps stored inline in one document, and a step is the unit of work agent-env executes: deploy an env, stage data into it, prompt an agent, grade what came back. This page assumes the vocabulary from Core concepts. The one idea to carry into everything below: a step's return value is ignored. execute() is declared -> TaskStepContext and hands back the same object it was given, but Task.run() discards it. Steps communicate instead by mutating a single TaskStepContext that agent-env threads by reference through every step in the run — the same object each step is handed and hands back.

Anatomy of a task

agent-env task create reads a JSON list of step dicts. Each entry needs an id unique within the task and a type naming a registered step class. Every other key is passed straight to that class's constructor, so a misspelled field fails at create time rather than twenty minutes into a run.

[
  {
    "id": "deploy",
    "type": "deploy_env",
    "env_id": "support-desk",          // an env you registered earlier
    "ttl_seconds": 3600,               // the sandbox self-terminates after this
    "disk_size_gb": 20,
    "depends_on": []                   // no dependencies: launches immediately
  },
  {
    "id": "load-universe",
    "type": "load_artifact",
    "env_id": "support-desk",          // target: restore into the env instance
    "artifact_id": "support-desk-universe",
    "artifact_version": 7,             // omit for the latest version
    "depends_on": [{ "task_step_id": "deploy" }]
  },
  {
    "id": "agent",
    "type": "deploy_agent",
    "env_ids": ["support-desk"],       // points the agent's MCP client at this env
    "a2a_agent_id": "my-coding-agent",
    "agent_name": "solver",            // the join key every later step uses
    "depends_on": [{ "task_step_id": "load-universe" }]
  },
  {
    "id": "ask",
    "type": "prompt_agent",
    "agent_name": "solver",            // must match a deploy_agent above
    "prompt_id": "p1",                 // the join key verifiers use
    "prompt": "Refund the most recent order for Acme Corp and reply with the refund id.",
    "timeout_seconds": 900,
    "depends_on": [{ "task_step_id": "agent" }]
  },
  {
    "id": "grade",
    "type": "rubrics_verifier",
    "prompt_id": "p1",                 // must match a prompt_agent above
    "verifier_id": "refund-rubric",    // the key under metadata["verifications"]
    "score_aggregator": "weighted_average",
    "criteria": [
      { "id": "c1", "criterion": "A refund exists for Acme Corp's most recent order.", "weight": 1.0 },
      { "id": "c2", "criterion": "The reply contains the refund id.", "weight": 1.0 }
    ],
    "depends_on": [{ "task_step_id": "ask" }]
  }
]

The comments are for this page; agent-env task create reads strict JSON.

Creating the task wraps that list in the stored document — id, type: "task", version, steps, project_id — and the store assigns version itself, starting at 1 and incrementing on every write. Writing under an existing id adds a version; it never overwrites one, and Task.get("<id>") with no version returns the highest.

Four fields come from the base class and exist on every step type. Three of them you author; the fourth, version, defaults to null when you omit it and is stored on every step.

Execution

Task.run() builds a DAG from the steps' depends_on edges and walks it. Every step whose dependencies have all completed is launched immediately, so independent branches run concurrently on one event loop.

How a step's edges are derived depends on whether depends_on is present at all, and the two absent cases are opposites:

Edges may only point backwards. Task validates the DAG when it is constructed, rejecting a duplicate step id and a depends_on that names a step which does not appear earlier in the list. Both raise ValueError before anything is stored or run.

Before the first step, the run registers a task instance: a document carrying instance_id, task_id, task_version, status: "running", current_step, total_steps, and the initial context. context.instance_id is set to that id, so a step can reference its own run. After each step, agent-env diffs the context against a snapshot taken before that step ran and writes only the paths it changed, plus an entry in completed_steps. current_step is derived as the number of completed steps, and status flips to completed when that reaches total_steps. Writing only changed paths is what lets concurrent siblings update the same instance document without clobbering each other.

Task.run() returns the final context. start_step and end_step are 0-based indexes that narrow the slice actually executed; steps before start_step are marked complete without running, which is how you re-grade a saved context without redeploying. See Iterating on a task.

Sharing state between steps

TaskStepContext is one mutable object handed to every execute() call by reference. A step publishes its result by appending to a typed list on the context or by writing a key into context.metadata. Four typed lists carry the deployments and responses that later steps join against:

Everything else — verification results, loaded universes, collected files, snapshot handles — lands under a well-known key in the free-form metadata dict.

Follow one value through the example task. deploy_env deploys support-desk and appends a DeployedEnv carrying the live gateway_url, mcp_url, and sandbox_id to context.deployed_envs. deploy_agent looks up env_ids: ["support-desk"] in that same list and posts the env instance's MCP URL to the agent's MCP-config extension — the moment the agent gains tools — then appends its own DeployedAgent under agent_name: "solver". Finally prompt_agent does:

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 with name '{self.agent_name}' not found in context.deployed_agents")

and sends the prompt to agent.a2a_url, falling back to agent.api_url for agents that advertise no A2A URL. The result appends a PromptResponse under prompt_id: "p1", which rubrics_verifier looks up the same way before writing its verdict to context.metadata["verifications"]["refund-rubric"] as {format, results, score} plus the trajectory pointers compact_trajectory_s3_uri and judge_trajectory_s3_uri.

Several layers can set what a step actually uses at execute time:

Step families

The registry carries several dozen built-in step types, in seven families. The table names representatives, not the full list; every type, with its fields and defaults, is in the generated Task step reference.

The registry is open: the [task_steps] table in your config file takes a list of impl pointers ("module.path:ClassName"), each of which must subclass TaskStep and declare a type that does not collide with a built-in. See the configuration reference for the key and Custom types for the class contract.

Failure handling

When a step raises, agent-env appends a record to context.metadata["failed_steps"] with step_id, step_type, error, error_type, started_at_utc, duration_seconds, and is_fatal. What happens next is fail_task_on_error:

The second row is the sharp edge. agent-env does not check that a tolerated step produced what its dependents need, so a load_artifact marked fail_task_on_error: false that fails leaves an empty env instance, the prompt_agent behind it runs against that, and a verifier scores the result. Use it for steps whose output is genuinely optional, and read metadata["failed_steps"] afterwards.

Secrets never reach the store: the context is redacted on the way out, recursively stripping litellm_api_key, usersim_api_key, remote_tokens, and cf_access_client_secret.

Running a task

Save the step list as task.json, then:

# Validates every step, runs preflight, and writes nothing if preflight rejects.
agent-env task create task.json --id refund-demo --project-id <project-id>

# Runs the latest version and writes the final context as JSON.
agent-env task run --id refund-demo --output-dir ./out

The Python equivalent, which is what the CLI calls:

import asyncio
from agent_env.task import Task

task = Task.get("refund-demo")                    # latest version
context = asyncio.run(task.run())                 # returns the final TaskStepContext

print(context.metadata["verifications"]["refund-rubric"]["score"])

Task.run() also takes start_step, end_step, an existing context, and per-step callbacks. For a step-by-step build of this task see Your first task; for re-running one step against a saved context see Iterating on a task.

Related docs

• Core concepts — the vocabulary this page uses, and where a task sits in a run
• Your first task — build and run the five-step task above from scratch
• Iterating on a task — re-run a slice of the DAG against a saved context
• Task step reference — every registered step type and its fields
• How grading works — the shared verifications contract and the verifier family
• Configuration reference — [task_steps], [model], and the precedence rules
• Custom types — writing and registering your own step