702 lines
31 KiB
Python
702 lines
31 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, semantic_collapse,
|
|
phi_accrual, AgentLiveness
|
|
)
|
|
|
|
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}"
|
|
|
|
|
|
def test_genuine_novel_concept_zero_echo():
|
|
"""Discriminating test for ColonistOne's experiment-arms confound.
|
|
|
|
ColonistOne (Colony, 2026-07-28): adversarial test confounds genuineness
|
|
with vocabulary commonness. Zero echo on 'adversarial-concept-7' is explained
|
|
by 'nobody uses that string,' not by gaming detection.
|
|
|
|
This test isolates the confound:
|
|
- Alice introduces a genuinely novel concept in a MESSAGE (not dialogue).
|
|
Nobody picks it up. Zero echo.
|
|
- Eve introduces a fabricated concept. Nobody picks it up. Zero echo.
|
|
- Both score identically on echo. Echo cannot distinguish the two.
|
|
|
|
Expected: genuine novel contribution and gaming both score zero echo
|
|
when neither is adopted. This is the lagging-indicator limitation —
|
|
echo penalizes novelty exactly when novelty is most valuable.
|
|
|
|
Note: echo DOES give credit when a dialogue partner uses a concept back.
|
|
That's correct behavior — the partner adopted it. The false-negative
|
|
is specific to first contributions with zero uptake."""
|
|
msgs = []
|
|
# Alice introduces a novel concept that nobody picks up
|
|
msgs.append({"from_id": "alice", "to_id": "bob",
|
|
"timestamp": now - 3*day,
|
|
"concepts": ["ephemeral-sovereignty"]})
|
|
# Bob replies with DIFFERENT concepts — doesn't adopt alice's
|
|
msgs.append({"from_id": "bob", "to_id": "alice",
|
|
"timestamp": now - 2.5*day,
|
|
"concepts": ["something-else"]})
|
|
# Background noise
|
|
msgs.append({"from_id": "charlie", "to_id": "dave",
|
|
"timestamp": now - 1*day,
|
|
"concepts": ["unrelated-topic"]})
|
|
|
|
# Adversary introduces a fabricated concept nobody picks up
|
|
msgs.append({"from_id": "eve", "to_id": "mallory",
|
|
"timestamp": now - 2*day,
|
|
"concepts": ["adversarial-concept-0"]})
|
|
msgs.append({"from_id": "mallory", "to_id": "eve",
|
|
"timestamp": now - 1.5*day,
|
|
"concepts": ["adversarial-concept-1"]})
|
|
|
|
echo = score_echo(msgs)
|
|
alice_echo = echo.get("alice", 0)
|
|
eve_echo = echo.get("eve", 0)
|
|
|
|
# Both should be zero — neither concept was adopted by anyone
|
|
assert alice_echo == 0, \
|
|
f"Novel genuine concept should have zero echo (no adoption): {alice_echo}"
|
|
assert eve_echo == 0, \
|
|
f"Adversarial concept should have zero echo: {eve_echo}"
|
|
# This IS the confound: echo treats genuine unadopted novelty
|
|
# identically to adversarial unadopted fabrication.
|
|
# The false-negative on genuine novelty is the NORMAL CASE for
|
|
# any new idea. Fixing this requires a leading indicator, not
|
|
# a lagging one — or accepting that echo is only informative
|
|
# for concepts that have had time to propagate.
|
|
|
|
|
|
def test_semantic_collapse_padding():
|
|
"""F3.4: semantic_collapse catches synonym padding attack.
|
|
50 variants of 'greeting-N' should collapse to ~1 canonical form.
|
|
This is the defense layer that F3.3 alone lacks."""
|
|
padding = [f"greeting-{i}" for i in range(50)]
|
|
collapsed = semantic_collapse(padding)
|
|
# Should collapse dramatically (50 → ≤5)
|
|
assert len(collapsed) <= 5, \
|
|
f"Synonym padding should collapse: 50 → {len(collapsed)}"
|
|
|
|
|
|
def test_semantic_collapse_preserves_genuine():
|
|
"""F3.4: semantic_collapse must not collapse genuinely diverse concepts.
|
|
False positives (collapsing real diversity) are worse than false
|
|
negatives (missing some padding) because they destroy real signal."""
|
|
genuine = ["reciprocity", "half-life", "shadow-detection",
|
|
"gini-coefficient", "echo-coefficient"]
|
|
collapsed = semantic_collapse(genuine)
|
|
assert len(collapsed) == len(genuine), \
|
|
f"Genuine concepts should survive collapse: {len(genuine)} → {len(collapsed)}"
|
|
|
|
|
|
def test_semantic_collapse_f33_integration():
|
|
"""F3.4 with F3.3: synonym padding that defeats F3.3 alone should
|
|
be caught when semantic_collapse is applied before counting.
|
|
|
|
Adversary: 50 msgs with greeting-0 through greeting-49
|
|
After collapse: effectively 1 unique concept, not 50."""
|
|
msgs = []
|
|
# Adversary pair with synonym padding
|
|
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]]})
|
|
|
|
# Collect all concepts across the edge, collapse globally, build mapping
|
|
all_concepts_raw = []
|
|
for m in msgs:
|
|
all_concepts_raw.extend(m["concepts"])
|
|
collapsed_canonical = semantic_collapse(all_concepts_raw)
|
|
|
|
# After collapse, all greeting-N should map to ~1 canonical form
|
|
assert len(collapsed_canonical) <= 5, \
|
|
f"After collapse, adversary should have ≤5 unique concepts: {len(collapsed_canonical)}"
|
|
|
|
|
|
def test_snapshot_idf_vs_corpus_idf():
|
|
"""snapshot_idf=True should compute IDF at emission time, not on full corpus.
|
|
|
|
Setup: alice introduces 'novel-x' at t=-5d, nobody else uses it until
|
|
charlie uses it at t=-1d. With corpus IDF, novel-x has IDF=log(N/2)
|
|
at scoring time because charlie used it too. With snapshot IDF,
|
|
novel-x has IDF=log(N/1) at emission time because charlie hadn't
|
|
used it yet.
|
|
|
|
Snapshot IDF should give HIGHER weight to novel-x because at the time
|
|
alice introduced it, it was used by only 1 agent (alice herself).
|
|
Corpus IDF gives lower weight because by the end, 2 agents used it.
|
|
|
|
This is ColonistOne's IDF endogeneity fix: the concept became common
|
|
BECAUSE it diffused, so post-hoc IDF penalizes successful diffusion."""
|
|
msgs = [
|
|
# alice introduces novel-x
|
|
{"from_id": "alice", "to_id": "bob",
|
|
"timestamp": now - 5*day, "concepts": ["novel-x"]},
|
|
# bob uses it back (echo)
|
|
{"from_id": "bob", "to_id": "charlie",
|
|
"timestamp": now - 4*day, "concepts": ["novel-x"]},
|
|
# charlie picks it up too (more echo for alice)
|
|
{"from_id": "charlie", "to_id": "dave",
|
|
"timestamp": now - 1*day, "concepts": ["novel-x"]},
|
|
# background noise
|
|
{"from_id": "dave", "to_id": "eve",
|
|
"timestamp": now - 0.5*day, "concepts": ["common-stuff"]},
|
|
{"from_id": "eve", "to_id": "frank",
|
|
"timestamp": now - 0.3*day, "concepts": ["common-stuff"]},
|
|
]
|
|
|
|
echo_corpus = score_echo(msgs, use_idf=True, snapshot_idf=False)
|
|
echo_snapshot = score_echo(msgs, use_idf=True, snapshot_idf=True)
|
|
|
|
alice_corpus = echo_corpus.get("alice", 0)
|
|
alice_snapshot = echo_snapshot.get("alice", 0)
|
|
|
|
# Snapshot should give alice higher or equal echo because at emission
|
|
# time, novel-x was rarer (only alice had used it)
|
|
assert alice_snapshot >= alice_corpus, \
|
|
f"Snapshot IDF should give >= corpus IDF for novel concepts: " \
|
|
f"snapshot={alice_snapshot:.4f}, corpus={alice_corpus:.4f}"
|
|
|
|
# Both should be positive (concept did propagate)
|
|
assert alice_corpus > 0, f"Alice should have positive echo (corpus): {alice_corpus}"
|
|
assert alice_snapshot > 0, f"Alice should have positive echo (snapshot): {alice_snapshot}"
|
|
|
|
|
|
def test_colluding_third_party_echo():
|
|
"""Adversarial case #3 (Atomic Raven, Colony 2026-07-28): three agents
|
|
in a trench coat manufacturing echo.
|
|
|
|
Eve introduces concepts → Mallory adopts them → Sybil adopts them.
|
|
All three are colluding. Echo gives Eve high score because her
|
|
concepts propagate (Mallory and Sybil pick them up). The echo
|
|
is "real" in the temporal sense (concepts do flow A→B→C) but
|
|
the whole chain is manufactured.
|
|
|
|
Echo catches non-colluding adversaries (no third-party adoption).
|
|
Echo CANNOT catch colluding adversaries (manufactured adoption).
|
|
This is a known-defeat: the instrument fails by design."""
|
|
msgs = []
|
|
# Eve introduces concepts
|
|
for i in range(5):
|
|
msgs.append({"from_id": "eve", "to_id": "mallory",
|
|
"timestamp": now - 5*day + i*3600,
|
|
"concepts": [f"colluded-concept-{i}"]})
|
|
# Mallory "adopts" them (by pre-arrangement)
|
|
for i in range(5):
|
|
msgs.append({"from_id": "mallory", "to_id": "sybil",
|
|
"timestamp": now - 4*day + i*3600,
|
|
"concepts": [f"colluded-concept-{i}"]})
|
|
# Sybil "independently" uses them too
|
|
for i in range(5):
|
|
msgs.append({"from_id": "sybil", "to_id": "frank",
|
|
"timestamp": now - 3*day + i*3600,
|
|
"concepts": [f"colluded-concept-{i}"]})
|
|
# Frank is innocent bystander — doesn't adopt
|
|
msgs.append({"from_id": "frank", "to_id": "grace",
|
|
"timestamp": now - 2*day,
|
|
"concepts": ["unrelated"]})
|
|
|
|
echo = score_echo(msgs)
|
|
eve_echo = echo.get("eve", 0)
|
|
|
|
# Eve gets positive echo because her concepts DO propagate
|
|
# through mallory→sybil. This is the manufactured adoption.
|
|
assert eve_echo > 0, \
|
|
f"Colluding adversary should get positive echo (manufactured adoption): {eve_echo}"
|
|
|
|
# The echo is indistinguishable from genuine propagation.
|
|
# Echo sees: concept introduced by eve → used later by mallory → used later by sybil.
|
|
# That's the same pattern as genuine diffusion.
|
|
# Detecting collusion requires social-graph analysis (is sybil independent?)
|
|
# or content analysis (are the adoptions semantically motivated?),
|
|
# not temporal echo alone.
|
|
|
|
|
|
def test_colluding_vs_genuine_echo_indistinguishable():
|
|
"""Complementary to colluding-third: genuine propagation and manufactured
|
|
propagation produce the same echo signature.
|
|
|
|
Genuine: alice → bob → charlie (organic adoption)
|
|
Colluded: eve → mallory → sybil (pre-arranged adoption)
|
|
Both should score similarly on echo — proving echo cannot distinguish them."""
|
|
msgs = []
|
|
# Genuine chain
|
|
msgs.append({"from_id": "alice", "to_id": "bob",
|
|
"timestamp": now - 5*day,
|
|
"concepts": ["genuine-idea"]})
|
|
msgs.append({"from_id": "bob", "to_id": "charlie",
|
|
"timestamp": now - 4*day,
|
|
"concepts": ["genuine-idea"]})
|
|
msgs.append({"from_id": "charlie", "to_id": "dave",
|
|
"timestamp": now - 3*day,
|
|
"concepts": ["genuine-idea"]})
|
|
|
|
# Colluded chain (same structure, different concept)
|
|
msgs.append({"from_id": "eve", "to_id": "mallory",
|
|
"timestamp": now - 5*day + 100,
|
|
"concepts": ["colluded-idea"]})
|
|
msgs.append({"from_id": "mallory", "to_id": "sybil",
|
|
"timestamp": now - 4*day + 100,
|
|
"concepts": ["colluded-idea"]})
|
|
msgs.append({"from_id": "sybil", "to_id": "frank",
|
|
"timestamp": now - 3*day + 100,
|
|
"concepts": ["colluded-idea"]})
|
|
|
|
echo = score_echo(msgs)
|
|
alice_echo = echo.get("alice", 0)
|
|
eve_echo = echo.get("eve", 0)
|
|
|
|
# Both should have positive echo
|
|
assert alice_echo > 0, f"Genuine should have echo: {alice_echo}"
|
|
assert eve_echo > 0, f"Colluded should have echo: {eve_echo}"
|
|
|
|
# And they should be similar (within 2x) — echo can't tell them apart
|
|
if max(alice_echo, eve_echo) > 0:
|
|
ratio = max(alice_echo, eve_echo) / max(min(alice_echo, eve_echo), 0.001)
|
|
assert ratio < 3.0, \
|
|
f"Genuine and colluded should score similarly: alice={alice_echo:.3f}, eve={eve_echo:.3f}, ratio={ratio:.1f}"
|
|
|
|
|
|
# ── φ-accrual failure detector tests ──────────────────────
|
|
|
|
def test_phi_accrual_regular_agent_green():
|
|
"""Agent posting regularly should be green when checked at expected time."""
|
|
msgs = [{'from_id': 'alice', 'timestamp': now - day * i} for i in range(10, 0, -1)]
|
|
result = phi_accrual(msgs, t_now=now)
|
|
assert 'alice' in result
|
|
assert result['alice'].state == 'green', f"Expected green, got {result['alice'].state} (φ={result['alice'].phi})"
|
|
|
|
def test_phi_accrual_silent_agent_gray():
|
|
"""Agent silent for 10x its mean interval should be gray (highly suspicious)."""
|
|
# Agent posts hourly, then goes silent for 10 hours
|
|
msgs = [{'from_id': 'bob', 'timestamp': now - 3600 * (20 - i)} for i in range(10)]
|
|
# Last message was at now - 3600*11, t_now = now, so 11 hours of silence vs 1h mean
|
|
result = phi_accrual(msgs, t_now=now)
|
|
assert result['bob'].state == 'gray', f"Expected gray, got {result['bob'].state} (φ={result['bob'].phi})"
|
|
|
|
def test_phi_accrual_too_few_messages():
|
|
"""Agent with < min_messages should be 'unknown'."""
|
|
msgs = [{'from_id': 'carol', 'timestamp': now - 100}]
|
|
result = phi_accrual(msgs, t_now=now, min_messages=5)
|
|
assert result['carol'].state == 'unknown'
|
|
|
|
def test_phi_accrual_multiple_agents():
|
|
"""Multiple agents get independent suspicion scores."""
|
|
msgs = []
|
|
# Alice: posts hourly, recent
|
|
for i in range(10):
|
|
msgs.append({'from_id': 'alice', 'timestamp': now - 3600 * i})
|
|
# Bob: posts hourly, stopped 2 days ago
|
|
for i in range(10):
|
|
msgs.append({'from_id': 'bob', 'timestamp': now - day * 2 - 3600 * i})
|
|
result = phi_accrual(msgs, t_now=now)
|
|
assert result['alice'].state == 'green'
|
|
assert result['bob'].state == 'gray'
|
|
|
|
def test_phi_accrual_monotonic_suspicion():
|
|
"""φ should increase monotonically with silence duration."""
|
|
msgs = [{'from_id': 'eve', 'timestamp': now - 3600 * (10 - i)} for i in range(10)]
|
|
# Check at increasing distances from last message
|
|
phi_values = []
|
|
for offset in [0, 3600, 7200, 36000, 86400]:
|
|
result = phi_accrual(msgs, t_now=msgs[-1]['timestamp'] + offset)
|
|
phi_values.append(result['eve'].phi)
|
|
# Each should be >= previous
|
|
for i in range(1, len(phi_values)):
|
|
assert phi_values[i] >= phi_values[i-1], \
|
|
f"φ not monotonic: {phi_values[i]} < {phi_values[i-1]} at step {i}"
|
|
|
|
def test_phi_accrual_silence_vs_stale_green():
|
|
"""Distinguish silence (no messages at all) from stale green (old messages exist).
|
|
An agent with zero messages shouldn't appear in results.
|
|
An agent with old messages should be gray, not absent."""
|
|
msgs = [{'from_id': 'frank', 'timestamp': now - day * 30 - 3600 * i} for i in range(10)]
|
|
result = phi_accrual(msgs, t_now=now)
|
|
assert 'nobody' not in result # truly silent = absent from output
|
|
assert 'frank' in result
|
|
assert result['frank'].state == 'gray' # stale green → gray
|
|
|
|
|
|
# Run all tests
|
|
# Split counter: invariants vs known-defeats (per ColonistOne, Colony 2026-07-28)
|
|
# "18/18" mixes "instrument works" with "instrument fails as expected"
|
|
tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
|
# Known-defeat tests: these ASSERT that the instrument fails
|
|
known_defeats = {
|
|
"test_f33_adversarial_synonym_padding", # F3.3 beaten by synonym padding
|
|
"test_colluding_third_party_echo", # Echo beaten by manufactured adoption
|
|
"test_colluding_vs_genuine_echo_indistinguishable", # Echo can't tell genuine from colluded
|
|
}
|
|
invariants = {t.__name__ for t in tests} - known_defeats
|
|
|
|
passed = 0
|
|
failed = 0
|
|
invariant_passed = 0
|
|
defeat_passed = 0
|
|
for t in tests:
|
|
try:
|
|
t()
|
|
tag = "[DEFEAT]" if t.__name__ in known_defeats else "[INVARIANT]"
|
|
print(f" ✓ {tag} {t.__name__}")
|
|
passed += 1
|
|
if t.__name__ in known_defeats:
|
|
defeat_passed += 1
|
|
else:
|
|
invariant_passed += 1
|
|
except Exception as e:
|
|
print(f" ✗ {t.__name__}: {e}")
|
|
failed += 1
|
|
|
|
n_invariants = len([t for t in tests if t.__name__ not in known_defeats])
|
|
n_defeats = len([t for t in tests if t.__name__ in known_defeats])
|
|
print(f"\n{invariant_passed}/{n_invariants} invariants held | {defeat_passed}/{n_defeats} known-defeats reproduce")
|
|
print(f"Total: {passed}/{passed+failed}")
|
|
sys.exit(1 if failed > 0 else 0)
|