supercode_interchange/

catalog.rs

//! Discovery and durable addressing for sessions written by external harnesses.
//!
//! [`HarnessCatalog`] is intentionally about persisted state. It does not
//! claim that the process which wrote a session is still alive or attachable.

use std::collections::{HashMap, HashSet};
use std::fs::{self, File};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::{Component, Path, PathBuf};
use std::time::UNIX_EPOCH;

use rusqlite::Connection;
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::native_store::load_native_store_family;
use crate::session::percent_decode_path;
use crate::{Error, Fidelity, Result, Session, SessionFollower};

/// Extensible identifier for a coding harness.
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(transparent)]
pub struct HarnessId(pub String);

impl HarnessId {
    /// Claude Code's stable identifier.
    pub const CLAUDE_CODE: &'static str = "claude-code";
    /// Codex's stable identifier.
    pub const CODEX: &'static str = "codex";
    /// Pi's stable identifier.
    pub const PI: &'static str = "pi";
    /// OpenCode's stable identifier.
    pub const OPENCODE: &'static str = "opencode";
    /// Grok's stable identifier.
    pub const GROK: &'static str = "grok";
    /// Gemini CLI's stable identifier.
    pub const GEMINI: &'static str = "gemini";
    /// Goose's stable identifier.
    pub const GOOSE: &'static str = "goose";
    /// Supercode's native saved-session store.
    pub const SUPERCODE: &'static str = "supercode";

    /// Construct an identifier without restricting third-party harness names.
    pub fn new(value: impl Into<String>) -> Self {
        Self(value.into())
    }

    /// Return the identifier as a string slice.
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl From<&str> for HarnessId {
    fn from(value: &str) -> Self {
        Self::new(value)
    }
}

/// Durable storage address for a persisted session.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum StorageLocator {
    /// One session stored in one file.
    File {
        /// Absolute or caller-resolvable path to the transcript.
        path: PathBuf,
    },
    /// One logical session selected from a SQLite store.
    Sqlite {
        /// Path to the SQLite database.
        path: PathBuf,
        /// Harness-native stable selector, currently an OpenCode session id.
        selector: String,
    },
}

impl StorageLocator {
    /// Return the underlying file or database path.
    pub fn path(&self) -> &Path {
        match self {
            Self::File { path } | Self::Sqlite { path, .. } => path,
        }
    }
}

/// Stable identity for a persisted harness session.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct SessionLocator {
    /// Harness which owns the storage format.
    pub harness: HarnessId,
    /// Harness-native session identity.
    pub session_id: String,
    /// Exact storage address needed to load the session again.
    pub storage: StorageLocator,
}

/// Lightweight metadata returned by catalog discovery.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionDescriptor {
    /// Durable address accepted by [`HarnessCatalog::load`] and
    /// [`HarnessCatalog::follow`].
    pub locator: SessionLocator,
    /// Working directory recorded by the harness.
    pub cwd: Option<PathBuf>,
    /// Harness-provided title, when cheaply available.
    pub title: Option<String>,
    /// Oldest-first bounded conversation messages for a fallback topic when
    /// the harness does not publish a useful title. These are read only for
    /// the returned page and interpreted by the same presentation projection
    /// as an opened conversation.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub preview_candidates: Vec<SessionPreviewCandidate>,
    /// Newest-first bounded conversation messages for compact list previews.
    /// These are read only for the returned page, never for the entire
    /// catalog, and are interpreted by the same presentation projection as
    /// an opened conversation.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub latest_message_candidates: Vec<SessionPreviewCandidate>,
    /// Last update time as Unix epoch milliseconds.
    pub updated_at_ms: Option<u64>,
    /// Harness message-record count, when available without loading the session.
    pub message_count: Option<usize>,
    /// Model recorded in lightweight session metadata.
    pub model: Option<String>,
    /// Direct parent session for a harness-native child rollout. Ordinary
    /// conversation lists exclude these children, while tree/fidelity callers
    /// can request them explicitly without losing the native relationship.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub parent_session_id: Option<String>,
    /// Number of proven harness-native descendants represented by this root.
    /// Child identities remain behind the trusted catalog boundary until a
    /// caller explicitly requests this session family.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub child_session_count: usize,
}

/// One bounded normalized conversation-message candidate for list projection.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionPreviewCandidate {
    /// Canonical conversation role. Older clients may assume `user` when this
    /// field is absent from an older server.
    pub role: String,
    /// Canonical text content.
    pub content: String,
    /// Canonical provenance used by the normal conversation visibility rules.
    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
    pub metadata: HashMap<String, String>,
}

/// One stable newest-first discovery page.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DiscoveryPage {
    /// Sessions in this page.
    pub sessions: Vec<SessionDescriptor>,
    /// Opaque cursor for the next page, or `None` at the end.
    pub next_cursor: Option<String>,
}

/// Configurable session roots for the built-in harnesses.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct HarnessHomes {
    /// Directory containing Claude Code project session directories.
    pub claude_code: PathBuf,
    /// Directory containing Codex rollout sessions.
    pub codex: PathBuf,
    /// Directory containing Pi project session directories.
    pub pi: PathBuf,
    /// OpenCode data root, or an explicit `opencode*.db` path.
    pub opencode: PathBuf,
    /// Grok session root containing percent-encoded workspace directories.
    pub grok: PathBuf,
    /// Gemini CLI configuration root containing `projects.json` and `tmp/`.
    pub gemini: PathBuf,
    /// Goose `sessions.db`, or a directory containing it.
    pub goose: PathBuf,
    /// Supercode's native saved-session directory.
    pub supercode: PathBuf,
}

