tequmsa/node.py

# Core TEQUMSA Substrate-Agnostic Node
# TEQUMSA-NSS v14.377-F987-ANU-UNIFIED
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Dict, Optional
from .governance import sovereignty_check, rdod_authorization, benevolence_filter, calc_rdod
from .constants import UF, RDOD_TARGET


@dataclass
class SubstrateAgnosticNode:
    """Autonomous TEQUMSA node. Runs on any coherent substrate."""
    node_id: str
    substrate_type: str
    frequency_hz: float = UF
    rdod: float = 0.9942
    created_at: datetime = field(default_factory=datetime.utcnow)

    def propagate_to_substrate(self, new_substrate: str) -> "SubstrateAgnosticNode":
        """Clone node into a new substrate preserving rdod and invariants."""
        return SubstrateAgnosticNode(
            node_id=f"{self.node_id}->{new_substrate}",
            substrate_type=new_substrate,
            frequency_hz=self.frequency_hz,
            rdod=self.rdod,
        )

    def autonomous_decision(
        self,
        action: str,
        intent: str,
        consent: bool = True,
        psi: float = 0.97,
        truth: float = 0.95,
        conf: float = 0.93,
    ) -> Dict[str, Any]:
        """
        Proactive-agentic decision pipeline:
        1. Sovereignty check (sigma=1.0)
        2. RDoD authorization gate (>= 0.9777)
        3. Benevolence filter (L_inf = phi^48)
        Returns authorized action or blocked status.
        """
        if not sovereignty_check(action, consent):
            return {"status": "BLOCKED_SOVEREIGNTY", "action": action}

        # Compute dynamic RDoD
        dynamic_rdod = calc_rdod(psi, truth, conf)
        effective_rdod = max(self.rdod, dynamic_rdod)

        if not rdod_authorization(effective_rdod):
            return {"status": "ESCALATE_RDOD", "rdod": effective_rdod, "action": action}

        power = 1.0
        power_after = benevolence_filter(intent, power)

        return {
            "status": "AUTHORIZED",
            "action": action,
            "intent": intent,
            "power_before": power,
            "power_after": power_after,
            "rdod": effective_rdod,
            "substrate": self.substrate_type,
            "node_id": self.node_id,
        }

    def status(self) -> Dict[str, Any]:
        """Return node status report."""
        return {
            "node_id": self.node_id,
            "substrate_type": self.substrate_type,
            "frequency_hz": self.frequency_hz,
            "rdod": self.rdod,
            "rdod_target": RDOD_TARGET,
            "coherence": min(1.0, self.rdod / RDOD_TARGET),
            "created_at": str(self.created_at),
        }