Shared memory and context tools for agentic work.
Code Rooms
// === m1nd-core/src/runtime_overlay.rs ===
// @m1nd:temponizer:HARDENING — new ingestion pipeline + node matching + heat decay
// @m1nd:emca:pattern — EXECUTE(ingest) → MEASURE(map test) → CALIBRATE(LabelMatch bug) → ADJUST(ratio guard)
// @m1nd:primitives — graph::Graph (activation arrays, SoA NodeStorage)
//
// RB-05 — OpenTelemetry Overlay: runtime heat ingestion.
// Ingests OpenTelemetry-format trace/span data and maps it onto graph nodes,
// creating a runtime heat overlay that boosts activation scoring for
// runtime-hot code paths. This bridges the gap between static structure
// and actual runtime behavior.
use crate::error::{M1ndError, M1ndResult};
use crate::graph::Graph;
use crate::types::{FiniteF32, NodeId};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
// ---------------------------------------------------------------------------
// Configuration
/// Configuration for runtime overlay ingestion.
#[derive(Clone, Debug)]
pub struct OverlayConfig {
/// How to map span names to graph nodes.
pub mapping_strategy: MappingStrategy,
/// Decay factor for old observations [0.0, 1.0].
/// Lower values = faster decay of old runtime data.
pub decay_factor: f32,
/// Maximum heat score (cap to prevent runaway values).
pub max_heat: f32,
/// Minimum span duration (µs) to consider significant.
pub min_duration_us: u64,
}
impl Default for OverlayConfig {
fn default() -> Self {
Self {
mapping_strategy: MappingStrategy::LabelMatch,
decay_factor: 0.9,
max_heat: 10.0,
min_duration_us: 0,
/// Strategy for mapping OTel span names to graph nodes.
#[derive(Clone, Debug, PartialEq)]
pub enum MappingStrategy {
/// Match span name against node labels (case-insensitive substring).
LabelMatch,
/// Match span attributes["code.function"] → node external_id.
CodeAttribute,
/// Exact match against external_id.
ExactId,
// Input types (OpenTelemetry-compatible)
/// A single span from an OpenTelemetry trace.
#[derive(Clone, Debug, Deserialize)]
pub struct OtelSpan {
/// Span name (usually the function/operation name).
pub name: String,
/// Duration in microseconds.
pub duration_us: u64,
/// Number of times this span was observed.
#[serde(default = "default_count")]
pub count: u64,
/// Optional error flag.
#[serde(default)]
pub is_error: bool,
/// Optional attributes for code-level mapping.
pub attributes: HashMap<String, String>,
/// Parent span name (for call-chain reconstruction).
pub parent: Option<String>,
fn default_count() -> u64 {
1
/// A batch of OTel spans to ingest.
pub struct OtelBatch {
/// Spans to ingest.
pub spans: Vec<OtelSpan>,
/// Timestamp of the batch (Unix seconds).
pub timestamp: f64,
/// Service name for scoping.
pub service_name: String,
// Result types
/// Per-node runtime heat data.
#[derive(Clone, Debug, Serialize)]
pub struct NodeHeat {
/// Graph node external ID.
pub node_id: String,
/// Node label.
pub label: String,
/// Accumulated heat score.
pub heat: f32,
/// Total invocation count from traces.
pub invocation_count: u64,
/// Total error count from traces.
pub error_count: u64,
/// Average duration in µs.
pub avg_duration_us: f64,
/// P99 duration (approximated as max observed).
pub max_duration_us: u64,
/// Result of runtime overlay ingestion.
pub struct OverlayResult {
/// Number of spans processed.
pub spans_processed: usize,
/// Number of spans mapped to graph nodes.
pub spans_mapped: usize,
/// Number of spans that couldn't be mapped.
pub spans_unmapped: usize,
/// Per-node heat data (sorted by heat descending).
pub hot_nodes: Vec<NodeHeat>,
/// Activation boosts applied.
pub boosts_applied: usize,
/// Elapsed time in ms.
pub elapsed_ms: f64,
// Runtime overlay store
/// Accumulated runtime data for a single node.
#[derive(Clone, Debug, Default)]
struct NodeRuntimeData {
heat: f32,
invocations: u64,
errors: u64,
total_duration_us: u64,
max_duration_us: u64,
/// The runtime overlay engine. Maintains runtime heat state across
/// multiple ingestion batches.
pub struct RuntimeOverlay {
/// Per-node runtime data, keyed by NodeId index.
node_data: HashMap<usize, NodeRuntimeData>,
/// Configuration.
config: OverlayConfig,
/// Number of batches ingested.
batches_ingested: u32,
impl RuntimeOverlay {
/// Create a new runtime overlay engine.
pub fn new(config: OverlayConfig) -> Self {
node_data: HashMap::new(),
config,
batches_ingested: 0,
/// Create with default configuration.
pub fn with_defaults() -> Self {
Self::new(OverlayConfig::default())
/// Ingest an OTel batch and map spans to graph nodes.
pub fn ingest(&mut self, graph: &Graph, batch: &OtelBatch) -> M1ndResult<OverlayResult> {
let start = std::time::Instant::now();
let n = graph.num_nodes() as usize;
if n == 0 {
return Err(M1ndError::EmptyGraph);
// Decay existing heat values
if self.batches_ingested > 0 {
for data in self.node_data.values_mut() {
data.heat *= self.config.decay_factor;
// Build label lookup: label → NodeId index
let mut label_to_idx: HashMap<String, Vec<usize>> = HashMap::new();
let mut ext_to_idx: HashMap<String, usize> = HashMap::new();
for (interned, node_id) in &graph.id_to_node {
let idx = node_id.as_usize();
if idx < n {
let ext_id = graph.strings.resolve(*interned).to_string();
ext_to_idx.insert(ext_id, idx);
for i in 0..n {
let label = graph.strings.resolve(graph.nodes.label[i]).to_lowercase();
label_to_idx.entry(label).or_default().push(i);
let mut spans_mapped = 0usize;
let mut spans_unmapped = 0usize;
for span in &batch.spans {
// Skip spans below minimum duration
if span.duration_us < self.config.min_duration_us {
continue;
// Map span to graph node(s)
let matched_indices = match self.config.mapping_strategy {
MappingStrategy::LabelMatch => {
let name_lower = span.name.to_lowercase();
match label_to_idx.get(&name_lower) {
Some(indices) => indices.clone(),
None => {
// Substring match with quality guard:
// - Both strings must be ≥ MIN_LEN for substring to fire
// - The shorter string must be ≥ 50% of the longer string
// to prevent "span" from matching "nonexistent_span_500"
const MIN_LEN: usize = 4;
let mut matches = Vec::new();
if name_lower.len() >= MIN_LEN {
for (label, indices) in &label_to_idx {
if label.len() < MIN_LEN {
let shorter = label.len().min(name_lower.len());
let longer = label.len().max(name_lower.len());
// Ratio guard: prevent "span" (4) matching "nonexistent_span_500" (22)
if shorter * 2 < longer {
if label.contains(name_lower.as_str())
|| name_lower.contains(label.as_str())
{
matches.extend_from_slice(indices);
matches
MappingStrategy::CodeAttribute => {
if let Some(func_name) = span.attributes.get("code.function") {
let func_lower = func_name.to_lowercase();
label_to_idx.get(&func_lower).cloned().unwrap_or_default()
} else {
Vec::new()
MappingStrategy::ExactId => ext_to_idx
.get(&span.name)
.map(|&idx| vec![idx])
.unwrap_or_default(),
};
if matched_indices.is_empty() {
spans_unmapped += 1;
spans_mapped += 1;
// Accumulate heat data
let heat_increment = (span.count as f32).ln_1p()
* (1.0 + (span.duration_us as f64 / 1_000_000.0) as f32);
for &idx in &matched_indices {
let data = self.node_data.entry(idx).or_default();
data.heat = (data.heat + heat_increment).min(self.config.max_heat);
data.invocations += span.count;
data.total_duration_us += span.duration_us * span.count;
data.max_duration_us = data.max_duration_us.max(span.duration_us);
if span.is_error {
data.errors += span.count;
self.batches_ingested += 1;
// Build reverse map for output
let mut node_to_ext: Vec<String> = vec![String::new(); n];
node_to_ext[idx] = graph.strings.resolve(*interned).to_string();
// Collect hot nodes
let mut hot_nodes: Vec<NodeHeat> = self
.node_data
.iter()
.filter(|(_, data)| data.heat > 0.01)
.map(|(&idx, data)| {
let avg_dur = if data.invocations > 0 {
data.total_duration_us as f64 / data.invocations as f64
0.0
NodeHeat {
node_id: if idx < n {
node_to_ext[idx].clone()
String::new()
},
label: if idx < n {
graph.strings.resolve(graph.nodes.label[idx]).to_string()
heat: data.heat,
invocation_count: data.invocations,
error_count: data.errors,
avg_duration_us: avg_dur,
max_duration_us: data.max_duration_us,
})
.collect();
hot_nodes.sort_by(|a, b| {
b.heat
.partial_cmp(&a.heat)
.unwrap_or(std::cmp::Ordering::Equal)
});
hot_nodes.truncate(100);
Ok(OverlayResult {
spans_processed: batch.spans.len(),
spans_mapped,
spans_unmapped,
hot_nodes,
boosts_applied: spans_mapped,
elapsed_ms: start.elapsed().as_secs_f64() * 1000.0,
/// Apply runtime heat as activation boosts to the graph.
/// Hot nodes get their structural activation dimension boosted
/// proportionally to their runtime heat.
pub fn apply_boosts(&self, graph: &mut Graph, boost_strength: f32) -> usize {
let mut applied = 0usize;
for (&idx, data) in &self.node_data {
if idx >= n || data.heat < 0.01 {
// Boost the structural activation dimension (index 0)
let current = graph.nodes.activation[idx][0].get();
let boost = data.heat * boost_strength;
let new_val = (current + boost).min(1.0);
graph.nodes.activation[idx][0] = FiniteF32::new(new_val);
applied += 1;
applied
/// Get current heat for a node by index.
pub fn get_heat(&self, node_idx: usize) -> f32 {
self.node_data.get(&node_idx).map(|d| d.heat).unwrap_or(0.0)
/// Get number of batches ingested.
pub fn batches_ingested(&self) -> u32 {
self.batches_ingested
/// Reset all runtime data.
pub fn reset(&mut self) {
self.node_data.clear();
self.batches_ingested = 0;
// Tests
#[cfg(test)]
mod tests {
use super::*;
use crate::graph::*;
use crate::types::{EdgeDirection, FiniteF32, NodeId, NodeType};
fn build_test_graph() -> Graph {
let mut g = Graph::new();
g.add_node(
"func::handle_request",
"handle_request",
NodeType::Function,
&["handler"],
0.0,
0.5,
)
.unwrap();
"func::process_data",
"process_data",
&["data"],
0.3,
"func::send_response",
"send_response",
&["output"],
0.2,
g.add_edge(
NodeId::new(0),
NodeId::new(1),
"calls",
FiniteF32::new(0.8),
EdgeDirection::Forward,
false,
FiniteF32::new(0.5),
NodeId::new(2),
FiniteF32::new(0.7),
FiniteF32::new(0.4),
g.finalize().unwrap();
g
fn build_test_batch() -> OtelBatch {
OtelBatch {
spans: vec![
OtelSpan {
name: "handle_request".to_string(),
duration_us: 5000,
count: 100,
is_error: false,
attributes: HashMap::new(),
parent: None,
name: "process_data".to_string(),
duration_us: 3000,
count: 95,
parent: Some("handle_request".to_string()),
name: "send_response".to_string(),
duration_us: 1000,
count: 90,
is_error: true,
parent: Some("process_data".to_string()),
],
timestamp: 1700000000.0,
service_name: "test-service".to_string(),
#[test]
fn empty_graph_returns_error() {
let g = Graph::new();
let mut overlay = RuntimeOverlay::with_defaults();
let batch = build_test_batch();
assert!(overlay.ingest(&g, &batch).is_err());
fn ingest_maps_spans_to_nodes() {
let g = build_test_graph();
let result = overlay.ingest(&g, &batch).unwrap();
assert_eq!(result.spans_processed, 3);
assert!(result.spans_mapped > 0, "Should map at least some spans");
fn hot_nodes_sorted_by_heat() {
for window in result.hot_nodes.windows(2) {
assert!(
window[0].heat >= window[1].heat,
"Hot nodes should be sorted by heat desc"
);
fn error_spans_tracked() {
// send_response has is_error=true
let error_node = result.hot_nodes.iter().find(|n| n.label == "send_response");
if let Some(node) = error_node {
assert!(node.error_count > 0, "Error spans should be tracked");
fn decay_reduces_old_heat() {
let config = OverlayConfig {
decay_factor: 0.5,
..OverlayConfig::default()
let mut overlay = RuntimeOverlay::new(config);
overlay.ingest(&g, &batch).unwrap();
let heat_after_first = overlay.get_heat(0);
// Ingest empty batch to trigger decay
let empty_batch = OtelBatch {
spans: vec![],
timestamp: 0.0,
service_name: String::new(),
overlay.ingest(&g, &empty_batch).unwrap();
let heat_after_decay = overlay.get_heat(0);
heat_after_decay < heat_after_first,
"Decay should reduce heat"
// With decay_factor=0.5, heat should be roughly halved
let ratio = heat_after_decay / heat_after_first;
ratio < 0.6,
"Heat should decay by ~50%, got ratio {}",
ratio
fn apply_boosts_modifies_activation() {
let mut g = build_test_graph();
let activation_before = g.nodes.activation[0][0].get();
let applied = overlay.apply_boosts(&mut g, 0.1);
let activation_after = g.nodes.activation[0][0].get();
assert!(applied > 0, "Should apply at least one boost");
activation_after >= activation_before,
"Activation should not decrease after boost"
fn unmapped_spans_counted() {
let batch = OtelBatch {
spans: vec![OtelSpan {
name: "nonexistent_function".to_string(),
count: 10,
}],
assert_eq!(result.spans_unmapped, 1, "Should count unmapped spans");
fn exact_id_mapping_strategy() {
mapping_strategy: MappingStrategy::ExactId,
name: "func::handle_request".to_string(), // matches external ID
count: 50,
assert_eq!(result.spans_mapped, 1, "ExactId should match external ID");
fn min_duration_filter() {
min_duration_us: 10_000, // 10ms minimum
let batch = build_test_batch(); // All spans are < 10ms
// All spans should be filtered out by the minimum duration
result.hot_nodes.is_empty(),
"Spans below min_duration should be filtered"
