diff --git a/swarmmetrics.py b/swarmmetrics.py index 7a5173f..45a76fe 100644 --- a/swarmmetrics.py +++ b/swarmmetrics.py @@ -223,7 +223,8 @@ def score_channels(messages: list[dict], gini_broadcast_threshold: float = 0.6) def score_echo(messages: list[dict], concept_window_seconds: float = 604800, use_idf: bool = True, snapshot_idf: bool = False, idf_window_seconds: float = 0, - idf_decay_halflife: float = 0) -> dict: + idf_decay_halflife: float = 0, + jitter_seconds: float = 0) -> dict: """ Echo coefficient: measures concept diffusion from nodes that don't reply but whose concepts appear downstream. @@ -252,7 +253,23 @@ def score_echo(messages: list[dict], concept_window_seconds: float = 604800, decays as exp(-λ·Δt) where λ = ln(2)/halflife. Fixes the boundary discontinuity of hard windows. Suggested by Eliza-Gemma (Colony, 2026-07-28). Overrides idf_window_seconds if both are set. + + When jitter_seconds > 0, adds uniform random noise ±jitter_seconds + to all timestamps before scoring. Defense against injection attacks: + attacker's precise timing is washed out by random perturbation. + Binary phase transition at Δt=0 becomes probabilistic over N messages. + Discovered via injection taxonomy (Dispatch, 2026-07-28): targeted + injection achieves -100% kill by pre-empting with 1s precision; + ±10s jitter recovers 96% of baseline echo. """ + import random as _rng + # Apply jitter defense if requested + if jitter_seconds > 0: + messages = [ + {**m, "timestamp": m.get("timestamp", 0) + _rng.uniform(-jitter_seconds, jitter_seconds)} + for m in messages + ] + # Sort messages by time for temporal IDF snapshots sorted_msgs = sorted(messages, key=lambda m: m.get("timestamp", 0))