[feat][SFT] Add native Fireworks training backend - #2057
Conversation
# Bounded-Memory Megatron Training and Colocation ## Summary This PR adds controls that bound temporary training memory and make colocated Megatron offload behavior complete. ## Changes | Change | Explanation | | --- | --- | | CPU-resident policy microbatches | Keeps the DP-sharded policy batch on CPU and transfers one nested microbatch to CUDA immediately before its forward step. | | Bounded vocabulary entropy | Chunks Megatron vocabulary entropy by an explicit size or an automatically calculated memory budget, and skips fully masked chunks. | CPU-resident policy microbatches is opt-in; automatic vocabulary-entropy chunk sizing, which defaults to a 512 MiB temporary-memory budget. ## Memory Validation | Variant | Memory Spike | Outcome | | --- | ---: | --- | | Earlier configuration | 9.3 GiB | OOM easily as the context/vocab size gets larger | | Bounded-memory configuration | 0.5 GiB | All memory spikes are under control | --------- Co-authored-by: Jinghan Yao <jinghan.yao1@anyscale.com>
…y-AI#1892) This PR adds an AMD ROCm path for running SkyRL’s Tinker-compatible API with FSDP training and vLLM ROCm inference. ## Changes - Add `docker/Dockerfile.amd` for a ROCm-based SkyRL image. - Add `docker/pyproject.amd.toml` to avoid CUDA-specific dependencies in the AMD image. - Add `examples/train/amd/run_tinker_server_amd.sh` to launch the Tinker API on AMD nodes. - Add AMD example clients: - `tinker_hello_world.py`: small LoRA smoke test. - `grpo_client.py`: GSM8K GRPO/PPO training example. - Add `examples/train/amd/README.md` with build/run instructions.
## What Adds a pretokenized data path to the SFT trainer: point `pretokenized_dataset_paths` at local store(s) of already-tokenized rows and train directly, skipping online tokenization (`tokenize_chat_example` / `tokenize_sft_example`) entirely. Useful when a data pipeline tokenizes offline (e.g. on a Spark/Ray data cluster). > **Scope note (per review):** this PR supports **local paths only**. Cloud-path (S3/GCS) ingestion is split into the follow-up draft NovaSky-AI#1933, where it lands once for both pretokenized stores and text-format datasets. ## How it works **New module `skyrl/train/dataset/pretokenized.py`** — `load_from_pretokenized(path, max_length)`, the pretokenized counterpart of `SFTTrainer._load_and_tokenize` (same `list[dict]` return shape, so everything downstream — collators, sequence packing, samplers, checkpoint/resume — works unchanged): - **Formats** (auto-detected, single file or directory of shards): Parquet, JSON-lines, raw Arrow IPC, or a HF `Dataset.save_to_disk` directory. - **Row schema**: unpadded `input_ids` plus a full-sequence 0/1 `loss_mask` (same length). `num_actions` is *inferred* from the first nonzero mask entry; window-form masks, `num_actions` columns, and HF-style `labels` are rejected with clear errors. `attention_mask` is optional and must be all-ones (padding stays collation-internal). The mask form covers instruction-following (1s on the response) and multi-turn conversational data (1s on every assistant turn). - **VLM ingestion**: rows carrying `pixel_values` + `image_grid_thw` pass through to the collator's `TensorList` path. Over-length VLM rows are dropped with a warning rather than truncated (consistent with the online VLM path); mixed text+VLM stores handle the null image columns parquet materializes on text rows. - **`max_length`** truncation mirrors the online path (prompt prefix kept, action window shrinks, empty-loss rows dropped). **Config (`SFTConfig`)** — integrated with multi-dataset SFT (NovaSky-AI#1883): - `pretokenized_dataset_paths: List[str]` — multiple stores are concatenated and mixed per `train_dataset_weights` via `DataMixingSampler`, exactly like `train_datasets`. Exclusive with `train_datasets`/`train_dataset_splits` (explicit error, no silent precedence): a pretokenized store is the output of a preprocessing job, so the user is assumed to have pre-split/filtered it — HF split syntax does not apply. - `eval_pretokenized_dataset_paths: List[str]` — each store becomes one eval set logged under `eval/{name}/`, named by `eval_dataset_names` (default: path basenames; collisions error). - `eval_interval` / `eval_before_train` accept pretokenized eval stores. ## Testing - **CPU**: 32 tests in `tests/train/test_sft_pretokenized.py` covering format detection, schema validation/normalization, truncation, VLM pass-through + collation, multi-store concatenation + mixing-sampler wiring, and config validation. Full SFT suite passes alongside. - **GPU e2e** (Qwen2.5-0.5B-Instruct, FSDP on 1×L4, alpaca tokenized offline with this repo's own tokenizer helpers into parquet, ingested from a local directory of shards + a separate local eval store): **https://wandb.ai/sky-posttraining-uc-berkeley/skyrl_sft/runs/lkpv0ip1** - Loss parity: pretokenized runs reproduce the online-tokenization runs' loss curves at identical seeds (e.g. step-5 train loss 1.084 vs 1.081). ## Notes / limitations - The offline pipeline must apply the same chat template as the trained model — token ids can't be verified at train time. - Streaming ingestion is out of scope (non-requirement). - VLM ingestion is covered at unit/collation level; no GPU VLM e2e in this PR. - Cloud paths (S3/GCS): follow-up draft NovaSky-AI#1933. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Avi Basnet <avigyabb@stanford.edu> Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
NovaSky-AI#1389 added model_role parameter to create_model but RayJaxBackend implementation was not updated Signed-off-by: Andrew Sy Kim <andrewsy@google.com>
…gher peak gpu_memory_utilization (NovaSky-AI#1934) ## Offload KV cache during non-colocated weight sync ### Motivation In non-colocated setups the inference engine stays fully awake during weight sync, so the KV cache occupies GPU memory the whole time. At high `gpu_memory_utilization` there's no headroom left for the weight-transfer scratch buffers, and the NCCL broadcast OOMs. This forces users to keep `gpu_memory_utilization` conservative. This PR adds an opt-in flag that sleeps the engine (freeing the KV cache) *during* the sync, so `gpu_memory_utilization` can be pushed much higher. For the fully-async trainer it also preserves in-flight generation across the sync — no aborts, no prefill recompute. ### What it does New flag: `generator.inference_engine.offload_kv_for_weight_sync` (bool, default `false`). Requires **non-colocated** placement and **non-LoRA** weight sync. It turns on vLLM sleep mode and changes `WorkerDispatch.save_weights_for_sampler`'s non-colocated path depending on the trainer: - **Synchronous trainer** (`fully_async.enabled=false`): generation is complete at sync time, so there are no in-flight requests. Plain `sleep() → wake_up(["weights"]) → broadcast → wake_up(["kv_cache"])` — the same three-phase pattern colocated mode already uses. - **Fully-async trainer** (`fully_async.enabled=true`): generation overlaps the sync. `pause_generation` (KEEP) freezes in-flight requests, then the per-worker `CuMemAllocator` is driven directly (so the scheduler is not resumed until KV is back). The KV cache is offloaded to CPU and restored so frozen requests resume seamlessly — **unless** `clear_kv_cache_on_weight_sync=true`, in which case the broadcast resets the prefix cache anyway, so KV is discarded (skipping the CPU copy) instead of offloaded. ### Implementation Driven entirely from SkyRL — **no vLLM patch**. The fully-async path deliberately avoids the `/sleep` + `/wake_up` HTTP endpoints (which route through `EngineCore.sleep`, force-clearing the prefix cache and preempting every running request at level ≥ 1). Instead it drives the allocator via two new `NewInferenceWorkerWrap` methods over `/collective_rpc`: - `skyrl_sleep_for_weight_sync(offload_kv)` — `allocator.sleep(offload_tags=("kv_cache",) if offload_kv else ())`. Discards the weights pool (the broadcast overwrites every parameter on wake) and either offloads or discards the KV cache. Model buffers live in the weights pool and aren't sent by the broadcast (e.g. non-persistent rotary `inv_freq`), so they're saved to CPU here and restored on wake — mirroring `GPUWorker.sleep(level=2)`. - `skyrl_wake_for_weight_sync(tags)` — `torch.cuda.empty_cache()` (return the broadcast's reserved blocks to CUDA so cumem can remap the KV pool at its fixed virtual addresses), then `allocator.wake_up(tags)`, which restores to the **same virtual addresses** so block tables stay valid. Restores buffers on the `weights` wake; re-inits fp8 KV scales on the `kv_cache` wake. Does not resume the scheduler. Files: - `config.py` — the flag. - `inference_servers/utils.py` — `enable_sleep_mode = colocate_all or offload_kv_for_weight_sync`. - `inference_servers/new_inference_worker_wrap.py` — the two worker methods. - `inference_servers/remote_inference_client.py` — `sleep_for_weight_sync` / `wake_for_weight_sync` client wrappers. - `workers/worker_dispatch.py` — orchestration in `save_weights_for_sampler`. - `train/utils/utils.py` — validation (non-colocated + non-LoRA). - `.claude/docs/weight_sync.md` — docs. ### Usage ```bash uv run ... \ trainer.placement.colocate_all=false \ generator.inference_engine.gpu_memory_utilization=0.95 \ generator.inference_engine.offload_kv_for_weight_sync=true ``` ### Validation Reproduced and fixed on the fully-async gsm8k example (Qwen2.5-1.5B, non-colocated, 1 inference + 1 policy GPU): | `gpu_memory_utilization` | `offload_kv_for_weight_sync` | Weight sync | |---|---|---| | 0.99 | `false` (baseline) | ❌ `CUDA out of memory. Tried to allocate 1.27 GiB. GPU 0 … 607 MiB free` in the broadcast | | 0.99 | `true` | ✅ passes; ran to `max_training_steps`, valid rewards, eval OK | The KV offload/restore is a GPU↔CPU copy of the whole KV pool each sync (its cost scales with the KV pool size), which is the tradeoff for zero lost generation work. ### Tests - **CPU** (`tests/train/test_config.py`): validation coverage — rejects colocated / LoRA; accepts sync trainer and both `clear_kv_cache_on_weight_sync` settings under fully-async. - **GPU** (`tests/backends/skyrl_train/gpu/gpu_ci/test_offload_kv_weight_sync.py`): a long request is kept in-flight while a real training step + weight sync run; asserts it finishes with a non-abort stop reason and that the offload path actually ran. Passes. The synchronous-trainer path reuses the standard `sleep()`/`wake_up()` sequence already exercised by the colocated `test_save_weights_for_sampler` case, so it's covered by CPU validation rather than a dedicated GPU test.
… env.step (NovaSky-AI#1900) ## What does this PR do? Splits each trajectory's rollout time into engine time and env time. Today only the total exists (`trajectory_time_completion_*`, NovaSky-AI#1804), so a slow rollout could be engine-bound or env-bound and the metrics cannot tell which. New metrics: - `generate/trajectory_time_llm_{mean,p90,max}`: time in the inference-engine call, summed over turns - `generate/trajectory_time_env_{mean,p90,max}`: time in `env.step()`, summed over turns - `generate/trajectory_time_other_{mean,p90,max}`: everything else in `e2e_time` (env construction and teardown, tokenization, chat templating, event-loop scheduling), so the three bands sum to the total and unattributed time is visible All time metrics share the `generate/trajectory_time_` prefix so one glob pulls the whole breakdown. ## How `agent_loop` accumulates one `time_splits` dict per trajectory (keys `llm`, `env`). It travels as a single `trajectory_time_splits` dict-of-lists on `GeneratorOutput`. Stats are recomputed from the raw lists in `concatenate_generator_outputs`, which every logging path goes through, because per-group aggregates like a p90 cannot be combined. The `other` band is e2e minus every recorded split component. ## Scope NovaSky-AI#1925 (env setup time band) is stacked on this PR. ## Tests - `test_llm_vs_env_time_split_metrics` (new): runs `generate()` with an env deliberately slower than the engine and asserts each sleep lands in the right component. - Exact-value tests for the metric math and its concatenation are in `test_generator_output_utils.py`. Step-wise replication and the `GeneratorOutput` field guardrail are asserted in existing tests. - `uv run --isolated --extra dev --extra skyrl-train pytest tests/train/generators/` passes, 51 tests. ruff and black clean on changed files. --------- Co-authored-by: Seiji Eicher <eicherseiji@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ner (NovaSky-AI#1809) ## Summary Overlap the next SFT `StatefulDataLoader` batch with the current GPU step. A single background thread advances the dataloader and performs collation while the current batch runs forward/backward. This rebased version integrates with the stateful dataloader and custom sampler support added in NovaSky-AI#1842: - `AsyncBatchCollator` provides a one-worker, one-future buffer with strict step matching. - `SFTTrainer.train()` consumes the buffered batch, handles dataloader exhaustion and epoch transitions, and always joins the worker on exit. - Checkpoints save the dataloader state after the current batch, not the already-collated next batch, so resume does not skip data. - `SFTConfig.async_batch_collation` defaults to `True`; set it to `False` for the serial path. ## Validation - Async and serial batches match across epoch boundaries for default and packed collators, including partial tail batches. - Checkpoint-resume coverage verifies that a step-N checkpoint resumes at the buffered step-N+1 batch. - `AsyncBatchCollator` unit tests cover step mismatch, single-slot enforcement, worker errors, draining, and shutdown. - Focused local suite: `54 passed` across `test_async_batch_collation.py` and `test_sft_dataloader.py`. - Ruff 0.11.9 and Black 24.10.0 pass on all touched files. --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ovaSky-AI#1934 (NovaSky-AI#1941) # What does this PR do? Fixes failing CPU tests after NovaSky-AI#1934 NovaSky-AI#1934 introduced a new `generator.inference_engine.offload_kv_for_weight_sync` flag that is used in `WorkerDispatch.save_weights_for_sampler`. We need to add this field to the mocked cfg object used in `tests/backends/skyrl_train/distributed/test_megatron_correctness::TestWeightSyncPauseFlush` Signed-off-by: SumanthRH <sumanthrh99@gmail.com>
- Add a native Megatron + local vLLM DAPO launcher for dense `Qwen/Qwen3.6-27B` with rank-32 LoRA on 4×8 H200s. - Preserve the validated workload: batch 128, group size 16, 8,192-token responses, TP4, eight inference engines, AIME 2024/2026 evals, and no checkpoint writes. - The LoRA path completed 17 consecutive training steps and entered step 18 without a training failure. The [W&B report](https://wandb.ai/trajectory-ai/qwen3_6_dapo_lora/reports/Qwen3.6-27B-SkyRL-DAPO-LoRA-%E2%80%94-reward-and-AIME-curves--VmlldzoxNzU3NjY5OA==) ## Testing ```bash bash -n examples/train/megatron/run_megatron_dapo_qwen3.6_27b_lora.sh git diff --check ``` The launcher passed shell syntax and whitespace checks; the linked 4-node run validates the native LoRA training path end to end. --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…Sky-AI#1923) ## What does this PR do? Publishes which macro-phase the fully async training loop is in as a Prometheus gauge. Today, joining "what was the loop doing" to "how busy were the GPUs" means hand-correlating the experiment tracker against Prometheus by wall-clock. New metric: - `skyrl_training_phase{phase=...}`: 1.0 for the active phase, 0.0 for the rest. Phases match the loop's `Timer()` keys (`wait_for_generation_buffer`, `convert_to_training_input`, `run_training`, `sync_weights`, `eval`, `save_checkpoints`) plus the default `generating`, so the wandb `timing/*` keys and the Prometheus label share one vocabulary. Ray exports it to the same Prometheus that scrapes node GPU metrics, so per-phase GPU utilization is a single-store query: avg(ray_node_gpus_utilization) and on() (ray_skyrl_training_phase{phase="eval"} == 1) Prometheus also survives a cluster restart, when the tracker may be unreachable. ## How New module `skyrl/train/utils/phase_metrics.py` with `TrainingPhaseGauge`, following the one-module-per-concern layout of `ray_gpu_monitor.py`. The trainer constructs it in `__init__` and sets the phase at each existing `Timer()` boundary. Each transition writes only the two changed series; construction seeds all series so PromQL selectors never hit a missing one. Best-effort: it no-ops when Ray metrics are unavailable and warns on unknown phase names, so it never breaks training or tests. ## Tests - `tests/train/utils/test_phase_metrics.py` (new): exactly one phase active, context-manager restore, unknown-phase rejection, no-op when Ray is unavailable. 5 passed. - ruff and black clean on changed files. End to end needs a GPU cluster.
…1944) The lint step used unpinned `uvx ruff check`, which picks up ruff's expanded default rule set in newer releases (0.16.0 enables ~414 rules vs ~60 in 0.11.9) and fails CI. Pin to 0.11.9 to match the rev in .pre-commit-config.yaml so CI and local pre-commit stay consistent. Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ovaSky-AI#1924) ## What does this PR do? Logs the rollout buffer state to Prometheus. Generation and training run concurrently: generators fill a buffer and the trainer pulls `mini_batch_size` batches per step. The buffer depth tells which side is the bottleneck. Near zero means the trainer is starving for rollouts; near the cap means generation is paused at the staleness limit. Same idle-GPU symptom, opposite cause. New metrics: - `skyrl_gen_buffer_qsize` (Prometheus, per step): buffer depth at step start. The W&B counterpart is NovaSky-AI#1930's `async/gen_buffer_qsize_at_wait_start`. - `skyrl_gen_buffer_maxsize`, `skyrl_mini_batch_size` (Prometheus, once at startup): run constants for dashboard panels, e.g. a fill ratio or capacity line. - `async/keep_rate` (W&B) and `skyrl_gen_group_keep_rate` (Prometheus): `mini_batch_size / (mini_batch_size + dropped)` under `sample_full_batch`, the zero-variance drop rate for the last mini-batch, so a deep buffer of mostly droppable groups is distinguishable from a usable one. ## How `ScalarGauges` (in `skyrl/train/utils/metrics.py`, alongside the phase gauge from NovaSky-AI#1923) lazily creates one Prometheus gauge per name at the first `set()`. The trainer publishes step-start values directly because the tracker flushes at step end, after training and weight sync, when the buffer has already moved on. ## Tests - `tests/train/utils/test_metrics.py` (extended): gauge created once per name with its first description, values coerced to float. - ruff and black clean on changed files. End to end needs a GPU cluster.
…y-AI#1838) ## Summary The Megatron `torch_dist` checkpoint save is fully parallel across ranks but **synchronous** — training blocks for the entire shard-write duration on every `ckpt_interval`. For large models checkpointing frequently, that stall is significant. This adds an opt-in `MegatronConfig.async_dist_ckpt_save` (default `False`, so existing behavior is unchanged). When enabled, `dist_checkpointing.save` stages each rank's shards to host memory and writes them to disk in a background process, letting training resume immediately. This replaces the long-standing `TODO(tgriggs): Support configurable async saves` in `megatron_strategy.py`. ## Changes - `MegatronConfig.async_dist_ckpt_save` knob (off by default). - `MegatronStrategy.save_checkpoint`: passes `async_sharded_save` through and schedules the returned request on the existing persistent `AsyncCallsQueue` instead of asserting it is `None`. Blocks on the previous async save before issuing a new one. - `MegatronStrategy.finalize_pending_saves()`: drains in-flight writes; called at the start of `load_checkpoint` and exposed up through `Worker` → `WorkerDispatch` → end-of-training in the sync, fully-async, and SFT trainers. ## Correctness - On-disk format is identical to the synchronous save. - The pending write is finalized before the next save, before any reload, and at end of training. - Async falls back to synchronous for cloud destinations, where `local_work_dir` uploads on context exit and would otherwise race a partial checkpoint. Only active for local/shared filesystems. ## Test plan - [ ] GPU run with `async_dist_ckpt_save=true`: verify checkpoints are byte-identical to sync saves and reload correctly. - [ ] Confirm training resumes before the disk write completes (timing/save_checkpoint drops). - [ ] Verify final checkpoint at end of training is fully written before exit. --------- Signed-off-by: Vu Dinh <vudinh@outlook.com>
…ts (NovaSky-AI#1947) # What does this PR do? The MTP tests from NovaSky-AI#1832 were authored for an 8-GPU and fail on the megatron GPU CI (l4_ci = single 4x L4, 24GB): they OOM or time out on placement groups. Fixes in this PR: - `test_mtp_grad_isolation`: parametrize (tp,dp) -> [(4,1),(2,2)] and skip a combo when tp*dp > device_count(). Drops the tp=1 case and the 8-GPU (4,2) case; (2,2) preserves the DistributedOptimizer DP-shard param-ownership coverage on a 4-GPU node. - `test_mtp_replay_vs_native`,`test_mtp_weight_roundtrip`, `test_mtp_packed_vs_unpacked`: forward-only / weight-export probes that never train. Set policy.inference_only_init=True to skip the DDP grad buffer + DistributedOptimizer (the OOM / grad-clip all_reduce crash site), so the model fits at TP=1 on a 24GB L4 (packed_vs_unpacked's MiMo-7B is ~14 GiB of bf16 params with no grad buffer). Validated on L4 CI: all four files pass after the fixes in this PR Also made an unrelated fix by removing a stale log line in `SkyRLTrainBackend` (it is no longer experimental) --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…-AI#1950) Signed-off-by: SumanthRH <sumanthrh@anyscale.com>
…vaSky-AI#1859) ## What Add `cispo.cispo_anchor: "old" | "rollout"` to select which behavior policy the CISPO clamped importance-sampling ratio is anchored on. Default `"old"` preserves current behavior exactly. ## Why CISPO currently anchors its clamped IS ratio on the recomputed old log-probs (π_θ / π_old), where π_old is recomputed fresh by the training backend at experience-prep. That ratio therefore only deviates from 1 once π_θ has moved away from π_old — i.e. only when the batch takes **more than one gradient update** (`update_epochs > 1`, or `train_batch_size > policy_mini_batch_size` so there are multiple minibatch steps). Fully-async training takes exactly **one** gradient step per batch (it hard-asserts `train_batch_size == policy_mini_batch_size` and `update_epochs == 1`), so at loss time π_θ == π_old exactly → ratio ≡ 1 → the CISPO clamp never bites. Stock CISPO is effectively inert under fully-async. With `cispo_anchor="rollout"` the ratio is anchored on the rollout/sampler log-probs (π_θ / π_rollout) instead. Unlike π_old, π_rollout is **not** a fresh backend recompute, so the ratio deviates from 1 even at a single gradient step, from two async-intrinsic sources: (1) staleness — the rollout was sampled by an older policy version than the current trainer weights; and (2) sampler/trainer engine mismatch — vLLM logprobs differ from the training-backend logprobs even for identical weights. So the clamped objective genuinely engages under async. This is the clamped (cap-not-zero) counterpart of the existing `rollout_is` loss, which anchors on π_rollout but HARD-ZEROS out-of-range tokens; CISPO caps them so every token keeps a bounded gradient. ## Notes - Default `"old"` is a no-op — existing runs are unchanged. - The rollout-anchored ratio is itself the off-policy correction, so stacking TIS (`off_policy_correction.tis_ratio_type`) is rejected to avoid double-counting. - `"rollout"` requires `rollout_logprobs` (asserted); the trainer's skip-fwd path is made anchor-aware so the unused old-logprobs forward is skipped when the rollout anchor makes it unnecessary. ## Tests - `test_losses.py`: rollout-anchor loss value, the `rollout_logprobs` requirement, the TIS double-count guard, and config validation. - `test_skip_fwd_logprobs.py`: the cispo+rollout skip-path. --------- Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: tbalestri-lila <tbalestri-lila@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: SumanthRH <sumanthrh@anyscale.com>
… + RL) (NovaSky-AI#1808) ## Reviewers: Where to Look The behavioral core is three vectorized collation rewrites — all must produce **bit-identical** tensors (dtype + layout) to the loops they replace. Focus review here: - **`skyrl/train/dataset/preprocess.py` — `convert_prompts_responses_to_batch_tensors` (RL).** The per-row `sequences` slice writes (`prompt` then `response`), the broadcast `attention_mask` / `action_mask` comparisons (`col >= pad_len`), and the right-aligned `loss_mask` / `rewards` / `logprobs` slice writes. Also the new `_reward_to_numpy` helper: confirm list vs tensor, `float32` cast, `detach().cpu()`, and the 1-D shape guard. - **`skyrl/train/dataset/collators.py` — `PackedDataCollator.__call__` (Megatron SFT FFD packing).** Highest-risk path. Check the per-sub-seq `sequences`/`attention_mask` slice copies, the right-shifted `loss_mask` write window (`full_mask[1:1+n_write]`) and its `write_end = min(row_offset + s - 1, loss_mask_width)` clamp reproducing the original `row_p < max_packed_len - 1` guard, and `total_nonpad` as a single reduction. - **`skyrl/train/sft_trainer.py` — `collate_sft_batch` (unpacked SFT).** The left-pad slice writes into preallocated arrays; confirm `sequences`/`attention_mask`/`loss_mask` dtypes stay `torch.long`. Sanity-check the equivalence oracle in **`tests/train/test_collation_vectorization_equivalence.py`**: each vectorized path is fuzzed against a faithful copy of the *original* loop with `torch.equal` plus explicit `.dtype` assertions (the packed test re-derives the FFD/DP-shard/`max_packed_len` decision inline). **Needs less scrutiny:** type-annotation widenings (`Union[List[float], torch.Tensor]`), the `import numpy as np` additions, docstrings/comments, and the untouched `rollout_expert_indices` MoE branch (intentionally out of scope — byte-identical to before). --- # What does this PR do? The controller builds every training batch on the main process before dispatching it to the workers. All three collation paths did so with per-token / per-sample Python loops that dominate the controller-side collate wall-time and serially block the GPU at large batch sizes. This PR replaces those loops with NumPy slice-assignments and broadcast comparisons. **Outputs are bit-identical** (same dtypes, same layout) for all inputs produced in practice — this is a pure CPU-side latency optimization, not a behavior change. ## Changes - **`PackedDataCollator` (Megatron SFT FFD packing), `skyrl/train/dataset/collators.py`** — the per-bin packed row tensors (`sequences` / `attention_mask` / `loss_mask`) were built with a per-token Python loop over the reconstructed full loss mask. Each sub-seq is now written with one C-level copy, and `total_nonpad` is a single vectorized reduction. - **`collate_sft_batch` / `DefaultCollator` (unpacked SFT), `skyrl/train/sft_trainer.py`** — each left-padded row is written with a single slice assignment into a preallocated array instead of building a per-example padded Python list. - **`convert_prompts_responses_to_batch_tensors` (RL), `skyrl/train/dataset/preprocess.py`** — the left-padded `sequences` are built with two slice copies per row, and the fixed-width `attention_mask` / `action_mask` / `loss_mask` / `rewards` / `logprobs` tensors are produced with broadcast comparisons / slice writes instead of per-token Python loops. This covers the SFT (packed + unpacked) and RL training-batch construction paths; the RL and SFT data paths are separate functions, so each is vectorized independently. The RL change is inherited unchanged by all `RayPPOTrainer` subclasses (sync / async / full-context / agentic), since none override `convert_to_training_input`. ### Intentionally out of scope - **MoE router-replay (`rollout_expert_indices`).** The optional `rollout_expert_indices` branch in `convert_prompts_responses_to_batch_tensors` is left exactly as-is — only the dense per-token batch tensors are vectorized. That branch is byte-identical to the prior implementation (zero correctness/regression risk); it is a narrow MoE-only path, so its residual per-sample loop is left for a follow-up rather than folded into this CPU-latency change. The new equivalence suite therefore does not exercise it (the oracle compares the six dense outputs; the 7th return value is intentionally discarded). - **Eval path.** Packing only fires on the training-step batch (`batch_size == self.batch_size`); on the eval path `PackedDataCollator` delegates to the un-packed `DefaultCollator`, so eval collation is unchanged by this PR. Notes on the bit-identical claim: - dtypes are preserved exactly: `int64` / `torch.long` for `sequences` / masks (incl. `action_mask`), `float32` / `torch.float` for `loss_mask` / `rewards` / `logprobs`. `dtype=np.int64` is pinned explicitly (NumPy's platform-default int is `int32` on Windows). - The RL reward path accepts Python lists and `float32` reward tensors (what the reward postprocessing produces today). A `requires_grad`, CUDA, or `bfloat16` reward tensor is not accepted; no reward producer in the repo emits those. - The `PackedDataCollator` loss-mask write window keeps the original `row_p < max_packed_len - 1` clamp. That `min()` is a defensive no-op — `max_packed_len` is `>=` every bin's packed length by construction — so the clamp never bites today; it is retained (and now commented) to preserve the original behavior exactly. ## Benchmarks Controller-side collate, single process, CPU, batch of 1024 (varying sequence lengths): | Path | Before | After | Speedup | |------|--------|-------|---------| | `PackedDataCollator` (FFD, dp=8) | 288.5 ms | 8.4 ms | ~34x | | `convert_prompts_responses_to_batch_tensors` (RL) | 92.8 ms | 14.5 ms | ~6.4x | | `collate_sft_batch` (unpacked SFT) | 98.4 ms | 16.2 ms | ~6x | ## Test plan - [x] New `tests/train/test_collation_vectorization_equivalence.py`: pins a faithful reference of each *original* loop and fuzzes the vectorized output against it with `torch.equal` plus explicit per-tensor `dtype` assertions — RL (with/without logprobs, list and `float32`-tensor rewards), unpacked SFT, and packed SFT across TP/PP/CP/DP configs. Because `torch.equal` is dtype-insensitive on matching values, the integer/float dtypes (`action_mask`/`attention_mask` `int64`/`long`, `loss_mask`/`rewards` `float32`) are pinned with explicit `.dtype` assertions. The packed test re-derives the FFD / DP-shard / `max_packed_len` decision inline as its own oracle, so any production drift surfaces as a `torch.equal` mismatch. (Mutation-checked: an injected off-by-one in any vectorized path fails the suite — note the unreachable `loss_mask` clamp is the one spot a localized off-by-one would not be caught, since it never fires under any in-practice input.) - [x] Existing `tests/train/dataset/test_preprocess.py`, `tests/train/test_sft_packing_collate.py`, `tests/train/test_packing_round_trip.py`, `tests/train/test_sft_tokenization.py` pass unchanged. - [x] `ruff` + `black` clean. ```bash uv run --isolated --extra dev --extra megatron -- pytest \ tests/train/test_collation_vectorization_equivalence.py \ tests/train/dataset/test_preprocess.py \ tests/train/test_sft_packing_collate.py \ tests/train/test_packing_round_trip.py ``` > Heads-up for reviewers: this overlaps open PR NovaSky-AI#1752 ([train] VLM SFT on Megatron), which edits the same `collate_sft_batch` loop and `TrainingInputBatch` dict to collect `pixel_values` / `image_grid_thw`. Whichever lands second needs a small rebase; if NovaSky-AI#1752 lands first, its per-sample VLM tensor collection should be reinstated inside the vectorized `for i, ex in enumerate(examples):` loop and its two keys re-added to the `from_numpy` batch dict. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
…he caller does not consume them (NovaSky-AI#1807) ## Summary Threads an explicit `return_per_token_outputs` keyword argument (default `True`) from the worker dispatch layer down to the `cross_entropy` branch of **both** backends on **both** the train and eval/forward paths, gating the per-token `loss_fn_outputs` build. When the flag is `False`, the per-token NLL, the two detached `[mb, seq]` D2H copies (logprobs + elementwise loss), and the `.tolist()` loop are skipped; each sequence gets an empty dict instead. The `loss` / `response_length` metrics and the `loss_fn_output_type` tag are unchanged. SkyRL's own `SFTTrainer` reads only `output.metrics` (loss / response_length), never `output.loss_fn_outputs`, so it now opts out — eliminating dead work on the SFT train + eval hot path. RL and Tinker callers pass nothing and keep the existing contract (default `True`). ## What changed - **FSDP** (`workers/worker.py`, `workers/fsdp/fsdp_worker.py`): accept and forward the flag through `forward_backward` / `forward`, and gate the per-token build in `_forward_backward_micro` (train) and `_forward_micro_with_loss` (eval). - **Megatron** (`workers/megatron/megatron_worker.py`, `workers/megatron/megatron_model_wrapper.py`): accept and forward the flag, gating the per-token build inside the shared `loss_func` (covers train + `forward_only` eval). - **Dispatch** (`workers/worker_dispatch.py`): plumb the flag through and document the reserved argument on `forward` / `forward_backward`. - **Caller wiring** (`train/sft_trainer.py`): `train_step` and `run_eval` pass `return_per_token_outputs=False`. The flag is a plain function argument rather than a `loss_fn_config` dict key or an `AlgorithmConfig` field: it is per-call request metadata, not algorithm configuration, and keeping it out of the config avoids both the key-validation issue in `build_nested_dataclass` and any pop/copy dance around the `AlgorithmConfig` merge. It is documented in the parameter docstring at each site that reads it — `_forward_backward_micro`, `_forward_micro_with_loss` (FSDP), and `forward_backward_mini_batch` (Megatron). ## Numerical equivalence / safety Byte-identical for all existing callers: - The default `True` at every level reproduces the exact prior code paths; no existing caller passes the flag. - `loss`, the backward pass, and all consumed scalar metrics (`loss`, `response_length`, `lr`) are computed before/independent of the gated block, so they are identical whether per-token outputs are kept or skipped. - `loss_fn_output_type` is a `WorkerOutput` field that always defaults to `"scalar"` (never set explicitly), so the type tag survives automatically — only the arrays become empty. The empty-dict-with-`"scalar"`-tag combination is only reachable behind the explicit opt-out whose sole caller ignores the payload. - RL's separate (non-`cross_entropy`) `loss_fn_outputs` else-branch is untouched; the RL trainer and Tinker backend pass no flag, so their contracts hold. Because the flag is a Python keyword argument and not a `loss_fn_config` key, it cannot be injected through the Tinker public API's user-supplied config at all. Note on the eval path: `run_eval` already iterates eval batches serially and reads only `output.metrics["loss"]`, so opting out there removes per-token work without changing any reported eval metric. ## Test plan - `tests/backends/skyrl_train/workers/test_sft_loss_fn_outputs_gate.py` (new, CPU): drives the real FSDP `_forward_backward_micro` / `_forward_micro_with_loss` `cross_entropy` builds on CPU; asserts default/explicit-`True` populate `logprobs` + `elementwise_loss`, `False` yields empty dicts, and `loss`/`response_length`/`lr` are identical across the flag. Includes a case where a real `eps_clip_low` override is passed via `loss_fn_config` alongside the flag, confirming the two are independent and the config merge still happens. Adds an RL-path test confirming the non-`cross_entropy` else-branch is ungated (logprobs still built; outputs + loss identical across the flag); that test disables `use_kl_loss`/`use_entropy_loss` explicitly (rather than relying on a default) so it isolates the gate from the KL/entropy terms. - `tests/train/test_sft_callbacks.py` (extended): assert `SFTTrainer.train_step` and `run_eval` pass `loss_fn="cross_entropy"` and `return_per_token_outputs=False` to the dispatch. - `tests/train/test_trainer.py`: the `_forward_backward_micro` mock in `test_forward_backward_batch_calculations` accepts the new argument. - `tests/backends/skyrl_train/gpu/gpu_ci/test_training_step.py` (extended, GPU): parametrized over FSDP + Megatron (the Megatron leg carries `@pytest.mark.megatron` like the sibling tests), DP=2; runs the real worker `forward_backward`/`forward` twice on the same dummy batch (flag default-`True` vs explicit-`False`) and asserts `loss` + `response_length` identical and `loss_fn_output_type == "scalar"` in both, with per-token outputs populated when kept vs empty when skipped. Run locally: ```bash uv run --isolated --extra skyrl-train --extra dev pytest \ tests/backends/skyrl_train/workers/test_sft_loss_fn_outputs_gate.py \ tests/train/test_trainer.py::test_forward_backward_batch_calculations ``` CPU run: 12 passed. The CPU suite for the new/extended tests runs automatically under the standard CPU CI job (`tests/backends/skyrl_train/` + `tests/train/`); `skyrl_train_tests` is green on this head. ## Generality & follow-ups Covered: both backends (Megatron `loss_func`; FSDP micro methods) and both the train (`forward_backward`) and eval/forward (`forward(loss_fn="cross_entropy")`) paths. RL and Tinker contracts preserved (default `True`). Intentionally out of scope: - The RL `loss_fn_outputs` build (separate else-branch, not `cross_entropy`) is untouched; the `forward(loss_fn=None)` pure-inference path is untouched. - The JAX backend builds `loss_fn_outputs` in a jit-traced path driven by a structured `LossFnConfig` dataclass, and is not reached by this argument — a possible follow-up, not a regression. - No config schema field is added — the flag is a per-call argument with a safe default, preferred over a global switch. Minor cleanup carried here: the `loss_config` merge guard in the two FSDP micro methods was relaxed from `if loss_fn_config is not None:` to `if loss_fn_config:` so an empty override dict skips a no-op `OmegaConf.merge`. This is a strict no-op for every existing caller (no caller passes `{}`, and `OmegaConf.merge(base, {})` is itself a no-op); it is incidental to the gate rather than required by it, and is easy to drop if reviewers would rather keep the diff minimal. ## Relationship to open PRs This gate is orthogonal/complementary to several in-flight efforts touching the same files: - **NovaSky-AI#1513** (SFT loss-aggregation rewrite of the same FSDP `cross_entropy` branch) — note it renames the SFT status key `loss`→`sft_loss`, so merge ordering matters; only the test coupling to the literal `loss` metric key would need a touch-up if it lands first. - **NovaSky-AI#1752** (VLM SFT on Megatron) — disjoint regions in the shared files. - **NovaSky-AI#1534** (preserve staged `forward_backward` loss_fn_outputs across DP ranks; `worker_dispatch.py`) — overlaps in `worker_dispatch.py`; the changes there are additive (a new forwarded argument plus docstring lines), so expect a trivial merge. The gate logic itself is composable with all of these. --------- Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: SumanthRH <sumanthrh@anyscale.com>
…reallocated buffer (lower peak memory) (NovaSky-AI#1806) ## Summary Refactor `ChunkedDistributedLogprob.backward` (the vocab-parallel chunked-logprob autograd function used by the Megatron worker for SFT cross-entropy and RL policy/ref losses) to stream each chunk's gradient into a single preallocated fp32 buffer instead of appending to a Python list and concatenating with `torch.cat` at the end. This lowers peak activation memory on the chunked backward path with **no change to numerics**. ## What changed `skyrl/backends/skyrl_train/distributed/megatron/model_utils.py`, `ChunkedDistributedLogprob.backward`: - Removed `all_grad_input = []` and the final `grad_input = torch.cat(all_grad_input, dim=1)`. - Preallocate once before the loop: `grad_input = torch.empty((batch_size, seq_size, partition_vocab_size), dtype=torch.float32, device=vocab_parallel_logits.device)`. - Renamed the per-chunk grad to `chunk_grad_input` and, after the `scatter_add_`, write it into its sequence slice: `grad_input[:, chunk_start:chunk_end, :] = chunk_grad_input`. The change is **unconditional** (no flag) because it is numerically byte-identical. It only engages on the chunked dispatch path (i.e. when `chunk_size < seq_len_local`). The non-chunked `DistributedLogprob`, `forward()`, and the vendored Triton fused-LCE path are untouched. `forward()` retains the list+`cat` form intentionally: its accumulator holds per-chunk `[batch_size, chunk_len]` log-prob tensors (tiny), not the `[batch_size, chunk_len, V//TP]` fp32 grads that make the backward `cat` expensive, so streaming it would not meaningfully lower peak. The old list-then-`cat` form kept every per-chunk `[B, chunk_len, V//TP]` fp32 grad alive **and** allocated the full concatenated output at the cat moment, so peak was ~2x the full `[B, seq, V//TP]` fp32 grad. Streaming drops peak to full-buffer + one live chunk = ~`(1 + 1/num_chunks)`x of the full grad. The win scales with chunk count; it is **not** a flat halving. ## Numerical equivalence / safety Byte-identical by construction: - The per-chunk math is unchanged: same `_compute_distributed_log_softmax` on the same fp32 slice, same `.exp()`, same `neg_`/`mul_`/`scatter_add_` formulation. `chunk_grad_input` is a fresh fp32 tensor, so the slice write is a same-dtype copy with no cast. - The chunks tile `[0, seq_size)` exactly and contiguously: chunk `i` covers `[i*chunk_size, min(seq_size, (i+1)*chunk_size))`, consecutive chunks meet with no gap/overlap, and `num_chunks = ceil(seq_size/chunk_size)`. `torch.cat(dim=1)` placed chunk `i`'s columns at those same `[chunk_start:chunk_end]` offsets, so values land at identical positions and each slice is written exactly once. - The new buffer is contiguous fp32 of shape `[B, seq_size, V//TP]` — identical shape/dtype/contiguity to the previous `torch.cat` output — and is fully overwritten by the full-coverage tiling, so no uninitialized `torch.empty` memory survives. Full-coverage tiling is a load-bearing invariant for the `torch.empty` buffer (a partial write would leak garbage silently); the prime-length coverage test below is the regression guard. A deliberate choice of a **separate fp32 buffer** (rather than writing in place into the autograd-saved `vocab_parallel_logits`): the separate buffer keeps the gradient in fp32 regardless of the saved logits' dtype, does not mutate an autograd-saved tensor (avoiding version-counter / double-backward hazards), and trades only ~`(1/num_chunks)`x extra memory for that safety. ## Test plan > Written to SkyRL CI conventions; `python -m py_compile` passes and lint is clean (`ruff`: all checks passed; `black --line-length 120`: unchanged) on all three files. - `tests/backends/skyrl_train/distributed/test_chunked_logprob_backward_streaming.py` (CPU lane): stubs `megatron.core.parallel_state` into `sys.modules` via a module-scoped autouse save/restore fixture (mirroring `test_preprocess_packed_seqs_cp.py`), uses a gloo `world_size=1` TP group (every `all_reduce` is the identity), and asserts `torch.equal` between the chunked-streamed grad and the single-shot `DistributedLogprob` grad across `chunk_size in {1,3,7,16,32,64}` x with/without OOV targets, plus edge cases (seq_len=1, all-in/all-out mask, a prime/ragged tiny-vocab config). This is the **byte-identity gate for the storage refactor**: the log-softmax + scatter-add backward math is purely per-position (reductions only over the vocab dim), so chunk boundaries cannot change any value, and the world_size=1 `torch.equal` fully validates the device/dtype-agnostic slice-write. The prime-length (`seq_len=17`, `chunk_size=5`) coverage test additionally pins that every sequence slice of the preallocated buffer is written (an unwritten `torch.empty` slice cannot coincidentally match the reference). Collected by the SkyRL-Train-CPU lane (`pytest tests/backends/skyrl_train/ --ignore=.../gpu`). Run with: ``` uv run --isolated --extra dev -- pytest -s \ tests/backends/skyrl_train/distributed/test_chunked_logprob_backward_streaming.py ``` - `tests/backends/skyrl_train/gpu/gpu_ci/megatron/test_chunked_logprob_backward_tp.py` (`@pytest.mark.megatron`): spawns TP NCCL ranks (parametrized `tp_size in {2, 4}`, guarded by a `device_count()` skip), runs `mpu.initialize_model_parallel`, shards the vocab across ranks, runs fwd+bwd on each rank's slice, and asserts (a) the rank vocab slices tile `[0, vocab)` exactly once via `torch.equal` on a coverage counter, and (b) each rank's local streamed grad matches the full-vocab single-process autograd reference columns **within fp32 tolerance** via `torch.testing.assert_close(atol=1e-5, rtol=1e-4)` — a tolerance is correct here because the cross-rank all-reduce reorders the fp32 reduction vs. the single-tensor reference (matching the existing GPU test's grad tolerance), while the no-overlap tiling check uses exact equality. Spawned ranks set the conftest-mandated NCCL env (`NCCL_CUMEM_ENABLE=0`, etc.) before `init_process_group`, since `mp.spawn` children do not inherit the runtime env set by the CI conftest. Run with (>=2 free GPUs; the `tp_size=4` case is skipped unless 4 are present): ``` uv run --isolated --extra dev --extra megatron -- \ pytest -s tests/backends/skyrl_train/gpu/gpu_ci/megatron/test_chunked_logprob_backward_tp.py ``` ## Scope & follow-ups - **Backends covered:** Megatron only, by nature — `ChunkedDistributedLogprob` is the vocab-parallel (TP/CP) chunked-logprob used on the Megatron worker, reached through the single `forward_backward_mini_batch` train-step chokepoint, so all Megatron-backed training pathways (SFT, sync/async/fully-async RL, full-context) inherit the optimization with no per-pathway edit. FSDP has no vocab-parallel chunked-logprob equivalent, so it is intentionally out of scope. - **Scope:** Limited to the single `backward` method; `DistributedLogprob`, `forward()`, and the fused-LCE path are unchanged. - **Deferred:** A direct streaming-vs-old-`cat` comparison at TP>1 was intentionally not added, since it would require shipping the pre-refactor `cat`-based backward as dead reference code. The byte-identity gate for the refactor itself lives on the CPU `world_size=1` lane (`torch.equal`); the GPU lane validates the distributed math against the reference (with tolerance) plus exact no-overlap tiling. ## Related PRs (merge ordering) - **NovaSky-AI#1765** (fused LM-head log-prob + entropy) edits the same `ChunkedDistributedLogprob.backward` and **keeps** the `all_grad_input = []` / `torch.cat` form. It refactors the per-chunk chosen-token scatter-add into a shared `_add_chosen_token_grad` helper — touching the exact lines this PR renames to `chunk_grad_input` and streams — so the two **will textually collide on this hunk**. NovaSky-AI#1765 does **not** subsume this PR's peak-memory win (it preserves the list+`cat`), and NovaSky-AI#1765 is otherwise complementary in intent. Whichever lands first forces a rebase of the other. **Recommended ordering:** land NovaSky-AI#1765 first, then rebase this streaming change on top of it (the preallocated-buffer write replaces the `append`+`cat` NovaSky-AI#1765 keeps); if the two are reviewed together they can be folded into one change. - **NovaSky-AI#1543** (WIP) independently removes the same `torch.cat` 2x-peak, but does so **in place** into the autograd-saved logits with no extra buffer. This PR deliberately does **not** take that approach: the in-place variant downcasts the fp32 grad to the saved logits' dtype (breaking byte-identity) and mutates an autograd-saved tensor (the version-counter / double-backward hazard). This PR's separate fp32 buffer preserves numeric fidelity and autograd-safety at the cost of ~`(1/num_chunks)`x extra memory. NovaSky-AI#1543 is also built on pre-merge code and currently conflicts with main, so it is not a viable supersession. --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
# What does this PR do? The Tinker E2E CI introduced in NovaSky-AI#1616 has been been failing with the following error: ```bash ValueError: Error retrieving result: Error code: 400 - {'detail': "Failed to merge the Job's runtime env .......} because of a conflict. Specifying the same runtime_env fields or the same environment variable keys is not allowed. Use RAY_OVERRIDE_JOB_RUNTIME_ENV=1 to instruct Ray to combine Job and Driver's runtime environment in the event of a conflict."} (status 400) for self.request_id='1' — exceeded 3 retries ``` The fix is to include `RAY_OVERRIDE_JOB_RUNTIME_ENV` in the job's `env_vars` - we propagate `WANDB_API_KEY` explicitly in `ray.init` and this conflicts with `WANDB_API_KEY` included in the job's runtime env. I also added the env var `RAY_OVERRIDE_JOB_RUNTIME_ENV` in all the CI YAMLs for consistency --------- Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…#1952) ## Summary Megatron async checkpoint finalization creates CUDA tensors for distributed completion collectives. CUDA current-device state belongs to each host thread, so integrations that drain pending saves from a helper thread can silently fall back to device 0 when every GPU is visible to each worker. Restore each worker’s `LOCAL_RANK` before finalization and route both finalization call sites through the same helper. ## Repro Run multiple Megatron workers on one node with all GPUs visible to every worker, then invoke `finalize_pending_saves()` from a new thread. Without this change, each thread may enter the completion all-reduce on device 0, causing NCCL to report duplicate GPUs. ## Testing - `pre-commit run --all-files` - Validated async checkpoint save/load in a multi-rank Megatron deployment with unmasked GPUs; checkpoint finalization completed and training resumed. ## Did this cause any problems? No. Revert this commit to restore previous finalization behavior.
## Summary Fix routed-expert replay (R3) correctness for RL training, and introduce a shared token-metadata layout that later routed-expert and sampler-support work builds on. This fixes a RouterReplay lifecycle bug that has been latent in the R3 implementation since it was added; recent padding work only made it easier to trigger. This PR is the first in a small stack of routed-expert / sampler-support improvements; it is the correctness base the rest build on. ## Results These curves come from an RL run training **GLM-4.7-Flash** (an MoE model) on the easy split of NVIDIA's open [Nemotron-Terminal-Synthetic-Tasks](https://huggingface.co/datasets/nvidia/Nemotron-Terminal-Synthetic-Tasks) dataset — a sandboxed terminal/coding agent environment. Enabling routed-expert replay both improves mean training reward and sharply reduces the gap between rollout (inference) logprobs and the trainer's recomputed logprobs: routes actually taken during sampling are replayed at train time instead of being re-derived, so training and inference stay aligned. Curves are lightly EMA-smoothed over the raw per-step values. | Mean training reward | Rollout vs. training logprob gap | |---|---| |  |  | With replay on, reward pulls ahead over the back half of training while the mean absolute rollout-vs-train logprob difference drops to roughly a third of the baseline (from ~0.023 to ~0.008 by end of training). ## Where the bugs arose ### 1. Forward-only R3 leaked backward replay state into the next schedule The R3 path calls `setup_per_microbatch_replay_forward()` from both forward-only logprob passes and training passes. Megatron's `set_target_indices()` appends every target tensor to `replay_backward_list`, which activation recomputation later consumes in FIFO order. A forward-only schedule has no backward phase, so its FIFO entries were never consumed, and `clear_router_replay()` was not called at schedule boundaries. The next training backward could therefore consume routes from an earlier forward-only microbatch. This is data-dependent: equal aligned token counts let the wrong routes pass silently, while different counts can leave fewer routing-map assignments than Megatron's dropless all-to-all expects and fail with `Split sizes doesn't match total dim 0 size`. **Fix:** scope global RouterReplay state to one Megatron pipeline schedule. Clear once before the schedule and again in `finally` after success or failure. Do not clear between training microbatches, whose backward FIFO intentionally spans the schedule. ### 2. vLLM routes describe a captured prefix, not every training token vLLM records routes for tokens actually executed by the rollout model. The final trajectory can be longer: the last sampled token has no later decode forward, a synthetic EOS may be appended, and multi-turn generation can append observations. Intermediate observations are captured when the next turn replays the prefix; a terminal uncaptured suffix is not. The old tensorization gave Megatron no way to distinguish an uncaptured row from a real route. The PPO loss mask does not solve this: the router and expert-bias accounting do not read it, and loss-masked observations that condition later actions still need their captured routes replayed. **Fix:** keep `rollout_expert_indices` ragged and use its length as the captured-prefix length. Derive `router_padding_mask` after left padding, marking only left/alignment padding and the uncaptured suffix. Carry that mask through `TrainingInput`, replay experiences, microbatch padding, and the Megatron model call. Synthetic padding rows use distinct dummy experts `[0, ..., topk - 1]`. The mask excludes them from expert-bias accounting, while distinct indices preserve Megatron's dropless `tokens * topk` dispatcher invariant without a dispatcher patch. ### 3. RL packing needs the same row layout for routes and masks With microbatch size greater than one, Megatron's RL packing produces layouts such as `[seq0, pad0, seq1, pad1]`, where each input row is one sequence and each row receives TP/CP alignment padding. **Fix:** build one `TokenMetadataLayout` per microbatch and apply it to both routes and `router_padding_mask`. The layout owns sequence lengths, row/segment alignment, and CP front/back placement. Generic construction, alignment, next-token shifting, and packed-output restoration live in `skyrl/utils/token_metadata.py`; RouterReplay-specific installation stays in `replay_utils.py`. ### 4. Expert-bias padding mask mis-broadcasts `TopKRouter._apply_expert_bias` combines a `[tokens, experts]` routing map with a one-dimensional `[tokens]` mask, which does not broadcast over experts. Freezing router parameters does not disable the separate expert-bias token counters, so dummy and uncaptured rows would corrupt bias updates when expert bias is enabled. **Fix:** pass Megatron's `padding_mask` through the model and apply one narrow compatibility shim that reshapes `[tokens]` to `[tokens, 1]` before calling the original method. GPTModel already scatters this mask beside its embedding on the first PP stage; HybridModel scatters only the embedding, so the mask is scattered explicitly. Intermediate PP stages are scattered to match sequence-parallel hidden states. ### 5. Dynamic sampling could detach routes from their trajectory Replacement copied a hard-coded subset of `GeneratorOutput` fields, while filtering rebuilt a different hard-coded subset, so `rollout_expert_indices` could remain from a rejected sample or disappear. **Fix:** slice every per-trajectory generator field generically for replacement and filtering, keeping route metadata, vision fields, trajectory IDs, and future per-row fields owned by the selected sample. ## Deliberately not included - No `rollout_expert_num_captured_tokens` field; `len(rollout_expert_indices[i])` is authoritative. - No sorting or deduplication of captured real-token routes. Only synthetic padding rows are constructed with deterministic distinct experts. - No change to synthetic-EOS loss semantics. ## Testing - `pytest tests/utils/test_token_metadata.py` - `pytest tests/backends/skyrl_train/utils/test_replay_utils.py tests/backends/skyrl_train/test_train_batch.py tests/backends/skyrl_train/test_token_based_batching_utils.py` - `pytest tests/train/dataset/test_preprocess.py tests/train/generators/test_skyrl_gym_generator.py tests/train/test_trainer_utils.py` Regressions cover captured prefixes, loss-masked observations, microbatch > 1 row packing, CP/SP layout transforms, packed next-token restoration, mocked GPTModel/HybridModel mask handling, expert-bias accounting, dynamic replacement/filter ownership, dummy batch padding, and route preprocessing. The GPU/Megatron path is exercised by `tests/backends/skyrl_train/gpu/gpu_ci/megatron/test_router_replay.py`. --------- Co-authored-by: Eric Tang <erictang000@gmail.com> Co-authored-by: lila-sync-bot <lila-sync-bot@users.noreply.github.com>
# What does this PR do? Adds delta weight sync to SkyRL following NovaSky-AI#1903 ## Implementation | Component | Category | Responsibility | | :---- | :---- | :---- | | **DeltaWeightSyncConfig** | Config | User-facing configuration for checkpoint-delta weight sync | | **DeltaTransferStrategy** | Weight Sync | Creates the sender, receiver init payload and vLLM transfer engine | | **DeltaWeightTransferSender** | Trainer | Owns the trainer-side weight send logic | | **DeltaPublisher** | Trainer | Maintains the CPU byte snapshot, computes XOR deltas, compresses payloads and publishes manifests | | **DeltaManifest / DeltaTensorRecord** | Data Model | Version and per-tensor metadata for a published delta | | **RemoteInferenceClient.fetch\_weights** | Control Plane | Runs the pre-pause receiver-side fetch+process phase on every inference worker | | **DeltaWeightTransferEngine** | Inference | vLLM [transfer engine](http://docs.vllm.ai/en/latest/api/vllm/distributed/weight_transfer) for delta weight sync | | **LocalCheckpointStore** | Inference | Maintains the local checkpoint and applies deltas into it | ## Test Plan ```bash # CPU unit tests uv run --isolated --extra dev --extra fsdp pytest tests/backends/skyrl_train/weight_sync/ uv run --isolated --extra dev --extra fsdp pytest tests/backends/skyrl_train/inference_servers/test_remote_inference_client.py # GPU end-to-end uv run --isolated --extra dev --extra fsdp pytest -s -vvv \ tests/backends/skyrl_train/gpu/gpu_ci/test_delta_weight_sync_e2e.py -m "not megatron" uv run --isolated --extra dev --extra megatron pytest -s -vvv \ tests/backends/skyrl_train/gpu/gpu_ci/test_delta_weight_sync_e2e.py -m megatron ``` ### Tests by component All CPU tests live in `tests/backends/skyrl_train/weight_sync/` unless noted otherwise. | Component | Category | Tests | | :---- | :---- | :---- | | **DeltaWeightSyncConfig** | Config | Covered indirectly through `DeltaTransferStrategy` below — `test_delta_create_init_info_requires_sync_dir` (required `sync_dir`) and `test_delta_create_init_info` (`checkpoint_load_format` validation). No direct unit test; see coverage gaps. | | **DeltaTransferStrategy** | Weight Sync | 1.`test_transfer_strategies.py::test_delta_create_init_info`: config → `DeltaInitInfo` field mapping, including `override_existing_receiver` for remote engines <br> 2. `test_transfer_strategies.py::test_delta_create_init_info_requires_sync_dir`: raises `ValueError` when the delta sub-config is absent or `sync_dir` is unset | | **DeltaWeightTransferSender** | Trainer | 1. `test_delta_sender_seed_sync_skips_chunk_iteration`: the first sync treats `base_model_path` as version 0 and skips chunk extraction/publish entirely <br> 2. `test_delta_checkpoint_non_source_rank_drains_without_publishing`: non-source ranks drain the chunk stream to drive the extractor's collectives without publishing | | **DeltaPublisher** | Trainer | 1. `test_delta_checkpoint_payload_stores_xor_patch`: payload is the XOR patch, and `base ^ patch == updated` byte-for-byte <br> 2.`test_delta_checkpoint_publisher_converts_to_base_checkpoint_dtype`: runtime bf16 tensors are normalized back to the base checkpoint's fp32 <br> 3. `test_delta_checkpoint_splits_payload_files_by_size`: `max_file_size_in_gb` starts a new safetensors file <br> `4. test_delta_checkpoint_skips_missing_lm_head_when_checkpoint_ties_embeddings`: tied-embedding models skip `lm_head.weight` instead of failing <br> 5. `test_delta_checkpoint_unchanged_publish_advances_version`: an unchanged publish is not a no-op — it advances the version and writes an empty delta | | **DeltaManifest / DeltaTensorRecord** | Data Model | 1. `test_delta_checkpoint_publisher_converts_to_base_checkpoint_dtype`: asserts `dtype`, `payload_key`, `checksum_algorithm == "xxh3-128"` and `uncompressed_num_bytes` <br> 2. `test_delta_checkpoint_unchanged_publish_advances_version`: asserts empty `tensors` / `payload_files` on a no-change version | | **RemoteInferenceClient.fetch\_weights** | Control Plane | `inference_servers/test_remote_inference_client.py::test_fetch_weights`: fan-out of the pre-pause fetch to every server | | **DeltaWeightTransferEngine** | Inference | No CPU unit test — the engine runs inside a vLLM worker process. Exercised by the GPU e2e test below. | | **LocalCheckpointStore** | Inference | 1. `test_delta_checkpoint_publish_fetch_and_reload_roundtrip`: publish → fetch → reload; changed and unchanged tensors both land correctly, state advances to v1 <br> 2. `test_delta_checkpoint_replays_multiple_versions_for_late_join`: a receiver joining at v0 replays v1 then v2 <br> 3. `test_local_checkpoint_store_fetch_is_single_writer_with_concurrent_ray_actors`: concurrent Ray actors sharing a cache dir serialize on the file lock and download once <br> 4. `test_delta_checkpoint_checksum_failure_marks_write_in_progress`: a corrupt manifest leaves `write_in_progress=True`, and a subsequent valid delta recovers <br> 5. `test_delta_checkpoint_vllm_multi_thread_safetensors_iterator_roundtrip`: `iter_tensors` round-trip under the multi-thread safetensors loader <br> 6. `test_safe_path_name_disambiguates_long_sibling_uris`: per-version cache keys stay distinct when the `sync_dir` URI is long enough to be truncated | | **Cloud transport** | Storage | 1. `test_delta_checkpoint_gcs_cli_publish_fetch_roundtrip`: full publish/fetch round-trip for GCS with mocked transfer <br> 2. `test_delta_checkpoint_s3_cli_publish_fetch_roundtrip`: similar test for S3 | ### Integration Tests `tests/backends/skyrl_train/gpu/gpu_ci/test_delta_weight_sync_e2e.py::test_delta_weight_sync_sparse_update_e2e`, parametrized over `fsdp` and `megatron` (the latter behind the `megatron` marker). Runs `Qwen/Qwen3-0.6B` non-colocated (1 trainer GPU, 1 vLLM engine at TP=1), applies a sparse weight perturbation between syncs to simulate a weight update on the trainer side. ### Other tests - `test_prefix_cache_reset.py`: Tests who resets the inference engines' prefix cache (PolicyWorker or WeightTransferSender) - `test_worker_dispatch.py::TestSaveWeights` : Tests the new dispatcher branch in `save_weights_for_sampler` for delta weight sync. ## E2E runs I've tested Delta weight sync with disk, GCS and S3 based weight transfer with a small model (Qwen 1.5B Instruct) on the GSM8K dataset on 4 GPUs. 1. `gsm8k-qwen1p5b-nccl`: Baseline, NCCL based weight sync 2. `gsm8k-qwen1p5b-delta-disk`: Delta based weight sync via shared disk 3. `gsm8k-qwen1p5b-delta-gcs`: Delta based weight sync via GCS 4. `gsm8k-qwen1p5b-delta-s3`: Delta based weight sync via S3 5. `gsm8k-qwen1p5b-delta-s3-disagg`: Delta based weight sync via S3 in a disaggregated setup: trainer and inference nodes are in separate clusters. <img width="461" height="253" alt="image" src="https://github.com/user-attachments/assets/c7f2594f-c7ba-4dd7-ae77-bc8221af266f" /> ## Performance Here are some performance numbers for delta weight sync for training `Qwen/Qwen3.5-35B-A3B` on the DAPO recipe with delta weight sync via GCS on 2 8xB200 nodes (1 trainer, 1 inference node in a non-colocated setup) : | Metric | Value | | :--- | ---: | | Total `sync_weights` | 57.31 s | | E2E publish (incl. upload) | 32.99 s | | — GCS upload | 6.86 s | | — non-upload publish (all-gather, CPU copy, delta, compression) | 26.13 s | | Compressed delta size | 1.99 GiB | | Uncompressed changed bytes | 64.56 GiB | | `/fetch_weights` | 13.00 s | | `/update_weights` | 8.49 s | There are some known low hanging fruits (ex: fetch and publish are both handled by a single rank right now, weight loading is currently disk -> CPU -> GPU instead of disk -> GPU because of a [vllm](vllm-project/vllm#48644) limitation) --------- Signed-off-by: SumanthRH <sumanthrh99@gmail.com> Signed-off-by: SumanthRH <sumanthrh@anyscale.com>
…age tensors to CPU on the same process before async save with `mcore` (NovaSky-AI#1960) # What does this PR do? Fixes hang with async save introduced in NovaSky-AI#1838 . Issue first noticed on CI ([job link](https://console.anyscale.com/cld_hxkifz7xa22mwicp21nzkds1lw/prj_4b6c498rypyq6g7yhk6vzgjevt/jobs/prodjob_uk2n7lmibfubmd19lizgamutui?job-tab=overview&job-logs-section-tabs=application_logs)) for test `tests/backends/skyrl_train/gpu/gpu_ci/test_save_load_checkpoint.py::test_save_load_checkpoint[megatron_async_dist_ckpt_save]` Currently, async saves with megatron will deadlock after the following failure on CI machines: ```bash �[36m(MegatronPolicyWorkerBase pid=51155)�[0m �[32m2026-07-28 13:31:00.200�[0m | �[33m�[1mWARNING �[0m | �[36mmegatron.core.dist_checkpointing.strategies.torch�[0m:�[36masync_save�[0m:�[36m675�[0m - �[33m�[1mMCore's async save is deprecated and will be removed in the future releases. Please, use NVRx async solution by setting `async_strategy` to `nvrx`.�[0m�[32m [repeated 3x across cluster]�[0m �[36m(MegatronPolicyWorkerBase pid=51155)�[0m PID 52839: Skipping CPU nice (current 15 already <= target 10; lowering requires superuser �[36m(MegatronPolicyWorkerBase pid=51155)�[0m Process SpawnProcess-2: �[36m(MegatronPolicyWorkerBase pid=51155)�[0m Traceback (most recent call last): �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/anaconda3/lib/python3.12/multiprocessing/process.py", line 314, in _bootstrap �[36m(MegatronPolicyWorkerBase pid=51155)�[0m self.run() �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/anaconda3/lib/python3.12/multiprocessing/process.py", line 108, in run �[36m(MegatronPolicyWorkerBase pid=51155)�[0m self._target(*self._args, **self._kwargs) �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/anaconda3/lib/python3.12/contextlib.py", line 81, in inner �[36m(MegatronPolicyWorkerBase pid=51155)�[0m return func(*args, **kwds) �[36m(MegatronPolicyWorkerBase pid=51155)�[0m ^^^^^^^^^^^^^^^^^^^ �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/.cache/uv/builds-v0/.tmpiQCzTt/lib/python3.12/site-packages/megatron/core/dist_checkpointing/strategies/async_utils.py", line 573, in async_loop �[36m(MegatronPolicyWorkerBase pid=51155)�[0m item = queue.get() �[36m(MegatronPolicyWorkerBase pid=51155)�[0m ^^^^^^^^^^^ �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/anaconda3/lib/python3.12/multiprocessing/queues.py", line 122, in get �[36m(MegatronPolicyWorkerBase pid=51155)�[0m return _ForkingPickler.loads(res) �[36m(MegatronPolicyWorkerBase pid=51155)�[0m ^^^^^^^^^^^^^^^^^^^^^^^^^^ �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/.cache/uv/builds-v0/.tmpiQCzTt/lib/python3.12/site-packages/torch/multiprocessing/reductions.py", line 180, in rebuild_cuda_tensor �[36m(MegatronPolicyWorkerBase pid=51155)�[0m storage = storage_cls._new_shared_cuda( �[36m(MegatronPolicyWorkerBase pid=51155)�[0m ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ �[36m(MegatronPolicyWorkerBase pid=51155)�[0m File "/home/ray/.cache/uv/builds-v0/.tmpiQCzTt/lib/python3.12/site-packages/torch/storage.py", line 1464, in _new_shared_cuda �[36m(MegatronPolicyWorkerBase pid=51155)�[0m return torch.UntypedStorage._new_shared_cuda(*args, **kwargs) �[36m(MegatronPolicyWorkerBase pid=51155)�[0m ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ �[36m(MegatronPolicyWorkerBase pid=51155)�[0m RuntimeError: pidfd_getfd: Operation not permitted ``` The error occurs because: 1. We have expandable segments true 2. GPU tensor handles are copied to ckpt worker processes via CUDA IPC 3. This requires the child process to be able to copy fd from parent , which translates to a `pidfd_getfd` 4. The CI container has a restricted `ptrace_scope`, so the operation fails The fix is to run the staging from GPU -> CPU on the parent process itself instead of having the worker process handle it. This is guarded by `async_save_prestage_to_cpu` flag and it is default `False` - prestaging to CPU affects the overall background write times (for a 0.6B model, this increased the write time by 5x) ## Test Plan Ran : ``` uv run --isolated --extra megatron -- pytest -s tests/backends/skyrl_train/gpu/gpu_ci/test_save_load_checkpoint.py::test_save_load_checkpoint[megatron_async_dist_ckpt_save] ``` Test reports the above error message and hangs on `main` . Test passes with the fixes in this PR. --------- Signed-off-by: SumanthRH <sumanthrh99@gmail.com>
…ky-AI#1957) ## What Closes NovaSky-AI#1921. Adds a new **SFT** section to the docs site covering the native `SFTTrainer`, which previously had no documentation outside `examples/train/sft/README.md`. ## Pages - **`sft/overview.mdx`** — quickstart (FSDP + Megatron), GPU placement (`placement.*`, Megatron TP×PP×CP divisibility, FSDP-only Ulysses `sequence_parallel_size`), data formats and tokenization worker pools (`num_workers` vs `dataloader_num_workers`), pretokenized dataset ingestion (NovaSky-AI#1927 row schema, formats, constraints), sequence packing, optimizer/LR schedule (`optimizer_config.*`, incl. Megatron supporting only `constant_with_warmup`), LoRA fine-tuning (`model.lora.*`), checkpointing/resume/HF export, evaluation, VLM SFT, and a key-config table. - **`sft/multi_dataset.mdx`** — weighted multi-source training (`train_datasets` / `train_dataset_weights`, `DataMixingSampler`), constraints, and per-dataset evaluation (`eval/{name}/loss`). - **`sft/custom_sampler.mdx`** — built-in samplers, writing a checkpointable custom sampler (`sampler=custom`, `sampler_class_path`, `sampler_kwargs`), the curriculum-learning example, and multi-dataset `lengths` injection. ## Scope notes - **LoRA is a shared feature, not SFT-specific.** `model.lora` is the common `SkyRLLoraConfig` also used by the RL trainer and Tinker; the SFT LoRA table deliberately lists only the fields that do something in an SFT run (`rank`/`alpha`/`dropout`/`target_modules`/`exclude_modules`/`init_method`), omitting the vLLM adapter-serving fields (`lora_sync_path`, `max_loras`, `max_cpu_loras`) that are irrelevant without an inference engine. - Config keys, defaults, and behaviors were checked against `sft_config.py`, `sft_trainer.py`, and the example scripts. - No code changes — docs only. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: Avigya Basnet <avigyabb@stanford.edu> Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: SumanthRH <sumanthrh@anyscale.com>
…e-parsing (NovaSky-AI#1962) # What does this PR do? Fixes NovaSky-AI#1567 `from_cli_overrides` accepts either a CLI dotlist or a dict. On the dict path values were serialized with f-strings and handed to `OmegaConf.from_cli`, which re-parses each value with YAML scalar rules. `str()` does not round-trip through that parser, so `None` became the string "None" and string values that look like YAML ("null", "true", "1e5", "[a]", "a: b", "") changed type. For `external_server_urls=None` this produced `list("None")` -> ['N','o','n','e'] and a request to the URL `N`, surfacing as `InvalidUrlClientError: N/get_world_size`. Serialize dict values as JSON instead, via a shared `overrides_dict_to_dotlist` helper used by both `SkyRLTrainConfig` and `SFTConfig`. `ensure_ascii=False` also keeps non-ASCII literal as a bonus for users using fancy Wandb run names, etc ## Test Plan 1. `tests/train/test_config.py::TestOverridesDictToDotlist`: Tests for new utility `overrides_dict_to_dotlist` 2. `tests/train/test_config.py::TestCliOverridesFromDict`: Tests for CLI override parsing with dicts and dotlists (`cfg.key=value`) including a regression test for the error reported in NovaSky-AI#1567 ```bash uv run --isolated --extra dev --extra skyrl-train tests/train/test_config.py ``` Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ings (NovaSky-AI#1963) ## What does this PR do? `docs/generate-api-docs.py` emitted **25 griffe warnings** across 21 sites while generating the API reference. All are now resolved — the build is at **0 warnings** with all 18 pages still generated. ```bash uv run --extra dev python docs/generate-api-docs.py 2>&1 | grep -E '^skyrl.*: ' ``` | | Warnings | Pages | |---|---|---| | Before | **25** | 18 | | After | **0** | 18 | ## Changes Annotations and docstring formatting only - `skyrl/backends/backend.py` — annotated `output_path` / `checkpoint_path` as `AnyPath` on the three abstract checkpoint methods, matching the concrete `jax.py` / `ray_jax.py` overrides and the `AnyPath`-derived paths `TinkerEngine` passes in. - `skyrl/train/entrypoints/main_base.py` — return annotations on `get_train_dataset`, `get_eval_dataset`, `get_generator`, `get_trainer`, `get_tracker` (+ private `_get_new_inference_client`, `_setup_trainer`); dropped redundant `Returns:` prefixes. `get_eval_dataset` is `Optional[PromptDataset]` since it returns `None` when eval is disabled. - `skyrl/backends/skyrl_train/workers/worker.py` — `bool` on the offload/backload flags (both `Worker` and `PPORayActorGroup`), `*args: Any, **kwargs: Any` on `async_run_ray_method`, fixed the unindented `nonblocking` continuation lines, and documented the previously undocumented offload/backload flags. - `skyrl/backends/skyrl_train/distributed/dispatch.py` — `**kwargs: Any` on `dispatch_from_staged`. - `skyrl/backends/skyrl_train/training_batch.py` — `-> "TensorBatch[DictType]"` on `repeat` / `repeat_interleave`, consistent with the already-annotated `select` / `slice` / `cat`. - `docs/.gitignore` — unrelated hygiene fix: no rule existed for the `content/docs/api-ref/skyrl/sft/` directory added by NovaSky-AI#1957, so its three generated pages appeared as untracked files after every build. ## Verification - `generate-api-docs.py` → 0 warnings, 18/18 pages, no `Could not load` / `Error rendering` markers in the generated MDX. - `cd docs && npm install && npm run build` → succeeds, 85 static pages, no warnings or errors. - `pre-commit run --all-files` → ruff, black, secret scan pass. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
…ce the single source of truth (NovaSky-AI#1966) ## Summary `docs/content/docs/configuration/config.mdx` duplicated the config dataclasses by hand and had drifted from them. Since there is no longer any YAML defaults file, the dataclasses in `skyrl/train/config/config.py` are already the only real source of truth, and `/docs/api-ref/skyrl/config` is generated from them. This removes the page and folds its content into the place that cannot drift. ### Drift the page had accumulated | Page said | Code says | |---|---| | `enforce_eager` defaults to `true` | `false` | | `fully_async.clear_kv_cache_on_weight_sync` defaults to `true` | `false` | | eval dumps to `dumped_eval/` | `dumped_evals/` | | `ppo_policy_loss` snippet | predated `safe_exp_delta`, off-policy correction, current `reduce_loss` signature | | — | missing `use_cache_salt`, `max_tokens_per_microbatch`, `mtp`, … | ## Changes in this PR **Detailed documentation in config.mdx moved Into docstrings** (115 fields newly documented, 34 enriched) — every parameter the page documented now has an attribute docstring, with pitfalls sitting next to the field they constrain: - FSDP `cpu_offload` vs. colocation offload - `use_precision_aware_optimizer` checkpointing bug - shared-filesystem requirement for distributed HF export - dataset in-memory limit - torch-profiler scope + FSDP restriction - reference model only instantiated when KL is used - `expandable_segments` fragmentation rationale - `policy_loss_type` / `loss_reduction` per-option descriptions and paper links **Documentation moved onto topic pages**: | Content | New home | |---|---| | Parallelism sizing rules, optimizer dtype aliases, `use_precision_aware_optimizer` warning | `examples/megatron` | | `clear_kv_cache_on_weight_sync`, simulated-trainer knobs | `tutorials/fully_async` | | CPU offloading vs. colocation offloading | `configuration/placement` | | Hard `Aborted`/SIGABRT from allocator fragmentation | `troubleshooting` | | Distributed HuggingFace export | `checkpointing` | | How the built-in PPO loss composes the config | `algorithms/custom_algorithms` | The loss formulation is **described rather than pasted**, so it cannot go stale the way the original snippet did. ## Additional changes - Added `DataLoaderConfig`, `MegatronHFExportConfig`, `DPPOConfig`, `DeltaWeightSyncConfig` to `api-pages.yaml` so the generated reference is complete. - Redirect `/docs/configuration/config` → `/docs/api-ref/skyrl/config` in `next.config.mjs`, keeping the published URL working. - Recorded the convention in `.claude/docs/training.md` (document new fields in the dataclass; griffe reads attribute docstrings but **not** `field(metadata={"help": ...})`; keep the first docstring line a complete sentence since the generated summary table truncates at the newline). - `placement.mdx` is moved to Tutorials now ## Verification - **All 333 config fields unchanged** in name, annotation, and default vs. `origin/main` — mechanically confirmed via AST diff, so this is a docstring-only change to `config.py`. - Coverage audit: all 187 config keys the deleted page documented are accounted for (179 via attribute docstrings; the other 8 are `policy_loss_type`/`loss_reduction` option *values*, all documented inline). - `ruff` / `black` / secret detection pass. - API docs regenerate: 18 pages, no errors; the 4 newly added classes render. - 139 config tests pass (`tests/train/test_config.py`, `step_wise/test_config.py`, `test_sft_config.py`). 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Signed-off-by: SumanthRH <sumanthrh@anyscale.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
There was a problem hiding this comment.
Code Review
This pull request introduces support for hosted Fireworks Supervised Fine-Tuning (SFT) training, allowing SkyRL to run its native SFT loop on a dedicated Fireworks trainer without requiring local GPUs or Ray workers. It adds a new direct entrypoint (main_fireworks_sft.py), SFT batch translation to Fireworks cross-entropy requests, configuration validation, and cleanup utilities. Feedback on the changes suggests optimizing the contiguous left padding check in the SFT batch processing to avoid multiple list allocations.
| present = [bool(value) for value in attention_mask.tolist()] | ||
| count = sum(present) | ||
| if present != [False] * (len(present) - count) + [True] * count: | ||
| raise ValueError(f"attention_mask[{row_index}] must describe contiguous left padding") | ||
| return [int(token) for token in sequences[attention_mask.bool()].tolist()] |
There was a problem hiding this comment.
The check for contiguous left padding can be made more efficient. The current implementation creates a new list from attention_mask, then iterates it with sum(), then creates two more lists and concatenates them for the comparison. This can be slow for long sequences in a performance-sensitive data loading path.
A more efficient approach is to iterate through the boolean mask once to validate its structure, avoiding multiple list allocations.
| present = [bool(value) for value in attention_mask.tolist()] | |
| count = sum(present) | |
| if present != [False] * (len(present) - count) + [True] * count: | |
| raise ValueError(f"attention_mask[{row_index}] must describe contiguous left padding") | |
| return [int(token) for token in sequences[attention_mask.bool()].tolist()] | |
| present = [bool(value) for value in attention_mask.tolist()] | |
| # Check for contiguous left padding efficiently. | |
| # The mask should be of the form [False, ..., False, True, ..., True]. | |
| # It can only transition from False to True once. | |
| seen_true = False | |
| is_valid = True | |
| for is_present in present: | |
| if seen_true and not is_present: | |
| is_valid = False | |
| break | |
| if is_present: | |
| seen_true = True | |
| if not is_valid: | |
| raise ValueError(f"attention_mask[{row_index}] must describe contiguous left padding") | |
| return [int(token) for token in sequences[attention_mask.bool()].tolist()] |
There was a problem hiding this comment.
Fixed in 284eb02. The converter now checks for a True-to-False tensor transition and reuses the boolean tensor for token extraction; the existing non-contiguous-padding regression test still passes.
Avoid allocating multiple Python lists while checking contiguous left padding in Fireworks SFT batch conversion. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
|
CI note: the current failures occur before this PRs feature tests run:
Focused Fireworks/SFT tests, the existing SFT regression subset, formatting on every changed Python file, and the docs production build pass locally; exact commands and live validation are recorded in the PR body. |
Avoid loading image and video processors when a multimodal-capable model is used with text-only Fireworks SFT data. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
| self.tokenizer = get_tokenizer(self.cfg.trainer.policy.model.path, **tokenizer_kwargs) | ||
| self.collator = self._build_collator(self.tokenizer) | ||
| self._init_tracker() | ||
| self._init_workers() |
There was a problem hiding this comment.
VLM models no longer fail closed
Medium Severity
The new setup() path sets is_vlm to False instead of detecting vision models, so the existing _init_workers rejection never runs. A VLM model.path now proceeds as text-only SFT, which can tokenize multimodal data without a processor and send incorrect sequences to a paid Fireworks trainer.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 2b34fb9. Configure here.
Reuse the native SFT checkpoint lifecycle and the existing Fireworks DCP dispatch so hosted runs can save provider, trainer, and dataloader state. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
Use cross-job references only when the checkpoint came from a different trainer namespace. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
Require checkpointed runs to keep the trainer namespace reconnectable and make smoke cleanup behavior explicit. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.
There are 2 total unresolved issues (including 1 from previous review).
Reviewed by Cursor Bugbot for commit 9dd3ab5. Configure here.
| " checkpoint path: ${CKPT_PATH:-disabled}" \ | ||
| " checkpoint interval: ${CKPT_INTERVAL}" \ | ||
| " resume from: ${RESUME_FROM:-fresh}" \ | ||
| " preserve trainer for resume: ${PRESERVE_TRAINER_FOR_RESUME}" \ |
There was a problem hiding this comment.
Checkpoint env vars skip trainer preserve
Medium Severity
The smoke script only forces CKPT_PATH when PRESERVE_TRAINER_FOR_RESUME=1, but cleanup_on_exit stays true whenever preserve is 0. Setting CKPT_PATH, CKPT_INTERVAL, or RESUME_FROM without preserve still prints a paid-run plan, then validate_fireworks_sft_cfg rejects the run because checkpointing now requires cleanup_on_exit=false. The confirmation text also still says the trainer will be deleted even when preserve is enabled.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 9dd3ab5. Configure here.
Save a separate base sampler artifact, promote its exact control-plane row, persist export evidence, and delete the trainer only after promotion succeeds. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
Expose the promoted model and trainer cleanup manifest to downstream lifecycle tracking. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> Signed-off-by: Bharat Mekala <155010017+bharatmekala@users.noreply.github.com>
…2040) Upgrades SkyRL to use cuda 13 wheels (upgraded from cuda 12.8) for all dependencies. Updates docker images (`novaskyai/skyrl-train-ray-2.57.0-py3.12-cu13.0-megatron` and `novaskyai/skyrl-train-ray-2.57.0-py3.12-cu13.0`), and new h100 ci image on anyscale (`anyscale/image/skyrl-train-ray-2.57.0-py312-cu13.0-megatron-efa-1.47:1`). upgrades ray to 2.57.0, so new anyscale base image is `anyscale/ray:2.57.0-py312-cu130` since there isn't a cuda 13 slim image. This increases the size of the image from 3.2 -> 8 GB, but then we remove the separate nvcc install from the images so it nets out to be slightly lower than before (10.7 now, vs 11.8 before for the megatron image) Prereqs: - [x] Upgrade ray to 2.57 which is the only ray version with cuda 13 image --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> Co-authored-by: SumanthRH <sumanthrh99@gmail.com>
…n setup failure (NovaSky-AI#2073) ^ also adds cache to skyrl cpu test
## Summary - Backport vllm-project/vllm#41602 for the pinned vLLM 0.26.0 release. - Handle hybrid-model KV-cache entries that contain a list of tensors. - Apply the patch only when the vLLM method matches the affected source. A future vLLM update can remove this patch. ## Before and after | Check | Before | After | | --- | --- | --- | | Qwen3.6-27B B300 lifecycle | KV-cache wake failed with `AttributeError: list object has no attribute zero_` | Four engines woke the KV cache in 1.21-1.25 s. The full create, sample, and unload check passed. | | Qwen3.8-27B B300 lifecycle | Same KV-cache wake failure | Four engines woke the KV cache in 1.12-1.15 s. The full create, sample, and unload check passed. | ## Testing `uv run ruff check skyrl/backends/skyrl_train/inference_servers/new_inference_worker_wrap.py skyrl/backends/skyrl_train/patches/vllm/patch_hybrid_fp8_kv_wake.py` `git diff --check upstream/main...HEAD` Both checks passed. The patched image also passed the two B300 lifecycle checks above. <!-- CURSOR_SUMMARY --> --- > [!NOTE] > **Medium Risk** > Runtime `exec` patching of vLLM's `init_fp8_kv_scales` affects inference worker lifecycle and KV-cache reinit; scope is narrow and gated on source matching, but wrong patch application could break wake/sleep for FP8 hybrid models. > > **Overview** > Adds a **vLLM 0.26.0 backport** of upstream **vllm#41602** so `wake_up(tags=["kv_cache"])` can zero FP8 KV-cache state for **hybrid models** (e.g. Qwen3.6/3.8) whose cache entries are **lists of tensors**, not single tensors—avoiding `AttributeError: 'list' object has no attribute 'zero_'`. > > The patch replaces the loop in `GPUModelRunner.init_fp8_kv_scales` only when the pinned vLLM source still matches the old 0.26.0 shape; otherwise it skips and logs. **`patch_hybrid_fp8_kv_wake()`** runs at import in **`new_inference_worker_wrap`** so inference workers get the fix before serving. > > <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 304ef55. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup> <!-- /CURSOR_SUMMARY --> --------- Co-authored-by: Dian Ang <23232359+yapdianang@users.noreply.github.com> Co-authored-by: avigyabb <98926738+avigyabb@users.noreply.github.com> Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…I#2056) ## Summary Megatron Core derives PackedSeqParams.seq_idx only when total_tokens is set. Mamba uses those labels to reset recurrent state between packed documents; without them, state can flow across document boundaries in a packed THD row. Pass the global padded token count when constructing PackedSeqParams. Using the global count keeps the labels aligned with cu_seqlens_q_padded under context parallelism. ## Testing - uv run --no-sync pytest -q tests/backends/skyrl_train/distributed/test_preprocess_packed_seqs_multiseq.py::TestSubSeqLengths::test_multiseq_row_emits_padded_cu_seqlens_entries tests/backends/skyrl_train/distributed/test_preprocess_packed_seqs_multiseq.py::TestMultiSeqCPLayout::test_roundtrip_recovers_full_layout_cp2 <!-- CURSOR_SUMMARY --> --- > [!NOTE] > **Medium Risk** > Fixes correctness of Mamba state in packed training (wrong labels can silently corrupt gradients); change is small and localized to packed-seq metadata construction. > > **Overview** > **Sets `total_tokens` on `PackedSeqParams`** when building packed THD inputs in `preprocess_packed_seqs`, using the global padded token count (`cu_seqlens_padded_cpu[-1]`). Megatron Core only derives per-token `seq_idx` when `total_tokens` is present; Mamba layers use those labels to reset recurrent state at document boundaries. > > Without this, **recurrent state can carry across packed sub-sequences** in the same THD row. Using the **global** padded total keeps `seq_idx` aligned with `cu_seqlens_q_padded` under context parallelism. > > Tests extend megatron stubs and assert `total_tokens` matches the padded cumulative length in multi-subseq and CP round-trip cases. > > <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 6558f94. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup> <!-- /CURSOR_SUMMARY -->
Let hosted SFT runs retain inference-formatted checkpoints at the same cadence as resumable DCP state, so selected intermediate steps can be promoted after training without reconnecting to the trainer. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Resolve dependency drift against current SkyRL and keep the hosted Fireworks checkpoint behavior covered by the updated SFT interfaces. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Keep the Fireworks branch compatible with the repository-wide Black check. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Replace the machine-local Harbor checkout with the pinned upstream source so isolated CI environments can resolve the Fireworks branch. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Record the exact resumable control-plane row returned after save so deleted-trainer resumes use a verified cross-job checkpoint instead of a possibly renamed request. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Capture the pre-save control-plane baseline and serialize save plus resolution so cross-job manifests cannot bind to an unrelated checkpoint. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
|
Paid deleted-trainer DCP resume validation passed on Qwen3-4B LoRA:
No deployment was created. |
Fireworks' forward-only call returns no loss metric, so every SFT eval batch fell back to NaN. Compute the loss client-side as the weight-weighted negative logprob sum over the datums we sent, which reproduces the loss:sum that forward_backward reports for identical datums. Any unexpected output shape omits the metric rather than reporting a guessed number.
Adds train_on_what=last_n_assistant_messages with train_on_last_n, for datasets where only the most recent turns of a long conversation should carry loss. Token ids are unchanged from all_assistant_messages: the mask returned by the turn-by-turn encoder is restricted to its last N runs of 1s, and the action window starts at the first surviving supervised token so no logprob compute is spent on unsupervised turns. train_on_last_n is part of the tokenization cache key, and rows whose selected turns fall outside max_length are counted and logged.
[fix][SFT] Compute Fireworks eval loss from per-token logprobs
[feat][SFT] Train on the last N assistant messages


What does this PR do?
Adds native supervised fine-tuning on dedicated Fireworks trainers while reusing the existing
SFTTrainerloop.Checkpoint contract
resumable=Truecorresponds to DCP weights and optimizer state.promotable=Truecorresponds to an inference-formatted sampler checkpoint.save_promotable_checkpoints=falsepreserves periodic DCP-only behavior.Validation
CI note
The
fireworksbase branch has repository-wide latest-Ruff drift in the separate SkyRL-CPU workflow, and fork GPU jobs lack Anyscale credentials. The scoped code-quality job passes and the hosted Fireworks suite passes locally.Generated with Devin