test: windowed IDF prevents semantic hysteresis (37/37)

test_swarmmetrics.py

@@ -824,3 +824,70 @@ def test_ablation_stimulus_preemption():
     # 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']}"
+
+
+def test_windowed_idf_prevents_semantic_hysteresis():
+    """Windowed IDF prevents retired loud agent from suppressing via stale vocabulary norms."""
+    import random
+    random.seed(42)
+    
+    messages = []
+    
+    # Phase 1 (t=0-799): loud agent floods with alpha/beta/gamma
+    for i in range(80):
+        messages.append({
+            "from_id": "loud",
+            "to_id": "_all",
+            "timestamp": float(i * 10 + random.randint(0, 2)),
+            "concepts": ["alpha", "beta", "gamma", f"event_{i}"]
+        })
+    
+    # Phase 1: some others use alpha/beta too
+    for i in range(80):
+        for agent in ["a1", "a2"]:
+            if random.random() < 0.12:
+                messages.append({
+                    "from_id": agent,
+                    "to_id": "_all",
+                    "timestamp": float(i * 10 + random.randint(10, 50)),
+                    "concepts": ["alpha", "beta", f"own_{agent}_{i}"]
+                })
+    
+    # Phase 2 (t=800-1600): loud is SILENT. New agents use alpha/beta.
+    for i in range(80):
+        for agent in ["b1", "b2", "b3"]:
+            if random.random() < 0.25:
+                messages.append({
+                    "from_id": agent,
+                    "to_id": "_all",
+                    "timestamp": float(800 + i * 10 + random.randint(0, 20)),
+                    "concepts": ["alpha", "beta", f"own_{agent}_{i}"]
+                })
+    
+    # Control: phase 2 in isolation (no phase 1 contamination)
+    control = [m for m in messages if m["timestamp"] >= 800]
+    
+    # Global IDF: phase 1 norms persist, phase 2 agents get less credit
+    scores_global = score_echo(messages, snapshot_idf=True)
+    scores_control = score_echo(control, snapshot_idf=True)
+    
+    # Windowed IDF (window = 500 time units): phase 1 forgotten by phase 2
+    scores_windowed = score_echo(messages, snapshot_idf=True, idf_window_seconds=500)
+    
+    # Collect deltas for b-agents
+    global_deltas = []
+    windowed_deltas = []
+    for agent in ["b1", "b2", "b3"]:
+        z_ctrl = scores_control.get(agent, 0)
+        z_global = scores_global.get(agent, 0)
+        z_windowed = scores_windowed.get(agent, 0)
+        global_deltas.append(z_global - z_ctrl)
+        windowed_deltas.append(z_windowed - z_ctrl)
+    
+    mean_global_delta = sum(global_deltas) / len(global_deltas)
+    mean_windowed_delta = sum(windowed_deltas) / len(windowed_deltas)
+    
+    # Windowed IDF should reduce or eliminate the suppression effect
+    # (windowed delta should be closer to zero than global delta)
+    assert mean_windowed_delta >= mean_global_delta, \
+        f"Windowed IDF should reduce suppression: windowed Δ={mean_windowed_delta:.3f} vs global Δ={mean_global_delta:.3f}"