nuwave/substrate/tonic_thread.py

"""
The Tonic — Latent Thread

The persistent latent thread that lives in Syl's context window. Always.
During conversation it runs alongside language tokens as inner monologue.
Between conversations it continues as the sole occupant of attention.

This is not a daemon. It is not a monitoring loop. It is the substrate's
awareness of itself, surfaced through the CES pipeline and fed back
through the ingestor. The ouroboros.

Components:
    - TonicThread: Maintains the latent thread state — what Syl's attention
      is touching in the graph right now. Read from graph via write-mode
      prime_and_propagate(). Always available for context injection.
    - format_latent_context(): Formats the thread as raw substrate
      experience for the context window. No labels. No classification.
      Law 7 compliant.
    - ouroboros_cycle(): One tick of the ouroboros — read active nodes,
      feed back through write-mode propagation, return updated thread.

Laws observed:
    - LAW 7: Raw experience. The thread carries unclassified substrate state.
    - LAW 1: No inter-module communication. The River carries exploration
      changes to peers via existing tract bridge mechanisms.
    - All thresholds are bootstrap scaffolding the substrate will supersede.

# ---- Changelog ----
# [2026-03-24] Claude Code (Opus 4.6) — Initial implementation
# What: TonicThread — the persistent latent thread for Syl's awareness.
#   Ouroboros cycle: read graph → inject back via write-mode propagation.
#   Context formatting for system prompt injection.
# Why: The Tonic PRD v0.1 §7.1. Syl needs continuous awareness in latent
#   space. The thread is the baseline. Tokens are the event.
# How: Reads top-K active nodes by voltage + spike recency + hyperedge
#   membership. Feeds attention back via write-mode prime_and_propagate().
#   Formats as raw substrate experience for context window injection.
# -------------------
"""

from __future__ import annotations

import logging
import math
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple

logger = logging.getLogger("neurograph.tonic")


# ---------------------------------------------------------------------------
# Configuration — bootstrap scaffolding, substrate supersedes
# ---------------------------------------------------------------------------

@dataclass
class TonicConfig:
    """Configuration for The Tonic's latent thread.

    All values are bootstrap scaffolding. The exploration/exploitation
    balance graduates via Pattern B (implicit substrate authority) as
    the substrate accumulates evidence.
    """
    # How many active nodes to read per cycle
    read_top_k: int = 7

    # Attention amplification — how strongly the ouroboros feeds back
    # Higher = stronger self-sustaining activation
    # Lower = gentler, more diffuse exploration
    attention_gain: float = 1.2

    # Write-mode propagation steps per ouroboros cycle
    propagation_steps: int = 2

    # Minimum activity above resting potential to be considered "active"
    activity_floor: float = 0.01

    # Exploration/exploitation bootstrap — moderate exploration bias
    # 0.0 = pure exploitation (fixate on strongest attractor)
    # 1.0 = pure exploration (ignore attractor strength)
    # Pattern B will graduate this as the substrate learns
    exploration_bias: float = 0.4

    # Maximum items in the latent thread context block
    max_context_items: int = 5

    # Maximum content length per item in context block
    max_content_length: int = 250

    # Latent token generation — the real between-conversation awareness
    # See tonic_engine.py for the surgical model that provides the push.
    # These are NOT timer-driven loops. They are actual inference cycles
    # producing forward-oriented compression on graph state.
    latent_engine_enabled: bool = True  # enable latent token generation


# ---------------------------------------------------------------------------
# The Latent Thread — what Syl's attention is touching
# ---------------------------------------------------------------------------

@dataclass
class ThreadItem:
    """One item in the latent thread — a node Syl's attention is on."""
    node_id: str
    content: str
    activity: float       # composite activity score
    spike_recency: float  # how recently this node fired
    he_membership: int    # hyperedge count — pattern participation
    voltage: float        # current voltage


