vtcode_memory/

event_log.rs

//! Append-only per-session `ThreadEvent` log plus index and manifest.

use std::collections::VecDeque;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use chrono::Utc;
use serde::{Deserialize, Serialize};
use vtcode_exec_events::{ThreadEvent, VersionedThreadEvent};

use crate::error::SessionStoreError;
use crate::manifest::ManifestStore;
use crate::session_dir;

/// Default maximum number of events retained per session before the oldest
/// completed turns are evicted.
pub const DEFAULT_MAX_EVENTS: usize = 10_000;

/// Maximum serialized event bytes retained before an append forces a write.
/// Turn boundaries and reads still flush immediately.
const MAX_WRITE_BUFFER_BYTES: usize = 64 * 1024;

/// Minimal envelope used while rebuilding the turn index.
///
/// The index only needs the event discriminator. Deserializing a complete
/// [`VersionedThreadEvent`] here would allocate every nested tool argument,
/// output, and thread item even though none of that payload is retained.
#[derive(Debug, Deserialize)]
struct VersionedEventKind<'a> {
    #[serde(rename = "schema_version", borrow)]
    _schema_version: &'a str,
    #[serde(borrow)]
    event: EventKind<'a>,
}

#[derive(Debug, Deserialize)]
struct EventKind<'a> {
    #[serde(rename = "type", borrow)]
    kind: &'a str,
}

/// In-memory state protected by a mutex (cheap; appends are infrequent relative
/// to model inference).
struct LogState {
    manifest: SessionManifest,
    index: TurnIndex,
    /// Whether we are currently inside a turn (between TurnStarted and
    /// TurnCompleted/TurnFailed). Used to update the last index entry's
    /// offsets as intermediate events arrive.
    in_turn: bool,
    /// Running byte offset of the next append. Avoids a `stat` syscall per
    /// event (the previous implementation re-statted the file twice on every
    /// `append`); initialized from the file length on `open`.
    next_offset: u64,
    /// Buffered pending writes to batch syscalls. Events are appended here
    /// and flushed to disk at turn boundaries or before read operations.
    write_buf: Vec<u8>,
}

impl LogState {
    fn new(session_id: &str) -> Self {
        Self {
            manifest: SessionManifest::new(session_id),
            index: TurnIndex::default(),
            in_turn: false,
            next_offset: 0,
            write_buf: Vec::with_capacity(65536),
        }
    }
}

/// Canonical append-only event log for a single session.
///
/// All session history is reconstructable from this log. Live conversation
/// state is never read back into context from here; the log is only consumed
/// for revert, compaction, analytics, and long-term-learning queries.
pub struct SessionEventLog {
    events_path: PathBuf,
    manifest_store: ManifestStore,
    file: Mutex<File>,
    state: Mutex<LogState>,
    max_events: usize,
}

impl SessionEventLog {
    /// Open the log for `session_id`, creating the session directory tree and
    /// rebuilding the index from `events.jsonl` if it already exists.
    pub(crate) fn open(workspace: &Path, session_id: &str, max_events: usize) -> Result<Self, SessionStoreError> {
        let dir = session_dir(workspace, session_id);
        std::fs::create_dir_all(dir.join(crate::DERIVED_DIR))
            .map_err(|e| SessionStoreError::CreateDir { path: dir.clone(), source: e })?;
        std::fs::create_dir_all(dir.join("index"))
            .map_err(|e| SessionStoreError::CreateDir { path: dir.clone(), source: e })?;
        let events_path = dir.join("events.jsonl");
        let file = OpenOptions::new()
            .create(true)
            .read(true)
            .append(true)
            .open(&events_path)
            .map_err(|e| SessionStoreError::io(events_path.clone(), e))?;
        let manifest_store = ManifestStore::new(dir.clone());
        let log = Self {
            events_path: events_path.clone(),
            manifest_store,
            file: Mutex::new(file),
            state: Mutex::new(LogState::new(session_id)),
            max_events,
        };
        // Try the fast path: read the persisted manifest + index and skip
        // the O(n) scan when they are present and consistent.
        let manifest_opt = log.manifest_store.load_manifest()?;
        let index_opt = log.manifest_store.load_turn_index()?;
        let file_len = std::fs::metadata(&events_path)
            .map_err(|e| SessionStoreError::io(events_path.clone(), e))?
            .len();
        match (manifest_opt, index_opt) {
            (Some(manifest), Some(index)) => {
                let mut st = log.state.lock().map_err(poison)?;
                st.manifest = manifest;
                st.index = index;
                st.next_offset = file_len;
            }
            _ => {
                log.scan()?;
                let mut st = log.state.lock().map_err(poison)?;
                st.next_offset = file_len;
            }
        }
        Ok(log)
    }

