feat(orchestrator): multi-agent native metrics

skills/training/monitor-run/SKILL.md

@@ -89,17 +89,25 @@ grep -E "WARNING|ERROR" {output_dir}/logs/envs/{train,eval}/*.log
 
 All metrics print to the console log (and W&B when configured).
 
-**Progress** — orchestrator log. Rollout metrics are keyed `{scope}/{subset}/<metric>/<stat>`: `scope` is `train/agg` (all train envs) or `train/<env>` (`eval/<env>` for eval); `subset` is `all` (every rollout) or `effective` (post-filter).
+**Progress** — orchestrator log. Rollout metrics mirror the episode/trace hierarchy, at two levels:
+
+- `{scope}/{subset}/<metric>/<stat>` — episode-level facts only: the token/turn/branch counts, summed over an episode's traces.
+- `{scope}/{subset}/<agent>/<metric>/<stat>` — every trace-level metric (reward, truncation, errors, timing, env metrics, filter verdicts, eval scores), keyed by agent name so seats never mix. Flat over that agent's traces: one sample is one trace, so an in-episode fan-out like n solvers contributes n samples.
+
+`scope` is `train/agg` (all train envs) or `train/<env>` (`eval/<env>` for eval); `subset` is `all` (every rollout) or `effective` (post-filter). Single-agent envs have one agent — usually `agent` — and one trace per episode, so both levels agree; multi-agent envs name each seat (`proposer`, `solver`, `judge`, …).
 
 | Metric | Description |
 |--------|-------------|
-| `train/agg/effective/reward/mean` | mean training reward (per env: `train/<env>/effective/reward/mean`) |
-| `train/agg/effective/num_total_tokens/mean` | avg tokens per rollout (also `num_input_tokens`, `num_output_tokens`) |
-| `train/agg/effective/num_turns/mean` | avg turns per rollout (multi-turn only) |
-| `train/agg/effective/is_truncated/mean` | fraction truncated |
-| `train/agg/all/has_error/mean` | fraction errored (per-type under `train/agg/all/error/<type>`; also `dispatcher/errored/{train,eval}`) |
-| `train/<env>/effective/metrics/<name>/mean` | env-specific metrics (e.g. pass rate) |
-| `eval/<env>/effective/{avg@k,pass@k}` | eval scores when configured |
+| `train/agg/effective/<agent>/reward/mean` | mean training reward for that agent (per env: `train/<env>/effective/<agent>/reward/mean`) |
+| `train/agg/effective/num_total_tokens/mean` | avg tokens per episode, summed over its agents (also `num_input_tokens`, `num_output_tokens`) |
+| `train/agg/effective/num_turns/mean` | avg turns per episode, summed over its agents |
+| `train/<env>/effective/<agent>/num_turns/mean` | avg turns for that agent alone (also token counts, `num_branches`) |
+| `train/agg/effective/<agent>/is_truncated/mean` | fraction of that agent's rollouts truncated |
+| `train/agg/all/<agent>/has_error/mean` | fraction of that agent's rollouts errored (per-type under `train/agg/all/<agent>/error/<type>`; also `dispatcher/errored/{train,eval}`) |
+| `train/agg/all/<agent>/is_trainable/mean` | fraction carrying a training signal — 0.0 for a frozen seat like a judge (also `is_filtered`, `filters/<name>`) |
+| `train/<env>/effective/<agent>/metrics/<name>/mean` | env-specific metrics for that agent (e.g. pass rate) |
+| `train/<env>/effective/<agent>/timing/agent/model/mean` | model vs harness share of that agent's phase |
+| `eval/<env>/effective/<agent>/{avg@k,pass@k}` | eval scores for that agent, when configured |
 
 **Stability** — trainer log:
 

src/prime_rl/orchestrator/dispatcher.py

@@ -5,8 +5,8 @@
   N rollouts in one call reserves N permits (each bridged v0 rollout is its own
   single-agent episode).
 - Optional rate limiting via ``AsyncLimiter(tasks_per_minute, 60)``.
-- Emit-everything invariant: every dispatched env-rollout eventually reaches
-  ``out_q`` exactly once, as one episode (a ``list[Rollout]``). Failures
+- Emit-everything invariant: every dispatched episode eventually reaches
+  ``out_q`` exactly once, as a ``list[Rollout]``. Failures
   (env error, empty trajectory, task exception, off-policy cancel) carry
   ``trace.last_error`` set; sinks decide drop / partial-train policy.
 - ``DispatcherMode.PREFER_TRAIN`` / ``PREFER_EVAL`` controls which kind to
@@ -495,7 +495,7 @@ def release(self, n: int) -> None:
         self.inflight_permits -= n
 
     async def handle_completed_rollout(self, task: asyncio.Task) -> None:
-        """Emit every dispatched env-rollout exactly once to ``out_q``: a ``run``
+        """Emit every dispatched episode exactly once to ``out_q``: a ``run``
         result as one episode, a legacy ``run_group`` result as ``rollout_count``
         single-trace episodes. Task exceptions synthesize ``rollout_count``
         error-marker episodes so the sink's count-to-``group_size`` finalization

src/prime_rl/orchestrator/envs.py

@@ -4,15 +4,15 @@
 external one pinned by ``config.serve.address``) and an ``EnvClient`` to drive it. The
 orchestrator never *runs* an environment — the agents and their runtimes live only
 in the server — but it does own the *taskset*: a v1 env's tasks are loaded here,
-once, and each dispatched env-rollout ships its task's data on the request
+once, and each dispatched episode ships its task's data on the request
 (``task_data``); the server pydantic-validates it into the taskset's declared
 ``TaskData`` type and runs it. That keeps the server (and every worker in its
 pool) stateless about data — no per-worker dataset loads, no idx-addressed task
 cache — and gives the orchestrator real tasks to cycle, shuffle, and filter. Only
 the legacy (v0) bridge, whose dataset genuinely lives server-side, is still driven
 by ``task_idx`` (its count comes from ``info``).
 
-The server answers one ``Episode`` per env-rollout, whose traces we validate into
+The server answers one ``Episode`` per run request, whose traces we validate into
 ``Trace[WireTaskData]`` — real ``vf.Trace``\\ s (never loose dicts) whose task
 keeps the env's task-specific fields as extras (``WireTaskData`` allows them).
 """
@@ -219,7 +219,7 @@ async def run(
         if not episode.traces:
             error = episode.last_error
             detail = f"{error.type}: {error.message}" if error is not None else "no traces and no error recorded"
-            raise RuntimeError(f"env-rollout failed before any trace was minted — {detail}")
+            raise RuntimeError(f"episode failed before any trace was produced — {detail}")
         rollouts = [ROLLOUT_TYPE.model_construct(**dict(wire)) for wire in episode.traces]
         for rollout in rollouts:
             rollout.episode_id = episode.id