fix(harness): assert empty claims per MESSAGE, not per second (BACKLOG #1101)

docs/BACKLOG.md

@@ -6746,7 +6746,7 @@ against the anchor.
 
 ## 1101. the connscale empty_claims_monotonic SLO reports runner contention as an engine defect
 
-> 🔢 **Filed 2026-08-07 - not started. Reproduced, not inferred.** Value **4/10** · Difficulty **2/10**.
+> ✅ **SHIPPED 2026-08-08 - the SLO now reads empty claims PER MESSAGE, and the latent `claim_mode` grouping defect went with it.** The asserted metric is `empty_claims_per_msg`, computed as the ratio of two rates taken over the SAME first-to-last in-hold samples, so the span cancels algebraically and the quantity is exactly `Δempty_claims / Δread` -- there is no wall clock left for runner contention or a mid-hold reload stall to move. `_monotonic_slo` now groups by `(sweep_mode, claim_mode)` rather than `sweep_mode` alone, so a profile combining `per_lane` and `pooled` can no longer chain-compare across claim modes. The per-second numbers are retained in the report as the operator-facing figures; they are simply no longer what gates a merge. **Verified against the failure mode, not just for green:** eight tests pin the invariance property AND the still-detects-a-real-regression property together, and both were shown to go RED under mutation - reverting the grouping fails 3, restoring the per-second metric fails 2. A metric that never fires would have passed a stability test alone, which is why the two are pinned as a pair. **Not done, and deliberately:** gating `reload_seconds` directly was raised below as a conditional ("if that cost is worth gating") and is a separate judgement, not part of this fix. Original filing follows. Value **4/10** · Difficulty **2/10**.
 > `tests/test_connscale_smoke.py:170` asserts `empty_claims_monotonic`: the N=24 empty-claim **rate per
 > second** must be at least 0.75x the N=12 rate. The metric has wall-clock in its denominator and a
 > deliberately un-gated O(N) probe in its numerator's way, so **CPU contention alone flips it red with

harness/load/connscale/batchbox.py

@@ -353,6 +353,7 @@ def _engine_gauge(key: str) -> int:
         empty_claims_per_s=0.0,
         idle_poll_per_s=0.0,
         wake_fanout_per_s=0.0,
+        empty_claims_per_msg=None,
         fd_count_peak=None,
         reload_seconds=None,
         ack_p50_ms=0.0,
@@ -409,6 +410,7 @@ def _failed_cell_record(cell: BatchCell, detail: str) -> ConnScaleRecord:
         empty_claims_per_s=0.0,
         idle_poll_per_s=0.0,
         wake_fanout_per_s=0.0,
+        empty_claims_per_msg=None,
         fd_count_peak=None,
         reload_seconds=None,
         ack_p50_ms=0.0,

harness/load/connscale/report.py

@@ -100,6 +100,14 @@ class ConnScaleRecord:
     wake_fanout_per_s: (
         float  # the per-commit thundering-herd cost (the herd slope vs N is read here)
     )
+    # Empty claims PER MESSAGE absorbed, over the same first→last in-hold window as the rates above.
+    # BACKLOG #1101: the per-SECOND form has wall clock in its denominator, so anything that slows the
+    # run — CPU contention on a shared CI runner, or the O(N) reload probe firing mid-hold — collapses
+    # it without the engine changing. Per-message is the quantity wall #3 actually means (the herd size
+    # per commit) and is immune to that: numerator and denominator are both deltas over the SAME
+    # samples, so the span cancels algebraically rather than by assumption. None when the window
+    # absorbed no messages, in which case the ratio is undefined and must not be invented as 0.
+    empty_claims_per_msg: float | None
 
     # --- wall #4: FD / socket count ---
     fd_count_peak: int | None  # None when the OS probe couldn't read the PID
@@ -195,6 +203,14 @@ def to_json_dict(self) -> dict[str, object]:
                 # SEPARATED (critic must-change #3): idle-poll re-SELECTs vs the per-commit herd.
                 "idle_poll_per_s": round(self.idle_poll_per_s, 2),
                 "wake_fanout_per_s": round(self.wake_fanout_per_s, 2),
+                # The ASSERTED form (BACKLOG #1101). The per-second numbers above are operator-facing
+                # and carry wall clock; this one is what the monotonicity SLO reads, because it does
+                # not move when the runner is merely slow. None when no messages were absorbed.
+                "total_per_msg": (
+                    None
+                    if self.empty_claims_per_msg is None
+                    else round(self.empty_claims_per_msg, 3)
+                ),
             },
             "wall4_fd": {"count_peak": self.fd_count_peak},
             "wall5_reload": {"seconds": self.reload_seconds},

