aether_sessions/store/
mod.rs

1mod prompt_history;
2
3use acp_utils::notifications::{PromptSearchResponse, SessionPreviewResponse, SessionPreviewRole, SessionPreviewTurn};
4use aether_core::events::{AgentEvent, MessageEvent, ToolEvent};
5use serde::Serialize;
6use std::fs::{self, File, OpenOptions};
7use std::io::Write;
8use std::path::{Path, PathBuf};
9use std::time::UNIX_EPOCH;
10use tracing::warn;
11use utils::settings::aether_home;
12
13use crate::error::SessionStoreError;
14use crate::{SessionEvent, SessionLog, SessionLogEntry, SessionMeta, UserEvent};
15use llm::ContentBlock;
16use prompt_history::PromptHistoryIndex;
17
18const PROMPT_HISTORY_FILE: &str = "prompt-history.jsonl";
19const PREVIEW_TRANSCRIPT_TURNS: usize = 8;
20const MAX_TITLE_LEN: usize = 80;
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
23pub struct FileFingerprint {
24    pub file_size: i64,
25    pub file_mtime_ns: i64,
26}
27
28#[derive(Debug, Clone, Serialize)]
29pub struct DiscoveredSessionFile {
30    pub path: PathBuf,
31    pub fingerprint: FileFingerprint,
32}
33
34impl FileFingerprint {
35    pub fn read(path: impl AsRef<Path>) -> std::io::Result<Self> {
36        let metadata = fs::metadata(path)?;
37        let modified = metadata.modified()?.duration_since(UNIX_EPOCH).unwrap_or_default();
38        Ok(Self {
39            file_size: metadata.len().try_into().unwrap_or(i64::MAX),
40            file_mtime_ns: modified.as_nanos().try_into().unwrap_or(i64::MAX),
41        })
42    }
43}
44
45#[derive(Debug, Clone, PartialEq)]
46pub struct SessionSummary {
47    pub meta: SessionMeta,
48    pub title: Option<String>,
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
52pub struct ScanLimits {
53    pub max_lines: usize,
54    pub max_bytes: usize,
55}
56
57impl ScanLimits {
58    pub const SUMMARY: Self = Self { max_lines: 64, max_bytes: 64 * 1024 };
59    pub const PREVIEW: Self = Self { max_lines: 200, max_bytes: 128 * 1024 };
60    pub const UNBOUNDED: Self = Self { max_lines: usize::MAX, max_bytes: usize::MAX };
61}
62
63pub fn discover_session_files(sessions_dir: impl AsRef<Path>) -> std::io::Result<Vec<DiscoveredSessionFile>> {
64    let mut files = Vec::new();
65    for entry in fs::read_dir(sessions_dir)? {
66        let Ok(entry) = entry else { continue };
67        let path = entry.path();
68        if !path.is_file() || path.file_name().and_then(|name| name.to_str()) == Some(PROMPT_HISTORY_FILE) {
69            continue;
70        }
71        if path.extension().and_then(|extension| extension.to_str()) == Some("jsonl") {
72            let Ok(path) = path.canonicalize() else { continue };
73            let Ok(fingerprint) = FileFingerprint::read(&path) else { continue };
74            files.push(DiscoveredSessionFile { path, fingerprint });
75        }
76    }
77    files.sort_by(|left, right| left.path.cmp(&right.path));
78    Ok(files)
79}
80
81pub struct SessionStore {
82    dir: PathBuf,
83    prompt_history: PromptHistoryIndex,
84}
85
86impl SessionStore {
87    pub fn new() -> Result<Self, SessionStoreError> {
88        let home = aether_home().ok_or(SessionStoreError::MissingAetherHome)?;
89        Ok(Self::from_path(home.join("sessions")))
90    }
91
92    pub fn from_path(dir: PathBuf) -> Self {
93        let prompt_history = PromptHistoryIndex::new(dir.join(PROMPT_HISTORY_FILE));
94        Self { dir, prompt_history }
95    }
96
97    pub fn append_meta(&self, session_id: &str, meta: &SessionMeta) -> Result<(), SessionStoreError> {
98        self.append_line(session_id, meta)
99    }
100
101    pub fn append_event(&self, session_id: &str, event: &SessionEvent) -> Result<(), SessionStoreError> {
102        if !event.is_persisted() {
103            return Ok(());
104        }
105        self.append_line(session_id, event)?;
106        if let Some(prompt) = event.user_content()
107            && let Some(meta) = self.session_meta(session_id)
108        {
109            let _ = self.prompt_history.append_prompt(&meta, prompt);
110        }
111
112        Ok(())
113    }
114
115    pub fn load(&self, session_id: &str) -> Result<(SessionMeta, Vec<SessionEvent>), SessionStoreError> {
116        let scan = read_bounded_session(&self.session_path(session_id), ScanLimits::UNBOUNDED)?;
117        Ok((scan.meta, scan.events))
118    }
119
120    pub fn session_cwd(&self, session_id: &str) -> Option<PathBuf> {
121        self.session_meta(session_id).map(|meta| meta.cwd)
122    }
123
124    /// Rewrites the session's recorded working directory to `new_cwd`, updating the
125    /// metadata line and any matching prompt-history entries. Event lines are left
126    /// unchanged.
127    pub fn relocate(&self, session_id: &str, new_cwd: &Path) -> Result<(), SessionStoreError> {
128        let path = self.session_path(session_id);
129        let content = fs::read_to_string(&path)?;
130        let mut lines = content.lines();
131        let first = lines.next().ok_or(SessionStoreError::MissingMetadata)?;
132
133        let mut meta: SessionMeta = serde_json::from_str(first.trim())
134            .map_err(|source| SessionStoreError::InvalidMetadata { line_number: 1, source })?;
135        meta.cwd = new_cwd.to_path_buf();
136        let meta_line = serde_json::to_string(&meta)?;
137
138        let tmp_path = path.with_extension("jsonl.tmp");
139        {
140            let mut file = File::create(&tmp_path)?;
141            writeln!(file, "{meta_line}")?;
142            for line in lines {
143                writeln!(file, "{line}")?;
144            }
145        }
146        fs::rename(tmp_path, &path)?;
147
148        let _ = self.prompt_history.relocate_session(session_id, new_cwd);
149        Ok(())
150    }
151
152    pub fn rebuild_prompt_history(&self) -> Result<(), SessionStoreError> {
153        let files = discover_session_files(&self.dir)?;
154        self.prompt_history
155            .rebuild(files.into_iter().filter_map(|file| read_all_prompts(&file.path).ok()).flatten())?;
156        Ok(())
157    }
158
159    pub fn list(&self) -> Vec<SessionSummary> {
160        let Ok(files) = discover_session_files(&self.dir) else {
161            return Vec::new();
162        };
163
164        let mut summaries: Vec<SessionSummary> =
165            files.into_iter().filter_map(|file| read_session_summary(&file.path).ok()).collect();
166
167        summaries.sort_by(|a, b| {
168            b.meta.created_at.cmp(&a.meta.created_at).then_with(|| b.meta.session_id.cmp(&a.meta.session_id))
169        });
170        summaries
171    }
172
173    pub fn preview(&self, session_id: &str) -> Result<SessionPreviewResponse, SessionStoreError> {
174        read_session_preview(&self.session_path(session_id), ScanLimits::PREVIEW)
175    }
176
177    pub fn search_prompts(&self, query: &str, limit: Option<usize>) -> Result<PromptSearchResponse, SessionStoreError> {
178        self.prompt_history.search(query, limit).map_err(SessionStoreError::PromptHistory)
179    }
180
181    fn session_meta(&self, session_id: &str) -> Option<SessionMeta> {
182        SessionLog::open(self.session_path(session_id)).ok().map(|log| log.meta)
183    }
184
185    fn append_line<T: Serialize>(&self, session_id: &str, value: &T) -> Result<(), SessionStoreError> {
186        fs::create_dir_all(&self.dir)?;
187        let path = self.session_path(session_id);
188        let mut file = OpenOptions::new().create(true).append(true).open(&path)?;
189        let line = serde_json::to_string(value)?;
190        writeln!(file, "{line}")?;
191        Ok(())
192    }
193
194    fn session_path(&self, session_id: &str) -> PathBuf {
195        self.dir.join(format!("{session_id}.jsonl"))
196    }
197}
198
199struct SessionLogScan {
200    meta: SessionMeta,
201    events: Vec<SessionEvent>,
202    truncated: bool,
203}
204
205fn read_session_summary(path: &Path) -> Result<SessionSummary, SessionStoreError> {
206    let scan = read_bounded_session(path, ScanLimits::SUMMARY)?;
207    let title = scan.events.iter().find_map(|event| match event {
208        SessionEvent::User(UserEvent::Message { content }) => Some(extract_title(content)),
209        _ => None,
210    });
211    Ok(SessionSummary { meta: scan.meta, title })
212}
213
214fn read_session_preview(path: &Path, limits: ScanLimits) -> Result<SessionPreviewResponse, SessionStoreError> {
215    let scan = read_bounded_session(path, limits)?;
216    let meta = scan.meta;
217    let mut truncated = scan.truncated;
218    let mut transcript = Vec::new();
219    let mut tool_call_count = 0;
220
221    for event in scan.events {
222        match event {
223            SessionEvent::User(UserEvent::Message { content }) => {
224                let text = ContentBlock::join_text(&content);
225                let text = if text.is_empty() { "[media prompt]".to_string() } else { text };
226                if !push_preview_turn(&mut transcript, SessionPreviewRole::User, &text) {
227                    truncated = true;
228                }
229            }
230            SessionEvent::Agent(AgentEvent::Message(MessageEvent::Text { chunk, .. }))
231                if !push_preview_turn(&mut transcript, SessionPreviewRole::Assistant, &chunk) =>
232            {
233                truncated = true;
234            }
235            SessionEvent::Agent(AgentEvent::Tool(ToolEvent::Call { .. })) => {
236                tool_call_count += 1;
237            }
238            _ => {}
239        }
240    }
241
242    Ok(SessionPreviewResponse {
243        session_id: meta.session_id,
244        cwd: meta.cwd,
245        created_at: meta.created_at,
246        model: meta.model,
247        selected_mode: meta.selected_mode,
248        transcript,
249        tool_call_count,
250        truncated,
251    })
252}
253
254fn read_bounded_session(path: &Path, limits: ScanLimits) -> Result<SessionLogScan, SessionStoreError> {
255    let mut log = SessionLog::open(path)?;
256    let meta = log.meta.clone();
257    let mut events = Vec::new();
258    let mut truncated = false;
259    let mut lines_since_meta = 0_usize;
260
261    while let Some(line) = log.next_line()? {
262        lines_since_meta = lines_since_meta.saturating_add(1);
263        let bytes_since_meta = log.bytes_read().saturating_sub(log.meta_line_bytes());
264        if lines_since_meta > limits.max_lines || bytes_since_meta > limits.max_bytes {
265            truncated = true;
266            break;
267        }
268        let entry = SessionLogEntry::parse(line);
269        match entry {
270            SessionLogEntry::Persisted { event, .. } => events.push(*event),
271            SessionLogEntry::Transient { .. } => {}
272            SessionLogEntry::Malformed { error, .. } => warn!("Skipping malformed session log line: {error}"),
273        }
274    }
275
276    Ok(SessionLogScan { meta, events, truncated })
277}
278
279fn read_all_prompts(path: &Path) -> Result<Vec<(SessionMeta, String)>, SessionStoreError> {
280    let (meta, events) = read_bounded_session(path, ScanLimits::UNBOUNDED).map(|scan| (scan.meta, scan.events))?;
281    Ok(events.into_iter().filter_map(|event| event.user_content().map(|prompt| (meta.clone(), prompt))).collect())
282}
283
284fn push_preview_turn(transcript: &mut Vec<SessionPreviewTurn>, role: SessionPreviewRole, text: &str) -> bool {
285    let text = text.lines().next().unwrap_or(text).trim();
286    if text.is_empty() {
287        return true;
288    }
289    if transcript.len() >= PREVIEW_TRANSCRIPT_TURNS {
290        return false;
291    }
292    transcript.push(SessionPreviewTurn { role, text: truncate_for_preview(text) });
293    true
294}
295
296fn truncate_for_preview(text: &str) -> String {
297    if text.len() <= MAX_TITLE_LEN {
298        text.to_string()
299    } else {
300        let end = text.floor_char_boundary(MAX_TITLE_LEN);
301        format!("{}…", &text[..end])
302    }
303}
304
305fn extract_title(content: &[ContentBlock]) -> String {
306    let first_line =
307        ContentBlock::first_text(content).and_then(|text| text.lines().next()).unwrap_or("Media prompt").trim();
308    truncate_for_preview(first_line)
309}
aether_sessions/store/mod.rs

aether_sessions/store/
mod.rs
