Give cost-stability the parent-switch ceiling its criterion described

The scenario's success criterion existed only as a comment: count
"Parent switched" in n04's log, expect at most 5. Nothing read it, and
the scenario declared no assertions at all, so its exit code could only
report that the mesh came up and nothing crashed.

Add a max_parent_switches assertion and wire that criterion to it.

The assertion takes an optional node scope, which is the part that
matters. The existing sibling counts mesh-wide, and a criterion written
about one node's log is a different quantity from the sum over every
node: across archived runs of this scenario n04 is 1 or 2 while the
mesh-wide total ranges 3 to 6. Asserting the sum against a per-node
threshold would check something other than what was specified, and at
this threshold would also have failed several runs that were fine.

Counting nothing must not read as stability. A node id absent from the
topology fails explicitly rather than matching zero log lines and
sailing under the ceiling, and a scenario that declares a parent-switch
assertion while setting a log level that suppresses the events it counts
is rejected at load rather than passing vacuously after a full run. The
parse rejects a mistyped key, a missing, negative, boolean or
non-integer ceiling, and a node written empty or as a non-string -- that
last one because YAML renders a bare `node:` as null, which would
silently revert to the mesh-wide count.

The scenario comment now records what the threshold is worth against
the 11 completed archived runs rather than a single sample: 5 is well
above anything observed, the daemon's own parent hold-down caps
switches near 6 per run, and nearly every switch lands during initial
tree formation before any mutation fires. So this catches only
near-pathological reparenting. Tightening to 3 would make it a real
detector, but that changes the stated criterion rather than fixing it,
so it is left as a decision.

Verified against a live run: the assertion evaluates n04 at 2 against
mesh-wide 4 and passes, and with the ceiling temporarily set to 0 the
scenario exits 3, the assertion-failure code, which it previously could
not reach.

testing/chaos/scenarios/cost-stability.yaml

@@ -17,8 +17,32 @@
 # stay within the 20% hysteresis band. n04 should pick one parent
 # and mostly stick with it — flapping indicates insufficient hysteresis.
 #
-# Validation: count "Parent switched" in n04 logs, expect <= 5 switches
+# Validation: n04 should switch parent at most 5 times. This is now the
-# over the full 180s duration.
+# max_parent_switches assertion below, scoped to n04. It previously
+# existed only as this comment and was checked by nothing, so the
+# scenario could not fail on it.
+#
+# The threshold of 5 is the one this comment always named, kept
+# deliberately rather than re-derived. What the number is worth, from 11
+# completed archived runs of this scenario: n04 was 1 or 2 every time and
+# never reached 3, while the mesh-wide total ranged 3 to 6. Two
+# consequences worth knowing before trusting a green result here.
+#
+# First, 5 is a long way above anything observed, so this catches only
+# near-pathological reparenting. The daemon's own 30s parent hold-down
+# caps non-mandatory switches at roughly 6 per 180s run, so the band
+# where this can fail is narrow. A hysteresis regression that added two
+# or three switches would pass.
+#
+# Second, the count spans the container's whole life, not just the
+# mutation phase: in the archived runs nearly every n04 switch lands in
+# the first 100ms during initial tree formation, before the first netem
+# mutation fires. So this number is dominated by convergence transients
+# rather than by the hysteresis behaviour the scenario is named for.
+#
+# Tightening to 3 would make it a real detector against the observed
+# distribution. That is a change to the stated criterion rather than a
+# fix, so it is left as a decision rather than taken here.
 
 scenario:
   name: "cost-stability"
@@ -67,6 +91,11 @@ traffic:
   duration_secs: {min: 5, max: 15}
   parallel_streams: 2
 
+assertions:
+  max_parent_switches:
+    node: n04
+    max_total: 5
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/sim/assertions.py

@@ -18,7 +18,11 @@ import logging
 from dataclasses import dataclass
 
 from .control import snapshot_all_bloom
