[Core] Add recurring D2H window scheduling

docs/config.md

@@ -77,6 +77,52 @@ By default, timeout-based flushing is disabled and the drain thread flushes at
 are explicitly set to 0, the drain thread only flushes when the ring is 100%
 full or at `stop()` time.
 
+## Recurring D2H Windows (`recurring_d2h_windows`)
+
+This grouped configuration is disabled by default and is supported only by the
+HookPointV1 encoded-record path. Construct `MonitoringEngine` with
+`record_mode_v1=True`, then create a `RecordRuntime` before defining a pattern.
+
+| Parameter | Type | Default | Description |
+|-----------|------|---------|-------------|
+| `enabled` | `bool` | false | Enable recurring-window scheduling. |
+| `progress` | native enum | packed version/counter | Fixed progress backend for this version; leave unchanged. |
+| `grant_policy` | native enum | binary adaptive | Fixed byte-grant policy for this version; leave unchanged. |
+| `minimum_record_probe_retry_interval_occurrences` | `uint64_t` | 0 | Window occurrences between minimum-record probes; must be positive when enabled. |
+| `timing_revalidation_retry_interval_occurrences` | `uint64_t` | 4 | Base window occurrences between failed timing-revalidation probes; must be positive when enabled. |
+| `capacity_flush_fallback_threshold` | `uint64_t` | 0 | Completed capacity-forced flushes before permanent batched fallback; must be positive when enabled. |
+| `capacity_flush_count_reset_interval_periods` | `uint64_t` | 32 | Complete pattern periods of progress after which isolated capacity-forced-flush evidence ages out; must be positive when enabled. |
+| `debug_enabled` | `bool` | false | Log every real window-scheduled D2H issue and completion, plus each pattern-version activation. |
+
+The integration defines sorted, non-overlapping half-open window offsets
+within a positive period, then calls `RecordRuntime.advance_boundary()` at each
+ordered framework boundary. Both operations must use the same framework CUDA
+stream, and the pattern must be defined before the first marker or CUDA Graph
+capture.
+
+```python
+from dmi.api import v1 as dmi
+
+windows = dmi.RecurringD2HWindowConfig()
+windows.enabled = True
+windows.minimum_record_probe_retry_interval_occurrences = 4
+windows.timing_revalidation_retry_interval_occurrences = 4
+windows.capacity_flush_fallback_threshold = 3
+
+cfg = dmi.RingConfig()
+cfg.recurring_d2h_windows = windows
+
+engine = dmi.MonitoringEngine(
+    model_id="my-run",
+    host_engine=host,
+    ring_config=cfg,
+    record_mode_v1=True,
+)
+runtime = engine.create_record_runtime(record_format)
+runtime.define_d2h_window_pattern(period=100, windows=[(20, 30), (70, 80)])
+runtime.advance_boundary()
+```
+
 ## P2P Thread / Output
 
 | Parameter | Type | Default | Description |

docs/integration-api-v1.md

