feat: adapt orchestrator to pull-based env server (lazy/infinite tasksets)

[mikasenghaas]

Summary

Adapt the orchestrator to the pull-based v1 env server. Companion to verifiers PR #1829, which replaces the index-addressed rollout RPC with sample() / run_rollout(task) / run_group(task, n) and moves task scheduling (shuffle + epoch) into the server. The orchestrator no longer addresses tasks by index — it samples one task per group, then runs it.

• Dispatcher samples once per group, then fans out at rollout granularity. schedule_group_rollout calls env.sample() once for the group (stored on GroupState.task) so all group_size rollouts share one task. A group-scored env then runs the whole group as one indivisible run_group(task, group_size); a non-group env fans out group_size independent run_rollout(task) calls, each holding a single permit. This is the key consolidation: a group is the aggregation unit, but a non-group env's concurrency is controlled per rollout — permits free one rollout at a time, so max_concurrent stays saturated instead of idling on a straggler-bound whole-group pull.
• Env gains sample() / run_rollout(client, task, ...) / run_group(client, task, group_size, ...) wrapping the new client RPCs; the task is carried as a vf.WireTask (not a dict).
• TrainSource collapses to a ratio-weighted env picker — next_example returns just {env_name}. The index range, per-env shuffle, and epoch cursors are gone; the server owns all of that now.
• EvalSource/EvalEnv examples become bare slots (a count); GroupState.task_idx is a -1 placeholder until the first sample() returns (it only labels error markers — successful rollouts carry the real idx on their own Trace).
• Env.num_tasks is int | None (None ⟺ infinite), used only to bound a finite eval.
• Shuffle is configured via --<env>.taskset.shuffle (a vf.ShuffleConfig), not a top-level TrainEnvConfig.shuffle bool (removed). Train envs default it on (a before-validator injects a ShuffleConfig when unset); eval envs leave it unset → deterministic. The legacy bridge gets the ShuffleConfig from the taskset too. (The env server's broker owns the task cursor, so a multi-worker pool serves one coherent shuffled stream — see Remove reload_weights_on_start dead code #1829.)

Requires the verifiers pin to include #1829 (the pull RPCs). This is not backward-compatible with the old index-addressed server — the two land together.

Breaking

• Default training weighting is now equal (...env[].ratio defaults to 1.0): previously, with no ratio set, envs were sampled proportional to dataset size; now every env defaults to ratio = 1.0, so they are sampled in equal parts (1:1:1). Configs still parse (ratio stays optional), but multi-env runs that relied on the implicit size weighting will sample differently — set an explicit ratio per env to restore the old mix. The all-or-nothing ratio validator and the infinite-needs-ratio error are removed. This changes training dynamics for existing multi-env configs.
• Rollouts are pulled, not addressed: the dispatcher sample()s a task then runs it; neither run_rollout nor run_group takes a task_idx. Custom code that drove rollouts by index must sample() first. Requires the verifiers pin to include Remove reload_weights_on_start dead code #1829.
• TrainEnvConfig.shuffle (top-level bool) is removed. Configure shuffling via --<env>.taskset.shuffle (a ShuffleConfig); training still shuffles by default (a before-validator injects a ShuffleConfig when unset). To pin a seed, set --<env>.taskset.shuffle.seed N.

Known follow-up

• Subset eval drifts across steps. The server cursor advances between eval steps, so a subset eval (num_examples < taskset size) repeated across steps sees a rotating subset, unlike the old fixed 0..n-1. Full-set eval is unaffected (each step pulls one reshuffled epoch = the same set). A clean fix is a per-eval-step cursor reset (a reset RPC) — not in this PR.

Verification

• Reverse-text e2e (this machine, 2× RTX PRO 6000). v1 reverse_text trained 20 steps (reward 0.20 → 0.78, 0 errors); the v0 path (legacy bridge) ran after a WireTask(prompt=None) fix to the bridge's sample(). A 2-worker static-pool run produced exactly 8 distinct tasks × 16 rollouts per step (clean groups, no duplication, 0 errors), confirming per-rollout fan-out + single-cursor pool routing.
• Source unit checks. TrainSource: equal-weight default ([1.0, 1.0], not size-proportional), explicit-ratio override, {env_name}-only examples, and permit gating (cost = group_size for group-scored envs, 1 otherwise — matching how the dispatcher actually opens each kind of group).

🤖 Generated with Claude Code