Give congestion-stress the assertions its criteria described

The scenario listed four success criteria "verified via post-run
congestion snapshot". Nothing read the snapshot, so the scenario could
not fail on any of them.

Add a congestion_signals assertion taking a floor on the number of nodes
reporting each counter, and encode three of the four. The floors are one
node each because that is what the criteria say; tightening them to the
counts recently observed would assert something nobody wrote down.

The fourth criterion is not encoded, and that is the finding rather than
an omission. No node has reported a non-zero kernel_drop_events in any of
the 182 archived runs, so asserting it would red the scenario
permanently. It is recorded at the site as an unmet criterion and a
coverage gap: the transport drop-detection path the scenario names as its
second signal is exercised by nothing.

Two things about the corpus are worth carrying, both recorded in the
scenario. Only six archived runs carry a status.txt and are therefore
provably completed; all six meet the three encoded criteria with five to
nine nodes reporting each signal. The 176 older runs meet none of them,
and they are not invalid samples: each reached teardown far enough to
write an analysis.txt, and ECN landed before all but one of them. What
changed on 2026-07-22 is not established, since neither the scenario nor
netem.py, traffic.py or control.py has been touched.

Count the nodes reporting a signal rather than the magnitude of any one
counter, since a single node with a large count would satisfy a magnitude
test while proving the signal never propagated. A missing snapshot fails
rather than reading as an absence of congestion.

testing/chaos/scenarios/congestion-stress.yaml

@@ -91,6 +91,44 @@ traffic:
 node_churn:
   enabled: false
 
+# Three of the four success criteria above, encoded. Until now all four
+# existed only as that comment and were checked by nothing, so the scenario
+# could not fail on any of them.
+#
+# The floors are 1 node each because that is what the criteria say ("on at
+# least one forwarding node", "on transit nodes", "on destination nodes").
+# They are deliberately not tightened to the observed counts: the criterion
+# is the spec, and a floor invented from six runs would assert something
+# nobody wrote down.
+#
+# What the floors are worth, from the archived corpus. The six runs that
+# carry a status.txt -- the only ones provably completed, all between
+# 2026-07-22 22:14 and 2026-07-23 02:02 -- meet all three with 5 to 9 nodes
+# reporting each signal, so the margin over a floor of 1 is comfortable.
+# The 176 older runs meet none of them. Those older runs did reach teardown
+# (each wrote an analysis.txt), and ECN landed 2026-03-05, before all but
+# one of them, so they are valid runs that observed no congestion rather
+# than runs that died early. What changed on 2026-07-22 is not established:
+# this file has not been touched since it was created, and neither have
+# netem.py, traffic.py or control.py. Worth knowing before trusting a green
+# result here, and worth its own investigation.
+assertions:
+  congestion_signals:
+    min_nodes_detected: 1
+    min_nodes_ce_forwarded: 1
+    min_nodes_ce_received: 1
+
+# The fourth criterion, "kernel_drop_events > 0 on at least one node", is
+# NOT asserted, because the implementation does not meet it: across all 182
+# archived runs of this scenario, including the six that meet the other
+# three, no node has ever reported a non-zero kernel_drop_events. Asserting
+# it would red the scenario permanently, and asserting a weakened version
+# would assert nothing. It stands here as an unmet criterion: either the
+# 4 KB recv_buf_size no longer provokes SO_RXQ_OVFL under this traffic, or
+# the counter does not reach the snapshot. Until that is settled it is a
+# coverage gap, because the transport drop-detection path this scenario
+# names as signal 2 is exercised by nothing.
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/sim/assertions.py

