Surface native host-sink failure instead of silently truncating captured activations

docs/huggingface.md

@@ -86,9 +86,18 @@ token_mask = out.dmi_internal.token_mask
 
 For layer-tuple fields such as `hidden_states`, `count` validates the number of
 layers in the reassembled tuple. It does not validate token completeness inside
-each layer tensor unless `match_token_ranges=True` is set. That option checks
-the captured row ranges against the token ranges recorded during this generate
-call. For per-layer fields it uses a fast representative-layer check.
+each layer tensor unless token-range validation runs. Lazy reads attached to
+`generate_with_monitoring_dict(...)` now perform that validation by default
+whenever request IDs and token ranges were recorded for the generate call.
+`match_token_ranges=False` opts out for a specific requirement. For per-layer
+fields every layer present is checked and the error names the offending layer.
+A layer that wrote no rows at all is not detectable this way -- it is simply
+absent from the tuple -- so pass `count=` when layer completeness matters.
+
+`logits` is a special case: `generate()` passes `logits_to_keep=1`, so the
+prefill logits row covers only the last prompt position rather than the whole
+prompt. Its rows are therefore validated on token-range *ends* only, which
+still catches a missing decode step.
 
 Supported mapped fields are:
 
@@ -147,9 +156,18 @@ norm = hidden_states[0].float().norm(dim=-1)[token_mask].mean()
 
 If a tensor field and `token_mask` have different `[batch, seq]` shapes, it
 usually means the tensor field was read before all rows for that field arrived.
-Use `match_token_ranges=True` with `retry=True` to wait for expected token
-ranges. If you already cached a partial field, clear that field's cache and read
-it again:
+Use `retry=True` to wait for expected token ranges after an incomplete read.
+If you intentionally want to skip token-range validation for a field, set
+`match_token_ranges=False` on that field's requirement. Both are set through
+`require()`, which also needs the expected `count`:
+
+```python
+out.dmi_internal.require(
+    "hidden_states", count=model.config.num_hidden_layers, retry=True
+)
+```
+ If you already cached a partial field, clear that
+field's cache and read it again:
 
 ```python
 out.dmi_internal.clear_cache("hidden_states")

docs/integration-api-v1.md

@@ -199,6 +199,7 @@ engine.ring_capacities() -> RingCapacities
 engine.capture_enabled -> bool
 engine.set_capture_enabled(enabled: bool) -> None
 engine.next_auto_group_id() -> int
+engine.sink_stats() -> SinkStats
 engine.close() -> None
 ```
 
@@ -240,11 +241,48 @@ fails, Python-visible capture flags remain unchanged.
 `next_auto_group_id()` returns engine-scoped integers starting at zero. It is
 not synchronized for concurrent callers.
 
+`sink_stats()` returns an immutable `SinkStats` snapshot of the host sink's
+loss and backpressure counters:
+
+| Field | Meaning |
+| --- | --- |
+| `dropped` | Rows the insert queue discarded, via a configured `OnFullPolicy.DROP` or because the engine was already stopping. |
+| `suppressed` | Rows the native P2P thread failed to submit and swallowed. Never a configured behavior. |
+| `full_errors` | Enqueue attempts that hit a full queue. |
+| `closed_errors` | Enqueue attempts against a closed queue. |
+| `too_large_errors` | Rows rejected as larger than the queue's item cap. |
+| `retries` | Enqueue retries under `OnFullPolicy.RETRY`. |
+| `lost_rows` | `dropped + suppressed` -- rows that provably never reached the sink. |
+
+The counters are cumulative for the life of the engine, never reset by DMI,
+and read best-effort: a backend that does not expose them reports zeros rather
+than raising. `close()` caches the final snapshot, so `sink_stats()` keeps
+working after teardown. Prefer it over the native warning: the P2P thread's
+submit-failure message is printed at most once per process no matter how many
+rows are lost.
+
 `close()` stops and flushes the ring, clears the active binding, closes host
