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

everruns_core/
compaction_checkpoint.rs