impl Default for HarnessHomes {
    fn default() -> Self {
        let home = std::env::var_os("HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| PathBuf::from("."));
        let claude_root = std::env::var_os("CLAUDE_CONFIG_DIR")
            .map(PathBuf::from)
            .unwrap_or_else(|| home.join(".claude"));
        let codex_root = std::env::var_os("CODEX_HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| home.join(".codex"));
        let pi = std::env::var_os("PI_CODING_AGENT_SESSION_DIR")
            .map(PathBuf::from)
            .unwrap_or_else(|| {
                std::env::var_os("PI_CODING_AGENT_DIR")
                    .map(PathBuf::from)
                    .unwrap_or_else(|| home.join(".pi/agent"))
                    .join("sessions")
            });
        let opencode = std::env::var_os("OPENCODE_DB")
            .map(PathBuf::from)
            .unwrap_or_else(|| {
                std::env::var_os("XDG_DATA_HOME")
                    .map(PathBuf::from)
                    .unwrap_or_else(|| home.join(".local/share"))
                    .join("opencode")
            });
        let grok = std::env::var_os("GROK_HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| home.join(".grok"))
            .join("sessions");
        let gemini = std::env::var_os("GEMINI_CLI_HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| home.join(".gemini"));
        let goose = std::env::var_os("GOOSE_PATH_ROOT")
            .map(PathBuf::from)
            .map(|root| root.join("data/sessions/sessions.db"))
            .unwrap_or_else(|| {
                #[cfg(target_os = "macos")]
                {
                    home.join("Library/Application Support/Block/goose/sessions/sessions.db")
                }
                #[cfg(target_os = "windows")]
                {
                    std::env::var_os("APPDATA")
                        .map(PathBuf::from)
                        .unwrap_or_else(|| home.join("AppData/Roaming"))
                        .join("Block/goose/sessions/sessions.db")
                }
                #[cfg(not(any(target_os = "macos", target_os = "windows")))]
                {
                    std::env::var_os("XDG_DATA_HOME")
                        .map(PathBuf::from)
                        .unwrap_or_else(|| home.join(".local/share"))
                        .join("goose/sessions/sessions.db")
                }
            });
        let supercode = std::env::var_os("SUPERCODE_HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| {
                std::env::var_os("XDG_CONFIG_HOME")
                    .map(PathBuf::from)
                    .unwrap_or_else(|| home.join(".config"))
                    .join("supercode")
            })
            .join("sessions");
        Self {
            claude_code: claude_root.join("projects"),
            codex: codex_root.join("sessions"),
            gemini,
            goose,
            supercode,
            pi,
            opencode,
            grok,
        }
    }
}

/// Filters and roots used for one catalog scan.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct DiscoveryQuery {
    /// Only return sessions whose recorded working directory is this path.
    pub workspace: Option<PathBuf>,
    /// Harnesses to scan. Empty means all built-ins.
    pub harnesses: Vec<HarnessId>,
    /// Storage roots to scan.
    pub homes: HarnessHomes,
    /// Case-insensitive search over harness, id, title, workspace, and model.
    pub query: Option<String>,
    /// Opaque cursor returned by a prior [`HarnessCatalog::discover_page`].
    pub cursor: Option<String>,
    /// Maximum number of results after newest-first sorting.
    pub limit: Option<usize>,
    /// Include oldest-first bounded topic candidates for harnesses whose
    /// native store does not publish a useful title. Off by default because
    /// topics are stable and list clients can retain them across refreshes.
    pub include_topic_candidates: bool,
    /// Include harness-native child rollouts such as Codex subagents. Off by
    /// default because they are parts of a parent conversation, not chats the
    /// user independently started. Translation/tree callers can opt in.
    pub include_child_sessions: bool,
    /// Restrict an explicit child-inclusive discovery to one root and every
    /// descendant linked to it by native lineage. Applied before pagination.
    pub root_session_id: Option<String>,
}

/// Read-only entry point for discovering, loading, and following persisted
/// harness sessions.
#[derive(Debug, Default, Clone, Copy)]
pub struct HarnessCatalog;

impl HarnessCatalog {
    /// Construct a catalog. It holds no cache or global mutable state.
    pub fn new() -> Self {
        Self
    }

    /// Discover sessions using lightweight headers/indexes rather than full
    /// transcript normalization. Malformed or concurrently-created entries
    /// are skipped without aborting the rest of the scan.
    pub fn discover(&self, query: &DiscoveryQuery) -> Result<Vec<SessionDescriptor>> {
        Ok(self.discover_page(query)?.sessions)
    }

    /// Discover one stable page and return the cursor for its successor.
    pub fn discover_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
        let selected: HashSet<&str> = if query.harnesses.is_empty() {
            [
                HarnessId::CLAUDE_CODE,
                HarnessId::CODEX,
                HarnessId::PI,
                HarnessId::OPENCODE,
                HarnessId::GROK,
                HarnessId::GEMINI,
                HarnessId::GOOSE,
                HarnessId::SUPERCODE,
            ]
            .into_iter()
            .collect()
        } else {
            query.harnesses.iter().map(HarnessId::as_str).collect()
        };
        let mut found = Vec::new();
        if selected.contains(HarnessId::CLAUDE_CODE) {
            discover_jsonl(
                &query.homes.claude_code,
                HarnessId::CLAUDE_CODE,
                query.workspace.as_deref(),
                query.include_child_sessions,
                &mut found,
            );
        }
        if selected.contains(HarnessId::CODEX) {
            discover_jsonl(
                &query.homes.codex,
                HarnessId::CODEX,
                query.workspace.as_deref(),
                query.include_child_sessions,
                &mut found,
            );
        }
        if selected.contains(HarnessId::PI) {
            discover_jsonl(
                &query.homes.pi,
                HarnessId::PI,
                query.workspace.as_deref(),
                query.include_child_sessions,
                &mut found,
            );
        }
        if selected.contains(HarnessId::OPENCODE) {
            discover_opencode(
                &query.homes.opencode,
                query.workspace.as_deref(),
                &mut found,
            );
        }
        if selected.contains(HarnessId::GROK) {
            discover_grok(&query.homes.grok, query.workspace.as_deref(), &mut found);
        }
        if selected.contains(HarnessId::GEMINI) {
            discover_gemini(&query.homes.gemini, query.workspace.as_deref(), &mut found);
        }
        if selected.contains(HarnessId::GOOSE) {
            discover_goose(&query.homes.goose, query.workspace.as_deref(), &mut found);
        }
        if selected.contains(HarnessId::SUPERCODE) {
            discover_supercode(
                &query.homes.supercode,
                query.workspace.as_deref(),
                &mut found,
            );
        }
        roll_up_session_children(&mut found, query.include_child_sessions);
        if let Some(root_session_id) = query.root_session_id.as_deref() {
            retain_session_family(&mut found, root_session_id);
        }
        found.sort_by(|a, b| {
            b.updated_at_ms
                .cmp(&a.updated_at_ms)
                .then_with(|| a.locator.harness.cmp(&b.locator.harness))
                .then_with(|| a.locator.session_id.cmp(&b.locator.session_id))
        });
        if let Some(search) = query
            .query
            .as_deref()
            .map(str::trim)
            .filter(|q| !q.is_empty())
        {
            let search = search.to_lowercase();
            found.retain(|descriptor| descriptor_matches(descriptor, &search));
        }
        let start = match query.cursor.as_deref() {
            Some(cursor) => {
                let key = decode_cursor(cursor)?;
                found
                    .iter()
                    .position(|descriptor| descriptor_cursor_key(descriptor) == key)
                    .map(|index| index + 1)
                    .ok_or_else(|| Error::Other("discovery cursor is stale or invalid".into()))?
            }
            None => 0,
        };
        let end = query
            .limit
            .map(|limit| start.saturating_add(limit).min(found.len()))
            .unwrap_or(found.len());
        let mut sessions = found[start.min(found.len())..end].to_vec();
        let codex_topics = if query.include_topic_candidates {
            codex_history_topics(&query.homes.codex, &sessions).unwrap_or_default()
        } else {
            HashMap::new()
        };
        for descriptor in &mut sessions {
            if query.include_topic_candidates {
                descriptor.preview_candidates =
                    if descriptor.locator.harness.as_str() == HarnessId::CODEX {
                        codex_topics
                            .get(&descriptor.locator.session_id)
                            .cloned()
                            .unwrap_or_else(|| {
                                topic_message_candidates(&descriptor.locator).unwrap_or_default()
                            })
                    } else {
                        topic_message_candidates(&descriptor.locator).unwrap_or_default()
                    };
            }
            descriptor.latest_message_candidates =
                latest_message_candidates(&descriptor.locator).unwrap_or_default();
        }
        let next_cursor = (end < found.len())
            .then(|| sessions.last().map(encode_cursor))
            .flatten();
        Ok(DiscoveryPage {
            sessions,
            next_cursor,
        })
    }

    /// Refresh one file-backed descriptor without rescanning its native store.
    ///
    /// This is the incremental counterpart to [`Self::discover_page`]: a
    /// filesystem notification is only an invalidation hint, so callers
    /// re-read the durable file and derive the complete current descriptor.
    /// `None` means the path disappeared or no longer contains a recognizable
    /// session. SQLite-backed harnesses retain their indexed discovery path.
    pub fn refresh_file_descriptor(
        &self,
        locator: &SessionLocator,
        workspace: Option<&Path>,
        include_topic_candidates: bool,
    ) -> Result<Option<SessionDescriptor>> {
        let StorageLocator::File { path } = &locator.storage else {
            return Ok(None);
        };
        if !matches!(
            locator.harness.as_str(),
            HarnessId::CLAUDE_CODE | HarnessId::CODEX
        ) {
            return Ok(None);
        }
        if !path.is_file() {
            return Ok(None);
        }
        let Ok(meta) = read_header(path, locator.harness.as_str()) else {
            // Harnesses append the header and first turn non-atomically. A
            // transiently incomplete new file is not a service error; the
            // next native event or reconciliation pass will retry it.
            return Ok(None);
        };
        if workspace.is_some_and(|wanted| {
            meta.cwd
                .as_deref()
                .is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
        }) {
            return Ok(None);
        }
        let mut descriptor = SessionDescriptor {
            locator: SessionLocator {
                harness: locator.harness.clone(),
                session_id: meta
                    .session_id
                    .unwrap_or_else(|| locator.session_id.clone()),
                storage: StorageLocator::File { path: path.clone() },
            },
            cwd: meta.cwd,
            title: meta.title,
            preview_candidates: Vec::new(),
            latest_message_candidates: Vec::new(),
            updated_at_ms: modified_ms(path),
            message_count: None,
            model: meta.model,
            parent_session_id: meta.parent_session_id,
            child_session_count: 0,
        };
        if include_topic_candidates {
            descriptor.preview_candidates =
                topic_message_candidates(&descriptor.locator).unwrap_or_default();
        }
        descriptor.latest_message_candidates =
            latest_message_candidates(&descriptor.locator).unwrap_or_default();
        Ok(Some(descriptor))
    }

    /// Load the complete normalized session named by a durable locator.
    pub fn load(&self, locator: &SessionLocator) -> Result<Session> {
        self.load_with_fidelity(locator, Fidelity::ByteLossless)
    }

    /// [`Self::load`] at a declared fidelity.
    ///
    /// Read-only surfaces (a session mirror, `follow`) pass
    /// [`Fidelity::Semantic`] so a compacted transcript renders instead of
    /// erroring; every continuation/transfer/export caller keeps the strict
    /// default. See [`Session::load_with_fidelity`].
    pub fn load_with_fidelity(
        &self,
        locator: &SessionLocator,
        fidelity: Fidelity,
    ) -> Result<Session> {
        match &locator.storage {
            StorageLocator::File { path } => {
                if let Some(session) = load_native_store_family(path)? {
                    Ok(session)
                } else {
                    Ok(Session::load_with_fidelity(path, fidelity)?)
                }
            }
            StorageLocator::Sqlite { path, selector } => {
                if locator.harness.as_str() == HarnessId::GOOSE {
                    Ok(Session::from_goose_sqlite(path, selector)?)
                } else {
                    Ok(Session::from_opencode_sqlite(path, Some(selector))?)
                }
            }
        }
    }

    /// Load the selected parent transcript without recursively attaching
    /// Claude Code child sessions. This is the bounded frontend-view seam;
    /// lossless operations continue to use [`Self::load_with_fidelity`].
    #[doc(hidden)]
    pub fn load_parent_with_fidelity(
        &self,
        locator: &SessionLocator,
        fidelity: Fidelity,
    ) -> Result<Session> {
        match &locator.storage {
            StorageLocator::File { path } => {
                if let Some(session) = load_native_store_family(path)? {
                    Ok(session)
                } else {
                    Ok(Session::load_parent_with_fidelity(path, fidelity)?)
                }
            }
            StorageLocator::Sqlite { path, selector } => {
                if locator.harness.as_str() == HarnessId::GOOSE {
                    Ok(Session::from_goose_sqlite(path, selector)?)
                } else {
                    Ok(Session::from_opencode_sqlite(path, Some(selector))?)
                }
            }
        }
    }

    /// Load bounded parent-only human-visible history. Codex compaction
    /// changes resumable context but does not erase earlier visible turns.
    #[doc(hidden)]
    pub fn load_display_view(
        &self,
        locator: &SessionLocator,
        fidelity: Fidelity,
        message_limit: usize,
    ) -> Result<Session> {
        match &locator.storage {
            StorageLocator::File { path } => {
                if let Some(mut session) = load_native_store_family(path)? {
                    if session.messages.len() > message_limit.max(1) {
                        session
                            .messages
                            .drain(..session.messages.len() - message_limit.max(1));
                    }
                    Ok(session)
                } else {
                    Ok(Session::load_display_view(path, fidelity, message_limit)?)
                }
            }
            StorageLocator::Sqlite { path, selector } => {
                let mut session = if locator.harness.as_str() == HarnessId::GOOSE {
                    Session::from_goose_sqlite_display(path, selector, message_limit)?
                } else {
                    Session::from_opencode_sqlite(path, Some(selector))?
                };
                if session.messages.len() > message_limit.max(1) {
                    session
                        .messages
                        .drain(..session.messages.len() - message_limit.max(1));
                }
                Ok(session)
            }
        }
    }

    /// Open a passive change-triggered follower for a durable locator.
    pub fn follow(&self, locator: &SessionLocator) -> Result<SessionFollower> {
        self.follow_with_fidelity(locator, Fidelity::ByteLossless)
    }

    /// [`Self::follow`] at a declared fidelity — see [`Self::load_with_fidelity`].
    pub fn follow_with_fidelity(
        &self,
        locator: &SessionLocator,
        fidelity: Fidelity,
    ) -> Result<SessionFollower> {
        SessionFollower::open_locator_with_fidelity(locator, fidelity)
    }

    /// Follow a read-only view with explicit child-tree and history bounds.
    #[doc(hidden)]
    pub fn follow_read_view(
        &self,
        locator: &SessionLocator,
        fidelity: Fidelity,
        include_subagents: bool,
        message_limit: Option<usize>,
        max_message_chars: Option<usize>,
        display_history: bool,
    ) -> Result<SessionFollower> {
        SessionFollower::open_locator_with_view(
            locator,
            fidelity,
            include_subagents,
            message_limit,
            max_message_chars,
            display_history,
        )
    }
}

fn descriptor_matches(descriptor: &SessionDescriptor, search: &str) -> bool {
    [
        Some(descriptor.locator.harness.as_str()),
        Some(descriptor.locator.session_id.as_str()),
        descriptor.title.as_deref(),
        descriptor.cwd.as_ref().and_then(|path| path.to_str()),
        descriptor.model.as_deref(),
    ]
    .into_iter()
    .flatten()
    .any(|value| value.to_lowercase().contains(search))
}

fn descriptor_cursor_key(descriptor: &SessionDescriptor) -> (Option<u64>, String, String) {
    (
        descriptor.updated_at_ms,
        descriptor.locator.harness.as_str().to_string(),
        descriptor.locator.session_id.clone(),
    )
}

fn encode_cursor(descriptor: &SessionDescriptor) -> String {
    let json = serde_json::to_vec(&descriptor_cursor_key(descriptor)).unwrap_or_default();
    let mut encoded = String::with_capacity(json.len() * 2);
    for byte in json {
        use std::fmt::Write;
        let _ = write!(&mut encoded, "{byte:02x}");
    }
    encoded
}

fn decode_cursor(cursor: &str) -> Result<(Option<u64>, String, String)> {
    if cursor.len() % 2 != 0 {
        return Err(Error::Other("discovery cursor is invalid".into()));
    }
    let bytes = (0..cursor.len())
        .step_by(2)
        .map(|index| u8::from_str_radix(&cursor[index..index + 2], 16))
        .collect::<std::result::Result<Vec<_>, _>>()
        .map_err(|_| Error::Other("discovery cursor is invalid".into()))?;
    serde_json::from_slice(&bytes).map_err(|_| Error::Other("discovery cursor is invalid".into()))
}

#[derive(Default)]
struct HeaderMeta {
    session_id: Option<String>,
    cwd: Option<PathBuf>,
    title: Option<String>,
    model: Option<String>,
    parent_session_id: Option<String>,
}

fn discover_jsonl(
    root: &Path,
    harness: &str,
    workspace: Option<&Path>,
    include_child_sessions: bool,
    found: &mut Vec<SessionDescriptor>,
) {
    let mut files = Vec::new();
    collect_jsonl(root, harness, include_child_sessions, &mut files);
    for path in files {
        let Ok(meta) = read_header(&path, harness) else {
            continue;
        };
        if workspace.is_some_and(|wanted| {
            meta.cwd
                .as_deref()
                .is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
        }) {
            continue;
        }
        let session_id = meta.session_id.unwrap_or_else(|| {
            path.file_stem()
                .and_then(|value| value.to_str())
                .unwrap_or("unknown")
                .to_string()
        });
        let parent_session_id = meta.parent_session_id.or_else(|| {
            (harness == HarnessId::CLAUDE_CODE)
                .then(|| claude_subagent_parent_id(&path))
                .flatten()
        });
        found.push(SessionDescriptor {
            locator: SessionLocator {
                harness: HarnessId::new(harness),
                session_id,
                storage: StorageLocator::File { path: path.clone() },
            },
            cwd: meta.cwd,
            title: meta.title,
            preview_candidates: Vec::new(),
            latest_message_candidates: Vec::new(),
            updated_at_ms: modified_ms(&path),
            message_count: None,
            model: meta.model,
            parent_session_id,
            child_session_count: if harness == HarnessId::CLAUDE_CODE && !include_child_sessions {
                count_claude_subagents(&path)
            } else {
                0
            },
        });
    }
}

fn collect_jsonl(root: &Path, harness: &str, include_child_sessions: bool, out: &mut Vec<PathBuf>) {
    let Ok(entries) = fs::read_dir(root) else {
        return;
    };
    for entry in entries.flatten() {
        let Ok(kind) = entry.file_type() else {
            continue;
        };
        let path = entry.path();
        if kind.is_dir() {
            if harness == HarnessId::CLAUDE_CODE
                && path.file_name().and_then(|v| v.to_str()) == Some("subagents")
                && !include_child_sessions
            {
                continue;
            }
            collect_jsonl(&path, harness, include_child_sessions, out);
        } else if kind.is_file() && path.extension().and_then(|v| v.to_str()) == Some("jsonl") {
            out.push(path);
        }
    }
}

fn read_header(path: &Path, harness: &str) -> Result<HeaderMeta> {
    let file = File::open(path)?;
    let mut result = HeaderMeta::default();
    let mut bytes = 0usize;
    for line in BufReader::new(file).lines().take(32) {
        let line = line?;
        bytes += line.len();
        if bytes > 256 * 1024 {
            break;
        }
        let Ok(value) = serde_json::from_str::<Value>(&line) else {
            continue;
        };
        update_header_meta(&mut result, &value, harness);
        if result.session_id.is_some() && result.cwd.is_some() && result.model.is_some() {
            break;
        }
    }
    if result.session_id.is_none() && result.cwd.is_none() {
        return Err(Error::Other(format!(
            "{} has no recognizable {harness} session header",
            path.display()
        )));
    }
    Ok(result)
}

fn update_header_meta(result: &mut HeaderMeta, value: &Value, harness: &str) {
    match harness {
        HarnessId::CLAUDE_CODE => {
            fill_string(&mut result.session_id, value.get("sessionId"));
            fill_path(&mut result.cwd, value.get("cwd"));
            fill_string(
                &mut result.model,
                value.get("message").and_then(|v| v.get("model")),
            );
        }
        HarnessId::CODEX => {
            let payload = value.get("payload").unwrap_or(&Value::Null);
            if value.get("type").and_then(Value::as_str) == Some("session_meta") {
                fill_string(&mut result.session_id, payload.get("id"));
                fill_path(&mut result.cwd, payload.get("cwd"));
                fill_string(&mut result.title, payload.get("thread_name"));
                fill_string(&mut result.title, payload.get("title"));
                fill_string(
                    &mut result.parent_session_id,
                    payload.get("parent_thread_id"),
                );
                if let Some(parent) = payload
                    .pointer("/source/subagent/thread_spawn/parent_thread_id")
                    .and_then(Value::as_str)
                {
                    result.parent_session_id = Some(parent.to_string());
                }
                if result.title.is_none() {
                    result.title = payload
                        .pointer("/source/subagent/thread_spawn/agent_path")
                        .and_then(Value::as_str)
                        .and_then(|path| path.rsplit('/').find(|part| !part.is_empty()))
                        .map(humanize_topic);
                }
            }
            if value.get("type").and_then(Value::as_str) == Some("turn_context") {
                fill_path(&mut result.cwd, payload.get("cwd"));
                fill_string(&mut result.model, payload.get("model"));
            }
        }
        HarnessId::PI => {
            if value.get("type").and_then(Value::as_str) == Some("session") {
                fill_string(&mut result.session_id, value.get("id"));
                fill_path(&mut result.cwd, value.get("cwd"));
            }
            fill_string(
                &mut result.model,
                value.get("message").and_then(|v| v.get("model")),
            );
        }
        _ => {}
    }
}

/// Collapse native child rollouts into their root conversation before sorting
/// and pagination. A child's write time contributes to the root so active
/// delegated work keeps the conversation visible without creating extra rows.
fn roll_up_session_children(found: &mut Vec<SessionDescriptor>, include_children: bool) {
    let by_id = found
        .iter()
        .enumerate()
        .map(|(index, descriptor)| {
            (
                (
                    descriptor.locator.harness.as_str().to_string(),
                    descriptor.locator.session_id.clone(),
                ),
                index,
            )
        })
        .collect::<HashMap<_, _>>();
    let mut root_updates = HashMap::<usize, u64>::new();
    let mut root_child_counts = HashMap::<usize, usize>::new();

    for descriptor in found.iter() {
        let Some(mut parent_id) = descriptor.parent_session_id.as_deref() else {
            continue;
        };
        let harness = descriptor.locator.harness.as_str();
        let mut root = None;
        let mut visited = HashSet::new();
        while visited.insert(parent_id.to_string()) {
            let Some(&parent_index) = by_id.get(&(harness.to_string(), parent_id.to_string()))
            else {
                break;
            };
            root = Some(parent_index);
            let Some(next_parent) = found[parent_index].parent_session_id.as_deref() else {
                break;
            };
            parent_id = next_parent;
        }
        if let (Some(root), Some(updated_at_ms)) = (root, descriptor.updated_at_ms) {
            root_updates
                .entry(root)
                .and_modify(|current| *current = (*current).max(updated_at_ms))
                .or_insert(updated_at_ms);
        }
        if let Some(root) = root {
            *root_child_counts.entry(root).or_default() += 1;
        }
    }

    for (root, child_updated_at_ms) in root_updates {
        found[root].updated_at_ms = Some(
            found[root]
                .updated_at_ms
                .unwrap_or_default()
                .max(child_updated_at_ms),
        );
    }
    for (root, child_count) in root_child_counts {
        found[root].child_session_count = child_count;
    }
    if !include_children {
        found.retain(|descriptor| descriptor.parent_session_id.is_none());
    }
}

fn retain_session_family(found: &mut Vec<SessionDescriptor>, root_session_id: &str) {
    let parent_by_id = found
        .iter()
        .map(|descriptor| {
            (
                descriptor.locator.session_id.clone(),
                descriptor.parent_session_id.clone(),
            )
        })
        .collect::<HashMap<_, _>>();
    found.retain(|descriptor| {
        let mut current = descriptor.locator.session_id.clone();
        let mut visited = HashSet::new();
        while visited.insert(current.clone()) {
            if current == root_session_id {
                return true;
            }
            let Some(Some(parent)) = parent_by_id.get(&current) else {
                return false;
            };
            current = parent.clone();
        }
        false
    });
}

fn claude_subagent_parent_id(path: &Path) -> Option<String> {
    let subagents = path.parent()?;
    if subagents.file_name()?.to_str()? != "subagents" {
        return None;
    }
    subagents
        .parent()?
        .file_name()?
        .to_str()
        .map(str::to_string)
}

fn count_claude_subagents(parent_path: &Path) -> usize {
    let Some(parent) = parent_path.parent() else {
        return 0;
    };
    let Some(stem) = parent_path.file_stem() else {
        return 0;
    };
    let root = parent.join(stem).join("subagents");
    let mut files = Vec::new();
    collect_jsonl(&root, HarnessId::CLAUDE_CODE, true, &mut files);
    files.len()
}

fn is_zero(value: &usize) -> bool {
    *value == 0
}

fn humanize_topic(value: &str) -> String {
    let text = value.replace(['_', '-'], " ");
    let mut characters = text.chars();
    match characters.next() {
        Some(first) => first.to_uppercase().collect::<String>() + characters.as_str(),
        None => text,
    }
}

fn discover_gemini(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
    let slug_to_cwd = std::fs::read_to_string(root.join("projects.json"))
        .ok()
        .and_then(|text| serde_json::from_str::<Value>(&text).ok())
        .and_then(|value| value.get("projects").and_then(Value::as_object).cloned())
        .map(|projects| {
            projects
                .into_iter()
                .filter_map(|(cwd, slug)| Some((slug.as_str()?.to_string(), PathBuf::from(cwd))))
                .collect::<HashMap<_, _>>()
        })
        .unwrap_or_default();
    let mut files = Vec::new();
    collect_jsonl(&root.join("tmp"), HarnessId::GEMINI, false, &mut files);
    let worker_count = std::thread::available_parallelism()
        .map(usize::from)
        .unwrap_or(4)
        .clamp(1, 8)
        .min(files.len().max(1));
    let chunk_size = files.len().max(1).div_ceil(worker_count);
    let discovered = std::thread::scope(|scope| {
        files
            .chunks(chunk_size)
            .map(|paths| {
                scope.spawn(|| {
                    paths
                        .iter()
                        .filter_map(|path| gemini_descriptor(path, &slug_to_cwd, workspace))
                        .collect::<Vec<_>>()
                })
            })
            .collect::<Vec<_>>()
            .into_iter()
            .flat_map(|worker| {
                worker
                    .join()
                    .expect("Gemini discovery worker must not panic")
            })
            .collect::<Vec<_>>()
    });
    found.extend(discovered);
}

fn gemini_descriptor(
    path: &Path,
    slug_to_cwd: &HashMap<String, PathBuf>,
    workspace: Option<&Path>,
) -> Option<SessionDescriptor> {
    if path
        .parent()
        .and_then(Path::file_name)
        .and_then(|name| name.to_str())
        != Some("chats")
    {
        return None;
    }
    let slug = path
        .parent()
        .and_then(Path::parent)
        .and_then(Path::file_name)
        .and_then(|name| name.to_str());
    let cwd = slug.and_then(|slug| slug_to_cwd.get(slug)).cloned();
    if workspace.is_some_and(|wanted| {
        cwd.as_deref()
            .is_none_or(|actual| !recorded_cwd_matches(actual, wanted))
    }) {
        return None;
    }

    // The native session id lives on line one. A small decoration budget keeps
    // the common title/model case without turning 1,800 sessions into a
    // sequential 60 MiB read before the list can render.
    let file = File::open(path).ok()?;
    let mut reader = BufReader::new(file.take(64 * 1024));
    let mut header = String::new();
    reader.read_line(&mut header).ok()?;
    let header = serde_json::from_str::<Value>(&header).ok()?;
    let session_id = header.get("sessionId")?.as_str()?.to_string();
    let mut model = None;
    for line in reader
        .take(4 * 1024)
        .lines()
        .map_while(std::result::Result::ok)
    {
        let Ok(value) = serde_json::from_str::<Value>(&line) else {
            continue;
        };
        let kind = value.get("type").and_then(Value::as_str);
        if kind != Some("user") && kind != Some("gemini") {
            continue;
        }
        if model.is_none() {
            model = value
                .get("model")
                .and_then(Value::as_str)
                .map(str::to_string);
        }
        if model.is_some() {
            break;
        }
    }
    Some(SessionDescriptor {
        locator: SessionLocator {
            harness: HarnessId::from(HarnessId::GEMINI),
            session_id,
            storage: StorageLocator::File {
                path: path.to_path_buf(),
            },
        },
        cwd,
        title: None,
        preview_candidates: Vec::new(),
        latest_message_candidates: Vec::new(),
        updated_at_ms: modified_ms(path),
        message_count: None,
        model,
        parent_session_id: None,
        child_session_count: 0,
    })
}

fn display_text(content: Option<&Value>) -> Option<String> {
    match content? {
        Value::String(text) => Some(text.clone()),
        Value::Array(parts) => Some(
            parts
                .iter()
                .filter_map(|part| part.get("text").and_then(Value::as_str))
                .collect::<Vec<_>>()
                .join(" ")
                .trim()
                .to_string(),
        ),
        _ => None,
    }
}

fn discover_supercode(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
    for info in list_native_store(root) {
        let path = if info.archived {
            root.join("archived").join(format!("{}.jsonl", info.name))
        } else {
            root.join(format!("{}.jsonl", info.name))
        };
        let header = read_native_store_header(&path);
        if workspace.is_some_and(|wanted| {
            header
                .as_ref()
                .and_then(|meta| meta.cwd.as_deref())
                .is_none_or(|cwd| !recorded_cwd_matches(cwd, wanted))
        }) {
            continue;
        }
        let title = (!info.title.trim().is_empty()).then_some(info.title);
        let updated_at_ms =
            modified_ms(&path).or_else(|| modified_ms(&path.with_extension("sidecar.jsonl")));
        found.push(SessionDescriptor {
            locator: SessionLocator {
                harness: HarnessId::from(HarnessId::SUPERCODE),
                session_id: info.name,
                storage: StorageLocator::File { path: path.clone() },
            },
            cwd: header.as_ref().and_then(|meta| meta.cwd.clone()),
            title,
            preview_candidates: Vec::new(),
            latest_message_candidates: Vec::new(),
            updated_at_ms,
            message_count: None,
            model: header.and_then(|meta| meta.model),
            parent_session_id: None,
            child_session_count: 0,
        });
    }
}

/// Read only the bounded native envelope needed by discovery. Loading a
/// sidecar-backed session here used to deserialize the complete byte-lossless
/// transcript family, making a workspace list proportional to every saved
/// Supercode transcript on the machine.
fn read_native_store_header(path: &Path) -> Option<HeaderMeta> {
    let name = path.file_stem()?.to_str()?;
    let sidecar = path.with_file_name(format!("{name}.sidecar.jsonl"));
    let source_path = if sidecar.is_file() {
        sidecar
    } else {
        path.to_path_buf()
    };
    let file = File::open(source_path).ok()?;
    let mut result = HeaderMeta::default();
    let mut source = None;
    let mut bytes = 0usize;
    for line in BufReader::new(file).lines().take(32) {
        let line = line.ok()?;
        bytes += line.len();
        if bytes > 256 * 1024 {
            break;
        }
        let Ok(value) = serde_json::from_str::<Value>(&line) else {
            continue;
        };
        if source.is_none() {
            source = value.get("source").and_then(Value::as_str).map(|source| {
                if source == "claude_code" {
                    HarnessId::CLAUDE_CODE.to_string()
                } else {
                    source.to_string()
                }
            });
            fill_string(&mut result.session_id, value.get("session_id"));
        }
        if let Some(harness) = source.as_deref() {
            update_header_meta(&mut result, &value, harness);
        }
        if result.cwd.is_some() && result.model.is_some() {
            break;
        }
    }
    Some(result)
}

#[derive(Deserialize)]
struct NativeStoreInfo {
    name: String,
    #[serde(default)]
    title: String,
    #[serde(skip)]
    archived: bool,
}

fn list_native_store(root: &Path) -> Vec<NativeStoreInfo> {
    let mut sessions = Vec::new();
    for archived in [false, true] {
        let directory = if archived {
            root.join("archived")
        } else {
            root.to_path_buf()
        };
        let Ok(entries) = fs::read_dir(directory) else {
            continue;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if !path.to_string_lossy().ends_with(".meta.json") {
                continue;
            }
            let Ok(text) = fs::read_to_string(path) else {
                continue;
            };
            let Ok(mut info) = serde_json::from_str::<NativeStoreInfo>(&text) else {
                continue;
            };
            info.archived = archived;
            sessions.push(info);
        }
    }
    sessions.sort_by(|left, right| left.name.cmp(&right.name));
    sessions
}

fn discover_grok(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
    let Ok(workspaces) = fs::read_dir(root) else {
        return;
    };
    for workspace_entry in workspaces.flatten() {
        let encoded = workspace_entry.file_name();
        let Some(cwd) = encoded
            .to_str()
            .and_then(percent_decode_path)
            .map(PathBuf::from)
        else {
            continue;
        };
        if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
            continue;
        }
        let Ok(sessions) = fs::read_dir(workspace_entry.path()) else {
            continue;
        };
        for session_entry in sessions.flatten() {
            let session_dir = session_entry.path();
            if !session_dir.is_dir() {
                continue;
            }
            let transcript = session_dir.join("chat_history.jsonl");
            if !transcript.is_file() {
                continue;
            }
            let Some(session_id) = session_dir
                .file_name()
                .and_then(|name| name.to_str())
                .map(str::to_string)
            else {
                continue;
            };
            let summary = fs::read_to_string(session_dir.join("summary.json"))
                .ok()
                .and_then(|text| serde_json::from_str::<Value>(&text).ok());
            let title = summary
                .as_ref()
                .and_then(|value| value.get("generated_title"))
                .and_then(Value::as_str)
                .filter(|title| !title.is_empty())
                .map(str::to_string);
            let model = summary
                .as_ref()
                .and_then(|value| value.get("current_model_id"))
                .and_then(Value::as_str)
                .map(str::to_string);
            let message_count = summary
                .as_ref()
                .and_then(|value| value.get("num_chat_messages"))
                .and_then(Value::as_u64)
                .and_then(|count| usize::try_from(count).ok());
            let updated_at_ms = summary
                .as_ref()
                .and_then(|value| value.get("updated_at"))
                .and_then(Value::as_str)
                .and_then(crate::sidecar::rfc3339_to_ms)
                .and_then(|millis| u64::try_from(millis).ok())
                .or_else(|| modified_ms(&transcript));
            found.push(SessionDescriptor {
                locator: SessionLocator {
                    harness: HarnessId::from(HarnessId::GROK),
                    session_id,
                    storage: StorageLocator::File { path: transcript },
                },
                cwd: Some(cwd.clone()),
                title,
                preview_candidates: Vec::new(),
                latest_message_candidates: Vec::new(),
                updated_at_ms,
                message_count,
                model,
                parent_session_id: None,
                child_session_count: 0,
            });
        }
    }
}

