Merge branch 'maint'

testing/chaos/README.md

@@ -124,8 +124,14 @@ detection.
 - **ecn-ab-on / ecn-ab-off**: Paired scenarios with identical conditions
   (6-node tree, 10 Mbps egress, 1000 kbps ingress policing, 10ms link
   delay, 8 KB recv buffer) differing only in `ecn.enabled`.
-  `ecn-ab-test.sh` runs both and compares throughput and congestion
-  counters. Initial results: +10.2% recv throughput with ECN enabled.
+  `ecn-ab-compare.sh` runs both and prints a side-by-side of throughput
+  and congestion counters. It is a manual tool, not a test: it asserts
+  nothing and no runner invokes it. The "+10.2% recv throughput with ECN
+  enabled" figure once recorded here is not reproducible from anything on
+  disk — the script read a fixed `sim-results/ecn-ab-on/` path while the
+  runner has written timestamped directories since 2026-03-20, and no
+  ecn-ab result directory survives. The path bug is fixed; the figure is
+  left out until a run produces one.
 
 ### Ingress Traffic Control
 

testing/chaos/ecn-ab-compare.sh

@@ -1,10 +1,27 @@
 #!/usr/bin/env bash
-# ECN A/B Throughput Test
+# ECN A/B Throughput Comparison  (a manual tool, NOT a test)
 #
 # Runs two identical chaos scenarios — one with ECN enabled, one disabled —
-# and compares iperf3 throughput and congestion counter results.
+# and prints a side-by-side of iperf3 throughput and congestion counters.
 #
-# Usage: ./ecn-ab-test.sh [--seed N] [--duration N]
+# Renamed from ecn-ab-test.sh on 2026-07-23. The old name claimed a verdict
+# this has never produced: it asserts nothing, applies no threshold, and no
+# runner invokes it. Naming it a test made it look like coverage.
+#
+# It also could not have worked. It read sim-results/ecn-ab-on/... while the
+# runner has written sim-results/<timestamp>-<scenario>/ since 2026-03-20, so
+# it has found neither input for at least four months, and there is not one
+# archived ecn-ab result directory on disk. That path bug is fixed below.
+#
+# WHAT IS STILL MISSING, and why it was not added: turning this into a real
+# test needs a threshold — how much throughput ECN should buy, or how much
+# lower the congestion counters should run — and there is no corpus to derive
+# one from, precisely because the tool has never produced a kept result. A
+# number invented here would assert the author's guess. The prerequisite is a
+# calibration run set, and that is a protocol question about what ECN is
+# expected to deliver, not a harness one.
+#
+# Usage: ./ecn-ab-compare.sh [--seed N] [--duration N]
 
 set -euo pipefail
 cd "$(dirname "$0")"
@@ -24,12 +41,12 @@ echo ""
 
 # --- Run A: ECN ON ---
 echo "--- Phase A: ECN ENABLED ---"
-sudo python3 -m sim scenarios/ecn-ab-on.yaml "${EXTRA_ARGS[@]}" || true
+sudo python3 -m sim scenarios/ecn-ab-on.yaml "${EXTRA_ARGS[@]}"
 echo ""
 
 # --- Run B: ECN OFF ---
 echo "--- Phase B: ECN DISABLED ---"
-sudo python3 -m sim scenarios/ecn-ab-off.yaml "${EXTRA_ARGS[@]}" || true
+sudo python3 -m sim scenarios/ecn-ab-off.yaml "${EXTRA_ARGS[@]}"
 echo ""
 
 # --- Compare results ---
@@ -37,7 +54,27 @@ echo "=== Results ==="
 echo ""
 
 python3 - <<'PYEOF'
+import glob
 import json
+
+
+def latest(scenario, filename):
+    """Newest run directory for a scenario, or None.
+
+    The runner writes sim-results/<timestamp>-<scenario>/, so a fixed path
+    such as sim-results/ecn-ab-on/ has never matched anything. Sorting the
+    glob works because the timestamp is the leading, fixed-width component.
+    """
+    dirs = sorted(glob.glob(f"sim-results/*-{scenario}"))
+    if not dirs:
+        print(f"  no run directory found for {scenario}")
+        return None
+    path = f"{dirs[-1]}/{filename}"
+    if not glob.glob(path):
+        print(f"  {scenario}: {filename} missing from {dirs[-1]}")
+        return None
+    return path
+
 import os
 import sys
 
