fix(harness): enforce the reconcile's read >= sent//2 floor unconditionally

CHANGELOG.md

@@ -6,6 +6,24 @@ All notable changes to MessageFoundry are documented here. The format follows
 
 ## [Unreleased]
 
+### Fixed
+- **The load harness's no-loss reconcile did not enforce the `read >= sent // 2` intake guarantee
+  0.3.2 documented.** The unconfirmed-send excusal is capped at `max(connections, half the run)`, but
+  that `max()` takes the connection count as a *floor*, and every call site passes a connection count
+  — so on a short, low-rate step (connscale-smoke's N=100 cell: ~105 sends, 100 connections) the count
+  won the max() and the intake bound degraded to `read >= 5`, the very vacuity the cap exists to
+  prevent. Nothing clamped the excusal to `sent` either, so `timeouts > sent` degraded it to
+  `read >= 0`. The half-the-run cap still decides the systemic no-ACK verdict (0.3.2's de-flake is
+  unchanged), and an **unconditional intake floor** the excusal cannot lower now enforces
+  `read >= sent // 2` in all three reconcile copies, at every call site. The estate copy also gained
+  the honest-reporting branch its two siblings had: it previously printed `read>=sent, …` on a
+  bounded-excused run whose read was demonstrably below `sent`, and its over-budget detail string now
+  matches theirs — a test pins the three in step, since nothing enforced the claim that they were.
+  *Known gap, unfixed:* the systemic no-ACK verdict is still gated on the same capped budget, so a
+  dead ACK path that nonetheless delivered everything still passes when `connections >= sent`; an
+  intake floor cannot catch a fault whose signature is a high read with no ACKs. Bounding that arm
+  needs its own change.
+
 ## [0.3.2] — 2026-07-28 — Early Access
 
 A patch release for one adopter-facing defect shipped in 0.3.1, plus two gates that were passing

harness/load/connscale/runner.py

@@ -715,8 +715,11 @@ def _build_record(
 ) -> ConnScaleRecord:
     c = metrics_counters.snapshot()
     base, final = poller.baseline, poller.final
-    # Budget = this step's connection count: at most ~one stranded in-flight per connection is a
-    # plausible teardown artifact; more is a systemic no-ACK fault the reconcile must fail.
+    # Budget = this step's connection count. NOT "~one stranded in-flight per connection" — that
+    # model was retired (this sender's `_inflight` is unbounded, so stranding scales with
+    # rate x ACK-latency): `_reconcile` treats the count only as a small-run FLOOR under its
+    # half-the-run fraction, and its separate intake floor keeps `read >= sent // 2` required here
+    # even when `count` exceeds half the step's sends (the short-hold smoke cells).
     no_loss = _reconcile(c, base, final, unconfirmed_budget=count)
     in_pipeline_peak = max((s.in_pipeline for s in samples), default=0)
 
@@ -821,21 +824,39 @@ def _reconcile(
     # a model this sender breaks (`_inflight` is an UNBOUNDED deque; open-loop sends are paced by the
     # offered rate, not an ACK slot), so genuine teardown stranding scales with rate x ACK-latency,
     # not the connection count. Bound it as a FRACTION instead — at most half the run, floored by the
-    # connection count — which keeps `read >= sent // 2` ALWAYS required. See harness/load/report.py's
-    # copy for the full rationale; the three copies are kept in step deliberately.
+    # connection count for tiny runs.
+    #
+    # That cap ALONE does NOT keep `read >= sent // 2` required, though the comment here used to claim
+    # it did: `max()` treats the connection count as a FLOOR, and THIS call site passes the step's
+    # `count` verbatim (_build_record above). A short, low-rate step sends barely more than one
+    # message per connection — connscale-smoke's N=100 cell is ~105 sends against budget
+    # max(100, 52) = 100 — so the count wins the max() and the bound collapses to `read >= 5`. So the
+    # guarantee is enforced SEPARATELY below as an intake floor the excusal cannot lower. See
+    # harness/load/report.py's copy for the full rationale; the three copies are kept in step
+    # deliberately.
     unconfirmed = c.timeouts
     budget = max(unconfirmed_budget, sent // 2)
     over_budget = unconfirmed > budget
     excused = 0 if over_budget else unconfirmed
     read_short = sent - excused - read
+    # The anti-vacuity guarantee, independent of the excusal: at least half the sends must be
+    # observed at intake whatever the budget forgives (nothing clamps `excused` to `sent`, so without
+    # this a `timeouts > sent` run would pass at `read >= 0`). Zero at sent == 0, rounds DOWN on an
+    # odd `sent`, so it is never stricter than the documented bound.
+    floor_short = sent // 2 - read
     deliver_short = written - sink_received
     drained = backlog == 0
-    ok = read_short <= 0 and deliver_short <= 0 and drained and not over_budget
+    ok = read_short <= 0 and floor_short <= 0 and deliver_short <= 0 and drained and not over_budget
     parts: list[str] = []
     if read_short > 0:
         parts.append(
             f"engine_read {read} < confirmed sent {sent - excused} (lost {read_short} on intake)"
         )
+    if floor_short > 0:
+        parts.append(
+            f"engine_read {read} < intake floor {sent // 2} (half of {sent} sent) — "
+            f"the unconfirmed-send excusal cannot lower this floor"
+        )
     if deliver_short > 0:
         parts.append(
             f"sink_received {sink_received} < engine_written {written} (lost {deliver_short})"

harness/load/estate/runner.py

@@ -356,6 +356,9 @@ def _build_record(
 ) -> EstateRecord:
     c = metrics.counters.snapshot()
     base, final = poller.baseline, poller.final
+    # Budget = the step's connection count — the small-run FLOOR under the reconcile's half-the-run
+    # excusal fraction, never a per-connection stranding allowance. `read >= sent // 2` holds here
+    # regardless of it (an estate step CAN send on the order of its connection count).
     no_loss = _reconcile(c, base, final, unconfirmed_budget=profile.count)
     in_pipeline_peak = max((s.in_pipeline for s in samples), default=0)
     read_per_s, written_per_s = _throughput_rates(samples)
@@ -432,19 +435,32 @@ def _reconcile(
     # budget it is a systemic no-ACK fault. The bound is a FRACTION of the run (at most half), floored
     # by the connection count — NOT "~one per connection", a model this sender breaks because
     # `_inflight` is an unbounded deque. Mirrors the connscale/report reconciles; see report.py.
+    #
+    # The floor arm means the budget does NOT by itself keep half the run unexcusable: this call site
+    # passes `profile.count` (the connection count), and an estate step whose sends are of the same
+    # order as its connection count lets the count win the max(), degrading the intake bound to
+    # `read >= sent - count`. The `read >= sent // 2` guarantee is therefore enforced separately below.
     unconfirmed = c.timeouts
     budget = max(unconfirmed_budget, sent // 2)
     over_budget = unconfirmed > budget
     excused = 0 if over_budget else unconfirmed
     read_short = sent - excused - read
+    # Anti-vacuity floor the excusal cannot lower (see report.py's copy for the derivation): zero at
+    # sent == 0, rounds DOWN on an odd `sent`, and holds even when `timeouts > sent`.
+    floor_short = sent // 2 - read
     deliver_short = written - sink_received
     drained = backlog == 0
-    ok = read_short <= 0 and deliver_short <= 0 and drained and not over_budget
+    ok = read_short <= 0 and floor_short <= 0 and deliver_short <= 0 and drained and not over_budget
     parts: list[str] = []
     if read_short > 0:
         parts.append(
             f"engine_read {read} < confirmed sent {sent - excused} (lost {read_short} on intake)"
         )
+    if floor_short > 0:
+        parts.append(
+            f"engine_read {read} < intake floor {sent // 2} (half of {sent} sent) — "
+            f"the unconfirmed-send excusal cannot lower this floor"
+        )
     if deliver_short > 0:
         parts.append(
             f"sink_received {sink_received} < engine_written {written} (lost {deliver_short})"
@@ -454,7 +470,16 @@ def _reconcile(
     if over_budget:
         parts.append(
             f"{unconfirmed} unconfirmed sends exceed the stranding budget "
-            f"({budget} = max(connections, half the run)) — systemic no-ACK fault"
+            f"({budget} = max(connections, half the run)) — systemic no-ACK fault "
+            f"(possible accepted-and-dropped); nothing excused"
+        )
+    elif unconfirmed > 0 and read < sent:
+        # Honest reporting, as in the two sibling copies: without this branch a bounded-excused run
+        # reported the flat "read>=sent, ..." detail while read was demonstrably BELOW sent — a claim
+        # false on its own numbers. The gap is attributed, never silently absorbed.
+        parts.append(
+            f"{unconfirmed} unconfirmed send(s) (no ACK before connection close) "
+            f"not observed at intake — not counted as loss"
         )
     detail = "; ".join(parts) if parts else "read>=sent, sink_received>=written, backlog drained"
     return NoLoss(ok, sent, read, written, sink_received, backlog, detail)

harness/load/multishard.py

@@ -135,8 +135,9 @@ class MultiShardRecord:
     ack_p99_ms: float
 
     # Unconfirmed sends (in-flight at a connection close with no ACK seen). The reconcile excuses
-    # these from the intake bound only up to ~one per connection (engines × count); surfaced here so
-    # the tolerance width is visible on a PASSING record too. Default 0 for older artifacts.
+    # these from the intake bound up to max(total connections, half the run), and never below its
+    # `read >= sent // 2` intake floor; surfaced here so the tolerance width is visible on a PASSING
+    # record too. Default 0 for older artifacts.
     timeouts: int = 0
 
     # --- per-engine disjoint-lane attribution (the no-cross-engine-steal proof) ---
@@ -560,8 +561,9 @@ def _build_record(
 ) -> MultiShardRecord:
     c = metrics.counters.snapshot()
     base, final = poller.baseline, poller.final
-    # Budget = total connection count across every engine (at most ~one stranded in-flight per
-    # connection is a plausible teardown artifact; more is a systemic no-ACK fault).
+    # Budget = total connection count across every engine — a small-run FLOOR under the reconcile's
+    # half-the-run excusal fraction, not the retired "~one stranded in-flight per connection" model.
+    # Its independent intake floor keeps `read >= sent // 2` required whatever this value is.
     no_loss = _reconcile(c, base, final, unconfirmed_budget=engines * count_per_engine)
     in_pipeline_peak = max((s.in_pipeline for s in samples), default=0)
     # Aggregate achieved/delivered rate = the read/written delta across EXACTLY the hold window

harness/load/report.py

@@ -371,7 +371,9 @@ def build_report(
             slos.extend(_phase_slos(rec, profile.slo_for(rec.phase)))
 
     # Unconfirmed-send budget = the run's total client connection count (one pool of pool_size per
-    # target): at most ~one stranded in-flight frame per connection is a plausible teardown artifact.
+    # target). It is only the small-run FLOOR under `_reconcile`'s half-the-run excusal fraction —
+    # NOT "~one stranded in-flight frame per connection", a model retired because this sender's
+    # in-flight deque is unbounded. `_reconcile` requires `read >= sent // 2` regardless of it.
     no_loss = _reconcile(
         final_counters,
         poller,
@@ -470,10 +472,15 @@ def _phase_slos(rec: PhaseRecord, slo: Slo) -> list[SloCheck]:
         # a single transport blip (one reconnect's failed open / stranded in-flights — client-side
         # noise, not loss) is >1%, so any sane threshold flips on one event. Below the floor the check
         # is not emitted at all (no verdict beats a noise-driven one); real load profiles run thousands
-        # of messages per phase and keep the gate. A mass reset/timeout FLOOD on a small phase is not
-        # un-gated by this floor: the reconcile's bounded unconfirmed-send budget fails zero_loss when
-        # timeouts exceed ~one per connection. (max_nak_rate below deliberately has no floor — a NAK
-        # is a deterministic engine verdict, not transport noise, so even one is signal.)
+        # of messages per phase and keep the gate. A mass reset/timeout FLOOD is not un-gated by this
+        # floor on the runs that need it: the reconcile fails zero_loss when timeouts exceed its
+        # stranding budget, or when intake drops below its unconditional `read >= sent // 2` floor —
+        # and on a CI smoke the whole run IS that phase. Mind the scope difference, though: the
+        # reconcile is computed ONCE over the run's final counters, so a flood confined to a sub-floor
+        # MEASURED phase inside a large multi-phase run is gated by neither. No shipped profile has
+        # such a phase today (reference's only sub-floor phase is `warmup`, which is unmeasured), so
+        # that is a known scope gap rather than a live hole. (max_nak_rate below deliberately has no
+        # floor — a NAK is a deterministic engine verdict, not transport noise, so even one is signal.)
         er = errs / sent
         out.append(
             SloCheck(
@@ -585,25 +592,51 @@ def _reconcile(
     # zero-loss run (14 stranded of 90 against a budget of 4) and red the required windows-2025 leg.
     #
     # So bound it as a FRACTION of the run, floored by the caller's connection count for tiny runs:
-    # at most half the sends may be excused, which keeps `read >= sent // 2` ALWAYS required. A dead
-    # ACK path (`timeouts == sent`) still blows the cap and fails loudly, and a run that genuinely
-    # loses more than half its sends at intake is still caught. Observed teardown stranding is ~16%,
-    # so this is ~3x the worst seen — wide enough to stop flaking, far from vacuous.
+    # at most half the sends may be excused. Observed teardown stranding is ~16%, so half is ~3x the
+    # worst seen — wide enough to stop flaking, far from vacuous. A dead ACK path (`timeouts == sent`)
+    # blows that cap and fails loudly — but ONLY while the connection-count floor does not dominate:
+    # once `unconfirmed_budget >= sent` the max() forgives even a 100%-dead ACK path, and the intake
+    # floor below cannot catch that one either, because its signature is a HIGH read with no ACKs.
+    # Closing that half needs the floor ARM of the budget bounded; the floor below closes the INTAKE
+    # half only. Known gap, deliberately not fixed here — it would re-open PR #17's de-flake.
+    #
+    # That cap ALONE does NOT deliver `read >= sent // 2`, and the comment here used to claim it did.
+    # `max()` takes the connection count as a FLOOR, not a ceiling, and every call site passes a
+    # connection count (connscale and estate pass the step's `count` verbatim; this one passes
+    # pool_size x targets). A short, low-rate step sends barely more than one message per connection,
+    # so the count WINS the max() and the bound degrades to `read >= sent - connections` — at
+    # connscale-smoke's N=100 cell (~105 sends, budget max(100, 52) = 100) that is `read >= 5`, the
+    # very vacuity the cap exists to prevent. Nothing clamps `excused` to `sent` either, so
+    # `timeouts > sent` drives it to `read >= 0` outright.
+    #
+    # Hence the guarantee is enforced SEPARATELY below, as an intake floor the excusal cannot lower.
     unconfirmed = counters.timeouts
     budget = max(unconfirmed_budget, sent // 2)
     over_budget = unconfirmed > budget
     excused = 0 if over_budget else unconfirmed
     read_short = sent - excused - read
+    # The anti-vacuity guarantee, enforced independently of the excusal AND of `tolerance` — the
+    # tolerance is an operator knob on the SHORTFALL, not a licence to lower the floor that makes
+    # "at least half the run was demonstrably ingested" true at every call site. `sent // 2` is 0 at
+    # sent == 0 (a run that sent nothing clears it trivially) and rounds DOWN on an odd `sent`, so
+    # the floor is never stricter than the documented bound.
+    floor_short = sent // 2 - read
     deliver_short = written - sink_received
     read_ok = read_short <= tolerance
+    floor_ok = floor_short <= 0
     deliver_ok = deliver_short <= tolerance
     drained = backlog == 0
-    ok = read_ok and deliver_ok and drained and not over_budget
+    ok = read_ok and floor_ok and deliver_ok and drained and not over_budget
     parts: list[str] = []
     if not read_ok:
         parts.append(
             f"engine_read {read} < confirmed sent {sent - excused} (lost {read_short} on intake)"
         )
+    if not floor_ok:
+        parts.append(
+            f"engine_read {read} < intake floor {sent // 2} (half of {sent} sent) — "
+            f"the unconfirmed-send excusal cannot lower this floor"
+        )
     if not deliver_ok:
         parts.append(
             f"sink_received {sink_received} < engine_written {written} (lost {deliver_short})"