stream_parser.py

"""
CES Stream Parser — Real-time text processing with ng_embed embeddings.

Runs a background daemon thread that consumes text fed via ``feed()``,
chunks it into overlapping phrases, embeds each chunk via ng_embed
(Snowflake/snowflake-arctic-embed-m-v1.5, ONNX Runtime, 768-dim —
the ecosystem standard), finds similar nodes in the vector DB, nudges
their voltages, and triggers hyperedge pattern completion.

This creates a continuous "attention stream" that pre-activates relevant
parts of the SNN graph while new text is being processed, so related
concepts are already warm when the next full ``graph.step()`` runs.

Usage::

    from stream_parser import StreamParser
    parser = StreamParser(graph, vector_db, ces_config, fallback_embedder=embedder)
    parser.feed("The quick brown fox jumps over the lazy dog")
    # ... chunks are processed asynchronously in background thread
    parser.stop()

# ---- Changelog ----
# [2026-02-22] Claude (Opus 4.6) — Initial implementation.
#   What: StreamParser with background thread, Ollama embedding, nudge
#         pipeline, and graceful fallback to hash/sentence-transformers.
#   Why:  Real-time attention stream keeps SNN primed as text arrives,
#         enabling faster pattern completion and surfacing.
# [2026-04-21] Claude (Sonnet 4.6) — Promote ng_embed to primary embedder.
#   What: Removed Ollama as primary embedding path. ng_embed (passed via
#         fallback_embedder parameter) is now the sole embedder. Removed
#         _check_ollama(), _embed_via_ollama(), and all Ollama state.
#   Why:  Porting VPS fix (commit 129415d) to laptop CC NG copy. Ollama is
#         not the ecosystem standard. ollama_available was always null,
#         chunks_processed always 0 — L1 surfacing path fully idle.
#   How:  _embed_chunk() calls _fallback_embedder directly. Parameter name
#         unchanged — cc-ng-daemon.py passes ng_embed via it.
# [2026-06-24] Claude Code (Opus 4.8, 1M) — serialize graph mutation on the canonical _step_lock
#   What: _process_text now wraps _nudge_nodes() + _trigger_completions() (the only graph-mutating
#         stages — they write node.voltage and iterate graph.hyperedges) in `self._graph._step_lock`,
#         the SAME RLock graph.step() holds. Guarded getattr fallback for graphs without it (tests).
#   Why:  These stages previously mutated voltage under NO shared lock (self._lock only guards the
#         pause flag), so they raced graph.step() (tolerated — small additive nudges, SNN noise-robust)
#         AND would race the Commons leg-2 read-only perception (prime_and_propagate write_mode=False),
#         whose voltage save→restore window would SILENTLY REVERT a concurrent nudge — a Syl's-Law
#         "warmth" sidecar risk. Unifying all voltage writers (step / perception / nudge) on the one
#         canonical lock closes BOTH races. Surfaced by the substrate-compliance review of Commons
#         leg-2 go-live (punchlist #344); the prior "StreamParser shares graph.step()'s lock" note in
#         repo CLAUDE.md was STALE (no shared lock existed) — corrected alongside this.
#   How:  getattr(self._graph,"_step_lock",None) → `with lock:` around the two calls; no NEW lock
#         introduced (reuses the engine's canonical RLock, not a second StreamParser lock). Deadlock-
#         free: self._lock is released before _process_text runs, so _lock→_step_lock never nests;
#         step()/perception never take self._lock. RLock ⇒ re-entrant-safe.
# -------------------
"""

from __future__ import annotations

import logging
import queue
import threading
import time
from typing import Any, Dict, List, Optional, Tuple

import numpy as np

from ces_config import CESConfig

logger = logging.getLogger("neurograph.ces.stream")