harness/load/connscale/runner.py

@@ -742,6 +742,13 @@ def _build_record(
     # (msg/s actually absorbed/delivered vs the OFFERED aggregate rate) — the A/B non-regression guard.
     achieved_read_per_s, achieved_written_per_s = _throughput_rates(samples)
 
+    # Wall #3, the ASSERTED form: empty claims per MESSAGE ABSORBED (BACKLOG #1101). Both rates above
+    # are Δ/span over the same first→last in-hold samples, so dividing them cancels the span exactly and
+    # leaves Δclaims/Δread — no wall clock, and therefore nothing for runner contention or a mid-hold
+    # reload stall to move. This is what the monotonicity SLO reads; the per-second form is retained for
+    # the report because it is the operator-facing number.
+    empty_per_msg = _empty_claims_per_msg(total_per_s, achieved_read_per_s)
+
     # Wall #4 + footprint: handle peak + CPU-seconds + working set, drained from the side map (each
     # None where the OS probe couldn't read).
     proc = _drain_proc(samples)
@@ -772,6 +779,7 @@ def _build_record(
         empty_claims_per_s=total_per_s,
         idle_poll_per_s=idle_per_s,
         wake_fanout_per_s=wake_per_s,
+        empty_claims_per_msg=empty_per_msg,
         fd_count_peak=proc.handles_peak,
         reload_seconds=reload_seconds,
         ack_p50_ms=ack.p50_ms,
@@ -898,6 +906,23 @@ def _empty_claim_rates(samples: list[EngineSample]) -> tuple[float, float, float
     return max(0.0, total), max(0.0, idle), max(0.0, wake)
 
 
+def _empty_claims_per_msg(total_per_s: float, achieved_read_per_s: float) -> float | None:
+    """Empty claims per message absorbed — the wall-clock-free form of wall #3 (BACKLOG #1101).
+
+    Both inputs are Δ/span over the SAME first→last in-hold samples, so ``span`` cancels and this is
+    exactly ``Δempty_claims / Δread``. That is why it survives runner contention: slowing the run
+    scales numerator and denominator identically. The per-SECOND form does not — it keeps ``span`` in
+    the denominator, so a slow arm reads as an improvement, which is the defect #1101 records.
+
+    Returns ``None`` when no messages were absorbed in the window. The ratio is genuinely undefined
+    there and returning 0.0 would be a fabricated reading that then chains through the monotonicity
+    comparison as a real one.
+    """
+    if achieved_read_per_s <= 0.0:
+        return None
+    return max(0.0, total_per_s / achieved_read_per_s)
+
+
 def _throughput_rates(samples: list[EngineSample]) -> tuple[float, float]:
     """Achieved (read/s, written/s) over the window, first→last sample (same span as the empty-claim
     rates), so both arms are measured identically for the A/B non-regression guard."""
@@ -1058,7 +1083,10 @@ def _evaluate_slos(profile: ConnScaleProfile, records: list[ConnScaleRecord]) ->
         out.append(_monotonic_slo("fd_count_monotonic", records, lambda r: r.fd_count_peak))
     if slo.empty_claims_monotonic:
         out.append(
-            _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_s)
+            # PER MESSAGE, not per second (BACKLOG #1101). The per-second form put wall clock in the
+            # denominator, so CPU contention alone flipped this red with no engine change — measured
+            # 0.451 to 2.49 across four replicates of one commit on one box.
+            _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_msg)
         )
     return out
 
@@ -1081,11 +1109,16 @@ def _monotonic_slo(  # type: ignore[no-untyped-def]
     skipped, not failed."""
     ok = True
     detail_parts: list[str] = []
-    by_mode: dict[str, list[ConnScaleRecord]] = {}
+    # Group by (sweep_mode, claim_mode), NOT sweep_mode alone (BACKLOG #1101). Chaining prev_val across
+    # claim modes compares per_lane against pooled, and compare.py:22-25 states pooled's empty-claim
+    # rate SHOULD be materially lower — so a correct engine would fail this the moment a profile set
+    # claim_modes = ["per_lane", "pooled"]. No shipped profile does, which is the only reason it has
+    # never fired; the grouping is wrong independently of that.
+    by_mode: dict[tuple[str, str], list[ConnScaleRecord]] = {}
     for r in records:
-        by_mode.setdefault(r.sweep_mode, []).append(r)
+        by_mode.setdefault((r.sweep_mode, r.claim_mode), []).append(r)
     floor = 1.0 - tolerance
-    for mode, rs in by_mode.items():
+    for (mode, claim_mode), rs in by_mode.items():
         ordered = sorted(rs, key=lambda r: r.count)
         prev_val: float | None = None
         for r in ordered:
@@ -1095,8 +1128,9 @@ def _monotonic_slo(  # type: ignore[no-untyped-def]
             v = float(val)
             if prev_val is not None and v < prev_val * floor:
                 ok = False
+                label = mode if claim_mode == "per_lane" else f"{mode}/{claim_mode}"
                 detail_parts.append(
-                    f"{mode}@N={r.count}: {v:.1f} < prior {prev_val:.1f} * {floor:.2f}"
+                    f"{label}@N={r.count}: {v:.3g} < prior {prev_val:.3g} * {floor:.2f}"
                 )
             prev_val = v
     observed = "monotonic" if ok else "; ".join(detail_parts)

tests/test_connscale_batch.py

@@ -236,6 +236,7 @@ def _rec(
         empty_claims_per_s=0.0,
         idle_poll_per_s=0.0,
         wake_fanout_per_s=0.0,
+        empty_claims_per_msg=None,
         fd_count_peak=100,
         reload_seconds=None,
         ack_p50_ms=ack_p50,

tests/test_connscale_compare.py

@@ -50,6 +50,7 @@ def _record(
         empty_claims_per_s=idle_poll,
         idle_poll_per_s=idle_poll,
         wake_fanout_per_s=0.0,
+        empty_claims_per_msg=None,
         fd_count_peak=100,
         reload_seconds=None,
         ack_p50_ms=1.0,

tests/test_connscale_empty_claims_per_msg.py

@@ -0,0 +1,186 @@
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# Copyright (C) 2026 MessageFoundry Organization and contributors
+"""BACKLOG #1101: the empty-claims monotonicity SLO must measure the engine, not the runner.
+
+The SLO used to read ``empty_claims_per_s``, which carries wall clock in its denominator. Anything
+that slowed the run -- CPU contention on a shared CI runner, or the O(N) reload probe firing mid-hold
+and stalling commits -- collapsed the numerator while the denominator kept ticking, so the metric
+fell and the gate went red with **no engine change**. Measured on one commit, one box: four contended
+replicates spread **0.451 to 2.49** against a 0.75 floor. That is a coin flip, not a detector.
+
+The fix reads ``empty_claims_per_msg`` instead. Both inputs are deltas over the SAME first-to-last
+in-hold samples, so the span cancels and the quantity is exactly ``Δempty_claims / Δread``.
+
+**These tests exist to stop the fix from being the WRONG kind of fix.** A metric that never fails is
+not an improvement on one that fails at random, and a correction is the easiest place to skip
+measuring because it feels like it has already paid its dues. So the invariance property and the
+still-detects-a-real-regression property are pinned TOGETHER: neither alone is evidence.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from harness.load.connscale.report import ConnScaleRecord, NoLoss
+from harness.load.connscale.runner import _empty_claims_per_msg, _monotonic_slo
+
+
+def _rec(
+    mode: str,
+    count: int,
+    *,
+    per_msg: float | None,
+    claim_mode: str = "per_lane",
+) -> ConnScaleRecord:
+    """A record carrying only the fields these SLOs read; the rest are inert placeholders."""
+    return ConnScaleRecord(
+        sweep_mode=mode,
+        count=count,
+        offered_aggregate_rate=35.0,
+        sent=1000,
+        acked=1000,
+        nak=0,
+        deferred=0,
+        timeouts=0,
+        no_loss=NoLoss(True, 1000, 1000, 1000, 1000, 0, "ok"),
+        in_pipeline_peak=3,
+        drain_seconds=1.2,
+        executor_queue_depth_peak=2,
+        executor_busy_peak=1,
+        pool_wait_p50_ms=None,
+        pool_wait_p95_ms=None,
+        pool_wait_p99_ms=None,
+        pool_wait_max_ms=None,
+        pool_idle_min=None,
+        pool_size_max=None,
+        empty_claims_per_s=10.0,
+        idle_poll_per_s=4.0,
+        wake_fanout_per_s=6.0,
+        empty_claims_per_msg=per_msg,
+        fd_count_peak=100,
+        reload_seconds=None,
+        ack_p50_ms=None,
+        ack_p95_ms=None,
+        ack_p99_ms=None,
+        claim_mode=claim_mode,
+    )
+
+
+# --------------------------------------------------------------------------------------------------
+# The property the fix exists for.
+# --------------------------------------------------------------------------------------------------
+
+
+def test_per_msg_is_invariant_when_the_whole_run_is_slowed() -> None:
+    """Contention scales numerator and denominator identically, so the ratio does not move.
+
+    This is the defect reproduced arithmetically. A run that is slowed by 3x reports a third of the
+    empty claims per SECOND and a third of the messages per second -- the engine behaved identically,
+    only the clock changed.
+    """
+    fast = _empty_claims_per_msg(total_per_s=450.0, achieved_read_per_s=9.0)
+    slowed = _empty_claims_per_msg(total_per_s=150.0, achieved_read_per_s=3.0)
+
+    assert fast == pytest.approx(50.0)
+    assert slowed == pytest.approx(50.0), (
+        "per-message must not move when the run is uniformly slowed; if it does, the fix has "
+        "reproduced the very wall-clock dependence #1101 records"
+    )
+
+
+def test_per_second_would_have_moved_on_the_same_data() -> None:
+    """The negative half of the pair: show the OLD metric fails on data the new one survives.
+
+    Without this, 'the new metric is stable' is unfalsifiable -- a constant would also pass.
+    """
+    fast_per_s, slowed_per_s = 450.0, 150.0
+    floor = 0.75
+    assert slowed_per_s < fast_per_s * floor, (
+        "the old per-second metric must visibly collapse here, otherwise this fixture does not "
+        "exercise the defect at all"
+    )
+    # ...while per-message, on the identical run, is unchanged.
+    assert _empty_claims_per_msg(fast_per_s, 9.0) == _empty_claims_per_msg(slowed_per_s, 3.0)
+
+
+def test_per_msg_is_none_when_no_messages_were_absorbed() -> None:
+    """Undefined must stay undefined. Returning 0.0 would chain through the comparison as a reading."""
+    assert _empty_claims_per_msg(total_per_s=12.0, achieved_read_per_s=0.0) is None
+    assert _empty_claims_per_msg(total_per_s=0.0, achieved_read_per_s=0.0) is None
+
+
+# --------------------------------------------------------------------------------------------------
+# The property that stops this being a gate that never fires.
+# --------------------------------------------------------------------------------------------------
+
+
+def test_slo_still_fails_on_a_genuine_herd_collapse() -> None:
+    """A REAL regression -- per-message herd size dropping with N -- must still go red.
+
+    The engine claim wall #3 makes is that the per-commit herd GROWS with connection count. A drop
+    means the instrumentation or the fanout changed. The fix must not buy stability by becoming blind.
+    """
+    records = [
+        _rec("fixed_aggregate", 12, per_msg=40.0),
+        _rec("fixed_aggregate", 24, per_msg=10.0),  # collapsed, far under the 0.75 floor
+    ]
+    check = _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_msg)
+    assert not check.ok, "a genuine per-message collapse must fail the SLO"
+    assert "fixed_aggregate@N=24" in str(check.observed)
+
+
+def test_slo_passes_on_the_healthy_shape() -> None:
+    """The measured healthy readings: 39.1 at N=12 rising to 77.8 at N=24, a clean 2.0x."""
+    records = [
+        _rec("fixed_aggregate", 12, per_msg=39.1),
+        _rec("fixed_aggregate", 24, per_msg=77.8),
+    ]
+    assert _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_msg).ok
+
+
+def test_slo_skips_undefined_readings_without_failing() -> None:
+    """None is 'no reading', not 'a reading of zero'."""
+    records = [
+        _rec("fixed_aggregate", 12, per_msg=40.0),
+        _rec("fixed_aggregate", 24, per_msg=None),
+    ]
+    assert _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_msg).ok
+
+
+# --------------------------------------------------------------------------------------------------
+# The latent grouping defect, fixed in the same pass.
+# --------------------------------------------------------------------------------------------------
+
+
+def test_monotonicity_does_not_chain_across_claim_modes() -> None:
+    """Grouping by sweep_mode ALONE compares per_lane against pooled.
+
+    ``compare.py`` states pooled's empty-claim rate SHOULD be materially lower, so a CORRECT engine
+    would fail this the moment a profile set ``claim_modes = ["per_lane", "pooled"]``. No shipped
+    profile does, which is the only reason it has never fired -- the grouping is wrong regardless.
+    Ordered so that a sweep_mode-only grouping sorts pooled's low N=12 after per_lane's high N=24.
+    """
+    records = [
+        _rec("fixed_aggregate", 12, per_msg=40.0, claim_mode="per_lane"),
+        _rec("fixed_aggregate", 24, per_msg=80.0, claim_mode="per_lane"),
+        _rec("fixed_aggregate", 12, per_msg=4.0, claim_mode="pooled"),
+        _rec("fixed_aggregate", 24, per_msg=8.0, claim_mode="pooled"),
+    ]
+    check = _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_msg)
+    assert check.ok, (
+        "each claim mode rises monotonically on its own; failing here means prev_val is being "
+        f"chained across claim modes, which is the #1101 latent defect. observed={check.observed}"
+    )
+
+
+def test_a_regression_inside_one_claim_mode_is_still_caught() -> None:
+    """The grouping fix must not become a way to hide a real drop behind a second claim mode."""
+    records = [
+        _rec("fixed_aggregate", 12, per_msg=40.0, claim_mode="per_lane"),
+        _rec("fixed_aggregate", 24, per_msg=5.0, claim_mode="per_lane"),  # real collapse
+        _rec("fixed_aggregate", 12, per_msg=4.0, claim_mode="pooled"),
+        _rec("fixed_aggregate", 24, per_msg=8.0, claim_mode="pooled"),
+    ]
+    check = _monotonic_slo("empty_claims_monotonic", records, lambda r: r.empty_claims_per_msg)
+    assert not check.ok
+    assert "N=24" in str(check.observed)

