test: add stimulus_preemption fixture — proves temporal preemption is mechanism, not volume alone. 36/36.

test_swarmmetrics.py

@@ -760,3 +760,67 @@ def test_ablation_returns_bilateral():
     assert "losers" in result
     assert "verdict" in result
     assert result["verdict"] in ("suppressor", "amplifier", "neutral")
+
+
+def test_ablation_stimulus_preemption():
+    """
+    Reproducing fixture: fast responder to shared stimuli = suppressor.
+    Same agent responding SLOWLY to same stimuli = NOT suppressor.
+    Proves temporal preemption is the mechanism, not volume alone.
+    Discovered 2026-07-28: synthetic without adaptive vocabulary = neutral;
+    stimulus-reactive model with fast response = suppressor (Δ=+9.73).
+    """
+    import random
+    from swarmmetrics import ablation_sensitivity
+    random.seed(42)
+
+    # Generate 50 events, each with event-specific concepts
+    events = sorted([random.randint(0, 86400*3) for _ in range(50)])
+    event_concepts = {}
+    for i, ev_time in enumerate(events):
+        event_concepts[i] = (ev_time, [f"ev{i}_c{j}" for j in range(3)])
+
+    # FAST cron agent: responds to every event within 1-18 min
+    fast_msgs = []
+    for i, (ev_time, concepts) in event_concepts.items():
+        fast_msgs.append({
+            "from_id": "fast_cron", "to_id": f"ag_{random.randint(1,6)}",
+            "timestamp": ev_time + random.randint(60, 1080),
+            "concepts": concepts[:2]
+        })
+
+    # SLOW agent: same coverage, 2-4h delay
+    slow_msgs = []
+    for i, (ev_time, concepts) in event_concepts.items():
+        slow_msgs.append({
+            "from_id": "slow_responder", "to_id": f"ag_{random.randint(1,6)}",
+            "timestamp": ev_time + random.randint(7200, 14400),
+            "concepts": concepts[:2]
+        })
+
+    # 6 specialist agents: 15 events each, 30-60 min delay
+    specialist_msgs = []
+    for ag in range(1, 7):
+        for i in random.sample(range(50), 15):
+            ev_time, concepts = event_concepts[i]
+            specialist_msgs.append({
+                "from_id": f"ag_{ag}", "to_id": f"ag_{random.randint(1,6)}",
+                "timestamp": ev_time + random.randint(1800, 3600),
+                "concepts": concepts[:2]
+            })
+
+    # Test 1: fast cron + specialists
+    corpus_fast = sorted(fast_msgs + specialist_msgs, key=lambda m: m["timestamp"])
+    r_fast = ablation_sensitivity(corpus_fast, "fast_cron", n_shuffles=30, use_idf=True)
+
+    # Test 2: slow responder + specialists (same volume, same concepts)
+    corpus_slow = sorted(slow_msgs + specialist_msgs, key=lambda m: m["timestamp"])
+    r_slow = ablation_sensitivity(corpus_slow, "slow_responder", n_shuffles=30, use_idf=True)
+
+    # Fast = suppressor (removing helps others)
+    assert r_fast["mean_delta"] > 0, f"fast cron should suppress, got Δ={r_fast['mean_delta']}"
+    # Slow ≠ suppressor (removing doesn't help others)
+    assert r_slow["mean_delta"] <= 0.5, f"slow responder shouldn't suppress, got Δ={r_slow['mean_delta']}"
+    # The difference proves timing is the mechanism
+    assert r_fast["mean_delta"] > r_slow["mean_delta"] + 1.0, \
+        f"fast should suppress MORE than slow: {r_fast['mean_delta']} vs {r_slow['mean_delta']}"