diff --git a/src/agent_env/bundle/plan.py b/src/agent_env/bundle/plan.py
index 948aeee16..392de9119 100644
--- a/src/agent_env/bundle/plan.py
+++ b/src/agent_env/bundle/plan.py
@@ -59,6 +59,7 @@
 from agent_env.store.routing import namespace_routing_enabled
 from agent_env.task.task import Task
 from agent_env.task_step.registry import get_task_step_registry
+from agent_env.task_step.roles import plays_user, prompts_agent_under_test
 from agent_env.task_step.task_step import TaskStep
 from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
 from agent_env.task_step.task_steps.install_agent import InstallAgentTaskStep
@@ -1219,14 +1220,22 @@ def folder_files(bundle: Bundle, entry: BundleEntry) -> Mapping[str, Path]:
 
 
 def solver_steps(steps: Any) -> tuple[str, ...]:
-    """The ids of the steps in ``steps``, a task's config, that deploy or install an agent a ``prompt_agent`` step
-    prompts, and not as its user simulator: the solver's."""
-    prompts = [step for step in steps if isinstance(step, dict) and step.get("type") == PromptAgentTaskStep.type]
-    prompted = {step.get("agent_name") or TaskStep.DEFAULT_AGENT_NAME for step in prompts}
-    prompted -= {step.get("user_agent_name") or PromptAgentTaskStep.DEFAULT_USER_AGENT_NAME for step in prompts}
+    """The ids of the steps in ``steps``, a task's config, that deploy or install an agent a step prompts as the agent
+    under test (``prompts_agent_under_test``), and not as its user simulator: the solver's."""
+    prompts = [step for step in steps if isinstance(step, dict) and prompts_agent_under_test(step)]
+    prompted = {step.get("agent_name") or TaskStep.DEFAULT_AGENT_NAME for step in prompts} - user_sim_agents(steps)
     return tuple(step["id"] for step in steps if isinstance(step, dict) and isinstance(step.get("id"), str)
                  and step.get("type") in (DeployAgentTaskStep.type, InstallAgentTaskStep.type)
-                 and (step.get("agent_name") or TaskStep.DEFAULT_AGENT_NAME) in prompted)
+                 and not plays_user(step) and (step.get("agent_name") or TaskStep.DEFAULT_AGENT_NAME) in prompted)
+
+
+def user_sim_agents(steps: Any) -> set[str]:
+    """The agents ``steps``, a task's config, name as a user simulator: each ``prompt_agent`` step's
+    ``user_agent_name``, and the agent each step whose class ``plays_user`` deploys."""
+    named = {step.get("user_agent_name") or PromptAgentTaskStep.DEFAULT_USER_AGENT_NAME for step in steps
+             if isinstance(step, dict) and step.get("type") == PromptAgentTaskStep.type}
+    return named | {step.get("agent_name") or TaskStep.DEFAULT_AGENT_NAME for step in steps
+                    if isinstance(step, dict) and plays_user(step)}
 
 
 def _leaves_an_agent_unset(task: ResolvedEntry) -> bool:
diff --git a/src/agent_env/bundle/resolve.py b/src/agent_env/bundle/resolve.py
index 71ff5c527..12b054957 100644
--- a/src/agent_env/bundle/resolve.py
+++ b/src/agent_env/bundle/resolve.py
@@ -29,6 +29,7 @@
 from agent_env.store.ids import HF_PREFIX, derive_id, is_local_id, validate_local_id
 from agent_env.store.image_store.oci_registry_credentials import names_registry
 from agent_env.task_step.registry import get_task_step_registry
+from agent_env.task_step.roles import plays_user, prompts_agent_under_test
 from agent_env.task_step.task_step import TaskStep, attach_retry_config, dependencies
 
 from ._fs import fold, relative, show
@@ -117,11 +118,13 @@ def resolve_bundle(bundle: Bundle) -> ResolvedBundle:
 
 
 def played_deploys(entry: ResolvedEntry) -> frozenset[str]:
-    """The ``deploy_agent`` steps of a task's only prompted agent, which a caller playing that agent itself
-    (``EpisodeSpec``) stands in for; none when the entry isn't a task, or prompts no agent or several."""
+    """The ``deploy_agent`` steps of a task's only agent under test, the one a step prompts as it and no step plays the
+    user as, which a caller playing that agent itself (``EpisodeSpec``) stands in for; none when the entry isn't a
+    task, or prompts no agent or several."""
     if entry.entry.kind is not BundleKind.TASK:
         return frozenset()