tests/test_connscale_fuse.py

@@ -193,6 +193,7 @@ def _rec(
         empty_claims_per_s=0.0,
         idle_poll_per_s=0.0,
         wake_fanout_per_s=0.0,
+        empty_claims_per_msg=None,
         fd_count_peak=100,
         reload_seconds=None,
         ack_p50_ms=1.0,

tests/test_connscale_fuse_replay.py

@@ -79,6 +79,7 @@ def _record_from_json(d: dict[str, Any]) -> ConnScaleRecord:
         empty_claims_per_s=w3["total_per_s"],
         idle_poll_per_s=w3["idle_poll_per_s"],
         wake_fanout_per_s=w3["wake_fanout_per_s"],
+        empty_claims_per_msg=None,
         fd_count_peak=d["wall4_fd"]["count_peak"],
         reload_seconds=d["wall5_reload"]["seconds"],
         ack_p50_ms=w6["p50"],

tests/test_connscale_report.py

@@ -48,6 +48,7 @@ def _record(
         empty_claims_per_s=empty_per_s,
         idle_poll_per_s=idle_per_s,
         wake_fanout_per_s=wake_per_s,
+        empty_claims_per_msg=None,
         fd_count_peak=fd,
         reload_seconds=0.05,
         ack_p50_ms=1.0,