@@ -92,8 +129,10 @@ def print_sessions(label, sessions):
     print(f"    {'':>14}  completed={n}  incomplete={incomplete}")
     return total_recv / n
 
-on_results = load_results("sim-results/ecn-ab-on/iperf3-results.json")
-off_results = load_results("sim-results/ecn-ab-off/iperf3-results.json")
+on_path = latest("ecn-ab-on", "iperf3-results.json")
+off_path = latest("ecn-ab-off", "iperf3-results.json")
+on_results = load_results(on_path) if on_path else []
+off_results = load_results(off_path) if off_path else []
 
 on_sessions = extract_throughput(on_results)
 off_sessions = extract_throughput(off_results)
@@ -111,8 +150,10 @@ if avg_on and avg_off:
 
 # Congestion counters
 print("Congestion Counters (final snapshot):")
-for label, path in [("ECN ON", "sim-results/ecn-ab-on/congestion-snapshot-final.json"),
-                     ("ECN OFF", "sim-results/ecn-ab-off/congestion-snapshot-final.json")]:
+for label, scenario in [("ECN ON", "ecn-ab-on"), ("ECN OFF", "ecn-ab-off")]:
+    path = latest(scenario, "congestion-snapshot-final.json")
+    if path is None:
+        continue
     snap = load_congestion(path)
     if not snap:
         print(f"  {label}: no snapshot")

testing/chaos/scenarios/bottleneck-parent.yaml

@@ -87,6 +87,21 @@ traffic:
   duration_secs: {min: 5, max: 10}
   parallel_streams: 2
 
+# The two test subjects named at the top of this file, encoded. Both
+# existed only as comments and were checked by nothing.
+#
+# n06 has a Bluetooth parent (n02) and a fiber one (n03) at the same depth
+# and should take the fiber. n09 has only Bluetooth, so asserting it keeps
+# n05 is asserting that the absence of an alternative is handled without
+# thrashing, not that Bluetooth was preferred.
+#
+# From the archived corpus: across the six provably-completed runs, n06's
+# parent is n03 in all six and n09's is n05 in all six.
+assertions:
+  tree_parents:
+    n06: n03
+    n09: n05
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/churn-mixed.yaml

@@ -67,6 +67,41 @@ bandwidth:
   enabled: true
   tiers_mbps: [1, 10, 100, 1000]
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about which nodes survive the churn; it exists so that
+# a run in which the mesh never formed cannot report success, which
+# until now it could, because this scenario carried no assertions at
+# all.
+#
+# This one is calibrated rather than derived, because a scenario that
+# stops and starts nodes on purpose does not hold a single spanning
+# tree. The six provably-completed archived runs end with two or three
+# distinct roots, seven or eight of ten nodes parented, and fifteen to
+# twenty sessions. The floors below sit one step outside those ranges
+# so an unlucky run does not go red, which leaves them catching a mesh
+# that collapsed rather than one that churned. Tighten them only
+# against a larger sample than six runs.
+#
+# READ THIS BEFORE RETUNING: the numbers above describe the invocation CI
+# gates on, which is not this file's own defaults. ci-local runs it as
+# "churn-mixed --nodes 10 --duration 120", and --nodes sed-patches
+# num_nodes in a copy of this file before the scenario is loaded, so the
+# gating run is a 10-node 120-second one while a bare `chaos.sh
+# churn-mixed` runs the 20-node 600-second scenario written here. This is
+# the only scenario CI overrides that way.
+#
+# The floors hold for both but are calibrated for the smaller. A bare
+# 20-node run clears them easily: 20 answering, 1 root, 19 parented, 69
+# sessions, measured 2026-07-23. Retuning against numbers like those would
+# put them past what the 10-node gating run can reach, and CI would go red
+# while a manual run stayed green.
+assertions:
+  baseline:
+    min_nodes_reporting: 10
+    max_roots: 4
+    min_nodes_parented: 6
+    min_sessions: 10
+
 logging:
   rust_log: "debug"
   output_dir: "./sim-results"

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/scenarios/cost-avoidance.yaml

@@ -61,6 +61,21 @@ traffic:
   duration_secs: {min: 5, max: 10}
   parallel_streams: 2
 
