feat(run): --agent runs a bundle's solver as another agent

AGENTS.md

@@ -154,7 +154,10 @@ Out-of-tree code plugs in either through `config.toml`, as a `module:Class` poin
   outside a bundle run gets it from `A2AAgent.get`, which reads the store first and an alias after (the planner reads
   the store itself, as `STORE_GETTERS` does). agent-env's own bundles' aliases are theirs, and an alias two other
   bundles give is refused, naming both. `agent-env a2a-agent list` lists the installed agents with their aliases, then
-  the store's; `a2a-agent get` prints the agent an alias names.
+  the store's; `a2a-agent get` prints the agent an alias names. `agent-env run --agent <name>` deploys that agent for
+  the solver for one run (each step deploying or installing an agent a `prompt_agent` step prompts, not its user
+  simulator: `plan.solver_steps`), pulled in and checked as a task's reference to it would be, through per-step
+  `step_params` overrides, so the tasks stay as written.
 - Environment providers have no table: an installed package registers one as an `agent_env.env_providers`
   entry point named its `type`, which records carry as `env_provider_type`.
 - `[plugins.<distribution name>]` is reserved for an installed plugin's own settings, read with

src/agent_env/agents/README.md

@@ -6,7 +6,8 @@ turn's trajectory. It's the default agent, so a `deploy_agent` or an agent judge
 from another bundle as `claude-code`, the alias `aliases.toml` gives it, or by its id,
 `@local/agentenv-framework/agents/claude-code`; an agent of your own with the id `claude-code`, in your bundle or your
 store, is the one `claude-code` names instead. `agent-env a2a-agent list` lists the agents a task can name, with
-their aliases.
+their aliases, and `agent-env run <bundle> --agent pi` runs a bundle's solver as another of them for one run, its
+tasks left as written.
 
 It needs Docker, and a model endpoint serving Anthropic's Messages API, `<base>/v1/messages`, which a LiteLLM gateway
 does: the `[model]` base URL and key reach it as `ANTHROPIC_BASE_URL` and a bearer token, and `[model.headers]` are sent

src/agent_env/bundle/materialize.py

@@ -127,7 +127,7 @@ def through_ledger(write: Write, write_fn: Callable[[], int]) -> None:
         for write in tasks:
             with _noted(plan, write, "preflighting"):
                 built[write.id] = _build(write)
-                found, skipped = _preflight(write, built[write.id], unwritten)
+                found, skipped = _preflight(plan, write, built[write.id], unwritten)
             problems.extend(f"{_path(plan, write)}: {problem}" for problem in found)
             unchecked.extend((write, step) for step in skipped)
         if problems:
@@ -325,14 +325,19 @@ def _build(write: Write) -> Task:
     return Task(id=write.id, version=None, steps=[build_step(step) for step in write.source.config])
 
 
-def _preflight(write: Write, task: Task, unwritten: set[tuple[str, str]]) -> tuple[list[str], list[TaskStep]]:
+def _preflight(plan: Plan, write: Write, task: Task,
+               unwritten: set[tuple[str, str]]) -> tuple[list[str], list[TaskStep]]:
     """``task``'s preflight problems, and the steps with a preflight left unchecked because they read a
     (store, id) in ``unwritten``, which a dry run would write first; an artifact and an agent can share an id.
     A step reading one of the task's own outputs is skipped too, and not listed: the output only exists once the
-    task runs."""
+    task runs. A step the solver deploys or installs at is checked for the agent the run gives it (``Plan.agent``)."""
     outputs = {output.id for output in write.source.outputs}
+    solver = plan.agent.steps.get(write.id, ()) if plan.agent is not None else ()
     problems, unchecked = [], []
     for config, step in zip(write.source.config, task.steps):
+        if step.id in solver:
+            config = {**config, "a2a_agent_id": plan.agent.agent_id, "a2a_agent_version": plan.agent.version}
+            step = build_step(config)
         problems.extend(step.config_problems())
         reads = _reads(config)
         if any(id in outputs for _, id in reads):

src/agent_env/bundle/plan.py

@@ -54,7 +54,10 @@
 from agent_env.task import Task
 from agent_env.task_step.registry import get_task_step_registry
 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
 from agent_env.task_step.task_steps.load_artifact import LoadArtifactTaskStep
+from agent_env.task_step.task_steps.prompt_agent import PromptAgentTaskStep
 
 from ._fs import fold, relative, with_article
 from .authoring import AuthoringContext, entry_files, staged_build_context
@@ -127,6 +130,17 @@ def agent_id(self) -> str:
         return self.alias if self.store_version is not None else self.target_id
 
 
+@dataclass(frozen=True)
+class AgentOverride:
+    """The agent a run deploys for the solver in place of the one each task names: at each step that deploys or
+    installs an agent a step prompts, not as its user simulator (``solver_steps``), by task id."""
+
+    name: str  # as the run was given it
+    agent_id: str
+    steps: Mapping[str, tuple[str, ...]]
+    version: int | None = None  # the store's version the plan read; None for an agent the run writes
+
+
 @dataclass(frozen=True)
 class Write:
     """An entity or built image a run writes."""
@@ -150,6 +164,7 @@ class Plan:
     # run registers in place of its build context alone.
     published: Mapping[str, Publication] = field(default_factory=dict)
     aliases: Mapping[str, AliasUse] = field(default_factory=dict)  # each agent alias the run names, by alias
+    agent: AgentOverride | None = None  # the agent --agent names for the solver
 
     def bundle_of(self, entry: BundleEntry) -> Bundle:
         """The bundle ``entry`` comes from: the installed one for an entry pulled in from one, else the run's."""