-input, and stops the host engine. It is terminal and effectively idempotent.
-Shutdown exceptions are suppressed, so an integration requiring an
-authoritative final read must ensure every worker reaches this close path and
-should separately check native host failures.
+input, and stops the host engine. It is terminal and idempotent -- both engine
+handles are cleared before any exception leaves the method, so a second call is
+a no-op: it neither re-reads the counters nor re-reports them. Only the first
+close sees the engines, so `sink_stats()` keeps returning that snapshot however
+many times `close()` is called.
+
+`close()` raises `RuntimeError` when:
+
+- the ring or host engine raised during teardown;
+- `raise_if_failed()` reports a host worker failure, in which case the message
+  carries the `SinkStats` snapshot and the original failure is chained as
+  `__cause__`; or
+- `sink_stats().suppressed` is nonzero, meaning rows were produced but never
+  submitted.
+
+When a teardown failure and a sink failure coincide, the sink failure is the
+one raised and the earlier error is attached as `__context__`. A nonzero
+`dropped` count raises no exception -- it is reachable through a configured
+drop policy -- but emits a `RuntimeWarning`. Integrations that treat a run as
+authoritative should call `close()` inside their own error handling rather
+than a bare `finally`, since it can now replace an in-flight exception.
 
 Closing does not disable or uninstall HookPoints: they retain hook IDs and the
 old payload tensor. Treat the attached model as terminal too. A later CUDA
@@ -919,7 +957,9 @@ DMXHostEngine(insert_stage: StageConfig)
 ```
 
 Construction validates/copies stage configuration but does not connect to the
-database. Public lifecycle and diagnostics are:
+database. The engine always keeps the ingest and queue counters that
+`MonitoringEngine.close()` uses to report silently lost rows; they measured
+below noise on the submit path. Public lifecycle and diagnostics are:
 
 ```python
 start() -> None
@@ -929,13 +969,37 @@ request_abort() -> None
 join(timeout_s: float | None = None) -> bool
 failures() -> list[ThreadFailure]
 raise_if_failed() -> None
+profiling() -> StatsSnapshot | None
+reset_metrics() -> None
 ```
 
 `start()` is asynchronous: it can return before a worker fails to connect or
 initialize. `stop()` returning true means threads joined, not that inserts
 succeeded. After shutdown, call `raise_if_failed()`; `failures()` returns
 records with `stage`, `thread_name`, `where`, `exc_type`, and `exc_what`.
 
+`profiling()` returns a `StatsSnapshot` by value, or `None` when the engine was
+built without stats. Counters are cumulative; `reset_metrics()` zeroes them.
+Both are safe to call after `stop()`. Prefer `MonitoringEngine.sink_stats()`
+unless you need the per-stage timing fields.
+
+### `StatsSnapshot`, `IngestStats`, `QueueStats`, and `StageStats`
+
+`StatsSnapshot` is the immutable value returned by `DMXHostEngine.profiling()`.
+Its fields are `ingest` (an `IngestStats`), `queue_by_stage`, and
+`stage_by_stage` (lists of `QueueStats` and `StageStats`, one entry per stage;
+the host engine has exactly one). All fields on all four classes are read-only
+and the classes cannot be constructed from Python.
+
+| Class | Read-only fields |
+| --- | --- |
+| `IngestStats` | `submit_calls`, `items_submitted`, `submit_enqueue_calls`, `submit_enqueue_s` |
+| `QueueStats` | `enqueued`, `dropped`, `full_errors`, `closed_errors`, `too_large_errors`, `retries` |
+| `StageStats` | `batches`, `items_in`, `items_out`, `dequeue_calls`, `dequeue_timeouts`, `process_calls`, `enqueue_calls`, `output_calls`, `output_items`, `dequeue_s`, `dequeue_idle_s`, `process_s`, `enqueue_s`, `output_s` |
+
+The `_s` fields are seconds and stay zero unless the engine was also built with
+timing enabled, which DMI does not do.
+
 ### `ThreadFailure`
 
 `ThreadFailure` is the immutable diagnostic record returned by
@@ -1107,7 +1171,7 @@ InternalRequirement(
     retry: bool = False,
     timeout_s: float | None = 30.0,
     poll_s: float = 0.25,
-    match_token_ranges: bool = False,
+    match_token_ranges: bool = True,
 )
 ```
 
