everruns_core/
compaction_checkpoint.rs

1//! Durable replacement context for compacted model history.
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use std::collections::{HashMap, VecDeque};
6use tokio::sync::Mutex;
7#[cfg(test)]
8use tokio::sync::RwLock;
9use uuid::Uuid;
10
11use crate::ProviderOpaqueContext;
12use crate::error::Result;
13use crate::typed_id::SessionId;
14
15pub const COMPACTION_CHECKPOINT_FORMAT_VERSION: u32 = 1;
16
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(tag = "type", rename_all = "snake_case")]
19pub enum CompactionCheckpointPayload {
20    ProviderOpaque { context: ProviderOpaqueContext },
21    Summary { text: String },
22}
23
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct CompactionCheckpoint {
26    pub id: Uuid,
27    pub session_id: SessionId,
28    pub source_sequence: i64,
29    pub provider_type: String,
30    pub model: String,
31    pub format_version: u32,
32    pub payload: CompactionCheckpointPayload,
33}
34
35#[derive(Debug, Clone, Copy, PartialEq, Eq)]
36pub struct ProactiveCompactionAttempt {
37    pub source_sequence: i64,
38    pub estimated_input_tokens: u64,
39    pub input_message_count: usize,
40    pub source_fingerprint: [u8; 32],
41}
42
43impl CompactionCheckpoint {
44    pub fn is_compatible(&self, provider_type: &str, model: &str) -> bool {
45        self.format_version == COMPACTION_CHECKPOINT_FORMAT_VERSION
46            && self.provider_type == provider_type
47            && self.model == model
48    }
49}
50
51#[async_trait]
52pub trait CompactionCheckpointStore: Send + Sync {
53    async fn get_latest(
54        &self,
55        session_id: SessionId,
56        provider_type: &str,
57        model: &str,
58    ) -> Result<Option<CompactionCheckpoint>>;
59
60    /// Install a checkpoint only if no newer source boundary is canonical.
61    async fn install(&self, checkpoint: CompactionCheckpoint) -> Result<bool>;
62
63    /// Latest pressure snapshot where proactive native compaction was attempted.
64    ///
65    /// This process-lifetime watermark prevents an ineffective or failed
66    /// provider call from being repeated on every reasoning iteration. The
67    /// source fingerprint makes it lineage-local across rollback/branch
68    /// selection. Durable checkpoints remain the semantic replacement
69    /// contract; this marker only controls retry pressure and never substitutes
70    /// for a checkpoint.
71    async fn get_proactive_attempt(
72        &self,
73        session_id: SessionId,
74        provider_type: &str,
75        model: &str,
76    ) -> Result<Option<ProactiveCompactionAttempt>>;
77
78    async fn record_proactive_attempt(
79        &self,
80        session_id: SessionId,
81        provider_type: &str,
82        model: &str,
83        attempt: ProactiveCompactionAttempt,
84    ) -> Result<()>;
85}
86
87#[cfg(test)]
88type CheckpointKey = (SessionId, String, String, u32);
89type AttemptKey = (SessionId, String, String);
90
91const DEFAULT_PROACTIVE_ATTEMPT_CAPACITY: usize = 4_096;
92
93/// Bounded process-local retry watermarks shared by successive reason atoms.
94/// Oldest session/provider/model entries are evicted once capacity is reached.
95#[derive(Debug)]
96pub struct ProactiveCompactionAttemptTracker {
97    capacity: usize,
98    state: Mutex<ProactiveAttemptState>,
99}
100
101#[derive(Debug, Default)]
102struct ProactiveAttemptState {
103    attempts: HashMap<AttemptKey, ProactiveCompactionAttempt>,
104    insertion_order: VecDeque<AttemptKey>,
105}
106
107impl Default for ProactiveCompactionAttemptTracker {
108    fn default() -> Self {
109        Self {
110            capacity: DEFAULT_PROACTIVE_ATTEMPT_CAPACITY,
111            state: Mutex::new(ProactiveAttemptState::default()),
112        }
113    }
114}
115
116impl ProactiveCompactionAttemptTracker {
117    pub async fn get(
118        &self,
119        session_id: SessionId,
120        provider_type: &str,
121        model: &str,
122    ) -> Option<ProactiveCompactionAttempt> {
123        self.state
124            .lock()
125            .await
126            .attempts
127            .get(&(session_id, provider_type.to_string(), model.to_string()))
128            .copied()
129    }
130
131    pub async fn record(
132        &self,
133        session_id: SessionId,
134        provider_type: &str,
135        model: &str,
136        attempt: ProactiveCompactionAttempt,
137    ) {
138        let key = (session_id, provider_type.to_string(), model.to_string());
139        let mut state = self.state.lock().await;
140        if let Some(current) = state.attempts.get_mut(&key) {
141            if attempt.source_sequence >= current.source_sequence {
142                *current = attempt;
143            }
144            return;
145        }
146        while state.attempts.len() >= self.capacity {
147            if let Some(oldest) = state.insertion_order.pop_front() {
148                state.attempts.remove(&oldest);
149            }
150        }
151        state.insertion_order.push_back(key.clone());
152        state.attempts.insert(key, attempt);
153    }
154}
155
156/// In-memory implementation used by collocated unit tests.
157#[cfg(test)]
158#[derive(Debug, Default)]
159pub struct InMemoryCompactionCheckpointStore {
160    checkpoints: RwLock<HashMap<CheckpointKey, CompactionCheckpoint>>,
161    proactive_attempts: ProactiveCompactionAttemptTracker,
162}
163
164#[cfg(test)]
165#[async_trait]
166impl CompactionCheckpointStore for InMemoryCompactionCheckpointStore {
167    async fn get_latest(
168        &self,
169        session_id: SessionId,
170        provider_type: &str,
171        model: &str,
172    ) -> Result<Option<CompactionCheckpoint>> {
173        Ok(self
174            .checkpoints
175            .read()
176            .await
177            .get(&(
178                session_id,
179                provider_type.to_string(),
180                model.to_string(),
181                COMPACTION_CHECKPOINT_FORMAT_VERSION,
182            ))
183            .cloned())
184    }
185
186    async fn install(&self, checkpoint: CompactionCheckpoint) -> Result<bool> {
187        let key = (
188            checkpoint.session_id,
189            checkpoint.provider_type.clone(),
190            checkpoint.model.clone(),
191            checkpoint.format_version,
192        );
193        let mut checkpoints = self.checkpoints.write().await;
194        if checkpoints
195            .get(&key)
196            .is_some_and(|current| current.source_sequence >= checkpoint.source_sequence)
197        {
198            return Ok(false);
199        }
200        checkpoints.insert(key, checkpoint);
201        Ok(true)
202    }
203
204    async fn get_proactive_attempt(
205        &self,
206        session_id: SessionId,
207        provider_type: &str,
208        model: &str,
209    ) -> Result<Option<ProactiveCompactionAttempt>> {
210        Ok(self
211            .proactive_attempts
212            .get(session_id, provider_type, model)
213            .await)
214    }
215
216    async fn record_proactive_attempt(
217        &self,
218        session_id: SessionId,
219        provider_type: &str,
220        model: &str,
221        attempt: ProactiveCompactionAttempt,
222    ) -> Result<()> {
223        self.proactive_attempts
224            .record(session_id, provider_type, model, attempt)
225            .await;
226        Ok(())
227    }
228}
229
230#[cfg(test)]
231mod tests {
232    use super::*;
233    use crate::{CompactOutputItem, ProviderOpaqueContext};
234
235    fn checkpoint(session_id: SessionId, source_sequence: i64) -> CompactionCheckpoint {
236        CompactionCheckpoint {
237            id: Uuid::now_v7(),
238            session_id,
239            source_sequence,
240            provider_type: "openai".to_string(),
241            model: "gpt-5.4".to_string(),
242            format_version: 1,
243            payload: CompactionCheckpointPayload::ProviderOpaque {
244                context: ProviderOpaqueContext::OpenResponsesCompact {
245                    reasoning_state: None,
246                    output: vec![CompactOutputItem::Compaction {
247                        encrypted_content: format!("opaque-{source_sequence}"),
248                    }],
249                },
250            },
251        }
252    }
253
254    #[tokio::test]
255    async fn install_is_monotonic_and_failed_cas_leaves_current_checkpoint_unchanged() {
256        let store = InMemoryCompactionCheckpointStore::default();
257        let session_id = SessionId::new();
258        let current = checkpoint(session_id, 12);
259        assert!(store.install(current.clone()).await.unwrap());
260
261        assert!(!store.install(checkpoint(session_id, 11)).await.unwrap());
262        assert!(!store.install(checkpoint(session_id, 12)).await.unwrap());
263        assert_eq!(
264            store
265                .get_latest(session_id, "openai", "gpt-5.4")
266                .await
267                .unwrap(),
268            Some(current)
269        );
270    }
271
272    #[test]
273    fn compatibility_requires_exact_provider_model_and_format() {
274        let mut checkpoint = checkpoint(SessionId::new(), 1);
275        assert!(checkpoint.is_compatible("openai", "gpt-5.4"));
276        assert!(!checkpoint.is_compatible("openrouter", "gpt-5.4"));
277        assert!(!checkpoint.is_compatible("openai", "gpt-5.5"));
278        checkpoint.format_version += 1;
279        assert!(!checkpoint.is_compatible("openai", "gpt-5.4"));
280    }
281
282    #[tokio::test]
283    async fn proactive_attempt_updates_preserve_monotonicity_and_fifo_eviction() {
284        let tracker = ProactiveCompactionAttemptTracker {
285            capacity: 2,
286            state: Mutex::default(),
287        };
288        let session = SessionId::from_seed(1);
289        let original = ProactiveCompactionAttempt {
290            source_sequence: 2,
291            estimated_input_tokens: 10_000,
292            input_message_count: 1,
293            source_fingerprint: [7; 32],
294        };
295        let newer = ProactiveCompactionAttempt {
296            source_sequence: 3,
297            estimated_input_tokens: 20_000,
298            input_message_count: 2,
299            source_fingerprint: [8; 32],
300        };
301        assert_eq!(tracker.get(session, "openai", "a").await, None);
302        tracker.record(session, "openai", "a", original).await;
303        tracker.record(session, "openai", "b", original).await;
304        tracker.record(session, "openai", "a", newer).await;
305        tracker.record(session, "openai", "a", original).await;
306        assert_eq!(tracker.get(session, "openai", "a").await, Some(newer));
307        let equal = ProactiveCompactionAttempt {
308            estimated_input_tokens: 30_000,
309            source_fingerprint: [9; 32],
310            ..newer
311        };
312        tracker.record(session, "openai", "a", equal).await;
313        assert_eq!(tracker.get(session, "openai", "a").await, Some(equal));
314        tracker.record(session, "other", "a", original).await;
315        assert_eq!(tracker.get(session, "openai", "a").await, None);
316        assert_eq!(tracker.get(session, "openai", "b").await, Some(original));
317        assert_eq!(tracker.get(session, "other", "a").await, Some(original));
318        tracker
319            .record(SessionId::from_seed(2), "other", "a", newer)
320            .await;
321        assert_eq!(tracker.get(session, "openai", "b").await, None);
322        assert_eq!(tracker.get(session, "other", "a").await, Some(original));
323        assert_eq!(
324            tracker.get(SessionId::from_seed(2), "other", "a").await,
325            Some(newer)
326        );
327        let state = tracker.state.lock().await;
328        assert_eq!(state.attempts.len(), 2);
329        assert_eq!(state.insertion_order.len(), 2);
330    }
331}
everruns_core/compaction_checkpoint.rs

everruns_core/
compaction_checkpoint.rs
