[DataFlow runtime] Two-axis staleness gate for online disaggregated training

[maocheng23]

Stacks on the online-disagg PR (dataflow-up-17-online-disagg) → zero-copy (#614) → #612. Merge bottom-up. Third PR of the series. No hot-switch.

Why

Online training keeps producing fresh draft weights while the rollout pool keeps generating samples. Without hot-swapping the rollout (deliberately out of scope), the pool drifts behind and its samples go stale — training on stale features hurts. So the consumer must drop stale refs before they reach the trainer.

What

The staleness half of the weight lifecycle (no hot-update, no rollback, no serving accept-length gate):

• contracts.WeightVersion (+ WeightStatus) — metadata-only published draft version; publish order is the staleness axis.
• control_plane/version_policy.py
  • WeightRegistry — publish-ordered history; draft_lag(sample) = distance from the newest published version (None = unknown → maximally stale). Optional durable metadata_store backing.
  • StalenessPolicy — two axes: draft (max_draft_lag) + target (require_target_match) → accept/reasons assessment.
  • DriftMonitor — rolling mean/max draft lag; drifting() for the orchestrator (pause / force-sync / alarm).
  • StalenessGatedQueue — wraps a ref queue so the loader only ever sees fresh refs; stale ones are acked (backpressure clears) and their features aborted (freed) instead of trained on.
• build_disagg_online_consumer gains optional staleness_policy / weight_registry / current_target_version / drift_monitor. When set, the streamed queue is gated so a drifted rollout can't poison training. Off by default (every streamed ref is trained, unchanged).

Tests

11 CPU tests: registry lag ordering + idempotent publish; both policy axes; unknown-version-is-maximally-stale; drift snapshot/threshold; and the gate (fresh passes, draft-stale + target-stale dropped → features aborted + acked).

Not in scope (deliberate)

Hot-update / weight hot-swap, rollback, and the serving accept-length gate. This PR only decides which streamed samples are fresh enough to train on.

🤖 Generated with Claude Code

[gemini-code-assist]

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

[maocheng23]

Closing — recreated cleanly on the rebased online stack (up-17/up-18 rebased onto the current zero-copy #621). Superseded by the new up-18 staleness PR below.