feat(orchestrator): retire the filter pipeline

configs/ci/nightly-fft/wiki-search.toml

@@ -19,9 +19,8 @@ batch_size = 512
 group_size = 16
 oversampling_factor = 2.0
 
-[[orchestrator.pre_batch_filters]]
-type = "zero_advantage"
-enforce = true
+[orchestrator.sampler]
+drop_degenerate_groups = true
 
 [[orchestrator.train.source]]
 name = "wiki-search"

configs/debug/algo/echo.toml

@@ -44,12 +44,6 @@ type = "subprocess"
 [orchestrator.train.sampling]
 max_completion_tokens = 512
 
-# ECHO learns from observation tokens even when the GRPO advantage collapses
-# to zero — keep zero-advantage rollouts in the batch.
-[[orchestrator.post_batch_filters]]
-type = "zero_advantage"
-enforce = false
-
 # Fine-tune inherits the PrimeIntellect Qwen3 template byte-for-byte.
 [orchestrator.renderer]
 name = "prime-qwen3"

docs/algorithms.md

@@ -1,6 +1,6 @@
 # Algorithms
 
-This page covers the math and the configurable algorithmic components: the algorithm abstraction and its algorithms, how off-policy training works, the loss components and advantage functions, how to plug in your own, the filters applied between rollout and training, and how multi-turn rollouts get merged into training samples.
+This page covers the math and the configurable algorithmic components: the algorithm abstraction and its algorithms, how off-policy training works, the loss components and advantage functions, how to plug in your own, the detections applied between rollout and training, and how multi-turn rollouts get merged into training samples.
 
 ## Table of Contents
 
