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, display_content, .. }) => {
209            Some(extract_title(display_content.as_deref().unwrap_or(content)))
210        }
211        _ => None,
212    });
213    Ok(SessionSummary { meta: scan.meta, title })
214}
215
216fn read_session_preview(path: &Path, limits: ScanLimits) -> Result<SessionPreviewResponse, SessionStoreError> {
217    let scan = read_bounded_session(path, limits)?;
218    let meta = scan.meta;
219    let mut truncated = scan.truncated;
220    let mut transcript = Vec::new();
221    let mut tool_call_count = 0;
222
223    for event in scan.events {
224        match event {
225            SessionEvent::User(UserEvent::Message { content, display_content, .. }) => {
226                let text = ContentBlock::join_text(display_content.as_deref().unwrap_or(&content));
227                let text = if text.is_empty() { "[media prompt]".to_string() } else { text };
228                if !push_preview_turn(&mut transcript, SessionPreviewRole::User, &text) {
229                    truncated = true;
230                }
231            }
232            SessionEvent::Agent(AgentEvent::Message(MessageEvent::Text { chunk, .. }))
233                if !push_preview_turn(&mut transcript, SessionPreviewRole::Assistant, &chunk) =>
234            {
235                truncated = true;
236            }
237            SessionEvent::Agent(AgentEvent::Tool(ToolEvent::Call { .. })) => {
238                tool_call_count += 1;
239            }
240            _ => {}
241        }
242    }
243
244    Ok(SessionPreviewResponse {
245        session_id: meta.session_id,
246        cwd: meta.cwd,
247        created_at: meta.created_at,
248        model: meta.model,
249        selected_mode: meta.selected_mode,
250        transcript,
251        tool_call_count,
252        truncated,
253    })
254}
255
256fn read_bounded_session(path: &Path, limits: ScanLimits) -> Result<SessionLogScan, SessionStoreError> {
257    let mut log = SessionLog::open(path)?;
258    let meta = log.meta.clone();
259    let mut events = Vec::new();
260    let mut truncated = false;
261    let mut lines_since_meta = 0_usize;
262
263    while let Some(line) = log.next_line()? {
264        lines_since_meta = lines_since_meta.saturating_add(1);
265        let bytes_since_meta = log.bytes_read().saturating_sub(log.meta_line_bytes());
266        if lines_since_meta > limits.max_lines || bytes_since_meta > limits.max_bytes {
267            truncated = true;
268            break;
269        }
270        let entry = SessionLogEntry::parse(line);
271        match entry {
272            SessionLogEntry::Persisted { event, .. } => events.push(*event),
273            SessionLogEntry::Transient { .. } => {}
274            SessionLogEntry::Malformed { error, .. } => warn!("Skipping malformed session log line: {error}"),
275        }
276    }
277
278    Ok(SessionLogScan { meta, events, truncated })
279}
280
281fn read_all_prompts(path: &Path) -> Result<Vec<(SessionMeta, String)>, SessionStoreError> {
282    let (meta, events) = read_bounded_session(path, ScanLimits::UNBOUNDED).map(|scan| (scan.meta, scan.events))?;
283    Ok(events.into_iter().filter_map(|event| event.user_content().map(|prompt| (meta.clone(), prompt))).collect())
284}
285
286fn push_preview_turn(transcript: &mut Vec<SessionPreviewTurn>, role: SessionPreviewRole, text: &str) -> bool {
287    let text = text.lines().next().unwrap_or(text).trim();
288    if text.is_empty() {
289        return true;
290    }
291    if transcript.len() >= PREVIEW_TRANSCRIPT_TURNS {
292        return false;
293    }
294    transcript.push(SessionPreviewTurn { role, text: truncate_for_preview(text) });
295    true
296}
297
298fn truncate_for_preview(text: &str) -> String {
299    if text.len() <= MAX_TITLE_LEN {
300        text.to_string()
301    } else {
302        let end = text.floor_char_boundary(MAX_TITLE_LEN);
303        format!("{}…", &text[..end])
304    }
305}
306
307fn extract_title(content: &[ContentBlock]) -> String {
308    let first_line =
309        ContentBlock::first_text(content).and_then(|text| text.lines().next()).unwrap_or("Media prompt").trim();
310    truncate_for_preview(first_line)
311}
aether_sessions/store/mod.rs

aether_sessions/store/
mod.rs