    /// Append an event to the log and update the in-memory index/manifest.
    pub fn append(&self, event: &ThreadEvent) -> Result<(), SessionStoreError> {
        let line = serde_json::to_string(&VersionedThreadEvent::new(event.clone()))?;
        let written = line.len() + 1;
        let mut st = self.state.lock().map_err(poison)?;
        let start = st.next_offset;
        writeln!(&mut st.write_buf, "{line}").map_err(|e| SessionStoreError::io(&self.events_path, e))?;
        let end = start + written as u64;
        st.next_offset = end;

        st.manifest.event_count += 1;
        st.manifest.updated_at = now_rfc3339();
        match event {
            ThreadEvent::TurnStarted(_) => {
                st.in_turn = true;
                let n = st.manifest.turn_count + 1;
                st.index.entries.push_back(TurnIndexEntry {
                    turn_number: n,
                    start_offset: start,
                    end_offset: end,
                    event_count: 1,
                    ts: now_rfc3339(),
                });
            }
            ThreadEvent::TurnCompleted(_) => {
                if st.in_turn {
                    if let Some(entry) = st.index.entries.back_mut() {
                        entry.end_offset = end;
                        entry.event_count += 1;
                    }
                    st.in_turn = false;
                    st.manifest.turn_count = st.index.entries.len() as u64;
                }
                st.manifest.status = "completed".to_string();
                self.persist_meta_locked(&mut st)?;
            }
            ThreadEvent::TurnFailed(_) => {
                if st.in_turn {
                    if let Some(entry) = st.index.entries.back_mut() {
                        entry.end_offset = end;
                        entry.event_count += 1;
                    }
                    st.in_turn = false;
                    st.manifest.turn_count = st.index.entries.len() as u64;
                }
                st.manifest.status = "failed".to_string();
                self.persist_meta_locked(&mut st)?;
            }
            _ => {
                if st.in_turn
                    && let Some(entry) = st.index.entries.back_mut()
                {
                    entry.end_offset = end;
                    entry.event_count += 1;
                }
            }
        }
        if st.write_buf.len() >= MAX_WRITE_BUFFER_BYTES {
            // Persist metadata with the bounded byte flush so a reopen after
            // a mid-turn crash does not trust an index that predates these
            // already-written events.
            self.persist_meta_locked(&mut st)?;
        }
        drop(st);
        self.enforce_event_cap()
    }

    /// Enforce the per-session event cap by evicting the oldest completed
    /// turns when the log exceeds [`Self::max_events`]. Returns `Ok(())` even
    /// when no truncation is needed or the cap is disabled (`max_events == 0`).
    fn enforce_event_cap(&self) -> Result<(), SessionStoreError> {
        if self.max_events == 0 {
            return Ok(());
        }
        let mut st = self.state.lock().map_err(poison)?;
        if st.manifest.event_count <= self.max_events as u64 {
            return Ok(());
        }

        // Keep ordinary appends in memory until a turn boundary or an
        // explicit read. Cap enforcement is the one append-time path that
        // needs the complete on-disk file before rewriting it.
        self.flush_write_buf_locked(&mut st)?;

        let _excess = st.manifest.event_count as i64 - self.max_events as i64;
        let mut evicted_event_count = 0u64;
        let mut truncate_offset = 0u64;

        while st.manifest.event_count > self.max_events as u64
            && let Some(oldest) = st.index.entries.front()
        {
            truncate_offset = oldest.end_offset;
            evicted_event_count += oldest.event_count;
            st.index.entries.pop_front();
        }

        if truncate_offset == 0 {
            return Ok(());
        }

        {
            let mut file = self.file.lock().map_err(poison)?;
            file.seek(SeekFrom::Start(truncate_offset))
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            let mut remaining = Vec::new();
            file.read_to_end(&mut remaining)
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            file.set_len(0).map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            file.seek(SeekFrom::Start(0))
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            file.write_all(&remaining)
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            file.flush().map_err(|e| SessionStoreError::io(&self.events_path, e))?;
        }

        for entry in &mut st.index.entries {
            entry.start_offset -= truncate_offset;
            entry.end_offset -= truncate_offset;
        }
        st.next_offset -= truncate_offset;
        st.manifest.event_count = st.manifest.event_count.saturating_sub(evicted_event_count);
        // The rewrite changed byte offsets and retained counts; persist the
        // derived metadata before exposing the append as successful.
        self.persist_meta_locked(&mut st)?;
        Ok(())
    }

