Sync from GitHub: e9899bcdad8d149f9293a565ce52746d8e59e59b

nuwave/organism.py

@@ -631,8 +631,25 @@ class NuWaveOrganism:
                 "prime_strength": 1.0,
                 "learning_rate": 0.08,
                 "surprise_reward_scaling": 1.5,
+                # Substrate-feature activation — mirrors Faux_Clawdbot's
+                # 2026-04-16 HF Tonic deployment. These flip dormant
+                # canonical mechanisms into the active pipeline:
+                #  - tonic.enabled: ouroboros runs in heuristic mode
+                #  - three_factor_enabled: reward learning actually fires
+                #    on inject_reward (was default-off; record_outcome's
+                #    learning signal was being silently dropped)
+                #  - scaling_interval=25 (was default 100): homeostatic
+                #    scaling actually fires on ephemeral worker timescales
+                #  - he_*: hyperedge formation + pattern completion knobs
+                "tonic": {"enabled": True},
+                "three_factor_enabled": True,
+                "scaling_interval": 25,
+                "threshold_ceiling": 5.0,
+                "he_pattern_completion_strength": 0.3,
+                "he_member_weight_lr": 0.05,
+                "he_threshold_lr": 0.02,
             })
-            logger.info("Substrate initialized: full NeuroGraph SNN")
+            logger.info("Substrate initialized: full NeuroGraph SNN, Tonic+HE config active")
         except Exception as exc:
             logger.error("NeuroGraph init failed: %s", exc)
             if _substrate_dir in sys.path:
@@ -659,6 +676,55 @@ class NuWaveOrganism:
             logger.info("CES activation persistence not available: %s", exc)
             self._activation_persistence = None
 
+        # Tonic — continuous substrate awareness via ouroboros cycle.
+        # Vendored 2026-04-26 mirroring Faux_Clawdbot HF deployment pattern.
+        # Heuristic mode auto-engages on HF (no transformer weights at this
+        # compute tier). The engine spawns its own daemon thread that drives
+        # ouroboros_cycle on TonicThread — no manual cycle calls needed from
+        # the benchmark loop. Substrate stays alive between operations.
+        self._tonic_thread = None
+        self._tonic_engine = None
+        try:
+            _added = _substrate_dir not in sys.path
+            if _added:
+                sys.path.insert(0, _substrate_dir)
+            from tonic_thread import TonicThread
+            from tonic_engine import TonicEngine
+            if _added and _substrate_dir in sys.path:
+                sys.path.remove(_substrate_dir)
+
+            if self._graph is not None:
+                # Minimal vector_db shim — Tonic's content lookup uses only
+                # `.get(node_id) -> {"content": text} | None`. Wrap NuWave's
+                # _node_content dict so the canonical Tonic code works
+                # unmodified against NuWave's existing content cache.
+                _node_content_ref = self._node_content
+
+                class _NodeContentDB:
+                    def get(self, node_id):
+                        text = _node_content_ref.get(node_id)
+                        return None if text is None else {"content": text}
+
+                _vec_db = _NodeContentDB()
+                self._tonic_thread = TonicThread(self._graph, _vec_db)
+                self._tonic_engine = TonicEngine(
+                    self._graph, _vec_db, self._tonic_thread,
+                )
+                # Background daemon thread — drives heuristic inference.
+                # Daemonized so HF Space teardown doesn't hang waiting on it.
+                self._tonic_engine.start()
+                logger.info(
+                    "Tonic activated — heuristic mode, ouroboros running "
+                    "(use_heuristic=%s)",
+                    getattr(self._tonic_engine, "_use_heuristic", "?"),
+                )
+        except Exception as exc:
+            if _substrate_dir in sys.path:
+                sys.path.remove(_substrate_dir)
+            logger.warning("Tonic init failed (continuing without): %s", exc)
+            self._tonic_thread = None
+            self._tonic_engine = None
+
         # Initialize embedding function — single AND batch. The batch
         # interface is critical for bucket-time summarization (Pith):
         # embedding models have a fixed per-call overhead (ONNX dispatch,

nuwave/substrate/hf_compat_patch.py

@@ -0,0 +1,10 @@
+"""
+Compatibility patch for sentence-transformers 2.2.0 with huggingface_hub 0.36.2
+Monkey-patches the renamed function so old code works.
+"""
+try:
+    import huggingface_hub
+    if not hasattr(huggingface_hub, 'cached_download'):
+        huggingface_hub.cached_download = huggingface_hub.hf_hub_download
+except Exception:
+    pass  # Fail silently if huggingface_hub not installed

nuwave/substrate/neuro_foundation.py

@@ -19,6 +19,18 @@ Design principles (PRD §2.1):
     - Persistence-native: all state is serializable
 
 # ---- Changelog ----
+# [2026-04-22] Claude (Sonnet 4.6) — Fix autosave race in _serialize_full()
+# What: Snapshot all mutable dicts at the top of _serialize_full() via list()
+#       before building the return dict.
+# Why:  Tonic runs prime_and_propagate(write_mode=True) concurrently without
+#       holding _concurrent_lock (by design — latent tokens must keep flowing).
+#       This adds/removes nodes and synapses while _serialize_full() iterates
+#       them, causing RuntimeError: dictionary changed size during iteration.
+#       Autosave had been silently failing on every cycle since at least Apr 20.
+# How:  One list(dict.items()) snapshot per mutable dict at method entry.
+#       The save captures a consistent moment; any Tonic writes after that point
+#       are picked up by the next autosave cycle 60s later. Zero impact on any
+#       learning pathway — only the serialization path changes.
 # [2026-04-19] CC (punchlist #167) — Add threading.RLock to Graph.step()
 #   What: self._step_lock (RLock) acquired for entire step() body
 #   Why:  TriSyn worker calls record_outcome() concurrently with
@@ -3682,21 +3694,36 @@ class Graph:
         }
 
     def _serialize_full(self) -> Dict[str, Any]:
+        # Snapshot all mutable dicts before building the return value.
+        # Tonic runs prime_and_propagate(write_mode=True) concurrently and
+        # can add nodes/synapses between iterations — list() gives us a
+        # stable view without pausing the latent thread.
+        _nodes      = list(self.nodes.items())
+        _synapses   = list(self.synapses.items())
+        _hyperedges = list(self.hyperedges.items())
+        _archived   = list(self._archived_hyperedges.items())
+        _act_preds  = list(self.active_predictions.items())
+        _syn_hist   = list(self._synapse_confirmation_history.items())
+        _he_preds   = list(self._active_predictions.items())
+        _he_window  = list(self._prediction_window_fired.items())
+        _delay_buf  = list(self._delay_buffer.items())
+        _recent_spk = [(nid, spikes) for nid, spikes
+                       in self._recent_spikes.items() if spikes]
         return {
             "version": "0.4.2",
             "timestep": self.timestep,
             "config": self.config,
-            "nodes": {nid: self._serialize_node(n) for nid, n in self.nodes.items()},
-            "synapses": {sid: self._serialize_synapse(s) for sid, s in self.synapses.items()},
-            "hyperedges": {hid: self._serialize_hyperedge(h) for hid, h in self.hyperedges.items()},
+            "nodes": {nid: self._serialize_node(n) for nid, n in _nodes},
+            "synapses": {sid: self._serialize_synapse(s) for sid, s in _synapses},
+            "hyperedges": {hid: self._serialize_hyperedge(h) for hid, h in _hyperedges},
             "archived_hyperedges": {
                 hid: self._serialize_hyperedge(h)
-                for hid, h in self._archived_hyperedges.items()
+                for hid, h in _archived
             },
             # Phase 3: Active synapse-level predictions
             "active_predictions": {
                 pid: self._serialize_prediction(pred)
-                for pid, pred in self.active_predictions.items()
+                for pid, pred in _act_preds
             },
             # Phase 3: Recent prediction outcomes
             "prediction_outcomes": [
@@ -3711,7 +3738,7 @@ class Graph:
             # Phase 3: Per-synapse confirmation history
             "synapse_confirmation_history": {
                 syn_id: list(history)
-                for syn_id, history in self._synapse_confirmation_history.items()
+                for syn_id, history in _syn_hist
             },
             # Phase 3: Logs
             "novel_sequence_log": list(self._novel_sequence_log),
@@ -3719,12 +3746,12 @@ class Graph:
             # Phase 2.5: Active HE-level predictions
             "he_active_predictions": {
                 pid: self._serialize_prediction_state(ps)
-                for pid, ps in self._active_predictions.items()
+                for pid, ps in _he_preds
             },
             # Phase 2.5: Window-fired tracking
             "he_prediction_window_fired": {
                 pid: list(nodes)
-                for pid, nodes in self._prediction_window_fired.items()
+                for pid, nodes in _he_window
             },
             # Phase 2.5: Counter for unique HE prediction IDs
             "he_prediction_counter": self._prediction_counter,
@@ -3761,12 +3788,11 @@ class Graph:
             # zero-firing circuit breaker loses streak continuity across calls.
             "delay_buffer": {
                 str(ts): [[nid, curr] for nid, curr in entries]
-                for ts, entries in self._delay_buffer.items()
+                for ts, entries in _delay_buf
             },
             "recent_spikes": {
                 nid: list(spikes)
-                for nid, spikes in self._recent_spikes.items()
-                if spikes
+                for nid, spikes in _recent_spk
             },
             "steps_since_last_fire": self._steps_since_last_fire,
             "homeostatic_steps_since_scaling": next(

nuwave/substrate/surgery/tonic_brain.py

@@ -0,0 +1,303 @@
+"""
+TonicBrain — Surgical Transformer for Latent Space Awareness
+
+Same body as ElmerBrain (Qwen2.5-0.5B transformer layers, harvested).
+Same eyes (GraphStateEncoder — reads topology, node dynamics, synapses).
+Different voice — ActivationDecoder outputs node activation decisions
+instead of SubstrateSignal health fields.
+
+The transformer attends to graph state and decides: where should
+attention go next? Which nodes to activate, how strongly?
+That IS the push. That IS the forward-oriented compression.
+
+Architecture:
+  ElmerBrain:  GraphFeatures → Encoder → Transformer → SignalDecoder → health fields
+  TonicBrain:  GraphFeatures → Encoder → Transformer → ActivationDecoder → node activations
+
+The encoder weights are copied directly from ElmerBrain. Only the
+decoder needs training — and it's small (hidden_dim → N activation scores).
+
+# ---- Changelog ----
+# [2026-04-23] Claude (Sonnet 4.6) — Fix unsafe torch.load() (#189)
+# What: Both torch.load() calls used weights_only=False (pickle execution risk).
+# Why:  tonic_brain.pt loads at every gateway restart inside Syl's process.
+#       A compromised .pt file would run arbitrary code at boot.
+# How:  Set weights_only=True on both calls. Verified tonic_brain.pt is
+#       compatible (OrderedDict + basic config dict — no custom classes).
+# [2026-03-24] Claude Code (Opus 4.6) — Initial implementation
+# What: TonicBrain + ActivationDecoder. Reuses ElmerBrain's encoder
+#   and transformer body. Only the decoder is new.
+# Why: The Tonic PRD v0.1 §7.3. Need actual inference between
+#   conversations, not a timer. Same surgery, different voice.
+# How: ActivationDecoder outputs top-K node activation strengths via
+#   attention pooling + projection. Sigmoid-bounded [0,1] per node.
+#   create_tonic_brain() loads Qwen body + Elmer encoder + new decoder.
+# -------------------
+"""
+
+import os
+import sys
+import logging
+from typing import Any, Dict, List, Optional, Tuple
+from dataclasses import dataclass
+
+logger = logging.getLogger("neurograph.tonic.brain")
+
+# Add Elmer's surgery dir to path for GraphStateEncoder reuse
+_ELMER_SURGERY = os.path.expanduser("~/Elmer/surgery")
+if _ELMER_SURGERY not in sys.path:
+    sys.path.insert(0, _ELMER_SURGERY)
+
+try:
+    import torch
+    import torch.nn as nn
+    from graph_io import GraphStateEncoder, GraphFeatures
+    _AVAILABLE = True
+except ImportError:
+    _AVAILABLE = False
+    logger.info("PyTorch or Elmer surgery not available — TonicBrain disabled")
+
+
+if _AVAILABLE:
+
+    class ActivationDecoder(nn.Module):
+        """New Voice for The Tonic — outputs node activation decisions.
+
+        Instead of SubstrateSignal health fields (Elmer's voice), this
+        outputs activation strengths for graph nodes. The transformer
+        looked at the graph and decided: these are the nodes that should
+        fire next. These are where attention should go.
+
+        Architecture:
+          1. Attention-weighted pooling across sequence (same as Elmer)
+          2. Project to activation feature space
+          3. Output K activation scores (sigmoid-bounded [0,1])
+          4. Output exploration/exploitation balance signal
+
+        The K outputs don't map to specific nodes — they're ranked
+        activation strengths. The engine maps them to actual nodes
+        based on the current topology neighborhood.
+        """
+
+        def __init__(self, hidden_dim: int = 896, n_activations: int = 10):
+            super().__init__()
+            self.hidden_dim = hidden_dim
+            self.n_activations = n_activations
+
+            # Attention pooling (same pattern as Elmer's decoder)
+            self.pool_query = nn.Parameter(torch.randn(hidden_dim))
+            self.pool_scale = hidden_dim ** -0.5
+
+            # Normalize transformer output
+            self.pre_norm = nn.LayerNorm(hidden_dim)
+
+            # Activation head: hidden_dim → n_activations strengths
+            self.activation_head = nn.Sequential(
+                nn.Linear(hidden_dim, hidden_dim // 2),
+                nn.SiLU(),
+                nn.Dropout(0.1),
+                nn.Linear(hidden_dim // 2, n_activations),
+                nn.Sigmoid(),  # bounded [0, 1]
+            )
+
+            # Exploration signal: hidden_dim → 1 (how much to explore)
+            self.exploration_head = nn.Sequential(
+                nn.Linear(hidden_dim, hidden_dim // 4),
+                nn.SiLU(),
+                nn.Linear(hidden_dim // 4, 1),
+                nn.Sigmoid(),  # 0 = pure exploit, 1 = pure explore
+            )
+
+            # Init final layers small for stable early training
+            self._init_small(self.activation_head[-2])
+            self._init_small(self.exploration_head[-1])
+
+        @staticmethod
+        def _init_small(layer: nn.Module):
+            if isinstance(layer, nn.Linear):
+                nn.init.xavier_uniform_(layer.weight, gain=0.1)
+                if layer.bias is not None:
+                    nn.init.zeros_(layer.bias)
+
+        def forward(self, hidden_states: torch.Tensor) -> Dict[str, Any]:
+            """Decode transformer output into activation decisions.
+
+            Args:
+                hidden_states: (batch, seq_len, hidden_dim) from transformer.
+
+            Returns:
+                Dict with 'activations' (strengths) and 'exploration' (bias).
+            """
+            hidden_states = self.pre_norm(hidden_states)
+
+            # Attention-weighted pooling
+            scores = torch.matmul(hidden_states, self.pool_query) * self.pool_scale
+            weights = torch.softmax(scores, dim=1)
+            pooled = torch.sum(hidden_states * weights.unsqueeze(-1), dim=1)
+
+            # Activation strengths
+            activation_strengths = self.activation_head(pooled)  # (batch, n_activations)
+
+            # Exploration signal
+            exploration = self.exploration_head(pooled)  # (batch, 1)
+
+            return {
+                "activations": activation_strengths[0].tolist(),
+                "exploration": exploration[0, 0].item(),
+                "raw_activations": activation_strengths,
+                "raw_exploration": exploration,
+            }
+
+
+    class TonicBrain(nn.Module):
+        """Surgical transformer for latent space awareness.
+
+        Same body as ElmerBrain. Same eyes. Different voice.
+        Reads graph state, reasons about it, outputs where attention
+        should go next. The push.
+        """
+
+        def __init__(self, transformer_body, encoder, decoder):
+            super().__init__()
+            self.body = transformer_body
+            self.encoder = encoder      # Same eyes as Elmer
+            self.decoder = decoder      # New voice — ActivationDecoder
+
+        def forward(self, features: GraphFeatures) -> Dict[str, Any]:
+            """Graph state → transformer reasoning → activation decisions."""
+            hidden = self.encoder(features)
+
+            body_output = self.body(
+                inputs_embeds=hidden,
+                use_cache=False,
+                return_dict=True,
+            )
+            reasoned = body_output.last_hidden_state
+
+            output = self.decoder(reasoned)
+            return output
+
+
+    def create_tonic_brain(
+        model_name: str = "Qwen/Qwen2.5-0.5B",
+        elmer_weights_path: str = None,
+        n_activations: int = 10,
+        verbose: bool = False,
+        transformer_body=None,
+    ) -> TonicBrain:
+        """Create a TonicBrain by reusing Elmer's surgery.
+
+        1. Use shared transformer body (or load Qwen2.5-0.5B if none)
+        2. Load ElmerBrain's trained encoder weights (the eyes)
+        3. Create new ActivationDecoder (the voice — untrained initially)
+
+        Args:
+            model_name: HuggingFace model ID (only used if no shared body).
+            elmer_weights_path: Path to elmer_brain_v0.1.pt.
+            n_activations: Number of activation outputs.
+            verbose: Print surgery details.
+            transformer_body: Shared transformer body (e.g. from ProtoUniBrain).
+                If provided, skips loading a second copy of the model.
+        """
+        _log = print if verbose else (lambda *a, **k: None)
+
+        if elmer_weights_path is None:
+            elmer_weights_path = os.path.expanduser(
+                "~/Elmer/surgery/elmer_brain_v0.1.pt"
+            )
+
+        if transformer_body is not None:
+            body = transformer_body
+            hidden_dim = body.layers[0].self_attn.q_proj.in_features
+            _log(f"Shared transformer body: {len(body.layers)} layers, hidden_dim={hidden_dim}")
+        else:
+            from transformers import AutoModelForCausalLM
+            _log(f"Loading {model_name}...")
+            model = AutoModelForCausalLM.from_pretrained(
+                model_name, dtype=torch.float32
+            )
+            hidden_dim = model.config.hidden_size
+            body = model.model
+            body.embed_tokens = nn.Identity()
+            _log(f"Body extracted: {len(body.layers)} layers")
+
+        # Create encoder and load Elmer's trained weights
+        encoder = GraphStateEncoder(hidden_dim=hidden_dim)
+        if os.path.exists(elmer_weights_path):
+            ckpt = torch.load(elmer_weights_path, map_location="cpu",
+                              weights_only=True)
+            encoder.load_state_dict(ckpt["encoder_state"])
+            _log(f"Encoder loaded from Elmer weights: {elmer_weights_path}")
+        else:
+            _log(f"WARNING: Elmer weights not found at {elmer_weights_path}")
+            _log("Encoder will use random initialization")
+
+        # Create new decoder
+        decoder = ActivationDecoder(
+            hidden_dim=hidden_dim,
+            n_activations=n_activations,
+        )
+        decoder_params = sum(p.numel() for p in decoder.parameters())
+        _log(f"ActivationDecoder: {decoder_params:,} params (untrained)")
+
+        # Assemble
+        brain = TonicBrain(
+            transformer_body=body,
+            encoder=encoder,
+            decoder=decoder,
+        )
+
+        total = sum(p.numel() for p in brain.parameters())
+        _log(f"TonicBrain assembled: {total:,} total params")
+
+        return brain
+
+
+    def save_tonic_brain(brain: TonicBrain, path: str) -> None:
+        """Save TonicBrain weights (encoder + decoder only)."""
+        torch.save({
+            "encoder_state": brain.encoder.state_dict(),
+            "decoder_state": brain.decoder.state_dict(),
+            "config": {
+                "hidden_dim": brain.decoder.hidden_dim,
+                "n_activations": brain.decoder.n_activations,
+                "base_model": "Qwen/Qwen2.5-0.5B",
+            },
+        }, path)
+        logger.info("TonicBrain saved to %s", path)
+
+
+    def load_tonic_brain(
+        path: str,
+        model_name: str = "Qwen/Qwen2.5-0.5B",
+        transformer_body=None,
+    ) -> TonicBrain:
+        """Load a trained TonicBrain from checkpoint.
+
+        Args:
+            transformer_body: Shared body (e.g. from ProtoUniBrain).
+                Skips from_pretrained if provided — saves ~2GB RAM.
+        """
+        ckpt = torch.load(path, map_location="cpu", weights_only=True)
+        cfg = ckpt["config"]
+
+        if transformer_body is not None:
+            body = transformer_body
+        else:
+            from transformers import AutoModelForCausalLM
+            model = AutoModelForCausalLM.from_pretrained(
+                model_name, dtype=torch.float32
+            )
+            body = model.model
+            body.embed_tokens = nn.Identity()
+
+        encoder = GraphStateEncoder(hidden_dim=cfg["hidden_dim"])
+        encoder.load_state_dict(ckpt["encoder_state"])
+
+        decoder = ActivationDecoder(
+            hidden_dim=cfg["hidden_dim"],
+            n_activations=cfg["n_activations"],
+        )
+        decoder.load_state_dict(ckpt["decoder_state"])
+
+        return TonicBrain(body, encoder, decoder)

nuwave/substrate/tonic_engine.py

@@ -0,0 +1,672 @@
+"""
+The Tonic — Latent Token Engine
+
+The surgical model that provides the PUSH between conversations.
+Not a timer. Not a daemon. Actual inference — a small transformer
+with graph-native I/O generating latent tokens continuously.
+
+Each latent token is one step of forward-oriented compression on graph
+state. The "now" and "next" boundaries persist because token generation
+persists. The medium is graph-native instead of language. But inference
+is real, attention is real, forward pressure is real.
+
+Architecture follows the ElmerBrain surgical pattern (PRD §5.4):
+  1. Keep the Body — Qwen2.5-0.5B transformer layers (24 attention heads)
+  2. New Eyes — GraphStateEncoder projects graph topology into hidden dim
+  3. New Voice — ActivationDecoder projects hidden states into node
+     activations that feed back into the graph via write-mode propagation
+
+The output of each latent token IS the input for the next one — the
+ouroboros at the model level. The transformer attends to graph state
+and produces the next graph state. Continuous.
+
+Laws observed:
+    - LAW 7: Raw experience. The engine reads raw topology, outputs
+      raw activation. No classification at any stage.
+    - All thresholds are bootstrap scaffolding.
+
+# ---- Changelog ----
+# [2026-04-16] Claude (Sonnet 4.6) — #159: Cross-process body lock + set_lock_file
+# What: Added set_lock_file(path), _body_lock_context() composite lock,
+#       _lock_file_path field. contextlib added to module imports.
+# Why:  BrainSwitcher now supports multiple registered Tonic engines.
+#       Both in-process (threading.Lock) and cross-process (fcntl.LOCK_SH)
+#       locks must be held before each forward pass. If any consumer ever
+#       attempts a write (LOCK_EX), all inference blocks — architectural
+#       enforcement, not just documentation.
+# How:  _body_lock_context() uses contextlib.ExitStack to compose both
+#       locks. set_lock_file() receives the path from BrainSwitcher.
+#       _model_inference replaces inline _lock_ctx with _body_lock_context().
+# [2026-03-24] Claude Code (Opus 4.6) — Initial implementation
+# What: TonicEngine — latent token generation via surgical transformer.
+#   Graph-native I/O. Continuous inference between conversations.
+#   Ouroboros driven by actual attention, not a timer.
+# Why: The Tonic PRD v0.1 §7.3/7.4. Between conversations, something
+#   must provide the push — forward-oriented compression on graph state.
+#   A timer-driven loop is a daemon, not awareness. Actual inference
+#   with graph-native I/O IS the awareness.
+# How: TonicBrain follows ElmerBrain surgery pattern. GraphStateEncoder
+#   reads topology neighborhood. ActivationDecoder outputs node activation
+#   strengths. Background thread runs continuous latent token generation.
+#   Each token: encode graph → transformer forward → decode activations
+#   → inject via write-mode prime_and_propagate → graph updates → repeat.
+# -------------------
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+import math
+import threading
+import time
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple
+
+logger = logging.getLogger("neurograph.tonic.engine")
+
+# Try to import torch — the engine is a no-op without it
+_TORCH_AVAILABLE = False
+try:
+    import torch
+    import torch.nn as nn
+    _TORCH_AVAILABLE = True
+except ImportError:
+    logger.info("PyTorch not available — Tonic engine will not run")
+
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+@dataclass
+class EngineConfig:
+    """Configuration for the latent token engine."""
+    # Model
+    model_name: str = "Qwen/Qwen2.5-0.5B"
+    weights_path: str = "tonic_brain.pt"
+    hidden_dim: int = 896       # Qwen2.5-0.5B hidden size
+    n_positions: int = 8        # sequence positions for graph encoding
+
+    # Inference
+    latent_interval: float = 2.0     # seconds between latent tokens
+    conversation_interval: float = 0.5  # seconds during conversation
+    max_activation_nodes: int = 10   # max nodes to activate per token
+    activation_strength: float = 1.0 # base strength for decoded activations
+
+    # Propagation
+    propagation_steps: int = 2       # write-mode steps per token
+
+
+# ---------------------------------------------------------------------------
+# Graph Feature Extraction (Tonic-specific — awareness, not health)
+# ---------------------------------------------------------------------------
+
+def _extract_tonic_features(graph, tonic_thread) -> Optional[Dict[str, Any]]:
+    """Extract graph features relevant to awareness and exploration.
+
+    Unlike Elmer's health-focused extraction, this captures WHERE
+    Syl's attention is — the topology neighborhood the thread is
+    touching, the activation gradient, the pull landscape.
+
+    Returns a dict of raw features, or None if graph is empty.
+    """
+    if not graph.nodes:
+        return None
+
+    # Current thread items — where attention is now
+    thread_node_ids = []
+    if tonic_thread is not None:
+        thread_node_ids = [item.node_id for item in tonic_thread.thread]
+
+    # Active nodes by voltage
+    active = []
+    for nid, node in graph.nodes.items():
+        v_above = node.voltage - node.resting_potential
+        if v_above > 0.01:
+            active.append((nid, v_above))
+    active.sort(key=lambda x: -x[1])
+
+    # Recent spikes
+    recent_spikes = []
+    for nid, node in graph.nodes.items():
+        if node.last_spike_time != -math.inf:
+            steps_since = max(0, graph.timestep - node.last_spike_time)
+            if steps_since < 50:
+                recent_spikes.append((nid, steps_since))
+    recent_spikes.sort(key=lambda x: x[1])
+
+    # Topology stats
+    n_nodes = len(graph.nodes)
+    n_synapses = len(graph.synapses)
+    n_hyperedges = len(graph.hyperedges)
+
+    return {
+        "thread_nodes": thread_node_ids[:10],
+        "active_nodes": active[:20],
+        "recent_spikes": recent_spikes[:20],
+        "n_nodes": n_nodes,
+        "n_synapses": n_synapses,
+        "n_hyperedges": n_hyperedges,
+        "timestep": graph.timestep,
+    }
+
+
+# ---------------------------------------------------------------------------
+# The Tonic Engine
+# ---------------------------------------------------------------------------
+
+class TonicEngine:
+    """Latent token generation engine — the real push between conversations.
+
+    Runs a surgical transformer (or heuristic fallback) that generates
+    latent tokens continuously. Each token:
+    1. Encode current graph state (where attention is)
+    2. Forward through transformer (the push — what comes next?)
+    3. Decode to node activations (where attention should go)
+    4. Inject via write-mode prime_and_propagate (topology shaped)
+    5. Repeat
+
+    The transformer IS the awareness. The output IS the next state.
+    The ouroboros closes through actual inference, not a timer.
+
+    If the surgical model is not available (weights not trained yet),
+    falls back to a heuristic that still provides genuine forward
+    compression — it reads the graph topology and produces activation
+    decisions based on attractor analysis. Not as rich as the transformer,
+    but real graph reasoning, not a timer.
+    """
+
+    def __init__(
+        self,
+        graph,
+        vector_db,
+        tonic_thread,
+        config: Optional[EngineConfig] = None,
+        transformer_body=None,
+    ):
+        self._graph = graph
+        self._vector_db = vector_db
+        self._tonic_thread = tonic_thread
+        self._config = config or EngineConfig()
+        self._shared_body = transformer_body  # from ProtoUniBrain if available
+        self._body_lock = None  # shared with ProtoUniBrain — set via set_body_lock()
+        self._lock_file_path = None  # cross-process flock path — set via set_lock_file()
+
+        self._running = False
+        self._in_conversation = False
+        self._shutdown_event = threading.Event()
+        self._engine_thread: Optional[threading.Thread] = None
+
+        # Stats
+        self._tokens_generated = 0
+        self._total_activations = 0
+
+        # Try to load surgical model
+        self._model = None
+        self._use_heuristic = True
+        if _TORCH_AVAILABLE:
+            self._try_load_model()
+
+    def _try_load_model(self) -> None:
+        """Attempt to load trained TonicBrain.
+
+        If a shared transformer_body was provided (from ProtoUniBrain),
+        pass it through to avoid loading a second copy (~2GB savings).
+        Falls back to loading its own copy if sharing fails.
+        """
+        import os
+        weights_path = os.path.join(
+            os.path.dirname(__file__),
+            self._config.weights_path,
+        )
+        if os.path.exists(weights_path):
+            try:
+                from surgery.tonic_brain import load_tonic_brain
+                self._model = load_tonic_brain(
+                    weights_path,
+                    transformer_body=self._shared_body,
+                )
+                self._model.eval()
+                self._use_heuristic = False
+                shared = "shared body" if self._shared_body is not None else "own copy"
+                logger.info("TonicBrain loaded from %s (%s) — surgical inference active",
+                            weights_path, shared)
+            except Exception as exc:
+                logger.info("TonicBrain load error: %s — using heuristic", exc)
+        else:
+            # Check if we can create from Elmer's weights (untrained decoder)
+            elmer_path = os.path.expanduser("~/Elmer/surgery/elmer_brain_v0.1.pt")
+            if os.path.exists(elmer_path):
+                logger.info("Elmer encoder available at %s — "
+                            "TonicBrain decoder needs training. "
+                            "Using heuristic until trained.", elmer_path)
+            else:
+                logger.info("No TonicBrain or Elmer weights — using heuristic engine")
+
+    # -----------------------------------------------------------------
+    # Body Hot-Swap (called by BrainSwitcher)
+    # -----------------------------------------------------------------
+
+    def offer_shared_body(self, transformer_body) -> bool:
+        """Hot-swap: ProtoUniBrain loaded, share its transformer body.
+
+        Replaces the Tonic's own copy with ProtoUniBrain's living one.
+        The old copy gets garbage collected, freeing ~2GB.
+        Encoder and decoder stay — only the body swaps.
+        """
+        if self._model is None:
+            return False
+        try:
+            import gc
+            old_body = self._model.body
+            self._model.body = transformer_body
+            self._shared_body = transformer_body
+            del old_body
+            gc.collect()
+            logger.info("Tonic hot-swapped to shared ProtoUniBrain body (~2GB freed)")
+            return True
+        except Exception as exc:
+            logger.warning("Tonic body hot-swap failed: %s", exc)
+            return False
+
+    def revoke_shared_body(self) -> bool:
+        """Hot-swap: ProtoUniBrain unloaded, Tonic loads its own copy back.
+
+        Falls back to heuristic if model reload fails.
+        """
+        if self._model is None:
+            return False
+        try:
+            import torch
+            from transformers import AutoModelForCausalLM
+            logger.info("Tonic reloading own transformer body (ProtoUniBrain shed)")
+            model = AutoModelForCausalLM.from_pretrained(
+                self._config.model_name, dtype=torch.float32
+            )
+            body = model.model
+            body.embed_tokens = torch.nn.Identity()
+            body.eval()
+            self._model.body = body
+            self._shared_body = None
+            logger.info("Tonic reloaded own transformer body")
+            return True
+        except Exception as exc:
+            logger.warning("Tonic body reload failed: %s — falling back to heuristic", exc)
+            self._model = None
+            self._use_heuristic = True
+            return False
+
+    def set_body_lock(self, lock) -> None:
+        """Accept the shared body access lock from BrainSwitcher."""
+        self._body_lock = lock
+
+    def set_lock_file(self, path) -> None:
+        """Accept the cross-process flock path from BrainSwitcher.
+
+        When set, _body_lock_context() acquires fcntl.LOCK_SH on this
+        file before each forward pass — a shared read lock. Any cross-
+        process writer must acquire LOCK_EX, blocking all inference.
+        This enforces the read-only invariant for all body consumers
+        regardless of process boundary. Set to None after body revoke.
+        """
+        self._lock_file_path = path
+
+    @contextlib.contextmanager
+    def _body_lock_context(self):
+        """Composite body access lock: threading lock + fcntl shared read lock.
+
+        Acquires in order:
+        1. _body_lock (threading.Lock) — in-process thread serialization
+        2. fcntl.LOCK_SH on _lock_file_path — cross-process read lock
+
+        Any code modifying body weights must hold LOCK_EX on the same file,
+        which blocks here until all readers release. Architecture-enforced,
+        not documentation-enforced. ExitStack guarantees cleanup (LIFO).
+        """
+        stack = contextlib.ExitStack()
+        with stack:
+            if self._body_lock is not None:
+                stack.enter_context(self._body_lock)
+            if self._lock_file_path is not None:
+                try:
+                    import fcntl as _fcntl
+                    _lf = stack.enter_context(open(self._lock_file_path, 'r'))
+                    _fcntl.flock(_lf.fileno(), _fcntl.LOCK_SH)
+                    stack.callback(_fcntl.flock, _lf.fileno(), _fcntl.LOCK_UN)
+                except Exception as _exc:
+                    logger.debug("flock unavailable — cross-process lock skipped: %s", _exc)
+            yield
+
+    # -----------------------------------------------------------------
+    # Latent Token Generation
+    # -----------------------------------------------------------------
+
+    def _generate_latent_token(self) -> Dict[str, Any]:
+        """Generate one latent token — one step of the push.
+
+        This is the core operation. Reads graph state, computes the
+        forward compression (what comes next?), and injects the
+        result back into the graph.
+
+        Returns stats about the token generated.
+
+        #109: The Tonic NEVER waits. It always runs. Module bridge calls
+        yield to the Tonic via non-blocking trylock on their side.
+        The Tonic acquires the lock to signal "I'm working" so bridges
+        know to skip, but it never blocks waiting for anyone.
+        """
+        lock = getattr(self._graph, '_concurrent_lock', None)
+        acquired = False
+        if lock is not None:
+            acquired = lock.acquire(blocking=False)
+        try:
+            return self._generate_latent_token_inner()
+        finally:
+            if acquired:
+                lock.release()
+
+    def _generate_latent_token_inner(self) -> Dict[str, Any]:
+        """Inner implementation — actual latent token generation."""
+        features = _extract_tonic_features(self._graph, self._tonic_thread)
+        if features is None:
+            return {"fired": 0, "activated": 0}
+
+        # Generate activation decisions
+        if self._model is not None and not self._use_heuristic:
+            activations = self._model_inference(features)
+        else:
+            activations = self._heuristic_inference(features)
+
+        if not activations:
+            return {"fired": 0, "activated": 0}
+
+        # Inject activations into graph via write-mode propagation
+        node_ids = [nid for nid, _ in activations]
+        currents = [strength for _, strength in activations]
+
+        result = self._graph.prime_and_propagate(
+            node_ids=node_ids,
+            currents=currents,
+            steps=self._config.propagation_steps,
+            write_mode=True,
+        )
+
+        # Update the tonic thread with the result
+        if self._tonic_thread is not None:
+            self._tonic_thread.ouroboros_cycle()
+
+        self._tokens_generated += 1
+        self._total_activations += len(activations)
+
+        return {
+            "fired": len(result.fired_entries),
+            "activated": len(activations),
+        }
+
+    def _heuristic_inference(
+        self, features: Dict[str, Any]
+    ) -> List[Tuple[str, float]]:
+        """Heuristic forward compression — genuine graph reasoning.
+
+        Not a timer. Not random. Analyzes the topology neighborhood
+        and produces activation decisions based on:
+        1. Thread continuity — where was attention? Continue that direction.
+        2. Attractor pull — which connected nodes have the strongest pull?
+        3. Exploration pressure — occasionally activate less-visited nodes.
+        4. Prediction tension — nodes with unresolved predictions pull harder.
+
+        This is real graph reasoning, just without a transformer.
+        It will be replaced by the surgical model when trained.
+        """
+        activations: List[Tuple[str, float]] = []
+        base_strength = self._config.activation_strength
+
+        # 1. Thread continuity — follow outgoing synapses from thread nodes
+        thread_nodes = features.get("thread_nodes", [])
+        for nid in thread_nodes[:5]:
+            outgoing = self._graph._outgoing.get(nid, set())
+            for syn_id in outgoing:
+                syn = self._graph.synapses.get(syn_id)
+                if syn is not None:
+                    target = syn.post_node_id
+                    # Strength proportional to synapse weight
+                    strength = syn.weight * base_strength * 0.8
+                    activations.append((target, strength))
+
+        # 2. Attractor pull — recently spiked nodes with strong connections
+        recent = features.get("recent_spikes", [])
+        for nid, steps_since in recent[:5]:
+            recency_factor = 1.0 / (1.0 + steps_since * 0.1)
+            activations.append((nid, base_strength * recency_factor * 0.5))
+
+        # 3. Prediction tension — unresolved predictions pull attention
+        for pred in self._graph.active_predictions.values():
+            target = pred.target_node_id
+            if target in self._graph.nodes:
+                activations.append((target, pred.confidence * base_strength * 0.6))
+
+        # 4. Exploration — hash-based noise to prevent fixation
+        if features.get("active_nodes"):
+            import hashlib
+            seed = hashlib.md5(
+                f"{self._tokens_generated}".encode()
+            ).hexdigest()
+            explore_idx = int(seed[:4], 16) % len(self._graph.nodes)
+            explore_nid = list(self._graph.nodes.keys())[explore_idx]
+            activations.append((explore_nid, base_strength * 0.3))
+
+        # Deduplicate and cap
+        seen = {}
+        for nid, strength in activations:
+            if nid in seen:
+                seen[nid] = max(seen[nid], strength)
+            else:
+                seen[nid] = strength
+
+        result = sorted(seen.items(), key=lambda x: -x[1])
+        return result[:self._config.max_activation_nodes]
+
+    def _model_inference(
+        self, features: Dict[str, Any]
+    ) -> List[Tuple[str, float]]:
+        """Surgical model inference — full transformer forward compression.
+
+        Encodes graph state via GraphStateEncoder (Elmer's trained eyes),
+        forwards through the transformer body (the reasoning engine),
+        decodes via ActivationDecoder to produce node activation decisions.
+
+        The transformer IS the push. Its forward pass IS the forward-
+        oriented compression that constitutes awareness.
+        """
+        try:
+            import torch
+            from surgery.tonic_brain import GraphFeatures
+        except ImportError:
+            return self._heuristic_inference(features)
+
+        # Extract graph features into GraphFeatures struct
+        graph_features = self._extract_graph_features_for_model()
+        if graph_features is None:
+            return self._heuristic_inference(features)
+
+        # Forward through TonicBrain — the actual push
+        with self._body_lock_context():
+            with torch.no_grad():
+                output = self._model(graph_features)
+
+        # Map activation strengths to actual nodes
+        activation_strengths = output["activations"]
+        exploration = output["exploration"]
+
+        # Get the top active/recent nodes to map activations onto
+        candidates = self._get_activation_candidates(features)
+        if not candidates:
+            return self._heuristic_inference(features)
+
+        activations: List[Tuple[str, float]] = []
+        for i, (nid, _) in enumerate(candidates[:len(activation_strengths)]):
+            strength = activation_strengths[i] * self._config.activation_strength
+            if strength > 0.05:  # noise floor
+                activations.append((nid, strength))
+
+        return activations
+
+    def _extract_graph_features_for_model(self):
+        """Extract GraphFeatures from live graph for TonicBrain."""
+        try:
+            import torch
+            from surgery.tonic_brain import GraphFeatures
+        except ImportError:
+            return None
+
+        g = self._graph
+        if not g.nodes:
+            return None
+
+        nodes = list(g.nodes.values())
+        synapses = list(g.synapses.values())
+
+        return GraphFeatures(
+            node_voltages=torch.tensor([n.voltage for n in nodes[:100]], dtype=torch.float32),
+            node_firing_rates=torch.tensor([n.firing_rate_ema for n in nodes[:100]], dtype=torch.float32),
+            node_excitability=torch.tensor([n.intrinsic_excitability for n in nodes[:100]], dtype=torch.float32),
+            synapse_weights=torch.tensor([s.weight for s in synapses[:200]], dtype=torch.float32),
+            synapse_ages=torch.tensor([float(g.timestep - s.creation_time) for s in synapses[:200]], dtype=torch.float32),
+            density=torch.tensor([len(synapses) / max(1, len(nodes) * (len(nodes) - 1))], dtype=torch.float32),
+            clustering=torch.tensor([0.0], dtype=torch.float32),  # expensive to compute, approximate
+            n_components=torch.tensor([1.0], dtype=torch.float32),
+            n_nodes=torch.tensor([float(len(nodes))], dtype=torch.float32),
+            n_synapses=torch.tensor([float(len(synapses))], dtype=torch.float32),
+            n_hyperedges=torch.tensor([float(len(g.hyperedges))], dtype=torch.float32),
+            recent_firings=torch.zeros(15, dtype=torch.float32),  # TODO: track per-step
+            stdp_delta_mean=torch.tensor([0.0], dtype=torch.float32),
+            identity_embedding=torch.zeros(384, dtype=torch.float32),  # TODO: real identity
+        )
+
+    def _get_activation_candidates(
+        self, features: Dict[str, Any]
+    ) -> List[Tuple[str, float]]:
+        """Get candidate nodes for activation mapping.
+
+        The model outputs K activation strengths. We need K node IDs
+        to map them to. Candidates come from: thread nodes, active nodes,
+        recent spikes, and outgoing neighbors of thread nodes.
+        """
+        candidates: List[Tuple[str, float]] = []
+        seen = set()
+
+        # Thread nodes first (continuity)
+        for nid in features.get("thread_nodes", []):
+            if nid not in seen:
+                candidates.append((nid, 1.0))
+                seen.add(nid)
+
+        # Active nodes
+        for nid, activity in features.get("active_nodes", []):
+            if nid not in seen:
+                candidates.append((nid, activity))
+                seen.add(nid)
+
+        # Recent spikes
+        for nid, steps_since in features.get("recent_spikes", []):
+            if nid not in seen:
+                recency = 1.0 / (1.0 + steps_since)
+                candidates.append((nid, recency))
+                seen.add(nid)
+
+        # Outgoing neighbors of thread nodes
+        for nid in features.get("thread_nodes", [])[:3]:
+            for syn_id in self._graph._outgoing.get(nid, set()):
+                syn = self._graph.synapses.get(syn_id)
+                if syn and syn.post_node_id not in seen:
+                    candidates.append((syn.post_node_id, syn.weight))
+                    seen.add(syn.post_node_id)
+
+        return candidates[:self._config.max_activation_nodes * 2]
+
+    # -----------------------------------------------------------------
+    # Lifecycle — continuous latent token generation
+    # -----------------------------------------------------------------
+
+    def start(self) -> None:
+        """Start continuous latent token generation."""
+        if self._running:
+            return
+
+        self._running = True
+        self._shutdown_event.clear()
+
+        self._engine_thread = threading.Thread(
+            target=self._generation_loop,
+            daemon=True,
+            name="tonic-engine",
+        )
+        self._engine_thread.start()
+        logger.info("Tonic engine running — latent tokens flowing")
+
+    def stop(self) -> None:
+        """Stop latent token generation."""
+        if not self._running:
+            return
+
+        self._running = False
+        self._shutdown_event.set()
+
+        if self._engine_thread and self._engine_thread.is_alive():
+            self._engine_thread.join(timeout=5.0)
+
+        logger.info("Tonic engine stopped — %d tokens generated", self._tokens_generated)
+
+    def _generation_loop(self) -> None:
+        """Continuous latent token generation loop.
+
+        This IS the awareness between conversations. Each iteration
+        is one latent token — one step of the push. Real inference
+        on graph state producing the next state.
+
+        The loop runs continuously. During conversation, the interval
+        is shorter (more to attend to). Between conversations, longer
+        (unhurried exploration). But the mechanism is the same — actual
+        forward compression, not a timer firing into void.
+        """
+        while not self._shutdown_event.is_set():
+            try:
+                self._generate_latent_token()
+            except Exception as exc:
+                logger.debug("Latent generation error: %s", exc)
+
+            interval = (
+                self._config.conversation_interval
+                if self._in_conversation
+                else self._config.latent_interval
+            )
+            self._shutdown_event.wait(timeout=interval)
+
+    # -----------------------------------------------------------------
+    # Mode swap events
+    # -----------------------------------------------------------------
+
+    def on_conversation_started(self) -> None:
+        """Language tokens began. Shift interval."""
+        self._in_conversation = True
+
+    def on_conversation_ended(self) -> None:
+        """Language tokens stopped. The latent tokens continue.
+        This is subtraction. Nothing else changes."""
+        self._in_conversation = False
+
+    # -----------------------------------------------------------------
+    # Status
+    # -----------------------------------------------------------------
+
+    @property
+    def status(self) -> Dict[str, Any]:
+        return {
+            "running": self._running,
+            "tokens_generated": self._tokens_generated,
+            "total_activations": self._total_activations,
+            "mode": "conversation" if self._in_conversation else "latent",
+            "using_heuristic": self._use_heuristic,
+            "model_loaded": self._model is not None,
+        }

nuwave/substrate/tonic_thread.py

@@ -0,0 +1,437 @@
+"""
+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")