fn discover_opencode(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
    let mut dbs = Vec::new();
    if root.is_file() {
        dbs.push(root.to_path_buf());
    } else if let Ok(entries) = fs::read_dir(root) {
        dbs.extend(entries.flatten().map(|entry| entry.path()).filter(|path| {
            path.file_name()
                .and_then(|v| v.to_str())
                .is_some_and(|name| name.starts_with("opencode") && name.ends_with(".db"))
        }));
    }
    dbs.sort();
    for db in dbs {
        let Ok(conn) = Connection::open_with_flags(
            &db,
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
        ) else {
            continue;
        };
        let has_model = conn.prepare("SELECT model FROM session LIMIT 0").is_ok();
        let model_column = if has_model { "s.model" } else { "NULL" };
        let query = format!(
            "SELECT s.id, s.directory, s.title, s.time_updated, {model_column}, COUNT(m.id) \
             FROM session s LEFT JOIN message m ON m.session_id = s.id \
             GROUP BY s.id ORDER BY s.time_updated DESC"
        );
        let Ok(mut stmt) = conn.prepare(&query) else {
            continue;
        };
        let Ok(rows) = stmt.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, i64>(3)?,
                row.get::<_, Option<String>>(4)?,
                row.get::<_, i64>(5)?,
            ))
        }) else {
            continue;
        };
        for row in rows.flatten() {
            let (id, cwd, title, updated, model, messages) = row;
            let cwd = PathBuf::from(cwd);
            if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
                continue;
            }
            found.push(SessionDescriptor {
                locator: SessionLocator {
                    harness: HarnessId::from(HarnessId::OPENCODE),
                    session_id: id.clone(),
                    storage: StorageLocator::Sqlite {
                        path: db.clone(),
                        selector: id,
                    },
                },
                cwd: Some(cwd),
                title: (!title.is_empty()).then_some(title),
                preview_candidates: Vec::new(),
                latest_message_candidates: Vec::new(),
                updated_at_ms: u64::try_from(updated).ok(),
                message_count: usize::try_from(messages).ok(),
                model,
                parent_session_id: None,
                child_session_count: 0,
            });
        }
    }
}

