1. 1//! Per-ledger recall cache (docs/design/23-memory.md M1): keyed by the
2. 2//! canonical ledger path and invalidated when mtime or length differs, so
3. 3//! warm queries skip the JSONL rescan while appends stay visible on the
4. 4//! very next search. Purely an accelerator — cold behavior is identical to
5. 5//! a fresh scan.
6. 6
7. 7use std::collections::{HashMap, VecDeque};
8. 8use std::io::BufRead;
9. 9use std::path::{Path, PathBuf};
10. 10use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
11. 11use std::time::{Instant, SystemTime};
12. 12
13. 13use chrono::{DateTime, Utc};
14. 14use vak_llm::Role;
15. 15
16. 16use crate::SearchError;
17. 17use crate::search::normalize_impl;
18. 18use crate::types::{Entry, EntryPayload};
19. 19
20. 20/// Bounded work: at most the trailing N message lines of a ledger are
21. 21/// scanned. Long sessions still recall their recent past.
22. 22pub(crate) const MAX_SCAN_LINES: usize = 4000;
23. 23const MAX_CACHED_LEDGERS: usize = 128;
24. 24
25. 25pub(crate) struct CachedMessage {
26. 26 pub entry_id: String,
27. 27 pub ts: DateTime<Utc>,
28. 28 pub role: &'static str,
29. 29 pub text: String,
30. 30 pub normalized: String,
31. 31}
32. 32
33. 33#[derive(Clone, Copy, PartialEq, Eq)]
34. 34struct Fingerprint {
35. 35 mtime: SystemTime,
36. 36 len: u64,
37. 37}
38. 38
39. 39struct CacheSlot {
40. 40 fingerprint: Fingerprint,
41. 41 last_used: Instant,
42. 42 ledger: Arc<Vec<CachedMessage>>,
43. 43}
44. 44
45. 45fn cache() -> &'static Mutex<HashMap<PathBuf, CacheSlot>> {
46. 46 static CACHE: OnceLock<Mutex<HashMap<PathBuf, CacheSlot>>> = OnceLock::new();
47. 47 CACHE.get_or_init(|| Mutex::new(HashMap::new()))
48. 48}
49. 49
50. 50fn lock(map: &Mutex<HashMap<PathBuf, CacheSlot>>) -> MutexGuard<'_, HashMap<PathBuf, CacheSlot>> {
51. 51 map.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
52. 52}
53. 53
54. 54fn evict_to_capacity(map: &mut HashMap<PathBuf, CacheSlot>) {
55. 55 while map.len() >= MAX_CACHED_LEDGERS {
56. 56 let oldest = map
57. 57 .iter()
58. 58 .min_by_key(|(_, slot)| slot.last_used)
59. 59 .map(|(k, _)| k.clone());
60. 60 match oldest {
61. 61 Some(k) => {
62. 62 map.remove(&k);
63. 63 }
64. 64 None => break,
65. 65 }
66. 66 }
67. 67}
68. 68
69. 69/// Tokenized representation of `path`'s trailing window, served from cache
70. 70/// when the file's mtime and length are unchanged since the last scan.
71. 71pub(crate) fn ledger(path: &Path) -> Result<Arc<Vec<CachedMessage>>, SearchError> {
72. 72 let key = path.canonicalize()?;
73. 73 let meta = std::fs::metadata(path)?;
74. 74 let fingerprint = Fingerprint {
75. 75 mtime: meta.modified()?,
76. 76 len: meta.len(),
77. 77 };
78. 78 {
79. 79 let mut map = lock(cache());
80. 80 if let Some(slot) = map.get_mut(&key)
81. 81 && slot.fingerprint == fingerprint
82. 82 {
83. 83 slot.last_used = Instant::now();
84. 84 return Ok(Arc::clone(&slot.ledger));
85. 85 }
86. 86 }
87. 87 let scanned = Arc::new(scan(path)?);
88. 88 let mut map = lock(cache());
89. 89 // A concurrent query may have cached a snapshot of this exact state
90. 90 // while we scanned; prefer it over re-inserting an equal copy.
91. 91 if let Some(slot) = map.get(&key)
92. 92 && slot.fingerprint == fingerprint
93. 93 {
94. 94 return Ok(Arc::clone(&slot.ledger));
95. 95 }
96. 96 evict_to_capacity(&mut map);
97. 97 map.insert(
98. 98 key,
99. 99 CacheSlot {
100. 100 fingerprint,
101. 101 last_used: Instant::now(),
102. 102 ledger: Arc::clone(&scanned),
103. 103 },
104. 104 );
105. 105 Ok(scanned)
106. 106}
107. 107
108. 108fn scan(path: &Path) -> Result<Vec<CachedMessage>, SearchError> {
109. 109 let file = std::fs::File::open(path)?;
110. 110 let reader = std::io::BufReader::new(file);
111. 111 // Ring buffer keeps memory bounded on huge ledgers while preserving
112. 112 // "trailing lines" semantics.
113. 113 let mut ring: VecDeque<String> = VecDeque::with_capacity(MAX_SCAN_LINES);
114. 114 for line in reader.lines() {
115. 115 let line = line?;
116. 116 if line.trim().is_empty() {
117. 117 continue;
118. 118 }
119. 119 if ring.len() == MAX_SCAN_LINES {
120. 120 ring.pop_front();
121. 121 }
122. 122 ring.push_back(line);
123. 123 }
124. 124 let mut messages = Vec::new();
125. 125 for line in ring {
126. 126 let Ok(entry) = serde_json::from_str::<Entry>(&line) else {
127. 127 continue;
128. 128 };
129. 129 let EntryPayload::Message(record) = entry.payload else {
130. 130 continue;
131. 131 };
132. 132 // A runtime nudge is not something the user said or the model
133. 133 // answered; leaving it out keeps it from surfacing in session search.
134. 134 if record.control_kind().is_some() {
135. 135 continue;
136. 136 }
137. 137 let role = match record.message.role {
138. 138 Role::User => "user",
139. 139 Role::Assistant => "assistant",
140. 140 };
141. 141 let text = record.message.text_content();
142. 142 if text.trim().is_empty() {
143. 143 continue;
144. 144 }
145. 145 messages.push(CachedMessage {
146. 146 normalized: normalize_impl(&text),
147. 147 text,
148. 148 role,
149. 149 entry_id: entry.id,
150. 150 ts: entry.ts,
151. 151 });
152. 152 }
153. 153 Ok(messages)
154. 154}
155. 155