    /// Reconstruct every event belonging to `turn`.
    pub(crate) fn reconstruct_turn(&self, turn: u64) -> Result<Vec<ThreadEvent>, SessionStoreError> {
        let entry = {
            let st = self.state.lock().map_err(poison)?;
            st.index
                .entries
                .iter()
                .find(|e| e.turn_number == turn)
                .cloned()
                .ok_or(SessionStoreError::TurnNotFound { session: st.manifest.session_id.clone(), turn })?
        };
        {
            let mut st = self.state.lock().map_err(poison)?;
            self.flush_write_buf_locked(&mut st)?;
        }
        let buf = {
            let mut file = self.file.lock().map_err(poison)?;
            file.seek(SeekFrom::Start(entry.start_offset))
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            let len = (entry.end_offset - entry.start_offset) as usize;
            let mut buf = vec![0u8; len];
            file.read_exact(&mut buf)
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            buf
        };
        let text = String::from_utf8_lossy(&buf);
        let mut events = Vec::new();
        for line in text.lines() {
            let line = line.trim();
            if line.is_empty() {
                continue;
            }
            // The index scan only validates the event envelope (plus the
            // lifecycle shape) so it can rebuild cheaply. A line accepted by
            // the scan can therefore still fail full decoding here; skip it
            // instead of failing the whole reconstruction (revert, compaction,
            // and analytics must not break on a single malformed record).
            let v: VersionedThreadEvent = match serde_json::from_str(line) {
                Ok(v) => v,
                Err(_) => continue,
            };
            events.push(v.into_event());
        }
        Ok(events)
    }

    /// Number of turns recorded.
    #[must_use]
    pub(crate) fn turn_count(&self) -> u64 {
        self.state.lock().map_err(poison).map_or(0, |s| s.manifest.turn_count)
    }

    /// Number of events recorded.
    #[must_use]
    pub fn event_count(&self) -> u64 {
        self.state.lock().map_err(poison).map_or(0, |s| s.manifest.event_count)
    }

    /// Flush pending event bytes and metadata to the session store.
    pub fn flush(&self) -> Result<(), SessionStoreError> {
        let mut st = self.state.lock().map_err(poison)?;
        self.persist_meta_locked(&mut st)
    }

    /// Snapshot of the session manifest.
    #[must_use]
    pub fn manifest(&self) -> SessionManifest {
        self.state
            .lock()
            .map_err(poison)
            .map(|s| s.manifest.clone())
            .unwrap_or_else(|_| SessionManifest::new(""))
    }

    /// Snapshot of the turn index.
    #[must_use]
    pub fn turn_index(&self) -> TurnIndex {
        self.state.lock().map_err(poison).map(|s| s.index.clone()).unwrap_or_default()
    }

    /// Mark the session completed and flush metadata.
    pub(crate) fn complete(&self) -> Result<(), SessionStoreError> {
        let mut st = self.state.lock().map_err(poison)?;
        st.manifest.status = "completed".to_string();
        st.manifest.updated_at = now_rfc3339();
        self.persist_meta_locked(&mut st)
    }

    /// Rebuild index + manifest by scanning `events.jsonl` (authoritative).
    ///
    /// Reads the file line-by-line via `BufReader` to avoid loading the entire
    /// log into memory. Long-lived sessions can otherwise produce multi-megabyte
    /// logs that spike memory on every reopen.
    fn scan(&self) -> Result<(), SessionStoreError> {
        let mut st = self.state.lock().map_err(poison)?;
        if !self.events_path.exists() {
            return Ok(());
        }
        let file = File::open(&self.events_path).map_err(|e| SessionStoreError::io(&self.events_path, e))?;
        let mut reader = std::io::BufReader::new(file);
        let mut buf = Vec::new();
        let mut pos = 0u64;
        let mut first_ts: Option<String> = None;
        let mut in_turn = false;
        loop {
            buf.clear();
            let n = reader
                .read_until(b'\n', &mut buf)
                .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
            if n == 0 {
                break;
            }
            let line_end = pos + n as u64;
            let trimmed = std::str::from_utf8(&buf).unwrap_or("").trim();
            if !trimmed.is_empty()
                && let Ok(v) = serde_json::from_str::<VersionedEventKind<'_>>(trimmed)
            {
                let kind = v.event.kind;
                if requires_full_lifecycle_validation(kind)
                    && serde_json::from_str::<VersionedThreadEvent>(trimmed).is_err()
                {
                    pos = line_end;
                    continue;
                }
                st.manifest.event_count += 1;
                match kind {
                    "thread.started" => {
                        if first_ts.is_none() {
                            first_ts = Some(now_rfc3339());
                        }
                    }
                    "turn.started" => {
                        in_turn = true;
                        let n = st.manifest.turn_count + 1;
                        st.index.entries.push_back(TurnIndexEntry {
                            turn_number: n,
                            start_offset: pos,
                            end_offset: line_end,
                            event_count: 1,
                            ts: now_rfc3339(),
                        });
                    }
                    "turn.completed" | "turn.failed" => {
                        if in_turn {
                            if let Some(entry) = st.index.entries.back_mut() {
                                entry.end_offset = line_end;
                                entry.event_count += 1;
                            }
                            in_turn = false;
                            st.manifest.turn_count = st.index.entries.len() as u64;
                        }
                        match kind {
                            "turn.completed" => {
                                st.manifest.status = "completed".to_string();
                            }
                            "turn.failed" => {
                                st.manifest.status = "failed".to_string();
                            }
                            _ => {}
                        }
                    }
                    _ => {
                        if in_turn && let Some(entry) = st.index.entries.back_mut() {
                            entry.end_offset = line_end;
                            entry.event_count += 1;
                        }
                    }
                }
            }
            pos = line_end;
        }
        if let Some(ts) = first_ts
            && st.manifest.created_at.is_empty()
        {
            st.manifest.created_at = ts;
        }
        Ok(())
    }