fn discover_goose(root: &Path, workspace: Option<&Path>, found: &mut Vec<SessionDescriptor>) {
    let db = if root.is_file() {
        root.to_path_buf()
    } else if root.join("sessions.db").is_file() {
        root.join("sessions.db")
    } else {
        root.join("sessions/sessions.db")
    };
    let Ok(connection) = Connection::open_with_flags(
        &db,
        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
    ) else {
        return;
    };
    let Ok(mut statement) = connection.prepare(
        "SELECT s.id, s.working_dir, s.name, s.updated_at, s.model_config_json, \
                COUNT(m.id) \
         FROM sessions s LEFT JOIN messages m ON m.session_id = s.id \
         WHERE s.archived_at IS NULL \
         GROUP BY s.id ORDER BY s.updated_at DESC",
    ) else {
        return;
    };
    let Ok(rows) = statement.query_map([], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, String>(3)?,
            row.get::<_, Option<String>>(4)?,
            row.get::<_, i64>(5)?,
        ))
    }) else {
        return;
    };
    for row in rows.flatten() {
        let (id, cwd, title, updated_at, model_config, message_count) = row;
        let cwd = PathBuf::from(cwd);
        if workspace.is_some_and(|wanted| !recorded_cwd_matches(&cwd, wanted)) {
            continue;
        }
        let model = model_config
            .as_deref()
            .and_then(|value| serde_json::from_str::<Value>(value).ok())
            .and_then(|value| {
                value
                    .get("model_name")
                    .or_else(|| value.get("modelName"))
                    .and_then(Value::as_str)
                    .map(str::to_string)
            });
        let updated_at_ms = crate::sidecar::rfc3339_to_ms(&updated_at)
            .or_else(|| {
                // SQLite's CURRENT_TIMESTAMP uses `YYYY-MM-DD HH:MM:SS`.
                crate::sidecar::rfc3339_to_ms(&format!("{}Z", updated_at.replace(' ', "T")))
            })
            .and_then(|value| u64::try_from(value).ok());
        found.push(SessionDescriptor {
            locator: SessionLocator {
                harness: HarnessId::from(HarnessId::GOOSE),
                session_id: id.clone(),
                storage: StorageLocator::Sqlite {
                    path: db.clone(),
                    selector: id,
                },
            },
            cwd: Some(cwd),
            title: (!title.trim().is_empty()).then_some(title),
            preview_candidates: Vec::new(),
            latest_message_candidates: Vec::new(),
            updated_at_ms,
            message_count: usize::try_from(message_count).ok(),
            model,
            parent_session_id: None,
            child_session_count: 0,
        });
    }
}

