fix(eval): preserve empty outcomes and exclude zero-turn traces from training

docs/eval.md

@@ -6,6 +6,7 @@ This page covers `uv run eval` — evaluating a model in one or more environment
 
 ## Table of Contents
 
+- [Outcome ownership](#outcome-ownership)
 - [Launch](#launch)
 - [Configuration](#configuration)
 - [Resume](#resume)
@@ -16,6 +17,45 @@ This page covers `uv run eval` — evaluating a model in one or more environment
 
 `uv run eval` runs one epoch of every configured source and exits. It reuses the orchestrator's eval pipeline: env servers are spawned per source, episodes are admitted under the concurrency controller, and every episode streams into the run's trace stream and metrics.
 
+## Outcome ownership
+
+Native `vf-eval` runs the environment in process. Prime-RL dispatches episodes to Verifiers workers over ZMQ. Both use Verifiers' `Env.run_slot()` for environment execution, agent runs, scoring, and retries.
+
+```mermaid
+flowchart TD
+    N[Native vf-eval] --> I[In-process environment]
+    P[Prime-RL eval dispatcher] --> Z[ZMQ environment worker]
+    I --> V[Verifiers Env.run_slot]
+    Z --> V
+    V --> E[Completed episode and trace verdicts]
+    E -->|Native| C[Completion callback]
+    C --> S[Save episode and scores]
+    E -->|Prime-RL| W[ZMQ response and Env.run adapter]
+    W --> D[Dispatcher records errors and stamps provenance]
+    D --> L[EvalRunner logs episode]
+    L --> B[EvalSink counts episode toward the epoch]
+```
+
+An episode's outcome and its usefulness for training are separate decisions. Emptiness alone does not change a returned verdict: Prime-RL preserves a successful episode with no traces and a successful episode's traces with zero sampled turns. Existing failures remain failures; Verifiers' default environment still rejects a `run()` that ran no agent. A trace-less episode carries no trace reward to aggregate.
+
+Training uses the existing `iter_trainable_traces()` filter. A trace must be successful, belong to a trainable agent, have at least one sampled turn, and contain tokens marked for training. A successful empty outcome can therefore be recorded without contributing a trainer sample.
+
+```mermaid
+flowchart TD
+    E[Returned episode] --> K{Consumer}
+    K -->|Evaluation| V[Keep verdict and scores]
+    V --> M[Record outcome and update eval metrics]
+    K -->|Training| T[Apply training timeout exclusion]
+    T --> F[iter_trainable_traces]
+    F --> G{Any eligible traces?}
+    G -->|No traces or no eligible traces| Z[Complete group and record zero output]
+    G -->|Yes| A[Algorithm scoring and curriculum admission]
+    A -->|Admitted| B[Build samples from eligible traces]
+    A -->|Rejected| Z
+```
+
+`TrainSink` completes groups with no eligible traces and advances its zero-output warning counter without adding samples. In a mixed group, only eligible traces can produce samples. The training timeout exclusion still applies; evaluation keeps a successful timeout as a scored outcome.
+
 ## Launch
 
 By default the model is served by Prime Inference; authenticate with `PRIME_API_KEY` or `prime login`:

src/prime_rl/orchestrator/algo/base.py

@@ -36,7 +36,7 @@ def iter_trainable_traces(episodes: list[vf.Episode]):
     """Yield clean trainable traces that contain sampled tokens."""
     for episode in episodes:
         for trace in episode.traces:
-            if trace.has_error or not trace.agent.trainable:
+            if trace.has_error or not trace.agent.trainable or trace.num_turns == 0:
                 continue
             if any(any(node.mask) for node in trace.nodes):
                 yield episode, trace

src/prime_rl/orchestrator/curriculum/base.py

@@ -7,6 +7,7 @@
 
 import verifiers.v1 as vf
 
+from prime_rl.orchestrator.algo.base import iter_trainable_traces
 from prime_rl.orchestrator.curriculum.gates import AdmissionGate
 from prime_rl.orchestrator.curriculum.samplers import TaskSampler
 
@@ -48,7 +49,7 @@ def __init__(
             self.gates[name] = gate
 
     def on_result(self, group: list[vf.Episode]) -> bool:
-        """Observe every result, evaluate every gate, and combine with AND."""
+        """Observe every result and admit trainable groups that pass every gate."""
         if not group:
             raise ValueError("Cannot report an empty rollout group")
         task_keys = {episode.task.key for episode in group}
@@ -63,7 +64,7 @@ def on_result(self, group: list[vf.Episode]) -> bool:
             if not isinstance(decision, bool):
                 raise TypeError(f"AdmissionGate {name!r}.admit() must return bool, got {type(decision).__name__}")
             decisions.append(decision)
-        return all(decisions)
+        return all(decisions) and any(iter_trainable_traces(group))
 
     def state_dict(self) -> dict[str, Any]:
         return {

src/prime_rl/orchestrator/curriculum/gates/adv.py

@@ -27,7 +27,7 @@ def admit(self, group: list[vf.Episode]) -> bool:
             advantage
             for episode in group
             for trace in episode.traces
-            if not trace.has_error and trace.agent.trainable
+            if not trace.has_error and trace.agent.trainable and trace.num_turns > 0
             for node in trace.nodes
             for advantage in node.advantages or []
         ]

src/prime_rl/orchestrator/curriculum/samplers/pool.py

@@ -68,7 +68,7 @@ def observe(self, group: list[vf.Episode]) -> None:
             trace.reward
             for episode in group
             for trace in episode.traces
-            if not trace.has_error and trace.agent.trainable
+            if not trace.has_error and trace.agent.trainable and trace.num_turns > 0
         ]
         if not rewards:
             return

src/prime_rl/orchestrator/dispatcher.py

@@ -665,25 +665,13 @@ async def handle_completed_request(self, task: asyncio.Task) -> None:
             )
             return
 
-        if not episode.traces and episode.ok:
-            episode.ok = False
-            episode.errors.append(vf.Error(type="EmptyEpisode", message="Episode returned with no traces"))
-
         for trace in episode.traces:
             if trace.is_timeout and not trace.has_error and meta.kind == "train":
                 # Training keeps a timed-out rollout out of the batch: an error,
                 # like any other failure. Eval keeps it as a scored outcome.
                 trace.errors.append(vf.Error(type="Timeout", message=f"Trace stopped by {trace.stop_condition}"))
                 trace.ok = False
                 episode.ok = False
-            if not trace.has_error and not trace.is_timeout and trace.num_turns == 0:
-                # Empty trajectory: promote to an explicit error so the sink
-                # treats it like any other failure (``has_error`` reads ``ok``).
-                # A timed-out eval trace with no turns stays a timeout: scored, no samples.
-                trace.errors.append(vf.Error(type="EmptyTrajectory", message="Trace returned with no trajectory steps"))
-                trace.ok = False
-                episode.ok = False
-                get_logger().warning(f"Empty trajectory in group {meta.group_id} ({meta.env_name})")
             if trace.has_error:
                 self.metrics.record_error(kind=meta.kind, env_name=meta.env_name)
                 if trace.last_error is not None:

tests/unit/orchestrator/test_curriculum.py

@@ -42,7 +42,7 @@ def make_rollout(
             )
         ]
         if advantages is not None
-        else []
+        else [vf.MessageNode(message=vf.AssistantMessage(content="answer"), sampled=True)]
     )
     trace = vf.Trace(
         task=vf.TraceTask(
@@ -111,7 +111,9 @@ def test_train_source_composes_sampler_and_all_gates_with_state_and_metrics() ->
     source = TrainSource([env])
 
     sampled = source.next_task(step=1).task
-    assert source.on_result(make_rollout(sampled, reward=0.25, advantages=[0.0])) is False
+    empty = make_rollout(sampled, reward=1.0, advantages=[1.0])
+    empty[0].traces[0].nodes[0].sampled = False
+    assert source.on_result(make_rollout(sampled, reward=0.25, advantages=[0.0]) + empty) is False
     assert source.metrics() == {
         "curriculum/test/admission_rate": 0.0,
         "curriculum/test/sampler/pool/unseen": 2.0,
@@ -127,6 +129,19 @@ def test_train_source_composes_sampler_and_all_gates_with_state_and_metrics() ->
         "admit": {},
     }
 
+    assert source.on_result(empty) is False
+    assert source.metrics()["curriculum/test/admission_rate"] == 0.0
+    assert source.curricula["test"].sampler.task_rewards == {sampled.key: 0.25}
+
+    # A scored rollout without token metadata still informs task difficulty.
+    assert source.on_result(make_rollout(sampled, reward=0.5)) is False
+    assert source.metrics()["curriculum/test/admission_rate"] == 0.0
+    assert source.curricula["test"].sampler.task_rewards == {sampled.key: 0.5}
+
+    assert source.on_result(make_rollout(sampled, reward=0.75, advantages=[1.0])) is True
+    assert source.metrics()["curriculum/test/admission_rate"] == 1.0
+    assert source.curricula["test"].sampler.task_rewards == {sampled.key: 0.75}
+
 
 def test_difficulty_pools_stack_with_advantage_gate_and_resume_sampling() -> None:
     tasks = [make_task(i) for i in range(3)]