@@ -167,6 +167,7 @@ MonitoringEngine(
     host_engine=None,
     db_config=None,
     enable_ring_transport: bool = True,
+    record_mode_v1: bool = False,
     ring_config=None,
     ring_payload_mb: int = 4096,
     ring_pinned_mb: int = 4096,
@@ -189,6 +190,9 @@ accessing those objects.
 - `host_engine` and `db_config` are mutually exclusive.
 - Supplying `ring_config` starts a ring even when
   `enable_ring_transport=False`.
+- `record_mode_v1=True` retains the ring configuration without constructing a
+  legacy ring. A later `create_record_runtime()` constructs the encoded-record
+  ring directly. Recurring D2H windows require this mode.
 - With no explicit `ring_config`, the wrapper defaults to a 4096 MiB GPU
   payload ring, 4096 MiB pinned staging ring, and 65,536 task entries. These
   are much larger than raw `RingConfig()` defaults.
@@ -350,7 +354,8 @@ engine.flush_and_wait(timeout_s=600.0)
 ```
 
 The public operations are `create_record_runtime()`, `bind_hook()`,
-`emit_output()`, `prepare_replay()`, and `flush_and_wait()`.
+`emit_output()`, `prepare_replay()`, `define_d2h_window_pattern()`,
+`advance_boundary()`, and `flush_and_wait()`.
 
 `RecordRuntime.bind_hook()` assigns stable output IDs and binds the hook to the
 record ring. `RecordRuntime.emit_output()` reserves and publishes one eager
@@ -359,6 +364,58 @@ descriptor before its host-selected producer occurrence.
 host-selected entries in an existing physical plan. Every accepted producer
 occurrence has exactly one descriptor and one task.
 
+### Recurring D2H windows
+
+Recurring D2H windows constrain record-mode transfers to framework-supplied,
+repeating execution windows. The integration defines the pattern once and
+publishes one boundary at each ordered execution point:
+
+```python
+windows = dmi.RecurringD2HWindowConfig()
+windows.enabled = True
+windows.minimum_record_probe_retry_interval_occurrences = 4
+windows.timing_revalidation_retry_interval_occurrences = 4
+windows.capacity_flush_fallback_threshold = 3
+windows.capacity_flush_count_reset_interval_periods = 32
+windows.debug_enabled = False
+
+ring = dmi.RingConfig()
+ring.recurring_d2h_windows = windows
+engine = dmi.MonitoringEngine(
+    model_id="my-run",
+    host_engine=host,
+    ring_config=ring,
+    record_mode_v1=True,
+)
+runtime = engine.create_record_runtime(record_format)
+
+accepted = runtime.define_d2h_window_pattern(
+    period=100,
+    windows=[(20, 30), (70, 80)],
+)
+
+# Enqueue this on the same ordered framework CUDA stream used by producers.
+runtime.advance_boundary()
+```
+
+`period` and every window endpoint are boundary-counter values. Windows must
+be nonempty, sorted, non-overlapping half-open ranges satisfying
+`0 <= begin < end <= period`. `initial_counter` defaults to zero and may be
+provided when defining a new pattern version. Pattern definition and every
+`advance_boundary()` call must use the same ordered framework CUDA stream.
+Define a pattern before the first boundary call and before CUDA Graph capture.
+The definition returns `True` when the pattern is queued. It returns `False`
+without reactivating window scheduling when the engine has already entered
+terminal fallback. Invalid arguments and CUDA/runtime failures still raise.
+
+The drain learns a transferable byte grant independently for each recurring
+window. While the feature is active, ordinary batch thresholds and timeout
+flushes do not initiate D2H. Full ring capacity can still force a flush; after
+`capacity_flush_fallback_threshold` such completed flushes, the engine
+permanently returns to existing batched draining. A too-small learned window
+periodically probes one complete record. Repeated failures wait increasingly
+long multiples of `minimum_record_probe_retry_interval_occurrences`.
+
 When a hook uses a device gate, the integration must apply the same selection
 before host reservation and descriptor publication. A gated-off occurrence is
 absent from the record stream: it has no reservation, descriptor, task,
@@ -954,6 +1011,7 @@ do not reconfigure it.
 | `clone_slices` | `False` | Clone multi-request slices so full assembled tensors can be released sooner. |
 | `insert_queue_max_bytes` | `4 GiB` | Reserved field; current v1 does not apply it to host queue limits. |
 | `insert_queue_max_items` | `65536` | Reserved field; current v1 does not apply it to host queue limits. |
+| `recurring_d2h_windows` | disabled configuration | Grouped `RecurringD2HWindowConfig`; supported only by the encoded-record ring. |
 
 Flush triggers are ORed, and full ring capacity always forces a flush. For
 live readback of small workloads, configure `drain_flush_timeout_us`; the poll
@@ -964,6 +1022,22 @@ Actual ClickHouse queue batching/backpressure is controlled by
 `StageConfig.input_queue` and its public `QueueConfig`; do not rely on the two
 reserved ring fields.
 
+### `RecurringD2HWindowConfig`
+
+The grouped recurring-window configuration is copied with `RingConfig` during
+ring construction.
+
+| Field | Default | Meaning |
+| --- | --- | --- |
+| `enabled` | `False` | Enable recurring-window scheduling. |
+| `progress` | packed version/counter | Fixed progress backend for v1; leave unchanged. |
+| `grant_policy` | binary adaptive | Fixed grant policy for v1; leave unchanged. |
+| `minimum_record_probe_retry_interval_occurrences` | `0` | Recurring-window occurrences between minimum-record probes; must be positive when enabled. |
+| `timing_revalidation_retry_interval_occurrences` | `4` | Base recurring-window occurrences between failed timing-revalidation probes; must be positive when enabled. |
+| `capacity_flush_fallback_threshold` | `0` | Completed capacity-forced flushes before permanent batched fallback; must be positive when enabled. |
+| `capacity_flush_count_reset_interval_periods` | `32` | Reset accumulated capacity-forced-flush evidence when consecutive events are separated by at least this many complete pattern periods; must be positive when enabled. |
+| `debug_enabled` | `False` | Log each real recurring-window D2H issue and completion. |
+
 ### `ClickHouseClientConfig`
 
 Construct `ClickHouseClientConfig()` and set all fields before passing it to

native/Makefile

@@ -12,7 +12,14 @@ SRCS := csrc/bindings.cpp \
         csrc/reference_python_capture_sink.cpp \
         csrc/clickhouse_client.cpp \
         csrc/ring/drain_thread.cpp \
+        csrc/ring/d2h_window_debug.cpp \
+        csrc/ring/d2h_window_grant_policy.cpp \
+        csrc/ring/d2h_window_mode.cpp \
+        csrc/ring/d2h_window_pattern.cpp \
+        csrc/ring/d2h_window_progress.cpp \
+        csrc/ring/d2h_window_subsystem.cpp \
         csrc/ring/p2p_thread.cpp \
+        csrc/ring/recurring_d2h_grant_controller.cpp \
         csrc/ring/record_consumer.cpp \
         csrc/ring/ring_torch_op.cpp
 HOST_SRCS := csrc/bindings.cpp csrc/clickhouse_client.cpp
@@ -22,6 +29,14 @@ HEADERS := csrc/dmx_host_utils.h \
            csrc/dmx_host_engine.h csrc/record_schema.h \
            csrc/ring/ring_config.h csrc/ring/ring_state.h csrc/ring/ring_alloc.h \
            csrc/ring/publication_word.h csrc/ring/drain_task.h csrc/ring/drain_thread.h \
+           csrc/ring/cuda_check.h \
+           csrc/ring/d2h_window_config.h csrc/ring/d2h_window_debug.h \
+           csrc/ring/d2h_window_grant_policy.h csrc/ring/d2h_window_marker.h \
+           csrc/ring/d2h_window_mode.h csrc/ring/d2h_window_pattern.h \
+           csrc/ring/d2h_window_pause.h csrc/ring/d2h_window_progress.h \
+           csrc/ring/d2h_window_progress_layout.h \
+           csrc/ring/d2h_window_subsystem.h \
+           csrc/ring/recurring_d2h_grant_controller.h \
            csrc/ring/record_descriptor.h csrc/ring/record_consumer.h \
            csrc/ring/record_sink.h \
            csrc/ring/p2p_thread.h csrc/ring/pinned_staging.h \
@@ -173,6 +188,7 @@ NVCCFLAGS := \
     -Icsrc
 
 RING_SRCS := csrc/ring/producer.cu \
+             csrc/ring/d2h_window_marker.cu \
              csrc/ring/ring_engine.cu \
              csrc/ring/ring_engine_py.cu
 RING_OBJS := $(patsubst csrc/%.cu,$(BUILD_DIR)/%.o,$(RING_SRCS))

native/csrc/bindings.cpp

@@ -605,6 +605,49 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
   // Ring offload engine
   // -------------------------------------------------------------------------
 #ifndef DMI_HOST_ONLY
+  py::enum_<ring::D2HWindowProgressKind>(m, "D2HWindowProgressKind")
+      .value("PACKED_VERSION_COUNTER",
+             ring::D2HWindowProgressKind::PACKED_VERSION_COUNTER);
+  py::enum_<ring::D2HWindowGrantPolicyKind>(m, "D2HWindowGrantPolicyKind")
+      .value("BINARY_ADAPTIVE",
+             ring::D2HWindowGrantPolicyKind::BINARY_ADAPTIVE);
+  py::class_<ring::RecurringD2HWindowConfig>(m, "RecurringD2HWindowConfig")
+      .def(py::init<>())
+      .def_readwrite("enabled", &ring::RecurringD2HWindowConfig::enabled)
+      .def_readwrite("progress", &ring::RecurringD2HWindowConfig::progress)
+      .def_readwrite("grant_policy",
+                     &ring::RecurringD2HWindowConfig::grant_policy)
+      .def_readwrite(
+          "minimum_record_probe_retry_interval_occurrences",
+          &ring::RecurringD2HWindowConfig::
+              minimum_record_probe_retry_interval_occurrences)
+      .def_readwrite(
+          "timing_revalidation_retry_interval_occurrences",
+          &ring::RecurringD2HWindowConfig::
+              timing_revalidation_retry_interval_occurrences)
+      .def_readwrite(
+          "capacity_flush_fallback_threshold",
+          &ring::RecurringD2HWindowConfig::capacity_flush_fallback_threshold)
+      .def_readwrite(
+          "capacity_flush_count_reset_interval_periods",
+          &ring::RecurringD2HWindowConfig::
+              capacity_flush_count_reset_interval_periods)
+      .def_readwrite("debug_enabled",
+                     &ring::RecurringD2HWindowConfig::debug_enabled);
+  py::enum_<ring::D2HWindowMode>(m, "D2HWindowMode")
+      .value("DISABLED", ring::D2HWindowMode::DISABLED)
+      .value("ENABLED_NO_PATTERN", ring::D2HWindowMode::ENABLED_NO_PATTERN)
+      .value("ENABLED_ACTIVE", ring::D2HWindowMode::ENABLED_ACTIVE)
+      .value("ENABLED_FALLBACK", ring::D2HWindowMode::ENABLED_FALLBACK);
+  py::class_<ring::D2HWindowRuntimeSnapshot>(m, "D2HWindowRuntimeSnapshot")
+      .def_readonly("mode", &ring::D2HWindowRuntimeSnapshot::mode)
+      .def_readonly(
+          "capacity_forced_flush_count",
+          &ring::D2HWindowRuntimeSnapshot::capacity_forced_flush_count)
+      .def_readonly(
+          "capacity_flush_fallback_threshold",
+          &ring::D2HWindowRuntimeSnapshot::capacity_flush_fallback_threshold);
+
   py::class_<ring_py::RingConfig>(m, "RingConfig")
       .def(py::init<>())
       .def_readwrite("task_ring_entries",         &ring_py::RingConfig::task_ring_entries)
@@ -618,7 +661,9 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def_readwrite("drain_flush_timeout_us",     &ring_py::RingConfig::drain_flush_timeout_us)
       .def_readwrite("clone_slices",              &ring_py::RingConfig::clone_slices)
       .def_readwrite("insert_queue_max_bytes",    &ring_py::RingConfig::insert_queue_max_bytes)
-      .def_readwrite("insert_queue_max_items",    &ring_py::RingConfig::insert_queue_max_items);
+      .def_readwrite("insert_queue_max_items",    &ring_py::RingConfig::insert_queue_max_items)
+      .def_readwrite("recurring_d2h_windows",
+                     &ring_py::RingConfig::recurring_d2h_windows);
 
   // Production sinks remain native-only.  The explicitly named reference
   // bridge below is opt-in and intentionally pays the Python GIL/copy cost.
@@ -764,6 +809,33 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def("staging_cap", &ring_py::RingEnginePy::staging_cap)
       .def("task_cap",    &ring_py::RingEnginePy::task_cap)
       .def("payload_tensor", &ring_py::RingEnginePy::payload_tensor)
+      .def("define_d2h_window_pattern",
+           [](ring_py::RingEnginePy& self,
+              uint64_t period,
+              const std::vector<std::pair<uint64_t, uint64_t>>& windows,
+              std::optional<uint64_t> initial_counter) {
+             std::vector<ring::D2HWindowOffset> offsets;
+             offsets.reserve(windows.size());
+             for (const auto& window : windows) {
+               offsets.push_back({window.first, window.second});
+             }
+             py::gil_scoped_release release;
+             return self.define_d2h_window_pattern(
+                 period, std::move(offsets), initial_counter);
+           },
+           py::arg("period"),
+           py::arg("windows"),
+           py::arg("initial_counter") = std::nullopt)
+      .def("_recurring_d2h_windows_enabled",
+           &ring_py::RingEnginePy::recurring_d2h_windows_enabled)
+      .def("_d2h_window_device_progress_tensor",
+           &ring_py::RingEnginePy::d2h_window_device_progress_tensor)
+      .def("_d2h_window_cpu_visible_progress_tensor",
+           &ring_py::RingEnginePy::d2h_window_cpu_visible_progress_tensor)
+      .def("_d2h_window_progress_kind",
+           &ring_py::RingEnginePy::d2h_window_progress_kind)
+      .def("d2h_window_runtime_snapshot",
+           &ring_py::RingEnginePy::d2h_window_runtime_snapshot)
       // Safety-net surface (eager only).  available_capacity() and
       // reserve_one() are CPU-only and fast -- no GIL release needed.
       // flush_and_wait() blocks on cudaStreamSynchronize + drain flush --

native/csrc/ring/cuda_check.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <cuda_runtime.h>
+
+#include <stdexcept>
+#include <string>
+
+namespace ring {
+
+// Shared by the D2H window translation units so a CUDA failure reports the
+// same "<operation> failed: <reason>" shape wherever it is raised.
+inline void check_cuda(cudaError_t error, const char* operation) {
+    if (error == cudaSuccess)
+        return;
+    throw std::runtime_error(std::string(operation) +
+                             " failed: " + cudaGetErrorString(error));
+}
+
+// Kernel launches report asynchronously, so their errors surface through
+// cudaGetLastError() rather than a return value.
+inline void check_cuda_launch(const char* operation) {
+    check_cuda(cudaGetLastError(), operation);
+}
+
+}  // namespace ring

native/csrc/ring/d2h_window_config.h

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <cstdint>
+
+namespace ring {
+
+enum class D2HWindowProgressKind : uint8_t {
+    PACKED_VERSION_COUNTER = 0,
+};
+
+enum class D2HWindowGrantPolicyKind : uint8_t {
+    BINARY_ADAPTIVE = 0,
+};
+
+struct D2HWindowOffset {
+    uint64_t begin{0};
+    uint64_t end{0};
+};
+
+struct RecurringD2HWindowConfig {
+    bool enabled{false};
+    D2HWindowProgressKind progress{D2HWindowProgressKind::PACKED_VERSION_COUNTER};
+    D2HWindowGrantPolicyKind grant_policy{D2HWindowGrantPolicyKind::BINARY_ADAPTIVE};
+    uint64_t minimum_record_probe_retry_interval_occurrences{0};
+    uint64_t timing_revalidation_retry_interval_occurrences{4};
+    uint64_t capacity_flush_fallback_threshold{0};
+    // Age out isolated fallback evidence after this many complete pattern
+    // periods. The progress-distance threshold is this value times T.
+    uint64_t capacity_flush_count_reset_interval_periods{32};
+    bool debug_enabled{false};
+};
+
+}  // namespace ring