@@ -20,6 +20,7 @@ from dataclasses import dataclass
 from .control import snapshot_all_bloom
 from .scenario import (
     BloomSendRateAssertion,
+    CongestionSignalsAssertion,
     MaxErrorsAssertion,
     MaxParentSwitchesAssertion,
     MinParentSwitchesAssertion,
@@ -171,6 +172,70 @@ def evaluate_max_parent_switches(
     )
 
 
+_CONGESTION_FLOORS = (
+    ("min_nodes_detected", "congestion_detected"),
+    ("min_nodes_ce_forwarded", "ce_forwarded"),
+    ("min_nodes_ce_received", "ce_received"),
+)
+
+
+def evaluate_congestion_signals(
+    cfg: CongestionSignalsAssertion,
+    snapshot: dict | None,
+) -> AssertionOutcome:
+    """Floors on how many nodes observed each congestion counter.
+
+    ``snapshot`` is the final congestion snapshot keyed by node id. None
+    means the snapshot never ran, which fails: a missing snapshot and a
+    mesh that observed no congestion produce the same zero counts, and
+    treating them alike is how an assertion comes to pass on an absence
+    of evidence.
+    """
+    if not snapshot:
+        return AssertionOutcome(
+            name="congestion_signals",
+            passed=False,
+            detail=(
+                "FAIL congestion_signals: no final congestion snapshot was "
+                "taken, so no node was observed at all. This is a harness "
+                "failure, not a statement about congestion."
+            ),
+        )
+
+    parts, failures = [], []
+    for attr, counter in _CONGESTION_FLOORS:
+        floor = getattr(cfg, attr)
+        if floor is None:
+            continue
+        hits = sorted(
+            nid for nid, data in snapshot.items()
+            if (data.get("congestion") or {}).get(counter, 0) > 0
+        )
+        parts.append(f"{counter}: {len(hits)} node(s) >0 (floor {floor})")
+        if len(hits) < floor:
+            failures.append(
+                f"{counter} non-zero on {len(hits)} node(s), need {floor}"
+            )
+        else:
+            parts[-1] += f" [{', '.join(hits)}]"
+
+    summary = "; ".join(parts)
+    if failures:
+        return AssertionOutcome(
+            name="congestion_signals",
+            passed=False,
+            detail=(
+                f"FAIL congestion_signals: {'; '.join(failures)} — across "
+                f"{len(snapshot)} node(s) sampled. Full counts: {summary}"
+            ),
+        )
+    return AssertionOutcome(
+        name="congestion_signals",
+        passed=True,
+        detail=f"PASS congestion_signals: {summary}",
+    )
+
+
 def evaluate_max_errors(
     cfg: MaxErrorsAssertion,
     errors: list[tuple[str, str]],

testing/chaos/sim/runner.py

@@ -15,6 +15,7 @@ from datetime import datetime
 from .assertions import (
     AssertionOutcome,
     BloomSendRateMonitor,
+    evaluate_congestion_signals,
     evaluate_max_errors,
     evaluate_max_parent_switches,
     evaluate_min_parent_switches,
@@ -75,6 +76,10 @@ class SimRunner:
         # Post-run assertion monitors (sampled near end of run).
         self.bloom_rate_monitor: BloomSendRateMonitor | None = None
         self.assertion_outcomes: list[AssertionOutcome] = []
+        # Set by the final snapshot. None means the snapshot never ran,
+        # which the congestion assertion must treat as a failure rather
+        # than as an absence of congestion.
+        self.final_congestion: dict | None = None
 
     def _evaluate_max_parent_switches(
         self, cfg, parent_switches: list[tuple[str, str]]
@@ -617,6 +622,17 @@ class SimRunner:
                 else:
                     log.error("%s", outcome.detail)
 
+            cong_cfg = self.scenario.assertions.congestion_signals
+            if cong_cfg is not None:
+                outcome = evaluate_congestion_signals(
+                    cong_cfg, self.final_congestion
+                )
+                self.assertion_outcomes.append(outcome)
+                if outcome.passed:
+                    log.info("%s", outcome.detail)
+                else:
+                    log.error("%s", outcome.detail)
+
             # Write assertion outcomes
             if self.assertion_outcomes:
                 assertions_path = os.path.join(self.output_dir, "assertions.txt")
@@ -677,6 +693,11 @@ class SimRunner:
         tree_path = os.path.join(self.output_dir, f"tree-snapshot-{label}.json")
         mmp_path = os.path.join(self.output_dir, f"mmp-snapshot-{label}.json")
         congestion_path = os.path.join(self.output_dir, f"congestion-snapshot-{label}.json")
+        # Retained for the congestion assertion, which needs the same
+        # responses the file gets. Reading the file back would let a write
+        # failure present as an assertion that saw nothing.
+        if label == "final":
+            self.final_congestion = congestion_snap
         os.makedirs(self.output_dir, exist_ok=True)
         with open(tree_path, "w") as f:
             json.dump(tree_snap, f, indent=2)

testing/chaos/sim/scenario.py

@@ -218,6 +218,26 @@ class MaxParentSwitchesAssertion:
     node: str | None = None
 
 
+@dataclass
+class CongestionSignalsAssertion:
+    """Floors on how many nodes observed each congestion signal.
+
+    Each floor is the number of nodes whose final congestion snapshot
+    reports a non-zero counter, not the counter's own magnitude: the
+    scenario's written criteria are about *whether* a signal reached a
+    class of node, and a single node with a huge count would satisfy a
+    magnitude test while proving the signal never propagated.
+
+    A floor left unset is not asserted. At least one must be set, since a
+    block that asserts nothing is the failure this assertion exists to
+    remove.
+    """
+
+    min_nodes_detected: int | None = None
+    min_nodes_ce_forwarded: int | None = None
+    min_nodes_ce_received: int | None = None
+
+
 @dataclass
 class MaxErrorsAssertion:
     """Ceiling on ERROR-level lines across every node's log.
@@ -248,6 +268,7 @@ class AssertionsConfig:
     min_parent_switches: MinParentSwitchesAssertion | None = None
     max_parent_switches: MaxParentSwitchesAssertion | None = None
     max_errors: MaxErrorsAssertion | None = None
+    congestion_signals: CongestionSignalsAssertion | None = None
 
 
 @dataclass
@@ -324,7 +345,7 @@ _SECTION_KEYS = {
     "link_swap.edges[]": {"edge", "policy"},
     "assertions": {
         "bloom_send_rate", "min_parent_switches", "max_parent_switches",
-        "max_errors",
+        "max_errors", "congestion_signals",
     },
     "logging": {"rust_log", "output_dir"},
 }
@@ -333,6 +354,9 @@ _ASSERTION_KEYS = {
     "min_parent_switches": {"min_total"},
     "max_parent_switches": {"max_total", "node"},
     "max_errors": {"max_total"},
+    "congestion_signals": {
+        "min_nodes_detected", "min_nodes_ce_forwarded", "min_nodes_ce_received",
+    },
 }
 _NETEM_POLICY_KEYS = {
     "delay_ms", "jitter_ms", "loss_pct", "duplicate_pct", "reorder_pct",
@@ -642,6 +666,37 @@ def load_scenario(path: str) -> Scenario:
         # run means, and a scenario that has to opt in is a scenario that
         # can forget to.
         s.assertions.max_errors = MaxErrorsAssertion()
+    if "congestion_signals" in asrt:
+        cs = asrt["congestion_signals"]
+        _reject_unknown(
+            cs, _ASSERTION_KEYS["congestion_signals"],
+            "assertions.congestion_signals",
+        )
+        floors = {}
+        for key in _ASSERTION_KEYS["congestion_signals"]:
+            if key not in cs:
+                continue
+            val = cs[key]
+            if isinstance(val, bool) or not isinstance(val, int):
+                raise ValueError(
+                    f"assertions.congestion_signals.{key}: must be a positive "
+                    f"integer number of nodes, got {val!r}"
+                )
+            if val < 1:
+                raise ValueError(
+                    f"assertions.congestion_signals.{key}: must be at least 1, "
+                    f"got {val}. A floor of 0 is satisfied by a mesh that "
+                    f"observed nothing, which is the case this assertion exists "
+                    f"to catch; omit the key instead."
+                )
+            floors[key] = val
+        if not floors:
+            raise ValueError(
+                "assertions.congestion_signals: set at least one floor "
+                "(min_nodes_detected, min_nodes_ce_forwarded, "
+                "min_nodes_ce_received); a block with none asserts nothing"
+            )
+        s.assertions.congestion_signals = CongestionSignalsAssertion(**floors)
 
     # Logging section
     lg = raw.get("logging", {})