feat(cli): task run and eval run always go through the configured [runner]

AGENTS.md

@@ -110,7 +110,7 @@ and deserialized through a registry.
 | `eval/` | `Eval` (`eval`), a versioned list of task references, and its store. |
 | `plugins/` | Entry-point plugins (see Extension points): the group names, `inventory()`, which `agent-env plugin list/show/check` report, `load_failures` and `settings`. The top level is public; the underscored modules are internal. |
 | `explorer/` | Optional local web UI: `agent-env up`, needs the `explorer` extra, binds loopback `:8234`. |
-| `cli/` | Click CLI with the groups `a2a-agent`, `artifact`, `bundle`, `config`, `env`, `eval`, `plugin`, `run`, `task`, `up`. |
+| `cli/` | Click CLI with the groups `a2a-agent`, `artifact`, `bundle`, `config`, `env`, `eval`, `plugin`, `run`, `task`, `up`. `task run` and `eval run` hand their runs to the configured `[runner]` and follow them. |
 | `examples/` | The bundles agent-env ships (`hello`, and `echo-mcp`, which needs Docker), registered under `agent_env.bundles` in `pyproject.toml`. Tests run them, but they are examples for users, not fixtures. |
 
 ## Extension points

src/agent_env/cli/eval/run.py

@@ -1,48 +1,20 @@
-"""Run all tasks in an eval concurrently, tearing down each run's sandboxes as it ends."""
+"""Run all tasks in an eval through the configured ``[runner]``, following each run to its end."""
 
-import asyncio
-import contextlib
 import copy
 import json
 import os
 from uuid import uuid4
 
 import click
 
+from agent_env.cli import submitted_runs
 from agent_env.cli.banner import print_banner
 from agent_env.cli.identity import get_agent_env_client_id
-from agent_env.cli.teardown_output import echo_teardown
 from agent_env.store.ids import fs_safe
 from agent_env.task.interrupts import Interrupts
-from agent_env.task.teardown import teardown_run
 from agent_env.task_step.context import TaskStepContext
 
 
-def _make_callbacks(tag: str):
-    """Return (on_step_start, on_step_complete) callbacks with a prefix tag."""
-    last_score = None
-
-    def on_step_start(i, total, step, context):
-        click.echo(click.style(
-            f"{tag} Running step [{i+1}/{total}]: {step.type} (id={step.id})...",
-            fg="yellow",
-        ))
-
-    def on_step_complete(i, total, step, context, duration):
-        nonlocal last_score
-        click.echo(click.style(
-            f"{tag} Completed step [{i+1}/{total}]: {step.type} (id={step.id}) [{duration:.1f}s]",
-            fg="green",
-        ))
-
-        score = context.metadata.get("score")
-        if score is not None and score != last_score:
-            click.echo(click.style(f"{tag} score: {score}", fg="green"))
-            last_score = score
-
-    return on_step_start, on_step_complete
-
-
 def _write_context(context, task_id, output_dir, prefix=""):
     """Write context JSON to output_dir."""
     os.makedirs(output_dir, exist_ok=True)
@@ -53,26 +25,6 @@ def _write_context(context, task_id, output_dir, prefix=""):
     click.echo(click.style(f"{prefix}Output written to: {output_path}", fg="blue"))
 
 
-async def _run_single(task, tag, output_dir, agent_model=None, agent_artifact_id=None, base_metadata=None):
-    """Execute a single task run with logging callbacks, then tear down what it deployed, even when it
-    raised or was cancelled."""
-    on_start, on_complete = _make_callbacks(tag)
-    context = TaskStepContext(metadata=copy.deepcopy(base_metadata or {}))
-    try:
-        await task.run(
-            on_step_start=on_start,
-            on_step_complete=on_complete,
-            agent_model=agent_model,
-            agent_artifact_id=agent_artifact_id,
-            context=context,
-        )
-    finally:
-        echo_teardown(tag, await teardown_run(context))
-    if output_dir:
-        _write_context(context, task.id, output_dir, prefix=f"{tag} ")
-    return task.id, tag, context
-
-
 @click.command()
 @click.option("--id", "eval_id", required=True, help="Eval id")
 @click.option("--version", "eval_version", default=None, type=int, help="Eval version (defaults to latest)")