+# The validation line at the top of this file, encoded. It existed only as
+# that comment and was checked by nothing, so the scenario could not fail
+# on the one thing it was built to test.
+#
+# From the archived corpus: of the six runs that carry a status.txt and are
+# therefore provably completed, n04's parent is n03 in all six. Among the
+# 176 older runs the picture is quite different -- n04 is absent from the
+# snapshot in 107 of them and is its own root in 49 -- but those are runs
+# whose tree had not converged when the snapshot was taken rather than runs
+# that chose the wrong parent, which is why the assertion fails an absent
+# or self-parented node explicitly instead of skipping it.
+assertions:
+  tree_parents:
+    n04: n03
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/depth-vs-cost.yaml

@@ -23,6 +23,23 @@
 #
 # Validation: tree snapshot shows n04's parent selection reflects the
 # actual cost tradeoff between depth and link quality.
+#
+# NO PARENT IS ASSERTED HERE, and unlike the other cost scenarios that is
+# not because the implementation disagrees with the comment. It is because
+# the validation line above does not name an outcome: "reflects the actual
+# cost tradeoff" is satisfied by either answer, so there is nothing to
+# encode. The comment above even says n04 "may still pick n01".
+#
+# The corpus agrees that both answers occur. Across the five provably
+# completed archived runs, n04's parent is n01 in three and n03 in two,
+# with the same seed each time, so the choice is not deterministic under
+# this configuration.
+#
+# Pinning whichever answer is more common would convert an unspecified
+# behaviour into a regression guard for something nobody decided, and
+# would go red roughly two runs in five besides. What this scenario needs
+# first is a decision about which parent is correct given the stated
+# effective depths, and that is a protocol question, not a harness one.
 
 scenario:
   name: "depth-vs-cost"
@@ -66,6 +83,21 @@ traffic:
   duration_secs: {min: 5, max: 10}
   parallel_streams: 2
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about which parent n04 chooses, which is unspecified;
+# it exists so that a run in which the mesh never formed cannot report
+# success, which until now it could,
+# because this scenario carried no assertions at all.
+#
+# Four nodes, one root, three parented and four sessions in all five
+# provably-completed archived runs.
+assertions:
+  baseline:
+    min_nodes_reporting: 4
+    max_roots: 1
+    min_nodes_parented: 3
+    min_sessions: 2
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/ethernet-mesh.yaml

@@ -62,6 +62,21 @@ link_flaps:
 traffic:
   enabled: false
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about Ethernet link behaviour under flaps; it exists
+# so that a run in which the mesh never formed cannot report success,
+# which until now it could, because this scenario carried no assertions
+# at all.
+#
+# Six nodes, one root, five parented in all six provably-completed
+# archived runs. The other assertion covering Ethernet transport in
+# CI; see ethernet-only for why that matters.
+assertions:
+  baseline:
+    min_nodes_reporting: 6
+    max_roots: 1
+    min_nodes_parented: 5
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/ethernet-only.yaml

@@ -41,6 +41,25 @@ link_flaps:
 traffic:
   enabled: false
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about what Ethernet transport does; it exists so that
+# a run in which the mesh never formed cannot report success, which
+# until now it could, because this scenario carried no assertions at
+# all.
+#
+# A 4-node mesh forms a spanning tree: one root and three nodes
+# with a parent. All six provably-completed archived runs show exactly
+# that, so these are the shape of a converged mesh rather than a
+# tolerance fitted to observations. This is currently the only
+# assertion covering Ethernet transport anywhere in CI: the three
+# tests in src/node/tests/ethernet.rs are #[ignore]d for requiring
+# CAP_NET_RAW and neither runner passes --ignored.
+assertions:
+  baseline:
+    min_nodes_reporting: 4
+    max_roots: 1
+    min_nodes_parented: 3
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/mixed-technology.yaml

@@ -30,6 +30,26 @@
 #   - n08 has fiber (n03) and Bluetooth (n04) parents — should pick n03
 #
 # Netem mutation shifts fiber-only links between normal and degraded.
+#
+# NEITHER TEST SUBJECT IS ASSERTED, because the implementation does not do
+# what the two lines above say. Across the five archived runs that carry a
+# status.txt and are therefore provably completed, n06's parent is n10 in
+# three and n03 in two, and n08's is n04 in four and n03 in one. Encoding
+# either as written would red the scenario most runs; encoding what it
+# actually does would pin behaviour nobody specified and would silently
+# bless it.
+#
+# So this is an open question rather than a missing assertion: either
+# cost-based selection is not preferring fiber here, or the criteria were
+# written for a topology this file no longer has. n06 taking n10 is the
+# more interesting half — n10 reaches n06 by fiber, so the choice may be
+# defensible and the comment simply stale. It needs someone to work out
+# which, and until then this scenario proves nothing about parent choice.
+#
+# That leaves this scenario asserting nothing about its own subject. The
+# baseline block further down holds it to forming a tree and carrying
+# traffic, which is a floor and not a substitute: nothing here checks that
+# the tree it forms is the one these comments describe.
 
 scenario:
   name: "mixed-technology"
