lean_ctx/core/
introspect.rs

1//! Cognition introspection — proves that each scientific subsystem is wired
2//! *and* actually active at runtime.
3//!
4//! The Cognition v2 stack (power-law decay, Hebbian eviction, global-workspace
5//! ignition, integration-aware Phi, immune detection, …) is only
6//! useful if it is genuinely reached on the hot path — not dead code. Every
7//! subsystem calls [`tick`] at its real call site; [`flush`] persists the
8//! counters to a small JSON file so a *separate* process (`lean-ctx introspect
9//! cognition`) can report, per subsystem: wired (present in this build), active
10//! (call count > 0), last-run, and total count.
11//!
12//! Determinism note (#498): this is diagnostic CLI output, never an MCP tool
13//! body, so wall-clock timestamps here do not affect provider prompt caching.
14
15use std::collections::HashMap;
16use std::sync::{LazyLock, Mutex};
17use std::time::Instant;
18
19use serde::{Deserialize, Serialize};
20
21/// One cognition subsystem in the activation registry. `key` is the stable
22/// identifier passed to [`tick`]; `label`/`science` are for human-readable
23/// reports only.
24pub struct Subsystem {
25    pub key: &'static str,
26    pub label: &'static str,
27    pub science: &'static str,
28}
29
30/// The full set of Cognition v2 subsystems that must be wired and active.
31/// Order is the natural reading order for the report (foundation → experiment).
32pub const SUBSYSTEMS: &[Subsystem] = &[
33    Subsystem {
34        key: "phi_recompute",
35        label: "Sticky-Phi fix",
36        science: "time-variant salience (attention)",
37    },
38    Subsystem {
39        key: "power_law_decay",
40        label: "Power-law decay",
41        science: "Ebbinghaus forgetting + spacing effect",
42    },
43    Subsystem {
44        key: "integration_phi",
45        label: "Integration-aware Phi",
46        science: "IIT non-redundancy (MMR)",
47    },
48    Subsystem {
49        key: "gwt_ignition",
50        label: "Global-workspace ignition",
51        science: "global workspace theory",
52    },
53    Subsystem {
54        key: "immune_detector",
55        label: "Context-poisoning screen (Control)",
56        science: "self/non-self screening of provider data",
57    },
58];
59
60/// Persisted activity record for a single subsystem.
61#[derive(Debug, Clone, Default, Serialize, Deserialize)]
62pub struct Activity {
63    pub count: u64,
64    pub last_unix: i64,
65}
66
67#[derive(Default)]
68struct Entry {
69    count: u64,
70    /// Count already merged to disk — lets [`flush`] write only the delta so
71    /// concurrent processes accumulate instead of clobbering each other.
72    flushed: u64,
73    last_unix: i64,
74}
75
76#[derive(Default)]
77struct Registry {
78    entries: HashMap<&'static str, Entry>,
79    last_flush: Option<Instant>,
80}
81
82static REGISTRY: LazyLock<Mutex<Registry>> = LazyLock::new(|| Mutex::new(Registry::default()));
83
84/// Minimum gap between debounced flushes, keeping `tick` cheap on the hot path.
85const FLUSH_DEBOUNCE: std::time::Duration = std::time::Duration::from_secs(2);
86
87fn now_unix() -> i64 {
88    std::time::SystemTime::now()
89        .duration_since(std::time::UNIX_EPOCH)
90        .map_or(0, |d| d.as_secs() as i64)
91}
92
93/// Record one activation of `key`. Cheap (one mutex + map update); call this at
94/// the *real* call site of each subsystem so the registry reflects reality.
95pub fn tick(key: &'static str) {
96    if let Ok(mut reg) = REGISTRY.lock() {
97        let e = reg.entries.entry(key).or_default();
98        e.count += 1;
99        e.last_unix = now_unix();
100    }
101}
102
103fn activity_path() -> Option<std::path::PathBuf> {
104    crate::core::data_dir::lean_ctx_data_dir()
105        .ok()
106        .map(|d| d.join("cognition_activity.json"))
107}
108
109fn read_disk() -> HashMap<String, Activity> {
110    activity_path()
111        .filter(|p| p.exists())
112        .and_then(|p| std::fs::read_to_string(p).ok())
113        .and_then(|s| serde_json::from_str(&s).ok())
114        .unwrap_or_default()
115}
116
117/// Persist in-memory deltas to the shared JSON file (delta-merge so multiple
118/// processes accumulate). Best-effort: failures are swallowed — diagnostics
119/// must never break the hot path.
120pub fn flush() {
121    let Ok(mut reg) = REGISTRY.lock() else {
122        return;
123    };
124    if reg.entries.values().all(|e| e.count == e.flushed) {
125        reg.last_flush = Some(Instant::now());
126        return;
127    }
128    let mut disk = read_disk();
129    for (key, e) in &mut reg.entries {
130        let delta = e.count - e.flushed;
131        if delta == 0 {
132            continue;
133        }
134        let rec = disk.entry((*key).to_string()).or_default();
135        rec.count += delta;
136        rec.last_unix = rec.last_unix.max(e.last_unix);
137        e.flushed = e.count;
138    }
139    if let Some(path) = activity_path() {
140        if let Some(parent) = path.parent() {
141            let _ = std::fs::create_dir_all(parent);
142        }
143        if let Ok(json) = serde_json::to_string_pretty(&disk) {
144            let _ = crate::config_io::write_atomic(&path, &json);
145        }
146    }
147    reg.last_flush = Some(Instant::now());
148}
149
150/// Debounced [`flush`]: persists at most once per `FLUSH_DEBOUNCE`. Safe to
151/// call from `post_dispatch` on every tool call.
152pub fn flush_if_due() {
153    let due = {
154        match REGISTRY.lock() {
155            Ok(reg) => reg.last_flush.is_none_or(|t| t.elapsed() >= FLUSH_DEBOUNCE),
156            Err(_) => false,
157        }
158    };
159    if due {
160        flush();
161    }
162}
163
164/// Snapshot of all known subsystems merged with persisted activity, for the
165/// in-process and CLI reports. Reads disk and overlays unflushed in-memory
166/// deltas so a freshly-active subsystem shows up even before the next flush.
167pub fn snapshot() -> Vec<(&'static Subsystem, Activity)> {
168    let mut disk = read_disk();
169    if let Ok(reg) = REGISTRY.lock() {
170        for (key, e) in &reg.entries {
171            let rec = disk.entry((*key).to_string()).or_default();
172            rec.count += e.count - e.flushed;
173            rec.last_unix = rec.last_unix.max(e.last_unix);
174        }
175    }
176    SUBSYSTEMS
177        .iter()
178        .map(|s| {
179            let act = disk.get(s.key).cloned().unwrap_or_default();
180            (s, act)
181        })
182        .collect()
183}
184
185/// Render a human-readable cognition activity report.
186pub fn format_report() -> String {
187    let snap = snapshot();
188    let active = snap.iter().filter(|(_, a)| a.count > 0).count();
189    let total = snap.len();
190    let mut out = String::new();
191    out.push_str(&format!(
192        "Cognition subsystems: {active}/{total} active ({total} wired)\n\n"
193    ));
194    for (sys, act) in snap {
195        let status = if act.count > 0 { "active" } else { "idle  " };
196        let last = if act.last_unix > 0 {
197            format_age(now_unix() - act.last_unix)
198        } else {
199            "never".to_string()
200        };
201        out.push_str(&format!(
202            "  [{status}] {label:<26} count={count:<7} last={last:<10} {science}\n",
203            label = sys.label,
204            count = act.count,
205            science = sys.science,
206        ));
207    }
208    out
209}
210
211/// Machine-readable variant for `--json`.
212pub fn snapshot_json() -> String {
213    let map: HashMap<&str, Activity> = snapshot().into_iter().map(|(s, a)| (s.key, a)).collect();
214    serde_json::to_string_pretty(&map).unwrap_or_else(|_| "{}".to_string())
215}
216
217fn format_age(secs: i64) -> String {
218    if secs < 0 {
219        return "just now".to_string();
220    }
221    if secs < 60 {
222        format!("{secs}s ago")
223    } else if secs < 3600 {
224        format!("{}m ago", secs / 60)
225    } else if secs < 86400 {
226        format!("{}h ago", secs / 3600)
227    } else {
228        format!("{}d ago", secs / 86400)
229    }
230}
231
232#[cfg(test)]
233pub mod tests {
234    use super::*;
235
236    #[test]
237    fn tick_increments_in_memory_snapshot() {
238        // Use a dedicated data dir so flush/snapshot stay isolated.
239        let _env = crate::core::data_dir::test_env_lock();
240        let dir = tempfile::tempdir().unwrap();
241        crate::test_env::set_var("LEAN_CTX_DATA_DIR", dir.path());
242
243        tick("phi_recompute");
244        tick("phi_recompute");
245        let snap = snapshot();
246        let phi = snap
247            .iter()
248            .find(|(s, _)| s.key == "phi_recompute")
249            .map(|(_, a)| a.count)
250            .unwrap();
251        assert!(phi >= 2, "tick should be reflected in snapshot, got {phi}");
252    }
253
254    #[test]
255    fn flush_persists_and_accumulates() {
256        let _env = crate::core::data_dir::test_env_lock();
257        let dir = tempfile::tempdir().unwrap();
258        crate::test_env::set_var("LEAN_CTX_DATA_DIR", dir.path());
259
260        tick("immune_detector");
261        flush();
262        let disk = read_disk();
263        assert!(disk.get("immune_detector").is_some_and(|a| a.count >= 1));
264
265        // A second flush of new ticks must add, not clobber.
266        tick("immune_detector");
267        flush();
268        let disk2 = read_disk();
269        assert!(
270            disk2.get("immune_detector").unwrap().count >= 2,
271            "deltas should accumulate across flushes"
272        );
273    }
274
275    #[test]
276    fn report_lists_all_subsystems() {
277        let _env = crate::core::data_dir::test_env_lock();
278        let dir = tempfile::tempdir().unwrap();
279        crate::test_env::set_var("LEAN_CTX_DATA_DIR", dir.path());
280        let report = format_report();
281        for sys in SUBSYSTEMS {
282            assert!(
283                report.contains(sys.label),
284                "report must mention {}",
285                sys.label
286            );
287        }
288    }
289}
lean_ctx/core/introspect.rs

lean_ctx/core/
introspect.rs
