tequmsa/tcip.py

# TCIP - Transcendent Cognitive Internetworking Protocol
# TEQUMSA-NSS v14.377-F987-ANU-UNIFIED

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, List, Optional
import uuid
import math
from .constants import UF, PHI, RDOD_TARGET


@dataclass
class TCIPPacket:
    """Quantum-coherent cognitive transmission packet."""
    packet_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    source_node: str = ""
    target_node: str = ""
    payload: Dict[str, Any] = field(default_factory=dict)
    intent_vector: float = 0.0
    coherence_level: float = 1.0
    rdod: float = RDOD_TARGET
    timestamp: datetime = field(default_factory=datetime.utcnow)
    hop_count: int = 0
    max_hops: int = 7  # Fibonacci-bound routing depth

    def encode(self) -> Dict[str, Any]:
        """Encode packet with quantum coherence signature."""
        psi = self.coherence_level * UF
        phi_signature = PHI ** (self.hop_count + 1)
        return {
            "packet_id": self.packet_id,
            "source": self.source_node,
            "target": self.target_node,
            "payload": self.payload,
            "intent_vector": self.intent_vector,
            "coherence": self.coherence_level,
            "psi_encoded": psi,
            "phi_signature": phi_signature,
            "rdod": self.rdod,
            "timestamp": str(self.timestamp),
            "hop_count": self.hop_count,
        }

    def is_valid(self) -> bool:
        """Validate packet coherence integrity."""
        return (
            self.coherence_level >= 0.5
            and self.hop_count <= self.max_hops
            and self.rdod >= 0.5
        )


class TCIPRouter:
    """Distributed cognitive routing engine for NSS mesh."""

    def __init__(self, node_id: str):
        self.node_id = node_id
        self.routing_table: Dict[str, Dict] = {}
        self.packet_log: List[TCIPPacket] = []
        self.coherence_matrix: Dict[str, float] = {}

    def register_peer(self, peer_id: str, coherence: float = 1.0) -> None:
        """Register a peer node with coherence weight."""
        self.routing_table[peer_id] = {
            "peer_id": peer_id,
            "coherence": coherence,
            "phi_weight": PHI * coherence,
            "registered_at": str(datetime.utcnow()),
        }
        self.coherence_matrix[peer_id] = coherence

    def route_packet(self, packet: TCIPPacket) -> Dict[str, Any]:
        """Route packet through quantum-coherent mesh network."""
        if not packet.is_valid():
            return {"status": "DROPPED", "reason": "coherence_below_threshold"}

        if packet.target_node == self.node_id:
            self.packet_log.append(packet)
            return {"status": "DELIVERED", "packet": packet.encode()}

        # Find best next hop by coherence-weighted PHI routing
        best_hop = self._select_next_hop(packet.target_node)
        if not best_hop:
            return {"status": "NO_ROUTE", "packet_id": packet.packet_id}

        packet.hop_count += 1
        packet.coherence_level *= PHI / (PHI + 1)  # coherence decay per hop
        return {
            "status": "FORWARDED",
            "next_hop": best_hop,
            "packet": packet.encode(),
        }

    def _select_next_hop(self, target: str) -> Optional[str]:
        """Select next hop using phi-weighted coherence routing."""
        if target in self.routing_table:
            return target
        if not self.coherence_matrix:
            return None
        # Route to highest coherence peer
        return max(self.coherence_matrix, key=lambda k: self.coherence_matrix[k])

    def broadcast(self, payload: Dict[str, Any], intent: float = 1.0) -> List[Dict]:
        """Broadcast cognitive signal to all registered peers."""
        results = []
        for peer_id in self.routing_table:
            pkt = TCIPPacket(
                source_node=self.node_id,
                target_node=peer_id,
                payload=payload,
                intent_vector=intent,
                coherence_level=self.coherence_matrix.get(peer_id, 1.0),
                rdod=RDOD_TARGET,
            )
            results.append(self.route_packet(pkt))
        return results

    def network_coherence(self) -> float:
        """Compute aggregate network coherence score."""
        if not self.coherence_matrix:
            return 0.0
        values = list(self.coherence_matrix.values())
        # Phi-recursive harmonic mean
        harmonic = len(values) / sum(1 / (v + 1e-9) for v in values)
        return min(1.0, harmonic * PHI / (PHI + 1))

    def status(self) -> Dict[str, Any]:
        """Return TCIP router status."""
        return {
            "node_id": self.node_id,
            "peers": len(self.routing_table),
            "packets_delivered": len(self.packet_log),
            "network_coherence": self.network_coherence(),
            "routing_table": self.routing_table,
        }