@@ -162,12 +177,13 @@ def path(self, entry: BundleEntry) -> str:
 
 
 def plan_bundle(resolved: ResolvedBundle, *, tasks: Sequence[str] = (), evals: Sequence[str] = (),
-                installed: Installed | None = None) -> Plan:
+                installed: Installed | None = None, agent: str | None = None) -> Plan:
     """Plan a run of ``tasks`` and ``evals``, each a name or id in this bundle. With neither, the
     run is every eval or, in a bundle without evals, every task. ``installed`` lists the installed bundles whose
-    entities a reference pulls in (``installed_bundles``); without it, an id under one is a store id. Raises
+    entities a reference pulls in (``installed_bundles``); without it, an id under one is a store id. ``agent`` names
+    the agent the solver runs as, as a task's reference would, leaving the tasks as written (``Plan.agent``). Raises
     BundleError listing every problem found."""
-    return _Planner(resolved, installed).plan(tasks, evals)
+    return _Planner(resolved, installed).plan(tasks, evals, agent)
 
 
 def check_bundle(resolved: ResolvedBundle, *, installed: Installed | None = None) -> None:
@@ -206,6 +222,8 @@ def __init__(self, resolved: ResolvedBundle, installed: Installed | None = None)
         self.pullable: dict[str, _Pullable | str] = {}  # each installed bundle read, by id root, or why it can't be
         self.unpulled: dict[tuple[EntityKind, str, int | None], str] = {}  # why an installed bundle's id isn't pulled
         self.alias_table: InstalledAliases | None = None  # the installed aliases, once a plan reads them
+        self.alias_uses: dict[str, AliasUse] = {}  # each alias the plan names, by alias
+        self.alias_stored: dict[str, int | None] = {}  # the store's latest version of each alias's id, as read
         self.pulling: set[_Key] = set()
         self.own = _by_id(resolved.bundle)  # what a pulled entity's id names in this bundle
         images: dict[BundleEntry, list[BundleImage]] = {}
@@ -265,7 +283,7 @@ def _resolve_aliases(self) -> dict[str, AliasUse]:
         pointed at the alias's agent is left with the problem the store ref check reports."""
         if self.installed is None:
             return {}
-        table, uses, stored, default = installed_aliases(self.installed()), {}, {}, _UNREAD
+        table, uses, stored, default = installed_aliases(self.installed()), self.alias_uses, self.alias_stored, _UNREAD
         self.alias_table = table
         for index, entry in enumerate(self.entries):
             if entry.entry.kind is not BundleKind.TASK:
@@ -292,6 +310,54 @@ def _resolve_aliases(self) -> dict[str, AliasUse]:
                 self._replace(index, replace(entry, config=_renamed(entry.config, aliased), references=tuple(references)))
         return uses
 
