lean_ctx_server/server/
compaction_sync.rs

1// SPDX-License-Identifier: Apache-2.0
2use std::path::Path;
3use std::sync::atomic::{AtomicU64, Ordering};
4
5use lean_ctx::core::cache::SessionCache;
6use lean_ctx::core::context_radar::RadarEvent;
7
8pub static LAST_COMPACTION_TS: AtomicU64 = AtomicU64::new(0);
9
10/// Effective cache policy: "aggressive" (default), "safe", or "off".
11pub fn effective_cache_policy() -> &'static str {
12    // Resolve on each call: tests and embedded callers may change the
13    // process override between engine instances; a process-wide OnceLock
14    // would permanently retain the first caller's policy.
15    let value = std::env::var("LEAN_CTX_CACHE_POLICY")
16        .ok()
17        .map(|v| v.trim().to_lowercase())
18        .filter(|v| matches!(v.as_str(), "aggressive" | "safe" | "off"))
19        .or_else(|| {
20            lean_ctx::core::config::Config::load()
21                .cache_policy
22                .map(|v| v.to_lowercase())
23        })
24        .unwrap_or_else(|| "aggressive".to_string());
25    match value.as_str() {
26        "safe" => "safe",
27        "off" => "off",
28        _ => "aggressive",
29    }
30}
31
32/// Check if a host compaction event occurred since our last check.
33/// If so, reset all `full_content_delivered` flags so the next read
34/// delivers full content instead of a stub.
35///
36/// Detection strategy (#808):
37/// 1. Check `last_compaction.json` marker file first (written atomically by
38///    the observe hook). This is only a few bytes, immune to being displaced
39///    by large events.
40/// 2. Fall back to scanning the tail of `context_radar.jsonl` (256 KB window)
41///    for older hook binaries that don't write the marker.
42pub fn sync_if_compacted(cache: &mut SessionCache, data_dir: &Path) -> bool {
43    let last_seen = LAST_COMPACTION_TS.load(Ordering::Relaxed);
44
45    let latest_ts = check_compaction_marker(data_dir, last_seen)
46        .or_else(|| scan_radar_tail(data_dir, last_seen));
47
48    let Some(latest_compaction_ts) = latest_ts else {
49        return false;
50    };
51
52    LAST_COMPACTION_TS.store(latest_compaction_ts, Ordering::Relaxed);
53    lean_ctx::core::search_delta::reset();
54    let reset_count = cache.reset_delivery_flags();
55    // Drop the persistent stub index too (#955): the conversation's context was
56    // summarised away, so neither a warm nor a cold stub may claim "you already
57    // have this". Writes the emptied index synchronously so a restart in the
58    // crash window can't resurrect a pre-compaction stub.
59    lean_ctx::core::read_stub_index::reset_in_dir(data_dir);
60    // The Context Kernel's content-dedup ledger answers re-reads with
61    // `unchanged (ref:…), already in context` on its own; it must forget the
62    // summarised deliveries as well, or the model re-reads with fresh=true (#2147).
63    lean_ctx::core::context_kernel::dedup_wiring::forget_deliveries();
64    if reset_count > 0 {
65        // #1781: `eprintln!` panics when the write fails, and a host closing the
66        // pipe is routine for an MCP server (restart, aborted call, IDE reload).
67        // On Windows that surfaced as `failed printing to stderr: The pipe is
68        // being closed. (os error 232)`. `tracing` returns the error instead of
69        // panicking, and the subscriber already targets stderr.
70        tracing::info!(
71            "[lean-ctx] compaction detected — reset {reset_count} delivery flags for re-read"
72        );
73    }
74
75    // The caller's own session, never the project's latest saved one, which
76    // may be another agent's (I07/S03).
77    let caller = lean_ctx::core::session::caller::current();
78    std::thread::spawn(move || {
79        lean_ctx::core::session::caller::scope(caller, || {
80            let root = lean_ctx::core::session::caller::with(|session| {
81                (!session.findings.is_empty() || !session.decisions.is_empty())
82                    .then(|| session.project_root.clone())
83                    .flatten()
84            })
85            .flatten();
86            if let Some(root) = root {
87                crate::tools::startup::auto_consolidate_knowledge(&root);
88            }
89        });
90    });
91
92    true
93}
94
95// ---------------------------------------------------------------------------
96// Detection: marker file (primary, #808)
97// ---------------------------------------------------------------------------
98
99/// Read `last_compaction.json` — a few-byte file written atomically by the
100/// observe hook. Returns `Some(ts)` if the marker exists and its timestamp
101/// is newer than `since_ts`.
102fn check_compaction_marker(data_dir: &Path, since_ts: u64) -> Option<u64> {
103    let marker_path = data_dir.join("last_compaction.json");
104    let content = std::fs::read_to_string(&marker_path).ok()?;
105    let parsed: serde_json::Value = serde_json::from_str(content.trim()).ok()?;
106    let ts = parsed.get("ts")?.as_u64()?;
107    if ts > since_ts { Some(ts) } else { None }
108}
109
110// ---------------------------------------------------------------------------
111// Detection: radar tail scan (fallback for older hook binaries)
112// ---------------------------------------------------------------------------
113
114/// Scan only the tail of radar JSONL for a compaction event newer than `since_ts`.
115/// #808: window widened from 4 KB to 256 KB so large `agent_response` /
116/// `thinking` events (up to 50 000 chars each) cannot bury the compaction entry.
117fn scan_radar_tail(data_dir: &Path, since_ts: u64) -> Option<u64> {
118    use std::io::{Read, Seek, SeekFrom};
119
120    let radar_path = data_dir.join("context_radar.jsonl");
121    let mut file = std::fs::File::open(&radar_path).ok()?;
122    let file_len = file.metadata().ok()?.len();
123
124    const TAIL_BYTES: u64 = 256 * 1024; // #808: was 4096
125    let content = if file_len <= TAIL_BYTES {
126        let mut s = String::new();
127        file.read_to_string(&mut s).ok()?;
128        s
129    } else {
130        file.seek(SeekFrom::End(-(TAIL_BYTES as i64))).ok()?;
131        let mut buf = vec![0u8; TAIL_BYTES as usize];
132        let n = file.read(&mut buf).ok()?;
133        let s = String::from_utf8_lossy(&buf[..n]).into_owned();
134        if let Some(idx) = s.find('\n') {
135            s[idx + 1..].to_string()
136        } else {
137            s
138        }
139    };
140
141    for line in content.lines().rev() {
142        if line.is_empty() {
143            continue;
144        }
145        let event: RadarEvent = match serde_json::from_str(line) {
146            Ok(e) => e,
147            Err(_) => continue,
148        };
149        if event.ts <= since_ts {
150            break;
151        }
152        if event.event_type == "compaction" {
153            return Some(event.ts);
154        }
155    }
156    None
157}
158
159#[cfg(test)]
160mod tests {
161    use super::*;
162    use serial_test::serial;
163    use std::io::Write;
164    use tempfile::TempDir;
165
166    fn make_cache_with_delivered(paths: &[&str]) -> SessionCache {
167        let mut cache = SessionCache::default();
168        for p in paths {
169            cache.store(p, "hello world");
170            cache.mark_full_delivered(p);
171        }
172        cache
173    }
174
175    #[test]
176    #[serial]
177    fn no_reset_without_compaction_event() {
178        let dir = TempDir::new().unwrap();
179        let radar = dir.path().join("context_radar.jsonl");
180        let mut f = std::fs::File::create(&radar).unwrap();
181        writeln!(f, r#"{{"ts":1000,"event_type":"mcp_call","tokens":50}}"#).unwrap();
182        drop(f);
183
184        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
185        let mut cache = make_cache_with_delivered(&["/tmp/a.rs"]);
186        assert!(!sync_if_compacted(&mut cache, dir.path()));
187        assert!(cache.is_full_delivered("/tmp/a.rs"));
188    }
189
190    #[test]
191    #[serial]
192    fn resets_after_compaction() {
193        let dir = TempDir::new().unwrap();
194        let radar = dir.path().join("context_radar.jsonl");
195        let mut f = std::fs::File::create(&radar).unwrap();
196        writeln!(f, r#"{{"ts":1000,"event_type":"mcp_call","tokens":50}}"#).unwrap();
197        writeln!(f, r#"{{"ts":2000,"event_type":"compaction","tokens":0}}"#).unwrap();
198        drop(f);
199
200        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
201        let mut cache = make_cache_with_delivered(&["/tmp/a.rs", "/tmp/b.rs"]);
202
203        assert!(cache.is_full_delivered("/tmp/a.rs"));
204        assert!(sync_if_compacted(&mut cache, dir.path()));
205        assert!(!cache.is_full_delivered("/tmp/a.rs"));
206        assert!(!cache.is_full_delivered("/tmp/b.rs"));
207    }
208
209    #[test]
210    #[serial]
211    fn does_not_double_reset() {
212        let dir = TempDir::new().unwrap();
213        let radar = dir.path().join("context_radar.jsonl");
214        let mut f = std::fs::File::create(&radar).unwrap();
215        writeln!(f, r#"{{"ts":2000,"event_type":"compaction","tokens":0}}"#).unwrap();
216        drop(f);
217
218        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
219        let mut cache = make_cache_with_delivered(&["/tmp/a.rs"]);
220        assert!(sync_if_compacted(&mut cache, dir.path()));
221        assert!(!cache.is_full_delivered("/tmp/a.rs"));
222
223        cache.mark_full_delivered("/tmp/a.rs");
224        assert!(!sync_if_compacted(&mut cache, dir.path()));
225        assert!(cache.is_full_delivered("/tmp/a.rs"));
226    }
227
228    // --- #808 new tests ---
229
230    /// The marker file alone (without a radar entry) triggers a reset,
231    /// and the same marker timestamp does not trigger it a second time.
232    #[test]
233    #[serial]
234    fn marker_alone_triggers_reset() {
235        let dir = TempDir::new().unwrap();
236        // No context_radar.jsonl — only the marker
237        let marker = dir.path().join("last_compaction.json");
238        std::fs::write(&marker, r#"{"ts":5000}"#).unwrap();
239        // Need a radar file to exist (sync_if_compacted checks marker OR radar)
240        let radar = dir.path().join("context_radar.jsonl");
241        std::fs::write(&radar, "").unwrap();
242
243        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
244        let mut cache = make_cache_with_delivered(&["/tmp/x.rs"]);
245        assert!(cache.is_full_delivered("/tmp/x.rs"));
246
247        // First call: marker is newer than last_seen (0) → reset
248        assert!(sync_if_compacted(&mut cache, dir.path()));
249        assert!(!cache.is_full_delivered("/tmp/x.rs"));
250        assert_eq!(LAST_COMPACTION_TS.load(Ordering::Relaxed), 5000);
251
252        // Re-deliver, then call again: same marker ts → no reset
253        cache.mark_full_delivered("/tmp/x.rs");
254        assert!(!sync_if_compacted(&mut cache, dir.path()));
255        assert!(cache.is_full_delivered("/tmp/x.rs"));
256    }
257
258    /// A compaction event buried under a large (10 KB+) event line is still
259    /// detected via the widened 256 KB fallback window. This reproduces the
260    /// original bug — the old 4 KB window would miss it.
261    #[test]
262    #[serial]
263    fn compaction_buried_under_large_event_line_is_still_detected() {
264        let dir = TempDir::new().unwrap();
265        let radar = dir.path().join("context_radar.jsonl");
266        let mut f = std::fs::File::create(&radar).unwrap();
267
268        // Write the compaction event
269        writeln!(f, r#"{{"ts":3000,"event_type":"compaction","tokens":0}}"#).unwrap();
270
271        // Bury it under a large agent_response event (10 KB content)
272        let large_content = "x".repeat(10_000);
273        writeln!(
274            f,
275            r#"{{"ts":3001,"event_type":"agent_response","tokens":2500,"content":"{large_content}"}}"#
276        )
277        .unwrap();
278        drop(f);
279
280        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
281        let mut cache = make_cache_with_delivered(&["/tmp/buried.rs"]);
282        assert!(cache.is_full_delivered("/tmp/buried.rs"));
283
284        // Must find the compaction despite the large event after it
285        assert!(sync_if_compacted(&mut cache, dir.path()));
286        assert!(!cache.is_full_delivered("/tmp/buried.rs"));
287    }
288
289    /// Marker file takes priority over the radar scan: even if the radar
290    /// has no compaction event, the marker triggers a reset.
291    #[test]
292    #[serial]
293    fn marker_takes_priority_over_empty_radar() {
294        let dir = TempDir::new().unwrap();
295        let radar = dir.path().join("context_radar.jsonl");
296        let mut f = std::fs::File::create(&radar).unwrap();
297        writeln!(f, r#"{{"ts":1000,"event_type":"mcp_call","tokens":50}}"#).unwrap();
298        drop(f);
299
300        let marker = dir.path().join("last_compaction.json");
301        std::fs::write(&marker, r#"{"ts":4000}"#).unwrap();
302
303        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
304        let mut cache = make_cache_with_delivered(&["/tmp/priority.rs"]);
305        assert!(sync_if_compacted(&mut cache, dir.path()));
306        assert!(!cache.is_full_delivered("/tmp/priority.rs"));
307        assert_eq!(LAST_COMPACTION_TS.load(Ordering::Relaxed), 4000);
308    }
309
310    /// A corrupt or empty marker file is ignored — falls back to radar scan.
311    #[test]
312    #[serial]
313    fn corrupt_marker_falls_back_to_radar() {
314        let dir = TempDir::new().unwrap();
315        let radar = dir.path().join("context_radar.jsonl");
316        let mut f = std::fs::File::create(&radar).unwrap();
317        writeln!(f, r#"{{"ts":6000,"event_type":"compaction","tokens":0}}"#).unwrap();
318        drop(f);
319
320        let marker = dir.path().join("last_compaction.json");
321        std::fs::write(&marker, "not valid json").unwrap();
322
323        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
324        let mut cache = make_cache_with_delivered(&["/tmp/fallback.rs"]);
325        // Corrupt marker ignored → radar scan finds the compaction
326        assert!(sync_if_compacted(&mut cache, dir.path()));
327        assert!(!cache.is_full_delivered("/tmp/fallback.rs"));
328    }
329
330    /// GitHub #2147: after compaction the kernel's content-dedup ledger still
331    /// answered a re-read with `unchanged (ref:…), already in context`, so the
332    /// model had to re-read with fresh=true.
333    #[test]
334    #[serial]
335    fn compaction_clears_the_kernel_dedup_ledger() {
336        use lean_ctx::core::context_kernel::{ctx_read_dedup, dedup_wiring, kernel_config};
337        let _kernel = kernel_config::KERNEL_TEST_LOCK
338            .lock()
339            .unwrap_or_else(std::sync::PoisonError::into_inner);
340        kernel_config::reset_features();
341        dedup_wiring::reset_dedup();
342
343        assert_eq!(ctx_read_dedup::try_dedup("/tmp/k.rs", "body", false), None);
344        assert!(ctx_read_dedup::try_dedup("/tmp/k.rs", "body", false).is_some());
345
346        let dir = TempDir::new().unwrap();
347        std::fs::write(dir.path().join("last_compaction.json"), r#"{"ts":7000}"#).unwrap();
348        std::fs::write(dir.path().join("context_radar.jsonl"), "").unwrap();
349        LAST_COMPACTION_TS.store(0, Ordering::Relaxed);
350        let mut cache = SessionCache::default();
351        assert!(sync_if_compacted(&mut cache, dir.path()));
352
353        assert_eq!(
354            ctx_read_dedup::try_dedup("/tmp/k.rs", "body", false),
355            None,
356            "a read after compaction must deliver the content, not a stub"
357        );
358        dedup_wiring::reset_dedup();
359    }
360}
lean_ctx_server/server/compaction_sync.rs

lean_ctx_server/server/
compaction_sync.rs