@@ -98,6 +118,23 @@ traffic:
   duration_secs: {min: 5, max: 15}
   parallel_streams: 4
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about parent choice, which this scenario cannot
+# currently assert; it exists so that a run in which the mesh never formed
+# cannot report success, which until now it could,
+# because this scenario carried no assertions at all.
+#
+# Ten nodes, one root, nine parented and six to eight sessions across
+# the five provably-completed archived runs. The session floor is set
+# below the observed minimum deliberately: it is here to catch a mesh
+# that carries no traffic at all, not to pin a throughput.
+assertions:
+  baseline:
+    min_nodes_reporting: 10
+    max_roots: 1
+    min_nodes_parented: 9
+    min_sessions: 3
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/smoke-10.yaml

@@ -22,6 +22,20 @@ link_flaps:
 traffic:
   enabled: false
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about any particular subsystem; it exists so that a
+# run in which the mesh never formed cannot report success, which until
+# now it could, because this scenario carried no assertions at all.
+#
+# Ten nodes, one root, nine parented in all six provably-completed
+# archived runs. As the smoke test this is the scenario whose silent
+# success was least defensible.
+assertions:
+  baseline:
+    min_nodes_reporting: 10
+    max_roots: 1
+    min_nodes_parented: 9
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/scenarios/tcp-mesh.yaml

@@ -61,6 +61,19 @@ link_flaps:
 traffic:
   enabled: false
 
+# Baseline: the mesh came up, agreed on a root, and took parents. This
+# asserts nothing about TCP transport specifically; it exists so that a
+# run in which the mesh never formed cannot report success, which until
+# now it could, because this scenario carried no assertions at all.
+#
+# Six nodes, one root, five parented in all six provably-completed
+# archived runs.
+assertions:
+  baseline:
+    min_nodes_reporting: 6
+    max_roots: 1
+    min_nodes_parented: 5
+
 logging:
   rust_log: "info"
   output_dir: "./sim-results"

testing/chaos/sim/assertions.py

@@ -19,9 +19,13 @@ from dataclasses import dataclass
 
 from .control import snapshot_all_bloom
 from .scenario import (
+    BaselineAssertion,
     BloomSendRateAssertion,
+    CongestionSignalsAssertion,
+    MaxErrorsAssertion,
     MaxParentSwitchesAssertion,
     MinParentSwitchesAssertion,
+    TreeParentsAssertion,
 )
 from .topology import SimTopology
 
@@ -170,6 +174,243 @@ def evaluate_max_parent_switches(
     )
 
 
