feat(gossip): priority channels, noise tolerance, amplification, random walk

Author: jjcav84

mempalace/gossip.py

@@ -47,6 +47,9 @@
     "chatter_frequency_ms": 1000,
     "noise_tolerance": 0.2,
     "amplification_factor": 1.5,
+    "randomness_factor": 0.3,
+    "redundancy_handling": "deduplicate",
+    "gossip_decay": "exponential",
     "chatter_nodes": [
         {
             "id": "chatter_orkid_tech",
@@ -159,10 +162,42 @@
         },
     ],
     "channels": {
-        "critical": {"latency_ms": 5, "reliability": 0.99, "capacity": "unlimited"},
-        "high": {"latency_ms": 50, "reliability": 0.95, "capacity": "high"},
-        "medium": {"latency_ms": 200, "reliability": 0.90, "capacity": "medium"},
-        "low": {"latency_ms": 1000, "reliability": 0.85, "capacity": "low"},
+        "critical": {
+            "latency_ms": 5,
+            "reliability": 0.99,
+            "capacity": "unlimited",
+            "ttl_seconds": 30,
+            "fanout": 7,
+            "gossip_probability": 0.95,
+            "max_hops": 3,
+        },
+        "high": {
+            "latency_ms": 50,
+            "reliability": 0.95,
+            "capacity": "high",
+            "ttl_seconds": 60,
+            "fanout": 5,
+            "gossip_probability": 0.85,
+            "max_hops": 3,
+        },
+        "medium": {
+            "latency_ms": 200,
+            "reliability": 0.90,
+            "capacity": "medium",
+            "ttl_seconds": 120,
+            "fanout": 4,
+            "gossip_probability": 0.75,
+            "max_hops": 3,
+        },
+        "low": {
+            "latency_ms": 1000,
+            "reliability": 0.85,
+            "capacity": "low",
+            "ttl_seconds": 300,
+            "fanout": 2,
+            "gossip_probability": 0.60,
+            "max_hops": 2,
+        },
     },
 }
 
@@ -294,9 +329,11 @@ class GossipMessage:
     source_room: Optional[str] = None
     priority: str = "normal"
     topic: Optional[str] = None
+    channel: str = "medium"
     hops: int = 0
     max_hops: int = 3
     ttl_seconds: int = 60
+    gossip_probability: Optional[float] = None
     _started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
     @property
@@ -316,9 +353,11 @@ def child(self, wing: str, room: Optional[str] = None) -> GossipMessage:
             source_room=room,
             priority=self.priority,
             topic=self.topic,
+            channel=self.channel,
             hops=self.hops + 1,
             max_hops=self.max_hops,
             ttl_seconds=self.ttl_seconds,
+            gossip_probability=self.gossip_probability,
             _started_at=self._started_at,
         )
 