@@ -21,7 +21,7 @@ This page covers the math and the configurable algorithmic components: the algor
   - [Self-Play Advantage (RAE)](#self-play-advantage-rae)
   - [Authoring an Algorithm](#authoring-an-algorithm)
   - [Reference Scoring](#reference-scoring)
-- [Filters](#filters)
+- [Detections](#detections)
 - [Multi-Turn Trajectories](#multi-turn-trajectories)
   - [Extension Property](#extension-property)
   - [Best-Effort Interleaving](#best-effort-interleaving)
@@ -164,12 +164,12 @@ At runtime, each env's resolved config builds two objects: a `RolloutSource` (`p
 | `hierarchical_grpo` | `HierarchicalGRPOAlgorithm` | `score_group`: GRPO baseline per episode for solvers, per group for the proposer |
 | `opd` | `OPDAlgorithm` | `score_rollout`: own-context prefill under the teacher |
 | `opsd` | `OPSDAlgorithm` | `score_rollout`: demo-conditioned prefill under the live policy |
-| `sft` | `SFTDistillAlgorithm` | `score_group`: group-norm credit (feeds filters) |
+| `sft` | `SFTDistillAlgorithm` | `score_group`: group-norm credit |
 
 Each class owns its hooks outright — reading one top to bottom reads the algorithm, and everything on the class is an override point. The two hooks are one scope-and-timing ladder — the wider scope is unlocked by a later barrier, so the two axes coincide. Each is handed the `Rollout` directly — the env's typed trace (`reward`, `nodes`, `num_turns`, ...) with `samples` attached, plus `assign_advantages` to write credit:
 
-- `async score_rollout(rollout)` — one rollout, **on arrival** (as it's tokenized, before its group is complete): rollout-local credit (`rollout.assign_advantages(...)`, scalar broadcast or per-token), observation ce weights, **or** model I/O — query a reference pool (e.g. `self.teacher_pool`, connected in `setup()` via `self.connect(...)`, or the live `self.policy_pool` for opsd) and attach per-token results (e.g. teacher logprobs) with bounded concurrency. No siblings. `echo` weights observation tokens here, identifying env-provided observation nodes by their non-sampled status and source step role attribution, applying the optional user filter, and writing the `ce_weights` stream. Model I/O runs *before* the pre-batch filters, so it pays compute on rollouts that may then be filtered out.
-- `score_group(group)` — the cohort, **before filtering** (filters read the streams), synchronous: group-relative credit (GRPO/MaxRL baselines). `group` is a list of `Rollout`.
+- `async score_rollout(rollout)` — one rollout, **on arrival** (as it's tokenized, before its group is complete): rollout-local credit (`rollout.assign_advantages(...)`, scalar broadcast or per-token), observation ce weights, **or** model I/O — query a reference pool (e.g. `self.teacher_pool`, connected in `setup()` via `self.connect(...)`, or the live `self.policy_pool` for opsd) and attach per-token results (e.g. teacher logprobs) with bounded concurrency. No siblings. `echo` weights observation tokens here, identifying env-provided observation nodes by their non-sampled status and source step role attribution, applying the optional user filter, and writing the `ce_weights` stream. Model I/O runs at arrival, so it pays compute on rollouts a detection may then exclude.
+- `score_group(group)` — the finalized cohort, synchronous: group-relative credit (GRPO/MaxRL baselines). `group` is a list of `Rollout`.
 
 The pipeline drives the hooks through two non-virtual methods it never looks inside: `algorithm.finalize_rollout(rollout)` per arrival (rollout-local scoring + reference I/O) and `algorithm.finalize_group(rollouts)` per group (scoring + wire stamping; after this the records are frozen — groups die at stamping). Sample construction (interleaving) is pure pipeline — observation-token provenance is available through structural attribution (`node.sampled`, `node.is_content`) for any algorithm that trains on env-provided tokens.
 
@@ -305,8 +305,8 @@ The per-token training signal is set by `algo.type` and the [algorithm](#the-alg
 | `rae` | `rl` | Reward minus a per-agent EMA baseline (SPIRAL's role-conditioned advantage estimation) — for multi-agent self-play envs. |
 | `hierarchical_grpo` | `rl` | GRPO for proposer-solver envs: solvers are compared within one proposed problem, while proposers are compared across proposals. |
 | `echo` | `rl` + `ce` | Group-norm on action tokens, plus weighted CE on env-provided tokens selected by message role (each role's `alpha` is its ECHO λ), optionally narrowed by a user filter. |
-| `opd` | `ref_kl` | On-policy distillation: per-token reverse KL to a reference model (`teacher`, an inline frozen hosted model), evaluated in the trainer from shipped reference logprobs. No credit — rollouts keep `advantages = None` (advantage-based filters never fire) and ship no advantage stream; `group_size` only fans out sampling. |
-| `opsd` | `ref_kl` | SDFT: per-token reverse KL to a demo-conditioned reference. No credit — rollouts keep `advantages = None` (advantage-based filters never fire) and ship no advantage stream. |
+| `opd` | `ref_kl` | On-policy distillation: per-token reverse KL to a reference model (`teacher`, an inline frozen hosted model), evaluated in the trainer from shipped reference logprobs. No credit — rollouts keep `advantages = None` (they always ship; their signal is not advantage-based) and ship no advantage stream; `group_size` only fans out sampling. |
+| `opsd` | `ref_kl` | SDFT: per-token reverse KL to a demo-conditioned reference. No credit — rollouts keep `advantages = None` (they always ship) and ship no advantage stream. |
 | `sft` | `ce` | Cross-entropy on the sampled tokens. Assigns no advantage — trains on every sampled token. |
 
 ### Default Advantage
@@ -382,7 +382,7 @@ id = "null"
 type = "subprocess"
 ```
 
-`group_size` controls how many problems are proposed from each source task. `env.n` controls how many solvers attempt each proposed problem. If a comparison contains only one trace—for example, a solver when `env.n = 1`—its advantage is zero and the zero-advantage filter removes it.
+`group_size` controls how many problems are proposed from each source task. `env.n` controls how many solvers attempt each proposed problem. If a comparison contains only one trace—for example, a solver when `env.n = 1`—its advantage is zero and it ships no samples.
 
 This algorithm is accepted only for proposer-solver envs. Use the env's `train_proposer` and `train_solver` settings if you want to train only one role.
 
@@ -439,7 +439,7 @@ class MyAlgorithm(Algorithm):
 
 Add a typed `MyAlgoConfig` to `prime_rl.configs.algorithm` and its discriminated union, then register `"my_algo": MyAlgorithm` in `ALGORITHM_CLASSES`. Pick the hook by *when* your signal is ready: `score_rollout` for per-arrival credit or credit that needs a model call (it's `async`), `score_group` for group-relative credit (GRPO/MaxRL). `assign_advantages` takes a scalar (broadcast over the rollout's trainable tokens — the common case) or a full-length per-token list aligned to the concatenated sample token_ids (process rewards, step-level credit; `0.0` off-mask).
 
-Each per-token list must match the rollout's completion-token count exactly — validated loudly when the view writes it. Advantage-based filters and metrics derive from the streams (the zero-advantage filter checks for all-zero streams; logged distributions use per-rollout means). Signals that depend on the live policy's weights (like OPD's reverse KL) cannot be precomputed here; those are reference-scoring algorithms, evaluated in the trainer.
+Each per-token list must match the rollout's completion-token count exactly — validated loudly when the view writes it. Shipping and metrics derive from the streams (an all-zero stream ships no samples unless the algorithm trains on zero advantage; logged distributions use per-rollout means). Signals that depend on the live policy's weights (like OPD's reverse KL) cannot be precomputed here; those are reference-scoring algorithms, evaluated in the trainer.
 
 ### Reference Scoring
 
@@ -454,30 +454,28 @@ type = "opsd"
 demo_key = "demonstration"
 ```
 
-Scoring runs at arrival, *before* the pre-batch filters, so a rollout that is later filtered still cost its reference compute — accepted for the simpler one-rollout-at-a-time shape (advantage-based filters never fire for opd/opsd anyway, since neither assigns an advantage).
+Scoring runs at rollout arrival, before the group-time detections, so an excluded rollout still cost its reference compute — accepted for the simpler one-rollout-at-a-time shape.
 
-## Filters
+## Detections
 
-Filters drop rollouts between scoring and training. Built-ins (composable):
+Detections flag generation pathology — the policy melting down mid-sample — by reading a rollout's own token ids and logprobs. They are rollout-granularity predicates, evaluated once at group finalization (the pipeline's single decision point). Built-ins:
 
-| Filter | Effect |
+| Detection | Fires when |
 |---|---|
-| `gibberish` | Drops rollouts whose mean log-prob fall below a threshold — usually a sign of degenerate output. |
-| `repetition` | Drops rollouts with high n-gram repetition. |
-| `zero_advantage` | Drops rollouts whose advantage is zero, so the trainer doesn't waste tokens on them. |
+| `gibberish` | any sampled token is both rare (`token_id > token_id_threshold`) and high-entropy (`logprob < -log(vocab_size) - logprob_offset`) |
+| `repetition` | `window` consecutive sampled tokens each exceed `prob_threshold` — a high-confidence repetition loop |
 
-The default `[orchestrator]` config registers all three in both filter slots: `post_batch_filters` enforce by default (flagged rollouts are recorded but not shipped to the trainer), while `pre_batch_filters` run in monitor mode (`enforce = false`); flip `enforce = true` there to drop matching rollouts before they consume a slot in the batch. Setting a slot replaces its defaults wholesale:
+Both run in monitor mode by default (results are recorded as metrics, nothing is dropped). With `enforce = true`, a detected rollout ships no training samples and never occupies a batch slot — the batch backfills from fresh groups — while its reward still counts toward the group baseline. Setting the list replaces the defaults wholesale:
 
 ```toml
-[[orchestrator.post_batch_filters]]
-type = "zero_advantage"
-
-[[orchestrator.post_batch_filters]]
-type = "repetition"
-threshold = 0.4
+[[orchestrator.detections]]
+type = "gibberish"
+enforce = true
 ```
 
-Filtered rollouts still appear in W&B distributions, just not in the trainer batch — useful for spotting whether filtering is doing its job.
+Zero-advantage handling needs no configuration: an all-zero advantage stream ships no samples unless the source's algorithm declares `trains_on_zero_advantage` (echo does — its observation CE trains through collapsed advantages), and `[orchestrator.sampler] drop_degenerate_groups = true` additionally keeps whole zero-signal groups out of the batch so it backfills (size the extra inference with `oversampling_factor`).
+
+Detected and degenerate rollouts still appear in W&B distributions and the `detections/*` / `sampler/*` metrics, just not in the trainer batch — useful for spotting whether the hygiene is doing its job.
 
 ## Multi-Turn Trajectories
 

docs/overview.md

@@ -40,6 +40,6 @@ The `rl` entrypoint reads `examples/basic/reverse-text/rl.toml`, splits it into
 - **[Training](training.md)** — Launch and observe RL and SFT runs.
 - **[Inference](inference.md)** — vLLM-backed server (or fleet) holding the current policy.
 - **[Scaling](scaling.md)** — Single-GPU through multi-node clusters via FSDP / EP / CP and SLURM.
-- **[Algorithms](algorithms.md)** — Async semantics, loss / advantage / filter plugins, trajectory merging.
+- **[Algorithms](algorithms.md)** — Async semantics, loss / advantage plugins, detections, trajectory merging.
 - **[Advanced](advanced.md)** — Custom modeling, multimodal, LoRA, P/D inference.
 - **[Development](development.md)** — Test suite, pre-commit hooks, adding a new model.

examples/advanced/glm-5.2/swe.toml

@@ -88,7 +88,7 @@ id = "bash"
 type = "prime"
 labels = ["glm5-pd-disag", "swe-bench-verified"]
 
-[[orchestrator.post_batch_filters]]
+[[orchestrator.detections]]
 type = "gibberish"
 enforce = true
 

examples/advanced/intellect-3.1/rl.toml

@@ -119,9 +119,8 @@ id = "null"
 [orchestrator.train.source.env.agent.runtime]
 type = "subprocess"
 
-[[orchestrator.pre_batch_filters]]
-type = "zero_advantage"
-enforce = true
+[orchestrator.sampler]
+drop_degenerate_groups = true
 
 [orchestrator.eval]
 interval = 25

examples/basic/wiki-search/rl.toml

@@ -52,9 +52,8 @@ id = "null"
 [orchestrator.train.source.env.agent.runtime]
 type = "subprocess"
 
-[[orchestrator.pre_batch_filters]]
-type = "zero_advantage"
-enforce = true
+[orchestrator.sampler]
+drop_degenerate_groups = true
 
 [ckpt] # Checkpoint at the end of training
 

packages/prime-rl-configs/src/prime_rl/configs/orchestrator.py

@@ -303,11 +303,12 @@ class CheckpointConfig(BaseConfig):
 
 
 # Flags rare tokens generated at high entropy (Section 5.2, https://arxiv.org/abs/2510.02387).
-class GibberishFilterConfig(BaseConfig):
+class GibberishDetectionConfig(BaseConfig):
     type: Literal["gibberish"] = "gibberish"
 
     enforce: bool = False
-    """When True, skip detected rollouts entirely so they are not sent to the trainer. When False, only track detection metrics."""
+    """When True, detected rollouts ship no training samples (their reward still counts toward
+    the group baseline). When False, only track detection metrics."""
 
     token_id_threshold: int = 100_000
     """Token IDs above this are candidates for gibberish. BPE tokens are sorted by merge order."""
@@ -319,11 +320,12 @@ class GibberishFilterConfig(BaseConfig):
 # Flags rollouts stuck in a repetition loop: emits high-confidence tokens for an extended stretch.
 # Flagged when `window` consecutive tokens are each sampled with probability above `prob_threshold`.
 # (Section 3.2, https://arxiv.org/abs/2506.13585)
-class RepetitionFilterConfig(BaseConfig):
+class RepetitionDetectionConfig(BaseConfig):
     type: Literal["repetition"] = "repetition"
 
     enforce: bool = False
-    """When True, skip detected rollouts entirely so they are not sent to the trainer. When False, only track detection metrics."""
+    """When True, detected rollouts ship no training samples (their reward still counts toward
+    the group baseline). When False, only track detection metrics."""
 
     window: int = Field(3_000, ge=1)
     """Consecutive high-probability steps required to flag the rollout."""
@@ -332,18 +334,20 @@ class RepetitionFilterConfig(BaseConfig):
     """Tokens sampled with probability above this are considered repetitive. Consecutive such tokens count toward the window."""
 
 
-# Flags rollouts with zero advantage.
-class ZeroAdvantageFilterConfig(BaseConfig):
-    type: Literal["zero_advantage"] = "zero_advantage"
+DetectionConfig: TypeAlias = Annotated[
+    GibberishDetectionConfig | RepetitionDetectionConfig,
+    Field(discriminator="type"),
+]
 
-    enforce: bool = True
-    """When True, skip detected rollouts entirely so they are not sent to the trainer. When False, only track detection metrics."""
 
+class SamplerConfig(BaseConfig):
+    """The task sampler's decision knobs (its stats collection is always on)."""
 
-FilterConfig: TypeAlias = Annotated[
-    GibberishFilterConfig | RepetitionFilterConfig | ZeroAdvantageFilterConfig,
-    Field(discriminator="type"),
-]
+    drop_degenerate_groups: bool = False
+    """Drop a finalized group that produced no training signal (every advantage stream all-zero)
+    instead of letting its rollouts occupy batch slots — the batch then backfills from fresh
+    groups, at inference cost bounded by ``oversampling_factor``. Groups of sources whose
+    algorithm declares ``trains_on_zero_advantage`` (echo) are never dropped."""
 
 
 class FileSystemWeightBroadcastConfig(BaseConfig):
@@ -416,23 +420,16 @@ class OrchestratorConfig(BaseConfig):
     eval: EvalConfig | None = None
     """Evaluation configuration."""
 
-    pre_batch_filters: list[FilterConfig] = [
-        GibberishFilterConfig(enforce=False),
-        RepetitionFilterConfig(enforce=False),
-        ZeroAdvantageFilterConfig(enforce=False),
-    ]
-    """Filters applied *before* a rollout enters the training batch buffer.
-    All three filter types are registered in monitor mode by default; flip ``enforce=true`` per type
-    to drop matching rollouts before they consume a slot in the batch (e.g. a zero-advantage group
-    never makes it into a training batch)."""
-
-    post_batch_filters: list[FilterConfig] = [
-        GibberishFilterConfig(),
-        RepetitionFilterConfig(),
-        ZeroAdvantageFilterConfig(),
+    detections: list[DetectionConfig] = [
+        GibberishDetectionConfig(),
+        RepetitionDetectionConfig(),
     ]
-    """Filters applied *after* a batch has been assembled. Each filter annotates each rollout;
-    rollouts flagged by an enforcing filter are still recorded but not shipped to the trainer."""
+    """Generation-pathology detections, evaluated per rollout at group finalization. Both registered in
+    monitor mode by default; flip ``enforce=true`` per type to keep detected rollouts out of
+    training (they never enter the batch, so the batch backfills; their reward still counts
+    toward the group baseline). Setting this replaces the defaults wholesale."""
+
+    sampler: SamplerConfig = SamplerConfig()
 
     log: LogConfig = LogConfig()
 
@@ -531,13 +528,10 @@ def auto_setup_prime_monitor_run_name(self):
         return self
 
     @model_validator(mode="after")
-    def validate_unique_filter_types(self):
-        for slot_name in ("pre_batch_filters", "post_batch_filters"):
-            types = [f.type for f in getattr(self, slot_name)]
-            if len(types) != len(set(types)):
-                raise ValueError(
-                    f"Duplicate filter types in {slot_name}: {types}. Each filter type may only appear once per slot."
-                )
+    def validate_unique_detection_types(self):
+        types = [d.type for d in self.detections]
+        if len(types) != len(set(types)):
+            raise ValueError(f"Duplicate detection types: {types}. Each detection type may only appear once.")
         return self
 
     @model_validator(mode="after")