[skycap] Upload each step's skycap documents to W&B

[jlee-lila]

Adds SkycapUploads, a trainer callback that uploads skycap documents to W&B when trainer.logger=wandb. It fetches each document from its skycap server and uploads on a background thread; on_train_end waits for it.

Train records upload once per step to skycap-records-train-<run id>, aliased step-N and latest. skycap.wandb.phases adds eval, which goes to its own skycap-records-eval-<run id> artifact. skycap.wandb.enabled=false turns uploads off.

Each version holds a step.json that marks every attempt trained or dropped (retries, dynamic sampling). The metadata counts uploaded and missing documents. For this, CallbackInput gains trajectory_ids at on_step_end.

The fully-async trainer fires no callbacks, so it uploads nothing. run_codecontests_modal.sh also takes the Modal credentials from the environment, and its default run names carry a UTC timestamp.

Tested: tests/integrations/harbor_skycap and tests/train/test_rl_callbacks.py pass on 8xB200 (21 passed).

🤖 Generated with Claude Code

[kouroshHakha]

Using a W&B artifact for this is the right tool, and it isn't slow: the heavy upload happens in wandb's background service, the call runs off the event loop, and errors are caught, so a W&B outage can't fail rollouts. The main issue is that it uploads once per generate() call, and that isn't once per training step. Three things to fix before merging:

1. Eval takes the step-N and latest aliases. Eval calls the same generate() with the same global_step, once per eval batch, right after training step N and at step 0. An alias belongs to one version at a time, so the eval upload moves step-N off the training records. training_phase is in batch_metadata and isn't checked.
2. Several versions per step. Dynamic sampling calls generate() repeatedly within one step, and each call moves step-N to itself, so the earlier versions are reachable only as vK.
3. Silent partial uploads across nodes. Documents that aren't on the trainer's disk are skipped quietly. With placement_strategy="SPREAD" and the node-local default record_dir, that's the likely case on multi-node, and num_trajectories still counts every id.

What I'd do:

• Upload once per training step, train phase only. Accumulate ids per global_step, upload when the step changes and at shutdown, and send eval to its own skycap-records-eval-<run> artifact or leave it off by default.
• Fetch documents from the server that owns them (GET /trajectories/{id} already reads from disk) rather than from the trainer's filesystem. The pool knows which server holds each id, so the "one node or a shared record_dir" constraint goes away.
• Don't await it on the step. Run it as a background task, optionally with sampling. Hashing 256 long documents costs roughly 10-100 ms per step at 32x8, which isn't big, but the step doesn't need to wait for it.

Smaller notes are inline. Size looks fine: roughly 10-80 MB per step at 32x8 with 32k-token trajectories, and memory stays flat since it streams files. A wandb.Table would have been much worse.

[kouroshHakha]

leaving some comments.

[kouroshHakha]

nit style: comments are not added

[kouroshHakha]

A thought on where this should live, before more iteration on the upload path. I think it splits cleanly in two, and neither half belongs in the Harbor example.

1. Getting records to durable storage is skycap's job. skycap already writes each trajectory once, when it ends, off the event loop. The semantics we want for the upload (async, fail open, bounded queue, timeouts, a deadline at shutdown) are exactly a record writer's. A remote store is just another destination for the same records. It also removes the multi-node problem at the source: today records land on each server's node-local disk, so this PR has to fetch them back over HTTP through the trainer and re-upload the bytes. If skycap writes to shared or remote storage (e.g. record_dir accepting a URL like s3://… or gs://…), there's nothing to fetch. Every skycap user gets it, and the viewer can read the same place. skycap stays free of SkyRL and W&B dependencies.

2. What goes to W&B is the trainer's job. "The trajectories of training step N, their phase, which were trained and which superseded" is SkyRL knowledge. A callback can log that per step through Tracking as a small index artifact: ids, phase, trained/superseded, and where each record lives. With remote storage those can be W&B reference entries, so no bytes are copied, and failing open is trivial. Any skycap-based generator wants this, so it belongs next to the skycap generator code rather than inside Harbor specifics.

With that split, most of this PR's fetch/re-upload/retry/timeout machinery goes away. I'm opening the two pieces as separate PRs and will link them here. Happy to fold anything from this PR's design into them (the SkycapWandbConfig shape, the phase handling, the eval opt-in).

[kouroshHakha]

Following up on the comment above, the two pieces are up:

• [skycap] Mirror records to remote storage, and say where each is at finish #2350, skycap: records are still written locally first, then mirrored to remote storage (--record-mirror s3://…, via fsspec as an optional skycap[remote] extra). The mirror is async and fails open, with a bounded queue, a per-copy timeout, bounded retries and a shutdown deadline. finish() now says where each record is.
• [skycap] Index each step's skycap records in W&B #2351, stacked on [skycap] Mirror records to remote storage, and say where each is at finish #2350: a TrainingCallback logs one W&B index artifact per step through Tracking: ids, phase, trained/superseded, and a reference entry per mirrored record. There's no fetching over HTTP and no re-uploading bytes. It reuses this PR's SkycapWandbConfig{enabled, phases}, per-phase log and eval opt-in, credited in the PR.

Together they replace this PR's upload path. Reviews and pushback welcome there.