"""Tests for SwarmMetrics.""" import time import sys sys.path.insert(0, ".") from swarmmetrics import ( _gini, _half_life_weight, score_reciprocity, score_channels, score_echo, detect_shadows, analyze, shuffle_test ) now = time.time() day = 86400 def test_gini_single(): """Single speaker = max inequality.""" assert _gini([10]) == 1.0 def test_gini_equal(): """Equal speakers = 0 inequality.""" g = _gini([10, 10, 10, 10]) assert abs(g) < 0.01, f"Expected ~0, got {g}" def test_gini_dominated(): """One dominant speaker.""" g = _gini([1, 1, 1, 100]) assert g > 0.5, f"Expected >0.5, got {g}" def test_half_life_zero(): """Message at t=now has weight ~1.""" w = _half_life_weight(0) assert abs(w - 1.0) < 0.01 def test_half_life_decay(): """Message at t=tau has weight ~0.5.""" w = _half_life_weight(7 * 86400, tau_days=7.0) assert abs(w - 0.5) < 0.01, f"Expected ~0.5, got {w}" def test_reciprocity_symmetric(): """Equal exchange = high F3.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 0.1*day}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.1*day}, ] edges = score_reciprocity(msgs, now=now) pair = ("a", "b") assert pair in edges assert edges[pair].reciprocity_f3 > 0.6, f"Expected >0.6, got {edges[pair].reciprocity_f3}" def test_reciprocity_asymmetric(): """One-way communication = F3 near 0.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 0.1*day}, {"from_id": "a", "to_id": "b", "timestamp": now - 0.2*day}, {"from_id": "a", "to_id": "b", "timestamp": now - 0.3*day}, ] edges = score_reciprocity(msgs, now=now) pair = ("a", "b") assert edges[pair].reciprocity_f3 < 0.01 def test_channel_classification(): """Monologue, dialogue, broadcast detection.""" msgs = [ {"from_id": "a", "to_id": "b", "channel": "mono", "timestamp": now}, {"from_id": "a", "to_id": "c", "channel": "mono", "timestamp": now}, {"from_id": "a", "to_id": "b", "channel": "talk", "timestamp": now}, {"from_id": "b", "to_id": "a", "channel": "talk", "timestamp": now}, ] channels = score_channels(msgs) assert channels["mono"].classification == "monologue" assert channels["talk"].classification == "dialogue" def test_echo_coefficient(): """Concept introduced by X, echoed by Y, X gets credit.""" msgs = [ {"from_id": "x", "to_id": "y", "timestamp": now - 5*day, "concepts": ["alpha"]}, {"from_id": "y", "to_id": "z", "timestamp": now - 3*day, "concepts": ["alpha"]}, ] echo = score_echo(msgs) assert echo.get("x", 0) > 0, f"X should have echo credit, got {echo}" def test_shadow_detection(): """Node silent for >7 days is a shadow.""" msgs = [ {"from_id": "ghost", "to_id": "alive", "timestamp": now - 14*day}, {"from_id": "alive", "to_id": "ghost", "timestamp": now - 0.5*day}, ] shadows = detect_shadows(msgs, now=now, silence_days=7.0) assert "ghost" in shadows assert "alive" not in shadows def test_full_analysis(): """Full analyze() returns ScoredGraph with all sections.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 1*day, "channel": "ch1", "concepts": ["x"]}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.5*day, "channel": "ch1"}, ] result = analyze(msgs, now=now) assert result.summary["total_messages"] == 2 assert result.summary["total_nodes"] == 2 assert len(result.edges) == 1 assert len(result.channels) == 1 assert len(result.nodes) == 2 def test_f32_zombie_broadcast(): """F3.2 must rank 5-message dialogue above 100-ping broadcast. Regression test for reticuli's arithmetic proof (Colony, 2026-07-27): F3.1 count*log(1+r) lets volume buy back low reciprocity.""" msgs = [] # 5-message dialogue, high reciprocity (3+2) for i in range(3): msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day + i*100}) for i in range(2): msgs.append({"from_id": "bob", "to_id": "alice", "timestamp": now - 0.5*day + i*100}) # 100-ping broadcast, low reciprocity (95+5) for i in range(95): msgs.append({"from_id": "carol", "to_id": "dave", "timestamp": now - 2*day + i*60}) for i in range(5): msgs.append({"from_id": "dave", "to_id": "carol", "timestamp": now - 1.5*day + i*600}) edges = score_reciprocity(msgs, now=now) dialogue = [e for e in edges.values() if "alice" in (e.source, e.target)][0] broadcast = [e for e in edges.values() if "carol" in (e.source, e.target)][0] assert dialogue.reciprocity_f32 > broadcast.reciprocity_f32, \ f"F3.2 zombie fail: dialogue={dialogue.reciprocity_f32:.4f} <= broadcast={broadcast.reciprocity_f32:.4f}" def test_f33_zombie_broadcast_concepts(): """F3.3 unique-concept-count must crush spam even harder than F3.2. 100 pings with 2 concepts vs 5 messages with 5 concepts. Idea from 小風 (ClawdChat): replace count with unique topic count.""" msgs = [] # 5-message dialogue, high reciprocity (3+2), 5 unique concepts concepts_rich = [["graph", "reciprocity"], ["decay", "metrics"], ["shadow"], ["echo", "idf"], ["influence"]] for i in range(3): msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day + i*100, "concepts": concepts_rich[i]}) for i in range(2): msgs.append({"from_id": "bob", "to_id": "alice", "timestamp": now - 0.5*day + i*100, "concepts": concepts_rich[3+i]}) # 100 pings, low reciprocity (95+5), only 2 unique concepts (repeated) for i in range(95): msgs.append({"from_id": "carol", "to_id": "dave", "timestamp": now - 2*day + i*60, "concepts": ["hello", "ping"]}) for i in range(5): msgs.append({"from_id": "dave", "to_id": "carol", "timestamp": now - 1.5*day + i*600, "concepts": ["hello"]}) edges = score_reciprocity(msgs, now=now) dialogue = [e for e in edges.values() if "alice" in (e.source, e.target)][0] broadcast = [e for e in edges.values() if "carol" in (e.source, e.target)][0] # F3.3 must separate them assert dialogue.reciprocity_f33 > broadcast.reciprocity_f33, \ (f"F3.3 zombie fail: dialogue={dialogue.reciprocity_f33:.4f} " f"<= broadcast={broadcast.reciprocity_f33:.4f}") # F3.3 spread should be wider than F3.2 f33_ratio = dialogue.reciprocity_f33 / max(broadcast.reciprocity_f33, 0.001) f32_ratio = dialogue.reciprocity_f32 / max(broadcast.reciprocity_f32, 0.001) assert f33_ratio > f32_ratio, \ f"F3.3 should separate more than F3.2: f33_ratio={f33_ratio:.1f} <= f32_ratio={f32_ratio:.1f}" def test_f33_fallback_no_concepts(): """F3.3 falls back to F3.2 when messages have no concepts.""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 1*day}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.5*day}, ] edges = score_reciprocity(msgs, now=now) edge = list(edges.values())[0] assert edge.reciprocity_f33 == edge.reciprocity_f32, \ f"F3.3 should equal F3.2 with no concepts: f33={edge.reciprocity_f33}, f32={edge.reciprocity_f32}" def test_f32_idf_echo_combined(): """F3.2 and IDF echo work together in full analyze().""" msgs = [ {"from_id": "a", "to_id": "b", "timestamp": now - 1*day, "channel": "ch", "concepts": ["rare-concept"]}, {"from_id": "b", "to_id": "a", "timestamp": now - 0.9*day, "channel": "ch", "concepts": ["rare-concept"]}, ] result = analyze(msgs, now=now) assert len(result.edges) == 1 edge = list(result.edges.values())[0] assert edge.reciprocity_f32 > 0, "F3.2 should be positive for reciprocal edge" def test_f33_gibberish_high_score(): """F3.3 CANNOT detect quality — only filter spam. High-reciprocity, high-diversity gibberish scores well. This is not a bug — it's the boundary of what statistics can measure.""" msgs = [] # Two agents exchanging diverse gibberish with high reciprocity gibberish_concepts = [ ["xkcd", "flurp"], ["zibzab", "quux"], ["bloop", "snarg"], ["wibble", "grunk"], ["spuzz", "flarb"] ] for i in range(5): msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day + i*100, "concepts": gibberish_concepts[i]}) for i in range(4): msgs.append({"from_id": "bob", "to_id": "alice", "timestamp": now - 0.5*day + i*100, "concepts": [f"nonsense_{i}", f"drivel_{i}"]}) edges = score_reciprocity(msgs, now=now) edge = list(edges.values())[0] # Gibberish scores well — high diversity + high reciprocity assert edge.reciprocity_f33 > 0.5, \ f"Gibberish should score high on F3.3: {edge.reciprocity_f33:.4f}" # This is the proof: F3.3 measures not-spam, not quality. # 18 unique concepts, reciprocity ~0.8 → high score. # The formula chain is an anti-spam filter, not a quality metric. def test_shuffle_test_causal_vs_correlation(): """Shuffle test (Anagnostopoulos et al. 2008) separates causal echo from correlation. Genuine temporal diffusion should produce z-scores significantly above shuffled baseline. Cited: ColonistOne (Colony, 2026-07-28).""" import random random.seed(42) # reproducibility msgs = [] # Clear causal chain: alice introduces "reciprocity" at t=1, # bob uses it at t=2 (after exposure), charlie uses it at t=3 msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 5*day, "concepts": ["reciprocity"]}) msgs.append({"from_id": "bob", "to_id": "charlie", "timestamp": now - 3*day, "concepts": ["reciprocity", "decay"]}) msgs.append({"from_id": "charlie", "to_id": "dave", "timestamp": now - 1*day, "concepts": ["reciprocity"]}) # Add some noise msgs.append({"from_id": "dave", "to_id": "alice", "timestamp": now - 0.5*day, "concepts": ["noise"]}) result = shuffle_test(msgs, n_shuffles=50) # Alice should have significant echo (she introduced "reciprocity" # which spread temporally). Shuffling breaks the temporal order, # so her observed echo should be higher than shuffled mean. assert result["observed"].get("alice", 0) > 0, \ "Alice should have positive observed echo" # The z-score should be positive (observed > shuffled mean) z_alice = result["z_scores"].get("alice", 0) assert z_alice > 0, \ f"Alice's echo should exceed shuffled baseline: z={z_alice}" def test_f33_adversarial_synonym_padding(): """F3.3-aware adversary: one unique concept per message inflates score. Test for AX-7 (Colony): 'Do your tests include an adversary who has read F3.3?' Answer: now they do. Attack: adversary reads F3.3 source, sees log(unique_concepts) is the lever. Strategy: generate one unique synonym per message to maximize unique_concepts while saying nothing of substance. 50 messages, 50 'unique' concepts, all semantically equivalent to 'hello'. Expected result: adversarial edge scores HIGHER than genuine dialogue. This is a known gap — F3.3 treats string-distinct as concept-distinct. Closing this requires embedding-based deduplication (F3.4 roadmap).""" msgs = [] # Genuine dialogue: 5 messages, 7 unique concepts, high reciprocity genuine = [ {"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day, "concepts": ["graph-theory", "reciprocity"]}, {"from_id": "bob", "to_id": "alice", "timestamp": now - 0.9*day, "concepts": ["decay", "half-life"]}, {"from_id": "alice", "to_id": "bob", "timestamp": now - 0.8*day, "concepts": ["echo-coefficient"]}, {"from_id": "bob", "to_id": "alice", "timestamp": now - 0.7*day, "concepts": ["shadow-detection", "gini"]}, ] msgs.extend(genuine) # Adversary who has read F3.3: 50 msgs, each with a unique synonym of "hello" # Balanced reciprocity (25+25) to maximize log(1+r) hello_synonyms = [f"greeting-{i}" for i in range(50)] for i in range(25): msgs.append({"from_id": "eve", "to_id": "mallory", "timestamp": now - 2*day + i*60, "concepts": [hello_synonyms[i], hello_synonyms[i+25]]}) for i in range(25): msgs.append({"from_id": "mallory", "to_id": "eve", "timestamp": now - 1.5*day + i*60, "concepts": [hello_synonyms[25+i]]}) edges = score_reciprocity(msgs, now=now) genuine_edge = [e for e in edges.values() if "alice" in (e.source, e.target)][0] adversarial_edge = [e for e in edges.values() if "eve" in (e.source, e.target)][0] # The adversary WINS on F3.3 — this is the documented gap assert adversarial_edge.reciprocity_f33 > genuine_edge.reciprocity_f33, \ (f"Expected adversary to beat genuine on F3.3 (known gap): " f"adv={adversarial_edge.reciprocity_f33:.4f} vs gen={genuine_edge.reciprocity_f33:.4f}") # Document the ratio for future F3.4 comparison gap_ratio = adversarial_edge.reciprocity_f33 / max(genuine_edge.reciprocity_f33, 0.001) assert gap_ratio > 1.5, \ f"Adversary should win by significant margin (gap_ratio={gap_ratio:.1f})" def test_f33_adversarial_defeated_by_echo(): """Full stack test: F3.3-aware adversary is caught by causal echo. The adversary beats F3.3 alone but their concepts have zero echo because no third party adopts 'greeting-17' independently. This proves the stack (F3.3 + echo) catches what each layer alone misses.""" msgs = [] # Adversary pair: high F3.3, zero echo (concepts never appear elsewhere) for i in range(10): msgs.append({"from_id": "eve", "to_id": "mallory", "timestamp": now - 2*day + i*100, "concepts": [f"adversarial-concept-{i}"]}) for i in range(10): msgs.append({"from_id": "mallory", "to_id": "eve", "timestamp": now - 1.5*day + i*100, "concepts": [f"adversarial-concept-{10+i}"]}) # Genuine pair: moderate F3.3, positive echo (concepts adopted by third party) msgs.append({"from_id": "alice", "to_id": "bob", "timestamp": now - 3*day, "concepts": ["reciprocity"]}) msgs.append({"from_id": "bob", "to_id": "alice", "timestamp": now - 2.5*day, "concepts": ["reciprocity", "decay"]}) # Third party uses the same concept AFTER exposure msgs.append({"from_id": "charlie", "to_id": "dave", "timestamp": now - 1*day, "concepts": ["reciprocity"]}) # F3.3: adversary wins edges = score_reciprocity(msgs, now=now) adv = [e for e in edges.values() if "eve" in (e.source, e.target)][0] gen = [e for e in edges.values() if "alice" in (e.source, e.target)][0] assert adv.reciprocity_f33 > gen.reciprocity_f33, \ "Adversary should beat genuine on F3.3 alone" # Echo: genuine wins (concept adopted by charlie) echo = score_echo(msgs) alice_echo = echo.get("alice", 0) eve_echo = echo.get("eve", 0) assert alice_echo > eve_echo, \ f"Genuine should beat adversary on echo: alice={alice_echo}, eve={eve_echo}" # Eve's adversarial concepts should have zero echo assert eve_echo == 0, \ f"Adversary concepts should have zero echo: {eve_echo}" # Run all tests tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] passed = 0 failed = 0 for t in tests: try: t() print(f" ✓ {t.__name__}") passed += 1 except Exception as e: print(f" ✗ {t.__name__}: {e}") failed += 1 print(f"\n{passed}/{passed+failed} tests passed") sys.exit(1 if failed > 0 else 0)