vtcode_session_store/
event_log.rs

1//! Append-only per-session `ThreadEvent` log plus index and manifest.
2
3use std::fs::{File, OpenOptions};
4use std::io::{Read, Seek, SeekFrom, Write};
5use std::path::{Path, PathBuf};
6use std::sync::Mutex;
7
8use chrono::Utc;
9use serde::{Deserialize, Serialize};
10use vtcode_exec_events::{ThreadEvent, VersionedThreadEvent};
11
12use crate::error::SessionStoreError;
13use crate::session_dir;
14
15use memchr::memchr;
16
17/// In-memory state protected by a mutex (cheap; appends are infrequent relative
18/// to model inference).
19struct LogState {
20    manifest: SessionManifest,
21    index: TurnIndex,
22    /// Whether we are currently inside a turn (between TurnStarted and
23    /// TurnCompleted/TurnFailed). Used to update the last index entry's
24    /// offsets as intermediate events arrive.
25    in_turn: bool,
26}
27
28impl LogState {
29    fn new(session_id: &str) -> Self {
30        Self {
31            manifest: SessionManifest::new(session_id),
32            index: TurnIndex::default(),
33            in_turn: false,
34        }
35    }
36}
37
38/// Canonical append-only event log for a single session.
39///
40/// All session history is reconstructable from this log. Live conversation
41/// state is never read back into context from here; the log is only consumed
42/// for revert, compaction, analytics, and long-term-learning queries.
43pub struct SessionEventLog {
44    session_dir: PathBuf,
45    events_path: PathBuf,
46    file: Mutex<File>,
47    state: Mutex<LogState>,
48}
49
50impl SessionEventLog {
51    /// Open the log for `session_id`, creating the session directory tree and
52    /// rebuilding the index from `events.jsonl` if it already exists.
53    pub fn open(workspace: &Path, session_id: &str) -> Result<Self, SessionStoreError> {
54        let dir = session_dir(workspace, session_id);
55        std::fs::create_dir_all(dir.join(crate::DERIVED_DIR)).map_err(|e| {
56            SessionStoreError::CreateDir {
57                path: dir.clone(),
58                source: e,
59            }
60        })?;
61        std::fs::create_dir_all(dir.join("index")).map_err(|e| SessionStoreError::CreateDir {
62            path: dir.clone(),
63            source: e,
64        })?;
65        let events_path = dir.join("events.jsonl");
66        let file = OpenOptions::new()
67            .create(true)
68            .read(true)
69            .append(true)
70            .open(&events_path)
71            .map_err(|e| SessionStoreError::io(events_path.clone(), e))?;
72        let log = Self {
73            session_dir: dir,
74            events_path,
75            file: Mutex::new(file),
76            state: Mutex::new(LogState::new(session_id)),
77        };
78        log.scan()?;
79        Ok(log)
80    }
81
82    /// Append an event to the log and update the in-memory index/manifest.
83    pub fn append(&self, event: &ThreadEvent) -> Result<(), SessionStoreError> {
84        let line = serde_json::to_string(&VersionedThreadEvent::new(event.clone()))?;
85        let start = {
86            let mut file = self.file.lock().map_err(poison)?;
87            let start = file
88                .metadata()
89                .map_err(|e| SessionStoreError::io(&self.events_path, e))?
90                .len();
91            writeln!(file, "{line}").map_err(|e| SessionStoreError::io(&self.events_path, e))?;
92            let end = file
93                .metadata()
94                .map_err(|e| SessionStoreError::io(&self.events_path, e))?
95                .len();
96            (start, end)
97        };
98        let (start, end) = start;
99
100        let mut st = self.state.lock().map_err(poison)?;
101        st.manifest.event_count += 1;
102        st.manifest.updated_at = now_rfc3339();
103        match event {
104            ThreadEvent::TurnStarted(_) => {
105                st.in_turn = true;
106                let n = st.manifest.turn_count + 1;
107                st.index.entries.push(TurnIndexEntry {
108                    turn_number: n,
109                    start_offset: start,
110                    end_offset: end,
111                    event_count: 1,
112                    ts: now_rfc3339(),
113                });
114            }
115            ThreadEvent::TurnCompleted(_) => {
116                if st.in_turn {
117                    if let Some(entry) = st.index.entries.last_mut() {
118                        entry.end_offset = end;
119                        entry.event_count += 1;
120                    }
121                    st.in_turn = false;
122                    st.manifest.turn_count = st.index.entries.len() as u64;
123                }
124                st.manifest.status = "completed".to_string();
125                self.persist_meta_locked(&st)?;
126            }
127            ThreadEvent::TurnFailed(_) => {
128                if st.in_turn {
129                    if let Some(entry) = st.index.entries.last_mut() {
130                        entry.end_offset = end;
131                        entry.event_count += 1;
132                    }
133                    st.in_turn = false;
134                    st.manifest.turn_count = st.index.entries.len() as u64;
135                }
136                st.manifest.status = "failed".to_string();
137                self.persist_meta_locked(&st)?;
138            }
139            _ => {
140                if st.in_turn {
141                    if let Some(entry) = st.index.entries.last_mut() {
142                        entry.end_offset = end;
143                        entry.event_count += 1;
144                    }
145                }
146            }
147        }
148        Ok(())
149    }
150
151    /// Reconstruct every event belonging to `turn`.
152    pub fn reconstruct_turn(&self, turn: u64) -> Result<Vec<ThreadEvent>, SessionStoreError> {
153        let entry = {
154            let st = self.state.lock().map_err(poison)?;
155            st.index
156                .entries
157                .iter()
158                .find(|e| e.turn_number == turn)
159                .cloned()
160                .ok_or(SessionStoreError::TurnNotFound {
161                    session: st.manifest.session_id.clone(),
162                    turn,
163                })?
164        };
165        let buf = {
166            let mut file = self.file.lock().map_err(poison)?;
167            file.seek(SeekFrom::Start(entry.start_offset))
168                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
169            let len = (entry.end_offset - entry.start_offset) as usize;
170            let mut buf = vec![0u8; len];
171            file.read_exact(&mut buf)
172                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
173            buf
174        };
175        let text = String::from_utf8_lossy(&buf);
176        let mut events = Vec::new();
177        for line in text.lines() {
178            let line = line.trim();
179            if line.is_empty() {
180                continue;
181            }
182            let v: VersionedThreadEvent =
183                serde_json::from_str(line).map_err(SessionStoreError::Json)?;
184            events.push(v.into_event());
185        }
186        Ok(events)
187    }
188
189    /// Number of turns recorded.
190    #[must_use]
191    pub fn turn_count(&self) -> u64 {
192        self.state
193            .lock()
194            .map_err(poison)
195            .map_or(0, |s| s.manifest.turn_count)
196    }
197
198    /// Number of events recorded.
199    #[must_use]
200    pub fn event_count(&self) -> u64 {
201        self.state
202            .lock()
203            .map_err(poison)
204            .map_or(0, |s| s.manifest.event_count)
205    }
206
207    /// Snapshot of the session manifest.
208    #[must_use]
209    pub fn manifest(&self) -> SessionManifest {
210        self.state
211            .lock()
212            .map_err(poison)
213            .map(|s| s.manifest.clone())
214            .unwrap_or_else(|_| SessionManifest::new(""))
215    }
216
217    /// Snapshot of the turn index.
218    #[must_use]
219    pub fn turn_index(&self) -> TurnIndex {
220        self.state
221            .lock()
222            .map_err(poison)
223            .map(|s| s.index.clone())
224            .unwrap_or_default()
225    }
226
227    /// Mark the session completed and flush metadata.
228    pub fn complete(&self) -> Result<(), SessionStoreError> {
229        let mut st = self.state.lock().map_err(poison)?;
230        st.manifest.status = "completed".to_string();
231        st.manifest.updated_at = now_rfc3339();
232        self.persist_meta_locked(&st)
233    }
234
235    /// Rebuild index + manifest by scanning `events.jsonl` (authoritative).
236    ///
237    /// Performs a single pass over the file, collecting both the event counts
238    /// and the byte offsets needed by [`reconstruct_turn`].
239    fn scan(&self) -> Result<(), SessionStoreError> {
240        let mut st = self.state.lock().map_err(poison)?;
241        if !self.events_path.exists() {
242            return Ok(());
243        }
244        let content = std::fs::read_to_string(&self.events_path)
245            .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
246        let bytes = content.as_bytes();
247        let mut first_ts: Option<String> = None;
248        let mut pos = 0usize;
249        let mut in_turn = false;
250        while pos < bytes.len() {
251            let line_end = memchr_newline(bytes, pos);
252            let trimmed = std::str::from_utf8(&bytes[pos..line_end])
253                .unwrap_or("")
254                .trim();
255            if !trimmed.is_empty() {
256                if let Ok(v) = serde_json::from_str::<VersionedThreadEvent>(trimmed) {
257                    let event = v.into_event();
258                    st.manifest.event_count += 1;
259                    match &event {
260                        ThreadEvent::ThreadStarted(_) => {
261                            if first_ts.is_none() {
262                                first_ts = Some(now_rfc3339());
263                            }
264                        }
265                        ThreadEvent::TurnStarted(_) => {
266                            in_turn = true;
267                            let n = st.manifest.turn_count + 1;
268                            st.index.entries.push(TurnIndexEntry {
269                                turn_number: n,
270                                start_offset: pos as u64,
271                                end_offset: 0,
272                                event_count: 1,
273                                ts: now_rfc3339(),
274                            });
275                        }
276                        ThreadEvent::TurnCompleted(_) | ThreadEvent::TurnFailed(_) => {
277                            if in_turn {
278                                if let Some(entry) = st.index.entries.last_mut() {
279                                    entry.end_offset = line_end as u64;
280                                    entry.event_count += 1;
281                                }
282                                in_turn = false;
283                                st.manifest.turn_count = st.index.entries.len() as u64;
284                            }
285                            match &event {
286                                ThreadEvent::TurnCompleted(_) => {
287                                    st.manifest.status = "completed".to_string();
288                                }
289                                ThreadEvent::TurnFailed(_) => {
290                                    st.manifest.status = "failed".to_string();
291                                }
292                                _ => {}
293                            }
294                        }
295                        _ => {
296                            if in_turn {
297                                if let Some(entry) = st.index.entries.last_mut() {
298                                    entry.end_offset = line_end as u64;
299                                    entry.event_count += 1;
300                                }
301                            }
302                        }
303                    }
304                }
305            }
306            pos = line_end;
307        }
308        if let Some(ts) = first_ts {
309            if st.manifest.created_at.is_empty() {
310                st.manifest.created_at = ts;
311            }
312        }
313        Ok(())
314    }
315
316    fn persist_meta_locked(&self, st: &LogState) -> Result<(), SessionStoreError> {
317        let mpath = self.session_dir.join("manifest.json");
318        let bytes = serde_json::to_string_pretty(&st.manifest)?;
319        std::fs::write(&mpath, bytes).map_err(|e| SessionStoreError::io(mpath, e))?;
320        let ipath = self.session_dir.join("index").join("turns.json");
321        let bytes = serde_json::to_string_pretty(&st.index)?;
322        std::fs::write(&ipath, bytes).map_err(|e| SessionStoreError::io(ipath, e))?;
323        Ok(())
324    }
325}
326
327/// Locate the next newline at or after `from`, returning a past-the-end index.
328///
329/// Uses the `memchr` crate's SIMD byte scan instead of a scalar
330/// `Iterator::position` loop.
331fn memchr_newline(bytes: &[u8], from: usize) -> usize {
332    let slice = &bytes[from..];
333    match memchr(b'\n', slice) {
334        Some(p) => from + p + 1,
335        None => bytes.len(),
336    }
337}
338
339fn poison<T>(_e: std::sync::PoisonError<T>) -> SessionStoreError {
340    SessionStoreError::Io {
341        path: PathBuf::new(),
342        source: std::io::Error::other("session store lock poisoned"),
343    }
344}
345
346fn now_rfc3339() -> String {
347    Utc::now().to_rfc3339()
348}
349
350/// Session-level metadata persisted to `manifest.json`.
351#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
352pub struct SessionManifest {
353    /// Stable session identifier (directory name).
354    pub session_id: String,
355    /// Layout schema version (`SESSION_STORE_SCHEMA_VERSION`).
356    pub schema_version: u32,
357    /// RFC3339 creation timestamp.
358    pub created_at: String,
359    /// RFC3339 last-update timestamp.
360    pub updated_at: String,
361    /// Number of completed turns.
362    pub turn_count: u64,
363    /// Total number of events recorded.
364    pub event_count: u64,
365    /// Lifecycle status (`active` | `completed`).
366    pub status: String,
367}
368
369impl SessionManifest {
370    /// Create a fresh manifest for a session.
371    #[must_use]
372    pub fn new(session_id: &str) -> Self {
373        let ts = now_rfc3339();
374        Self {
375            session_id: session_id.to_string(),
376            schema_version: crate::SESSION_STORE_SCHEMA_VERSION,
377            created_at: ts.clone(),
378            updated_at: ts,
379            turn_count: 0,
380            event_count: 0,
381            status: "active".to_string(),
382        }
383    }
384}
385
386/// Byte-offset index of a single turn within `events.jsonl`.
387#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
388pub struct TurnIndexEntry {
389    /// Turn ordinal (1-based).
390    pub turn_number: u64,
391    /// Byte offset of the turn's first event.
392    pub start_offset: u64,
393    /// Byte offset just past the turn's last event.
394    pub end_offset: u64,
395    /// Number of events in the turn.
396    pub event_count: u64,
397    /// RFC3339 timestamp of turn start.
398    pub ts: String,
399}
400
401/// Ordered index of all turns in a session.
402#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
403pub struct TurnIndex {
404    /// Turn entries in ordinal order.
405    pub entries: Vec<TurnIndexEntry>,
406}
407
408impl TurnIndex {
409    /// Number of indexed turns.
410    #[must_use]
411    pub fn len(&self) -> usize {
412        self.entries.len()
413    }
414
415    /// Whether the index is empty.
416    #[must_use]
417    pub fn is_empty(&self) -> bool {
418        self.entries.is_empty()
419    }
420}
vtcode_session_store/event_log.rs

vtcode_session_store/
event_log.rs
