[rollout] feat: schedule owned fully async work

[ashtonchew]

Summary

• Integrates source reservations and exact execution receipts into FullyAsyncRolloutFn while preserving the composed SubmissionScheduler from PR #1673 and GroupBuffer from PR #2030.
• Preserves group- and sample-completion submission policy, FIFO/LIFO consumption, and stale or overflow eviction while binding owned work to exact reservation settlement.
• Bounds executing work, all retained reservations, and completed prefetch independently, and reopens each capacity only at its corresponding terminal or settlement transition.
• Keeps each completed group owned until its train-batch lease commits or rolls back. Receipt-bound inference execution is defined separately in PR #52.

Context

PR #50 defines the reservation ownership state machine and its exact-attempt receipts, but it does not yet change how the scheduler acquires or hands off work. The composed integration base uses SubmissionScheduler from the current open PR #1673 to choose group- or sample-completion backfill and GroupBuffer from the current open PR #2030 to order, bound, and evict completed groups.

Owned scheduling must extend both components rather than introduce a parallel scheduler or buffer. A submission still follows the selected scheduler policy, and a completed group still follows the configured buffer order and eviction policy. The difference is that every owned entry carries the exact terminal receipt needed to acknowledge or requeue its source reservation.

Execution capacity can reopen when an attempt becomes terminal, but the source reservation must remain retained until training accepts the batch or its lease requeues it. Completed prefetch needs a third bound because terminal work may wait independently of both execution and training.

Description

• Enables reservation-backed scheduling when all three owned capacity limits are present and retains the existing scheduling behavior otherwise.
• Sends legacy and owned submissions through the same SubmissionScheduler, including its sample-completion callback and backfill accounting.
• Extends GroupBuffer to hold ordinary source-result pairs or receipt-bound owned completions while preserving its FIFO/LIFO order, staleness metrics, and stale or overflow selection.
• Validates each reservation's parent count and complete, unique parent identities before running generation on a pristine copy.
• Requires every one-to-many output to preserve the identity of its parent slot before the group can enter a training batch.
• Applies separate limits to executing samples, retained source groups, and completed prefetched groups.
• Returns leased training batches whose commit acknowledges the exact terminal receipts and whose rollback requeues the corresponding pristine reservations.
• Routes stale or overflow eviction through exact reservation rollback before releasing retained or completed capacity.
• Requeues aborted, malformed, and failed terminal work without releasing ownership before source settlement succeeds.
• Drains active and queued owned work before surfacing a fatal worker failure.

Test Plan

• At exact head dc22978d1d6e3a385f6f18e54d88fd90295f0b4f, python -m pytest -q -p no:cacheprovider tests/fast/rollout/test_fully_async_rollout.py passed 44 tests.
• Validation head d6cd677a89759d6ea5316d2805a3d9cb6f392a1b passed the 8xB200 composition run for candidate 7d53849102b1804fd4a8b21515aa01ca23f89b48: one real actor optimization step, exact trained-weight reset and rebroadcast recovery, update-boundary rollback and replay, one nonempty admission hold, and clean inner and outer Ray completion. The validation head is two validation-only commits above the candidate and includes the router startup-deadline patch, so it is composition-level evidence; it does not establish exact branch-head GPU proof.