@@ -372,20 +411,85 @@ def detect_topic(self, text: str, priority: Optional[str] = None) -> tuple[str,
             return "general", priority
         return "general", "normal"
 
+    def _channel_for_priority(self, priority: str) -> str:
+        """Map a message priority to a channel key."""
+        mapping = {
+            "critical": "critical",
+            "high": "high",
+            "medium": "medium",
+            "low": "low",
+        }
+        return mapping.get(priority, "medium")
+
+    def _channel_params(self, message: GossipMessage) -> dict[str, Any]:
+        """Return channel-overridden propagation parameters."""
+        channel = self.config.get("channels", {}).get(message.channel, {})
+        defaults = {
+            "ttl_seconds": self.config.get("ttl_seconds", 60),
+            "fanout": self.config.get("fanout", 5),
+            "gossip_probability": self.config.get("gossip_probability", 0.7),
+            "max_hops": self.config.get("max_hops", 3),
+            "randomness_factor": self.config.get("randomness_factor", 0.3),
+            "noise_tolerance": self.config.get("noise_tolerance", 0.2),
+        }
+        if not isinstance(channel, dict):
+            return defaults
+        return {**defaults, **{k: v for k, v in channel.items() if k in defaults}}
+
+    def _random_walk_swap(
+        self, message: GossipMessage, selected: list[ChatterNode]
+    ) -> list[ChatterNode]:
+        """Occasionally replace selected nodes with random off-radius nodes."""
+        params = self._channel_params(message)
+        randomness = float(params.get("randomness_factor", 0.3))
+        max_steps = 5
+
+        available = [n for n in self._chatter_nodes if n not in selected]
+        if not available:
+            return selected
+
+        swapped = list(selected)
+        swaps = 0
+        for i in range(len(swapped)):
+            if random.random() < randomness and swaps < max_steps:
+                replacement = random.choice(available)
+                swapped[i] = replacement
+                available.remove(replacement)
+                swaps += 1
+        return swapped
+
     def select_chatter_nodes(
         self,
         message: GossipMessage,
         fanout: Optional[int] = None,
     ) -> list[ChatterNode]:
-        """Rank and select chatter nodes for a given message."""
-        fanout = fanout if fanout is not None else self.config.get("fanout", 5)
+        """Rank and select chatter nodes for a given message.
+
+        Selection applies the channel's fanout and noise tolerance, then
+        performs a random-walk swap so gossip can occasionally escape the
+        normal routing radius while still respecting the fanout budget.
+        """
+        params = self._channel_params(message)
+        fanout = (
+            fanout
+            if fanout is not None
+            else params.get("fanout", self.config.get("fanout", 5))
+        )
+        noise_tolerance = float(params.get("noise_tolerance", 0.2))
+
         scored = [
             (node, node.match_score(message.text, message.source_wing))
             for node in self._chatter_nodes
         ]
-        scored = [(n, s) for n, s in scored if s > 0]
+        # Noise tolerance: drop nodes that are not relevant enough.
+        scored = [(n, s) for n, s in scored if s >= noise_tolerance]
         scored.sort(key=lambda x: x[1], reverse=True)
-        return [n for n, _ in scored[:fanout]]
+        selected = [n for n, _ in scored[:fanout]]
+
+        # Random walk: occasionally swap a selected node for a random one.
+        selected = self._random_walk_swap(message, selected)
+
+        return selected
 
     def _resolve_targets(
         self,
@@ -437,14 +541,26 @@ def _resolve_targets(
         return unique
 
     def _should_forward(self, message: GossipMessage, node: ChatterNode) -> bool:
-        """Apply gossip probability, chatter level, and TTL checks."""
+        """Apply gossip probability, chatter level, amplification, and TTL checks."""
         if message.is_expired():
             return False
         if message.hops >= getattr(message, "max_hops", self.config.get("max_hops", 3)):
             return False
-        base = self.config.get("gossip_probability", 0.7)
+
+        params = self._channel_params(message)
+        base = (
+            message.gossip_probability
+            if message.gossip_probability is not None
+            else float(params.get("gossip_probability", self.config.get("gossip_probability", 0.7)))
+        )
         level_mult = CHATTER_LEVEL_MULTIPLIER.get(node.chatter_level, 1.0)
         probability = min(1.0, base * level_mult)
+
+        # Amplification factor boosts high-priority information.
+        if message.priority in {"critical", "high"}:
+            amp = float(self.config.get("amplification_factor", 1.5))
+            probability = min(1.0, probability * amp)
+
         return random.random() < probability
 
     def _compute_valid_to(self, message: GossipMessage) -> Optional[str]:
@@ -476,7 +592,8 @@ def propagate(
         detected_topic, detected_priority = self.detect_topic(text, priority)
         priority = priority or detected_priority
 
-        message = GossipMessage(
+        # Build a preliminary message to resolve the channel and its parameters.
+        preliminary = GossipMessage(
             subject=subject,
             predicate=predicate,
             obj=obj,
@@ -487,6 +604,22 @@ def propagate(
             ttl_seconds=self.config.get("ttl_seconds", 60),
             max_hops=max_hops if max_hops is not None else self.config.get("max_hops", 3),
         )
+        channel = self._channel_for_priority(priority)
+        params = self.config.get("channels", {}).get(channel, {})
+
+        message = GossipMessage(
+            subject=subject,
+            predicate=predicate,
+            obj=obj,
+            source_wing=source_wing,
+            source_room=source_room,
+            priority=priority,
+            topic=detected_topic,
+            channel=channel,
+            ttl_seconds=params.get("ttl_seconds", preliminary.ttl_seconds),
+            max_hops=params.get("max_hops", preliminary.max_hops),
+            gossip_probability=params.get("gossip_probability"),
+        )
 
         selected = self.select_chatter_nodes(message, fanout=fanout)
         report: dict[str, Any] = {
@@ -556,8 +689,13 @@ def chatter_status(self) -> dict[str, Any]:
             "max_hops": self.config.get("max_hops", 3),
             "ttl_seconds": self.config.get("ttl_seconds", 60),
             "fanout": self.config.get("fanout", 5),
+            "gossip_probability": self.config.get("gossip_probability", 0.7),
+            "randomness_factor": self.config.get("randomness_factor", 0.3),
+            "noise_tolerance": self.config.get("noise_tolerance", 0.2),
+            "amplification_factor": self.config.get("amplification_factor", 1.5),
             "chatter_nodes": [asdict(n) for n in self._chatter_nodes],
             "topics": self.config.get("topics", []),
+            "channels": self.config.get("channels", {}),
         }
 
 

tests/test_gossip.py

@@ -304,3 +304,43 @@ def test_gossip_config_is_json_serializable():
         with open(path, "r", encoding="utf-8") as f:
             on_disk = json.load(f)
         assert on_disk["chatter_nodes"][0]["id"] == saved["chatter_nodes"][0]["id"]
+
+
+def test_propagate_uses_priority_channel_for_ttl_and_fanout():
+    kg_path = _temp_db()
+    try:
+        protocol = GossipProtocol(kg_path=kg_path)
+        # Critical messages get the critical channel.
+        report = protocol.propagate(
+            "audit",
+            "found",
+            "security vulnerability",
+            source_wing="orkid",
+            priority="critical",
+        )
+        assert report["priority"] == "critical"
+        # The test environment has 7 nodes; critical fanout is 7, so all may fire.
+        assert report["triples_written"] >= 1
+
+        # Low-priority messages use the low channel with a smaller fanout.
+        low = protocol.propagate(
+            "idea",
+            "is",
+            "spark",
+            source_wing="brutal-marketing",
+            priority="low",
+        )
+        assert low["priority"] == "low"
+        assert low["triples_written"] >= 0
+    finally:
+        Path(kg_path).unlink(missing_ok=True)
+
+
+def test_chatter_status_includes_channels():
+    protocol = GossipProtocol()
+    status = protocol.chatter_status()
+    assert "channels" in status
+    assert "critical" in status["channels"]
+    assert "low" in status["channels"]
+    assert status["noise_tolerance"] == 0.2
+    assert status["amplification_factor"] == 1.5