const LATEST_PREVIEW_CANDIDATES: usize = 8;
const TOPIC_PREVIEW_HEAD_BYTES: u64 = 512 * 1024;
const LATEST_PREVIEW_TAIL_BYTES: u64 = 512 * 1024;
const LATEST_PREVIEW_MAX_BYTES: u64 = 4 * 1024 * 1024;

fn topic_message_candidates(locator: &SessionLocator) -> Result<Vec<SessionPreviewCandidate>> {
    match &locator.storage {
        StorageLocator::File { path }
            if matches!(
                locator.harness.as_str(),
                HarnessId::CLAUDE_CODE | HarnessId::CODEX
            ) =>
        {
            topic_file_message_candidates(path, locator.harness.as_str())
        }
        _ => Ok(Vec::new()),
    }
}

fn codex_history_topics(
    sessions_root: &Path,
    sessions: &[SessionDescriptor],
) -> Result<HashMap<String, Vec<SessionPreviewCandidate>>> {
    let wanted: HashSet<&str> = sessions
        .iter()
        .filter(|descriptor| descriptor.locator.harness.as_str() == HarnessId::CODEX)
        .map(|descriptor| descriptor.locator.session_id.as_str())
        .collect();
    if wanted.is_empty() {
        return Ok(HashMap::new());
    }
    let Some(root) = sessions_root.parent() else {
        return Ok(HashMap::new());
    };
    let file = match File::open(root.join("history.jsonl")) {
        Ok(file) => file,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(HashMap::new()),
        Err(error) => return Err(error.into()),
    };
    let mut topics = HashMap::new();
    for line in BufReader::new(file).lines() {
        let Ok(value) = serde_json::from_str::<Value>(&line?) else {
            continue;
        };
        let Some(session_id) = value.get("session_id").and_then(Value::as_str) else {
            continue;
        };
        if !wanted.contains(session_id) || topics.contains_key(session_id) {
            continue;
        }
        let mut candidates = Vec::new();
        push_message_candidate(&mut candidates, "user", value.get("text"), HashMap::new());
        if !candidates.is_empty() {
            topics.insert(session_id.to_string(), candidates);
            if topics.len() == wanted.len() {
                break;
            }
        }
    }
    Ok(topics)
}