+def evaluate_baseline(
+    cfg: BaselineAssertion,
+    snapshot: dict | None,
+    sessions: int,
+) -> AssertionOutcome:
+    """Floor on the mesh having formed: nodes answered, agreed a root, took parents."""
+    if not snapshot:
+        return AssertionOutcome(
+            name="baseline",
+            passed=False,
+            detail=(
+                "FAIL baseline: no final tree snapshot was taken, so nothing "
+                "about the mesh was observed. This is a harness failure."
+            ),
+        )
+
+    reporting = len(snapshot)
+    roots = {v.get("root") for v in snapshot.values() if v.get("root")}
+    parented = sum(
+        1 for v in snapshot.values()
+        if v.get("parent") and v.get("parent") != v.get("my_node_addr")
+    )
+
+    parts, failures = [], []
+
+    def note(ok, text):
+        parts.append(text)
+        if not ok:
+            failures.append(text)
+
+    if cfg.min_nodes_reporting is not None:
+        note(reporting >= cfg.min_nodes_reporting,
+             f"{reporting} node(s) answered (need {cfg.min_nodes_reporting})")
+    if cfg.max_roots is not None:
+        note(len(roots) <= cfg.max_roots and len(roots) >= 1,
+             f"{len(roots)} distinct root(s) (allowed {cfg.max_roots})")
+    if cfg.min_nodes_parented is not None:
+        note(parented >= cfg.min_nodes_parented,
+             f"{parented} node(s) have a parent (need {cfg.min_nodes_parented})")
+    if cfg.min_sessions is not None:
+        note(sessions >= cfg.min_sessions,
+             f"{sessions} session(s) established (need {cfg.min_sessions})")
+
+    summary = "; ".join(parts)
+    if failures:
+        return AssertionOutcome(
+            name="baseline",
+            passed=False,
+            detail=(
+                f"FAIL baseline: {'; '.join(failures)}. Full: {summary}"
+            ),
+        )
+    return AssertionOutcome(
+        name="baseline", passed=True, detail=f"PASS baseline: {summary}"
+    )
+
+
+def evaluate_tree_parents(
+    cfg: TreeParentsAssertion,
+    snapshot: dict | None,
+) -> AssertionOutcome:
+    """Check each node's parent in the final tree snapshot.
+
+    Parents are compared by node address, resolved from the snapshot's own
+    ``my_node_addr`` fields, so the check does not depend on the display
+    name a node happened to publish.
+
+    Every way of not knowing the answer is a failure: no snapshot, the
+    node absent from it, the expected parent absent from it, or the node
+    still claiming to be its own root. Each of those produces the same
+    "no match" that a genuinely wrong parent does, and only saying so
+    separately keeps a harness problem from reading as a routing verdict.
+    """
+    if not snapshot:
+        return AssertionOutcome(
+            name="tree_parents",
+            passed=False,
+            detail=(
+                "FAIL tree_parents: no final tree snapshot was taken, so no "
+                "node's parent was observed. This is a harness failure, not "
+                "a statement about the tree."
+            ),
+        )
+
+    addr_of = {
+        nid: data.get("my_node_addr")
+        for nid, data in snapshot.items()
+        if data.get("my_node_addr")
+    }
+    id_of = {addr: nid for nid, addr in addr_of.items()}
+
+    good, bad = [], []
+    for child, want_parent in sorted(cfg.expected.items()):
+        entry = snapshot.get(child)
+        if entry is None:
+            bad.append(
+                f"{child} is absent from the snapshot ({len(snapshot)} node(s) "
+                f"present: {', '.join(sorted(snapshot))})"
+            )
+            continue
+        want_addr = addr_of.get(want_parent)
+        if want_addr is None:
+            bad.append(
+                f"{child}: expected parent {want_parent} is absent from the "
+                f"snapshot, so its address cannot be resolved"
+            )
+            continue
+        got_addr = entry.get("parent")
+        if got_addr == entry.get("my_node_addr"):
+            bad.append(
+                f"{child} is its own parent — it still believes it is root, "
+                f"so the tree never converged around it (wanted {want_parent})"
+            )
+            continue
+        if got_addr == want_addr:
+            good.append(f"{child}->{want_parent}")
+            continue
+        got_id = id_of.get(got_addr) or entry.get("parent_display_name") or got_addr
+        bad.append(f"{child} chose {got_id}, wanted {want_parent}")
+
+    if bad:
+        detail = f"FAIL tree_parents: {'; '.join(bad)}"
+        if good:
+            detail += f". Correct: {', '.join(good)}"
+        return AssertionOutcome(name="tree_parents", passed=False, detail=detail)
+    return AssertionOutcome(
+        name="tree_parents",
+        passed=True,
+        detail=f"PASS tree_parents: {', '.join(good)}",
+    )
+
+
+_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]],
+) -> AssertionOutcome:
+    """Ceiling on ERROR-level log lines across the whole mesh.
+
+    ``errors`` is the ``AnalysisResult.errors`` list of ``(source, line)``
+    pairs rather than a bare count, so a failure can name the nodes and
+    quote the lines. A ceiling breach that only reports a number sends the
+    reader back to the logs it was supposed to save them reading.
+    """
+    count = len(errors)
+    if count <= cfg.max_total:
+        return AssertionOutcome(
+            name="max_errors",
+            passed=True,
+            detail=(
+                f"PASS max_errors: {count} ERROR line(s) mesh-wide <= "
+                f"ceiling {cfg.max_total}"
+            ),
+        )
+
+    per_node: dict[str, int] = {}
+    for source, _line in errors:
+        per_node[source] = per_node.get(source, 0) + 1
+    worst = sorted(per_node.items(), key=lambda kv: -kv[1])
+    breakdown = ", ".join(f"{src}={n}" for src, n in worst)
+    samples = "\n".join(
+        f"    [{src}] {line.strip()}" for src, line in errors[:5]
+    )
+    return AssertionOutcome(
+        name="max_errors",
+        passed=False,
+        detail=(
+            f"FAIL max_errors: {count} ERROR line(s) mesh-wide > ceiling "
+            f"{cfg.max_total} — per node: {breakdown}. First "
+            f"{min(5, count)}:\n{samples}"
+        ),
+    )
+
+
 def evaluate_min_parent_switches(
     cfg: MinParentSwitchesAssertion,
     parent_switch_count: int,

testing/chaos/sim/runner.py

@@ -15,8 +15,12 @@ from datetime import datetime
 from .assertions import (
     AssertionOutcome,
     BloomSendRateMonitor,
+    evaluate_baseline,
+    evaluate_congestion_signals,
+    evaluate_max_errors,
     evaluate_max_parent_switches,
     evaluate_min_parent_switches,
+    evaluate_tree_parents,
 )
 from .compose import generate_compose
 from .config_gen import write_configs
@@ -74,6 +78,11 @@ 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
+        self.final_tree: dict | None = None
 
     def _evaluate_max_parent_switches(
         self, cfg, parent_switches: list[tuple[str, str]]
@@ -604,6 +613,49 @@ class SimRunner:
                 else:
                     log.error("%s", outcome.detail)
 
+            # Applied to every scenario by default, so this is the one
+            # assertion that is present even when the YAML declares no
+            # assertions block at all.
+            err_cfg = self.scenario.assertions.max_errors
+            if err_cfg is not None:
+                outcome = evaluate_max_errors(err_cfg, result.errors)
+                self.assertion_outcomes.append(outcome)
+                if outcome.passed:
+                    log.info("%s", outcome.detail)
+                else:
+                    log.error("%s", outcome.detail)
+
+            bl_cfg = self.scenario.assertions.baseline
+            if bl_cfg is not None:
+                outcome = evaluate_baseline(
+                    bl_cfg, self.final_tree, len(result.sessions_established)
+                )
+                self.assertion_outcomes.append(outcome)
+                if outcome.passed:
+                    log.info("%s", outcome.detail)
+                else:
+                    log.error("%s", outcome.detail)
+
+            tp_cfg = self.scenario.assertions.tree_parents
+            if tp_cfg is not None:
+                outcome = evaluate_tree_parents(tp_cfg, self.final_tree)
+                self.assertion_outcomes.append(outcome)
+                if outcome.passed:
+                    log.info("%s", outcome.detail)
+                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")
@@ -664,6 +716,12 @@ 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
+            self.final_tree = tree_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

@@ -3,10 +3,16 @@
 from __future__ import annotations
 
 import os
+import re
 from dataclasses import dataclass, field
 
 import yaml
 
+# Node ids are rendered nNN by the topology generator, zero-padded to two
+# digits. Matching the shape here keeps a typo such as "no4" or "n4x" from
+# reaching an assertion as a node that simply never appears.
+_NODE_ID_RE = re.compile(r"n\d+")
+
 
 @dataclass
 class Range:
@@ -218,6 +224,88 @@ class MaxParentSwitchesAssertion:
     node: str | None = None
 
 
+@dataclass
+class BaselineAssertion:
+    """Floor on the mesh having formed at all.
+
+    Deliberately weak and deliberately universal. It does not describe any
+    scenario's subject; it says the nodes came up, agreed on a root, and
+    took parents. Its value is that a scenario with no other assertion
+    still cannot pass while the mesh is dead, which is the state twelve of
+    thirteen scenarios were previously unable to distinguish from success.
+
+    ``max_roots`` is how many distinct root values the snapshot may carry.
+    One means the whole mesh agreed; a churn scenario that partitions on
+    purpose needs a higher number, and setting it above one is a statement
+    that partition is expected rather than an oversight.
+    """
+
+    min_nodes_reporting: int | None = None
+    max_roots: int | None = None
+    min_nodes_parented: int | None = None
+    min_sessions: int | None = None
+
+
+@dataclass
+class TreeParentsAssertion:
+    """Expected parent per node in the final tree snapshot.
+
+    ``expected`` maps a node id to the node id its parent must be, e.g.
+    ``{"n04": "n03"}``. Both sides are node ids rather than node
+    addresses: an address is a per-run key derived from the generated
+    identity, so a scenario could not name one ahead of time.
+
+    A node absent from the snapshot fails rather than being skipped. That
+    case is not hypothetical — more than half the archived runs of these
+    scenarios have no entry for the node under test, and a skip would
+    have reported those as satisfied.
+    """
+
+    expected: dict[str, str] = field(default_factory=dict)
+
+
+@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.
+
+    Unlike the other assertions this one is applied to every scenario
+    whether or not the YAML asks for it, because it is a floor on what a
+    green run means rather than a property of one scenario: without it a
+    run in which every node errors on every line still exits 0.
+
+    ``max_total`` defaults to 0, which is what the archived corpus
+    supports. Across 2416 archived run directories no node log contains a
+    single ERROR-level line, while the sibling WARN counter extracted by
+    the same code path ranges from 0 to 1135 — so 0 is an observed value
+    and not an aspiration, and the counter is known to discriminate.
+
+    A scenario that legitimately induces errors raises the ceiling in its
+    own YAML and must say at that site why the errors are expected.
+    """
+
+    max_total: int = 0
+
+
 @dataclass
 class AssertionsConfig:
     """Optional post-run assertions evaluated against control-socket data."""
@@ -225,6 +313,10 @@ class AssertionsConfig:
     bloom_send_rate: BloomSendRateAssertion | None = None
     min_parent_switches: MinParentSwitchesAssertion | None = None
     max_parent_switches: MaxParentSwitchesAssertion | None = None
+    max_errors: MaxErrorsAssertion | None = None
+    congestion_signals: CongestionSignalsAssertion | None = None
+    tree_parents: TreeParentsAssertion | None = None
+    baseline: BaselineAssertion | None = None
 
 
 @dataclass
@@ -260,10 +352,13 @@ class Scenario:
 # YAML still looks correct — a mistyped "assertion:" disarms a scenario's only
 # assertions, a mistyped "link_flap:" turns off the chaos it was meant to inject.
 #
-# Three mappings are deliberately NOT listed, because their keys are names the
+# Four mappings are deliberately NOT listed, because their keys are names the
 # scenario author chooses rather than a schema: netem.mutation.policies,
-# link_swap.policies and topology.transport_mix. Two sub-trees are passed through
-# whole and are likewise not checked: fips_overrides and topology.params.
+# link_swap.policies, topology.transport_mix and assertions.tree_parents (whose
+# keys are node ids). Two sub-trees are passed through whole and are likewise
+# not checked: fips_overrides and topology.params. tree_parents is not thereby
+# unchecked -- both sides of every entry are validated as node ids that exist in
+# this scenario's topology, which is the check that matters for it.
 #
 # NOTE: adding a new assertion type means registering it in TWO places below —
 # _SECTION_KEYS["assertions"] (so the block accepts its name) and _ASSERTION_KEYS
@@ -301,6 +396,7 @@ _SECTION_KEYS = {
     "link_swap.edges[]": {"edge", "policy"},
     "assertions": {
         "bloom_send_rate", "min_parent_switches", "max_parent_switches",
+        "max_errors", "congestion_signals", "tree_parents", "baseline",
     },
     "logging": {"rust_log", "output_dir"},
 }
@@ -308,6 +404,13 @@ _ASSERTION_KEYS = {
     "bloom_send_rate": {"window_secs", "max_per_node"},
     "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",
+    },
+    "baseline": {
+        "min_nodes_reporting", "max_roots", "min_nodes_parented", "min_sessions",
+    },
 }
 _NETEM_POLICY_KEYS = {
     "delay_ms", "jitter_ms", "loss_pct", "duplicate_pct", "reorder_pct",
@@ -589,6 +692,134 @@ def load_scenario(path: str) -> Scenario:
             max_total=max_total,
             node=node,
         )
+    if "max_errors" in asrt:
+        mev = asrt["max_errors"]
+        _reject_unknown(
+            mev, _ASSERTION_KEYS["max_errors"], "assertions.max_errors",
+        )
+        if "max_total" not in mev:
+            raise ValueError(
+                "assertions.max_errors: max_total is required (the point of "
+                "overriding the default ceiling is to name the new number)"
+            )
+        err_total = mev["max_total"]
+        if isinstance(err_total, bool) or not isinstance(err_total, int):
+            raise ValueError(
+                f"assertions.max_errors: max_total must be a non-negative "
+                f"integer, got {err_total!r}"
+            )
+        if err_total < 0:
+            raise ValueError(
+                f"assertions.max_errors: max_total must be non-negative, "
+                f"got {err_total}"
+            )
+        s.assertions.max_errors = MaxErrorsAssertion(max_total=err_total)
+    else:
+        # Default-on. See MaxErrorsAssertion for why this one assertion is
+        # applied without being asked for: it is the floor on what a green
+        # 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)
+    if "tree_parents" in asrt:
+        tp = asrt["tree_parents"]
+        if not isinstance(tp, dict) or not tp:
+            raise ValueError(
+                "assertions.tree_parents: give it at least one "
+                "'<node>: <expected parent>' entry; an empty block asserts "
+                "nothing"
+            )
+        expected = {}
+        for child, parent in tp.items():
+            for role, val in (("node", child), ("parent", parent)):
+                if not isinstance(val, str) or not _NODE_ID_RE.fullmatch(val):
+                    raise ValueError(
+                        f"assertions.tree_parents: {role} {val!r} is not a node "
+                        f"id of the form 'n04'"
+                    )
+                idx = int(val[1:])
+                if idx < 1 or idx > s.topology.num_nodes:
+                    raise ValueError(
+                        f"assertions.tree_parents: {role} '{val}' is outside "
+                        f"this scenario's {s.topology.num_nodes} nodes. An "
+                        f"assertion about a node that cannot exist would fail "
+                        f"for the wrong reason every run."
+                    )
+            if child == parent:
+                raise ValueError(
+                    f"assertions.tree_parents: '{child}' is given itself as "
+                    f"its parent. A node is its own parent only when it "
+                    f"believes it is root, which is what an unconverged tree "
+                    f"looks like; assert that some other way."
+                )
+            expected[child] = parent
+        s.assertions.tree_parents = TreeParentsAssertion(expected=expected)
+    if "baseline" in asrt:
+        bl = asrt["baseline"]
+        _reject_unknown(bl, _ASSERTION_KEYS["baseline"], "assertions.baseline")
+        vals = {}
+        for key in _ASSERTION_KEYS["baseline"]:
+            if key not in bl:
+                continue
+            val = bl[key]
+            if isinstance(val, bool) or not isinstance(val, int):
+                raise ValueError(
+                    f"assertions.baseline.{key}: must be an integer, "
+                    f"got {val!r}"
+                )
+            floor = 1 if key == "max_roots" else 0
+            if val < floor:
+                raise ValueError(
+                    f"assertions.baseline.{key}: must be at least {floor}, "
+                    f"got {val}"
+                )
+            vals[key] = val
+        if not vals:
+            raise ValueError(
+                "assertions.baseline: set at least one of "
+                + ", ".join(sorted(_ASSERTION_KEYS["baseline"]))
+                + "; a block with none asserts nothing"
+            )
+        if vals.get("min_nodes_parented", 0) > 0 or vals.get("max_roots"):
+            n = s.topology.num_nodes
+            if vals.get("min_nodes_parented", 0) > n - 1:
+                raise ValueError(
+                    f"assertions.baseline.min_nodes_parented: "
+                    f"{vals['min_nodes_parented']} exceeds {n - 1}, the most a "
+                    f"{n}-node mesh can reach — the root is its own parent, so "
+                    f"this could never pass"
+                )
+        s.assertions.baseline = BaselineAssertion(**vals)
 
     # Logging section
     lg = raw.get("logging", {})
@@ -639,9 +870,31 @@ def _validate_parent_switch_observability(s: Scenario):
         )
 
 
+def _validate_error_observability(s: Scenario):
+    """Refuse an error ceiling the log level cannot observe.
+
+    Narrower than the parent-switch guard on purpose: ERROR lines survive
+    every level except ``off``, so ``off`` is the only setting that turns
+    this assertion into one that counts zero whatever the mesh did. Since
+    the ceiling is applied by default, a scenario silencing its logs would
+    otherwise acquire an assertion that cannot fail.
+    """
+    if s.assertions.max_errors is None:
+        return
+    default_level = s.logging.rust_log.split(",")[0].strip().lower()
+    if default_level == "off":
+        raise ValueError(
+            "logging.rust_log is 'off', which suppresses the ERROR-level "
+            "lines the max_errors assertion counts. The assertion would see "
+            "zero errors regardless of what the mesh did. Raise the level, "
+            "or state a deliberate override in assertions.max_errors."
+        )
+
+
 def _validate(s: Scenario):
     """Validate scenario constraints."""
     _validate_parent_switch_observability(s)
+    _validate_error_observability(s)
     if s.topology.num_nodes < 2:
         raise ValueError("topology.num_nodes must be >= 2")
     if s.topology.num_nodes > 250: