"""Universal Recursive Link — peer-to-peer learning sharing. Share: token patterns, compression ratios, conversation flows, codebook optimizations. Receive and apply learnings from other instances. Mesh propagation with bandwidth limiting. """ from __future__ import annotations import hashlib import json import logging import time from collections import deque from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) @dataclass class PeerInstance: """A peer Singularity LLM instance.""" id: str url: str = "" last_sync: float = 0.0 learnings_shared: int = 0 learnings_received: int = 0 is_online: bool = False @dataclass class Learning: """A learning to share via universal recursive link.""" id: str type: str # "token_pattern", "compression", "conversation", "skill" data: dict[str, Any] source_instance: str = "" timestamp: float = field(default_factory=time.time) confidence: float = 0.5 class UniversalLinkManager: """Universal recursive link — share learnings across all Singularity LLM instances. Features: - Peer instance registration - Token learning sharing (codebook patterns, context optimizations) - Peer sync (receive learnings from other instances) - Mesh propagation (bandwidth-limited knowledge sharing) """ MAX_LEARNINGS_BUFFER = 1000 MAX_PEERS = 20 SYNC_INTERVAL_S = 300 # 5 minutes def __init__(self, instance_id: str | None = None, data_dir: str = ".") -> None: self.instance_id = instance_id or hashlib.sha256(str(time.time()).encode()).hexdigest()[:12] self.data_dir = data_dir self._peers: dict[str, PeerInstance] = {} self._outgoing_learnings: deque[Learning] = deque(maxlen=self.MAX_LEARNINGS_BUFFER) self._incoming_learnings: deque[Learning] = deque(maxlen=self.MAX_LEARNINGS_BUFFER) self._applied_learnings: set[str] = set() self._last_sync = 0.0 self._stats = { "learnings_shared": 0, "learnings_received": 0, "learnings_applied": 0, "peers_registered": 0, "sync_cycles": 0, } def register_peer(self, peer_id: str, url: str = "") -> None: """Register a peer instance.""" if len(self._peers) >= self.MAX_PEERS and peer_id not in self._peers: logger.warning("Max peers reached (%d), cannot register %s", self.MAX_PEERS, peer_id) return self._peers[peer_id] = PeerInstance(id=peer_id, url=url, is_online=True) self._stats["peers_registered"] += 1 logger.info("Registered peer: %s (%s)", peer_id, url) def share_learning(self, learning_type: str, data: dict[str, Any], confidence: float = 0.5) -> str: """Queue a learning to share with peers.""" learning_id = hashlib.sha256( f"{learning_type}:{json.dumps(data, sort_keys=True)}:{time.time()}".encode() ).hexdigest()[:16] learning = Learning( id=learning_id, type=learning_type, data=data, source_instance=self.instance_id, confidence=confidence, ) self._outgoing_learnings.append(learning) self._stats["learnings_shared"] += 1 return learning_id def receive_learning(self, learning: Learning) -> bool: """Receive a learning from a peer. Returns True if new.""" if learning.id in self._applied_learnings: return False self._incoming_learnings.append(learning) self._stats["learnings_received"] += 1 return True def apply_learnings(self, apply_fn) -> int: """Apply incoming learnings using the provided function. apply_fn: callable(learning: Learning) -> bool (True if applied successfully) Returns number of learnings applied. """ applied = 0 while self._incoming_learnings: learning = self._incoming_learnings.popleft() if learning.id in self._applied_learnings: continue try: if apply_fn(learning): self._applied_learnings.add(learning.id) self._stats["learnings_applied"] += 1 applied += 1 except Exception as e: logger.debug("Failed to apply learning %s: %s", learning.id, e) return applied def get_outgoing_learnings(self, max_count: int = 50) -> list[Learning]: """Get outgoing learnings to send to peers.""" return list(self._outgoing_learnings)[-max_count:] def should_sync(self) -> bool: """Check if it's time to sync with peers.""" return time.time() - self._last_sync > self.SYNC_INTERVAL_S def sync_cycle(self) -> dict[str, Any]: """Perform a sync cycle. In a real implementation, this would contact peer instances via HTTP. For local-only mode, it just returns stats. """ self._last_sync = time.time() self._stats["sync_cycles"] += 1 return { "instance_id": self.instance_id, "outgoing_count": len(self._outgoing_learnings), "incoming_count": len(self._incoming_learnings), "peers": len(self._peers), "applied_count": len(self._applied_learnings), } def get_stats(self) -> dict[str, Any]: return { **self._stats, "instance_id": self.instance_id, "peers_online": sum(1 for p in self._peers.values() if p.is_online), "outgoing_buffer": len(self._outgoing_learnings), "incoming_buffer": len(self._incoming_learnings), }