lean_ctx_activity/
store.rs

1// SPDX-License-Identifier: Apache-2.0
2//! The on-disk log: `<root>/<project key>/<session key>.jsonl`, one
3//! [`Event`] per line, appended under the project's `.lock`.
4//!
5//! Bounds: a session log over [`MAX_SESSION_BYTES`] keeps its newest half
6//! (sequence numbers stay, so the gap shows the trim); a project keeps its
7//! [`MAX_SESSIONS_PER_PROJECT`] most recently written sessions.
8
9use std::fs::{File, OpenOptions};
10use std::io::{Read, Seek, SeekFrom, Write};
11use std::path::{Path, PathBuf};
12use std::time::Duration;
13
14use chrono::{DateTime, SecondsFormat, Utc};
15
16use crate::model::{
17    Draft, Event, Kind, MAX_LABEL_BYTES, MAX_ROOT_BYTES, MAX_TARGET_BYTES, MAX_TEXT_BYTES, bound,
18    present,
19};
20
21/// A session log larger than this keeps its newest half.
22pub const MAX_SESSION_BYTES: u64 = 1024 * 1024;
23/// Sessions kept per project (most recently written).
24pub const MAX_SESSIONS_PER_PROJECT: usize = 64;
25/// Events returned when no `limit` is given.
26pub const DEFAULT_LIMIT: usize = 200;
27/// The largest `limit` a listing accepts.
28pub const MAX_LIMIT: usize = 1_000;
29
30/// How long an append waits for the project lock before dropping the event.
31const LOCK_TIMEOUT: Duration = Duration::from_millis(250);
32/// The tail read to find the last sequence number; larger than any line.
33const TAIL_BYTES: u64 = 64 * 1024;
34
35/// What a listing selects. `session` and `project_root` narrow, both may be
36/// given; with neither the listing is empty.
37#[derive(Debug, Clone, Default)]
38pub struct Filter {
39    pub session: Option<String>,
40    pub project_root: Option<String>,
41    /// Strictly newer than this instant.
42    pub since: Option<DateTime<Utc>>,
43    /// Empty = every kind.
44    pub kinds: Vec<Kind>,
45    /// Newest `limit` are kept (`DEFAULT_LIMIT` when 0).
46    pub limit: usize,
47}
48
49/// A listing: the newest `limit` matches, sorted oldest → newest.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct Listing {
52    /// Matches before the limit.
53    pub total: usize,
54    pub truncated: bool,
55    pub events: Vec<Event>,
56}
57
58/// The activity store under one root directory.
59#[derive(Debug, Clone)]
60pub struct Store {
61    root: PathBuf,
62}
63
64fn key(value: &str) -> String {
65    lean_ctx_fs::hasher::hash_short(value)
66}
67
68fn label(value: &str) -> String {
69    bound(value.trim(), MAX_LABEL_BYTES).0
70}
71
72impl Store {
73    pub fn new(root: impl Into<PathBuf>) -> Self {
74        Self { root: root.into() }
75    }
76
77    fn project_dir(&self, project_root: &str) -> PathBuf {
78        self.root.join(key(project_root))
79    }
80
81    /// Append `draft` at `now`. `redact` masks secrets in every text field
82    /// before anything is bounded or written. Returns the stored event.
83    pub fn append(
84        &self,
85        draft: Draft,
86        now: DateTime<Utc>,
87        redact: &dyn Fn(&str) -> String,
88    ) -> std::io::Result<Event> {
89        let session_id = label(&draft.session_id);
90        let project_root = bound(draft.project_root.trim(), MAX_ROOT_BYTES).0;
91        if session_id.is_empty() || project_root.is_empty() {
92            return Err(std::io::Error::new(
93                std::io::ErrorKind::InvalidInput,
94                "an activity event needs a session id and a project root",
95            ));
96        }
97        let mut truncated = false;
98        let mut bounded = |value: Option<String>, max: usize| {
99            present(value).map(|text| {
100                let (text, cut) = bound(&redact(&text), max);
101                truncated |= cut;
102                text
103            })
104        };
105        let target = bounded(draft.target, MAX_TARGET_BYTES);
106        let text = bounded(draft.text, MAX_TEXT_BYTES);
107        let agent = label(&draft.agent);
108
109        let dir = self.project_dir(&project_root);
110        std::fs::create_dir_all(&dir)?;
111        let lock = OpenOptions::new()
112            .create(true)
113            .truncate(false)
114            .write(true)
115            .open(dir.join(".lock"))?;
116        lean_ctx_fs::file_lock::acquire_exclusive_timeout(&lock, LOCK_TIMEOUT)
117            .map_err(|reason| std::io::Error::new(std::io::ErrorKind::TimedOut, reason))?;
118
119        let path = dir.join(format!("{}.jsonl", key(&session_id)));
120        let is_new = !path.exists();
121        let event = Event {
122            seq: last_seq(&path)? + 1,
123            at: now.to_rfc3339_opts(SecondsFormat::Millis, true),
124            session_id,
125            agent: if agent.is_empty() {
126                "agent".to_owned()
127            } else {
128                agent
129            },
130            project_root,
131            kind: draft.kind,
132            source: draft.source,
133            tool: present(draft.tool).map(|tool| label(&tool)),
134            target,
135            outcome: draft.outcome,
136            text,
137            truncated,
138        };
139        let mut line = serde_json::to_string(&event).map_err(std::io::Error::other)?;
140        line.push('\n');
141        let mut file = OpenOptions::new().create(true).append(true).open(&path)?;
142        file.write_all(line.as_bytes())?;
143        let len = file.metadata()?.len();
144        drop(file);
145        if len > MAX_SESSION_BYTES {
146            trim(&path)?;
147        }
148        if is_new {
149            prune_sessions(&dir)?;
150        }
151        Ok(event)
152    }
153
154    /// The events `filter` selects. Read only; unreadable lines are skipped.
155    pub fn list(&self, filter: &Filter) -> std::io::Result<Listing> {
156        let mut files = Vec::new();
157        match (&filter.session, &filter.project_root) {
158            (Some(session), Some(root)) => {
159                files.push(
160                    self.project_dir(root)
161                        .join(format!("{}.jsonl", key(session))),
162                );
163            }
164            (Some(session), None) => {
165                let name = format!("{}.jsonl", key(session));
166                for dir in sorted_entries(&self.root)? {
167                    files.push(dir.join(&name));
168                }
169            }
170            (None, Some(root)) => {
171                files.extend(
172                    sorted_entries(&self.project_dir(root))?
173                        .into_iter()
174                        .filter(|path| path.extension().is_some_and(|ext| ext == "jsonl")),
175                );
176            }
177            (None, None) => {}
178        }
179
180        let mut events = Vec::new();
181        for path in files {
182            let Ok(text) = std::fs::read_to_string(&path) else {
183                continue;
184            };
185            for line in text.lines() {
186                let Ok(event) = serde_json::from_str::<Event>(line) else {
187                    continue;
188                };
189                if matches(&event, filter) {
190                    events.push(event);
191                }
192            }
193        }
194        // Deterministic: time, then session, then sequence.
195        events.sort_by(|a, b| {
196            (a.at.as_str(), a.session_id.as_str(), a.seq).cmp(&(
197                b.at.as_str(),
198                b.session_id.as_str(),
199                b.seq,
200            ))
201        });
202        let limit = match filter.limit {
203            0 => DEFAULT_LIMIT,
204            n => n.min(MAX_LIMIT),
205        };
206        let total = events.len();
207        let truncated = total > limit;
208        if truncated {
209            events.drain(..total - limit);
210        }
211        Ok(Listing {
212            total,
213            truncated,
214            events,
215        })
216    }
217}
218
219fn matches(event: &Event, filter: &Filter) -> bool {
220    filter
221        .session
222        .as_ref()
223        .is_none_or(|session| &event.session_id == session)
224        && filter
225            .project_root
226            .as_ref()
227            .is_none_or(|root| &event.project_root == root)
228        && (filter.kinds.is_empty() || filter.kinds.contains(&event.kind))
229        && filter.since.is_none_or(|since| {
230            DateTime::parse_from_rfc3339(&event.at).is_ok_and(|at| at.with_timezone(&Utc) > since)
231        })
232}
233
234/// Entries of `dir` sorted by name; none when it does not exist.
235fn sorted_entries(dir: &Path) -> std::io::Result<Vec<PathBuf>> {
236    let entries = match std::fs::read_dir(dir) {
237        Ok(entries) => entries,
238        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
239        Err(error) => return Err(error),
240    };
241    let mut out: Vec<PathBuf> = entries.filter_map(|e| e.ok().map(|e| e.path())).collect();
242    out.sort();
243    Ok(out)
244}
245
246/// The sequence number of the last line of `path`, 0 when there is none.
247fn last_seq(path: &Path) -> std::io::Result<u64> {
248    let mut file = match File::open(path) {
249        Ok(file) => file,
250        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0),
251        Err(error) => return Err(error),
252    };
253    let len = file.metadata()?.len();
254    file.seek(SeekFrom::Start(len.saturating_sub(TAIL_BYTES)))?;
255    let mut tail = Vec::new();
256    file.read_to_end(&mut tail)?;
257    let tail = String::from_utf8_lossy(&tail);
258    Ok(tail
259        .lines()
260        .rev()
261        .find_map(|line| serde_json::from_str::<Event>(line).ok())
262        .map_or(0, |event| event.seq))
263}
264
265/// Keep the newest half (by bytes) of a session log, whole lines only.
266fn trim(path: &Path) -> std::io::Result<()> {
267    let text = std::fs::read_to_string(path)?;
268    let cut = text.len() / 2;
269    let start = text[cut..].find('\n').map_or(text.len(), |i| cut + i + 1);
270    lean_ctx_fs::atomic_fs::try_atomic_write(path, &text.as_bytes()[start..], None)
271}
272
273/// Keep the `MAX_SESSIONS_PER_PROJECT` most recently written session logs.
274fn prune_sessions(dir: &Path) -> std::io::Result<()> {
275    let mut sessions: Vec<(std::time::SystemTime, PathBuf)> = sorted_entries(dir)?
276        .into_iter()
277        .filter(|path| path.extension().is_some_and(|ext| ext == "jsonl"))
278        .filter_map(|path| {
279            let modified = path.metadata().and_then(|m| m.modified()).ok()?;
280            Some((modified, path))
281        })
282        .collect();
283    if sessions.len() <= MAX_SESSIONS_PER_PROJECT {
284        return Ok(());
285    }
286    sessions.sort();
287    let excess = sessions.len() - MAX_SESSIONS_PER_PROJECT;
288    for (_, path) in sessions.into_iter().take(excess) {
289        let _ = std::fs::remove_file(path);
290    }
291    Ok(())
292}
293
294#[cfg(test)]
295mod tests {
296    use super::*;
297    use crate::model::{Outcome, Source};
298    use chrono::TimeZone;
299
300    fn at(second: u32) -> DateTime<Utc> {
301        Utc.with_ymd_and_hms(2026, 11, 4, 9, 0, second)
302            .single()
303            .unwrap_or_default()
304    }
305
306    fn mask(text: &str) -> String {
307        text.replace("sk-secret", "[REDACTED]")
308    }
309
310    fn draft(session: &str, kind: Kind) -> Draft {
311        Draft::new(session, "claude", "/code/app", kind, Source::Hook)
312    }
313
314    #[test]
315    fn append_numbers_redacts_and_bounds_every_event() {
316        let dir = tempfile::tempdir().expect("tempdir");
317        let store = Store::new(dir.path());
318        let first = store
319            .append(
320                draft("s1", Kind::Shell)
321                    .tool("Bash")
322                    .target("curl -H 'Authorization: sk-secret' x")
323                    .outcome(Outcome::Ok),
324                at(1),
325                &mask,
326            )
327            .expect("append");
328        assert_eq!(first.seq, 1);
329        assert_eq!(first.at, "2026-11-04T09:00:01.000Z");
330        assert_eq!(
331            first.target.as_deref(),
332            Some("curl -H 'Authorization: [REDACTED]' x")
333        );
334        assert!(!first.truncated);
335
336        let long = "é".repeat(MAX_TEXT_BYTES);
337        let second = store
338            .append(draft("s1", Kind::AgentMessage).text(long), at(2), &mask)
339            .expect("append");
340        assert_eq!(second.seq, 2);
341        assert!(second.truncated);
342        let text = second.text.expect("text");
343        assert!(text.len() <= MAX_TEXT_BYTES + '…'.len_utf8());
344        assert!(text.ends_with('…'));
345
346        let empty = store
347            .append(draft("s1", Kind::TurnEnd).text("  "), at(3), &mask)
348            .expect("append");
349        assert_eq!(empty.text, None);
350
351        let missing = store.append(draft(" ", Kind::TurnEnd), at(4), &mask);
352        assert!(missing.is_err());
353    }
354
355    #[test]
356    fn list_selects_by_session_project_kind_and_since_deterministically() {
357        let dir = tempfile::tempdir().expect("tempdir");
358        let store = Store::new(dir.path());
359        for (session, kind, second) in [
360            ("s1", Kind::UserMessage, 1),
361            ("s2", Kind::FileRead, 1),
362            ("s1", Kind::FileWrite, 2),
363            ("s2", Kind::TurnEnd, 3),
364        ] {
365            store
366                .append(draft(session, kind), at(second), &mask)
367                .expect("append");
368        }
369        let other = Draft::new("s3", "codex", "/code/other", Kind::Shell, Source::Mcp);
370        store.append(other, at(1), &mask).expect("append");
371
372        let project = Filter {
373            project_root: Some("/code/app".into()),
374            ..Filter::default()
375        };
376        let listing = store.list(&project).expect("list");
377        let order: Vec<(&str, u64)> = listing
378            .events
379            .iter()
380            .map(|e| (e.session_id.as_str(), e.seq))
381            .collect();
382        assert_eq!(order, [("s1", 1), ("s2", 1), ("s1", 2), ("s2", 2)]);
383        assert_eq!(store.list(&project).expect("list"), listing);
384
385        let session = Filter {
386            session: Some("s3".into()),
387            ..Filter::default()
388        };
389        let listing = store.list(&session).expect("list");
390        assert_eq!(listing.events.len(), 1);
391        assert_eq!(listing.events[0].project_root, "/code/other");
392
393        let narrowed = Filter {
394            since: Some(at(1)),
395            kinds: vec![Kind::TurnEnd, Kind::FileRead],
396            limit: 1,
397            ..project
398        };
399        let listing = store.list(&narrowed).expect("list");
400        assert_eq!(listing.total, 1);
401        assert!(!listing.truncated);
402        assert_eq!(listing.events[0].kind, Kind::TurnEnd);
403
404        let limited = Filter {
405            project_root: Some("/code/app".into()),
406            limit: 2,
407            ..Filter::default()
408        };
409        let listing = store.list(&limited).expect("list");
410        assert_eq!((listing.total, listing.truncated), (4, true));
411        assert_eq!(listing.events[1].kind, Kind::TurnEnd);
412
413        assert!(
414            store
415                .list(&Filter::default())
416                .expect("list")
417                .events
418                .is_empty()
419        );
420    }
421
422    #[test]
423    fn an_oversized_session_keeps_its_newest_half_and_its_numbering() {
424        let dir = tempfile::tempdir().expect("tempdir");
425        let store = Store::new(dir.path());
426        let text = "x".repeat(MAX_TEXT_BYTES);
427        let lines = (MAX_SESSION_BYTES / MAX_TEXT_BYTES as u64) + 10;
428        for _ in 0..lines {
429            store
430                .append(
431                    draft("s1", Kind::UserMessage).text(text.clone()),
432                    at(1),
433                    &mask,
434                )
435                .expect("append");
436        }
437        let path = store
438            .project_dir("/code/app")
439            .join(format!("{}.jsonl", key("s1")));
440        assert!(std::fs::metadata(&path).expect("log").len() <= MAX_SESSION_BYTES);
441        let listing = store
442            .list(&Filter {
443                session: Some("s1".into()),
444                limit: MAX_LIMIT,
445                ..Filter::default()
446            })
447            .expect("list");
448        let last = listing.events.last().expect("events");
449        assert_eq!(last.seq, lines);
450        assert!(listing.events[0].seq > 1);
451    }
452}
lean_ctx_activity/store.rs

lean_ctx_activity/
store.rs
