[Core] Add step-gated record drain and build rings on use

[XbzOnGit]

Summary

Add an opt-in drain policy for record rings in which the integration, not a timer or ratio threshold, decides when records are flushed: once per model step, after the step's producers. Also make MonitoringEngine build a ring only on first use, so a recording engine no longer builds a legacy ring just to replace it.

Configuration

ring = dmi.RingConfig()
ring.drain_policy = "step_gated"   # default: "autonomous"

engine = dmi.MonitoringEngine(model_id="my-run", ring_config=ring)
runtime = engine.create_record_runtime(record_format, record_sink=sink)

# once per model step
runtime.begin_step()
...                                 # emit the step's records
runtime.allow_flush_step()          # after the step's producers are enqueued

allow_flush_step() grants the drain every ring record reserved since the previous call; it records an event on the current stream and returns without waiting. A step with no ring records records no event (including a step whose only records were oversized and went through the CPU-direct path). Calling begin_step() while the previous step still has ungranted ring records raises. step_gated is for record rings only: a legacy ring refuses it; mirror-only rings have no drain.

Implementation details

allow_flush_step(): K = ring records since the previous call; record event on the current stream; queue (K, event)
  -> drain thread: sleep on its CV (grants, forced flushes, stop)
  -> pop (K, event); wait on the event (blocking sync); allowed += K
  -> flush whole published records while record tail < allowed
  -> existing D2H / reconstruction / P2P / sink path

• In step-gated mode the drain never wakes on a timer; forced-flush handling remains first in the loop.
• Grants are credits, not positions: a forced flush (reservation that does not fit, flush_and_wait(), shutdown) may deliver granted records early, and later grants then flush only what is still below allowed. No duplicates, no lost records.
• A granted record that is not published after its event completed fails the drain (protocol error), instead of waiting.
• Lazy rings: the constructor only keeps the ring config. The first legacy entry point (adaptor attach, generation with monitoring, ring_capacities(), capture_enabled, set_capture_enabled()) builds the legacy ring; create_record_runtime() builds the record ring directly. enable_ring_transport() still builds at once; close() and a failed record-ring creation leave no ring configured, as before.

Compatibility and scope

"autonomous" (the default) keeps the existing drain behavior. The only observable change for existing users is ring construction timing: code reading engine._ring_transport / engine._ring_engine immediately after construction, before any entry point, now sees None. One test helper (test_eager_safety_net_gpu.py) is updated accordingly; the HF adapter, BackendAdaptor subclasses and the released vLLM integration go through the entry points.

Validation

• Native ring suite: 941 assertions passed, 0 failed (test_rings, test_ring_engine, test_null_mode, test_producer, test_record_consumer).
• CPU suite: 2202 passed (12 modules fail to collect without botocore, as on main).
• GPU suite: 174 passed; the 2 test_e2e_correctness_vs_hf failures also occur on main.
• New test_record_step_gated_drain_gpu.py: 12 passed — no flush without a grant, per-step grants, event wait without host synchronization, ordered grants, empty steps, the ungranted-records error, forced flush without duplicates, oversized records in order, policy validation, ring construction on use for both policies.

Performance

Engine plus record-runtime setup, 1 GiB payload ring and 1 GiB pinned staging, sampled outside the process:

	Peak GPU	Peak host RSS	Setup
Before	2653 MiB	2652 MiB	1114 ms
After, autonomous	1627 MiB	1637 MiB	546 ms
After, step-gated	1627 MiB	1638 MiB	495 ms

Step-gated: with ~256 ms of GPU work queued ahead of the producers, allow_flush_step() returns in ~14 µs and records reach the sink only after the GPU passes the step's event.

Setup: one RTX 4090, PyTorch 2.13.0+cu130, CUDA 13.0, Python 3.12.

[Samfisheryu]

I reproduced two regressions below that should be fixed before merging. Validation passed: 363 targeted Python tests, 26 GPU tests, 1,650 records through wraparound/forced flushes, and 100 CUDA Graph replays. Storage-extension tests were partly skipped; this was not a full storage-chain check.

[Samfisheryu]

Rechecked a1e6240 using the original repros. Both issues are fixed: the lazy legacy ring stays on its construction device, and disabling capture no longer drops a pending step's grant. The next begin_step succeeds and the record is delivered once.

363 targeted Python tests and 28 GPU tests passed, including both new regressions on two GPUs. Four storage-extension tests were skipped. No new blocking issues found in the fix. CI is still running.