-from .scenario import BloomSendRateAssertion, MinParentSwitchesAssertion
+from .scenario import (
+    BloomSendRateAssertion,
+    MaxParentSwitchesAssertion,
+    MinParentSwitchesAssertion,
+)
 from .topology import SimTopology
 
 log = logging.getLogger(__name__)
@@ -131,6 +135,41 @@ class BloomSendRateMonitor:
         )
 
 
+def evaluate_max_parent_switches(
+    cfg: MaxParentSwitchesAssertion,
+    parent_switch_count: int,
+    scope: str,
+) -> AssertionOutcome:
+    """Stability ceiling on parent switches over the run.
+
+    ``scope`` describes what was counted and appears in the message, so a
+    reader can tell a per-node result from a mesh-wide one. Resolving a
+    per-node scope to a real node is the caller's job, and so is failing
+    loudly when it cannot: an unresolvable node would count zero switches
+    and sail under any ceiling without having observed anything.
+    """
+    if parent_switch_count <= cfg.max_total:
+        return AssertionOutcome(
+            name="max_parent_switches",
+            passed=True,
+            detail=(
+                f"PASS max_parent_switches: {parent_switch_count} switches "
+                f"({scope}) <= ceiling {cfg.max_total}"
+            ),
+        )
+    return AssertionOutcome(
+        name="max_parent_switches",
+        passed=False,
+        detail=(
+            f"FAIL max_parent_switches: {parent_switch_count} switches "
+            f"({scope}) > ceiling {cfg.max_total} — the tree is reparenting "
+            f"more than the hysteresis band should allow. Check whether a "
+            f"cost change smaller than the hysteresis margin is still "
+            f"triggering a switch."
+        ),
+    )
+
+
 def evaluate_min_parent_switches(
     cfg: MinParentSwitchesAssertion,
     parent_switch_count: int,
@@ -147,7 +186,7 @@ def evaluate_min_parent_switches(
             passed=True,
             detail=(
                 f"PASS min_parent_switches: {parent_switch_count} switches "
-                f">= floor {cfg.min_total}"
+                f"(mesh-wide) >= floor {cfg.min_total}"
             ),
         )
     return AssertionOutcome(
@@ -155,7 +194,8 @@ def evaluate_min_parent_switches(
         passed=False,
         detail=(
             f"FAIL min_parent_switches: {parent_switch_count} switches "
-            f"< floor {cfg.min_total} — harness did not induce sufficient "
+            f"(mesh-wide) < floor {cfg.min_total} — harness did not induce "
+            f"sufficient "
             f"parent flapping; bloom-rate assertion would be trivially "
             f"true. Check tree-snapshot-warmup.json: did the expected "
             f"node win the root election?"

testing/chaos/sim/runner.py

@@ -12,7 +12,12 @@ import sys
 import time
 from datetime import datetime
 
-from .assertions import AssertionOutcome, BloomSendRateMonitor, evaluate_min_parent_switches
+from .assertions import (
+    AssertionOutcome,
+    BloomSendRateMonitor,
+    evaluate_max_parent_switches,
+    evaluate_min_parent_switches,
+)
 from .compose import generate_compose
 from .config_gen import write_configs
 from .control import snapshot_all_congestion, snapshot_all_mmp, snapshot_all_trees
@@ -70,6 +75,46 @@ class SimRunner:
         self.bloom_rate_monitor: BloomSendRateMonitor | None = None
         self.assertion_outcomes: list[AssertionOutcome] = []
 
+    def _evaluate_max_parent_switches(
+        self, cfg, parent_switches: list[tuple[str, str]]
+    ) -> AssertionOutcome:
+        """Count parent switches in the configured scope and apply the ceiling.
+
+        A per-node scope is resolved through the topology and fails
+        explicitly if the node id is not in it. That is the whole reason
+        this lives here rather than in assertions.py: filtering log lines
+        by an unknown container name yields zero matches, and zero
+        trivially satisfies a ceiling, so a typo in ``node:`` would turn
+        the assertion into an unconditional pass.
+
+        Note the limit of that guard. It covers an unresolvable node id
+        and nothing else. A node that is in the topology but whose log is
+        empty, or whose log level suppressed the event, still counts zero
+        and still passes. Only the misspelling is caught here; the log
+        level is guarded at load time, and an empty log for a live node
+        is not guarded at all.
+        """
+        if cfg.node is None:
+            return evaluate_max_parent_switches(
+                cfg, len(parent_switches), "mesh-wide"
+            )
+
+        if cfg.node not in self.topology.nodes:
+            known = ", ".join(sorted(self.topology.nodes))
+            return AssertionOutcome(
+                name="max_parent_switches",
+                passed=False,
+                detail=(
+                    f"FAIL max_parent_switches: node '{cfg.node}' is not in "
+                    f"this topology (nodes: {known}). Nothing was counted, so "
+                    f"the ceiling was never actually tested."
+                ),
+            )
+
+        source = self.topology.container_name(cfg.node)
+        count = sum(1 for src, _ in parent_switches if src == source)
+        return evaluate_max_parent_switches(cfg, count, f"node {cfg.node}")
+
     @staticmethod
     def _resolve_output_dir(scenario: Scenario) -> str:
         """Build a timestamped output directory path.
@@ -533,7 +578,10 @@ class SimRunner:
             print(result.summary())
 
             # Log-derived assertions (evaluated after analyze_logs so
-            # parent_switches and similar are populated).
+            # parent_switches and similar are populated). See
+            # _evaluate_max_parent_switches for why the per-node variant
+            # resolves its node against the topology rather than filtering
+            # optimistically.
             mps_cfg = self.scenario.assertions.min_parent_switches
             if mps_cfg is not None:
                 outcome = evaluate_min_parent_switches(
@@ -545,6 +593,17 @@ class SimRunner:
                 else:
                     log.error("%s", outcome.detail)
 
+            xps_cfg = self.scenario.assertions.max_parent_switches
+            if xps_cfg is not None:
+                outcome = self._evaluate_max_parent_switches(
+                    xps_cfg, result.parent_switches
+                )
+                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")

testing/chaos/sim/scenario.py

@@ -203,12 +203,28 @@ class MinParentSwitchesAssertion:
     min_total: int = 1
 
 
+@dataclass
+class MaxParentSwitchesAssertion:
+    """Stability ceiling: fail if parent switches exceed ``max_total``.
+
+    ``node`` scopes the count to a single node id (e.g. ``n04``) instead
+    of the mesh-wide total. The distinction is not cosmetic: a criterion
+    written about one node's log is a different number from the sum over
+    every node, and checking the sum against a per-node threshold
+    silently tests something other than what was specified.
+    """
+
+    max_total: int = 0
+    node: str | None = None
+
+
 @dataclass
 class AssertionsConfig:
     """Optional post-run assertions evaluated against control-socket data."""
 
     bloom_send_rate: BloomSendRateAssertion | None = None
     min_parent_switches: MinParentSwitchesAssertion | None = None
+    max_parent_switches: MaxParentSwitchesAssertion | None = None
 
 
 @dataclass
@@ -283,12 +299,15 @@ _SECTION_KEYS = {
     "ingress": {"enabled", "tiers_kbps", "burst_bytes"},
     "link_swap": {"enabled", "interval_secs", "policies", "edges"},
     "link_swap.edges[]": {"edge", "policy"},
-    "assertions": {"bloom_send_rate", "min_parent_switches"},
+    "assertions": {
+        "bloom_send_rate", "min_parent_switches", "max_parent_switches",
+    },
     "logging": {"rust_log", "output_dir"},
 }
 _ASSERTION_KEYS = {
     "bloom_send_rate": {"window_secs", "max_per_node"},
     "min_parent_switches": {"min_total"},
+    "max_parent_switches": {"max_total", "node"},
 }
 _NETEM_POLICY_KEYS = {
     "delay_ms", "jitter_ms", "loss_pct", "duplicate_pct", "reorder_pct",
@@ -525,6 +544,51 @@ def load_scenario(path: str) -> Scenario:
         s.assertions.min_parent_switches = MinParentSwitchesAssertion(
             min_total=int(mps.get("min_total", 1)),
         )
+    if "max_parent_switches" in asrt:
+        xps = asrt["max_parent_switches"]
+        _reject_unknown(
+            xps, _ASSERTION_KEYS["max_parent_switches"],
+            "assertions.max_parent_switches",
+        )
+        if "max_total" not in xps:
+            raise ValueError(
+                "assertions.max_parent_switches: max_total is required "
+                "(a defaulted ceiling would assert an arbitrary number)"
+            )
+        max_total = xps["max_total"]
+        # bool is a subclass of int, and `max_total: yes` would coerce to 1
+        # -- a ceiling low enough to change the verdict, arrived at by typo.
+        if isinstance(max_total, bool) or not isinstance(max_total, int):
+            raise ValueError(
+                f"assertions.max_parent_switches: max_total must be a "
+                f"non-negative integer, got {max_total!r}"
+            )
+        if max_total < 0:
+            raise ValueError(
+                f"assertions.max_parent_switches: max_total must be "
+                f"non-negative, got {max_total}"
+            )
+        # Distinguish "key absent" from "key present but empty". Only the
+        # first means mesh-wide. YAML renders a bare `node:` as None, and
+        # treating that as absent would silently swap a per-node ceiling
+        # for a mesh-wide one -- the exact conflation this assertion was
+        # added to stop, and a live flake rather than a theoretical one:
+        # archived runs of this scenario reach 6 mesh-wide against an n04
+        # maximum of 2.
+        node = None
+        if "node" in xps:
+            node = xps["node"]
+            if not isinstance(node, str) or not node.strip():
+                raise ValueError(
+                    f"assertions.max_parent_switches: node must be a node id "
+                    f"string such as 'n04', got {node!r}. Omit the key "
+                    f"entirely for the mesh-wide total."
+                )
+            node = node.strip()
+        s.assertions.max_parent_switches = MaxParentSwitchesAssertion(
+            max_total=max_total,
+            node=node,
+        )
 
     # Logging section
     lg = raw.get("logging", {})
@@ -541,8 +605,43 @@ def load_scenario(path: str) -> Scenario:
     return s
 
 
+_SUPPRESSING_LOG_LEVELS = ("off", "error", "warn")
+
+
+def _validate_parent_switch_observability(s: Scenario):
+    """Refuse a parent-switch assertion the log level cannot observe.
+
+    The events these assertions count are emitted at ``info``. A scenario
+    that declares one while setting ``logging.rust_log`` to ``warn`` or
+    below counts zero switches: the minimum assertion then fails for the
+    wrong reason, and the maximum assertion passes without observing
+    anything, which is the failure mode the whole assertion effort exists
+    to remove. Catch it at load rather than after the run.
+
+    Deliberately conservative. It rejects only a default level that is
+    demonstrably too coarse, and says nothing about per-target directives
+    such as ``info,fips::node=debug``, which raise verbosity rather than
+    lower it.
+    """
+    if (
+        s.assertions.min_parent_switches is None
+        and s.assertions.max_parent_switches is None
+    ):
+        return
+    default_level = s.logging.rust_log.split(",")[0].strip().lower()
+    if default_level in _SUPPRESSING_LOG_LEVELS:
+        raise ValueError(
+            f"logging.rust_log is '{s.logging.rust_log}', whose default level "
+            f"'{default_level}' suppresses the info-level parent-switch events "
+            f"that a parent-switch assertion counts. The assertion would see "
+            f"zero switches regardless of what the tree did. Use 'info' or "
+            f"more verbose, or drop the assertion."
+        )
+
+
 def _validate(s: Scenario):
     """Validate scenario constraints."""
+    _validate_parent_switch_observability(s)
     if s.topology.num_nodes < 2:
         raise ValueError("topology.num_nodes must be >= 2")
     if s.topology.num_nodes > 250: