[faresobeid]

What

No behavior change unless an algorithm opts in. This adds an opt-in Algorithm.finalize_in_background: ClassVar[bool] = False. An algorithm that does not opt in runs the same inline process_group path as main: same order, same accounting, same metrics. Only the inline code was split into process_group (pop the pending state) and _settle_group (the unchanged body), so both paths can share it.

When an env's algorithm opts in (gar in #3803), a completed group with trainable survivors runs finalize_group as a background asyncio task:

• When the task finishes, it puts the group id on a queue owned by the sink, TrainSink.finalized_q (an unbounded asyncio.Queue[str]; it is not dispatcher.out_q).
• Before reading out_q, the orchestrator main loop settles each queued id with TrainSink.finalized(group_id). This runs the rest of the old process_group: accounting, curriculum admission, trace_to_samples, pruning, pending_batch insert, the insertion stale sweep, and _maybe_batch.
• finalized() returns None for ids that were already dropped (stale cancel, drain, stop).

Why

The gar score_group awaits a grader call of 10-30 minutes. If that call is inline, the whole orchestrator stops for that time.

Background-path behavior

• Cap / backpressure: Algorithm.max_finalizing_groups (default 64; an instance can override it from config). When any env has that many groups finalizing, update_dispatch_gate() closes the orchestrator's existing train dispatch gate (dispatcher.dispatch_allowed). Train scheduling then stops, while evals and in-flight episodes keep flowing. The gate reopens when the count falls below the cap. Only groups already in flight can still complete and start finalizing, so finalizing is bounded by the cap plus the groups in flight.
  • The main loop re-checks the gate at the top of each iteration, but only while something is finalizing or the gate is closed. So for algorithms that do not opt in, the gate calls are the same as on main.
  • I chose this over blocking out_q because it does not stall evals, and it also bounds out_q when no inflight ceiling is configured.
• max_off_policy_steps: right after progress.step advances, cancel_stale_finalizing() cancels finalizing groups that would already be stale on insertion. They are accounted exactly like a stale group cancellation: episodes cancelled and not admitted, the zero-output tally advanced, failures and cancelled attempts booked. A group whose task has already finished is not cancelled; the insertion sweep _drop_stale handles it and remains the hard guarantee.
• Drain: start_draining calls train_sink.stop(). Finalizing groups whose batches would be dropped anyway are cancelled; drain does not wait for them.
• Stop: Orchestrator.stop() also calls train_sink.stop().
• Errors: an exception in a background finalize_group is stored by the task's done-callback. _raise_if_component_stopped() re-raises it at the top of the next loop iteration (at most ~0.5 s later), with the original traceback. Errors are never swallowed.
• Checkpoint/resume: the checkpoint holds only Progress and TrainSource (curriculum) state. The curriculum sees a group only when it is settled, so finalizing groups are lost on restart like in-flight rollouts. No changes were needed in ckpt.py.
• Gauges: the (+N buffered) pipeline count includes the episodes and failures of finalizing groups.
• Latency: a finalized group is settled the next time the main loop wakes. With no other traffic, that is at most 0.5 s later because of the out_q poll timeout. That is negligible next to minute-scale finalization.
• Docs: docs/algorithms.md now says that background-finalized algorithms run finalize_group concurrently and complete in any order, so stateful ones must tolerate interleaved groups.

What I ran

• pytest tests/unit/orchestrator: 92 passed. ruff check and ruff format --check on src are clean. (On the earlier revision, the full tests/unit run had the same failure set as origin/main; all of those are trainer/GPU tests, and no GPU is available here.)
• A throwaway script (not committed) that feeds the same random episode stream through origin/main's TrainSink and this branch's. The stream mixes interleaved groups, errored traces, dispatch failures, overload and stale cancellations, policy-version bumps, and a curriculum that rejects some groups. It uses real GRPOAlgorithm and covers 12 seeds × {trace, token batching} × {constant_trainer_batch_size on/off} × max_off_policy_steps ∈ {1, 2, 8}.
  • GRPO, not opted in: all 2288 shipped batches are bit-identical to main. That covers samples (token ids, advantages, rl_weights, temperatures), the episode / admitted / cancelled / sampled-trace sets, cohort, failures, cancelled and stale attempts, buffered ids, stale_drops, the zero-output tally, the final pending state, and the order of curriculum calls.
  • GRPO opted in, each group settled before the next item: also identical to main.
  • Opted-in fake slow algorithm (random delays): batch membership differs because groups land in a different order. Totals of episodes, failures and cancelled attempts match main, and stale cancels happen.
  • Opted-in algorithm that hangs: cancel_stale_finalizing() cancels only the stale group, with exact stale accounting. finalizing_full follows max_finalizing_groups. stop() cancels the rest and leaves the queue empty. finalized() on a dropped id returns None.
  • Opted-in algorithm that raises: raise_if_failed() re-raises the error.

GPU / e2e checks not run here

• GRPO compared with main: reward, step time, and off_policy/* metrics, including a constant_trainer_batch_size run.
• A clean drain at max_steps, the final checkpoint, and resume from it.
• With gar (feat(orchestrator): gar — grader-ranked group advantage redistribution #3803): the size of finalizing (pipeline log +N buffered) and orchestrator RSS over many steps; the dispatch pause and resume at max_finalizing_groups.

Landing order

verifiers + prime-envs -> prime-rl PR1 (this PR, background finalization) -> prime-rl PR4 (gar, #3803)