[faresobeid]

The orchestrator checkpoint now saves queued samples and open train groups together with the train-source cursor, so a resume does not lose them.

Why

On main, progress.pt holds only progress and train-source state. A resume loses accepted samples queued for the next batch, the finished rollouts of open groups, and the groups themselves; the sampler has already moved past their tasks, so those tasks are silently skipped.

What

• Save. progress.pt also holds the train sink's state, in one file with the cursor:
  • pending_batch: TrainingSamples plus a QueuedTrace record each (env, dispatch policy version);
  • every open train group: env, task, finished episodes (as JSON; parametrized pydantic episode classes do not pickle);
  • groups fully emitted by the dispatcher whose last results still wait in out_q (kept in emitted_train_groups until the sink consumes them).
  • The snapshot is taken on the event loop and pickled in a thread; the write is fsync'ed before the rename. A pause saves the same snapshot.
• Load.
  • Queued samples go back into pending_batch, the staleness sweep runs against the resumed step, and a full batch ships as soon as the main loop starts.
  • Each saved group keeps the finished episodes still within max_off_policy_steps (min_fresh_version, same rule as the queue sweep). TrainSource.resumed asks the dispatcher for the missing group_size - kept members (same task, same group id) before any new task; pending resume requests are saved too.
  • A saved member that ages out before its group completes is left out of the group's scoring, so advantages only use members that train.
  • A checkpoint without the new key loads as before.
• Batch cut. process_batch sorts the queue oldest-first by QueuedTrace.policy_start (stable; a group stays together) before cutting batch_size, so the overflow is the youngest traces, with the most staleness budget left.
• Not saved: in-flight rollouts (they rerun with their group) and out_q results (regenerated).

Validation

• test_train_sink_resume.py: resume-time staleness drop; test_resumed_group_scores_only_members_within_max_off_policy_steps: scoring of a resumed group.
• pytest tests/unit/orchestrator tests/unit/test_configs.py tests/unit/utils -m "not gpu" on current main: 418 passed. ruff check, ruff format --check and uv lock --check are clean.
• GPU, 1 node, reverse-text Qwen3-0.6B, ckpt.interval=5, SIGKILL of rl right after v9, then rl --resume (not rerun after the main merge):
  • NCCL: Resuming with 0 queued traces and 6 open groups (6 finished episodes kept), trained to v14.
  • Filesystem, 2-rank trainer: 3 open groups (1 finished episodes kept), trained to v14.
  • The queue overflow was empty at every checkpoint, so queued-sample replay is covered only by unit tests.

Related

• fix(resume): resume from the latest complete checkpoint, not the newest step dir #3906 is compatible: everything stays in progress.pt, its completeness marker.
• fix(orchestrator): drop restored work that points at deleted by-handle payload files #4044 (stacked on feat: pass router-replay ids and sampling masks by handle on multi-node runs #3902 and this PR) drops restored samples whose by-handle payload files are gone.
• Likely conflicts with refactor: orchestrator as hook-wired components with per-component configs #3652 (ckpt.py, train_sink.py).

---

Note

Medium Risk
Changes orchestrator resume semantics and batch assembly (staleness, group scoring, checkpoint I/O); incorrect state could skip work or ship wrong off-policy data, but behavior is guarded by existing max_off_policy_steps rules and new tests.

Overview
Orchestrator checkpoints now persist in-flight training work alongside progress and dataset cursor state, so --resume no longer drops queued samples or partially finished rollout groups.

progress.pt gains a train_sink blob: compiled samples in pending_batch, per-trace QueuedTrace metadata (env, dispatch policy version), and open train groups (env, task, finished episodes as JSON). Saves use snapshot_state() on the event loop and async thread pickle with fsync + atomic rename. On load, queued traces are restored (and can ship immediately), stale members are dropped per max_off_policy_steps, and TrainSource.resumed re-dispatches missing rollouts for saved groups before new tasks. The dispatcher tracks fully emitted but not yet consumed groups via emitted_train_groups; the sink sorts the batch queue oldest policy version first and scores resumed groups only on members still fresh enough.

Docs in training.md describe resume behavior; unit tests cover resume staleness and mixed-version group scoring.

^{Reviewed by Cursor Bugbot for commit 7e5b946. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Stale Bugbot comment from a previous run.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit b2b4a96. Configure here.}