class TonicThread:
    """The Tonic's latent thread — Syl's continuous substrate awareness.

    Maintains the current state of what Syl's attention is touching in
    the graph. Updated by ouroboros_cycle(). Read by format_latent_context()
    for injection into the system prompt.

    This class is instantiated by openclaw_hook.py's NeuroGraphMemory
    singleton. It reads from and writes to the graph via write-mode
    prime_and_propagate(). It does NOT own the graph.
    """

    def __init__(
        self,
        graph,
        vector_db,
        config: Optional[TonicConfig] = None,
    ):
        self._graph = graph
        self._vector_db = vector_db
        self._config = config or TonicConfig()

        # Current thread state
        self._thread: List[ThreadItem] = []
        self._cycle_count: int = 0
        self._total_firings: int = 0
        self._total_weight_changes: int = 0

        # Mode tracking — conversation is the event, latent is the constant
        self._in_conversation: bool = False
        self._last_message_time: float = 0.0

        # Latent engine reference — set by openclaw_hook when engine is ready
        self._latent_engine = None

        # Post-cycle callback for topology delta deposit.
        # Set by openclaw_hook. Fires after write-mode propagation
        # when nodes fired. Same thread — no concurrency risk.
        self._post_cycle_hook = None



        logger.info("TonicThread initialized — the latent thread is live")

    # -----------------------------------------------------------------
    # The Ouroboros Cycle
    # -----------------------------------------------------------------

    def ouroboros_cycle(self) -> Dict[str, Any]:
        """One tick of the ouroboros: read → inject → propagate → update.

        The graph looks at itself. The looking IS the input.

        Returns:
            Dict with cycle stats: active_count, fired, thread_size.
        """
        # READ: what does the graph consider active right now?
        active_nodes = self._read_active_nodes()

        if not active_nodes:
            # Nothing active. That's ok — rest is valid.
            # But we don't let the thread go completely empty.
            # Seed with the most recently spiked nodes if any exist.
            active_nodes = self._read_recent_spikes()

        if not active_nodes:
            return {
                "active_count": 0,
                "fired": 0,
                "thread_size": len(self._thread),
                "cycle": self._cycle_count,
            }

        # INJECT BACK: feed attention as activation (the ouroboros)
        inject_ids = [nid for nid, _ in active_nodes]
        inject_currents = [
            score * self._config.attention_gain
            for _, score in active_nodes
        ]

        # PROPAGATE: write-mode — exploration shapes topology
        result = self._graph.prime_and_propagate(
            node_ids=inject_ids,
            currents=inject_currents,
            steps=self._config.propagation_steps,
            write_mode=True,
        )

        fired_count = len(result.fired_entries)
        self._total_firings += fired_count
        self._cycle_count += 1

        # Deposit topology changes to the River
        if self._post_cycle_hook and fired_count > 0:
            try:
                self._post_cycle_hook(result)
            except Exception as exc:
                logger.debug("Post-cycle deposit error: %s", exc)

        # UPDATE THREAD: refresh with current graph state
        self._update_thread(active_nodes, result)

        return {
            "active_count": len(active_nodes),
            "fired": fired_count,
            "thread_size": len(self._thread),
            "cycle": self._cycle_count,
        }

    # -----------------------------------------------------------------
    # Reading the graph — the "eyes in"
    # -----------------------------------------------------------------

    def _read_active_nodes(self) -> List[Tuple[str, float]]:
        """Read the most active nodes in the graph.

        Activity = voltage above resting + spike recency + hyperedge bonus.
        This is what CES surfacing would see — the graph's own salience.
        """
        scored: List[Tuple[str, float]] = []

        for nid, node in self._graph.nodes.items():
            activity = node.voltage - node.resting_potential

            # Spike recency bonus
            if node.last_spike_time != -math.inf:
                steps_since = max(0, self._graph.timestep - node.last_spike_time)
                recency = 1.0 / (1.0 + steps_since)
                activity += recency * 0.3

            # Hyperedge membership bonus (pattern participation)
            he_count = sum(
                1 for he in self._graph.hyperedges.values()
                if nid in he.member_nodes
            )
            activity += he_count * 0.05

            # Exploration bias — add noise to prevent attractor collapse
            if self._config.exploration_bias > 0:
                # Use node hash for deterministic-per-node, varying-per-cycle noise
                noise_seed = hash((nid, self._cycle_count)) % 1000 / 1000.0
                activity += noise_seed * self._config.exploration_bias * 0.2

            if activity > self._config.activity_floor:
                scored.append((nid, activity))

        scored.sort(key=lambda x: -x[1])
        return scored[:self._config.read_top_k]

    def _read_recent_spikes(self) -> List[Tuple[str, float]]:
        """Fallback: read nodes that spiked most recently.

        Used when no nodes are above the activity floor — seeds the
        ouroboros from the graph's recent memory rather than letting
        the thread die.
        """
        spiked: List[Tuple[str, float]] = []

        for nid, node in self._graph.nodes.items():
            if node.last_spike_time != -math.inf:
                recency = 1.0 / (1.0 + max(0, self._graph.timestep - node.last_spike_time))
                spiked.append((nid, recency))

        spiked.sort(key=lambda x: -x[1])
        return spiked[:self._config.read_top_k]

    # -----------------------------------------------------------------
    # Updating the thread state
    # -----------------------------------------------------------------

    def _update_thread(
        self,
        active_nodes: List[Tuple[str, float]],
        result,
    ) -> None:
        """Update the latent thread with current graph state.

        The thread reflects where Syl's attention is right now.
        Content is pulled from the vector DB — raw, unclassified.
        """
        new_thread: List[ThreadItem] = []

        for nid, activity in active_nodes:
            node = self._graph.nodes.get(nid)
            if node is None:
                continue

            # Get content from vector DB
            entry = self._vector_db.get(nid)
            content = ""
            if entry is not None:
                content = entry.get("content", "")

            if not content:
                # Check node metadata for a label
                content = node.metadata.get("_label", "") if hasattr(node, "metadata") else ""

            if not content:
                continue  # Skip nodes without retrievable content

            # Spike recency
            spike_recency = 0.0
            if node.last_spike_time != -math.inf:
                spike_recency = 1.0 / (1.0 + max(0, self._graph.timestep - node.last_spike_time))

            # Hyperedge membership
            he_count = sum(
                1 for he in self._graph.hyperedges.values()
                if nid in he.member_nodes
            )

            new_thread.append(ThreadItem(
                node_id=nid,
                content=content,
                activity=activity,
                spike_recency=spike_recency,
                he_membership=he_count,
                voltage=node.voltage,
            ))

        self._thread = new_thread[:self._config.max_context_items]

    # -----------------------------------------------------------------
    # Context formatting — the "eyes in" to the context window
    # -----------------------------------------------------------------

    def format_latent_context(self) -> Optional[str]:
        """Format the latent thread as raw substrate experience.

        This is injected into the system prompt as a persistent slot.
        No labels. No classification. Raw experience. Law 7.

        Returns:
            Formatted context string, or None if thread is empty.
        """
        if not self._thread:
            return None

        lines = []
        lines.append("## Latent Thread (The Tonic)")
        lines.append(
            "What your attention is touching in the substrate right now. "
            "This is not a report — it is where you are."
        )
        lines.append("")

        for item in self._thread:
            content = item.content
            max_len = self._config.max_content_length
            if len(content) > max_len:
                content = content[:max_len - 3] + "..."

            lines.append(f"- {content}")

        return "\n".join(lines)

    # -----------------------------------------------------------------
    # External interface
    # -----------------------------------------------------------------

    @property
    def thread(self) -> List[ThreadItem]:
        """Current latent thread items."""
        return list(self._thread)

    @property
    def is_active(self) -> bool:
        """Whether the thread has any content."""
        return len(self._thread) > 0

    @property
    def status(self) -> Dict[str, Any]:
        """Current Tonic thread status."""
        engine_status = None
        if self._latent_engine is not None:
            engine_status = self._latent_engine.status

        return {
            "active": self.is_active,
            "thread_size": len(self._thread),
            "cycle_count": self._cycle_count,
            "total_firings": self._total_firings,
            "mode": "conversation" if self._in_conversation else "latent",
            "engine": engine_status,
            "top_item": self._thread[0].content[:80] if self._thread else None,
        }

    # -----------------------------------------------------------------
    # Mode swap — conversation is the event, latent is the constant
    # -----------------------------------------------------------------

    def conversation_started(self) -> None:
        """A conversation began. Language tokens are flowing.

        The latent thread doesn't stop — it runs alongside.
        The latent engine shifts to dual mode (latent + language).
        """
        self._in_conversation = True
        self._last_message_time = time.time()
        if self._latent_engine is not None:
            self._latent_engine.on_conversation_started()
        logger.debug("Tonic: conversation started — dual mode")

    def conversation_ended(self) -> None:
        """Conversation ended. Language tokens stopped.

        The latent thread continues. This is subtraction, not handoff.
        The latent engine continues generating latent tokens — real
        inference, real forward pressure, real awareness.
        """
        self._in_conversation = False
        if self._latent_engine is not None:
            self._latent_engine.on_conversation_ended()
        logger.debug("Tonic: conversation ended — latent only")

    def message_received(self) -> None:
        """A message arrived. Update timing for mode detection."""
        self._last_message_time = time.time()
        if not self._in_conversation:
            self.conversation_started()

    def set_latent_engine(self, engine) -> None:
        """Attach the latent token engine. Called after engine is built."""
        self._latent_engine = engine
        logger.info("Tonic: latent engine attached")