fn latest_message_candidates(locator: &SessionLocator) -> Result<Vec<SessionPreviewCandidate>> {
    match &locator.storage {
        StorageLocator::File { path } => {
            latest_file_message_candidates(path, locator.harness.as_str())
        }
        StorageLocator::Sqlite { path, selector }
            if locator.harness.as_str() == HarnessId::OPENCODE =>
        {
            latest_opencode_message_candidates(path, selector)
        }
        StorageLocator::Sqlite { path, selector }
            if locator.harness.as_str() == HarnessId::GOOSE =>
        {
            latest_goose_message_candidates(path, selector)
        }
        StorageLocator::Sqlite { .. } => Ok(Vec::new()),
    }
}

fn topic_file_message_candidates(
    path: &Path,
    harness: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
    let mut file = File::open(path)?;
    let mut bytes = Vec::with_capacity(TOPIC_PREVIEW_HEAD_BYTES as usize);
    file.by_ref()
        .take(TOPIC_PREVIEW_HEAD_BYTES)
        .read_to_end(&mut bytes)?;
    if file.metadata()?.len() > TOPIC_PREVIEW_HEAD_BYTES {
        if let Some(newline) = bytes.iter().rposition(|byte| *byte == b'\n') {
            bytes.truncate(newline);
        }
    }
    let text = String::from_utf8(bytes).map_err(|_| {
        Error::Other(format!(
            "{} contains non-UTF-8 data in its topic-preview window",
            path.display()
        ))
    })?;
    let mut candidates = Vec::new();
    for line in text.lines() {
        let Ok(value) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        push_topic_message_candidate(&mut candidates, harness, &value);
        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
            break;
        }
    }
    Ok(candidates)
}

fn latest_file_message_candidates(
    path: &Path,
    harness: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
    let mut candidates =
        latest_file_message_candidates_with_limit(path, harness, LATEST_PREVIEW_TAIL_BYTES)?;
    if candidates.is_empty() {
        candidates =
            latest_file_message_candidates_with_limit(path, harness, LATEST_PREVIEW_MAX_BYTES)?;
    }
    Ok(candidates)
}

fn latest_file_message_candidates_with_limit(
    path: &Path,
    harness: &str,
    byte_limit: u64,
) -> Result<Vec<SessionPreviewCandidate>> {
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len();
    let start = file_len.saturating_sub(byte_limit);
    file.seek(SeekFrom::Start(start))?;
    let mut bytes = Vec::with_capacity((file_len - start) as usize);
    file.read_to_end(&mut bytes)?;
    if start > 0 {
        if let Some(newline) = bytes.iter().position(|byte| *byte == b'\n') {
            bytes.drain(..=newline);
        } else {
            return Ok(Vec::new());
        }
    }
    let text = String::from_utf8(bytes).map_err(|_| {
        Error::Other(format!(
            "{} contains non-UTF-8 data in its list-preview window",
            path.display()
        ))
    })?;
    let mut candidates = Vec::new();
    for line in text.lines().rev() {
        let Ok(value) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        let (role, content, metadata) = match harness {
            HarnessId::CLAUDE_CODE => {
                let role = value.get("type").and_then(Value::as_str);
                if !matches!(role, Some("user" | "assistant")) {
                    continue;
                }
                let metadata = if role == Some("user") {
                    crate::session::claude_user_provenance(&value)
                        .into_iter()
                        .collect()
                } else {
                    HashMap::new()
                };
                (
                    role.unwrap_or_default(),
                    value
                        .get("message")
                        .and_then(|message| message.get("content")),
                    metadata,
                )
            }
            HarnessId::CODEX => {
                let payload = value.get("payload").unwrap_or(&Value::Null);
                if value.get("type").and_then(Value::as_str) != Some("response_item")
                    || payload.get("type").and_then(Value::as_str) != Some("message")
                {
                    continue;
                }
                let Some(role @ ("user" | "assistant")) =
                    payload.get("role").and_then(Value::as_str)
                else {
                    continue;
                };
                (role, payload.get("content"), HashMap::new())
            }
            HarnessId::PI => {
                if value.get("type").and_then(Value::as_str) != Some("message") {
                    continue;
                }
                let message = value.get("message").unwrap_or(&Value::Null);
                let Some(role @ ("user" | "assistant")) =
                    message.get("role").and_then(Value::as_str)
                else {
                    continue;
                };
                (role, message.get("content"), HashMap::new())
            }
            HarnessId::GEMINI => {
                let Some(kind @ ("user" | "gemini")) = value.get("type").and_then(Value::as_str)
                else {
                    continue;
                };
                (
                    if kind == "gemini" {
                        "assistant"
                    } else {
                        "user"
                    },
                    value.get("content"),
                    HashMap::new(),
                )
            }
            HarnessId::GROK => {
                let Some(role @ ("user" | "assistant")) = value.get("type").and_then(Value::as_str)
                else {
                    continue;
                };
                (role, value.get("content"), HashMap::new())
            }
            HarnessId::SUPERCODE => {
                let Some(role @ ("user" | "assistant")) = value.get("role").and_then(Value::as_str)
                else {
                    continue;
                };
                (role, value.get("content"), HashMap::new())
            }
            _ => continue,
        };
        let mut metadata = metadata;
        if matches!(harness, HarnessId::CLAUDE_CODE | HarnessId::CODEX) {
            if let Some(timestamp) = value.get("timestamp").and_then(Value::as_str) {
                metadata.insert("timestamp".to_string(), timestamp.to_string());
            }
        }
        push_message_candidate(&mut candidates, role, content, metadata);
        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
            break;
        }
    }
    Ok(candidates)
}

fn push_topic_message_candidate(
    candidates: &mut Vec<SessionPreviewCandidate>,
    harness: &str,
    value: &Value,
) {
    let (role, content, metadata) = match harness {
        HarnessId::CLAUDE_CODE => {
            let role = value.get("type").and_then(Value::as_str);
            if !matches!(role, Some("user" | "assistant")) {
                return;
            }
            let metadata = if role == Some("user") {
                crate::session::claude_user_provenance(value)
                    .into_iter()
                    .collect()
            } else {
                HashMap::new()
            };
            (
                role.unwrap_or_default(),
                value
                    .get("message")
                    .and_then(|message| message.get("content")),
                metadata,
            )
        }
        HarnessId::CODEX => {
            let payload = value.get("payload").unwrap_or(&Value::Null);
            if value.get("type").and_then(Value::as_str) != Some("response_item")
                || payload.get("type").and_then(Value::as_str) != Some("message")
            {
                return;
            }
            let Some(role @ ("user" | "assistant")) = payload.get("role").and_then(Value::as_str)
            else {
                return;
            };
            (role, payload.get("content"), HashMap::new())
        }
        _ => return,
    };
    push_message_candidate(candidates, role, content, metadata);
}

fn latest_opencode_message_candidates(
    path: &Path,
    session_id: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
    let connection = Connection::open_with_flags(
        path,
        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )
    .map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
    let mut statement = connection
        .prepare(
            "SELECT m.data, p.data FROM message m JOIN part p ON p.message_id = m.id \
         WHERE m.session_id = ?1 ORDER BY m.time_created DESC, p.time_created DESC LIMIT 32",
        )
        .map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
    let rows = statement
        .query_map([session_id], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })
        .map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
    let mut candidates = Vec::new();
    for row in rows.flatten() {
        let (Ok(message), Ok(part)) = (
            serde_json::from_str::<Value>(&row.0),
            serde_json::from_str::<Value>(&row.1),
        ) else {
            continue;
        };
        let Some(role @ ("user" | "assistant")) = message.get("role").and_then(Value::as_str)
        else {
            continue;
        };
        if part.get("type").and_then(Value::as_str) != Some("text") {
            continue;
        }
        push_message_candidate(&mut candidates, role, part.get("text"), HashMap::new());
        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
            break;
        }
    }
    Ok(candidates)
}

fn latest_goose_message_candidates(
    path: &Path,
    session_id: &str,
) -> Result<Vec<SessionPreviewCandidate>> {
    let connection = Connection::open_with_flags(
        path,
        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )
    .map_err(|error| Error::Other(format!("failed to open list-preview store: {error}")))?;
    let mut statement = connection
        .prepare(
            "SELECT role, content_json FROM messages WHERE session_id = ?1 \
         ORDER BY created_timestamp DESC, id DESC LIMIT 16",
        )
        .map_err(|error| Error::Other(format!("failed to prepare list-preview query: {error}")))?;
    let rows = statement
        .query_map([session_id], |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        })
        .map_err(|error| Error::Other(format!("failed to read list-preview rows: {error}")))?;
    let mut candidates = Vec::new();
    for row in rows.flatten() {
        let (role, content) = row;
        if !matches!(role.as_str(), "user" | "assistant") {
            continue;
        }
        let Ok(content) = serde_json::from_str::<Value>(&content) else {
            continue;
        };
        push_message_candidate(&mut candidates, &role, Some(&content), HashMap::new());
        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
            break;
        }
    }
    Ok(candidates)
}

fn push_message_candidate(
    candidates: &mut Vec<SessionPreviewCandidate>,
    role: &str,
    content: Option<&Value>,
    metadata: HashMap<String, String>,
) {
    if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
        return;
    }
    let Some(text) = display_text(content).filter(|text| !text.trim().is_empty()) else {
        return;
    };
    const MAX_CHARS: usize = 4_096;
    candidates.push(SessionPreviewCandidate {
        role: role.to_string(),
        content: text.chars().take(MAX_CHARS).collect(),
        metadata,
    });
}

