feat(orchestrator): add composable curricula

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

@@ -19,12 +19,9 @@ batch_size = 512
 group_size = 16
 oversampling_factor = 2.0
 
-[[orchestrator.pre_batch_filters]]
-type = "zero_advantage"
-enforce = true
-
 [[orchestrator.train.source]]
 name = "wiki-search"
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "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, curricula, 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)
+- [Curricula](#curricula)
 - [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 `Sampler` (`prime_r
 | `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` | no credit assignment; CE on sampled tokens |
 
 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.
+- `score_group(group)` — the completed 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` 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` 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.
 
 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. Curriculum admission and metrics can inspect these streams after group scoring. 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,50 @@ 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 arrival, before curriculum admission, so a rollout that is later rejected still costs its reference compute.
 
-## Filters
+## Curricula
 
-Filters drop rollouts between scoring and training. Built-ins (composable):
+Each training source has a `Curriculum` composed from one `TaskSampler` and any number of named `AdmissionGate`s. The sampler chooses tasks and observes every finalized result. All gates also observe every result; the group trains only if every gate admits it. Rejected groups remain observable while the orchestrator samples again to fill the batch.
 
-| Filter | Effect |
-|---|---|
-| `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. |
+Samplers and gates can be stateful. Their `state_dict`, `load_state_dict`, and `metrics` methods are included in orchestrator checkpoints and logged under `curriculum/<env>/`. Custom implementations are ordinary subclasses loaded by import path:
 
-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:
+```python
+from prime_rl.orchestrator.curriculum import AdmissionGate, TaskSampler
 
-```toml
-[[orchestrator.post_batch_filters]]
-type = "zero_advantage"
 
-[[orchestrator.post_batch_filters]]
-type = "repetition"
-threshold = 0.4
+class MySampler(TaskSampler):
+    def observe(self, result):
+        ...
+
+
+class MyGate(AdmissionGate):
+    def admit(self, result):
+        return True
 ```
 
-Filtered rollouts still appear in W&B distributions, just not in the trainer batch — useful for spotting whether filtering is doing its job.
+Two small implementations are included:
+
+- `DifficultyPools` samples every finite task once, tracks its latest valid mean group reward, then samples a named reward pool by weight and a task uniformly within that pool. Tasks without a valid result do not block pool sampling, and the final pool is a catch-all.
+- `AdvantageRangeGate` rejects a group when every trainable-token advantage falls inside `reject_min` through `reject_max`. Its default `[0, 0]` range rejects zero-advantage groups. Groups without an advantage stream are admitted.
+
+```toml
+[orchestrator.train.source.curriculum.sampler]
+import_path = "prime_rl.orchestrator.curriculum.DifficultyPools"
+
+[orchestrator.train.source.curriculum.sampler.kwargs.thresholds]
+hard = 0.25
+medium = 0.75
+easy = 1.0
+
+[orchestrator.train.source.curriculum.sampler.kwargs.weights]
+hard = 0.2
+medium = 0.6
+easy = 0.2
+
+[orchestrator.train.source.curriculum.gates.zero_advantage]
+import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate"
+```
 
 ## Multi-Turn Trajectories
 

examples/advanced/glm-5.2/swe.toml

@@ -88,11 +88,6 @@ id = "bash"
 type = "prime"
 labels = ["glm5-pd-disag", "swe-bench-verified"]
 
-[[orchestrator.post_batch_filters]]
-type = "gibberish"
-enforce = true
-
 [inference]
 # we need <0.85 bc glm5 layers are too large for 0.85
 use_deep_gemm = true
-

examples/advanced/intellect-3.1/rl.toml

@@ -50,6 +50,7 @@ oversampling_factor = 2
 [[orchestrator.train.source]]
 name = "swe"
 ratio = 0.3
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "r2e-gym"
@@ -64,6 +65,7 @@ labels = ["intellect-3.1", "swe"]
 [[orchestrator.train.source]]
 name = "deepdive"
 ratio = 0.2
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "deepdive"
@@ -80,6 +82,7 @@ labels = ["intellect-3.1", "deepdive"]
 [[orchestrator.train.source]]
 name = "math"
 ratio = 0.3
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "i3_math"
@@ -93,6 +96,7 @@ type = "subprocess"
 [[orchestrator.train.source]]
 name = "logic"
 ratio = 0.2
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "i3_logic"
@@ -109,6 +113,7 @@ type = "subprocess"
 [[orchestrator.train.source]]
 name = "code"
 ratio = 0.2
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "i3_code"
@@ -119,10 +124,6 @@ id = "null"
 [orchestrator.train.source.env.agent.runtime]
 type = "subprocess"
 
-[[orchestrator.pre_batch_filters]]
-type = "zero_advantage"
-enforce = true
-
 [orchestrator.eval]
 interval = 25
 

examples/basic/wiki-search/rl.toml

@@ -42,6 +42,7 @@ max_completion_tokens = 512
 
 [[orchestrator.train.source]]
 name = "wiki-search"
+curriculum = { gates = { zero_advantage = { import_path = "prime_rl.orchestrator.curriculum.AdvantageRangeGate" } } }
 
 [orchestrator.train.source.env.taskset]
 id = "wiki-search"
@@ -52,10 +53,6 @@ id = "null"
 [orchestrator.train.source.env.agent.runtime]
 type = "subprocess"
 
-[[orchestrator.pre_batch_filters]]
-type = "zero_advantage"
-enforce = true
-
 [ckpt] # Checkpoint at the end of training
 
 [inference.vllm]

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

@@ -234,8 +234,7 @@ class MaxRLAlgoConfig(BaseAlgoConfig):
     objective: low-pass-rate examples get ~1/p weight, and ``group_size`` is
     the truncation order interpolating REINFORCE (1) → exact maximum
     likelihood (∞). Designed for non-negative (canonically binary) rewards;
-    a group with mean reward 0 carries zero advantages everywhere (the
-    zero-advantage filter drops it, matching the paper's K=0 convention)."""
+    a group with mean reward 0 carries zero advantages everywhere."""
 
     action_loss_type: ClassVar[ActionLossType] = "rl"
 
@@ -300,8 +299,8 @@ class OPDAlgoConfig(BaseAlgoConfig):
     a reference model, evaluated in the trainer from reference prefill
     logprobs scored over each sample's own context (``ref_logprobs`` on the
     wire, ``ref_kl`` loss component). No scalar advantage is assigned —
-    rollouts keep ``advantages=None`` (advantage-based filters never fire) and
-    samples ship no advantage stream. ``group_size`` only fans out sampling."""
+    rollouts keep ``advantages=None`` and samples ship no advantage stream.
+    ``group_size`` only fans out sampling."""
 
     action_loss_type: ClassVar[ActionLossType] = "ref_kl"
 
@@ -322,8 +321,7 @@ class OPSDAlgoConfig(BaseAlgoConfig):
     prepended as a leading system message. The sample is scored verbatim (no
     re-rendering), so it's robust to tool/multimodal prompts and works for any
     number of turns. No scalar advantage is assigned — rollouts keep
-    ``advantages=None`` (advantage-based filters never fire) and samples ship no
-    advantage stream."""
+    ``advantages=None`` and samples ship no advantage stream."""
 
     action_loss_type: ClassVar[ActionLossType] = "ref_kl"
 
@@ -347,8 +345,8 @@ class SFTAlgoConfig(BaseAlgoConfig):
     type: Literal["sft"] = "sft"
     """SFT distillation: cross-entropy on the sampled tokens. The ``ce`` loss
     ignores advantages and SFT assigns none — it trains on every sampled token.
-    Reward-based filtering, if wanted, is an explicit filter, not smuggled
-    through an unused advantage stream."""
+    A curriculum can reject results using reward or any other finalized
+    rollout data."""
 
     action_loss_type: ClassVar[ActionLossType] = "ce"