claude_agent/session/

persistence_jsonl.rs

//! JSONL-based persistence backend compatible with Claude Code CLI.
//!
//! This module provides file-based session persistence using the JSONL (JSON Lines) format,
//! matching the Claude Code CLI's storage format for full interoperability.
//!
//! # Features
//!
//! - **Claude Code CLI Compatible**: Uses the same file structure and format as the official CLI
//! - **DAG Structure**: Messages form a directed acyclic graph via parent_uuid references
//! - **Incremental Writes**: Only new entries are appended, avoiding full file rewrites
//! - **Project-Based Organization**: Sessions organized by encoded project paths
//! - **Async I/O**: Non-blocking file operations via tokio
//!
//! # File Structure
//!
//! ```text
//! ~/.claude/
//! └── projects/
//!     └── {encoded-project-path}/
//!         ├── {session-uuid}.jsonl    # Conversation history
//!         └── ...
//! ```

use std::collections::{HashMap, HashSet};
use std::hash::{Hash, Hasher};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;

use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use uuid::Uuid;

use super::state::{
    MessageId, Session, SessionConfig, SessionId, SessionMessage, SessionMode, SessionState,
    SessionType,
};
use super::types::{
    CompactRecord, CompactTrigger, EnvironmentContext, Plan, PlanStatus, QueueItem, QueueStatus,
    SummarySnapshot, TodoItem, TodoStatus,
};
use super::{Persistence, SessionError, SessionResult};
use crate::types::{ContentBlock, Role, TokenUsage};

// ============================================================================
// Configuration
// ============================================================================

/// Sync mode for file operations.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum SyncMode {
    /// No explicit sync (OS buffering only).
    #[default]
    None,
    /// Sync after every write (safest, slowest).
    OnWrite,
}

/// Configuration for JSONL persistence.
#[derive(Clone, Debug)]
pub struct JsonlConfig {
    /// Base directory for storage (default: ~/.claude).
    pub base_dir: PathBuf,
    /// Log retention period in days (default: 30).
    pub retention_days: u32,
    /// File sync mode for durability.
    pub sync_mode: SyncMode,
}

impl Default for JsonlConfig {
    fn default() -> Self {
        Self {
            base_dir: dirs::home_dir()
                .unwrap_or_else(|| PathBuf::from("."))
                .join(".claude"),
            retention_days: 30,
            sync_mode: SyncMode::default(),
        }
    }
}

impl JsonlConfig {
    pub fn builder() -> JsonlConfigBuilder {
        JsonlConfigBuilder::default()
    }

    fn projects_dir(&self) -> PathBuf {
        self.base_dir.join("projects")
    }

    /// Encode a project path for use as a directory name.
    /// Cross-platform: handles both Unix and Windows path separators.
    fn encode_project_path(&self, path: &Path) -> String {
        path.to_string_lossy()
            .replace(['/', '\\'], "-")
            .replace(':', "_") // Windows drive letters
    }

    fn project_dir(&self, project_path: &Path) -> PathBuf {
        self.projects_dir()
            .join(self.encode_project_path(project_path))
    }
}

/// Builder for JsonlConfig.
#[derive(Default)]
pub struct JsonlConfigBuilder {
    base_dir: Option<PathBuf>,
    retention_days: Option<u32>,
    sync_mode: Option<SyncMode>,
}

impl JsonlConfigBuilder {
    pub fn base_dir(mut self, path: impl Into<PathBuf>) -> Self {
        self.base_dir = Some(path.into());
        self
    }

    pub fn retention_days(mut self, days: u32) -> Self {
        self.retention_days = Some(days);
        self
    }

    pub fn sync_mode(mut self, mode: SyncMode) -> Self {
        self.sync_mode = Some(mode);
        self
    }

    pub fn build(self) -> JsonlConfig {
        let default = JsonlConfig::default();
        JsonlConfig {
            base_dir: self.base_dir.unwrap_or(default.base_dir),
            retention_days: self.retention_days.unwrap_or(default.retention_days),
            sync_mode: self.sync_mode.unwrap_or(default.sync_mode),
        }
    }
}

// ============================================================================
// JSONL Entry Types (Claude Code CLI Compatible)
// ============================================================================

/// JSONL entry types matching Claude Code CLI format.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum JsonlEntry {
    User(UserEntry),
    Assistant(AssistantEntry),
    System(SystemEntry),
    QueueOperation(QueueOperationEntry),
    Summary(SummaryEntry),
    SessionMeta(SessionMetaEntry),
    Todo(TodoEntry),
    Plan(PlanEntry),
    Compact(CompactEntry),
}

/// Common fields shared by message entries.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct EntryCommon {
    pub uuid: String,
    #[serde(rename = "parentUuid", skip_serializing_if = "Option::is_none")]
    pub parent_uuid: Option<String>,
    #[serde(rename = "sessionId")]
    pub session_id: String,
    pub timestamp: DateTime<Utc>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cwd: Option<PathBuf>,
    pub version: String,
    #[serde(rename = "gitBranch", default)]
    pub git_branch: String,
    #[serde(rename = "isSidechain", default)]
    pub is_sidechain: bool,
}

impl EntryCommon {
    fn from_message(session_id: &SessionId, msg: &SessionMessage) -> Self {
        Self {
            uuid: msg.id.to_string(),
            parent_uuid: msg.parent_id.as_ref().map(|id| id.to_string()),
            session_id: session_id.to_string(),
            timestamp: msg.timestamp,
            cwd: msg.environment.as_ref().and_then(|e| e.cwd.clone()),
            version: env!("CARGO_PKG_VERSION").to_string(),
            git_branch: msg
                .environment
                .as_ref()
                .and_then(|e| e.git_branch.clone())
                .unwrap_or_default(),
            is_sidechain: msg.is_sidechain,
        }
    }

