feat(controller): support graceful application-requested stop

[shuangwu]

Readiness

Ready for review. Full-job GPU validation covers colocated RL and two-replica SFT, including stopping during active validation and injected final-checkpoint failure. Full CI is still required before merge.

Summary

Add controller-owned await controller.request_stop(reason) for successful early completion across disaggregated RL, colocated RL (including colocated-separated execution), and single-/multi-replica SFT.

• First reason wins. Acceptance is not a completion acknowledgement.
• Finish issued work and active validation, save at the actual completed step (including zero), and preserve the configured training horizon.
• Disaggregated RL closes new prompt/result admission and reuses the existing terminal-command protocol. Colocated RL allows prompts needed by the issued iteration and observes the terminal command before another generation iteration, including while waiting for weight synchronization.
• SFT replica leaders rendezvous at a controller-owned step boundary and broadcast one decision to all ranks. Stop cannot revoke permission already granted to an update. This adds one controller round trip per replica per update.
• SFT flushes asynchronous saves before completion acknowledgement. Rank-local save failure cannot produce a success acknowledgement; all participating replicas must confirm the same final step.
• Freeze stop membership and reject requests before initialization. No elastic recovery, transport recovery, or immediate cancellation is introduced.
• Applications own stop thresholds and checkpoint contents. Existing trainer checkpoint interfaces are retained.

Includes the controller API, HTTP forwarding for colocated facades, documentation, regression tests, and portable validation/failure-injection fixtures.

Validation

• Focused requested-stop, terminal-drain, and checkpoint regression suite: 63 tests passed on the current change. Earlier broader regression coverage passed 131 tests.
• Two-process CPU/Gloo canary passed shared stop decisions, rank-local final-save failure suppressing success ACKs, and successful save/finalize acknowledgement. Reproduce with PYTHONPATH=. torchrun --standalone --nproc-per-node=2 tests/generic_stop_rank_canary.py.
• Full-job two-replica SFT: partial-progress stop saved step 2 of 8; stopping during active validation saved step 0 of 8 after validation completed. Both cases received both replica completion ACKs and exited cleanly. Injected final-save failure withheld the failed replica's ACK and never reached terminal success; the finite outer timeout ended remaining processes.
• Full-job colocated RL: partial-progress stop saved step 2 of 8; stop during active validation completed the already-issued update and saved step 1 of 8. Both reached terminal success and exited cleanly. Injected final-save failure exited nonzero with no completion ACK or terminal success.
• Prior live disaggregated custom-trainer canaries covered step-zero stop, in-flight update stop, active validation, and final-save failure.
• Portable fixtures: tests/requested_stop_sft_live.py, tests/requested_stop_canary.py, and tests/configs/requested_stop_colocated.toml; reproduction and required assertions are in docs/requested_stop.md. The SFT fixture generates a tiny local model and dataset and uses the real checkpoint saver. The colocated fixture uses the companion custom trainer and test-only checkpoint format.
• Formatting, lint, and whitespace checks passed.

Scope and independence

Based directly on upstream main. Does not contain or replace #747. The new colocated and SFT live canaries ran without #747. Prior disaggregated failure-containment validation used a separate combined snapshot with an earlier #747 implementation; it does not establish behavior for redesigned #747.

The live cases do not establish multi-node, sharded-model, or colocated-separated lifecycle coverage. A missing participant or stalled collective cannot be reported as successful completion. This change adds no scheduler-wide failure propagation or recovery guarantee; use a finite outer job limit.