""" CES Surfacing Monitor — Surface relevant knowledge from the SNN for prompt injection. Monitors node activations after each ``graph.step()``, maintains a bounded priority queue of "surfaced" concepts that score above threshold, and formats them as a context block for prompt injection. The queue decays each step so stale concepts fade out and fresh firings take priority. Usage:: from surfacing import SurfacingMonitor monitor = SurfacingMonitor(graph, vector_db, ces_config) step_result = graph.step() monitor.after_step(step_result) context = monitor.format_context() # ---- Changelog ---- # [2026-07-08] Candor (TQB/QB build) — honest salience label # What: format line now "(salience: {score:.2f})" instead of "(confidence: {score:.0%})" # Why: PRD 2026-07-08-codemine-surfacing-parity §2.2/§7.2 — salience is not a probability; >100% strings eroded worker trust # How: display-only change, canonical parity (e6eb2f2); pinned by tests/test_surfacing_salience.py # [2026-04-08] Claude (Opus 4.6) — Punchlist #55: Read attention params from substrate # What: _decay_queue() reads surfacing_decay_rate and surfacing_min_confidence # from graph.config (substrate) instead of frozen CES dataclass. # Why: CES attention params are now in TUNABLE_PARAMS (ng_lite.py). Elmer # tunes them via update_tunable(). SurfacingMonitor must read the live # substrate value, not the bootstrap default. CES config is fallback only. # How: getattr(self._graph, 'config', {}).get() with self._cfg fallback. # [2026-04-07] Claude (Opus 4.6) — Fix temporal stuttering: decay before read # What: get_surfaced() now calls _decay_queue() before returning items. # Why: Lifecycle ordering bug — assemble() reads the surfacing queue before # afterTurn() applies decay, so stale concepts persisted at inflated # scores across multiple turns. A concept firing at score 1.0 would # remain above min_confidence (0.3) for ~16 turns. With decay-before-read, # each read applies one decay step, roughly halving the persistence window. # How: Single _decay_queue() call at the top of get_surfaced(). Safe because # _decay_queue() is idempotent within a step — double-decay in the same # step is bounded by the 0.95 multiplier and won't drop items incorrectly. # [2026-03-25] Claude (Opus 4.6) — Salience weights from config (SVG Phase 3) # What: _score_node() reads weight_voltage/weight_excitability/weight_he_membership # from CES SurfacingConfig instead of hardcoded 0.5/0.3/0.2. # Why: Static Value Graduation — salience weights are the substrate's concern. # Named config values are tunable by Elmer's TuningSocket. Bootstrap # defaults preserved (0.5/0.3/0.2). # How: Weights read from self._cfg at scoring time. ces_config.py updated # with weight_voltage, weight_excitability, weight_he_membership fields. # [2026-02-22] Claude (Opus 4.6) — Initial implementation. # What: SurfacingMonitor with bounded heap, composite scoring, # per-step decay, and context block formatting. # Why: Surfaces knowledge the SNN considers relevant based on # activation dynamics — not explicit search, but associative # "remembering". # ------------------- """ from __future__ import annotations import heapq import logging from typing import Any, Dict, List, Optional from ces_config import CESConfig logger = logging.getLogger("neurograph.ces.surfacing") class _SurfacedItem: """Internal queue entry with negated score for max-heap via heapq.""" __slots__ = ("node_id", "score", "content", "metadata", "age") def __init__( self, node_id: str, score: float, content: str = "", metadata: Optional[Dict[str, Any]] = None, age: int = 0, ) -> None: self.node_id = node_id self.score = score self.content = content self.metadata = metadata or {} self.age = age def __lt__(self, other: _SurfacedItem) -> bool: # heapq is a min-heap; negate score for max-heap behavior return self.score > other.score def __eq__(self, other: object) -> bool: if not isinstance(other, _SurfacedItem): return NotImplemented return self.node_id == other.node_id class SurfacingMonitor: """Monitors SNN firings and surfaces relevant knowledge. Args: graph: The NeuroGraph ``Graph`` instance. vector_db: ``SimpleVectorDB`` for content retrieval. ces_config: ``CESConfig`` with surfacing parameters. """ def __init__( self, graph: Any, vector_db: Any, ces_config: CESConfig, ) -> None: self._graph = graph self._vector_db = vector_db self._cfg = ces_config.surfacing # Bounded priority queue (max-heap via __lt__ inversion) self._queue: List[_SurfacedItem] = [] self._node_ids_in_queue: set = set() # Decay tracking — prevents double-decay when get_surfaced() and # after_step() are both called in the same logical step. self._last_decay_step: int = -1 # Stats self._total_surfaced: int = 0 # ── Public API ───────────────────────────────────────────────────── def after_step(self, step_result: Any) -> None: """Process a step result and update the surfacing queue. Should be called after each ``graph.step()``. Scans fired nodes, scores them, and inserts into the bounded queue. Also applies decay to existing entries. Note: fired nodes have already had their voltage reset to resting potential by ``graph.step()`` before this method is called. We do NOT filter by current voltage — being in ``fired_node_ids`` already means the node exceeded its firing threshold during the step. Args: step_result: ``StepResult`` from ``graph.step()``. """ # Decay existing entries self._decay_queue() # Process newly fired nodes for node_id in step_result.fired_node_ids: node = self._graph.nodes.get(node_id) if node is None: continue score = self._score_node(node_id, node) if score < self._cfg.min_confidence: continue # Fetch content from vector DB db_entry = self._vector_db.get(node_id) content = "" metadata: Dict[str, Any] = {} if db_entry is not None: content = db_entry.get("content", "") metadata = db_entry.get("metadata", {}) if not content: continue # Skip nodes without retrievable content item = _SurfacedItem( node_id=node_id, score=score, content=content, metadata=metadata, ) self._insert_item(item) self._total_surfaced += 1 def get_surfaced(self, max_items: Optional[int] = None) -> List[Dict[str, Any]]: """Return the top surfaced items sorted by score. Applies decay before reading so that scores reflect elapsed steps since the last ``after_step()`` call. This prevents stale concepts from being injected into the prompt at their pre-decay score when ``assemble()`` runs before the next ``after_step()``. Args: max_items: Maximum items to return. Defaults to ``ces_config.surfacing.max_surfaced``. Returns: List of dicts with node_id, content, metadata, score. """ # Decay before read — fixes lifecycle ordering where assemble() # reads the queue before afterTurn() applies decay, causing # stale concepts to persist at inflated scores. self._decay_queue() if max_items is None: max_items = self._cfg.max_surfaced # Sort by score descending sorted_items = sorted(self._queue, key=lambda x: x.score, reverse=True) results = [] for item in sorted_items[:max_items]: entry: Dict[str, Any] = { "node_id": item.node_id, "content": item.content, "score": round(item.score, 4), } if self._cfg.include_metadata: entry["metadata"] = item.metadata results.append(entry) return results def format_context( self, surfaced_items: Optional[List[Dict[str, Any]]] = None ) -> str: """Format surfaced items as a context block for prompt injection. Args: surfaced_items: Items to format. If ``None``, calls ``get_surfaced()`` automatically. Returns: Formatted context block string, or empty string if nothing surfaced. """ if surfaced_items is None: surfaced_items = self.get_surfaced() if not surfaced_items: return "" lines = ["[NeuroGraph Surfaced Knowledge]"] for item in surfaced_items: content = item.get("content", "") score = item.get("score", 0.0) # Truncate long content for context blocks if len(content) > 200: content = content[:197] + "..." # Salience, NOT a probability: _score_node()'s designed range is # ~[0.8, 1.8] (floored >=0.8 for any fired node). The old # "(confidence: {score:.0%})" rendering produced ">100%" strings # that read as fictitious and eroded worker trust in the block. lines.append(f"- {content} (salience: {score:.2f})") return "\n".join(lines) def clear(self) -> None: """Empty the surfacing queue.""" self._queue.clear() self._node_ids_in_queue.clear() def get_stats(self) -> Dict[str, Any]: """Return surfacing statistics.""" scores = [item.score for item in self._queue] avg_score = sum(scores) / len(scores) if scores else 0.0 return { "total_surfaced": self._total_surfaced, "queue_depth": len(self._queue), "avg_score": round(avg_score, 4), } # ── Internal ─────────────────────────────────────────────────────── def _score_node(self, node_id: str, node: Any) -> float: """Compute a composite relevance score for a fired node. Weights read from CES SurfacingConfig (bootstrap defaults: 0.5/0.3/0.2). These are tunable by Elmer's TuningSocket as the substrate learns which signals matter for surfacing. Components: - voltage_normalized = effective_voltage / node.threshold (capped at 2.0) - excitability = node.intrinsic_excitability (capped at 2.0) - he_membership = number of hyperedges containing this node / 10 (capped at 1.0) Note: ``graph.step()`` resets fired nodes' voltage to resting potential before this method is called, so we use ``max(voltage, threshold)`` as the effective voltage — fired nodes reached at least their threshold. """ effective_voltage = max(node.voltage, node.threshold) voltage_norm = min(effective_voltage / max(node.threshold, 0.01), 2.0) excitability = min(node.intrinsic_excitability, 2.0) # Count hyperedge memberships he_count = 0 for he in self._graph.hyperedges.values(): if node_id in getattr(he, 'member_nodes', getattr(he, 'member_node_ids', [])): he_count += 1 he_norm = min(he_count / 10.0, 1.0) w_v = self._cfg.weight_voltage w_e = self._cfg.weight_excitability w_h = self._cfg.weight_he_membership return w_v * voltage_norm + w_e * excitability + w_h * he_norm def _insert_item(self, item: _SurfacedItem) -> None: """Insert or update an item in the bounded queue.""" # If already in queue, update if higher score if item.node_id in self._node_ids_in_queue: for i, existing in enumerate(self._queue): if existing.node_id == item.node_id: if item.score > existing.score: self._queue[i] = item heapq.heapify(self._queue) return # If queue at capacity, replace weakest if new item is stronger capacity = self._cfg.queue_capacity if len(self._queue) >= capacity: # heapq[0] is the "smallest" — but our __lt__ inverts, so # heapq[0] is actually the highest score. We want the lowest. weakest_idx = max(range(len(self._queue)), key=lambda i: -self._queue[i].score) if item.score > self._queue[weakest_idx].score: removed = self._queue[weakest_idx] self._node_ids_in_queue.discard(removed.node_id) self._queue[weakest_idx] = item self._node_ids_in_queue.add(item.node_id) heapq.heapify(self._queue) return heapq.heappush(self._queue, item) self._node_ids_in_queue.add(item.node_id) def _decay_queue(self) -> None: """Apply decay to all items in the queue, removing those below threshold. Idempotent within a single graph timestep — tracks the last step at which decay was applied and skips if already done this step. This allows both ``get_surfaced()`` and ``after_step()`` to call decay without double-penalising queue items. """ current_step = getattr(self._graph, 'timestep', -2) if current_step == self._last_decay_step: return # Already decayed this step self._last_decay_step = current_step # Read from substrate config (tunable via Elmer) with CES config fallback graph_config = getattr(self._graph, 'config', {}) decay_rate = graph_config.get("surfacing_decay_rate", self._cfg.decay_rate) min_confidence = graph_config.get("surfacing_min_confidence", self._cfg.min_confidence) surviving: List[_SurfacedItem] = [] for item in self._queue: item.score *= decay_rate item.age += 1 if item.score >= min_confidence: surviving.append(item) else: self._node_ids_in_queue.discard(item.node_id) self._queue = surviving heapq.heapify(self._queue)