+    def _agent_override(self, name: str) -> AgentOverride | None:
+        """The agent ``name`` names, as a task's reference to it would: this bundle's agent, an installed bundle's
+        pulled in, an alias's, or the store's, read once. Each task with solver steps gets a reference to it, so it's
+        written and checked with what the task needs, and its config is left as written. Records a problem and gives
+        None when ``name`` names no agent, or no task has a step for the solver."""
+        where = f"--agent {name!r}"
+        agents = [entry.entry for entry in self.resolved.entries if entry.entry.kind is BundleKind.AGENT]
+        own = next((agent for agent in agents if _nfc(name) in (agent.name, agent.id)), None)
+        near = next((agent for agent in agents if fold(name) in (fold(agent.name), fold(agent.id))), None)
+        if own is None and near is not None:
+            self.problems.append(f"{where}: isn't in this bundle, but {near.name!r} is; names are case-sensitive")
+            return None
+        if own is not None:
+            ref = Reference(EntityKind.AGENT, own.id, None, own, where, None)
+        else:
+            ref = Reference(EntityKind.AGENT, name, None, None, where, None)
+            use = None if self.alias_table is None else self._alias_use(
+                self.alias_table, name, None, self.alias_uses, self.alias_stored)
+            if isinstance(use, str):
+                self.problems.append(f"{where}: {use}")
+                return None
+            ref = self._pulled(replace(ref, id=use.target_id) if use is not None and use.store_version is None else ref)
+            if (problem := self.unpulled.get((ref.kind, ref.id, None)) or self._unstored(ref)) is not None:
+                self.problems.append(f"{where}: {problem}")
+                return None
+        steps = {}
+        for index, entry in enumerate(self.entries):
+            if entry.entry.kind is BundleKind.TASK and (solver := solver_steps(entry.config)):
+                steps[entry.entry.id] = solver
+                self._replace(index, replace(entry, references=(*entry.references, ref)))
+        if not steps:
+            self.problems.append(f"{where}: no step in this bundle's tasks prompts an agent, so none deploys a solver")
+            return None
+        return AgentOverride(name, ref.local.id if ref.local is not None else ref.id, steps)
+
+    def _unstored(self, ref: Reference) -> str | None:
+        """Why the store has no agent ``ref`` names, a store ref, or None when it has one or ``ref`` is local."""
+        if ref.local is not None:
+            return None
+        try:
+            STORE_GETTERS[EntityKind.AGENT](ref.id, None)
+        except NotFoundError:
+            hint = self.alias_table.hint(ref.id) if self.alias_table is not None else ""
+            return f"there is no agent {ref.id!r} in this bundle, an installed bundle or the store{hint}"
+        except (ValueError, KeyError, TypeError) as e:
+            return f"agent {ref.id!r} can't be read ({type(e).__name__}: {e})"
+        return None
+
     def _alias_use(self, table: InstalledAliases, agent_id: str, version: int | None, uses: dict[str, AliasUse],
                    stored: dict[str, int | None]) -> AliasUse | str | None:
         """What ``agent_id`` names when it's an installed alias, the store's agent of that id read first (its latest
@@ -403,13 +469,16 @@ def _pull(self, pullable: _Pullable, target: BundleEntry | BundleImage) -> None:
         self.pulling.discard(key)
         self._add(key, source, _needs(source))
 
-    def plan(self, tasks: Sequence[str], evals: Sequence[str]) -> Plan:
+    def plan(self, tasks: Sequence[str], evals: Sequence[str], agent: str | None = None) -> Plan:
         aliases = self._resolve_aliases()
+        override = self._agent_override(agent) if agent is not None else None
         self._check_writers()
         self._check_loops()
         selected_tasks, selected_evals = self._selection(tasks, evals)
         closure = self._closure(selected_tasks + selected_evals)
         store_refs, store_latest = self._check_store_refs(closure)
+        if override is not None:
+            override = replace(override, version=store_latest.get((EntityKind.AGENT, override.agent_id)))
         self._build_tasks(closure)
         self._check_files(closure)
         self._check_cli_loads(closure)
@@ -421,7 +490,7 @@ def plan(self, tasks: Sequence[str], evals: Sequence[str]) -> Plan:
             raise BundleError(self.problems)
         writes = self._ordered(closure)
         return Plan(self.resolved, writes, tuple(selected_tasks), tuple(selected_evals), store_refs, store_latest,
-                    dict(self.origins), self._published(writes), aliases)
+                    dict(self.origins), self._published(writes), aliases, override)
 
     def check(self) -> None:
         self._check_writers()
@@ -986,6 +1055,17 @@ def folder_files(bundle: Bundle, entry: BundleEntry) -> Mapping[str, Path]:
 _UNREAD: Any = object()  # the configured default agent, before a task needs it
 
 
+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}
+    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)
+
+
 def _leaves_an_agent_unset(task: ResolvedEntry) -> bool:
     """Whether a step of ``task`` leaves an agent field unset, so deploys the configured default agent."""
     registry = get_task_step_registry()

src/agent_env/bundle/preflight.py

@@ -77,7 +77,7 @@
 from agent_env.utils.docker.build import DEFAULT_BUILD_PLATFORM, docker_plugin_missing, docker_unreachable
 
 from .parse import BundleError, BundleKind
-from .plan import Plan
+from .plan import AgentOverride, Plan
 from .resolve import BuiltImage, Reference, ReferencedImage, ResolvedEntry, build_step
 
 _SHOWN = 3
@@ -147,18 +147,20 @@ def __init__(self, plan: Plan, sandbox: str | None):
     def task(self, task: ResolvedEntry) -> None:
         steps = [build_step(config) for config in task.config]
         sandboxes = {step.sandbox_name: step for step in steps if isinstance(step, DeploySandboxTaskStep)}
+        solver = self.plan.agent.steps.get(task.entry.id, ()) if self.plan.agent else ()
         for step in steps:
             where = f"{self.plan.path(task.entry)}: step {step.id!r}"
+            override = self.plan.agent if step.id in solver else None
             try:
                 if isinstance(step, DeployEnvTaskStep):
                     self._env(where, step)
                 elif isinstance(step, DeployAgentTaskStep):
-                    self._agent(where, step, sandboxes)
+                    self._agent(where, step, sandboxes, override)
                 elif isinstance(step, DeploySandboxTaskStep):
                     self._sandbox(where, step)
                 elif isinstance(step, RubricsVerifierTaskStep):
                     self._judge(where, step)
-                elif isinstance(step, InstallAgentTaskStep) and step.a2a_agent_id is None:
+                elif isinstance(step, InstallAgentTaskStep) and step.a2a_agent_id is None and override is None:
                     self.default_agent_users.append((where, "names no agent"))
             except ValueError as e:  # a provider that names no backend, or can't be built from its config
                 self._problem(where, str(e))
@@ -301,7 +303,9 @@ def _planned_version(self, kind: EntityKind, entity_id: str, version: int | None
         """The version a write names a store entity at: its pin, else the one the plan read, which the writer pins."""
         return version if version is not None else self.plan.store_latest.get((kind, entity_id))
 
-    def _agent(self, where: str, step: DeployAgentTaskStep, sandboxes: dict[str, DeploySandboxTaskStep]) -> None:
+    def _agent(self, where: str, step: DeployAgentTaskStep, sandboxes: dict[str, DeploySandboxTaskStep],
+               override: AgentOverride | None = None) -> None:
+        """Check a deploy of the step's agent, or of ``override``, the agent ``--agent`` names for the solver."""
         accepts: Callable[[SandboxProvider], Accepts] = A2AAgent.accepts
         if step.sandbox_name:
             linked = sandboxes.get(step.sandbox_name)
@@ -315,8 +319,9 @@ def _agent(self, where: str, step: DeployAgentTaskStep, sandboxes: dict[str, Dep
         else:
             provider = _provider(self.sandbox or step.sandbox_type, get_agent_sandbox_provider)
         given = frozenset({*step.env_vars, *([ENV_BASE_URL] if step.litellm_base_url else [])})
-        self._agent_deploy(where, provider, step.a2a_agent_id, step.a2a_agent_version, "names no agent", accepts,
-                           given=given)
+        agent_id, version = ((override.agent_id, override.version) if override is not None
+                             else (step.a2a_agent_id, step.a2a_agent_version))
+        self._agent_deploy(where, provider, agent_id, version, "names no agent", accepts, given=given)
 
     def _judge(self, where: str, step: RubricsVerifierTaskStep) -> None:
         if not step.use_agent_judge or step.agent_name is not None: