talos_session/

manager.rs

use crate::sqlite::{ForkInfo, IndexError, SearchResult, SessionIndex};
use crate::store::{CompactTextSessionStore, JsonlSessionStore, SessionStore};
use crate::todo::{TodoError, TodoRepository};
use crate::topology::{workspace_dir_name, workspace_root_from_dir_name};
use crate::{
    DurableSession, OrphanSidecarReconciliationPolicy, OrphanSidecarReconciliationReport, Session,
    SessionArtifactCleanupReport, SessionError, SessionInfo,
    remove_session_sidecars_for_transcript, remove_session_transcript,
};
use chrono::{DateTime, Duration, Utc};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use uuid::Uuid;

const KNOWN_EXTENSIONS: &[&str] = &["jsonl", "tlog"];

/// Policy for selecting session cleanup candidates.
#[derive(Debug, Clone, Default)]
pub struct SessionCleanupPolicy {
    /// Workspace root to limit cleanup to. `None` scans all workspaces.
    pub workspace_root: Option<String>,
    /// Keep at most this many newest sessions after excluding protected IDs.
    pub max_sessions_per_workspace: Option<usize>,
    /// Delete sessions older than this many days after excluding protected IDs.
    pub max_age_days: Option<i64>,
    /// Session IDs that must never be selected for cleanup.
    pub protected_session_ids: Vec<Uuid>,
}

/// A session selected by a cleanup policy.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionCleanupCandidate {
    /// Session identifier.
    pub id: Uuid,
    /// Workspace root associated with the session.
    pub workspace_root: String,
    /// JSONL file path to remove if cleanup is applied.
    pub file_path: PathBuf,
    /// File size in bytes at selection time.
    pub size_bytes: u64,
    /// Last modified timestamp used for retention decisions.
    pub timestamp: DateTime<Utc>,
    /// Human-readable reason this session was selected.
    pub reason: String,
}

/// Result of applying a session cleanup policy.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SessionCleanupReport {
    /// Candidates selected by the policy.
    pub candidates: Vec<SessionCleanupCandidate>,
    /// Number of candidates actually removed.
    pub removed: usize,
    /// Total bytes removed from complete Session-owned artifact sets.
    pub bytes_removed: u64,
}

/// Manages sessions on disk.
pub struct SessionManager {
    pub(crate) sessions_dir: PathBuf,
    index: Arc<Mutex<Option<SessionIndex>>>,
    store: Arc<dyn SessionStore>,
    jsonl_store: Arc<dyn SessionStore>,
}

impl std::fmt::Debug for SessionManager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SessionManager")
            .field("sessions_dir", &self.sessions_dir)
            .finish()
    }
}

impl Clone for SessionManager {
    fn clone(&self) -> Self {
        Self {
            sessions_dir: self.sessions_dir.clone(),
            index: Arc::clone(&self.index),
            store: Arc::clone(&self.store),
            jsonl_store: Arc::clone(&self.jsonl_store),
        }
    }
}

impl SessionManager {
    /// Create a new `SessionManager` with the default sessions directory (`~/.talos/sessions/`).
    pub fn new() -> Result<Self, SessionError> {
        let dir = Self::default_sessions_dir()?;
        let manager = Self {
            sessions_dir: dir,
            index: Arc::new(Mutex::new(None)),
            store: Arc::new(CompactTextSessionStore),
            jsonl_store: Arc::new(JsonlSessionStore),
        };
        if let Err(error) = manager.reconcile_index() {
            eprintln!("Session index reconciliation failed during startup: {error}");
        }
        match manager.reconcile_orphan_sidecars(&OrphanSidecarReconciliationPolicy::default()) {
            Ok(report) => {
                if report.bounded {
                    eprintln!(
                        "Session orphan-sidecar reconciliation reached its safety bound after scanning {} entries; continuation state was saved. Run `talos storage maintenance --reconcile` to continue.",
                        report.scanned_entries,
                    );
                }
                for failure in report.failures {
                    eprintln!(
                        "Session orphan-sidecar reconciliation failed for {} at {}: {}",
                        failure.session_id,
                        failure.path.display(),
                        failure.error,
                    );
                }
            }
            Err(error) => {
                eprintln!("Session orphan-sidecar reconciliation failed during startup: {error}");
            }
        }
        Ok(manager)
    }

    /// Return the default sessions directory without opening indexes or reconciling state.
    ///
    /// # Errors
    ///
    /// Returns an error if no user home directory environment variable is available.
    pub fn default_sessions_dir() -> Result<PathBuf, SessionError> {
        let home = home_dir_from_env()?;
        Ok(PathBuf::from(home).join(".talos").join("sessions"))
    }

    /// Create a new `SessionManager` with a custom sessions directory.
    pub fn with_dir(sessions_dir: PathBuf) -> Self {
        Self {
            sessions_dir,
            index: Arc::new(Mutex::new(None)),
            store: Arc::new(CompactTextSessionStore),
            jsonl_store: Arc::new(JsonlSessionStore),
        }
    }

    /// Creates or opens a UUID-backed durable session for a host logical ID.
    ///
    /// The external ID is stored only in a colocated binding index; it is never
    /// used as a path component or filename.
    pub fn create_or_open_session(
        &self,
        external_id: &str,
    ) -> Result<DurableSession, SessionError> {
        crate::durable::create_or_open(&self.sessions_dir, external_id)
    }

    /// Looks up a durable session by its host logical ID without creating one.
    pub fn get_session_by_external_id(
        &self,
        external_id: &str,
    ) -> Result<Option<DurableSession>, SessionError> {
        crate::durable::get_by_external_id(&self.sessions_dir, external_id)
    }

    /// Returns whether a UUID-backed session exists without accepting a path.
    pub fn session_exists(&self, id: &Uuid) -> bool {
        self.get_session(id).is_ok()
    }

    /// Reads all normalized entries for a UUID-backed session.
    pub fn read_session(&self, id: &Uuid) -> Result<Vec<crate::SessionEntry>, SessionError> {
        self.get_session(id)?.read_entries()
    }

    /// Returns the byte size of a UUID-backed session file.
    pub fn session_size(&self, id: &Uuid) -> Result<u64, SessionError> {
        Ok(fs::metadata(self.find_session_file(id)?)?.len())
    }

    /// Return the root directory used for session JSONL files and the colocated index.
    #[must_use]
    pub fn sessions_dir(&self) -> &Path {
        &self.sessions_dir
    }

    fn is_session_file(&self, path: &Path) -> bool {
        path.extension()
            .and_then(|e| e.to_str())
            .is_some_and(|ext| KNOWN_EXTENSIONS.contains(&ext))
    }

    fn store_for_path(&self, path: &Path) -> &dyn SessionStore {
        match path.extension().and_then(|e| e.to_str()) {
            Some("tlog") => self.store.as_ref(),
            _ => self.jsonl_store.as_ref(),
        }
    }

    /// Open the colocated session todo repository and initialize its schema.
    ///
    /// The repository is session-scoped by data, not by database file: all todo rows carry a
    /// `session_id` and share one SQLite file under the sessions directory.
    ///
    /// # Errors
    ///
    /// Returns an error if the SQLite database cannot be opened or initialized.
    pub fn todo_repository(&self) -> Result<TodoRepository, TodoError> {
        let repo = TodoRepository::new(&self.sessions_dir.join("todos.sqlite"))?;
        repo.init_schema()?;
        Ok(repo)
    }

    /// Create a new session for the given project and workspace.
    ///
    /// The session file is created at `~/.talos/sessions/<workspace_dir>/<uuid>.tlog`,
    /// where `workspace_dir` is a hash of the workspace root path.
    pub fn create_session(
        &self,
        project: &str,
        workspace_root: &str,
    ) -> Result<Session, SessionError> {
        let id = Uuid::new_v4();
        let project_dir = self.sessions_dir.join(workspace_dir_name(workspace_root));
        fs::create_dir_all(&project_dir)?;

        let file_path = project_dir.join(format!("{id}.{}", self.store.file_extension()));
        fs::File::create(&file_path)?;

        Ok(Session::new(
            id,
            project.to_string(),
            workspace_root.to_string(),
            file_path,
        ))
    }

    /// Prepare a new session without creating the file on disk.
    ///
    /// Returns a [`Session`] with `persisted = false`. The JSONL file is
    /// not created until [`Session::ensure_persisted`] is called (triggered
    /// automatically on the first message append).
    pub fn defer_create_session(
        &self,
        project: &str,
        workspace_root: &str,
    ) -> Result<Session, SessionError> {
        let id = Uuid::new_v4();
        let project_dir = self.sessions_dir.join(workspace_dir_name(workspace_root));
        let file_path = project_dir.join(format!("{id}.{}", self.store.file_extension()));

        Ok(Session::new_deferred(
            id,
            project.to_string(),
            workspace_root.to_string(),
            file_path,
        ))
    }

    /// Load an existing session by ID.
    ///
    /// Scans all workspace directories for a file matching `<id>.<ext>`
    /// where `<ext>` is any known session file extension (`.tlog` or `.jsonl`).
    /// If both exist for the same UUID, returns a duplicate-format error.
    pub fn get_session(&self, id: &Uuid) -> Result<Session, SessionError> {
        if !self.sessions_dir.exists() {
            return Err(SessionError::SessionNotFound(*id));
        }

        for entry in fs::read_dir(&self.sessions_dir)? {
            let entry = entry?;
            if !entry.file_type()?.is_dir() {
                continue;
            }
            let project_dir = entry.path();
            let dir_name = project_dir
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("unknown")
                .to_string();

            let mut found: Option<(PathBuf, Arc<dyn SessionStore>)> = None;
            for ext in KNOWN_EXTENSIONS {
                let candidate = project_dir.join(format!("{id}.{ext}"));
                if candidate.exists() {
                    if found.is_some() {
                        return Err(SessionError::ParseError(format!(
                            "duplicate session files for {id}: both .tlog and .jsonl exist"
                        )));
                    }
                    let store = if ext == &"tlog" {
                        Arc::clone(&self.store)
                    } else {
                        Arc::clone(&self.jsonl_store)
                    };
                    found = Some((candidate, store));
                }
            }

            if let Some((file_path, store)) = found {
                let metadata = fs::metadata(&file_path)?;
                let created_at = metadata
                    .modified()
                    .ok()
                    .map(DateTime::<Utc>::from)
                    .unwrap_or_else(Utc::now);

                let mut session = Session::with_store(
                    *id,
                    dir_name.clone(),
                    workspace_root_from_dir_name(&dir_name),
                    file_path,
                    store,
                );
                session.created_at = created_at;

                let entries = session.read_entries()?;
                if !entries.is_empty()
                    && let Some(branch) = session.branches.get_mut(&session.current_branch)
                {
                    branch.entries = entries;
                }

                return Ok(session);
            }
        }

        Err(SessionError::SessionNotFound(*id))
    }

    /// List all sessions across all workspace directories.
    pub fn list_sessions(&self) -> Result<Vec<SessionInfo>, SessionError> {
        let mut sessions = Vec::new();

        if !self.sessions_dir.exists() {
            return Ok(sessions);
        }

        for entry in fs::read_dir(&self.sessions_dir)? {
            let entry = entry?;
            if !entry.file_type()?.is_dir() {
                continue;
            }
            let project_dir = entry.path();
            let dir_name = project_dir
                .file_name()
                .and_then(|n| n.to_str())
                .unwrap_or("unknown")
                .to_string();

            for file_entry in fs::read_dir(&project_dir)? {
                let file_entry = file_entry?;
                let path = file_entry.path();
                if !self.is_session_file(&path) {
                    continue;
                }

                let file_stem = path
                    .file_stem()
                    .and_then(|s| s.to_str())
                    .and_then(|s| Uuid::parse_str(s).ok());

                if let Some(id) = file_stem {
                    let metadata = fs::metadata(&path)?;
                    let timestamp = metadata
                        .modified()
                        .ok()
                        .map(DateTime::<Utc>::from)
                        .unwrap_or_else(Utc::now);

                    let store = self.store_for_path(&path);
                    let info = store.scan_file(&path)?;

                    sessions.push(SessionInfo {
                        id,
                        project: dir_name.clone(),
                        workspace_root: String::new(),
                        last_message_preview: info.last_message_preview,
                        timestamp,
                        message_count: info.message_count,
                    });
                }
            }
        }

        Ok(sessions)
    }

    /// List sessions for one workspace directory.
    pub fn list_workspace_sessions(
        &self,
        workspace_root: &str,
    ) -> Result<Vec<SessionInfo>, SessionError> {
        let workspace_dir = self.sessions_dir.join(workspace_dir_name(workspace_root));
        if !workspace_dir.exists() {
            return Ok(Vec::new());
        }

        let mut sessions = Vec::new();
        for file_entry in fs::read_dir(&workspace_dir)? {
            let file_entry = file_entry?;
            let path = file_entry.path();
            if !self.is_session_file(&path) {
                continue;
            }

            let file_stem = path
                .file_stem()
                .and_then(|s| s.to_str())
                .and_then(|s| Uuid::parse_str(s).ok());

            if let Some(id) = file_stem {
                let metadata = fs::metadata(&path)?;
                let timestamp = metadata
                    .modified()
                    .ok()
                    .map(DateTime::<Utc>::from)
                    .unwrap_or_else(Utc::now);

                let store = self.store_for_path(&path);
                let info = store.scan_file(&path)?;

                sessions.push(SessionInfo {
                    id,
                    project: String::new(),
                    workspace_root: workspace_root.to_string(),
                    last_message_preview: info.last_message_preview,
                    timestamp,
                    message_count: info.message_count,
                });
            }
        }

        Ok(sessions)
    }

    /// Return the most recently modified session for one workspace directory.
    pub fn latest_workspace_session(
        &self,
        workspace_root: &str,
    ) -> Result<Option<SessionInfo>, SessionError> {
        let sessions = self.list_workspace_sessions(workspace_root)?;
        Ok(sessions.into_iter().max_by_key(|s| s.timestamp))
    }

    /// Resume a session by ID, loading all entries from the session file.
    ///
    /// This is equivalent to [`SessionManager::get_session`] but with a clearer
    /// name for the "resume" use case.
    pub fn resume_session(&self, session_id: &str) -> Result<Session, SessionError> {
        let id = Uuid::parse_str(session_id)
            .map_err(|_| SessionError::SessionNotFound(Uuid::new_v4()))?;
        self.get_session(&id)
    }

    fn get_or_create_index(
        &self,
    ) -> Result<std::sync::MutexGuard<'_, Option<SessionIndex>>, IndexError> {
        let mut guard = self.index.lock().expect("index lock poisoned");
        if guard.is_none() {
            let db_path = self.sessions_dir.join("index.db");

            let index = SessionIndex::new(&db_path)?;
            index.init_schema()?;
            *guard = Some(index);
        }
        Ok(guard)
    }

    /// Perform a full-text search across all indexed session messages.
    ///
    /// Returns results ranked by relevance. The index is created lazily if it
    /// does not exist.
    pub fn search(&self, query: &str, limit: usize) -> Result<Vec<SearchResult>, IndexError> {
        let guard = self.get_or_create_index()?;
        let index = guard.as_ref().expect("index just created");
        index.search(query, limit)
    }

    /// List the most recently updated sessions from the index.
    ///
    /// Returns sessions ordered by last update time, descending.
    pub fn list_recent(&self, limit: usize) -> Result<Vec<SessionInfo>, IndexError> {
        let guard = self.get_or_create_index()?;
        let index = guard.as_ref().expect("index just created");
        index.list_recent(limit)
    }

    /// Update the search index for a session.
    ///
    /// If the index has not been initialized, this is a no-op.
    pub fn update_index(&self, session: &Session) -> Result<(), IndexError> {
        let mut guard = self.get_or_create_index()?;
        let index = guard.as_mut().expect("index just created");
        index.index_session(session)
    }

    /// Checkpoint the session index WAL and truncate it where possible.
    pub fn checkpoint_index(&self) -> Result<(), IndexError> {
        let guard = self.get_or_create_index()?;
        let index = guard.as_ref().expect("index just created");
        index.checkpoint_truncate()
    }

    /// Vacuum the session index database.
    pub fn vacuum_index(&self) -> Result<(), IndexError> {
        let guard = self.get_or_create_index()?;
        let index = guard.as_ref().expect("index just created");
        index.vacuum()
    }

    /// Return forks originating from the given session ID.
    pub fn get_forks(&self, session_id: &str) -> Result<Vec<ForkInfo>, IndexError> {
        let guard = self.get_or_create_index()?;
        let index = guard.as_ref().expect("index just created");
        index.get_forks(session_id)
    }

    /// Record one source/child fork relationship through the manager-owned index.
    pub fn record_fork(
        &self,
        source_session_id: &Uuid,
        forked_session_id: &Uuid,
        fork_entry_id: &str,
    ) -> Result<(), IndexError> {
        let mut guard = self.get_or_create_index()?;
        let index = guard.as_mut().expect("index just created");
        index.record_fork(
            &source_session_id.to_string(),
            &forked_session_id.to_string(),
            fork_entry_id,
        )
    }

    #[allow(clippy::collapsible_if)]
    pub fn reconcile_index(&self) -> Result<usize, IndexError> {
        let mut guard = self.get_or_create_index()?;
        let index = guard.as_mut().expect("index just created");

        let mut fixed = 0usize;

        let indexed_ids: std::collections::HashSet<String> =
            index.list_all_session_ids()?.into_iter().collect();

        let mut on_disk_ids: std::collections::HashSet<String> = std::collections::HashSet::new();

        if self.sessions_dir.exists() {
            for ws_entry in fs::read_dir(&self.sessions_dir)? {
                let ws_entry = ws_entry?;
                if !ws_entry.file_type()?.is_dir() {
                    continue;
                }
                let ws_dir = ws_entry.path();
                let workspace_root = workspace_root_from_dir_name(
                    &ws_dir.file_name().unwrap_or_default().to_string_lossy(),
                );

                for file_entry in fs::read_dir(&ws_dir)? {
                    let file_entry = file_entry?;
                    let path = file_entry.path();
                    if !self.is_session_file(&path) {
                        continue;
                    }
                    let stem = match path.file_stem().and_then(|s| s.to_str()) {
                        Some(s) => s.to_string(),
                        None => continue,
                    };
                    on_disk_ids.insert(stem.clone());

                    let existing = index.get_session_info(&stem)?;
                    let store = self.store_for_path(&path);
                    let info = store.scan_file(&path).unwrap_or(SessionInfo {
                        id: Uuid::nil(),
                        project: String::new(),
                        workspace_root: String::new(),
                        last_message_preview: String::new(),
                        timestamp: Utc::now(),
                        message_count: 0,
                    });
                    let msg_count = info.message_count;
                    let needs_reindex = match &existing {
                        None => true,
                        Some(info) => info.message_count != msg_count,
                    };
                    if needs_reindex && Uuid::parse_str(&stem).is_ok() {
                        if let Ok(id) = Uuid::parse_str(&stem) {
                            let project = ws_dir
                                .file_name()
                                .and_then(|n| n.to_str())
                                .unwrap_or("unknown")
                                .to_string();
                            let session_store =
                                if path.extension().and_then(|e| e.to_str()) == Some("tlog") {
                                    Arc::clone(&self.store)
                                } else {
                                    Arc::clone(&self.jsonl_store)
                                };
                            let mut session = Session::with_store(
                                id,
                                project,
                                workspace_root.to_string(),
                                path.clone(),
                                session_store,
                            );
                            if let Ok(entries) = session.read_entries() {
                                if let Some(branch) =
                                    session.branches.get_mut(&session.current_branch)
                                {
                                    branch.entries = entries;
                                }
                            }
                            index.index_session(&session)?;
                            fixed += 1;
                        }
                    }
                }
            }
        }

        for orphan_id in indexed_ids.difference(&on_disk_ids) {
            index.delete_session(orphan_id)?;
            fixed += 1;
        }

        Ok(fixed)
    }

    /// Delete one discoverable Session through the retryable ownership boundary.
    pub fn delete_session(&self, id: &Uuid) -> Result<(), SessionError> {
        let file_path = self.find_session_file(id)?;
        self.remove_owned_session_artifacts(id, &file_path)
            .map(|_| ())
    }

    /// Roll back all artifacts owned by a prepared or partially created Session.
    ///
    /// Unlike `delete_session`, this does not require the transcript to exist, so
    /// post-identity initialization failures remain recoverable.
    pub fn rollback_session_artifacts(
        &self,
        session: &Session,
    ) -> Result<SessionArtifactCleanupReport, SessionError> {
        self.remove_owned_session_artifacts(&session.id, &session.file_path)
    }

    fn remove_owned_session_artifacts(
        &self,
        id: &Uuid,
        transcript_path: &Path,
    ) -> Result<SessionArtifactCleanupReport, SessionError> {
        let mut report = remove_session_sidecars_for_transcript(transcript_path)?;
        crate::durable::remove_binding_for_session(&self.sessions_dir, id)?;
        let mut guard = self
            .get_or_create_index()
            .map_err(|error| SessionError::IndexCleanup {
                session_id: *id,
                message: error.to_string(),
            })?;
        if let Some(index) = guard.as_mut() {
            index
                .delete_session(&id.to_string())
                .map_err(|error| SessionError::IndexCleanup {
                    session_id: *id,
                    message: error.to_string(),
                })?;
        }
        report.merge(remove_session_transcript(transcript_path)?);
        Ok(report)
    }

    /// Discover and remove safe transcript-less pending SQLite artifacts.
    pub fn reconcile_orphan_sidecars(
        &self,
        policy: &OrphanSidecarReconciliationPolicy,
    ) -> Result<OrphanSidecarReconciliationReport, SessionError> {
        crate::artifacts::reconcile_orphan_sidecars_in_root(&self.sessions_dir, policy)
    }

    /// Return sessions that would be removed by `policy` without deleting files.
    pub fn cleanup_candidates(
        &self,
        policy: &SessionCleanupPolicy,
    ) -> Result<Vec<SessionCleanupCandidate>, SessionError> {
        let mut by_workspace = self.collect_cleanup_sessions(policy)?;
        let protected: std::collections::HashSet<Uuid> =
            policy.protected_session_ids.iter().copied().collect();
        let cutoff = policy
            .max_age_days
            .map(|days| Utc::now() - Duration::days(days.max(0)));

        let mut candidates = Vec::new();
        for (workspace_root, sessions) in by_workspace.iter_mut() {
            sessions.sort_by(|a, b| b.timestamp.cmp(&a.timestamp).then_with(|| a.id.cmp(&b.id)));

            for session in sessions.iter() {
                if protected.contains(&session.id) {
                    continue;
                }
                if let Some(cutoff) = cutoff
                    && session.timestamp < cutoff
                {
                    candidates.push(SessionCleanupCandidate {
                        id: session.id,
                        workspace_root: workspace_root.clone(),
                        file_path: session.file_path.clone(),
                        size_bytes: session.size_bytes,
                        timestamp: session.timestamp,
                        reason: format!(
                            "older than {} day(s)",
                            policy.max_age_days.unwrap_or_default().max(0)
                        ),
                    });
                }
            }

            if let Some(max_sessions) = policy.max_sessions_per_workspace {
                let mut unprotected: Vec<_> = sessions
                    .iter()
                    .filter(|session| !protected.contains(&session.id))
                    .collect();
                unprotected
                    .sort_by(|a, b| b.timestamp.cmp(&a.timestamp).then_with(|| a.id.cmp(&b.id)));
                for session in unprotected.into_iter().skip(max_sessions) {
                    if candidates
                        .iter()
                        .any(|candidate| candidate.id == session.id)
                    {
                        continue;
                    }
                    candidates.push(SessionCleanupCandidate {
                        id: session.id,
                        workspace_root: workspace_root.clone(),
                        file_path: session.file_path.clone(),
                        size_bytes: session.size_bytes,
                        timestamp: session.timestamp,
                        reason: format!("exceeds max_sessions_per_workspace={max_sessions}"),
                    });
                }
            }
        }

        candidates.sort_by(|a, b| a.timestamp.cmp(&b.timestamp).then_with(|| a.id.cmp(&b.id)));
        Ok(candidates)
    }

    /// Apply a cleanup policy by deleting selected sessions and index rows.
    ///
    /// This is an explicit maintenance operation. It never removes IDs listed in
    /// `protected_session_ids`, even if callers accidentally include an active
    /// session in an otherwise matching policy.
    pub fn apply_cleanup(
        &self,
        policy: &SessionCleanupPolicy,
    ) -> Result<SessionCleanupReport, SessionError> {
        let candidates = self.cleanup_candidates(policy)?;
        let mut report = SessionCleanupReport {
            candidates,
            removed: 0,
            bytes_removed: 0,
        };

        for candidate in &report.candidates {
            let cleanup =
                self.remove_owned_session_artifacts(&candidate.id, &candidate.file_path)?;
            report.removed = report.removed.saturating_add(1);
            report.bytes_removed = report.bytes_removed.saturating_add(cleanup.bytes_removed);
        }

        Ok(report)
    }

    #[allow(clippy::collapsible_if)]
    fn find_session_file(&self, id: &Uuid) -> Result<PathBuf, SessionError> {
        if self.sessions_dir.exists() {
            for ws_entry in fs::read_dir(&self.sessions_dir)? {
                let ws_entry = ws_entry?;
                if !ws_entry.file_type()?.is_dir() {
                    continue;
                }
                let mut found: Option<PathBuf> = None;
                for ext in KNOWN_EXTENSIONS {
                    let candidate = ws_entry.path().join(format!("{id}.{ext}"));
                    if candidate.exists() {
                        if found.is_some() {
                            return Err(SessionError::ParseError(format!(
                                "duplicate session files for {id}: both .tlog and .jsonl exist"
                            )));
                        }
                        found = Some(candidate);
                    }
                }
                if let Some(path) = found {
                    return Ok(path);
                }
            }
        }
        Err(SessionError::SessionNotFound(*id))
    }

    fn collect_cleanup_sessions(
        &self,
        policy: &SessionCleanupPolicy,
    ) -> Result<std::collections::HashMap<String, Vec<CleanupSession>>, SessionError> {
        let mut by_workspace: std::collections::HashMap<String, Vec<CleanupSession>> =
            std::collections::HashMap::new();

        if !self.sessions_dir.exists() {
            return Ok(by_workspace);
        }

        if let Some(target) = &policy.workspace_root {
            let workspace_dir = self.sessions_dir.join(workspace_dir_name(target));
            if workspace_dir.exists() {
                self.collect_cleanup_workspace(target, &workspace_dir, &mut by_workspace)?;
            }
            return Ok(by_workspace);
        }

        for ws_entry in fs::read_dir(&self.sessions_dir)? {
            let ws_entry = ws_entry?;
            if !ws_entry.file_type()?.is_dir() {
                continue;
            }
            let ws_dir = ws_entry.path();
            let workspace_root = workspace_root_from_dir_name(
                &ws_dir.file_name().unwrap_or_default().to_string_lossy(),
            );
            self.collect_cleanup_workspace(&workspace_root, &ws_dir, &mut by_workspace)?;
        }

        Ok(by_workspace)
    }

    fn collect_cleanup_workspace(
        &self,
        workspace_root: &str,
        workspace_dir: &Path,
        by_workspace: &mut std::collections::HashMap<String, Vec<CleanupSession>>,
    ) -> Result<(), SessionError> {
        for file_entry in fs::read_dir(workspace_dir)? {
            let file_entry = file_entry?;
            let path = file_entry.path();
            if !self.is_session_file(&path) {
                continue;
            }
            let Some(id) = path
                .file_stem()
                .and_then(|s| s.to_str())
                .and_then(|s| Uuid::parse_str(s).ok())
            else {
                continue;
            };
            let metadata = fs::metadata(&path)?;
            let timestamp = metadata
                .modified()
                .ok()
                .map(DateTime::<Utc>::from)
                .unwrap_or_else(Utc::now);
            by_workspace
                .entry(workspace_root.to_string())
                .or_default()
                .push(CleanupSession {
                    id,
                    file_path: path,
                    size_bytes: metadata.len(),
                    timestamp,
                });
        }

        Ok(())
    }
}

#[derive(Debug, Clone)]
struct CleanupSession {
    id: Uuid,
    file_path: PathBuf,
    size_bytes: u64,
    timestamp: DateTime<Utc>,
}

impl Default for SessionManager {
    fn default() -> Self {
        let home = home_dir_from_env()
            .unwrap_or_else(|_| std::env::temp_dir().to_string_lossy().into_owned());
        Self {
            sessions_dir: PathBuf::from(home).join(".talos").join("sessions"),
            index: Arc::new(Mutex::new(None)),
            store: Arc::new(CompactTextSessionStore),
            jsonl_store: Arc::new(JsonlSessionStore),
        }
    }
}

/// Resolve home directory env vars in cross-platform precedence.
///
/// Order:
/// 1. `HOME`
/// 2. `USERPROFILE`
/// 3. `HOMEDRIVE` + `HOMEPATH`
fn home_dir_from_env() -> Result<String, SessionError> {
    // Empty env vars should behave as unset vars so Windows fallback order still applies.
    home_dir_from_getter(|key| std::env::var(key).ok().filter(|value| !value.is_empty()))
}

fn home_dir_from_getter<F>(mut get_var: F) -> Result<String, SessionError>
where
    F: FnMut(&str) -> Option<String>,
{
    if let Some(home) = get_var("HOME") {
        return Ok(home);
    }
    if let Some(profile) = get_var("USERPROFILE") {
        return Ok(profile);
    }
    let drive = get_var("HOMEDRIVE").unwrap_or_default();
    let path = get_var("HOMEPATH").unwrap_or_default();
    if !drive.is_empty() && !path.is_empty() {
        return Ok(format!("{drive}{path}"));
    }
    Err(SessionError::IoError(std::io::Error::new(
        std::io::ErrorKind::NotFound,
        "home directory environment variable not found",
    )))
}