    fn persist_meta_locked(&self, st: &mut LogState) -> Result<(), SessionStoreError> {
        self.flush_write_buf_locked(st)?;
        self.manifest_store.write_manifest(&st.manifest)?;
        self.manifest_store.write_turn_index(&st.index)?;
        Ok(())
    }

    /// Flush the in-memory write buffer to the underlying file.
    fn flush_write_buf_locked(&self, st: &mut LogState) -> Result<(), SessionStoreError> {
        if st.write_buf.is_empty() {
            return Ok(());
        }
        let mut file = self.file.lock().map_err(poison)?;
        file.write_all(&st.write_buf)
            .map_err(|e| SessionStoreError::io(&self.events_path, e))?;
        st.write_buf.clear();
        Ok(())
    }
}

fn requires_full_lifecycle_validation(kind: &str) -> bool {
    matches!(kind, "thread.started" | "turn.started" | "turn.completed" | "turn.failed")
}

impl Drop for SessionEventLog {
    fn drop(&mut self) {
        if let Ok(mut st) = self.state.lock() {
            // The fallible `flush` method is the authoritative shutdown path;
            // Drop only provides a best-effort byte flush for callers that do
            // not explicitly close the log. Rewriting metadata here could
            // overwrite a manifest update made by another owner after the
            // last append.
            let _ = self.flush_write_buf_locked(&mut st);
        }
    }
}

/// Locate the next newline at or after `from`, returning a past-the-end index.
fn poison<T>(_e: std::sync::PoisonError<T>) -> SessionStoreError {
    SessionStoreError::Io {
        path: PathBuf::new(),
        source: std::io::Error::other("session store lock poisoned"),
    }
}

fn now_rfc3339() -> String {
    Utc::now().to_rfc3339()
}

/// Session-level metadata persisted to `manifest.json`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct SessionManifest {
    /// Stable session identifier (directory name).
    pub session_id: String,
    /// Layout schema version (`SESSION_STORE_SCHEMA_VERSION`).
    schema_version: u32,
    /// RFC3339 creation timestamp.
    pub created_at: String,
    /// RFC3339 last-update timestamp.
    pub updated_at: String,
    /// Number of completed turns.
    pub turn_count: u64,
    /// Total number of events recorded.
    pub event_count: u64,
    /// Lifecycle status (`active` | `completed`).
    pub status: String,
}

impl SessionManifest {
    /// Create a fresh manifest for a session.
    #[must_use]
    pub(crate) fn new(session_id: &str) -> Self {
        let ts = now_rfc3339();
        Self {
            session_id: session_id.to_string(),
            schema_version: crate::SESSION_STORE_SCHEMA_VERSION,
            created_at: ts.clone(),
            updated_at: ts,
            turn_count: 0,
            event_count: 0,
            status: "active".to_string(),
        }
    }
}

/// Byte-offset index of a single turn within `events.jsonl`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TurnIndexEntry {
    /// Turn ordinal (1-based).
    turn_number: u64,
    /// Byte offset of the turn's first event.
    start_offset: u64,
    /// Byte offset just past the turn's last event.
    end_offset: u64,
    /// Number of events in the turn.
    event_count: u64,
    /// RFC3339 timestamp of turn start.
    ts: String,
}

/// Ordered index of all turns in a session.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct TurnIndex {
    /// Turn entries in ordinal order.
    entries: VecDeque<TurnIndexEntry>,
}

impl TurnIndex {
    /// Number of indexed turns.
    #[must_use]
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Whether the index is empty.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }
}