    fn to_environment(&self) -> EnvironmentContext {
        EnvironmentContext {
            cwd: self.cwd.clone(),
            git_branch: if self.git_branch.is_empty() {
                None
            } else {
                Some(self.git_branch.clone())
            },
            ..Default::default()
        }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct UserMessageContent {
    pub role: String,
    pub content: serde_json::Value,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct UserEntry {
    #[serde(flatten)]
    pub common: EntryCommon,
    pub message: UserMessageContent,
    #[serde(rename = "isCompactSummary", default)]
    pub is_compact_summary: bool,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AssistantMessageContent {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub model: Option<String>,
    pub role: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub stop_reason: Option<String>,
    pub content: serde_json::Value,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub usage: Option<UsageInfo>,
}

#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct UsageInfo {
    pub input_tokens: u64,
    pub output_tokens: u64,
    #[serde(default)]
    pub cache_creation_input_tokens: u64,
    #[serde(default)]
    pub cache_read_input_tokens: u64,
}

impl From<&TokenUsage> for UsageInfo {
    fn from(u: &TokenUsage) -> Self {
        Self {
            input_tokens: u.input_tokens,
            output_tokens: u.output_tokens,
            cache_creation_input_tokens: u.cache_creation_input_tokens,
            cache_read_input_tokens: u.cache_read_input_tokens,
        }
    }
}

impl From<&UsageInfo> for TokenUsage {
    fn from(u: &UsageInfo) -> Self {
        Self {
            input_tokens: u.input_tokens,
            output_tokens: u.output_tokens,
            cache_creation_input_tokens: u.cache_creation_input_tokens,
            cache_read_input_tokens: u.cache_read_input_tokens,
        }
    }
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct AssistantEntry {
    #[serde(flatten)]
    pub common: EntryCommon,
    pub message: AssistantMessageContent,
    #[serde(rename = "requestId", skip_serializing_if = "Option::is_none")]
    pub request_id: Option<String>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SystemEntry {
    #[serde(flatten)]
    pub common: EntryCommon,
    pub subtype: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub content: Option<String>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct QueueOperationEntry {
    pub operation: String,
    #[serde(rename = "sessionId")]
    pub session_id: String,
    pub timestamp: DateTime<Utc>,
    pub content: String,
    pub priority: i32,
    #[serde(rename = "itemId")]
    pub item_id: String,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SummaryEntry {
    #[serde(rename = "sessionId")]
    pub session_id: String,
    pub summary: String,
    #[serde(rename = "leafUuid", skip_serializing_if = "Option::is_none")]
    pub leaf_uuid: Option<String>,
    pub timestamp: DateTime<Utc>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SessionMetaEntry {
    #[serde(rename = "sessionId")]
    pub session_id: String,
    #[serde(rename = "parentSessionId", skip_serializing_if = "Option::is_none")]
    pub parent_session_id: Option<String>,
    #[serde(rename = "tenantId", skip_serializing_if = "Option::is_none")]
    pub tenant_id: Option<String>,
    #[serde(rename = "sessionType")]
    pub session_type: serde_json::Value,
    pub mode: String,
    pub state: String,
    pub config: serde_json::Value,
    #[serde(rename = "permissionPolicy")]
    pub permission_policy: serde_json::Value,
    #[serde(rename = "totalUsage", default)]
    pub total_usage: UsageInfo,
    #[serde(rename = "totalCostUsd", default)]
    pub total_cost_usd: f64,
    #[serde(rename = "staticContextHash", skip_serializing_if = "Option::is_none")]
    pub static_context_hash: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    #[serde(rename = "createdAt")]
    pub created_at: DateTime<Utc>,
    #[serde(rename = "updatedAt")]
    pub updated_at: DateTime<Utc>,
    #[serde(rename = "expiresAt", skip_serializing_if = "Option::is_none")]
    pub expires_at: Option<DateTime<Utc>>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TodoEntry {
    pub id: String,
    #[serde(rename = "sessionId")]
    pub session_id: String,
    pub content: String,
    #[serde(rename = "activeForm")]
    pub active_form: String,
    pub status: String,
    #[serde(rename = "planId", skip_serializing_if = "Option::is_none")]
    pub plan_id: Option<String>,
    #[serde(rename = "createdAt")]
    pub created_at: DateTime<Utc>,
    #[serde(rename = "startedAt", skip_serializing_if = "Option::is_none")]
    pub started_at: Option<DateTime<Utc>>,
    #[serde(rename = "completedAt", skip_serializing_if = "Option::is_none")]
    pub completed_at: Option<DateTime<Utc>>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct PlanEntry {
    pub id: String,
    #[serde(rename = "sessionId")]
    pub session_id: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    pub content: String,
    pub status: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    #[serde(rename = "createdAt")]
    pub created_at: DateTime<Utc>,
    #[serde(rename = "approvedAt", skip_serializing_if = "Option::is_none")]
    pub approved_at: Option<DateTime<Utc>>,
    #[serde(rename = "startedAt", skip_serializing_if = "Option::is_none")]
    pub started_at: Option<DateTime<Utc>>,
    #[serde(rename = "completedAt", skip_serializing_if = "Option::is_none")]
    pub completed_at: Option<DateTime<Utc>>,
}

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct CompactEntry {
    pub id: String,
    #[serde(rename = "sessionId")]
    pub session_id: String,
    pub trigger: String,
    #[serde(rename = "preTokens")]
    pub pre_tokens: usize,
    #[serde(rename = "postTokens")]
    pub post_tokens: usize,
    #[serde(rename = "savedTokens")]
    pub saved_tokens: usize,
    pub summary: String,
    #[serde(rename = "originalCount")]
    pub original_count: usize,
    #[serde(rename = "newCount")]
    pub new_count: usize,
    #[serde(rename = "logicalParentId", skip_serializing_if = "Option::is_none")]
    pub logical_parent_id: Option<String>,
    #[serde(rename = "createdAt")]
    pub created_at: DateTime<Utc>,
}

// ============================================================================
// Conversion: SessionMessage <-> JsonlEntry
// ============================================================================

impl JsonlEntry {
    fn from_message(session_id: &SessionId, msg: &SessionMessage) -> Self {
        let common = EntryCommon::from_message(session_id, msg);

        match msg.role {
            Role::User => JsonlEntry::User(UserEntry {
                common,
                message: UserMessageContent {
                    role: "user".to_string(),
                    content: serde_json::to_value(&msg.content).unwrap_or_default(),
                },
                is_compact_summary: msg.is_compact_summary,
            }),
            Role::Assistant => JsonlEntry::Assistant(AssistantEntry {
                common,
                message: AssistantMessageContent {
                    id: msg.metadata.request_id.clone(),
                    model: msg.metadata.model.clone(),
                    role: "assistant".to_string(),
                    stop_reason: None,
                    content: serde_json::to_value(&msg.content).unwrap_or_default(),
                    usage: msg.usage.as_ref().map(UsageInfo::from),
                },
                request_id: msg.metadata.request_id.clone(),
            }),
        }
    }

    fn to_session_message(&self) -> Option<SessionMessage> {
        match self {
            JsonlEntry::User(entry) => {
                let content: Vec<ContentBlock> =
                    serde_json::from_value(entry.message.content.clone()).unwrap_or_default();
                let mut msg = SessionMessage::user(content);
                msg.id = MessageId::from_string(&entry.common.uuid);
                msg.parent_id = entry
                    .common
                    .parent_uuid
                    .as_ref()
                    .map(MessageId::from_string);
                msg.timestamp = entry.common.timestamp;
                msg.is_sidechain = entry.common.is_sidechain;
                msg.is_compact_summary = entry.is_compact_summary;
                msg.environment = Some(entry.common.to_environment());
                Some(msg)
            }
            JsonlEntry::Assistant(entry) => {
                let content: Vec<ContentBlock> =
                    serde_json::from_value(entry.message.content.clone()).unwrap_or_default();
                let mut msg = SessionMessage::assistant(content);
                msg.id = MessageId::from_string(&entry.common.uuid);
                msg.parent_id = entry
                    .common
                    .parent_uuid
                    .as_ref()
                    .map(MessageId::from_string);
                msg.timestamp = entry.common.timestamp;
                msg.is_sidechain = entry.common.is_sidechain;
                msg.usage = entry.message.usage.as_ref().map(TokenUsage::from);
                msg.metadata.model.clone_from(&entry.message.model);
                msg.metadata.request_id.clone_from(&entry.request_id);
                msg.environment = Some(entry.common.to_environment());
                Some(msg)
            }
            _ => None,
        }
    }

    #[cfg(test)]
    fn message_uuid(&self) -> Option<&str> {
        match self {
            JsonlEntry::User(e) => Some(&e.common.uuid),
            JsonlEntry::Assistant(e) => Some(&e.common.uuid),
            _ => None,
        }
    }
}

// ============================================================================
// Session Index
// ============================================================================

#[derive(Clone, Debug)]
struct SessionMeta {
    path: PathBuf,
    project_path: Option<PathBuf>,
    tenant_id: Option<String>,
    parent_id: Option<SessionId>,
    updated_at: DateTime<Utc>,
    /// IDs of persisted messages and compacts
    persisted_ids: HashSet<String>,
    /// Hash of last persisted todos state (to detect changes)
    todos_hash: u64,
    /// Hash of last persisted plan state (to detect changes)
    plan_hash: u64,
}

#[derive(Default)]
struct SessionIndex {
    sessions: HashMap<SessionId, SessionMeta>,
    by_project: HashMap<PathBuf, Vec<SessionId>>,
    by_tenant: HashMap<String, Vec<SessionId>>,
    by_parent: HashMap<SessionId, Vec<SessionId>>,
}

impl SessionIndex {
    fn insert(&mut self, session_id: SessionId, meta: SessionMeta) {
        // Remove old entries if updating
        self.remove(&session_id);

        if let Some(ref project) = meta.project_path {
            self.by_project
                .entry(project.clone())
                .or_default()
                .push(session_id);
        }
        if let Some(ref tenant) = meta.tenant_id {
            self.by_tenant
                .entry(tenant.clone())
                .or_default()
                .push(session_id);
        }
        if let Some(parent) = meta.parent_id {
            self.by_parent.entry(parent).or_default().push(session_id);
        }
        self.sessions.insert(session_id, meta);
    }

    fn remove(&mut self, session_id: &SessionId) -> Option<SessionMeta> {
        let meta = self.sessions.remove(session_id)?;

        if let Some(ref project) = meta.project_path
            && let Some(ids) = self.by_project.get_mut(project)
        {
            ids.retain(|id| id != session_id);
        }
        if let Some(ref tenant) = meta.tenant_id
            && let Some(ids) = self.by_tenant.get_mut(tenant)
        {
            ids.retain(|id| id != session_id);
        }
        if let Some(parent) = meta.parent_id
            && let Some(ids) = self.by_parent.get_mut(&parent)
        {
            ids.retain(|id| id != session_id);
        }
        Some(meta)
    }
}

// ============================================================================
// File Operations (blocking, run via spawn_blocking)
// ============================================================================

fn read_entries_sync(path: &Path) -> SessionResult<Vec<JsonlEntry>> {
    if !path.exists() {
        return Ok(Vec::new());
    }

    let file = std::fs::File::open(path).map_err(|e| SessionError::Storage {
        message: format!("Failed to open {}: {}", path.display(), e),
    })?;

    let reader = BufReader::with_capacity(64 * 1024, file);
    let mut entries = Vec::with_capacity(128);

    for (line_num, line) in reader.lines().enumerate() {
        let line = line.map_err(|e| SessionError::Storage {
            message: format!("Read error at line {}: {}", line_num + 1, e),
        })?;

        if line.trim().is_empty() {
            continue;
        }

        match serde_json::from_str::<JsonlEntry>(&line) {
            Ok(entry) => entries.push(entry),
            Err(e) => {
                tracing::warn!(
                    path = %path.display(),
                    line = line_num + 1,
                    error = %e,
                    "Skipping malformed JSONL entry"
                );
            }
        }
    }

    Ok(entries)
}

fn append_entries_sync(path: &Path, entries: &[JsonlEntry], sync: bool) -> SessionResult<()> {
    if entries.is_empty() {
        return Ok(());
    }

    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| SessionError::Storage {
            message: format!("Failed to create directory {}: {}", parent.display(), e),
        })?;
    }

    let file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
        .map_err(|e| SessionError::Storage {
            message: format!("Failed to open {} for writing: {}", path.display(), e),
        })?;

    let mut writer = std::io::BufWriter::with_capacity(64 * 1024, file);

    for entry in entries {
        serde_json::to_writer(&mut writer, entry)?;
        writeln!(writer).map_err(|e| SessionError::Storage {
            message: format!("Write failed: {}", e),
        })?;
    }

    writer.flush().map_err(|e| SessionError::Storage {
        message: format!("Flush failed: {}", e),
    })?;

    if sync {
        writer
            .into_inner()
            .map_err(|e| SessionError::Storage {
                message: format!("Buffer error: {}", e.error()),
            })?
            .sync_all()
            .map_err(|e| SessionError::Storage {
                message: format!("Sync failed: {}", e),
            })?;
    }

    Ok(())
}

// ============================================================================
// JSONL Persistence Implementation
// ============================================================================

pub struct JsonlPersistence {
    config: JsonlConfig,
    index: Arc<RwLock<SessionIndex>>,
    summaries: Arc<RwLock<HashMap<SessionId, Vec<SummarySnapshot>>>>,
    queue: Arc<RwLock<HashMap<SessionId, Vec<QueueItem>>>>,
}

impl JsonlPersistence {
    pub async fn new(config: JsonlConfig) -> SessionResult<Self> {
        tokio::fs::create_dir_all(config.projects_dir())
            .await
            .map_err(|e| SessionError::Storage {
                message: format!("Failed to create projects directory: {}", e),
            })?;

        let persistence = Self {
            config,
            index: Arc::new(RwLock::new(SessionIndex::default())),
            summaries: Arc::new(RwLock::new(HashMap::new())),
            queue: Arc::new(RwLock::new(HashMap::new())),
        };

        persistence.rebuild_index().await?;
        Ok(persistence)
    }

    pub async fn default_config() -> SessionResult<Self> {
        Self::new(JsonlConfig::default()).await
    }

    async fn rebuild_index(&self) -> SessionResult<()> {
        let projects_dir = self.config.projects_dir();
        if !projects_dir.exists() {
            return Ok(());
        }

        let mut index = self.index.write().await;
        let mut summaries = self.summaries.write().await;

        let mut entries =
            tokio::fs::read_dir(&projects_dir)
                .await
                .map_err(|e| SessionError::Storage {
                    message: format!("Failed to read projects dir: {}", e),
                })?;

        while let Some(project_entry) =
            entries
                .next_entry()
                .await
                .map_err(|e| SessionError::Storage {
                    message: format!("Failed to read entry: {}", e),
                })?
        {
            let file_type = project_entry.file_type().await.ok();
            if !file_type.map(|t| t.is_dir()).unwrap_or(false) {
                continue;
            }

            let project_path = project_entry.path();
            let mut files =
                tokio::fs::read_dir(&project_path)
                    .await
                    .map_err(|e| SessionError::Storage {
                        message: format!("Failed to read project dir: {}", e),
                    })?;

            while let Some(file_entry) =
                files
                    .next_entry()
                    .await
                    .map_err(|e| SessionError::Storage {
                        message: format!("Failed to read file entry: {}", e),
                    })?
            {
                let file_path = file_entry.path();
                if file_path.extension().and_then(|s| s.to_str()) != Some("jsonl") {
                    continue;
                }

                let session_id = match file_path
                    .file_stem()
                    .and_then(|s| s.to_str())
                    .and_then(SessionId::parse)
                {
                    Some(id) => id,
                    None => continue,
                };

                // Read file in blocking context
                let path_clone = file_path.clone();
                let parsed = tokio::task::spawn_blocking(move || read_entries_sync(&path_clone))
                    .await
                    .map_err(|e| SessionError::Storage {
                        message: format!("Task join error: {}", e),
                    })??;

                let (meta, session_summaries) =
                    Self::parse_file_metadata(session_id, file_path, &parsed);

                index.insert(session_id, meta);
                if !session_summaries.is_empty() {
                    summaries.insert(session_id, session_summaries);
                }
            }
        }

        Ok(())
    }

    fn parse_file_metadata(
        session_id: SessionId,
        path: PathBuf,
        entries: &[JsonlEntry],
    ) -> (SessionMeta, Vec<SummarySnapshot>) {
        let mut project_path: Option<PathBuf> = None;
        let mut tenant_id: Option<String> = None;
        let mut parent_id: Option<SessionId> = None;
        let mut updated_at = Utc::now();
        let mut summaries = Vec::new();
        let mut persisted_ids = HashSet::with_capacity(entries.len());

        for entry in entries {
            match entry {
                JsonlEntry::User(e) => {
                    if project_path.is_none() {
                        project_path = e.common.cwd.clone();
                    }
                    updated_at = e.common.timestamp;
                    persisted_ids.insert(e.common.uuid.clone());
                }
                JsonlEntry::Assistant(e) => {
                    if project_path.is_none() {
                        project_path = e.common.cwd.clone();
                    }
                    updated_at = e.common.timestamp;
                    persisted_ids.insert(e.common.uuid.clone());
                }
                JsonlEntry::SessionMeta(m) => {
                    tenant_id.clone_from(&m.tenant_id);
                    parent_id = m
                        .parent_session_id
                        .as_ref()
                        .and_then(|s| SessionId::parse(s));
                    updated_at = m.updated_at;
                }
                JsonlEntry::Summary(s) => {
                    summaries.push(SummarySnapshot {
                        id: Uuid::new_v4(),
                        session_id,
                        summary: s.summary.clone(),
                        leaf_message_id: s.leaf_uuid.as_ref().map(MessageId::from_string),
                        created_at: s.timestamp,
                    });
                }
                _ => {}
            }
        }

        (
            SessionMeta {
                path,
                project_path,
                tenant_id,
                parent_id,
                updated_at,
                persisted_ids,
                todos_hash: 0, // Will be computed on first save
                plan_hash: 0,  // Will be computed on first save
            },
            summaries,
        )
    }

    fn session_file_path(&self, session_id: &SessionId, project_path: Option<&Path>) -> PathBuf {
        let dir = match project_path {
            Some(p) => self.config.project_dir(p),
            None => self.config.projects_dir().join("_default"),
        };
        dir.join(format!("{}.jsonl", session_id))
    }

    fn get_project_path(session: &Session) -> Option<PathBuf> {
        session
            .messages
            .first()
            .and_then(|m| m.environment.as_ref())
            .and_then(|e| e.cwd.clone())
    }

    /// Compute a simple hash of todos for change detection.
    fn compute_todos_hash(todos: &[TodoItem]) -> u64 {
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        for todo in todos {
            todo.id.hash(&mut hasher);
            format!("{:?}", todo.status).hash(&mut hasher);
            todo.content.hash(&mut hasher);
        }
        hasher.finish()
    }

    /// Compute a simple hash of plan for change detection.
    fn compute_plan_hash(plan: Option<&Plan>) -> u64 {
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        if let Some(p) = plan {
            p.id.hash(&mut hasher);
            format!("{:?}", p.status).hash(&mut hasher);
            p.content.hash(&mut hasher);
        }
        hasher.finish()
    }

    fn session_to_meta_entry(session: &Session) -> JsonlEntry {
        JsonlEntry::SessionMeta(SessionMetaEntry {
            session_id: session.id.to_string(),
            parent_session_id: session.parent_id.map(|p| p.to_string()),
            tenant_id: session.tenant_id.clone(),
            session_type: serde_json::to_value(&session.session_type).unwrap_or_default(),
            mode: format!("{:?}", session.mode).to_lowercase(),
            state: format!("{:?}", session.state).to_lowercase(),
            config: serde_json::to_value(&session.config).unwrap_or_default(),
            permission_policy: serde_json::to_value(&session.permission_policy).unwrap_or_default(),
            total_usage: UsageInfo::from(&session.total_usage),
            total_cost_usd: session.total_cost_usd,
            static_context_hash: session.static_context_hash.clone(),
            error: session.error.clone(),
            created_at: session.created_at,
            updated_at: session.updated_at,
            expires_at: session.expires_at,
        })
    }

    fn reconstruct_session(session_id: SessionId, entries: Vec<JsonlEntry>) -> Session {
        let mut session = Session::new(SessionConfig::default());
        session.id = session_id;

        let mut messages: HashMap<String, SessionMessage> = HashMap::with_capacity(entries.len());
        let mut todos_map: HashMap<String, TodoItem> = HashMap::new();
        let mut latest_plan: Option<Plan> = None;
        let mut compacts: Vec<CompactRecord> = Vec::new();

        for entry in entries {
            match entry {
                JsonlEntry::User(_) | JsonlEntry::Assistant(_) => {
                    if let Some(msg) = entry.to_session_message() {
                        messages.insert(msg.id.to_string(), msg);
                    }
                }
                JsonlEntry::SessionMeta(m) => {
                    session.tenant_id = m.tenant_id;
                    session.parent_id = m
                        .parent_session_id
                        .as_ref()
                        .and_then(|s| SessionId::parse(s));
                    session.session_type =
                        serde_json::from_value(m.session_type).unwrap_or(SessionType::Main);
                    session.mode = serde_json::from_str(&format!("\"{}\"", m.mode))
                        .unwrap_or(SessionMode::default());
                    session.state = SessionState::from_str_lenient(&m.state);
                    session.config = serde_json::from_value(m.config).unwrap_or_default();
                    session.permission_policy =
                        serde_json::from_value(m.permission_policy).unwrap_or_default();
                    session.total_usage = TokenUsage::from(&m.total_usage);
                    session.total_cost_usd = m.total_cost_usd;
                    session.static_context_hash = m.static_context_hash;
                    session.error = m.error;
                    session.created_at = m.created_at;
                    session.updated_at = m.updated_at;
                    session.expires_at = m.expires_at;
                }
                JsonlEntry::Summary(s) => {
                    session.summary = Some(s.summary);
                }
                JsonlEntry::Todo(t) => {
                    let todo = TodoItem {
                        id: Uuid::parse_str(&t.id).unwrap_or_else(|_| Uuid::new_v4()),
                        session_id,
                        content: t.content,
                        active_form: t.active_form,
                        status: TodoStatus::from_str_lenient(&t.status),
                        plan_id: t.plan_id.and_then(|s| Uuid::parse_str(&s).ok()),
                        created_at: t.created_at,
                        started_at: t.started_at,
                        completed_at: t.completed_at,
                    };
                    // Use map to get latest version of each todo
                    todos_map.insert(t.id, todo);
                }
                JsonlEntry::Plan(p) => {
                    let plan = Plan {
                        id: Uuid::parse_str(&p.id).unwrap_or_else(|_| Uuid::new_v4()),
                        session_id,
                        name: p.name,
                        content: p.content,
                        status: PlanStatus::from_str_lenient(&p.status),
                        error: p.error,
                        created_at: p.created_at,
                        approved_at: p.approved_at,
                        started_at: p.started_at,
                        completed_at: p.completed_at,
                    };
                    // Keep the latest plan entry
                    latest_plan = Some(plan);
                }
                JsonlEntry::Compact(c) => {
                    compacts.push(CompactRecord {
                        id: Uuid::parse_str(&c.id).unwrap_or_else(|_| Uuid::new_v4()),
                        session_id,
                        trigger: CompactTrigger::from_str_lenient(&c.trigger),
                        pre_tokens: c.pre_tokens,
                        post_tokens: c.post_tokens,
                        saved_tokens: c.saved_tokens,
                        summary: c.summary,
                        original_count: c.original_count,
                        new_count: c.new_count,
                        logical_parent_id: c.logical_parent_id.as_ref().map(MessageId::from_string),
                        created_at: c.created_at,
                    });
                }
                _ => {}
            }
        }

        // Topological sort preserving order
        let ordered = Self::topological_sort(&messages);
        session.messages = Vec::with_capacity(ordered.len());
        for msg in ordered {
            session.add_message(msg);
        }

        // Restore todos, plan, and compacts
        session.todos = todos_map.into_values().collect();
        session
            .todos
            .sort_by(|a, b| a.created_at.cmp(&b.created_at));
        session.current_plan = latest_plan;
        session.compact_history = compacts;

        session
    }

    fn topological_sort(messages: &HashMap<String, SessionMessage>) -> Vec<SessionMessage> {
        if messages.is_empty() {
            return Vec::new();
        }

        // Build parent -> children mapping, sorted by timestamp to preserve order
        let mut children: HashMap<Option<String>, Vec<&SessionMessage>> = HashMap::new();
        for msg in messages.values() {
            children
                .entry(msg.parent_id.as_ref().map(|p| p.to_string()))
                .or_default()
                .push(msg);
        }

        // Sort each group by timestamp
        for group in children.values_mut() {
            group.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
        }

        // BFS traversal using VecDeque for FIFO order
        let mut result = Vec::with_capacity(messages.len());
        let mut queue = std::collections::VecDeque::new();

        // Start with root messages (no parent)
        if let Some(roots) = children.remove(&None) {
            queue.extend(roots);
        }

        while let Some(msg) = queue.pop_front() {
            let id = msg.id.to_string();
            result.push(msg.clone());
            if let Some(child_msgs) = children.remove(&Some(id)) {
                queue.extend(child_msgs);
            }
        }

        result
    }
}

#[async_trait::async_trait]
impl Persistence for JsonlPersistence {
    fn name(&self) -> &str {
        "jsonl"
    }

    async fn save(&self, session: &Session) -> SessionResult<()> {
        let project_path = Self::get_project_path(session);
        let file_path = self.session_file_path(&session.id, project_path.as_deref());

        // Get persisted state from index (avoid re-reading file)
        let (persisted_ids, prev_todos_hash, prev_plan_hash) = {
            let index = self.index.read().await;
            match index.sessions.get(&session.id) {
                Some(m) => (m.persisted_ids.clone(), m.todos_hash, m.plan_hash),
                None => (HashSet::new(), 0, 0),
            }
        };

        let mut new_entries = Vec::new();
        let mut new_ids = HashSet::new();
        let is_first_save = persisted_ids.is_empty();

        // Session meta - always write on first save
        if is_first_save {
            new_entries.push(Self::session_to_meta_entry(session));
        }

        // Only new messages (incremental)
        for msg in &session.messages {
            let id = msg.id.to_string();
            if !persisted_ids.contains(&id) {
                new_entries.push(JsonlEntry::from_message(&session.id, msg));
                new_ids.insert(id);
            }
        }

        // Compute current hashes
        let current_todos_hash = Self::compute_todos_hash(&session.todos);
        let current_plan_hash = Self::compute_plan_hash(session.current_plan.as_ref());

        // Persist todos only if changed (including when cleared)
        if current_todos_hash != prev_todos_hash {
            for todo in &session.todos {
                new_entries.push(JsonlEntry::Todo(TodoEntry {
                    id: todo.id.to_string(),
                    session_id: session.id.to_string(),
                    content: todo.content.clone(),
                    active_form: todo.active_form.clone(),
                    status: format!("{:?}", todo.status).to_lowercase(),
                    plan_id: todo.plan_id.map(|id| id.to_string()),
                    created_at: todo.created_at,
                    started_at: todo.started_at,
                    completed_at: todo.completed_at,
                }));
            }
        }

        // Persist plan only if changed
        if current_plan_hash != prev_plan_hash
            && let Some(ref plan) = session.current_plan
        {
            new_entries.push(JsonlEntry::Plan(PlanEntry {
                id: plan.id.to_string(),
                session_id: session.id.to_string(),
                name: plan.name.clone(),
                content: plan.content.clone(),
                status: format!("{:?}", plan.status).to_lowercase(),
                error: plan.error.clone(),
                created_at: plan.created_at,
                approved_at: plan.approved_at,
                started_at: plan.started_at,
                completed_at: plan.completed_at,
            }));
        }

        // Persist compact history (incremental by ID)
        for compact in &session.compact_history {
            let compact_id = format!("compact:{}", compact.id);
            if !persisted_ids.contains(&compact_id) {
                new_entries.push(JsonlEntry::Compact(CompactEntry {
                    id: compact.id.to_string(),
                    session_id: session.id.to_string(),
                    trigger: format!("{:?}", compact.trigger).to_lowercase(),
                    pre_tokens: compact.pre_tokens,
                    post_tokens: compact.post_tokens,
                    saved_tokens: compact.saved_tokens,
                    summary: compact.summary.clone(),
                    original_count: compact.original_count,
                    new_count: compact.new_count,
                    logical_parent_id: compact.logical_parent_id.as_ref().map(|id| id.to_string()),
                    created_at: compact.created_at,
                }));
                new_ids.insert(compact_id);
            }
        }

        if new_entries.is_empty() {
            return Ok(());
        }

        // Write in blocking context
        let path_clone = file_path.clone();
        let sync = self.config.sync_mode == SyncMode::OnWrite;
        tokio::task::spawn_blocking(move || append_entries_sync(&path_clone, &new_entries, sync))
            .await
            .map_err(|e| SessionError::Storage {
                message: format!("Task join error: {}", e),
            })??;

        // Update index with new hashes
        let mut index = self.index.write().await;
        let mut persisted = persisted_ids;
        persisted.extend(new_ids);

        index.insert(
            session.id,
            SessionMeta {
                path: file_path,
                project_path,
                tenant_id: session.tenant_id.clone(),
                parent_id: session.parent_id,
                updated_at: session.updated_at,
                persisted_ids: persisted,
                todos_hash: current_todos_hash,
                plan_hash: current_plan_hash,
            },
        );

        Ok(())
    }

    async fn load(&self, id: &SessionId) -> SessionResult<Option<Session>> {
        let path = {
            let index = self.index.read().await;
            match index.sessions.get(id) {
                Some(m) => m.path.clone(),
                None => return Ok(None),
            }
        };

        let entries = tokio::task::spawn_blocking(move || read_entries_sync(&path))
            .await
            .map_err(|e| SessionError::Storage {
                message: format!("Task join error: {}", e),
            })??;

        if entries.is_empty() {
            return Ok(None);
        }

        let session = Self::reconstruct_session(*id, entries);
        Ok(Some(session))
    }

    async fn delete(&self, id: &SessionId) -> SessionResult<bool> {
        let meta = {
            let mut index = self.index.write().await;
            index.remove(id)
        };

        let Some(meta) = meta else {
            return Ok(false);
        };

        if meta.path.exists() {
            tokio::fs::remove_file(&meta.path)
                .await
                .map_err(|e| SessionError::Storage {
                    message: format!("Failed to delete {}: {}", meta.path.display(), e),
                })?;
        }

        self.summaries.write().await.remove(id);
        self.queue.write().await.remove(id);
        Ok(true)
    }

    async fn list(&self, tenant_id: Option<&str>) -> SessionResult<Vec<SessionId>> {
        let index = self.index.read().await;
        Ok(match tenant_id {
            Some(tid) => index.by_tenant.get(tid).cloned().unwrap_or_default(),
            None => index.sessions.keys().copied().collect(),
        })
    }

    async fn list_children(&self, parent_id: &SessionId) -> SessionResult<Vec<SessionId>> {
        let index = self.index.read().await;
        Ok(index.by_parent.get(parent_id).cloned().unwrap_or_default())
    }

    async fn add_summary(&self, snapshot: SummarySnapshot) -> SessionResult<()> {
        let path = {
            let index = self.index.read().await;
            index
                .sessions
                .get(&snapshot.session_id)
                .map(|m| m.path.clone())
        };

        if let Some(path) = path {
            let entry = JsonlEntry::Summary(SummaryEntry {
                session_id: snapshot.session_id.to_string(),
                summary: snapshot.summary.clone(),
                leaf_uuid: snapshot.leaf_message_id.as_ref().map(|id| id.to_string()),
                timestamp: snapshot.created_at,
            });

            let sync = self.config.sync_mode == SyncMode::OnWrite;
            tokio::task::spawn_blocking(move || append_entries_sync(&path, &[entry], sync))
                .await
                .map_err(|e| SessionError::Storage {
                    message: format!("Task join error: {}", e),
                })??;
        }

        self.summaries
            .write()
            .await
            .entry(snapshot.session_id)
            .or_default()
            .push(snapshot);

        Ok(())
    }

    async fn get_summaries(&self, session_id: &SessionId) -> SessionResult<Vec<SummarySnapshot>> {
        Ok(self
            .summaries
            .read()
            .await
            .get(session_id)
            .cloned()
            .unwrap_or_default())
    }

    async fn enqueue(
        &self,
        session_id: &SessionId,
        content: String,
        priority: i32,
    ) -> SessionResult<QueueItem> {
        let item = QueueItem::enqueue(*session_id, content.clone()).with_priority(priority);

        let path = {
            let index = self.index.read().await;
            index.sessions.get(session_id).map(|m| m.path.clone())
        };

        if let Some(path) = path {
            let entry = JsonlEntry::QueueOperation(QueueOperationEntry {
                operation: "enqueue".to_string(),
                session_id: session_id.to_string(),
                timestamp: Utc::now(),
                content,
                priority,
                item_id: item.id.to_string(),
            });

            let sync = self.config.sync_mode == SyncMode::OnWrite;
            tokio::task::spawn_blocking(move || append_entries_sync(&path, &[entry], sync))
                .await
                .map_err(|e| SessionError::Storage {
                    message: format!("Task join error: {}", e),
                })??;
        }

        self.queue
            .write()
            .await
            .entry(*session_id)
            .or_default()
            .push(item.clone());

        Ok(item)
    }

    async fn dequeue(&self, session_id: &SessionId) -> SessionResult<Option<QueueItem>> {
        let mut queue = self.queue.write().await;
        let items = match queue.get_mut(session_id) {
            Some(items) => items,
            None => return Ok(None),
        };

        items.sort_by(|a, b| b.priority.cmp(&a.priority));

        for item in items.iter_mut() {
            if item.status == QueueStatus::Pending {
                item.start_processing();
                return Ok(Some(item.clone()));
            }
        }

        Ok(None)
    }

    async fn cancel_queued(&self, item_id: Uuid) -> SessionResult<bool> {
        let mut queue = self.queue.write().await;
        for items in queue.values_mut() {
            if let Some(item) = items.iter_mut().find(|i| i.id == item_id) {
                item.cancel();
                return Ok(true);
            }
        }
        Ok(false)
    }

    async fn pending_queue(&self, session_id: &SessionId) -> SessionResult<Vec<QueueItem>> {
        Ok(self
            .queue
            .read()
            .await
            .get(session_id)
            .map(|items| {
                items
                    .iter()
                    .filter(|i| i.status == QueueStatus::Pending)
                    .cloned()
                    .collect()
            })
            .unwrap_or_default())
    }

    async fn cleanup_expired(&self) -> SessionResult<usize> {
        let cutoff = Utc::now() - chrono::Duration::days(self.config.retention_days as i64);

        // Collect expired sessions and remove from index in single lock
        let expired_paths: Vec<PathBuf> = {
            let mut index = self.index.write().await;
            let expired_ids: Vec<SessionId> = index
                .sessions
                .iter()
                .filter(|(_, m)| m.updated_at < cutoff)
                .map(|(id, _)| *id)
                .collect();

            let mut paths = Vec::with_capacity(expired_ids.len());
            for id in &expired_ids {
                if let Some(meta) = index.remove(id) {
                    paths.push(meta.path);
                }
            }
            paths
        };

        let count = expired_paths.len();

        // Delete files without holding the lock
        for path in expired_paths {
            let _ = tokio::fs::remove_file(&path).await;
        }

        Ok(count)
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::ContentBlock;
    use tempfile::TempDir;

    async fn create_test_persistence() -> (JsonlPersistence, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let config = JsonlConfig::builder()
            .base_dir(temp_dir.path().to_path_buf())
            .build();
        let persistence = JsonlPersistence::new(config).await.unwrap();
        (persistence, temp_dir)
    }

    #[tokio::test]
    async fn test_save_and_load_session() {
        let (persistence, _temp) = create_test_persistence().await;

        let mut session = Session::new(SessionConfig::default());
        session.add_message(SessionMessage::user(vec![ContentBlock::text("Hello")]));
        session.add_message(SessionMessage::assistant(vec![ContentBlock::text(
            "Hi there!",
        )]));

        persistence.save(&session).await.unwrap();

        let loaded = persistence.load(&session.id).await.unwrap().unwrap();
        assert_eq!(loaded.id, session.id);
        assert_eq!(loaded.messages.len(), 2);
    }

    #[tokio::test]
    async fn test_incremental_save() {
        let (persistence, _temp) = create_test_persistence().await;

        let mut session = Session::new(SessionConfig::default());
        session.add_message(SessionMessage::user(vec![ContentBlock::text("First")]));
        persistence.save(&session).await.unwrap();

        session.add_message(SessionMessage::assistant(vec![ContentBlock::text(
            "Second",
        )]));
        persistence.save(&session).await.unwrap();

        let loaded = persistence.load(&session.id).await.unwrap().unwrap();
        assert_eq!(loaded.messages.len(), 2);
    }

    #[tokio::test]
    async fn test_delete_session() {
        let (persistence, _temp) = create_test_persistence().await;

        let session = Session::new(SessionConfig::default());
        persistence.save(&session).await.unwrap();

        assert!(persistence.delete(&session.id).await.unwrap());
        assert!(persistence.load(&session.id).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn test_list_sessions() {
        let (persistence, _temp) = create_test_persistence().await;

        let s1 = Session::new(SessionConfig::default());
        let s2 = Session::new(SessionConfig::default());

        persistence.save(&s1).await.unwrap();
        persistence.save(&s2).await.unwrap();

        let list = persistence.list(None).await.unwrap();
        assert_eq!(list.len(), 2);
    }

    #[tokio::test]
    async fn test_tenant_filtering() {
        let (persistence, _temp) = create_test_persistence().await;

        let mut s1 = Session::new(SessionConfig::default());
        s1.tenant_id = Some("tenant-a".to_string());

        let mut s2 = Session::new(SessionConfig::default());
        s2.tenant_id = Some("tenant-b".to_string());

        persistence.save(&s1).await.unwrap();
        persistence.save(&s2).await.unwrap();

        let list = persistence.list(Some("tenant-a")).await.unwrap();
        assert_eq!(list.len(), 1);
        assert_eq!(list[0], s1.id);
    }

    #[tokio::test]
    async fn test_summaries() {
        let (persistence, _temp) = create_test_persistence().await;

        let session = Session::new(SessionConfig::default());
        persistence.save(&session).await.unwrap();

        persistence
            .add_summary(SummarySnapshot::new(session.id, "Summary 1"))
            .await
            .unwrap();
        persistence
            .add_summary(SummarySnapshot::new(session.id, "Summary 2"))
            .await
            .unwrap();

        let summaries = persistence.get_summaries(&session.id).await.unwrap();
        assert_eq!(summaries.len(), 2);
    }

    #[tokio::test]
    async fn test_queue_operations() {
        let (persistence, _temp) = create_test_persistence().await;

        let session = Session::new(SessionConfig::default());
        persistence.save(&session).await.unwrap();

        persistence
            .enqueue(&session.id, "Low priority".to_string(), 1)
            .await
            .unwrap();
        persistence
            .enqueue(&session.id, "High priority".to_string(), 10)
            .await
            .unwrap();

        let next = persistence.dequeue(&session.id).await.unwrap().unwrap();
        assert_eq!(next.content, "High priority");
    }

    #[tokio::test]
    async fn test_dag_reconstruction() {
        let (persistence, _temp) = create_test_persistence().await;

        let mut session = Session::new(SessionConfig::default());
        session.add_message(SessionMessage::user(vec![ContentBlock::text("Q1")]));
        session.add_message(SessionMessage::assistant(vec![ContentBlock::text("A1")]));
        session.add_message(SessionMessage::user(vec![ContentBlock::text("Q2")]));
        session.add_message(SessionMessage::assistant(vec![ContentBlock::text("A2")]));

        persistence.save(&session).await.unwrap();

        let loaded = persistence.load(&session.id).await.unwrap().unwrap();

        assert_eq!(loaded.messages.len(), 4);
        assert!(
            loaded.messages[0]
                .content
                .iter()
                .any(|c| c.as_text() == Some("Q1"))
        );
        assert!(
            loaded.messages[1]
                .content
                .iter()
                .any(|c| c.as_text() == Some("A1"))
        );
        assert!(
            loaded.messages[2]
                .content
                .iter()
                .any(|c| c.as_text() == Some("Q2"))
        );
        assert!(
            loaded.messages[3]
                .content
                .iter()
                .any(|c| c.as_text() == Some("A2"))
        );
    }

    #[tokio::test]
    async fn test_project_path_encoding() {
        let config = JsonlConfig::default();

        assert_eq!(
            config.encode_project_path(Path::new("/home/user/project")),
            "-home-user-project"
        );
        assert_eq!(
            config.encode_project_path(Path::new("/Users/alice/work/app")),
            "-Users-alice-work-app"
        );
    }

    #[test]
    fn test_jsonl_entry_serialization() {
        let msg = SessionMessage::user(vec![ContentBlock::text("Hello")]);
        let session_id = SessionId::new();
        let entry = JsonlEntry::from_message(&session_id, &msg);

        let json = serde_json::to_string(&entry).unwrap();
        assert!(json.contains("\"type\":\"user\""));

        let parsed: JsonlEntry = serde_json::from_str(&json).unwrap();
        assert!(matches!(parsed, JsonlEntry::User(_)));
    }

    #[tokio::test]
    async fn test_no_duplicate_writes() {
        let (persistence, _temp) = create_test_persistence().await;

        let mut session = Session::new(SessionConfig::default());
        session.add_message(SessionMessage::user(vec![ContentBlock::text("Hello")]));
        persistence.save(&session).await.unwrap();
        persistence.save(&session).await.unwrap(); // Save same data twice
        persistence.save(&session).await.unwrap(); // And again

        // Check file has only one message entry + session meta
        let file_path = persistence.session_file_path(&session.id, None);
        let entries = read_entries_sync(&file_path).unwrap();
        let message_count = entries
            .iter()
            .filter(|e| e.message_uuid().is_some())
            .count();
        assert_eq!(message_count, 1, "Should not duplicate message entries");
    }

    #[test]
    fn test_windows_path_encoding() {
        let config = JsonlConfig::default();
        // Windows-style path
        assert_eq!(
            config.encode_project_path(Path::new("C:\\Users\\alice\\project")),
            "C_-Users-alice-project"
        );
    }
}