#[cfg(test)]
mod manager_env_tests {
    use super::home_dir_from_getter;

    #[test]
    fn home_dir_prefers_home() {
        let value = home_dir_from_getter(|key| match key {
            "HOME" => Some("/home/test".to_string()),
            "USERPROFILE" => Some("C:\\Users\\test".to_string()),
            _ => None,
        })
        .expect("HOME should be used");
        assert_eq!(value, "/home/test");
    }

    #[test]
    fn home_dir_falls_back_to_userprofile() {
        let value = home_dir_from_getter(|key| match key {
            "HOME" => None,
            "USERPROFILE" => Some("C:\\Users\\test".to_string()),
            _ => None,
        })
        .expect("USERPROFILE should be used");
        assert_eq!(value, "C:\\Users\\test");
    }

    #[test]
    fn home_dir_falls_back_to_drive_and_path() {
        let value = home_dir_from_getter(|key| match key {
            "HOME" => None,
            "USERPROFILE" => None,
            "HOMEDRIVE" => Some("C:".to_string()),
            "HOMEPATH" => Some("\\Users\\test".to_string()),
            _ => None,
        })
        .expect("HOMEDRIVE/HOMEPATH should be used");
        assert_eq!(value, "C:\\Users\\test");
    }

    #[test]
    fn home_dir_errors_when_all_missing() {
        let err = home_dir_from_getter(|_| None).expect_err("missing vars should error");
        assert!(matches!(err, crate::SessionError::IoError(_)));
    }
}