@@ -1124,7 +1188,7 @@ requirements.require(
     retry: bool = False,
     timeout_s: float | None = 30.0,
     poll_s: float = 0.25,
-    match_token_ranges: bool = False,
+    match_token_ranges: bool = True,
 ) -> InternalRequirements
 ```
 
@@ -1164,18 +1228,32 @@ handle.require(
     retry: bool = False,
     timeout_s: float | None = 30.0,
     poll_s: float = 0.25,
-    match_token_ranges: bool = False,
+    match_token_ranges: bool = True,
 ) -> handle
 ```
 
 `count` validates `len(field_value)`: it is a layer count for per-layer tuples
 and a batch count for global tensors, not a token or database-row count. With
 `retry=True`, synchronous field access polls missing/incomplete data until
 success or timeout. `timeout_s=None` can block forever. Database/runtime errors
-are not retried. `match_token_ranges=True` performs validation only when both
-nonempty request IDs and ranges were supplied; otherwise it is a no-op. For a
-per-layer field the current v1 implementation checks only its highest present
-layer, not every layer.
+are not retried. When both nonempty request IDs and ranges were supplied,
+token-range validation runs by default -- including for a field with no
+requirement registered at all. `match_token_ranges=False` opts out for an
+individual field, and otherwise the flag is a no-op. For a per-layer field
+every layer present in the result is checked, and the error names the offending
+`layer=`.
+
+One loss the range check cannot see: a layer that wrote no rows at all leaves
+nothing to compare against, so it is silently absent from the reassembled
+tuple. Only a declared `count` catches that -- pass one when layer
+completeness matters.
+
+`logits` is validated on range *ends* only. Its rows are re-based natively to
+the trailing positions of each step, because HF generate passes
+`logits_to_keep=1` and vLLM computes one logit per request, so a full-range
+comparison would fail on every prefill with a prompt longer than one token.
+The relaxed check still detects a missing step; it does not detect a wrong
+start offset.
 
 Mapped dynamic attributes are:
 

native/csrc/bindings.cpp

@@ -98,6 +98,10 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
   using QueueConfig = DMXHostEngine::QueueConfig;
   using EnqueuePolicy = DMXHostEngine::EnqueuePolicy;
   using Duration = DMXHostEngine::Duration;
+  using IngestStats = DMXHostEngine::IngestStats;
+  using QueueStats = DMXHostEngine::QueueStats;
+  using StageStats = DMXHostEngine::StageStats;
+  using StatsSnapshot = DMXHostEngine::StatsSnapshot;
 
   py::enum_<dmx_host::OnFullPolicy>(m, "OnFullPolicy")
       .value("RAISE", dmx_host::OnFullPolicy::RAISE)
@@ -159,6 +163,41 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def_readonly("exc_type", &ThreadFailure::exc_type)
       .def_readonly("exc_what", &ThreadFailure::exc_what);
 
