# Task Step Plugins (https://www.agentenvframework.com/docs/plugins/task-step-plugins)

> Write task steps in a package, register them with an entry point and run them in a task, with OpenCiv3's match and recording steps as the example

`agentenv_openciv3.steps` ships three step types: `openciv3_match` seats the agents a run deployed,
and any people, in a new game; `openciv3_await_game` waits for a game people play to end; and
`save_env_recording` stores the game's recording. This page follows the first and the last, which run
against the `openciv3` env, or `openciv3-live` from [Environment Plugins](https://www.agentenvframework.com/docs/plugins/environment-plugins.md).

[![Opus, Sonnet and Haiku, seated by openciv3_match, play one 300-turn game of OpenCiv3: the whole map, the scoreboard and the score chart](https://www.agentenvframework.com/plugins/openciv3-three-agents-poster.jpg)](https://www.agentenvframework.com/plugins/openciv3-three-agents.mp4)

## When to Write a Step

Write a step when a task needs something done between deploy and teardown that no step in
[More task steps](https://www.agentenvframework.com/docs/tasks/more-steps.md) does. `apply_server_config` can start a game with fixed
args, as the `smoke` task does, but can't seat the deployed agents, and no built-in step stores the
files a recording extension returns.

## The Step Contract

A step is a `TaskStep` subclass, as [Your own step](https://www.agentenvframework.com/docs/tasks/task-steps.md#your-own-step) shows.
`save_env_recording` has fields of its own and passes the base fields to `super().__init__`:

```python title="src/agentenv_openciv3/steps.py (trimmed)"
class SaveEnvRecordingTaskStep(TaskStep):
    type: ClassVar[str] = "save_env_recording"
    entity_refs = (EntityRef.env("env_id"),)

    def __init__(self, id: str, version: int | None, env_id: str, extension_uri: str = RECORDING_EXTENSION,
                 formats: list[str] | None = None, view: str = "spectator", timeout_seconds: int = 300,
                 depends_on: list | None = None, fail_task_on_error: bool = False):
        super().__init__(id, version, depends_on=depends_on, fail_task_on_error=fail_task_on_error)
        self.env_id = env_id
        # and the other four fields

    def to_dict(self) -> dict:
        return {**super().to_dict(), "env_id": self.env_id, "extension_uri": self.extension_uri,
                "formats": self.formats, "view": self.view, "timeout_seconds": self.timeout_seconds}

    @classmethod
    def from_dict(cls, data: dict) -> SaveEnvRecordingTaskStep:
        return cls(**{**cls._base_from_dict(data), "fail_task_on_error": data.get("fail_task_on_error", False)},
                   env_id=data["env_id"], extension_uri=data.get("extension_uri", RECORDING_EXTENSION),
                   formats=data.get("formats"), view=data.get("view", "spectator"),
                   timeout_seconds=data.get("timeout_seconds", 300))
```

`type` is the string task JSON names and every stored step document carries, so the class needs its
own. `_base_from_dict` returns `id`, `version`, `depends_on` and `fail_task_on_error`, defaulting
the last to `true`; this step defaults it to `false`, so a failed recording never fails the task.
Core attaches `retry_config` itself, so the constructor does not take it.

## Preflight

`preflight()` returns one message per problem it can find before a run, and must not need a sandbox
or a deployed env. The base class returns `[]`, and neither step overrides it: `deploy_env`'s
preflight already loads the env, and the agents `openciv3_match` seats exist only once deployed, so
its `execute` checks them:

```text title="Output (civs names haiku, which wasn't deployed)"
RuntimeError: agents ['haiku'] are not deployed in this run (deployed: ['opus', 'sonnet'])
```

## Declare the Env Reference

`entity_refs` names the fields that hold an env, agent or artifact id, so a tool that rewrites ids
before a run, such as the bundle resolver, can find them. It is not inherited, and when a bundle
passes one of its own names to a step type that leaves it `None`, the resolver reports that the type
`doesn't declare its entity_refs, so it can't be rewritten`. Nothing reads it at run time.

## Reach the Deployed Env

Each step finds the env instance in `context.deployed_envs`, where `deploy_env` put it, then the card
that advertises the extension it needs: the env's own, or a child's behind a gateway.

```python title="src/agentenv_openciv3/steps.py"
def _extension_card(deployed: DeployedEnv, uri: str) -> dict:
    own = deployed.environment_card or {}
    card = next((c for c in [own, *(own.get("children_environments") or [])] if client.find_extension(c, uri)), None)
    if card is None:
        raise RuntimeError(f"env {deployed.env_id!r} does not advertise {uri}")
    return card
```

Rather than call MCP tools as an agent would, it passes that card to `client.invoke_extension`, which
reads the extension's endpoint from it. Core derives `environment_url` from the card URL the
[environment provider](https://www.agentenvframework.com/docs/plugins/environment-provider-plugins.md) recorded, so the step knows
nothing about the machine behind it.

## Store the Recording

`execute` asks the `urn:openciv3:recording/v1` extension for the game's MP4 and HTML replay, stores
each as a `file` artifact, and records them in `metadata["recordings"]` by step id. The bytes go to
the object store and only artifact ids to the context, which is persisted after every step.

```python title="src/agentenv_openciv3/steps.py (trimmed)"
    async def execute(self, context: TaskStepContext) -> TaskStepContext:
        deployed = _deployed_env(context, self.env_id)
        card = _extension_card(deployed, self.extension_uri)
        args = {"view": self.view, **({"formats": self.formats} if self.formats is not None else {})}
        result = await client.invoke_extension(deployed.environment_url, card, self.extension_uri, args,
                                               timeout=self.timeout_seconds)
        stem = f"{context.metadata.get('task_id', self.env_id)}-recording-{context.instance_id or uuid.uuid4().hex}"
        saved = []
        for f in result["files"]:
            content = base64.b64decode(f["base64"])
            artifact = await asyncio.to_thread(
                FileArtifact.put_bytes, f"{stem}.{f['name'].partition('.')[2]}",
                description=f"Recording of env {self.env_id!r}: {f['name']}", filename=f["name"],
                content=content, content_type=f["content_type"])
            saved.append({"name": f["name"], "artifact_id": artifact.id, "version": artifact.version,
                          "bytes": len(content), "content_type": f["content_type"]})
        context.metadata.setdefault("recordings", {})[self.id] = saved
        return context
```

`put_bytes` is synchronous, so the step runs it in a thread to keep other steps of the run moving.
With the store from [Registry Plugins](https://www.agentenvframework.com/docs/plugins/registry-plugins.md), a registered task's object URLs start `cas://openciv3/`.

## Write a Verifier

A verifier is an ordinary step that writes `metadata["verifications"][verifier_id]` with a `score`
from 0 to 1 and the `results` rows behind it. The plugin writes none: its tasks use the built-in
`env_outcome_verifier`, which runs `verify(mcp_url)` from a script the bundle ships as a file
artifact, such as `victor-verifier`, which names a match's winner.

```json title="tasks/three-agents-quick.json (the grade step)"
{"id": "grade", "type": "env_outcome_verifier", "env_id": "openciv3", "file_artifact_id": "victor-verifier",
 "verifier_id": "victor", "score_aggregator": "weighted_average", "depends_on": ["opus", "sonnet", "haiku"]}
```

`agent-env task run` prints the rows and `score` of every entry in `verifications`, so a plugin
verifier needs no output code of its own.

## Register the Entry Points

Each step is an entry point in the `agent_env.task_steps` group, named after its `type`:

```toml title="pyproject.toml"
[project.entry-points."agent_env.task_steps"]
openciv3_match = "agentenv_openciv3.steps:OpenCiv3MatchTaskStep"
openciv3_await_game = "agentenv_openciv3.steps:OpenCiv3AwaitGameTaskStep"
save_env_recording = "agentenv_openciv3.steps:SaveEnvRecordingTaskStep"
```

The name must equal the `type`, because documents are written and read by it. A class under another
name, one that keeps the base `type`, or one that leaves `execute` or `from_dict` unimplemented is
reported `failed` (`invalid-plugin`) and skipped, with a reason such as
`OpenCiv3MatchTaskStep has type 'openciv3_match'; its entry point must be named 'openciv3_match'`.

## Confirm the Plugin Loaded

`agent-env plugin show` lists each contribution with its status:

```text title="agent-env plugin show agentenv-openciv3 (trimmed)"
agentenv-openciv3 0.1.0
  task step    openciv3_await_game  agentenv_openciv3.steps:OpenCiv3AwaitGameTaskStep  active
  task step    openciv3_match       agentenv_openciv3.steps:OpenCiv3MatchTaskStep      active
  task step    save_env_recording   agentenv_openciv3.steps:SaveEnvRecordingTaskStep   active
```

```text title="agent-env plugin check"
ok: 2 plugin package(s), 6 contribution(s), all in effect
```

## Use the Steps in a Task

The bundle's `three-agents-quick` task seats three agents for 10 turns, then grades the game and
saves its recording in parallel. One agent's steps are shown:

```json title="tasks/three-agents-quick.json (trimmed)"
[
  {"id": "deploy", "type": "deploy_env", "env_id": "openciv3", "ttl_seconds": 7200},
  {"id": "agent-opus", "type": "deploy_agent", "agent_name": "opus", "env_ids": ["openciv3"], "depends_on": ["deploy"]},
  {"id": "match", "type": "openciv3_match", "env_id": "openciv3", "turns": 10,
   "civs": {"opus": "Rome", "sonnet": "Greece", "haiku": "Egypt"}, "depends_on": ["agent-opus", "agent-sonnet", "agent-haiku"]},
  {"id": "opus", "type": "prompt_agent", "agent_name": "opus", "model": "anthropic/claude-opus-5-5",
   "prompt": "You lead a civilization in a game of OpenCiv3, ...", "depends_on": ["match"]},
  {"id": "recording", "type": "save_env_recording", "env_id": "openciv3", "timeout_seconds": 1200,
   "depends_on": ["opus", "sonnet", "haiku"]}
]
```

`agent-env task create` runs each step's `preflight()`, so an unregistered env stops the create
before anything is saved:

```text title="agent-env task create --id three-agents-quick --project-id demo three-agents-quick.json"
Preflight found 1 problem(s):
  - deploy_env 'deploy': env 'openciv3' can't be loaded: Env openciv3 not found
Nothing was saved. Re-run with --skip-validation to save anyway.
```

`agent-env run openciv3 --task three-agents-quick` runs it from the bundle; it passed with grade 1
in 182.4s. The `smoke` task needs no model and ends with the same `save_env_recording` step:

```text title="agent-env run openciv3 --task smoke (trimmed)"
[tasks/smoke.json] step 5/5 recording (save_env_recording)
[tasks/smoke.json] step 5/5 recording done in 4.1s
[tasks/smoke.json] passed in 8.7s

Tasks:
  tasks/smoke.json v1: passed (grade: 1), 8.7s, instance @local/agentenv-openciv3/openciv3/smoke-49qpygc4

Tore down 1 sandbox.
```

Its context holds what the step wrote:

```json title="metadata (trimmed)"
{
  "recordings": {
    "recording": [
      {"name": "openciv3-seed7.mp4", "artifact_id": "@local/agentenv-openciv3/openciv3/smoke-recording-@local/agentenv-openciv3/openciv3/smoke-49qpygc4.mp4", "version": 1, "bytes": 184289, "content_type": "video/mp4"},
      {"name": "openciv3-seed7.html", "artifact_id": "@local/agentenv-openciv3/openciv3/smoke-recording-@local/agentenv-openciv3/openciv3/smoke-49qpygc4.html", "version": 1, "bytes": 933081, "content_type": "text/html"}
    ]
  }
}
```