tsift_quality/
cycle_packet_cache.rs

1//! Cycle-scoped packet reuse cache (#gpackreuse).
2//!
3//! Caches graph/context packets across a single agent-doc cycle so that
4//! repeated tsift CLI invocations sharing the same source/document/staged-diff
5//! watermarks skip redundant computation and report stable packet ids.
6//!
7//! Packet kinds:
8//! - `evidence`: graph-db evidence reports keyed by `packet_id`
9//! - `context_pack`: context-pack reports keyed by watermark triple
10//! - `impact`: impact reports keyed by source watermark + revision
11//! - `conflict_matrix`: conflict-matrix reports keyed by prepared inputs + targets
12//!
13//! Spec: see specs/graph.md § "Cycle Packet Cache".
14
15use serde::{Deserialize, Serialize};
16use std::fs;
17use std::path::{Path, PathBuf};
18use std::time::SystemTime;
19
20pub const CYCLE_PACKET_CACHE_VERSION: &str = "cycle-packet-cache-v1";
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
23#[serde(rename_all = "snake_case")]
24pub enum CyclePacketKind {
25    Evidence,
26    ContextPack,
27    Impact,
28    ConflictMatrix,
29}
30
31impl CyclePacketKind {
32    pub fn dir_name(&self) -> &'static str {
33        match self {
34            CyclePacketKind::Evidence => "evidence",
35            CyclePacketKind::ContextPack => "context-pack",
36            CyclePacketKind::Impact => "impact",
37            CyclePacketKind::ConflictMatrix => "conflict-matrix",
38        }
39    }
40}
41
42#[derive(Debug, Clone, Serialize, Deserialize)]
43pub struct CyclePacketCacheEntry {
44    pub version: String,
45    pub kind: CyclePacketKind,
46    pub key: String,
47    pub packet_id: String,
48    pub source_watermark: String,
49    pub document_watermark: String,
50    pub staged_diff_watermark: String,
51    pub skipped_phases: Vec<String>,
52    pub compute_micros: u128,
53}
54
55#[derive(Debug, Clone, Serialize, Deserialize)]
56pub struct CyclePacketCacheHitReport {
57    pub kind: CyclePacketKind,
58    pub key: String,
59    pub packet_id: String,
60    pub hit_status: String,
61    pub skipped_phases: Vec<String>,
62    pub lookup_micros: u128,
63}
64
65pub fn cycle_packet_cache_dir(root: &Path) -> PathBuf {
66    root.join(".tsift/cycle-packet-cache")
67}
68
69pub fn cycle_packet_cache_path(root: &Path, kind: CyclePacketKind, key: &str) -> PathBuf {
70    cycle_packet_cache_dir(root)
71        .join(kind.dir_name())
72        .join(format!("{key}.json"))
73}
74
75pub fn cycle_packet_read_cache<T: for<'de> Deserialize<'de>>(
76    root: &Path,
77    kind: CyclePacketKind,
78    key: &str,
79) -> Option<T> {
80    let path = cycle_packet_cache_path(root, kind, key);
81    let bytes = fs::read(path).ok()?;
82    serde_json::from_slice(&bytes).ok()
83}
84
85pub fn cycle_packet_write_cache<T: Serialize>(
86    root: &Path,
87    kind: CyclePacketKind,
88    key: &str,
89    value: &T,
90) {
91    let path = cycle_packet_cache_path(root, kind, key);
92    let Some(parent) = path.parent() else {
93        return;
94    };
95    if fs::create_dir_all(parent).is_err() {
96        return;
97    }
98    if let Ok(bytes) = serde_json::to_vec(value) {
99        let _ = fs::write(path, bytes);
100    }
101}
102
103pub fn cycle_packet_watermark_key(
104    source_watermark: &str,
105    document_watermark: &str,
106    staged_diff_watermark: &str,
107    extra: &[&str],
108) -> String {
109    let mut parts = vec![
110        format!("version:{CYCLE_PACKET_CACHE_VERSION}"),
111        format!("source:{source_watermark}"),
112        format!("document:{document_watermark}"),
113        format!("staged_diff:{staged_diff_watermark}"),
114    ];
115    for e in extra {
116        parts.push(e.to_string());
117    }
118    blake3::hash(parts.join("\n").as_bytes())
119        .to_hex()
120        .to_string()
121}
122
123pub fn cycle_packet_evidence_key(packet_id: &str) -> String {
124    cycle_packet_watermark_key("evidence", "evidence", "evidence", &[packet_id])
125}
126
127pub fn build_cache_hit_report(
128    kind: CyclePacketKind,
129    key: &str,
130    packet_id: &str,
131    status: &str,
132    skipped_phases: &[&str],
133    lookup_micros: u128,
134) -> CyclePacketCacheHitReport {
135    CyclePacketCacheHitReport {
136        kind,
137        key: key.to_string(),
138        packet_id: packet_id.to_string(),
139        hit_status: status.to_string(),
140        skipped_phases: skipped_phases.iter().map(|s| s.to_string()).collect(),
141        lookup_micros,
142    }
143}
144
145pub const CYCLE_PACKET_CACHE_DEFAULT_TTL_SECS: u64 = 24 * 60 * 60;
146pub const CYCLE_PACKET_CACHE_DEFAULT_MAX_BYTES: u64 = 50 * 1024 * 1024;
147
148#[derive(Debug, Clone, Serialize, Deserialize)]
149pub struct CyclePacketCacheEvictionReport {
150    pub scanned_entries: usize,
151    pub evicted_entries: usize,
152    pub evicted_bytes: u64,
153    pub remaining_entries: usize,
154    pub remaining_bytes: u64,
155    pub ttl_secs: u64,
156    pub max_bytes: u64,
157}
158
159pub fn cycle_packet_cache_stats(root: &Path) -> (usize, u64) {
160    let cache_dir = cycle_packet_cache_dir(root);
161    if !cache_dir.exists() {
162        return (0, 0);
163    }
164    let mut count = 0usize;
165    let mut total_bytes = 0u64;
166    if let Ok(entries) = fs::read_dir(&cache_dir) {
167        for entry in entries.flatten() {
168            if entry.file_type().map(|t| t.is_dir()).unwrap_or(false)
169                && let Ok(files) = fs::read_dir(entry.path()) {
170                    for file in files.flatten() {
171                        if file.path().extension().is_some_and(|ext| ext == "json")
172                            && let Ok(meta) = file.metadata() {
173                                count += 1;
174                                total_bytes += meta.len();
175                            }
176                    }
177                }
178        }
179    }
180    (count, total_bytes)
181}
182
183pub fn cycle_packet_cache_evict(
184    root: &Path,
185    ttl_secs: u64,
186    max_bytes: u64,
187) -> CyclePacketCacheEvictionReport {
188    let cache_dir = cycle_packet_cache_dir(root);
189    if !cache_dir.exists() {
190        return CyclePacketCacheEvictionReport {
191            scanned_entries: 0,
192            evicted_entries: 0,
193            evicted_bytes: 0,
194            remaining_entries: 0,
195            remaining_bytes: 0,
196            ttl_secs,
197            max_bytes,
198        };
199    }
200    let now = SystemTime::now();
201    let cutoff = now
202        .duration_since(SystemTime::UNIX_EPOCH)
203        .unwrap_or_default()
204        .as_secs()
205        .saturating_sub(ttl_secs);
206    let mut all_files: Vec<(PathBuf, u64, u64)> = Vec::new();
207    if let Ok(entries) = fs::read_dir(&cache_dir) {
208        for entry in entries.flatten() {
209            if entry.file_type().map(|t| t.is_dir()).unwrap_or(false)
210                && let Ok(files) = fs::read_dir(entry.path()) {
211                    for file in files.flatten() {
212                        let path = file.path();
213                        if path.extension().is_some_and(|ext| ext == "json")
214                            && let Ok(meta) = file.metadata() {
215                                let size = meta.len();
216                                let mtime_secs = meta
217                                    .modified()
218                                    .ok()
219                                    .and_then(|t| t.duration_since(SystemTime::UNIX_EPOCH).ok())
220                                    .map(|d| d.as_secs())
221                                    .unwrap_or(u64::MAX);
222                                all_files.push((path, size, mtime_secs));
223                            }
224                    }
225                }
226        }
227    }
228    let scanned = all_files.len();
229    all_files.sort_by_key(|(_, _, mtime)| *mtime);
230    let mut evicted = 0usize;
231    let mut evicted_bytes = 0u64;
232    for (path, size, mtime) in &all_files {
233        if *mtime < cutoff {
234            let _ = fs::remove_file(path);
235            evicted += 1;
236            evicted_bytes += size;
237        }
238    }
239    let remaining: u64 = all_files
240        .iter()
241        .filter(|(_, _, mtime)| *mtime >= cutoff)
242        .map(|(_, size, _)| *size)
243        .sum();
244    if remaining > max_bytes {
245        let expired: Vec<_> = all_files
246            .iter()
247            .filter(|(_, _, mtime)| *mtime >= cutoff)
248            .collect();
249        let mut kept_bytes = 0u64;
250        for (path, size, _) in expired {
251            if kept_bytes.saturating_add(*size) > max_bytes {
252                let _ = fs::remove_file(path);
253                evicted += 1;
254                evicted_bytes += size;
255            } else {
256                kept_bytes = kept_bytes.saturating_add(*size);
257            }
258        }
259    }
260    let (remaining_count, remaining_bytes) = cycle_packet_cache_stats(root);
261    CyclePacketCacheEvictionReport {
262        scanned_entries: scanned,
263        evicted_entries: evicted,
264        evicted_bytes,
265        remaining_entries: remaining_count,
266        remaining_bytes,
267        ttl_secs,
268        max_bytes,
269    }
270}
271
272#[cfg(test)]
273mod tests {
274    use super::*;
275
276    #[test]
277    fn cycle_packet_watermark_key_is_stable() {
278        let a = cycle_packet_watermark_key("s1", "d1", "sd1", &["extra"]);
279        let b = cycle_packet_watermark_key("s1", "d1", "sd1", &["extra"]);
280        assert_eq!(a, b);
281    }
282
283    #[test]
284    fn cycle_packet_watermark_key_differs_for_different_inputs() {
285        let a = cycle_packet_watermark_key("s1", "d1", "sd1", &["extra"]);
286        let b = cycle_packet_watermark_key("s2", "d1", "sd1", &["extra"]);
287        assert_ne!(a, b);
288    }
289
290    #[test]
291    fn cycle_packet_evidence_key_is_stable() {
292        let a = cycle_packet_evidence_key("gevd:abc123");
293        let b = cycle_packet_evidence_key("gevd:abc123");
294        assert_eq!(a, b);
295    }
296
297    #[test]
298    fn cycle_packet_evidence_key_differs_for_different_ids() {
299        let a = cycle_packet_evidence_key("gevd:abc123");
300        let b = cycle_packet_evidence_key("gevd:def456");
301        assert_ne!(a, b);
302    }
303
304    #[test]
305    fn cache_dir_uses_tsift_subdirectory() {
306        let dir = cycle_packet_cache_dir(Path::new("/project"));
307        assert_eq!(
308            dir,
309            PathBuf::from("/project/.tsift/cycle-packet-cache")
310        );
311    }
312
313    #[test]
314    fn cache_path_includes_kind_and_key() {
315        let path = cycle_packet_cache_path(
316            Path::new("/project"),
317            CyclePacketKind::Evidence,
318            "abc123",
319        );
320        assert_eq!(
321            path,
322            PathBuf::from("/project/.tsift/cycle-packet-cache/evidence/abc123.json")
323        );
324    }
325
326    #[test]
327    fn disk_roundtrip_preserves_entry() {
328        let dir = tempfile::tempdir().unwrap();
329        let root = dir.path();
330        let entry = CyclePacketCacheEntry {
331            version: CYCLE_PACKET_CACHE_VERSION.to_string(),
332            kind: CyclePacketKind::Evidence,
333            key: "test-key".to_string(),
334            packet_id: "gevd:abc123".to_string(),
335            source_watermark: "sw1".to_string(),
336            document_watermark: "dw1".to_string(),
337            staged_diff_watermark: "sdw1".to_string(),
338            skipped_phases: vec!["graph_db_evidence".to_string()],
339            compute_micros: 1234,
340        };
341        cycle_packet_write_cache(root, CyclePacketKind::Evidence, "test-key", &entry);
342        let loaded: CyclePacketCacheEntry =
343            cycle_packet_read_cache(root, CyclePacketKind::Evidence, "test-key").unwrap();
344        assert_eq!(loaded.version, entry.version);
345        assert_eq!(loaded.kind, entry.kind);
346        assert_eq!(loaded.key, entry.key);
347        assert_eq!(loaded.packet_id, entry.packet_id);
348        assert_eq!(loaded.source_watermark, entry.source_watermark);
349        assert_eq!(loaded.document_watermark, entry.document_watermark);
350        assert_eq!(loaded.staged_diff_watermark, entry.staged_diff_watermark);
351        assert_eq!(loaded.skipped_phases, entry.skipped_phases);
352        assert_eq!(loaded.compute_micros, entry.compute_micros);
353    }
354
355    #[test]
356    fn disk_read_returns_none_for_missing() {
357        let dir = tempfile::tempdir().unwrap();
358        let root = dir.path();
359        let result: Option<CyclePacketCacheEntry> =
360            cycle_packet_read_cache(root, CyclePacketKind::Evidence, "nonexistent");
361        assert!(result.is_none());
362    }
363
364    #[test]
365    fn kind_dir_names_are_lowercase_with_hyphens() {
366        assert_eq!(CyclePacketKind::Evidence.dir_name(), "evidence");
367        assert_eq!(CyclePacketKind::ContextPack.dir_name(), "context-pack");
368        assert_eq!(CyclePacketKind::Impact.dir_name(), "impact");
369        assert_eq!(CyclePacketKind::ConflictMatrix.dir_name(), "conflict-matrix");
370    }
371
372    #[test]
373    fn build_cache_hit_report_captures_fields() {
374        let report = build_cache_hit_report(
375            CyclePacketKind::Evidence,
376            "key1",
377            "gevd:abc",
378            "disk_hit",
379            &["phase_a", "phase_b"],
380            500,
381        );
382        assert_eq!(report.kind, CyclePacketKind::Evidence);
383        assert_eq!(report.key, "key1");
384        assert_eq!(report.packet_id, "gevd:abc");
385        assert_eq!(report.hit_status, "disk_hit");
386        assert_eq!(report.skipped_phases, vec!["phase_a", "phase_b"]);
387        assert_eq!(report.lookup_micros, 500);
388    }
389
390    #[test]
391    fn cache_stats_returns_zero_for_missing_dir() {
392        let dir = tempfile::tempdir().unwrap();
393        let (count, bytes) = cycle_packet_cache_stats(dir.path());
394        assert_eq!(count, 0);
395        assert_eq!(bytes, 0);
396    }
397
398    #[test]
399    fn cache_stats_counts_entries() {
400        let dir = tempfile::tempdir().unwrap();
401        let root = dir.path();
402        cycle_packet_write_cache(
403            root,
404            CyclePacketKind::Evidence,
405            "key1",
406            &serde_json::json!({"test": 1}),
407        );
408        cycle_packet_write_cache(
409            root,
410            CyclePacketKind::Evidence,
411            "key2",
412            &serde_json::json!({"test": 2}),
413        );
414        cycle_packet_write_cache(
415            root,
416            CyclePacketKind::ContextPack,
417            "key3",
418            &serde_json::json!({"test": 3}),
419        );
420        let (count, bytes) = cycle_packet_cache_stats(root);
421        assert_eq!(count, 3);
422        assert!(bytes > 0);
423    }
424
425    #[test]
426    fn evict_removes_expired_entries() {
427        let dir = tempfile::tempdir().unwrap();
428        let root = dir.path();
429        let old_entry = serde_json::json!({"old": true});
430        cycle_packet_write_cache(root, CyclePacketKind::Evidence, "old-key", &old_entry);
431        let old_path = cycle_packet_cache_path(root, CyclePacketKind::Evidence, "old-key");
432        let old_time = std::time::SystemTime::now() - std::time::Duration::from_secs(7200);
433        let file_time = filetime::FileTime::from_system_time(old_time);
434        filetime::set_file_mtime(&old_path, file_time).unwrap();
435
436        let new_entry = serde_json::json!({"new": true});
437        cycle_packet_write_cache(root, CyclePacketKind::Evidence, "new-key", &new_entry);
438
439        let report = cycle_packet_cache_evict(root, 3600, 1024 * 1024 * 1024);
440        assert_eq!(report.evicted_entries, 1);
441        assert_eq!(report.remaining_entries, 1);
442        assert!(
443            cycle_packet_read_cache::<serde_json::Value>(root, CyclePacketKind::Evidence, "old-key")
444                .is_none(),
445            "old entry should be evicted"
446        );
447        assert!(
448            cycle_packet_read_cache::<serde_json::Value>(root, CyclePacketKind::Evidence, "new-key")
449                .is_some(),
450            "new entry should survive"
451        );
452    }
453
454    #[test]
455    fn evict_enforces_max_bytes() {
456        let dir = tempfile::tempdir().unwrap();
457        let root = dir.path();
458        for i in 0..5 {
459            let data = serde_json::json!({"payload": "x".repeat(200), "idx": i});
460            cycle_packet_write_cache(
461                root,
462                CyclePacketKind::Evidence,
463                &format!("key-{i}"),
464                &data,
465            );
466        }
467        let (count, bytes) = cycle_packet_cache_stats(root);
468        assert_eq!(count, 5);
469        assert!(bytes > 500);
470        let max_bytes = 500u64;
471        let report = cycle_packet_cache_evict(root, 0, max_bytes);
472        assert!(
473            report.evicted_entries > 0,
474            "expected evictions for max_bytes={max_bytes}, got {report:?}"
475        );
476        let (_, remaining_bytes) = cycle_packet_cache_stats(root);
477        assert!(
478            remaining_bytes <= max_bytes + 300,
479            "remaining bytes should be near max_bytes, got {remaining_bytes}"
480        );
481    }
482
483    #[test]
484    fn evict_noop_on_empty_cache() {
485        let dir = tempfile::tempdir().unwrap();
486        let report = cycle_packet_cache_evict(dir.path(), 3600, 1024);
487        assert_eq!(report.scanned_entries, 0);
488        assert_eq!(report.evicted_entries, 0);
489    }
490}
tsift_quality/cycle_packet_cache.rs

tsift_quality/
cycle_packet_cache.rs