-    prompted = {_agent_name(step) for step in entry.config if step.get("type") == "prompt_agent"}
+    prompted = ({_agent_name(step) for step in entry.config if prompts_agent_under_test(step)}
+                - {_agent_name(step) for step in entry.config if plays_user(step)})
     if len(prompted) != 1:
         return frozenset()
     return frozenset(step.get("id") for step in entry.config
diff --git a/src/agent_env/cli/task/run.py b/src/agent_env/cli/task/run.py
index 72c6ca818..66d7bbc22 100644
--- a/src/agent_env/cli/task/run.py
+++ b/src/agent_env/cli/task/run.py
@@ -28,6 +28,7 @@
 from agent_env.task.teardown import TeardownReport, kind, teardown_run
 from agent_env.task_step.context import TaskStepContext, read_dual_keyed
 from agent_env.task_step.context_ops import build_context_update_ops
+from agent_env.task_step.roles import prompts_agent_under_test
 from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep
 
 
@@ -172,8 +173,11 @@ def _step_output_lines(step, context) -> list[str]:
         if match:
             lines.append(_format_loaded_snapshot(match))
         return lines
-    if step.type == "prompt_agent" and context.prompt_responses:
-        return [_format_prompt_response(context.prompt_responses[-1])]
+    if prompts_agent_under_test(step):
+        # This step's own response: a step running alongside it may have added a later one.
+        match = next((r for r in reversed(context.prompt_responses) if r.step_id == step.id), None)
+        if match:
+            return [_format_prompt_response(match)]
     if step.type == "snapshot_agent_state":
         snapshots = context.metadata.get("agent_snapshots") or []
         match = next((s for s in reversed(snapshots) if s.get("capture_step_id") == step.id), None)
@@ -355,7 +359,8 @@ def _stored(task, context: TaskStepContext) -> str:
 @click.option("--litellm-api-key", default=None, type=str, help="Override LITELLM_API_KEY for this run; a secret:KEY or env:NAME reference is resolved where it is used")
 @click.option("--judge-litellm-api-key", default=None, type=str, help="Override LITELLM_API_KEY used by rubrics_verifier judge agents (falls back to --litellm-api-key, then to config secret); a secret:KEY or env:NAME reference is resolved where it is used")
 @click.option("--apply-trajectory-filter/--no-trajectory-filter", "apply_trajectory_filter", default=None, help="Force trajectory filtering on/off for rubrics_verifier (overrides the step's trajectory_filter setting)")
-@click.option("--a2a-agent-id", default=None, type=str, help="Override A2A agent id for deploy_agent steps")
+@click.option("--a2a-agent-id", default=None, type=str,
+              help="Override A2A agent id for deploy_agent steps; a user sim's agent keeps its own")
 @click.option("--agent-sandbox", default=None,
               help="Override agent sandbox backend(s) for deploy_agent steps; comma-separated for fallback chain (e.g. 'modal,local')")
 @click.option("--env-sandbox", default=None,
diff --git a/src/agent_env/task/episode/rules.py b/src/agent_env/task/episode/rules.py
index 70e882d90..3fbf7db9a 100644
--- a/src/agent_env/task/episode/rules.py
+++ b/src/agent_env/task/episode/rules.py
@@ -2,8 +2,9 @@
 
 An outside agent is played through a proxy agent deployed in its place (the ``agents`` bundle's ``proxy``), so steps
 that act on the agent, or in its sandbox, act on the proxy. 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.
+here (``CHECKS``) or listed as untouched by an outside agent (``UNAFFECTED``); a step type whose class ``plays_user``
+not in ``CHECKS`` is checked as the user sim's deploy. A test holds the registry to these, so a new step type is
+classified before an episode runs it. Plugin step types aren't inspected.
 """
 
 from __future__ import annotations
@@ -12,6 +13,8 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from agent_env.task_step.roles import plays_user
+
 if TYPE_CHECKING:
     from agent_env.task_step.task_step import TaskStep
     from agent_env.task_step.task_steps.collect_artifacts import CollectArtifactsTaskStep
@@ -43,7 +46,7 @@ def problems(steps: list[TaskStep], outside: list[str]) -> list[str]:
     })
     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):
+        if check := CHECKS.get(step.type) or (_plays_the_user if plays_user(step) else None):
             why += check(step, scope)
         found.extend(f"step {step.id!r}: {reason}" for reason in why)
     return found
@@ -99,6 +102,12 @@ def _needs_the_agent(step: TaskStep, outside: Outside) -> list[str]:
     return [f"{step.type} needs {step.agent_name!r} as a real agent, and it is played from outside"]
 
 
+def _plays_the_user(step: TaskStep, outside: Outside) -> list[str]:
+    if step.agent_name not in outside.agents:
+        return []
+    return [f"{step.type} deploys {step.agent_name!r} to play the user, 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"]
 
diff --git a/src/agent_env/task/episode/spec.py b/src/agent_env/task/episode/spec.py
index 2eafc412b..087b1f1c6 100644
--- a/src/agent_env/task/episode/spec.py
+++ b/src/agent_env/task/episode/spec.py
@@ -10,6 +10,7 @@
 from agent_env.task.store import new_instance_id
 from agent_env.task.task import Task, substitute_seed_in_steps
 from agent_env.task_step.context import TaskStepContext, sandbox_overrides
+from agent_env.task_step.roles import prompts_agent_under_test
 from agent_env.task_step.task_step import TaskStep
 from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep
 
@@ -74,7 +75,7 @@ async def start(self, *, sandbox: str | None = None) -> Episode:
 
 
 def _trainer_agents(task: Task, steps: list[TaskStep], named: Sequence[str] | None) -> list[str]:
-    prompted = list(dict.fromkeys(step.agent_name for step in steps if step.type == "prompt_agent"))
+    prompted = list(dict.fromkeys(step.agent_name for step in steps if prompts_agent_under_test(step)))
     if named is None:
         if len(prompted) != 1:
             raise EpisodeError(
diff --git a/src/agent_env/task_step/roles.py b/src/agent_env/task_step/roles.py
new file mode 100644
index 000000000..6c8cac306
--- /dev/null
+++ b/src/agent_env/task_step/roles.py
@@ -0,0 +1,31 @@
+"""Which part a task's step plays: whether it prompts the agent under test, and whether it plays the user."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from agent_env.task_step.registry import get_task_step_registry
+
+# The step types that prompt the agent under test: its agent_name is the agent a run tests.
+_PROMPTS_AGENT_UNDER_TEST = frozenset({"prompt_agent", "always_on"})
+
+
+def _type_of(step: Any) -> str | None:
+    """The type of ``step``, a step or a step's config, or None when it names none."""
+    kind = step.get("type") if isinstance(step, dict) else getattr(step, "type", None)
+    return kind if isinstance(kind, str) else None
+
+
+def prompts_agent_under_test(step: Any) -> bool:
+    """Whether ``step``, a step or a step's config, prompts the agent under test (its ``agent_name``)."""
+    return _type_of(step) in _PROMPTS_AGENT_UNDER_TEST
+
+
+def plays_user(step: Any) -> bool:
+    """Whether ``step``, a step or a step's config, deploys the agent that plays the user: its class's ``plays_user``,
+    the class looked up by type for a config; False for a type that isn't registered."""
+    if isinstance(step, dict):
+        kind = _type_of(step)
+        cls = get_task_step_registry().get(kind) if kind is not None else None
+        return bool(getattr(cls, "plays_user", False))
+    return bool(getattr(step, "plays_user", False))
diff --git a/src/agent_env/training/harness.py b/src/agent_env/training/harness.py
index 807eddde0..237971f25 100644
--- a/src/agent_env/training/harness.py
+++ b/src/agent_env/training/harness.py
@@ -35,6 +35,7 @@
 from agent_env.task.task import Task, recorded_scores
 from agent_env.task.teardown import teardown_run
 from agent_env.task_step.context import TaskStepContext, sandbox_overrides
+from agent_env.task_step.roles import plays_user, prompts_agent_under_test
 from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
 from agent_env.task_step.task_steps.verifiers.rubrics_verifier import RubricsVerifierTaskStep
 from agent_env.training._openenv import on_session_loop
@@ -231,7 +232,17 @@ def _index_named(prompt: Any, ids: list[str]) -> int:
 
 
 def _first_prompt(task: Task) -> str:
-    return next((step.prompt for step in task.steps if step.type == "prompt_agent"), "")
+    """The user's first message to the agent under test: its step's prompt, else the opening of the user sim the step
+    names (an always_on step's ``user_sim``, a prompt_agent step's ``user_agent_name``), as it was written."""
+    step = next((each for each in task.steps if prompts_agent_under_test(each)), None)
+    if step is None:
+        return ""
+    if getattr(step, "prompt", None):
+        return step.prompt
+    user = getattr(step, "user_sim", None) or getattr(step, "user_agent_name", None)
+    sim = next((each for each in task.steps if plays_user(each) and user and getattr(each, "agent_name", None) == user),
+               None)
+    return getattr(sim, "opening_prompt", None) or ""
 
 
 def _free_port() -> int:
diff --git a/tst/unit/bundle/agent_override_test.py b/tst/unit/bundle/agent_override_test.py
index f4c670791..1e9f9dd31 100644
--- a/tst/unit/bundle/agent_override_test.py
+++ b/tst/unit/bundle/agent_override_test.py
@@ -9,7 +9,7 @@
 from agent_env.bundle import BundleError, dry_run_bundle, parse_bundle, run_bundle
 from agent_env.bundle import preflight as preflight_module
 from agent_env.bundle.installed import InstalledBundle
-from agent_env.bundle.plan import AgentOverride, plan_bundle, solver_steps
+from agent_env.bundle.plan import AgentOverride, plan_bundle, solver_steps, user_sim_agents
 from agent_env.bundle.resolve import resolve_bundle
 from agent_env.cli import cli
 from agent_env.entity.refs import EntityKind
@@ -18,6 +18,7 @@
 from agent_env.task_step.context import TaskStepContext
 from agent_env.task_step.task_steps.install_agent import InstallAgentTaskStep
 from tst.unit.bundle._support import layout, store_agent
+from tst.unit.task_step.plays_user_step import PlaysUser, plays_user_registered  # noqa: F401  (a fixture)
 
 ROOT = "@local/~/triage"
 SOLVER = "@local/demo-dist/kit/solver"
@@ -119,6 +120,44 @@ def test_only_the_steps_deploying_or_installing_the_agent_a_step_prompts_are_the
     assert solver_steps(steps) == ("solver", "installed")
 
 
+@pytest.mark.usefixtures("plays_user_registered")
+def test_the_agent_an_always_on_step_wakes_is_the_solver():
+    steps = [
+        {"id": "solver", "type": "deploy_agent"},
+        {"id": "user", "type": PlaysUser.type, "agent_name": "linda"},
+        {"id": "loop", "type": "always_on", "prompt": "Help Linda."},
+    ]
+
+    assert solver_steps(steps) == ("solver",)
+
+
+@pytest.mark.usefixtures("plays_user_registered")
+def test_the_agent_a_step_whose_class_plays_the_user_deploys_is_never_the_solvers():
+    """Even when a step prompts it, as a briefing would, and another step installs it."""
+    steps = [
+        {"id": "solver", "type": "deploy_agent"},
+        {"id": "user", "type": PlaysUser.type, "agent_name": "linda"},
+        {"id": "installed", "type": "install_agent", "agent_name": "linda", "sandbox_name": "box"},
+        {"id": "brief", "type": "prompt_agent", "agent_name": "linda"},
+        {"id": "loop", "type": "always_on", "prompt": "Help Linda."},
+    ]
+
+    assert solver_steps(steps) == ("solver",)
+    assert user_sim_agents(steps) == {"linda", "human_agent"}
+
+
+@pytest.mark.usefixtures("plays_user_registered")
+def test_a_prompt_agent_tasks_solver_and_user_agents_are_as_on_main():
+    steps = [
+        {"id": "solver", "type": "deploy_agent"},
+        {"id": "user", "type": "deploy_agent", "agent_name": "human_agent"},
+        {"id": "ask", "type": "prompt_agent", "max_conversation_turns": 3},
+    ]
+
+    assert solver_steps(steps) == ("solver",)
+    assert user_sim_agents(steps) == {"human_agent"}
+
+
 def test_preflight_checks_the_agent_the_solver_deploys(bundle_dir, monkeypatch):
     checked = []
     original = preflight_module._Walk._agent_deploy
diff --git a/tst/unit/bundle/resolve_test.py b/tst/unit/bundle/resolve_test.py
index 23cec79f7..0dac945d7 100644
--- a/tst/unit/bundle/resolve_test.py
+++ b/tst/unit/bundle/resolve_test.py
@@ -2,16 +2,18 @@
 
 import json
 import random
+from types import SimpleNamespace
 
 import pytest
 
 from agent_env.bundle import BundleError, BundleKind, parse_bundle
 from agent_env.bundle import resolve as resolve_module
-from agent_env.bundle.resolve import BuiltImage, Output, Reference, ReferencedImage, resolve_bundle
+from agent_env.bundle.resolve import BuiltImage, Output, Reference, ReferencedImage, played_deploys, resolve_bundle
 from agent_env.entity.refs import EntityKind, EntityRef
 from agent_env.env.env import Env
 from agent_env.task import Task
 from agent_env.task_step.task_step import TaskStep
+from tst.unit.task_step.plays_user_step import PlaysUser, plays_user_registered  # noqa: F401  (a fixture)
 
 ROOT = "@local/~/triage"
 LAYOUT = {
@@ -575,3 +577,25 @@ def test_a_bad_depends_on_on_a_step_deriving_its_output_is_named_once(make, depe
     assert problems(make(tasks={"t": [
         {"id": "cli", "type": "build_mcp_cli", "env_id": "tickets", "command_name": "c", "depends_on": depends_on},
     ]})) == (f"tasks/t.json: step 'cli': {problem}",)
+
+
+@pytest.mark.usefixtures("plays_user_registered")
+def test_the_agent_an_always_on_step_wakes_is_the_one_a_caller_plays():
+    task = SimpleNamespace(entry=SimpleNamespace(kind=BundleKind.TASK), config=[
+        {"id": "agent", "type": "deploy_agent"},
+        {"id": "user", "type": PlaysUser.type, "agent_name": "linda"},
+        {"id": "brief", "type": "prompt_agent", "agent_name": "linda"},
+        {"id": "loop", "type": "always_on", "prompt": "Help Linda."},
+    ])
+
+    assert played_deploys(task) == frozenset({"agent"})
+
+
+def test_a_prompt_agent_tasks_played_deploy_is_as_on_main():
+    task = SimpleNamespace(entry=SimpleNamespace(kind=BundleKind.TASK), config=[
+        {"id": "agent", "type": "deploy_agent"},
+        {"id": "user", "type": "deploy_agent", "agent_name": "human_agent"},
+        {"id": "ask", "type": "prompt_agent", "max_conversation_turns": 3},
+    ])
+
+    assert played_deploys(task) == frozenset({"agent"})
diff --git a/tst/unit/cli/always_on_step_output_test.py b/tst/unit/cli/always_on_step_output_test.py
new file mode 100644
index 000000000..a6d50b52a
--- /dev/null
+++ b/tst/unit/cli/always_on_step_output_test.py
@@ -0,0 +1,30 @@
+"""`task run` shows an always_on step its response as it shows a prompt_agent step's."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+from agent_env.cli.task.run import _step_output_lines
+from agent_env.task_step.context import PromptResponse, TaskStepContext
+
+
+def test_an_always_on_steps_response_is_shown():
+    context = TaskStepContext(prompt_responses=[PromptResponse(prompt_id="loop", response="Recap sent.", step_id="loop")])
+
+    [lines] = _step_output_lines(SimpleNamespace(type="always_on", id="loop"), context)
+
+    assert "prompt_id: loop" in lines and "Recap sent." in lines
+
+
+def test_an_always_on_step_without_a_response_shows_nothing():
+    assert _step_output_lines(SimpleNamespace(type="always_on", id="loop"), TaskStepContext()) == []
+
+
+def test_an_always_on_step_shows_its_own_response_when_a_parallel_step_added_a_later_one():
+    context = TaskStepContext(prompt_responses=[
+        PromptResponse(prompt_id="loop", response="Recap sent.", step_id="loop"),
+        PromptResponse(prompt_id="side", response="Side answer.", step_id="side")])
+
+    [lines] = _step_output_lines(SimpleNamespace(type="always_on", id="loop"), context)
+
+    assert "Recap sent." in lines and "Side answer." not in lines
diff --git a/tst/unit/cli/task_run_agent_pick_test.py b/tst/unit/cli/task_run_agent_pick_test.py
new file mode 100644
index 000000000..c2b557162
--- /dev/null
+++ b/tst/unit/cli/task_run_agent_pick_test.py
@@ -0,0 +1,64 @@
+"""``task run --a2a-agent-id`` is a run-wide pick: the CLI leaves it run-wide, at the deploy steps' stored versions, for
+each step to apply (tst/unit/task_step/run_wide_agent_pick_test.py); a step that plays the user keeps its own agent by
+its class, as it does for a hub run's pick."""
+
+import importlib
+import logging
+
+import pytest
+from click.testing import CliRunner
+
+from agent_env.cli import cli, submitted_runs
+from agent_env.runner import local_runner
+from agent_env.task import Task
+from agent_env.task.teardown import TeardownReport
+
+task_run = importlib.import_module("agent_env.cli.task.run")  # the package's ``run`` is the click command
+pytestmark = pytest.mark.usefixtures("sigint_handled", "local_stores")
+
+AGENT = {"id": "agent", "type": "deploy_agent", "env_ids": [], "a2a_agent_id": "theirs", "a2a_agent_version": 4}
+USER = {"id": "user", "type": "deploy_agent", "env_ids": [], "agent_name": "human_agent", "a2a_agent_id": "sim"}
+ASK = {"id": "ask", "type": "prompt_agent", "prompt_id": "ask", "prompt": "hi", "max_conversation_turns": 3}
+
+
+class _Recorded:
+    """A stored task that records each context it runs with."""
+
+    id, version, requires = "t", 1, None
+
+    def __init__(self, steps):
+        self.steps = Task.from_dict({"id": "t", "version": 1, "steps": steps}).steps
+        self.contexts = []
+
+    async def run(self, context, **_):
+        self.contexts.append(context)
+        return context
+
+
+@pytest.fixture
+def stored(monkeypatch):
+    async def teardown_run(context):
+        return TeardownReport()
+
+    monkeypatch.setattr(task_run, "teardown_run", teardown_run)
+    monkeypatch.setattr(local_runner, "teardown_run", teardown_run)
+    monkeypatch.setattr(submitted_runs, "POLL_SECONDS", 0.02)
+
+    def store(*steps):
+        recorded = _Recorded(list(steps))
+        monkeypatch.setattr(Task, "get", classmethod(lambda cls, id, version=None: recorded))
+        return recorded
+
+    logging.disable(logging.CRITICAL)
+    yield store
+    logging.disable(logging.NOTSET)
+
+
+def test_the_pick_stays_run_wide_with_a_user_agent_deployed(stored):
+    recorded = stored(AGENT, USER, ASK)
+    result = CliRunner().invoke(cli, ["task", "run", "--id", "t", "--a2a-agent-id", "mine"])
+    assert result.exit_code == 0, result.output
+
+    [context] = recorded.contexts
+    overrides = context.metadata["user_overrides"]
+    assert overrides["a2a_agent_id"] == "mine" and "step_params" not in overrides
diff --git a/tst/unit/task/episode/spec_test.py b/tst/unit/task/episode/spec_test.py
index c85d90a61..43334ebb5 100644
--- a/tst/unit/task/episode/spec_test.py
+++ b/tst/unit/task/episode/spec_test.py
@@ -10,6 +10,7 @@
 from agent_env.task_step.registry import _builtin_registry
 from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
 from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep
+from tst.unit.task_step.plays_user_step import PlaysUser, plays_user_registered  # noqa: F401  (a fixture)
 
 ENV = {"id": "env", "type": "deploy_env", "env_id": "portsim"}
 LOAD = {"id": "load", "type": "load_artifact", "env_id": "portsim", "artifact_id": "week-<week>"}
@@ -265,13 +266,45 @@ def test_steps_that_need_only_what_an_outside_player_leaves_are_allowed(replace)
     EpisodeSpec.from_task(_portsim(**replace))
 
 def test_every_built_in_step_type_is_checked_or_unaffected():
-    registered, checked = set(_builtin_registry()), set(CHECKS)
+    registry = _builtin_registry()
+    registered, checked = set(registry), set(CHECKS)
+    plays_the_user = {kind for kind, cls in registry.items() if getattr(cls, "plays_user", False)}
 
-    assert sorted(registered - checked - UNAFFECTED) == []
-    assert sorted((checked | UNAFFECTED) - registered) == []
+    assert sorted(registered - checked - UNAFFECTED - plays_the_user) == []
+    # always_on is classified ahead of its step class (FD-5364).
+    assert sorted((checked | UNAFFECTED) - registered - {"always_on"}) == []
     assert sorted(checked & UNAFFECTED) == []
 
 
+class _AlwaysOn(PromptAgentTaskStep):
+    """An always_on step as these rules read one: the agent it wakes, by name."""
+
+    type = "always_on"
+
+
+def test_the_agent_an_always_on_step_wakes_is_the_one_played_and_is_refused_for_now():
+    loop = _AlwaysOn.from_dict({**PLAY, "id": "loop"})
+    task = Task(id="dock", version=4, steps=[*_portsim(play=None).steps, loop])
+
+    assert ("step 'loop': always_on needs 'default-agent' as a real agent, and it is played from outside"
+            in _refusal(task))
+
+
+@pytest.mark.usefixtures("plays_user_registered")
+def test_a_step_whose_class_plays_the_user_is_refused_when_its_agent_is_played_from_outside():
+    task = _portsim({"id": "user", "type": PlaysUser.type, "agent_name": "default-agent"})
+
+    assert ("step 'user': test_plays_user deploys 'default-agent' to play the user, and it is played from outside"
+            in _refusal(task))
+
+
+@pytest.mark.usefixtures("plays_user_registered")
+def test_a_step_whose_class_plays_the_user_with_its_own_agent_is_allowed():
+    task = _portsim({"id": "user", "type": PlaysUser.type, "agent_name": "linda"})
+
+    assert EpisodeSpec.from_task(task, seed=SEED).trainer_agents == ("default-agent",)
+
+
 def test_the_seed_pass_rebuilds_each_step_as_its_own_class():
     class _Unregistered(PromptAgentTaskStep):
         pass
diff --git a/tst/unit/task_step/plays_user_step.py b/tst/unit/task_step/plays_user_step.py
new file mode 100644
index 000000000..a919bbb72
--- /dev/null
+++ b/tst/unit/task_step/plays_user_step.py
@@ -0,0 +1,24 @@
+"""A deploy step whose class plays the user (``plays_user``), registered under a test type, for tests of the lookups
+that tell the agent under test from the user sim's."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+import pytest
+
+from agent_env.config.runtime import Config
+from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
+
+
+class PlaysUser(DeployAgentTaskStep):
+    type: ClassVar[str] = "test_plays_user"
+    plays_user: ClassVar[bool] = True
+
+
+@pytest.fixture
+def plays_user_registered(monkeypatch):
+    """The task step registry with ``PlaysUser`` registered under its type."""
+    steps = {**Config().task_step_registry(), PlaysUser.type: PlaysUser}
+    monkeypatch.setattr(Config, "task_step_registry", lambda self: steps)
+    return PlaysUser
diff --git a/tst/unit/task_step/run_wide_agent_pick_test.py b/tst/unit/task_step/run_wide_agent_pick_test.py
new file mode 100644
index 000000000..c57ac26f9
--- /dev/null
+++ b/tst/unit/task_step/run_wide_agent_pick_test.py
@@ -0,0 +1,37 @@
+"""A run-wide agent pick (``user_overrides.a2a_agent_id``: ``task run --a2a-agent-id``, a hub run's agent pick) reaches
+every ``deploy_agent`` and ``install_agent`` step, a ``prompt_agent`` task's user agent included, exactly as on main. A
+step that plays the user keeps its own agent by its own class (``plays_user``), not by anything these steps read."""
+
+from __future__ import annotations
+
+from agent_env.task_step.context import TaskStepContext
+from agent_env.task_step.task_steps.deploy_agent import DeployAgentTaskStep
+from agent_env.task_step.task_steps.install_agent import InstallAgentTaskStep
+
+SOLVER = {"id": "deploy-solver", "type": "deploy_agent", "env_ids": [], "agent_name": "solver",
+          "a2a_agent_id": "theirs", "a2a_agent_version": 4}
+HUMAN = {"id": "deploy-human", "type": "deploy_agent", "env_ids": [], "agent_name": "human_agent",
+         "a2a_agent_id": "sim", "a2a_agent_version": 2}
+
+
+def _picked(**overrides) -> TaskStepContext:
+    return TaskStepContext(instance_id="run-1", metadata={"user_overrides": {"a2a_agent_id": "mine", **overrides}})
+
+
+def test_the_pick_reaches_a_prompt_agent_tasks_user_agent_too_at_each_steps_stored_version():
+    ctx = _picked()
+    assert DeployAgentTaskStep.from_dict(SOLVER)._resolve_agent_ref(ctx) == ("mine", 4)
+    assert DeployAgentTaskStep.from_dict(HUMAN)._resolve_agent_ref(ctx) == ("mine", 2)
+
+
+def test_a_picked_version_applies_to_every_deploy():
+    ctx = _picked(a2a_agent_version=9)
+    assert DeployAgentTaskStep.from_dict(SOLVER)._resolve_agent_ref(ctx) == ("mine", 9)
+    assert DeployAgentTaskStep.from_dict(HUMAN)._resolve_agent_ref(ctx) == ("mine", 9)
+
+
+def test_install_agent_takes_the_pick_at_its_stored_version():
+    installed = InstallAgentTaskStep.from_dict({"id": "install-human", "type": "install_agent",
+                                                "agent_name": "human_agent", "a2a_agent_id": "sim",
+                                                "a2a_agent_version": 2, "sandbox_name": "world"})
+    assert installed._agent_ref(_picked()) == ("mine", 2)
diff --git a/tst/unit/training/harness_test.py b/tst/unit/training/harness_test.py
index defddc230..43dd2b0a2 100644
--- a/tst/unit/training/harness_test.py
+++ b/tst/unit/training/harness_test.py
@@ -8,6 +8,7 @@
 import logging
 import threading
 from concurrent.futures import ThreadPoolExecutor
+from types import SimpleNamespace
 
 import httpx
 import pytest
@@ -24,6 +25,7 @@
 from agent_env.training.harness import AgentEnvSessionFactory, prompt_rows
 from tst.unit.bundle._support import layout
 from tst.unit.conftest import local_stores  # noqa: F401
+from tst.unit.task_step.plays_user_step import PlaysUser, plays_user_registered  # noqa: F401  (a fixture)
 from tst.util.fake_vllm import fake_vllm, says, token_ids
 
 MODEL = "Qwen/Qwen3-4B"
@@ -246,3 +248,37 @@ def eval_only(llm_url, model, **_):
 
     with pytest.raises(RuntimeError, match="eval-only"):
         AgentEnvSessionFactory(bundle, llm_url="http://127.0.0.1:9", model=MODEL)
+
+
+def test_a_rows_prompt_is_an_always_on_steps_as_it_is_a_prompt_agent_steps():
+    task = SimpleNamespace(steps=[SimpleNamespace(type="deploy_agent"),
+                                  SimpleNamespace(type="always_on", prompt="Watch Tess's shop this month.")])
+
+    assert harness._first_prompt(task) == "Watch Tess's shop this month."
+
+
+def test_a_rows_prompt_with_no_step_prompt_is_the_named_user_sims_opening():
+    def user_sim(name, opening):
+        return SimpleNamespace(type="deploy_user_sim_agent", plays_user=True, agent_name=name, opening_prompt=opening)
+
+    always_on = SimpleNamespace(type="always_on", prompt=None, user_sim="nora")
+    task = SimpleNamespace(steps=[user_sim("other", "Not this one."), user_sim("nora", "Find me a sofa."), always_on])
+    assert harness._first_prompt(task) == "Find me a sofa."
+
+    prompt_agent = SimpleNamespace(type="prompt_agent", prompt="", user_agent_name="nora")
+    assert harness._first_prompt(SimpleNamespace(steps=[user_sim("nora", "Find me a sofa."), prompt_agent])) \
+        == "Find me a sofa."
+
+    unnamed = SimpleNamespace(type="always_on", prompt=None, user_sim=None)
+    assert harness._first_prompt(SimpleNamespace(steps=[user_sim("nora", "Find me a sofa."), unnamed])) == ""
+
+
+@pytest.mark.usefixtures("plays_user_registered")
+def test_a_task_whose_user_sim_step_deploys_the_user_trains_only_its_solver():
+    config = [{"id": "solver", "type": "deploy_agent", "a2a_agent_id": "theirs"},
+              {"id": "user", "type": PlaysUser.type, "agent_name": "linda", "a2a_agent_id": "sim"},
+              {"id": "brief", "type": "prompt_agent", "agent_name": "linda", "prompt": "You are Linda."},
+              {"id": "ask", "type": "prompt_agent", "prompt": "Help Linda."}]
+
+    assert harness._untrainable(config) is None
+    assert harness.solver_steps(config) == ("solver",)