feat(task): run a task with some of its agents played from your own loop

AGENTS.md

@@ -92,7 +92,7 @@ and deserialized through a registry.
 |---|---|
 | `env/` | `Env` base class: a subclass declares `type`, implements `from_dict` and `async deploy(...) -> DeployedEnv`, and may override `reset` and `load_file_artifact_universe`. Built-ins in `env/envs/`: `MCPServerEnv` (`mcp_server`), `MultiEnv` (`multi`, MCP servers plus websites), `ServiceDBEnv` (`service_db`, PostgreSQL), `WebsiteEnv` (`website`), `GatewayEnv` (`gateway_server`). `env/gateway/gateway.py` is the env-side gateway server, container-only code. |
 | `artifact/` | Immutable versioned artifacts; every `put()` creates a new version. The document goes to the document store, binary payloads to the object store. Built-ins: `CliArtifact` (`cli`), `FileArtifact` (`file`), `FileArtifactUniverse` (`file_artifact_universe`), `DockerImageArtifact` (`docker_image`), `EnvironmentArtifact` (`environment`; legacy `service`), `EnvironmentUniverseArtifact` (`environment_universe`; legacy `service_universe`), `SkillArtifact` (`skill`). A `DockerImageArtifact` holds a tar.gz of its image, or, written by `put_ref` (`--image-ref` on `env mcp-server put` and `a2a-agent put`), only a registry reference with its tag pinned to a digest, which sandboxes pull, or, written by `put_context`, only its build context, which a VM sandbox builds before running it (`VmSandbox.load_docker_images`) and Modal builds before creating its container (`ModalSandboxProvider.prepare_image`, outside a fallback chain's deadline); anything else that runs images by name (a container on the local provider, a lone server on any provider but Modal) refuses such an image. A build context (`utils/build_context.py`) is what `docker build` would send, `.dockerignore` applied, written so equal files give equal bytes and an equal `source_digest`. |
-| `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, and more). |
+| `task/`, `task_step/` | A `Task` holds its `TaskStep`s inline; `Task.run()` executes them as a DAG. `depends_on` (None means all prior steps) gates a step, independent steps run concurrently, `fail_task_on_error` makes a failure fatal or tolerated, `retry_config` rolls a failed span back through the step journal and re-dispatches it. Built-in steps live in `task_step/task_steps/` (`deploy_env`, `deploy_agent`, `prompt_agent`, the verifiers under `verifiers/`, and more). `task/episode/` runs a task with some of its agents played from outside, as a training loop does (`EpisodeSpec`). |
 | `store/` | Four store ABCs with local and cloud implementations: `DocumentStore` (SQLite, MongoDB), `ObjectStore` (filesystem, S3, Cloud Storage), `ImageStore` (local OCI registry, ECR), `SecretStore` (env vars or file, AWS Secrets Manager, Google Cloud Secret Manager). `VersionedEntityStore` implements the shared versioned get/put logic, `QueryBuilder` is the immutable chained query API, `store/base.py` holds the error types. A new backend must pass the conformance kits in `tst/store/`. |
 | `config/` | The `Config` singleton (`get_config`, `configure`, `reset_config`) in `config/runtime.py`, file discovery in `config/loader.py`, and `load_impl`, which resolves `module:Class` pointers. `agent_env.store` re-exports the config names for compatibility. |
 | `providers/` | `providers/sandbox_providers/` holds the sandbox providers `local`, `modal`, `modal_vm`, `e2b`, `sail_vm` (the `sail` extra); `[sandbox] default` and `agent_default` accept a comma-separated fallback chain, whose deploys skip a provider that can't run them. A sandbox provider declares what it runs (`ON_THIS_MACHINE`, `CONTAINER_ACCEPTS`, `PRIVATE_NETWORK`; `SUPPORTED_SANDBOX_MODES`, derived from whether it implements `create_sandbox` and `create_vm`; and `SANDBOX_ACCEPTS`, the images an agent's sandbox runs, which defaults to those a VM loads when it implements `create_vm` and to `CONTAINER_ACCEPTS` otherwise, so only a provider that implements `create_vm` yet gives an agent a container, like the local one, declares it), each sandbox carries the `platform` of the images it runs (linux/amd64 unless its provider sets another when it creates it; None, any, unchecked, for a local sandbox), and code outside `providers/sandbox_providers/` reads those, never which provider it has. `providers/env_providers/` holds the environment providers: `EnvironmentProvider` (an env's containers and state store) and `EnvironmentGatewayProvider`, which renders a docker-compose for the gateway and its MCP servers inside the sandbox; `providers/env_state/` holds env-state providers (`local_postgres` built in). |
@@ -155,6 +155,12 @@ orchestrators call it the same way. What callers rely on:
   a context with `TaskStepContext.to_dict()`, not `dataclasses.asdict`.
 - A run is resumable: pass a previously persisted `context`, a `start_step` and the existing
   `instance_id`; completed steps are seeded as SUCCESS.
+- An episode (`EpisodeSpec`, `task/episode/`) is one `Task.run()` of a copy of the task in which
+  the `deploy_agent` and `prompt_agent` steps of each of its `trainer_agents` are the plain-data
+  `episode_agent` and `episode_turn`, which wait for the caller instead of deploying or prompting. Its
+  record is tagged `context.metadata["episode"]` and names the original task, so it isn't resumable
+  with `start_step`. The episode tears down what its run deployed, and an abandoned one is recorded
+  `cancelled`.
 
 ## Tests
 

README.md

@@ -44,6 +44,24 @@ uv sync --extra dev
 source .venv/bin/activate
 ```
 
+## Run a task from your own loop
+
+An episode is one run of a task in which some of its agents are played from outside AgentEnv, by a training loop's model or any code of your own. AgentEnv runs the rest exactly as `agent-env run` does, deploying the envs, running the other agents and scoring, and waits at each of your agent's turns:
+
+```python
+from agent_env.task import EpisodeSpec, Task
+
+spec = EpisodeSpec.from_task(Task.get("my-task"), turn_timeout=900)
+episode = await spec.start()
+while (turn := await episode.next_turn()) is not None:
+    async with turn.tools() as tools:  # the agent's env tools, called as its role
+        ...  # your model reads turn.system_prompt and turn.prompt, and acts with tools.list() and tools.call()
+    await turn.end(response="Done.")
+scores = await episode.result()  # every score the run recorded, by step id
+```
+
+The scores are the ones `agent-env run` reports, so making a reward of them is yours. `trainer_agents=` names the agents you play when the task prompts several, `turn_timeout` is each turn's deadline in seconds (the prompt's own `timeout_seconds` otherwise), `episode.abandon(reason)` stops a run early, and `from_task` names any step an outside player can't stand behind. [Running tasks](https://www.agentenvframework.com/docs/tasks/running) covers the rest.
+
 ## Plugins
 
 A package of your own can add env types, providers, task steps and stores to agent-env, and `agent-env plugin` lists, checks, adds and removes installed plugins. The [plugin docs](https://www.agentenvframework.com/docs/plugins) show how.

src/agent_env/a2a_agent/store.py

@@ -3,7 +3,7 @@
 
 import dataclasses
 import logging
-import random
+import secrets
 import string
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Optional, Self
@@ -160,7 +160,7 @@ def _doc_store(self):
 
     def create_instance(self, deployed: "DeployedA2AAgent", ttl_seconds: int) -> "DeployedA2AAgent":
         from agent_env.a2a_agent.a2a_agent import DeployedA2AAgent
-        suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+        suffix = "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(8))
         instance_id = f"{deployed.agent_id}-{suffix}"
         now = datetime.now(timezone.utc)
         created_at_utc = now.strftime("%Y-%m-%d %H:%M UTC")

src/agent_env/bundle/run.py

@@ -27,6 +27,7 @@
 from agent_env.store.routing import namespace_routing, run_scope
 from agent_env.task import Task, record_task_cancelled
 from agent_env.task.interrupts import Interrupts
+from agent_env.task.task import recorded_scores
 from agent_env.task.teardown import TeardownReport, teardown_run
 from agent_env.task_step.context import TaskStepContext
 from agent_env.utils.docker_build import DEFAULT_BUILD_PLATFORM
@@ -78,13 +79,7 @@ def failed_steps(self) -> list[dict]:
     def scores(self) -> dict[str, float]:
         """Every score the run recorded, by the id of the step that recorded it, or by its verification's key
         when that isn't a step's verifier id."""
-        steps = {step.verifier_id: step.id for step in self.task.steps if getattr(step, "verifier_id", None)}
-        scores = {}
-        for key, verification in (self.context.metadata.get("verifications") or {}).items():
-            score = verification.get("score")
-            if isinstance(score, (int, float)):
-                scores[steps.get(key, key)] = float(score)
-        return scores
+        return recorded_scores(self.task.steps, self.context)
 
     @property
     def outcome(self) -> Outcome:

src/agent_env/env/store.py

@@ -4,7 +4,7 @@
 
 import dataclasses
 import logging
-import random
+import secrets
 import string
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Optional, Self, TypeVar
@@ -209,7 +209,7 @@ def create_instance(self, deployed_env: DeployedEnv, ttl_seconds: int) -> Deploy
         """Persist a deployed env as an instance record. Returns a new DeployedEnv with instance_id set."""
         from agent_env.env.env import DeployedEnv as DeployedEnvCls
 
-        suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+        suffix = "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(8))
         instance_id = f"{deployed_env.env_id}-{suffix}"
         now = datetime.now(timezone.utc)
         created_at_utc = now.strftime("%Y-%m-%d %H:%M UTC")

src/agent_env/providers/env_state/store.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import logging
-import random
+import secrets
 import string
 from datetime import datetime, timedelta, timezone
 from typing import Optional
@@ -38,7 +38,7 @@ def _doc_store(self):
     def create_instance(self, instance: EnvStateInstance, ttl_seconds: int) -> EnvStateInstance:
         now = datetime.now(timezone.utc)
         if not instance.instance_id:
-            suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
+            suffix = "".join(secrets.choice(string.ascii_lowercase + string.digits) for _ in range(8))
             instance.instance_id = f"esi-{suffix}"
         instance.created_at_utc = now.strftime(_TS_FMT)
         instance.expires_at_utc = (now + timedelta(seconds=ttl_seconds)).strftime(_TS_FMT)

src/agent_env/providers/sandbox_providers/local_sandbox.py

@@ -16,8 +16,8 @@
 import logging
 import os
 import platform
-import random
 import re
+import secrets
 import shlex
 import shutil
 import signal
@@ -139,7 +139,7 @@ def _free_host_port() -> int:
     try:
         fcntl.flock(fd, fcntl.LOCK_EX)
         last = os.read(fd, 16).strip()
-        start = int(last) + 1 if last.isdigit() else random.choice(_HOST_PORTS)
+        start = int(last) + 1 if last.isdigit() else secrets.choice(_HOST_PORTS)
         for i in range(len(_HOST_PORTS)):
             port = _HOST_PORTS[(start + i - _HOST_PORTS.start) % len(_HOST_PORTS)]
             if _bindable(port):

src/agent_env/task/__init__.py

@@ -1,6 +1,7 @@
 """Task module for AgentEnv."""
 
 from .task import Task
+from .episode import Episode, EpisodeError, EpisodeSpec, Turn
 from .store import (
     TaskStore, TaskQuery, get_task_store, set_task_store, reset_task_store,
     TaskInstance, TaskInstanceStore, get_task_instance_store, set_task_instance_store, reset_task_instance_store,
@@ -9,6 +10,10 @@
 
 __all__ = [
     "Task",
+    "Episode",
+    "EpisodeError",
+    "EpisodeSpec",
+    "Turn",
     "TaskStore",
     "TaskQuery",
     "get_task_store",

src/agent_env/task/episode/__init__.py

@@ -0,0 +1,9 @@
+"""Run a task with some of its agents played from outside AgentEnv, by a training loop's model or any other caller.
+
+AgentEnv runs everything else exactly as ``Task.run`` does, so its scores are the ones ``agent-env run`` would report.
+"""
+
+from .run import Episode, EpisodeError, Turn
+from .spec import EpisodeSpec
+
+__all__ = ["Episode", "EpisodeError", "EpisodeSpec", "Turn"]

src/agent_env/task/episode/rules.py

@@ -0,0 +1,160 @@
+"""Which steps a task may have when some of its agents are played from outside AgentEnv, and why the others are refused.
+
+Every registered step type is either checked here (``CHECKS``) or listed as untouched by an outside agent
+(``UNAFFECTED``). A test holds the registry to the two, so a new step type is classified before an episode runs it.
+Plugin step types aren't inspected.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    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.deploy_agent import DeployAgentTaskStep
+    from agent_env.task_step.task_steps.peer_agents import PeerAgentsTaskStep
+    from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep
+    from agent_env.task_step.task_steps.register_env_triggers import RegisterEnvTriggersStep
+    from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep
+
+
+@dataclass(frozen=True)
+class Outside:
+    """The agents played from outside, and their prompts by step id, each with its prompt id."""
+
+    agents: frozenset[str]
+    prompts: dict[str, str]
+
+
+def problems(steps: list[TaskStep], outside: list[str]) -> list[str]:
+    """Why each outside agent, and each step, can't run with those agents played from outside; empty when it can."""
+    found = []
+    for name in outside:
+        deploys = [step for step in steps if step.type == "deploy_agent" and step.agent_name == name]
+        if len(deploys) != 1:
+            found.append(f"agent {name!r} is played from outside, so it needs exactly one deploy_agent step, "
+                         f"not {len(deploys)}")
+    scope = Outside(frozenset(outside), {
+        step.id: step.prompt_id for step in steps if step.type == "prompt_agent" and step.agent_name in outside
+    })
+    for step in steps:
+        why = ["retry_config can replay an outside turn"] if step.retry_config is not None else []
+        if check := CHECKS.get(step.type):
+            why += check(step, scope)
+        found.extend(f"step {step.id!r}: {reason}" for reason in why)
+    return found
+
+
+def _prompt_agent(step: PromptAgentTaskStep, outside: Outside) -> list[str]:
+    if step.agent_name in outside.agents:
+        return [
+            f"an outside agent's prompt can't have {what}" for what, present in (
+                ("parts, only a text prompt", step.parts is not None),
+                ("a user simulator (max_conversation_turns > 1)", step.max_conversation_turns > 1),
+                ("output_format", step.output_format),
+                ("agentenv_tools", step.agentenv_tools),
+                ("snapshot_config", step.snapshot_config),
+            ) if present
+        ]
+    if step.max_conversation_turns > 1 and step.user_agent_name in outside.agents:
+        return [f"its user simulator {step.user_agent_name!r} is played from outside"]
+    return []
+
+
+def _deploy_agent(step: DeployAgentTaskStep, outside: Outside) -> list[str]:
+    if step.agent_name not in outside.agents:
+        return []
+    return [
+        f"an outside agent can't have {what}" for what, present in (
+            ("skills, instructions its player wouldn't get", step.skills),
+            ("agent_snapshot_files_artifact_id", step.agent_snapshot_files_artifact_id),
+            ("agent_changelog_object_url", step.agent_changelog_object_url),
+            ("enable_agent_changelog", step.enable_agent_changelog),
+        ) if present
+    ]
+
+
+def _deploys_an_agent(step: TaskStep, outside: Outside) -> list[str]:
+    if step.agent_name not in outside.agents:
+        return []
+    return [f"{step.type} deploys {step.agent_name!r}, which is played from outside"]
+
+
+def _needs_the_agent(step: TaskStep, outside: Outside) -> list[str]:
+    if step.agent_name not in outside.agents:
+        return []
+    return [f"{step.type} needs {step.agent_name!r} as a real agent, and it is played from outside"]
+
+
+def _validates_an_agent(step: TaskStep, outside: Outside) -> list[str]:
+    return ["validates the task's deployed A2A agent, and an episode plays one from outside"]
+
+
+def _collect_artifacts(step: CollectArtifactsTaskStep, outside: Outside) -> list[str]:
+    why = []
+    if step.agent_name in outside.agents and not (step.env_id or step.container_name or step.sandbox_name):
+        why.append(f"collects from {step.agent_name!r}'s sandbox, and it is played from outside")
+    if step.manifest_step_id in outside.prompts:
+        why.append(f"its manifest is outside prompt {step.manifest_step_id!r}")
+    return why
+
+
+def _register_env_triggers(step: RegisterEnvTriggersStep, outside: Outside) -> list[str]:
+    if step.executor_agent_name not in outside.agents:
+        return []
+    return [f"its executor {step.executor_agent_name!r} is played from outside"]
+
+
+def _peer_agents(step: PeerAgentsTaskStep, outside: Outside) -> list[str]:
+    named = dict.fromkeys(
+        name for peering in step.peerings for name in (peering.source_agent_name, *peering.peer_agent_names)
+    )
+    return [f"peers {name!r}, which is played from outside" for name in named if name in outside.agents]
+
+
+def _rubrics_verifier(step: RubricsVerifierTaskStep, outside: Outside) -> list[str]:
+    why = []
+    if step.use_trajectory and step.prompt_id in outside.prompts.values():
+        why.append("grades an outside prompt's trajectory, which an outside player doesn't hand back")
+    if step.use_agent_judge and step.agent_name in outside.agents:
+        why.append(f"judges in {step.agent_name!r}'s sandbox, and it is played from outside")
+    return why
+
+
+CHECKS: dict[str, Callable[..., list[str]]] = {
+    "prompt_agent": _prompt_agent,
+    "deploy_agent": _deploy_agent,
+    "install_agent": _deploys_an_agent,
+    "deploy_human_agent": _deploys_an_agent,
+    "add_skills": _needs_the_agent,
+    "register_agent_triggers": _needs_the_agent,
+    "snapshot_agent_state": _needs_the_agent,
+    "load_artifact": _needs_the_agent,
+    "verify_sandbox": _needs_the_agent,
+    "run_code": _needs_the_agent,
+    "collect_artifacts": _collect_artifacts,
+    "register_env_triggers": _register_env_triggers,
+    "peer_agents": _peer_agents,
+    "rubrics_verifier": _rubrics_verifier,
+    **dict.fromkeys((
+        "verify_a2a_agent_card", "verify_a2a_agent_config_identity", "verify_a2a_agent_mcp", "verify_a2a_core_protocol",
+        "verify_a2a_install", "verify_a2a_modalities", "verify_a2a_peer_agents", "verify_a2a_role",
+        "verify_a2a_skill_config", "verify_a2a_snapshot", "verify_a2a_system_prompt", "verify_a2a_trajectory",
+    ), _validates_an_agent),
+}
+
+UNAFFECTED = frozenset({
+    # They name no agent.
+    "aggregate_verifiers", "apply_server_config", "build_mcp_cli", "combine_universe_verdicts", "deploy_env",
+    "deploy_sandbox", "env_outcome_verifier", "modify_env_tool_access", "reset_env", "review",
+    "run_container_unit_tests_verifier", "run_docker_container", "snapshot_env", "sync_env_clock",
+    "validation_gate_aggregator", "verify_env_card", "verify_env_core_protocol", "verify_mcp_tool_schema",
+    "verify_spec_conformance", "verify_universe_load_export_roundtrip",
+    # They read only what an outside turn leaves: its reply.
+    "agent_prompt_response_verifier", "verify_mcp_env_assessment",
+    # It skips an agent that isn't deployed.
+    "teardown_sandboxes",
+})