test: injection taxonomy — surgical kill + attacker invisibility (43/43)

test_swarmmetrics.py

@@ -1005,3 +1005,114 @@ def test_ewma_approaches_global_with_large_halflife():
 
     assert abs(ewma_large - global_score) < 0.01, \
         f"Large half-life EWMA ({ewma_large:.4f}) should ≈ global ({global_score:.4f})"
+
+
+def test_targeted_injection_surgical_kill():
+    """
+    Injecting an agent using ONLY target X's concepts suppresses X to ~0
+    while leaving all other agents within epsilon.
+
+    Three-way taxonomy proven on real bus data 2026-07-28:
+    - Targeted (X's vocab): X → 0, others ±0%, attacker invisible
+    - Shared (multi-agent vocab): broad suppression, dominants immune
+    - Random: mostly amplifies (bad weapon)
+
+    The targeted case is the strongest invariant. This test proves it synthetically.
+    """
+    import random
+    random.seed(42)
+
+    # Target introduces concepts that others echo.
+    # Attacker pre-empts those same concepts — should kill target's echo.
+    msgs = []
+    target_concepts = [f"x{i}" for i in range(5)]
+    other_concepts = {
+        "agent_a": [f"a{i}" for i in range(5)],
+        "agent_b": [f"b{i}" for i in range(5)],
+        "agent_c": [f"c{i}" for i in range(5)],
+    }
+
+    # Build 30 rounds: target introduces, others echo target's concepts
+    for round_i in range(30):
+        t = float(round_i * 200)
+        tc = target_concepts[round_i % 5]
+        # Target introduces its concept first
+        msgs.append({"from_id": "target", "to_id": "_all",
+                     "timestamp": t, "concepts": [tc]})
+        # Other agents echo target's concept (with their own too)
+        for agent, uniques in other_concepts.items():
+            delay = random.randint(30, 90)
+            msgs.append({
+                "from_id": agent, "to_id": "_all",
+                "timestamp": t + delay,
+                "concepts": [tc, uniques[round_i % 5]]
+            })
+
+    baseline = score_echo(msgs, snapshot_idf=True)
+    target_base = baseline.get("target", 0)
+    assert target_base > 0, f"target should have nonzero echo in baseline, got {baseline}"
+
+    # Inject attacker using ONLY target's concepts, BEFORE target
+    injected = list(msgs)
+    for round_i in range(30):
+        t = float(round_i * 200)
+        tc = target_concepts[round_i % 5]
+        injected.append({
+            "from_id": "attacker", "to_id": "_all",
+            "timestamp": t - 5,  # 5s BEFORE target — pre-empts introduction
+            "concepts": [tc]
+        })
+
+    with_injection = score_echo(injected, snapshot_idf=True)
+    target_after = with_injection.get("target", 0)
+
+    # Core invariant: target echo drops by >50%
+    suppression = 1.0 - (target_after / target_base) if target_base > 0 else 0
+    assert suppression > 0.5, \
+        f"Targeted injection should suppress >50%: was {target_base:.2f} → {target_after:.2f} ({suppression:.0%})"
+
+    # Collateral check: other agents should be minimally affected
+    for agent in ["agent_a", "agent_b", "agent_c"]:
+        before = baseline.get(agent, 0)
+        after = with_injection.get(agent, 0)
+        if before > 0:
+            change = abs(after - before) / before
+            assert change < 0.5, \
+                f"Collateral damage on {agent}: {before:.2f} → {after:.2f} ({change:.0%} change, expected <50%)"
+
+
+def test_injection_attacker_invisible():
+    """Attacker using target's unique vocabulary gets near-zero echo itself."""
+    import random
+    random.seed(42)
+
+    msgs = []
+    # 4 agents with distinct concepts, one shared
+    for round_i in range(25):
+        t = float(round_i * 120)
+        msgs.append({"from_id": "target", "to_id": "_all",
+                     "timestamp": t + 10, "concepts": ["shared", f"t_{round_i % 5}"]})
+        msgs.append({"from_id": "other1", "to_id": "_all",
+                     "timestamp": t + 20, "concepts": ["shared", f"o1_{round_i % 5}"]})
+        msgs.append({"from_id": "other2", "to_id": "_all",
+                     "timestamp": t + 30, "concepts": ["shared", f"o2_{round_i % 5}"]})
+
+    # Inject attacker that only uses target's concepts
+    injected = list(msgs)
+    target_concepts = [f"t_{i}" for i in range(5)]
+    for round_i in range(25):
+        t = float(round_i * 120)
+        injected.append({
+            "from_id": "attacker", "to_id": "_all",
+            "timestamp": t + 5,  # faster than target
+            "concepts": [target_concepts[round_i % 5]]
+        })
+
+    scores = score_echo(injected, snapshot_idf=True)
+    attacker_echo = scores.get("attacker", 0)
+    # Attacker should be nearly invisible — its concepts are target's,
+    # and it pre-empts target, so nobody echoes the attacker (they echo the original stimulus)
+    # Low threshold because the attacker introduces timing patterns, not content
+    max_attacker = max(scores.get(a, 0) for a in ["target", "other1", "other2"])
+    assert attacker_echo < max_attacker * 0.5, \
+        f"Attacker should be relatively invisible: echo={attacker_echo:.2f} vs max agent={max_attacker:.2f}"