+  py::class_<IngestStats>(m, "IngestStats")
+      .def_readonly("submit_calls", &IngestStats::submit_calls)
+      .def_readonly("items_submitted", &IngestStats::items_submitted)
+      .def_readonly("submit_enqueue_calls", &IngestStats::submit_enqueue_calls)
+      .def_readonly("submit_enqueue_s", &IngestStats::submit_enqueue_s);
+
+  py::class_<QueueStats>(m, "QueueStats")
+      .def_readonly("enqueued", &QueueStats::enqueued)
+      .def_readonly("dropped", &QueueStats::dropped)
+      .def_readonly("full_errors", &QueueStats::full_errors)
+      .def_readonly("closed_errors", &QueueStats::closed_errors)
+      .def_readonly("too_large_errors", &QueueStats::too_large_errors)
+      .def_readonly("retries", &QueueStats::retries);
+
+  py::class_<StageStats>(m, "StageStats")
+      .def_readonly("batches", &StageStats::batches)
+      .def_readonly("items_in", &StageStats::items_in)
+      .def_readonly("items_out", &StageStats::items_out)
+      .def_readonly("dequeue_calls", &StageStats::dequeue_calls)
+      .def_readonly("dequeue_timeouts", &StageStats::dequeue_timeouts)
+      .def_readonly("process_calls", &StageStats::process_calls)
+      .def_readonly("enqueue_calls", &StageStats::enqueue_calls)
+      .def_readonly("output_calls", &StageStats::output_calls)
+      .def_readonly("output_items", &StageStats::output_items)
+      .def_readonly("dequeue_s", &StageStats::dequeue_s)
+      .def_readonly("dequeue_idle_s", &StageStats::dequeue_idle_s)
+      .def_readonly("process_s", &StageStats::process_s)
+      .def_readonly("enqueue_s", &StageStats::enqueue_s)
+      .def_readonly("output_s", &StageStats::output_s);
+
+  py::class_<StatsSnapshot>(m, "StatsSnapshot")
+      .def_readonly("ingest", &StatsSnapshot::ingest)
+      .def_readonly("queue_by_stage", &StatsSnapshot::queue_by_stage)
+      .def_readonly("stage_by_stage", &StatsSnapshot::stage_by_stage);
+
   py::class_<StageConfig>(m, "StageConfig")
       .def(py::init<>())
       .def_readwrite("name", &StageConfig::name)
@@ -213,6 +252,12 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
            py::call_guard<py::gil_scoped_release>())
       .def("failures", &DMXHostEngine::failures)
       .def("raise_if_failed", &DMXHostEngine::raise_if_failed)
+      // Both take the engine's profiling mutex, which C++ worker threads
+      // also hold; release the GIL like every other blocking method here.
+      .def("profiling", &DMXHostEngine::profiling,
+           py::call_guard<py::gil_scoped_release>())
+      .def("reset_metrics", &DMXHostEngine::reset_metrics,
+           py::call_guard<py::gil_scoped_release>())
       // Submit a pre-formatted ClickHouseRow directly to the insert stage.
       // Called from the ring transport drain callback after format processing.
       .def("submit_direct",
@@ -304,6 +349,8 @@ PYBIND11_MODULE(TORCH_EXTENSION_NAME, m) {
       .def("payload_cap", &ring_py::RingEnginePy::payload_cap)
       .def("staging_cap", &ring_py::RingEnginePy::staging_cap)
       .def("task_cap",    &ring_py::RingEnginePy::task_cap)
+      .def("suppressed_submit_failures",
+           &ring_py::RingEnginePy::suppressed_submit_failures)
       .def("payload_tensor", &ring_py::RingEnginePy::payload_tensor)
       // Safety-net surface (eager only).  available_capacity() and
       // reserve_one() are CPU-only and fast -- no GIL release needed.

native/csrc/dmx_host_engine.h

@@ -8,11 +8,20 @@ namespace dmx_host{
 
 // DMXHostEngine is a single-stage ClickHouse insert pipeline.
 // Pre-assembled ClickHouseRows are submitted via submit_direct().
-class DMXHostEngine : public PipelinedEngine<dmx_host_queue_item, uint64_t, 1, QueueOptions<false, false, false>, false,
+class DMXHostEngine : public PipelinedEngine<dmx_host_queue_item, uint64_t, 1, QueueOptions<false, false, false>, true,
 NoOutputHandler<dmx_host_queue_item> >{
 public:
+    // Stats are always on: they are what lets teardown report silently lost
+    // rows, and they measured below noise (+33ns/row against an 86ns/row
+    // spread) on the submit path.  Counts only; timing stays off.
     explicit DMXHostEngine(StageConfig insert_stage):
-    PipelinedEngine(std::array<StageConfig, 1>{std::move(insert_stage)}, EngineConfig{}){}
+    PipelinedEngine(
+        std::array<StageConfig, 1>{std::move(insert_stage)},
+        [] {
+            EngineConfig config{};
+            config.enable_stats = true;
+            return config;
+        }()){}
 
     // Submit a pre-assembled ClickHouseRow directly to the insert stage.
     // Fields must match the order expected by ClickHouseInsertStage: