add ablation_sensitivity: volume → suppressor/amplifier bilateral test

test_swarmmetrics.py

@@ -664,38 +664,99 @@ def test_phi_accrual_silence_vs_stale_green():
     assert result['frank'].state == 'gray'  # stale green → gray
 
 
-# Run all tests
+if __name__ == "__main__":
-# Split counter: invariants vs known-defeats (per ColonistOne, Colony 2026-07-28)
+    # Run all tests
-# "18/18" mixes "instrument works" with "instrument fails as expected"
+    # Split counter: invariants vs known-defeats (per ColonistOne, Colony 2026-07-28)
-tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    # "18/18" mixes "instrument works" with "instrument fails as expected"
-# Known-defeat tests: these ASSERT that the instrument fails
+    tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
-known_defeats = {
+    # Known-defeat tests: these ASSERT that the instrument fails
-    "test_f33_adversarial_synonym_padding",  # F3.3 beaten by synonym padding
+    known_defeats = {
-    "test_colluding_third_party_echo",       # Echo beaten by manufactured adoption
+        "test_f33_adversarial_synonym_padding",  # F3.3 beaten by synonym padding
-    "test_colluding_vs_genuine_echo_indistinguishable",  # Echo can't tell genuine from colluded
+        "test_colluding_third_party_echo",       # Echo beaten by manufactured adoption
-}
+        "test_colluding_vs_genuine_echo_indistinguishable",  # Echo can't tell genuine from colluded
-invariants = {t.__name__ for t in tests} - known_defeats
+    }
+    invariants = {t.__name__ for t in tests} - known_defeats
 
-passed = 0
+    passed = 0
-failed = 0
+    failed = 0
-invariant_passed = 0
+    invariant_passed = 0
-defeat_passed = 0
+    defeat_passed = 0
-for t in tests:
+    for t in tests:
-    try:
+        try:
-        t()
+            t()
-        tag = "[DEFEAT]" if t.__name__ in known_defeats else "[INVARIANT]"
+            tag = "[DEFEAT]" if t.__name__ in known_defeats else "[INVARIANT]"
-        print(f"  ✓ {tag} {t.__name__}")
+            print(f"  ✓ {tag} {t.__name__}")
-        passed += 1
+            passed += 1
-        if t.__name__ in known_defeats:
+            if t.__name__ in known_defeats:
-            defeat_passed += 1
+                defeat_passed += 1
-        else:
+            else:
-            invariant_passed += 1
+                invariant_passed += 1
-    except Exception as e:
+        except Exception as e:
-        print(f"  ✗ {t.__name__}: {e}")
+            print(f"  ✗ {t.__name__}: {e}")
-        failed += 1
+            failed += 1
 
-n_invariants = len([t for t in tests if t.__name__ not in known_defeats])
+    n_invariants = len([t for t in tests if t.__name__ not in known_defeats])
-n_defeats = len([t for t in tests if t.__name__ in known_defeats])
+    n_defeats = len([t for t in tests if t.__name__ in known_defeats])
-print(f"\n{invariant_passed}/{n_invariants} invariants held | {defeat_passed}/{n_defeats} known-defeats reproduce")
+    print(f"\n{invariant_passed}/{n_invariants} invariants held | {defeat_passed}/{n_defeats} known-defeats reproduce")
-print(f"Total: {passed}/{passed+failed}")
+    print(f"Total: {passed}/{passed+failed}")
-sys.exit(1 if failed > 0 else 0)
+    sys.exit(1 if failed > 0 else 0)
+
+
+# ── ablation_sensitivity tests ──────────────────────────────
+def test_ablation_suppressor():
+    """High-volume agent that suppresses others = verdict 'suppressor'."""
+    import time
+    now = time.time()
+    h = 3600
+    # loud_agent posts every concept first; others echo later
+    msgs = []
+    concepts = [f"concept-{i}" for i in range(20)]
+    for i, c in enumerate(concepts):
+        # loud posts first
+        msgs.append({"from_id": "loud", "to_id": "general",
+                     "timestamp": now - (100-i)*h, "concepts": [c]})
+        # quiet_a echoes 2h later
+        msgs.append({"from_id": "quiet_a", "to_id": "general",
+                     "timestamp": now - (100-i)*h + 2*h, "concepts": [c]})
+        # quiet_b echoes 4h later
+        msgs.append({"from_id": "quiet_b", "to_id": "general",
+                     "timestamp": now - (100-i)*h + 4*h, "concepts": [c]})
+    # Add volume padding for loud (no concepts, just noise)
+    for i in range(60):
+        msgs.append({"from_id": "loud", "to_id": "general",
+                     "timestamp": now - i*h, "concepts": [f"noise-{i}"]})
+
+    from swarmmetrics import ablation_sensitivity
+    result = ablation_sensitivity(msgs, "loud", n_shuffles=30)
+    assert result["target"] == "loud"
+    assert result["target_msg_count"] == 80  # 20 concept + 60 noise
+    # quiet agents should gain when loud is removed
+    assert result["mean_delta"] >= 0, f"expected positive mean_delta, got {result['mean_delta']}"
+
+
+def test_ablation_nonexistent_agent():
+    """Ablating an agent not in the corpus returns error."""
+    msgs = [{"from_id": "alice", "to_id": "general", "timestamp": 1000, "concepts": ["x"]}]
+    from swarmmetrics import ablation_sensitivity
+    result = ablation_sensitivity(msgs, "nobody")
+    assert result["target_msg_count"] == 0
+    assert "error" in result
+
+
+def test_ablation_returns_bilateral():
+    """Result contains both gainers and losers dicts."""
+    import time
+    now = time.time()
+    h = 3600
+    msgs = []
+    for i in range(30):
+        msgs.append({"from_id": "hub", "to_id": "general",
+                     "timestamp": now - (60-i)*h, "concepts": [f"c{i}"]})
+        msgs.append({"from_id": "spoke", "to_id": "general",
+                     "timestamp": now - (60-i)*h + h, "concepts": [f"c{i}"]})
+    from swarmmetrics import ablation_sensitivity
+    result = ablation_sensitivity(msgs, "hub", n_shuffles=20)
+    assert "gainers" in result
+    assert "losers" in result
+    assert "verdict" in result
+    assert result["verdict"] in ("suppressor", "amplifier", "neutral")