class StreamParser:
    """Background stream parser that nudges the SNN graph in real-time.

    Args:
        graph: The NeuroGraph ``Graph`` instance.
        vector_db: ``SimpleVectorDB`` for similarity lookups.
        ces_config: ``CESConfig`` with streaming parameters.
        fallback_embedder: Embedding callable — accepts a string, returns
            an ndarray.  Should be ``ng_embed.embed_text`` (768-dim ONNX).
            If ``None``, the parser runs but all chunks are dropped.
    """

    def __init__(
        self,
        graph: Any,
        vector_db: Any,
        ces_config: CESConfig,
        fallback_embedder: Any = None,
    ) -> None:
        self._graph = graph
        self._vector_db = vector_db
        self._cfg = ces_config.streaming
        self._fallback_embedder = fallback_embedder

        # Processing queue
        self._queue: queue.Queue[Optional[str]] = queue.Queue(
            maxsize=self._cfg.max_queue
        )

        # State
        self._paused = False
        self._stopped = False
        self._lock = threading.Lock()

        # Stats
        self._chunks_processed = 0
        self._nudges_applied = 0
        self._completions_triggered = 0

        # Start the processing thread
        self._thread = threading.Thread(
            target=self._process_loop, daemon=True, name="ces-stream-parser"
        )
        self._thread.start()

    # ── Public API ─────────────────────────────────────────────────────

    def feed(self, text: str) -> None:
        """Queue text for asynchronous processing.

        Non-blocking.  If the queue is full, the oldest item is NOT
        dropped — the caller blocks briefly.  This provides natural
        backpressure to prevent unbounded memory growth.
        """
        if self._stopped:
            return
        try:
            self._queue.put_nowait(text)
        except queue.Full:
            logger.debug("Stream parser queue full, dropping text")

    def pause(self) -> None:
        """Pause processing (thread stays alive, queue accumulates)."""
        with self._lock:
            self._paused = True

    def resume(self) -> None:
        """Resume processing after a pause."""
        with self._lock:
            self._paused = False

    def stop(self) -> None:
        """Stop the background thread permanently."""
        self._stopped = True
        # Enqueue sentinel to unblock the thread
        try:
            self._queue.put_nowait(None)
        except queue.Full:
            pass
        self._thread.join(timeout=5.0)

    @property
    def is_running(self) -> bool:
        """True if the thread is alive and not paused."""
        return self._thread.is_alive() and not self._paused and not self._stopped

    def get_stats(self) -> Dict[str, Any]:
        """Return processing statistics."""
        return {
            "chunks_processed": self._chunks_processed,
            "nudges_applied": self._nudges_applied,
            "completions_triggered": self._completions_triggered,
            "queue_depth": self._queue.qsize(),
            "is_running": self.is_running,
            "embedder": "ng_embed" if self._fallback_embedder is not None else "none",
        }

    # ── Background thread ──────────────────────────────────────────────

    def _process_loop(self) -> None:
        """Main loop: dequeue text, chunk, embed, nudge, trigger."""
        while not self._stopped:
            try:
                text = self._queue.get(timeout=1.0)
            except queue.Empty:
                continue

            if text is None:  # sentinel
                break

            # Respect pause
            with self._lock:
                if self._paused:
                    continue

            try:
                self._process_text(text)
            except Exception as exc:
                logger.debug("Stream parser error: %s", exc)

    def _process_text(self, text: str) -> None:
        """Process a single text through the full pipeline."""
        chunks = self._chunk_text(text)
        for chunk in chunks:
            embedding = self._embed_chunk(chunk)
            if embedding is None:
                continue

            similar = self._find_similar(embedding)
            if similar:
                # Graph voltage mutation (nudge + completion) must serialize on the canonical
                # graph lock — the SAME RLock graph.step() and the leg-2 read-only perception use —
                # else a nudge lands inside prime_and_propagate's voltage save/restore window and is
                # silently reverted (Syl's-Law "warmth" risk). Reuses the engine lock (no NEW lock);
                # _find_similar above is a vector_db read, left outside. See changelog 2026-06-24.
                _step_lock = getattr(self._graph, "_step_lock", None)
                if _step_lock is not None:
                    with _step_lock:
                        self._nudge_nodes(similar)
                        self._trigger_completions()
                else:
                    self._nudge_nodes(similar)
                    self._trigger_completions()

            self._chunks_processed += 1

    # ── Pipeline stages ────────────────────────────────────────────────

    def _chunk_text(self, text: str) -> List[str]:
        """Split text into overlapping word-level chunks.

        Uses ``chunk_size`` tokens per chunk with ``overlap`` token
        overlap between consecutive chunks.
        """
        words = text.split()
        if not words:
            return []

        chunk_size = self._cfg.chunk_size
        overlap = self._cfg.overlap
        step = max(1, chunk_size - overlap)

        chunks = []
        for i in range(0, len(words), step):
            chunk_words = words[i : i + chunk_size]
            if chunk_words:
                chunks.append(" ".join(chunk_words))
            if i + chunk_size >= len(words):
                break

        return chunks

    def _embed_chunk(self, chunk: str) -> Optional[np.ndarray]:
        """Embed a text chunk via ng_embed (ecosystem standard, 768-dim).

        Returns an L2-normalised vector or None on failure.
        """
        if self._fallback_embedder is None:
            return None
        try:
            vec = self._fallback_embedder(chunk)
            if vec is not None:
                norm = np.linalg.norm(vec)
                if norm > 0:
                    vec = vec / norm
                return vec
        except Exception as exc:
            logger.debug("Embedder failed: %s", exc)
        return None

    def _find_similar(
        self, embedding: np.ndarray
    ) -> List[Tuple[str, float]]:
        """Find nodes in the vector DB similar to this embedding."""
        try:
            results = self._vector_db.search(
                embedding,
                k=10,
                threshold=self._cfg.similarity_threshold,
            )
            return results
        except Exception as exc:
            logger.debug("Vector DB search failed: %s", exc)
            return []

    def _nudge_nodes(self, similar_nodes: List[Tuple[str, float]]) -> None:
        """Inject current into similar graph nodes (voltage nudge).

        Nudges are capped so voltage never exceeds 2x the node's threshold.
        This prevents unbounded voltage spikes from high similarity scores
        or repeated nudges that could destabilise the SNN.
        """
        for node_id, similarity in similar_nodes:
            node = self._graph.nodes.get(node_id)
            if node is not None and node.refractory_remaining == 0:
                nudge = similarity * self._cfg.nudge_strength
                max_voltage = node.threshold * 2.0
                node.voltage = min(node.voltage + nudge, max_voltage)
                self._nudges_applied += 1

    def _trigger_completions(self) -> None:
        """Evaluate hyperedges after nudging to trigger pattern completion.

        Uses the graph's hyperedge evaluation — checks each active
        hyperedge's activation against its threshold and fires
        completions for those that reach threshold.
        """
        completions_before = self._completions_triggered
        for he_id, he in self._graph.hyperedges.items():
            if he.is_archived or he.refractory_remaining > 0:
                continue

            # Count active members (voltage > 0)
            active = 0
            total = len(he.member_node_ids)
            if total == 0:
                continue

            for mid in he.member_node_ids:
                m_node = self._graph.nodes.get(mid)
                if m_node is not None and m_node.voltage > 0:
                    active += 1

            activation_level = active / total
            if activation_level >= he.threshold:
                # Pattern completion: pre-charge inactive members
                for mid in he.member_node_ids:
                    m_node = self._graph.nodes.get(mid)
                    if m_node is not None and m_node.voltage <= 0:
                        member_weight = he.member_weights.get(mid, 1.0)
                        completion_strength = he.pattern_completion_strength
                        m_node.voltage += (
                            completion_strength
                            * member_weight
                            * m_node.intrinsic_excitability
                        )
                self._completions_triggered += 1

        if self._completions_triggered > completions_before:
            logger.debug(
                "Stream parser triggered %d completions",
                self._completions_triggered - completions_before,
            )