fix(test): the intake audit could clear intake over an empty comparison (BACKLOG #1292)

docs/LOAD-TESTING.md

@@ -297,6 +297,43 @@ headroom denominator (in + out events, not messages). Exit codes match `--load`.
   backlog stayed low) so engine numbers are never silently the harness's own ceiling.
 - The `zero_loss` gate is **exact** by default (no message may be lost). At-least-once re-deliveries
   (`sink_received > engine_written`) are reported as a count and are *not* treated as loss.
+- **`intake_audit` (connection-scale runs only) is the per-MESSAGE companion to `zero_loss`, and it
+  answers a question `zero_loss` cannot ask.** `zero_loss` compares COUNTS, so its
+  `engine_read N < confirmed sent M (lost K on intake)` reads identically whether the engine lost an
+  acknowledged message or the harness's own `engine_read` gauge was short — and `engine_read` is
+  itself a `COUNT(*)` sampled through two HTTP layers, so a second count could not separate them.
+  The audit records each send's control id as its response frame comes back and then asks the step's
+  own store, per message, whether that row is there. Its verdict is on the console, in the JSON
+  artifact under `records[].intake_audit`, and appended to a failing `no_loss` detail:
+  - `INTAKE_COMPLETE` — every confirmed send has a row.
+  - `SAMPLING_LAG` — a shortfall was reported, every confirmed send has a row anyway, **and the
+    shortfall is larger than the never-confirmed sends can account for**, so the unexplained
+    remainder is in the gauge (sample attribution or per-inbound sum coverage). A harness defect.
+  - `UNCONFIRMED_SHORTFALL` — the shortfall is no larger than the set of sends the harness never got
+    a response frame for, so it implicates **neither** intake **nor** the gauge. Split out from
+    `SAMPLING_LAG` because the excusal clamps its allowance to zero once the unconfirmed count
+    exceeds its budget, and the shortfall then consists of sends the engine may never have received;
+    blaming the gauge for those accused an instrument that was exactly right, and contradicted the
+    `no_loss` line this verdict is appended to, which already calls that step a systemic no-ACK fault.
+  - `INVARIANT_SUSPECT` — a send the engine accept-ACKed has no row in its own stopped, committed
+    store. The engine branch: on a deployment an acknowledged message would be lost at intake. The
+    verdict names the sequence numbers, so it is reproducible rather than statistical.
+  - `CORRELATION_SUSPECT` — only *rejected* sends are unmatched. Not an engine finding: several NAK
+    paths record their row with a NULL control id, so a rejected message is expected to be
+    unmatchable by control id.
+  - `PROBE_UNUSABLE` — the audit could not answer. At least: its own read came back empty or
+    truncated, the send ledger was incomplete, or **nothing was ever confirmed**, which leaves the
+    compared set empty. That last one is the ledger-side positive control and it is not redundant
+    with the store-side one — a ledger holding only unconfirmed sends is non-empty by total and
+    still compares nothing, so without it the audit returned a conclusive "not in intake" computed
+    over zero elements. Deliberately **not** rendered as "everything is missing", and deliberately
+    **not** a pass either.
+
+  It runs at two moments: LIVE (engine still up, only on a shortfall) and POST-MORTEM (engine
+  stopped, always). The post-mortem one is authoritative — a live read can be explained away as
+  early sampling; a read of a stopped engine's store cannot. Set `intake_audit = false` in the
+  `[connscale]` profile to skip it on a heavy operator sweep; it is on by default, because a check
+  an operator has to remember to enable is absent on exactly the run that needed it.
 
 ## Known limitations
 

harness/load/__init__.py

@@ -8,10 +8,23 @@
 fan-out and times each message end-to-end; an engine poller samples the HTTP API for throughput,
 backlog, and drain. See ``docs/LOAD-TESTING.md``.
 
-Like :mod:`harness.scenarios`, this package imports no PySide6 and never imports the engine's
-``pipeline``/``store``/``config`` internals — only the **pure** surfaces the harness is allowed to
-use: the MLLP framing primitives (:mod:`messagefoundry.transports.mllp`), the parsing library, the
-generators, and the HTTP :class:`~messagefoundry.apiclient.EngineClient`.
+Like :mod:`harness.scenarios`, this package imports no PySide6, and drives the engine through the
+**pure** surfaces a client is allowed to use: the MLLP framing primitives
+(:mod:`messagefoundry.transports.mllp`), the parsing library, the generators, and the HTTP
+:class:`~messagefoundry.apiclient.EngineClient`.
+
+**The store carve-out, and it is not the client rule being bent.** The rigs that OWN the engine
+subprocess they measure — they spawn it, hand it a store, and stop it — are test rigs rather than
+clients, and some of their jobs are only doable against the store directly. In
+:mod:`harness.load.connscale` that is at least emptying a shared server store between sweep steps
+(``runner._reset_server_store``) and the BACKLOG #1292 intake audit's per-message read
+(``runner._store_reader``); :mod:`harness.load.shardcert` provisions its own store the same way.
+Each goes through the ``Store`` protocol via ``open_store``, lazily imported inside the function so
+the import graph of everything else is unchanged, and each is a read/reset path on a store the rig
+itself provisioned — never a shortcut around the API for something the API could answer. (Separately
+and harmlessly, several modules import the ``AckMode`` enum from ``config``; that is a value type,
+not engine state.) The Qt-free client rule itself is unchanged: nothing here imports PySide6, and
+the monitoring path is still the HTTP API.
 """
 
 from __future__ import annotations

harness/load/connscale/driver.py

@@ -22,6 +22,7 @@
 
 import asyncio
 
+from harness.load.connscale.intake_audit import IntakeLedger
 from harness.load.corpus import Corpus, Outgoing
 from harness.load.correlator import Correlator
 from harness.load.metrics import LiveMetrics
@@ -50,13 +51,18 @@ def __init__(
         correlator: Correlator,
         metrics: LiveMetrics,
         queue_max: int = 256,
+        ledger: IntakeLedger | None = None,
     ) -> None:
         if count < 1:
             raise ValueError("connection count must be >= 1")
         self._host = host
         self._base_port = base_port
         self._count = count
         self._m = metrics
+        # BACKLOG #1292: ONE ledger SHARED across the N connections, exactly as the correlator and
+        # LiveMetrics are. The reconcile it discriminates aggregates across all N, so a per-connection
+        # ledger would answer a narrower question than the assertion it has to explain.
+        self._ledger = ledger
         # One persistent, pipelined connection per inbound port; expect_ack so each send→ACK is timed.
         self._conns = [
             PersistentConnection(
@@ -66,6 +72,7 @@ def __init__(
                 metrics,
                 expect_ack=True,
                 queue_max=queue_max,
+                ledger=ledger,
             )
             for i in range(count)
         ]