Claude Sonnet 4.6
Port torn-read fix + grace_period fix + CES to Codemine's openclaw_hook.py
a904336 | """ | |
| 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) | |
| 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, | |
| ) | |