@@ -84,7 +36,7 @@ async def _run_single(task, tag, output_dir, agent_model=None, agent_artifact_id
 @click.option("--judge-model", default=None, type=str, help="Override the model rubrics_verifier judges grade with")
 @click.option("--agent-artifact-id", default=None, type=str, help="Override the agent image artifact for deploy_agent steps")
 def run(eval_id: str, eval_version: int | None, output_dir: str | None, k: int, max_concurrency: int | None, agent_model: str | None, user_model: str | None, judge_model: str | None, agent_artifact_id: str | None):
-    """Run all tasks in an eval concurrently."""
+    """Run all tasks in an eval through the configured [runner], concurrently."""
     if k < 1:
         raise click.BadParameter("must be at least 1", param_hint="'--k'")
     if max_concurrency is not None and max_concurrency < 1:
@@ -119,71 +71,43 @@ def run(eval_id: str, eval_version: int | None, output_dir: str | None, k: int,
     if overrides := {key: value for key, value in (("user_model", user_model), ("judge_model", judge_model)) if value}:
         base_metadata["user_overrides"] = overrides
 
-    async def _run_safe(task, tag, slot):
-        """Wrap _run_single so failures, and a cancel while it waits for its slot, carry task identity."""
-        try:
-            async with slot:
-                return await _run_single(
-                    task,
-                    tag,
-                    output_dir,
-                    agent_model=agent_model,
-                    agent_artifact_id=agent_artifact_id,
-                    base_metadata=base_metadata,
-                )
-        except BaseException as exc:
-            return (task.id, tag, exc)
-
-    def _on_signal(count):
-        message = "Cancelling the runs and tearing them down (Ctrl-C again to stop now)" if count == 1 else (
-            "Stopping the teardown now")
-        click.echo(click.style(message, fg="yellow"))
-
-    async def _run_all(interrupts):
-        slot = asyncio.Semaphore(max_concurrency) if max_concurrency else contextlib.nullcontext()
-        coros = []
-        for task in tasks:
-            for run_idx in range(1, k + 1):
-                if k > 1:
-                    tag = f"[{task.id} run {run_idx}]"
-                else:
-                    tag = f"[{task.id}]"
-                coros.append(_run_safe(task, tag, slot))
-        return [run.result() for run in await interrupts.gather(coros, _on_signal)]
-
+    submissions = [(f"[{task.id} run {i}]" if k > 1 else f"[{task.id}]", task, copy.deepcopy(base_metadata), None)
+                   for task in tasks for i in range(1, k + 1)]
+    runs: list[submitted_runs.SubmittedRun] = []
     with Interrupts() as interrupts:
-        results = interrupts.run(_run_all(interrupts))
+        submitted_runs.follow(interrupts, submissions, runs, agent_model=agent_model,
+                              agent_artifact_id=agent_artifact_id, limit=max_concurrency)
 
     click.echo()
 
     # Summarize results
     failures = []
     successes = []
-    for result in results:
-        task_id, tag, payload = result
-        if isinstance(payload, BaseException):
-            failures.append((task_id, tag, payload))
+    for run in runs:
+        context = run.instance.context if run.instance else None
+        if output_dir and context is not None:
+            _write_context(TaskStepContext.from_dict(context), run.task.id, output_dir, prefix=f"{run.tag} ")
+        if run.completed:
+            successes.append((run.tag, (context or {}).get("metadata", {}).get("score")))
         else:
-            score = payload.metadata.get("score")
-            successes.append((task_id, tag, score))
+            failures.append((run.tag, run.outcome if run.done else "CANCELLED"))
 
     if successes:
         click.echo(click.style("Results:", fg="blue"))
-        for task_id, tag, score in successes:
+        for tag, score in successes:
             score_str = f" score={score}" if score is not None else ""
             click.echo(click.style(f"  {tag} PASSED{score_str}", fg="green"))
 
-    if failures:
-        for task_id, tag, exc in failures:
-            outcome = "CANCELLED" if isinstance(exc, asyncio.CancelledError) else f"FAILED: {exc}"
+    if len(successes) < total_runs:
+        for tag, outcome in failures:
             click.echo(click.style(f"  {tag} {outcome}", fg="red"))
         click.echo(click.style(
-            f"\n{len(successes)}/{total_runs} runs completed, {len(failures)}/{total_runs} failed.",
+            f"\n{len(successes)}/{total_runs} runs completed, {total_runs - len(successes)}/{total_runs} failed.",
             fg="red",
         ))
     else:
         click.echo(click.style(f"\nAll {total_runs} runs completed!", fg="blue"))
     if interrupts.count:
         raise SystemExit(128 + interrupts.signum)
-    if failures:
+    if len(successes) < total_runs:
         raise SystemExit(1)

src/agent_env/cli/submitted_runs.py

@@ -0,0 +1,148 @@
+"""Hand runs to the configured ``[runner]`` and follow them to the end. Progress comes from the run index and the
+task instance, the records every runner keeps."""
+
+from __future__ import annotations
+
+import asyncio
+import functools
+from dataclasses import dataclass, field
+
+import click
+
+from agent_env.config import get_runner
+from agent_env.runner.runner import RunHandle, RunRecord, RunStatus
+from agent_env.task import Task
+from agent_env.task.interrupts import Interrupts, tearing_down
+from agent_env.task.step_journal import replay_context
+from agent_env.task.store import TaskInstance, TaskStepStatus, find_task_instance, get_task_instance_store
+from agent_env.task.task import substitute_seed_in_steps
+from agent_env.task_step.context import TaskStepContext
+
+POLL_SECONDS = 2.0
+
+
+@dataclass
+class SubmittedRun:
+    tag: str
+    task: Task
+    handle: RunHandle
+    record: RunRecord | None = None
+    instance: TaskInstance | None = None
+    steps: set[tuple[str, str]] = field(default_factory=set)
+    definitions: list | None = None
+
+    @property
+    def done(self) -> bool:
+        return self.record is not None and self.record.status.is_terminal
+
+    @property
+    def completed(self) -> bool:
+        return self.record is not None and self.record.status is RunStatus.COMPLETED
+
+    @property
+    def outcome(self) -> str:
+        return f"{self.record.status}" + (f": {self.record.error}" if self.record.error else "")
+
+
+def follow(interrupts: Interrupts, submissions, runs: list[SubmittedRun], *, step_lines=None, **submit_kwargs) -> None:
+    """Submit each ``(tag, task, metadata, instance_id)`` into ``runs`` and follow the runs to the end, echoing each step
+    that ends (with ``step_lines(step, context)`` when given, else a verifier's score). A first Ctrl-C cancels the
+    runs; a second stops waiting for them. ``runs`` holds what was submitted even when following raises."""
+    [outcome] = interrupts.run(interrupts.gather([_submit_and_follow(submissions, runs, step_lines, **submit_kwargs)],
+                                                 _on_signal))
+    if not outcome.cancelled() and outcome.exception() is not None:
+        raise outcome.exception()
+
+
+async def _submit_and_follow(submissions, runs, step_lines, *, limit=None, **submit_kwargs) -> None:
+    """``stop()`` is always awaited, spared from a first Ctrl-C so a ``LocalRunner`` finishes tearing its runs down,
+    short of a second. Never ``start()``: that is the serving process's."""
+    runner = get_runner()
+    submit = functools.partial(runner.submit, **submit_kwargs)
+    pending = list(submissions)
+    try:
+        try:
+            await _follow(runner, submit, pending, runs, limit, step_lines)
+        except asyncio.CancelledError:
+            pending.clear()
+            await asyncio.gather(*(runner.cancel(run.handle.run_id) for run in runs if not run.done))
+            await _follow(runner, submit, pending, runs, limit, step_lines)
+            raise
+    finally:
+        if asyncio.current_task().cancelling() < 2:
+            tearing_down()
+            await runner.stop()
+
+
+async def _follow(runner, submit, pending, runs: list[SubmittedRun], limit, step_lines) -> None:
+    while True:
+        await _poll(runner, runs, step_lines)
+        while pending and (limit is None or sum(not run.done for run in runs) < limit):
+            tag, task, metadata, instance_id = pending.pop(0)
+            handle = await submit(task.id, task.version, metadata=metadata, instance_id=instance_id)
+            _echo(tag, f"Submitted run {handle.run_id} (instance {handle.instance_id})", "blue")
+            runs.append(SubmittedRun(tag, task, handle))
+        if not pending and all(run.done for run in runs):
+            return
+        await asyncio.sleep(POLL_SECONDS)
+
+
+def _on_signal(count: int) -> None:
+    message = "Cancelling the runs (Ctrl-C again to stop waiting for them)" if count == 1 else (
+        "Stopped waiting; what the runs deployed may still be up")
+    click.echo(click.style(message, fg="yellow"))
+
+
+async def _poll(runner, runs: list[SubmittedRun], step_lines) -> None:
+    """Reads every run before raising the first read that failed, so a run that ended is seen as ended."""
+    errors = []
+    for run in runs:
+        if not run.done:
+            try:
+                await _poll_run(runner, run, step_lines)
+            except Exception as e:
+                errors.append(e)
+    if errors:
+        raise errors[0]
+
+
+async def _poll_run(runner, run: SubmittedRun, step_lines) -> None:
+    record = await runner.status(run.handle.run_id)
+    run.instance = find_task_instance(run.handle.instance_id) or run.instance
+    for step in run.instance.completed_steps if run.instance else []:
+        if (step.step_id, step.status) not in run.steps:
+            run.steps.add((step.step_id, step.status))
+            _echo_step(run, step.step_id, step.status == TaskStepStatus.SUCCESS, step_lines)
+    if record is not None and (run.record is None or record.status != run.record.status):
+        run.record = record
+        _echo(run.tag, f"Run {run.outcome}",
+              "green" if record.status is RunStatus.COMPLETED else "red" if record.status.is_terminal else "yellow")
+
+
+def _echo_step(run: SubmittedRun, step_id: str, passed: bool, step_lines) -> None:
+    if run.definitions is None:
+        run.definitions = substitute_seed_in_steps(
+            run.task.steps, ((run.instance.context or {}).get("metadata") or {}).get("seed"))
+    i, step = next((i, step) for i, step in enumerate(run.definitions) if step.id == step_id)
+    _echo(run.tag, f"{'Completed' if passed else 'Failed'} step [{i + 1}/{len(run.definitions)}]: {step.type} "
+                   f"(id={step.id})", "green" if passed else "red")
+    vid = getattr(step, "verifier_id", None)
+    if not passed or not (step_lines or vid):
+        return
+    context = _step_context(run.instance, step_id)
+    if step_lines:
+        for line in step_lines(step, context):
+            _echo(run.tag, line, "green")
+    elif (score := context.metadata.get("verifications", {}).get(vid, {}).get("score")) is not None:
+        _echo(run.tag, f"score (verifier_id={vid}): {score}", "green")
+
+
+def _step_context(instance: TaskInstance, step_id: str) -> TaskStepContext:
+    """What ``step_id`` wrote, replayed from its journal entry; the instance's whole context when it has none."""
+    entries = get_task_instance_store().journal_entries_sync(instance.instance_id)
+    entry = next((e for e in entries if e["step_id"] == step_id), None)
+    return TaskStepContext.from_dict(replay_context([entry], ()) if entry else instance.context or {})
+
+
+def _echo(tag: str, message: str, color: str) -> None:
+    click.echo(click.style(f"{tag} {message}" if tag else message, fg=color))