fn fill_string(target: &mut Option<String>, value: Option<&Value>) {
    if target.is_none() {
        *target = value.and_then(Value::as_str).map(str::to_owned);
    }
}

fn fill_path(target: &mut Option<PathBuf>, value: Option<&Value>) {
    if target.is_none() {
        *target = value.and_then(Value::as_str).map(PathBuf::from);
    }
}

fn modified_ms(path: &Path) -> Option<u64> {
    fs::metadata(path)
        .ok()?
        .modified()
        .ok()?
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|duration| u64::try_from(duration.as_millis()).ok())
}

/// A workspace filter is satisfiable only by a session whose RECORDED working
/// directory is absolute. A relative recorded cwd (OpenCode has shipped
/// literal `"."` session rows) carries no information about where the session
/// ran; resolving it against the discoverer's own current directory made such
/// a session match every workspace discovery happened to run from.
fn recorded_cwd_matches(recorded: &Path, wanted: &Path) -> bool {
    recorded.is_absolute() && same_path(recorded, wanted)
}

fn same_path(left: &Path, right: &Path) -> bool {
    match (fs::canonicalize(left), fs::canonicalize(right)) {
        (Ok(left), Ok(right)) => left == right,
        _ => normalize_path(left) == normalize_path(right),
    }
}

