feat: store by-handle sampling masks as CSR

[faresobeid]

Stacked on #3904.

What

By-handle sampling masks (and score-centering mask logprobs, when they come back) were written as [tokens, widest row] arrays padded to each request's widest row. At top_p 0.97 a sampled token keeps 1.7 ids on average but requests run ~32 wide, so a mask cost ~128 B per completion token (~35% of payload bytes on math, plus the same again for mask logprobs) against ~11 B inline. This stores them as CSR instead.

• PayloadField.ragged marks fields with rows of varying width (sampling_mask, sampling_mask_logprobs).
• Layout (ragged_bytes): rows + 1 uint32 file offsets where each row's values start (the last one is where the last row ends), then the values (int32 for an int fill, float32 for a float fill). The segment's rows are the offsets (dtype="uint32", shape=[]), so the orchestrator and verifiers clip segments exactly as before, with no knowledge of row widths and no PayloadSegment change.
• Writer: _PackedOutputs passes (counts, ids) straight from perf(inference): pack /generate per-token payloads and tune API-server GC #3900's PackedSamplingMask; _write_payload lays ragged fields out with ragged_bytes. Score centering would add ("sampling_mask_logprobs", prompt_len, (mask.counts, logprobs)).
• Reader: for a ragged field each trainer rank reads the offsets of all of the micro batch's segments (4 B per completion token) to get the widest row, then only its window's values. The width is the micro batch's widest row, as inline (before, it was the widest request's), so every CP rank gets the same width. A cut-out mask's zero-row placeholder still yields a width-1 fill tensor.
• csr_rows is gone; the fixed-width and ragged paths share one concurrent _pread over (file, offset, size) reads with the short-read retry.

Size: ~4 B per completion token + 4 B per kept id (~11 B at 1.7 ids) instead of ~128 B.

Tests

• test_payload_by_handle_matches_inline now writes masks and mask logprobs with ragged_bytes, one CSR region per sample split into two segments mid-request (a clipped head), and still checks bitwise equality with inline through truncation, packing, padding, 2 DP ranks and CP 1/2.
• test_inline_and_by_handle_samples_keep_their_masks, test_payload_dir_writes_routing_and_masks_by_handle (reads the written file back through read_field), test_pread_retries_short_reads.
• pytest tests/unit/orchestrator tests/unit/test_configs.py tests/unit/inference tests/unit/train/rl -m "not gpu" on vLLM 0.31.0 with feat: write router-replay ids and sampling masks by handle on multi-node runs #3903's pinned verifiers/renderers on the path, ruff check, ruff format --check: pass.
• GPU (1 node, vLLM 0.31.0, Qwen3-30B-A3B TP1, VLLM_BATCH_INVARIANT=1, router replay + sampling masks, top_p 0.95 / top_k 50):
  • 16 requests (500-8000 prompt tokens, 256 new, concurrency 8), once inline and once with payload_dir; tokens equal, response 33.4 MB → 0.06 MB.
  • Each written CSR mask read back with read_field equals the inline (ids, counts), and each routing segment equals the inline routing.
  • The same rows pushed through prepare_batch + DataLoader._micro_batch_to_tensor (seq_len 16384/4096 with truncation, 2 DP, CP 1/2/4; every other sample's segments split at a random position) give 140/140 micro batches whose routing and masks are bitwise equal to inline (masks at the micro batch's width, no trimming).
  • At top_k 50 a completion token kept 3.7 ids on average; the CSR mask took 18.8 B per completion token.
  • Mask logprobs are not written on this stack (score centering is not in it), so they were not checked on GPU; the unit test covers them.