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

vtcode_session_store/
event_log.rs
