feat(rollout): persist and recover partial rollout groups

[macandro96]

What does this PR do?

Persists partial-rollout lineage with native TQ state and recovers it through the Single Controller restart lifecycle.

• Saves rollout lineage beside native TQ state, replay metadata, the dataloader cursor, and controller checkpoint metadata.
• Restores TQ through the bootstrap data-plane client before runtime consumers are constructed.
• Reconstructs finalized replay ownership from metadata while tensor payloads remain authoritative in TQ.
• Validates canonical rows, live staging ownership, receipt references, and lineage digest before recovery.
• Clears unreferenced staging rows and drops unfinished groups that are stale under the restored sampler policy.
• Reuses sealed siblings and redispatches only missing or abandoned generation indices.
• Applies bounded infrastructure retries during startup recovery.
• Resolves compact prompt references through the restored input stream rather than serializing large prompts.
• Makes the checkpoint barrier re-entrant.
• Runs restored recovery beside fresh dataloader dispatch after ownership and buffer permits are reconstructed; both share the bounded rollout-worker budget.

Frequent rollout-only snapshots are added by #3508.

Stack

• Layer 2 of 4.
• Previous: feat(rollout): add stable partial-rollout lineage #3506.
• Next: feat(rollout): add periodic partial-rollout recovery snapshots #3508.

Recovery contract

1. Load the selected trainer checkpoint and native TQ snapshot.
2. Restore replay metadata, lineage, sampler state, and dataloader position.
3. Validate TQ inventory against finalized and unfinished ownership.
4. Reserve restored buffer capacity before fresh dispatch.
5. Recover unfinished groups concurrently with fresh work.
6. Reuse sealed siblings and redispatch only incomplete siblings.

The persisted controller sidecar remains metadata-only.

Validation

• Unit coverage includes persistence, inventory validation, selective redispatch, finalizer replay, stale cleanup, capacity ownership, sampler restore, and recovery/fresh-dispatch overlap.
• The PR remains draft while the refreshed focused suite, recovery functionals, and full CI are rerun.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[github-actions]

✅ Submodule Fast-Forward Check Results

Check based on commit: c232604 (PR #3507 from amahishi/partial-rollout-recovery)

✅ Submodules that are properly updated:

Gym: ✅ PR branch is ahead of amahishi/sc-tq-token-capture-recovery branch (fast-forward)

All submodule changes look good! ✨

[tianyi-zhang-02]

@macandro96 — heads-up that this draft and my #3783 rewrite the same three lines in SingleControllerActor.run(), and I'd rather flag it now than have you hit it at rebase.

Both replace

done, _ = await asyncio.wait(set(tasks), return_when=asyncio.FIRST_COMPLETED)

with a loop that keeps waiting on the still-pending set, closing the same window: the rollout pump exhausting normally is the first completion, so under the current code a watchdog raise during the train pump's drain lands on a task nobody awaits and gets swallowed by the finally: gather(..., return_exceptions=True).

Two differences worth knowing before you rebase:

• probe_task. Your version has no probe_task branch — grep probe_task on this diff returns nothing — but main now creates one when generation_fleet_health is enabled (single_controller.py:353-366). As written the rewrite would drop that branch.
• Ordering. for task in done: await task iterates a set, so if the watchdog and a pump complete in the same wait batch, which exception surfaces is nondeterministic. fix(sc): keep the watchdog armed while the train pump drains #3783 keeps the explicit probe → watchdog → rollout → train order, which is what makes a stall get reported ahead of a pump whose only symptom is "waiting".

#3783 is 177/-16 on two files, based on main, mergeable, and adds tests/unit/single_controller/test_run_supervision.py covering the drain case and the probe path — this draft has no test of the supervision loop (its test_single_controller.py addition is a _train_pump test).

Suggestion, entirely yours to take or leave: let #3783 land the loop on its own, then drop your copy at rebase — it's incidental to the rollout-recovery feature this PR is actually about, and it'd stop gating a small fix behind a 1877-line stack. No new-file collision either way. :)

[macandro96]

Will be superseded by #3923