367 lines
16 KiB
Python
367 lines
16 KiB
Python
"""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)
|