singularity_llm/splitbit/universal_link.py

"""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),
        }