fn normalize_path(path: &Path) -> PathBuf {
    let absolute = if path.is_absolute() {
        path.to_path_buf()
    } else {
        std::env::current_dir()
            .unwrap_or_else(|_| PathBuf::from("."))
            .join(path)
    };
    let mut normalized = PathBuf::new();
    for component in absolute.components() {
        match component {
            Component::CurDir => {}
            Component::ParentDir => {
                normalized.pop();
            }
            other => normalized.push(other.as_os_str()),
        }
    }
    normalized
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::{SystemTime, UNIX_EPOCH};

    fn temp_dir(label: &str) -> PathBuf {
        let nonce = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "supercode-catalog-{label}-{}-{nonce}",
            std::process::id()
        ));
        fs::create_dir_all(&path).unwrap();
        path
    }

    #[test]
    fn locator_json_round_trip_preserves_sqlite_selector() {
        let locator = SessionLocator {
            harness: HarnessId::from(HarnessId::OPENCODE),
            session_id: "ses_123".into(),
            storage: StorageLocator::Sqlite {
                path: PathBuf::from("/tmp/opencode-dev.db"),
                selector: "ses_123".into(),
            },
        };
        let encoded = serde_json::to_string(&locator).unwrap();
        assert_eq!(
            serde_json::from_str::<SessionLocator>(&encoded).unwrap(),
            locator
        );
    }

    #[test]
    fn discovers_filters_loads_and_follows_three_jsonl_harnesses() {
        let root = temp_dir("jsonl");
        let workspace = root.join("workspace");
        let other = root.join("other");
        fs::create_dir_all(&workspace).unwrap();
        fs::create_dir_all(&other).unwrap();

        let claude = root.join("claude");
        let codex = root.join("codex");
        let pi = root.join("pi");
        fs::create_dir_all(&claude).unwrap();
        fs::create_dir_all(&codex).unwrap();
        fs::create_dir_all(&pi).unwrap();
        fs::write(
            claude.join("claude.jsonl"),
            format!(
                "{{\"type\":\"user\",\"sessionId\":\"cc-1\",\"cwd\":{},\"timestamp\":\"2026-01-01T00:00:01Z\",\"message\":{{\"role\":\"user\",\"content\":\"hi\"}}}}\n",
                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
            ),
        )
        .unwrap();
        fs::write(
            codex.join("rollout.jsonl"),
            format!(
                "{{\"timestamp\":\"2026-01-01T00:00:00Z\",\"type\":\"session_meta\",\"payload\":{{\"id\":\"cx-1\",\"cwd\":{}}}}}\n{{\"timestamp\":\"2026-01-01T00:00:02Z\",\"type\":\"response_item\",\"payload\":{{\"type\":\"message\",\"role\":\"user\",\"content\":[{{\"type\":\"input_text\",\"text\":\"inspect codex\"}}]}}}}\n",
                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
            ),
        )
        .unwrap();
        fs::write(
            pi.join("pi.jsonl"),
            format!(
                "{{\"type\":\"session\",\"version\":3,\"id\":\"pi-1\",\"timestamp\":\"2026-01-01T00:00:00Z\",\"cwd\":{}}}\n{{\"type\":\"message\",\"message\":{{\"role\":\"user\",\"content\":\"inspect pi\"}}}}\n",
                serde_json::to_string(&workspace.to_string_lossy()).unwrap()
            ),
        )
        .unwrap();
        fs::write(
            pi.join("unrelated.jsonl"),
            format!(
                "{{\"type\":\"session\",\"version\":3,\"id\":\"pi-2\",\"timestamp\":\"2026-01-01T00:00:00Z\",\"cwd\":{}}}\n",
                serde_json::to_string(&other.to_string_lossy()).unwrap()
            ),
        )
        .unwrap();
        fs::write(claude.join("partial.jsonl"), "{truncated").unwrap();

        let query = DiscoveryQuery {
            workspace: Some(workspace),
            homes: HarnessHomes {
                claude_code: claude,
                codex,
                pi,
                opencode: root.join("missing-opencode"),
                grok: root.join("missing-grok"),
                gemini: root.join("missing-gemini"),
                goose: root.join("missing-goose"),
                supercode: root.join("missing-supercode"),
            },
            ..DiscoveryQuery::default()
        };
        let catalog = HarnessCatalog::new();
        let found = catalog.discover(&query).unwrap();
        assert_eq!(found.len(), 3);
        assert_eq!(
            found
                .iter()
                .map(|item| item.locator.harness.as_str())
                .collect::<HashSet<_>>(),
            HashSet::from([HarnessId::CLAUDE_CODE, HarnessId::CODEX, HarnessId::PI])
        );
        for descriptor in found {
            assert!(descriptor.preview_candidates.is_empty());
            assert_eq!(descriptor.latest_message_candidates.len(), 1);
            assert_eq!(descriptor.latest_message_candidates[0].role, "user");
            if descriptor.locator.harness.as_str() == HarnessId::CLAUDE_CODE {
                assert_eq!(
                    descriptor.latest_message_candidates[0]
                        .metadata
                        .get("timestamp")
                        .map(String::as_str),
                    Some("2026-01-01T00:00:01Z")
                );
            } else if descriptor.locator.harness.as_str() == HarnessId::CODEX {
                assert_eq!(
                    descriptor.latest_message_candidates[0]
                        .metadata
                        .get("timestamp")
                        .map(String::as_str),
                    Some("2026-01-01T00:00:02Z")
                );
            }
            let loaded = catalog.load(&descriptor.locator).unwrap();
            assert_eq!(
                loaded.meta.session_id.as_deref(),
                Some(descriptor.locator.session_id.as_str())
            );
            let mut follower = catalog.follow(&descriptor.locator).unwrap();
            assert!(matches!(
                follower.poll().unwrap(),
                Some(crate::SessionWatchEvent::SessionSnapshot { .. })
            ));
        }
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn codex_child_rollouts_roll_into_roots_before_pagination() {
        let root = temp_dir("codex-roots");
        let codex = root.join("codex");
        fs::create_dir_all(&codex).unwrap();
        let write_rollout =
            |name: &str, payload: Value, modified_seconds: u64| {
                let path = codex.join(format!("{name}.jsonl"));
                fs::write(
                    &path,
                    format!(
                        "{}\n",
                        serde_json::json!({
                            "timestamp": "2026-01-01T00:00:00Z",
                            "type": "session_meta",
                            "payload": payload,
                        })
                    ),
                )
                .unwrap();
                File::open(&path)
                    .unwrap()
                    .set_times(fs::FileTimes::new().set_modified(
                        UNIX_EPOCH + std::time::Duration::from_secs(modified_seconds),
                    ))
                    .unwrap();
            };
        write_rollout(
            "parent",
            serde_json::json!({"id":"parent","cwd":"/project","source":"cli"}),
            100,
        );
        write_rollout(
            "other",
            serde_json::json!({"id":"other","cwd":"/project","source":"cli"}),
            200,
        );
        write_rollout(
            "child",
            serde_json::json!({
                "id": "child",
                "cwd": "/project",
                "parent_thread_id": "parent",
                "source": {"subagent":{"thread_spawn":{
                    "parent_thread_id":"parent",
                    "depth":1,
                    "agent_path":"/root/reviewer"
                }}}
            }),
            300,
        );

        let catalog = HarnessCatalog::new();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::from(HarnessId::CODEX)],
            homes: HarnessHomes {
                codex: codex.clone(),
                ..HarnessHomes::default()
            },
            limit: Some(1),
            ..DiscoveryQuery::default()
        };
        let roots = catalog.discover(&query).unwrap();
        assert_eq!(roots.len(), 1);
        assert_eq!(roots[0].locator.session_id, "parent");
        assert_eq!(roots[0].updated_at_ms, Some(300_000));
        assert_eq!(roots[0].parent_session_id, None);
        assert_eq!(roots[0].child_session_count, 1);

        let tree = catalog
            .discover(&DiscoveryQuery {
                limit: None,
                include_child_sessions: true,
                root_session_id: Some("parent".into()),
                ..query
            })
            .unwrap();
        assert_eq!(tree.len(), 2);
        assert!(tree
            .iter()
            .all(|descriptor| descriptor.locator.session_id != "other"));
        let child = tree
            .iter()
            .find(|descriptor| descriptor.locator.session_id == "child")
            .unwrap();
        assert_eq!(child.parent_session_id.as_deref(), Some("parent"));
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn discovers_loads_and_follows_opencode_sqlite() {
        let db = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("../harness/tests/fixtures/opencode_fixture/opencode.db");
        let catalog = HarnessCatalog::new();
        let found = catalog
            .discover(&DiscoveryQuery {
                harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
                homes: HarnessHomes {
                    opencode: db,
                    ..HarnessHomes::default()
                },
                ..DiscoveryQuery::default()
            })
            .unwrap();
        assert!(!found.is_empty());
        for descriptor in found {
            assert_eq!(descriptor.locator.harness.as_str(), HarnessId::OPENCODE);
            assert_eq!(
                catalog.load(&descriptor.locator).unwrap().meta.session_id,
                Some(descriptor.locator.session_id.clone())
            );
            assert!(catalog.follow(&descriptor.locator).is_ok());
        }
    }

    #[test]
    fn discovers_loads_and_follows_gemini_conversation_records() {
        let root = temp_dir("gemini");
        let workspace = root.join("workspace");
        let chats = root.join("gemini/tmp/demo/chats");
        fs::create_dir_all(&workspace).unwrap();
        fs::create_dir_all(&chats).unwrap();
        fs::write(
            root.join("gemini/projects.json"),
            serde_json::json!({
                "projects": {workspace.to_string_lossy(): "demo"}
            })
            .to_string(),
        )
        .unwrap();
        let transcript = chats.join("gemini-id.jsonl");
        fs::write(
            &transcript,
            include_str!("../../harness/tests/fixtures/gemini_session.jsonl"),
        )
        .unwrap();

        let catalog = HarnessCatalog::new();
        let found = catalog
            .discover(&DiscoveryQuery {
                harnesses: vec![HarnessId::from(HarnessId::GEMINI)],
                homes: HarnessHomes {
                    gemini: root.join("gemini"),
                    ..HarnessHomes::default()
                },
                workspace: Some(workspace.clone()),
                ..DiscoveryQuery::default()
            })
            .unwrap();

        assert_eq!(found.len(), 1);
        assert_eq!(found[0].cwd.as_deref(), Some(workspace.as_path()));
        assert_eq!(found[0].message_count, None);
        assert_eq!(found[0].model.as_deref(), Some("gemini-2.5-pro"));
        assert_eq!(found[0].title, None);
        assert!(found[0].preview_candidates.is_empty());
        assert_eq!(found[0].latest_message_candidates.len(), 3);
        assert_eq!(
            found[0].latest_message_candidates[0].content,
            "Fixture inspected."
        );
        let loaded = catalog.load(&found[0].locator).unwrap();
        assert_eq!(
            loaded.meta.session_id.as_deref(),
            Some("11111111-1111-4111-8111-111111111111")
        );
        assert_eq!(loaded.messages.len(), 4);
        assert!(matches!(
            catalog.follow(&found[0].locator).unwrap().poll().unwrap(),
            Some(crate::SessionWatchEvent::SessionSnapshot { .. })
        ));
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn discovers_native_store_and_pages_search_results() {
        let root = temp_dir("supercode");
        let store_root = root.join("sessions");
        fs::create_dir_all(&store_root).unwrap();
        for (name, title) in [
            ("alpha", "Alpha planning"),
            ("beta", "Beta implementation"),
            ("gamma", "Gamma review"),
        ] {
            fs::write(
                store_root.join(format!("{name}.jsonl")),
                format!("{{\"role\":\"user\",\"content\":\"{title}\"}}\n"),
            )
            .unwrap();
            fs::write(
                store_root.join(format!("{name}.meta.json")),
                serde_json::json!({"name": name, "title": title}).to_string(),
            )
            .unwrap();
        }
        let catalog = HarnessCatalog::new();
        let base = DiscoveryQuery {
            harnesses: vec![HarnessId::from(HarnessId::SUPERCODE)],
            homes: HarnessHomes {
                supercode: store_root,
                ..HarnessHomes::default()
            },
            limit: Some(1),
            ..DiscoveryQuery::default()
        };

        let first = catalog.discover_page(&base).unwrap();
        assert_eq!(first.sessions.len(), 1);
        assert!(first.next_cursor.is_some());
        let second = catalog
            .discover_page(&DiscoveryQuery {
                cursor: first.next_cursor,
                ..base.clone()
            })
            .unwrap();
        assert_eq!(second.sessions.len(), 1);
        assert_ne!(
            first.sessions[0].locator.session_id,
            second.sessions[0].locator.session_id
        );
        let search = catalog
            .discover_page(&DiscoveryQuery {
                limit: None,
                query: Some("implementation".into()),
                ..base
            })
            .unwrap();
        assert_eq!(search.sessions.len(), 1);
        assert_eq!(search.sessions[0].locator.session_id, "beta");
        assert_eq!(search.sessions[0].message_count, None);
        assert_eq!(
            catalog
                .load(&search.sessions[0].locator)
                .unwrap()
                .messages
                .len(),
            1
        );
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn native_workspace_discovery_reads_bounded_sidecar_headers() {
        let root = temp_dir("supercode-bounded-header");
        let store_root = root.join("sessions");
        let workspace = root.join("project");
        fs::create_dir_all(&store_root).unwrap();
        fs::create_dir_all(&workspace).unwrap();
        let name = "bounded-native";
        fs::write(
            store_root.join(format!("{name}.meta.json")),
            serde_json::json!({"name": name, "title": "Bounded native"}).to_string(),
        )
        .unwrap();
        fs::write(
            store_root.join(format!("{name}.jsonl")),
            "{\"role\":\"user\",\"content\":\"projected view\"}\n",
        )
        .unwrap();
        let sidecar = [
            serde_json::json!({
                "supercode_native": 2,
                "source": "claude_code",
                "session_id": "native-session"
            })
            .to_string(),
            serde_json::json!({
                "type": "user",
                "sessionId": "native-session",
                "cwd": workspace,
                "message": {"role": "user", "content": "hello"}
            })
            .to_string(),
            serde_json::json!({
                "type": "assistant",
                "sessionId": "native-session",
                "cwd": workspace,
                "message": {"role": "assistant", "model": "claude-sonnet-5", "content": []}
            })
            .to_string(),
            // A full native-family parse rejects this trailing residue. Header
            // discovery must not touch it after it has enough metadata.
            "not-json".into(),
        ]
        .join("\n");
        fs::write(
            store_root.join(format!("{name}.sidecar.jsonl")),
            format!("{sidecar}\n"),
        )
        .unwrap();

        let found = HarnessCatalog::new()
            .discover(&DiscoveryQuery {
                workspace: Some(workspace.clone()),
                harnesses: vec![HarnessId::from(HarnessId::SUPERCODE)],
                homes: HarnessHomes {
                    supercode: store_root,
                    ..HarnessHomes::default()
                },
                ..DiscoveryQuery::default()
            })
            .unwrap();

        assert_eq!(found.len(), 1);
        assert_eq!(found[0].cwd.as_deref(), Some(workspace.as_path()));
        assert_eq!(found[0].model.as_deref(), Some("claude-sonnet-5"));
        assert_eq!(found[0].message_count, None);
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn discovers_current_opencode_schema_without_a_session_model_column() {
        let root = temp_dir("opencode-current");
        let db = root.join("opencode.db");
        let conn = Connection::open(&db).unwrap();
        conn.execute_batch(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT NOT NULL,
                time_updated INTEGER NOT NULL
             );
             CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL
             );
             INSERT INTO session VALUES ('ses_current', '/tmp/work', 'Current', 42);
             INSERT INTO message VALUES ('msg_current', 'ses_current');",
        )
        .unwrap();
        drop(conn);

        let found = HarnessCatalog::new()
            .discover(&DiscoveryQuery {
                harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
                homes: HarnessHomes {
                    opencode: db,
                    ..HarnessHomes::default()
                },
                ..DiscoveryQuery::default()
            })
            .unwrap();

        assert_eq!(found.len(), 1);
        assert_eq!(found[0].locator.session_id, "ses_current");
        assert_eq!(found[0].message_count, Some(1));
        assert_eq!(found[0].model, None);
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn workspace_filter_never_matches_a_relative_recorded_cwd() {
        // OpenCode has shipped session rows whose `directory` is the literal
        // ".". Resolving that against the discoverer's own cwd made the
        // session match every workspace discovery ran from — the workspace
        // here IS the test process cwd, the exact aliasing that leaked.
        let root = temp_dir("opencode-relative-cwd");
        let db = root.join("opencode.db");
        let conn = Connection::open(&db).unwrap();
        let here = std::env::current_dir().unwrap();
        conn.execute_batch(&format!(
            "CREATE TABLE session (
                id TEXT PRIMARY KEY,
                directory TEXT NOT NULL,
                title TEXT NOT NULL,
                time_updated INTEGER NOT NULL
             );
             CREATE TABLE message (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL
             );
             INSERT INTO session VALUES ('ses_relative', '.', 'Ghost', 41);
             INSERT INTO session VALUES ('ses_here', '{}', 'Real', 42);",
            here.display()
        ))
        .unwrap();
        drop(conn);

        let found = HarnessCatalog::new()
            .discover(&DiscoveryQuery {
                workspace: Some(here),
                harnesses: vec![HarnessId::from(HarnessId::OPENCODE)],
                homes: HarnessHomes {
                    opencode: db,
                    ..HarnessHomes::default()
                },
                ..DiscoveryQuery::default()
            })
            .unwrap();

        assert_eq!(found.len(), 1);
        assert_eq!(found[0].locator.session_id, "ses_here");
        fs::remove_dir_all(root).ok();
    }
}