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::{BTreeSet, 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::ontology::{Binding, OrchestratorBindingRow};
use crate::session::{
    hermes_capture_nouns, openclaw_agent_id_from_path, openclaw_capture_header_nouns,
    percent_decode_path, OrchestrationNouns, SessionMeta, SessionSource,
};
use crate::{Error, Fidelity, Result, Session, SessionFollower};

pub use crate::ontology::HarnessId;

/// 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>,
    /// Time of the last recorded TURN as Unix epoch milliseconds.
    ///
    /// Not the transcript file's mtime: a harness rewrites its transcript for
    /// reasons that are not conversation (Claude Code appends untimestamped
    /// `bridge-session` records while a session merely sits open, and moves
    /// mtime again on resume), so mtime ranks idle-but-open sessions above
    /// genuinely active ones. Falls back to mtime only when no timestamped
    /// record is readable.
    pub updated_at_ms: Option<u64>,
    /// Harness message-record count, when available without loading the session.
    /// STORED message-record count in the native store (cheap line scan) —
    /// for branched record-tree dialects this can exceed the active-path
    /// message count a full load renders; `inspect` labels both (SUP-58).
    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,
    /// ORCH-6: the ORCH-3 conversation nouns (`trigger`, `surface`,
    /// `profile`, `recurrence`, `cross_surface`, `workspace`), flattened onto
    /// the row so the wire stays additive. Filled by `finalize_nouns` for
    /// every harness; only Hermes and OpenClaw publish more than the
    /// `trigger`/`workspace` defaults today.
    #[serde(flatten)]
    pub nouns: OrchestrationNouns,
}

/// One bounded normalized conversation-message candidate for list projection.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SessionPreviewCandidate {
    /// Opaque identity of this native message boundary. Consumers may retain
    /// it to reconcile bounded discovery windows without treating a growing
    /// preview or a native-store heartbeat as a new conversation message.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub cursor: Option<String>,
    /// 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>,
    /// Coverage receipt (UNI-7 / SUP-54): what was asked and what came back,
    /// so incomplete coverage can never read as complete.
    #[serde(default)]
    pub receipt: DiscoveryReceipt,
}

/// Coverage evidence for one discovery page.
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(default)]
pub struct DiscoveryReceipt {
    /// True only when bounded topic/latest message search was applied before
    /// pagination. Omitted for ordinary metadata-only discovery.
    #[serde(skip_serializing_if = "is_false")]
    pub searched_previews: bool,
    /// The requested lower time bound (epoch ms), when one was given.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub requested_after_ms: Option<u64>,
    /// The requested upper time bound (epoch ms), when one was given.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub requested_before_ms: Option<u64>,
    /// The requested page limit, when one was given.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub requested_limit: Option<usize>,
    /// Oldest `updated_at_ms` among the RETURNED sessions.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub oldest_returned_ms: Option<u64>,
    /// Newest `updated_at_ms` among the RETURNED sessions.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub newest_returned_ms: Option<u64>,
    /// Sessions in this page.
    pub returned: usize,
    /// Sessions matching the query across ALL pages.
    pub total_matched: usize,
    /// True when matches beyond this page exist (`next_cursor` is the resume
    /// point). Explicit so a consumer cannot mistake a capped page for the
    /// complete result set.
    pub truncated: bool,
}

/// Append-aware index of the stable topic records in Codex `history.jsonl`.
///
/// Long-lived session-list clients can retain this index and refresh it after
/// filesystem invalidations. Ordinary appends read and parse only the new
/// bytes; truncation, replacement, and in-place rewrites rebuild the index so
/// the result remains identical to a fresh catalog discovery.
#[derive(Debug)]
pub struct CodexHistoryTopicIndex {
    path: PathBuf,
    fingerprint: Option<CodexHistoryFingerprint>,
    offset: u64,
    trailing: Vec<u8>,
    topics: HashMap<String, Vec<SessionPreviewCandidate>>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct CodexHistoryFingerprint {
    len: u64,
    modified_ns: u128,
    identity: u128,
}

impl CodexHistoryTopicIndex {
    /// Create an empty index for the Codex sessions root from
    /// [`HarnessHomes::codex`]. Call [`Self::refresh`] before first use.
    pub fn new(sessions_root: &Path) -> Self {
        let root = sessions_root.parent().unwrap_or(sessions_root);
        Self {
            path: root.join("history.jsonl"),
            fingerprint: None,
            offset: 0,
            trailing: Vec::new(),
            topics: HashMap::new(),
        }
    }

    /// Return the native history file watched by this index.
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Refresh from durable state and return session ids whose topic changed.
    ///
    /// A missing file is a valid empty history. Read errors leave the previous
    /// successful index intact so a transient filesystem error cannot erase
    /// topics from a live session list.
    pub fn refresh(&mut self) -> Result<BTreeSet<String>> {
        let metadata = match fs::metadata(&self.path) {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                let changed = self.topics.keys().cloned().collect();
                self.fingerprint = None;
                self.offset = 0;
                self.trailing.clear();
                self.topics.clear();
                return Ok(changed);
            }
            Err(error) => return Err(error.into()),
        };
        let fingerprint = codex_history_fingerprint(&metadata)?;
        if self.fingerprint == Some(fingerprint) {
            return Ok(BTreeSet::new());
        }

        let is_append = self.fingerprint.is_some_and(|previous| {
            previous.identity == fingerprint.identity
                && previous.len < fingerprint.len
                && self.offset <= previous.len
        });
        if is_append {
            let mut file = File::open(&self.path)?;
            file.seek(SeekFrom::Start(self.offset))?;
            let mut bytes = Vec::with_capacity(
                usize::try_from(fingerprint.len.saturating_sub(self.offset)).unwrap_or(0),
            );
            file.read_to_end(&mut bytes)?;
            self.offset = file.stream_position()?;
            let changed = self.ingest(bytes);
            self.fingerprint = Some(CodexHistoryFingerprint {
                len: self.offset,
                ..fingerprint
            });
            return Ok(changed);
        }

        let previous = std::mem::take(&mut self.topics);
        let mut file = File::open(&self.path)?;
        let mut bytes = Vec::with_capacity(usize::try_from(fingerprint.len).unwrap_or(0));
        file.read_to_end(&mut bytes)?;
        self.offset = file.stream_position()?;
        self.trailing.clear();
        self.ingest(bytes);
        self.fingerprint = Some(CodexHistoryFingerprint {
            len: self.offset,
            ..fingerprint
        });
        Ok(changed_topic_ids(&previous, &self.topics))
    }

    fn ingest(&mut self, bytes: Vec<u8>) -> BTreeSet<String> {
        let mut input = std::mem::take(&mut self.trailing);
        input.extend(bytes);
        let complete_len = input
            .iter()
            .rposition(|byte| *byte == b'\n')
            .map_or(0, |index| index + 1);
        let mut changed = BTreeSet::new();
        for line in input[..complete_len].split(|byte| *byte == b'\n') {
            self.ingest_line(line, &mut changed);
        }
        self.trailing.extend_from_slice(&input[complete_len..]);
        if !self.trailing.is_empty() {
            let trailing = self.trailing.clone();
            if self.ingest_line(&trailing, &mut changed) {
                self.trailing.clear();
            }
        }
        changed
    }

    fn ingest_line(&mut self, line: &[u8], changed: &mut BTreeSet<String>) -> bool {
        let Ok(value) = serde_json::from_slice::<Value>(line) else {
            return false;
        };
        let Some(session_id) = value.get("session_id").and_then(Value::as_str) else {
            return true;
        };
        if self.topics.contains_key(session_id) {
            return true;
        }
        let mut candidates = Vec::new();
        push_message_candidate(&mut candidates, "user", value.get("text"), HashMap::new());
        if !candidates.is_empty() {
            self.topics.insert(session_id.to_string(), candidates);
            changed.insert(session_id.to_string());
        }
        true
    }
}

/// 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,
    /// OpenClaw home root (contains `agents/<agentId>/sessions/*.jsonl`,
    /// openclaw >= 2026.7 — plain pi-v3 dialect files; see
    /// `SessionSource::OpenClaw`). READ-ONLY discovery (UNI-16).
    pub openclaw: PathBuf,
    /// Hermes `state.db` SQLite store path (see `SessionSource::Hermes`).
    /// READ-ONLY discovery (UNI-15).
    pub hermes: PathBuf,
    /// The orchestrator's home FOLDER (`SUPERCODE_ORCHESTRATOR_HOME`, default
    /// `~/.supercode/orchestrator`). Unlike every other entry this addresses a
    /// directory, because the orchestrator's serialization IS its folder
    /// (`docs/ORCHESTRATOR-IR.md` §6): the root is the `default` profile and
    /// `profiles/<name>/` are the named ones. READ-ONLY (ORC-7).
    pub orchestrator: PathBuf,
}

/// Every profile folder under an orchestrator home, in listing order: the
/// root (the implicit `default` profile) then each `profiles/<name>/`.
///
/// One helper, used by every reader that answers `--harness orchestrator`, so
/// the folder layout of `docs/ORCHESTRATOR-IR.md` §6 is stated once.
pub fn orchestrator_profile_dirs(root: &Path) -> Vec<(String, PathBuf)> {
    if !root.is_dir() {
        return Vec::new();
    }
    let mut dirs = vec![("default".to_string(), root.to_path_buf())];
    if let Ok(entries) = fs::read_dir(root.join("profiles")) {
        let mut named: Vec<(String, PathBuf)> = entries
            .flatten()
            .filter(|entry| entry.path().is_dir())
            .filter_map(|entry| {
                entry
                    .file_name()
                    .into_string()
                    .ok()
                    .map(|name| (name, entry.path()))
            })
            // A dot-prefixed folder is never a profile: `profiles/.trash/`
            // holds the homes `profiles.delete` moved aside (ORC-13), and the
            // package's own loader skips it for the same reason.
            .filter(|(name, _)| !name.starts_with('.'))
            .collect();
        named.sort();
        dirs.extend(named);
    }
    dirs
}

/// A Hermes home's session stores: its own `state.db` and each named profile's (Hermes gives every profile
/// its own home and store under `profiles/<name>/`; a linked profile folder counts). The same folder walk as
/// the orchestrator's profiles, since both lay profiles out as Hermes does.
pub fn hermes_session_stores(db_path: &Path) -> Vec<PathBuf> {
    let mut stores = vec![db_path.to_path_buf()];
    let (Some(root), Some(name)) = (db_path.parent(), db_path.file_name()) else {
        return stores;
    };
    for (_, dir) in orchestrator_profile_dirs(root).into_iter().skip(1) {
        stores.push(dir.join(name));
    }
    stores
}

impl Default for HarnessHomes {
    fn default() -> Self {
        let home = crate::user_home()
            .map(std::path::PathBuf::into_os_string)
            .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"));
        // Blind-walk finding 2026-08-31: openclaw itself treats
        // OPENCLAW_HOME as a HOME replacement (state lives at
        // `$OPENCLAW_HOME/.openclaw`), and names the state dir directly with
        // OPENCLAW_STATE_DIR. Mirror those semantics exactly so an inherited
        // environment means the same thing to us and to any openclaw process
        // we spawn.
        let openclaw = std::env::var_os("OPENCLAW_STATE_DIR")
            .map(PathBuf::from)
            .or_else(|| {
                std::env::var_os("OPENCLAW_HOME").map(|root| PathBuf::from(root).join(".openclaw"))
            })
            .unwrap_or_else(|| home.join(".openclaw"));
        let orchestrator = std::env::var_os("SUPERCODE_ORCHESTRATOR_HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| home.join(".supercode/orchestrator"));
        let hermes = std::env::var_os("HERMES_HOME")
            .map(PathBuf::from)
            .unwrap_or_else(|| home.join(".hermes"))
            .join("state.db");
        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,
            openclaw,
            hermes,
            orchestrator,
            pi,
            opencode,
            grok,
        }
    }
}

fn codex_history_fingerprint(metadata: &fs::Metadata) -> Result<CodexHistoryFingerprint> {
    let modified_ns = metadata
        .modified()?
        .duration_since(UNIX_EPOCH)
        .map_err(|error| Error::Other(format!("history timestamp predates Unix epoch: {error}")))?
        .as_nanos();
    #[cfg(unix)]
    let identity = {
        use std::os::unix::fs::MetadataExt;
        (u128::from(metadata.dev()) << 64) | u128::from(metadata.ino())
    };
    #[cfg(not(unix))]
    let identity = 0;
    Ok(CodexHistoryFingerprint {
        len: metadata.len(),
        modified_ns,
        identity,
    })
}

fn changed_topic_ids(
    before: &HashMap<String, Vec<SessionPreviewCandidate>>,
    after: &HashMap<String, Vec<SessionPreviewCandidate>>,
) -> BTreeSet<String> {
    before
        .keys()
        .chain(after.keys())
        .filter(|session_id| before.get(*session_id) != after.get(*session_id))
        .cloned()
        .collect()
}

/// 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>,
    /// Widen `workspace` from that exact folder to the folder and every folder
    /// under it (a Teams sync rule's root). Matched the way the exact filter
    /// matches: only an absolute recorded cwd qualifies, both sides are
    /// canonicalized (symlinks, macOS `/var` vs `/private/var`, Windows `\\?\`
    /// prefixes), and containment is per path component, so `/work` does not
    /// admit `/workshop`. Applied before any per-session enrichment. Ignored
    /// without `workspace`; absent (old clients) means the exact match.
    #[serde(skip_serializing_if = "std::ops::Not::not")]
    pub workspace_subtree: bool,
    /// Only return sessions whose recorded working directory belongs to the
    /// same REPOSITORY FAMILY as this path (UNI-7 / SUP-54 decision): the
    /// family identity is the realpath of the git common directory, so a main
    /// checkout and all its worktrees join as one family; the origin URL is
    /// the tie-breaker that also joins separate clones of the same remote. A
    /// non-repository path degrades to exact-realpath matching. Combines with
    /// `workspace` as AND when both are set (exact-cwd stays available).
    pub workspace_family: Option<PathBuf>,
    /// Only return sessions updated at or after this epoch-ms instant.
    pub updated_after_ms: Option<u64>,
    /// Only return sessions updated at or before this epoch-ms instant.
    pub updated_before_ms: Option<u64>,
    /// 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>,
    /// Also match bounded opening/latest message candidates. Explicitly opt-in:
    /// this scans previews of eligible sessions before pagination, not just the
    /// returned page. Requires a nonempty query; not a full-transcript search.
    pub search_previews: bool,
    /// 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>,
    /// ORCH-6: only return sessions routed through this config home (Hermes
    /// `profile_name` / gateway key namespace, OpenClaw agent id). Exact
    /// match; a session with no profile never matches. `harnesses` is the
    /// harness filter and needs no second spelling.
    pub profile: Option<String>,
}

/// Repository-family identity (UNI-7 / SUP-54 decision): the realpath of the
/// git common directory joins a main checkout with all its worktrees, and the
/// origin URL is the tie-breaker that also joins separate clones of the same
/// remote. Resolved from the filesystem alone (`.git` file/dir + config), no
/// subprocess, so discovery stays deterministic and sandbox-friendly.
#[derive(Debug, Clone, PartialEq, Eq)]
struct RepoFamily {
    /// Realpath of the git common directory, or the realpath of the queried
    /// path itself when it is not inside a git repository.
    identity: PathBuf,
    /// True when `identity` is a git common directory (not a bare-path
    /// degradation) — only then may origin tie-breaking apply.
    is_repository: bool,
    /// `[remote "origin"] url` from the common directory's config, if any.
    origin_url: Option<String>,
}

impl RepoFamily {
    fn of(path: &Path) -> Self {
        let start = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
        let mut current = Some(start.as_path());
        while let Some(dir) = current {
            let dot_git = dir.join(".git");
            if dot_git.is_dir() {
                let identity = std::fs::canonicalize(&dot_git).unwrap_or_else(|_| dot_git.clone());
                let origin_url = read_origin_url(&identity);
                return Self {
                    identity,
                    is_repository: true,
                    origin_url,
                };
            }
            if dot_git.is_file() {
                // Worktree checkout: `.git` is a one-line pointer file
                // `gitdir: <main>/.git/worktrees/<name>`; the common directory
                // is everything before `/worktrees/<name>`.
                if let Ok(text) = std::fs::read_to_string(&dot_git) {
                    if let Some(gitdir) = text
                        .lines()
                        .find_map(|line| line.trim().strip_prefix("gitdir:"))
                    {
                        let gitdir = PathBuf::from(gitdir.trim());
                        let gitdir = if gitdir.is_absolute() {
                            gitdir
                        } else {
                            dir.join(gitdir)
                        };
                        let common = gitdir
                            .parent()
                            .filter(|parent| parent.ends_with("worktrees"))
                            .and_then(Path::parent)
                            .map(Path::to_path_buf)
                            .unwrap_or(gitdir);
                        let identity = std::fs::canonicalize(&common).unwrap_or(common);
                        let origin_url = read_origin_url(&identity);
                        return Self {
                            identity,
                            is_repository: true,
                            origin_url,
                        };
                    }
                }
            }
            current = dir.parent();
        }
        Self {
            identity: start,
            is_repository: false,
            origin_url: None,
        }
    }

    /// Two paths join one family when their common directories match, or —
    /// for genuine repositories only — when both declare the same origin URL
    /// (the clone tie-breaker). Bare-path degradations never origin-match.
    fn joins(&self, other: &Self) -> bool {
        if self.identity == other.identity {
            return true;
        }
        self.is_repository
            && other.is_repository
            && matches!((&self.origin_url, &other.origin_url), (Some(a), Some(b)) if a == b)
    }
}

/// Minimal git-config scan for `[remote "origin"] url = ...`.
fn read_origin_url(common_dir: &Path) -> Option<String> {
    let text = std::fs::read_to_string(common_dir.join("config")).ok()?;
    let mut in_origin = false;
    for line in text.lines() {
        let line = line.trim();
        if line.starts_with('[') {
            in_origin = line.starts_with("[remote \"origin\"]");
            continue;
        }
        if in_origin {
            if let Some(value) = line.strip_prefix("url") {
                let value = value.trim_start();
                if let Some(url) = value.strip_prefix('=') {
                    let url = url.trim();
                    if !url.is_empty() {
                        return Some(url.to_string());
                    }
                }
            }
        }
    }
    None
}

/// 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)
    }

    /// Scan file-backed session metadata without pagination or conversation
    /// previews. Child sessions are retained so a long-lived caller can keep
    /// an exact in-memory lineage index and project it without rescanning the
    /// native stores.
    pub fn discover_raw_index(&self, query: &DiscoveryQuery) -> Vec<SessionDescriptor> {
        self.scan_descriptors(query, true)
    }

    /// Project a raw metadata index through the query's lineage, search, and
    /// pagination rules without reading any transcript content.
    pub fn project_index(
        &self,
        query: &DiscoveryQuery,
        descriptors: impl IntoIterator<Item = SessionDescriptor>,
    ) -> Result<Vec<SessionDescriptor>> {
        Ok(self.project_index_page(query, descriptors)?.sessions)
    }

    /// Project a retained metadata index without reading transcripts, preserving
    /// the full-match count and successor cursor of ordinary discovery.
    pub fn project_index_page(
        &self,
        query: &DiscoveryQuery,
        descriptors: impl IntoIterator<Item = SessionDescriptor>,
    ) -> Result<DiscoveryPage> {
        if query.search_previews {
            return Err(Error::Other(
                "preview search requires discover_page, not a metadata-only index projection"
                    .into(),
            ));
        }
        let mut found = descriptors.into_iter().collect::<Vec<_>>();
        project_descriptors(query, &mut found);
        let total_matched = found.len();
        let (sessions, next_cursor) = paginate_descriptors(query, found)?;
        let receipt = DiscoveryReceipt {
            searched_previews: false,
            requested_after_ms: query.updated_after_ms,
            requested_before_ms: query.updated_before_ms,
            requested_limit: query.limit,
            oldest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).min(),
            newest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).max(),
            returned: sessions.len(),
            total_matched,
            truncated: next_cursor.is_some(),
        };
        Ok(DiscoveryPage {
            sessions,
            next_cursor,
            receipt,
        })
    }

    /// Add the bounded topic/latest-message previews used by list clients to
    /// an already projected metadata page.
    pub fn enrich_index_page(
        &self,
        query: &DiscoveryQuery,
        mut sessions: Vec<SessionDescriptor>,
    ) -> Result<Vec<SessionDescriptor>> {
        enrich_descriptors(query, &mut sessions, None)?;
        Ok(sessions)
    }

    /// Add list previews using a retained append-aware Codex history index.
    ///
    /// Results are identical to [`Self::enrich_index_page`], while a
    /// long-lived caller avoids reparsing all of `history.jsonl` for every
    /// active transcript write.
    pub fn enrich_index_page_with_codex_history(
        &self,
        query: &DiscoveryQuery,
        mut sessions: Vec<SessionDescriptor>,
        codex_history: &CodexHistoryTopicIndex,
    ) -> Result<Vec<SessionDescriptor>> {
        enrich_descriptors(query, &mut sessions, Some(codex_history))?;
        Ok(sessions)
    }

    /// Discover one stable page and return the cursor for its successor.
    pub fn discover_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
        if query.search_previews {
            return self.discover_preview_page(query);
        }
        let mut page = self.project_index_page(
            query,
            self.scan_descriptors(query, query.include_child_sessions),
        )?;
        if page.sessions.is_empty() && query.cursor.is_none() {
            if let Some(named) = self.claude_session_named(query)? {
                page.sessions.push(named);
            }
        }
        enrich_descriptors(query, &mut page.sessions, None)?;
        Ok(page)
    }

    /// The Claude Code session a query names exactly, when nothing else matched
    /// it. A name given on resume (`claude --resume <id> -n <name>`, `/rename`)
    /// is a record appended wherever the conversation had got to, past the
    /// header discovery reads; so, and only on a miss, the transcripts are
    /// searched for the record itself, and the most recently written one that
    /// carries it wins. The query's other filters still apply.
    fn claude_session_named(&self, query: &DiscoveryQuery) -> Result<Option<SessionDescriptor>> {
        let Some(name) = query
            .query
            .as_deref()
            .map(str::trim)
            .filter(|name| !name.is_empty())
        else {
            return Ok(None);
        };
        if !query.harnesses.is_empty()
            && !query
                .harnesses
                .iter()
                .any(|harness| harness.as_str() == HarnessId::CLAUDE_CODE)
        {
            return Ok(None);
        }
        let needle = format!("\"customTitle\":{}", Value::String(name.to_string()));
        let mut best: Option<(std::time::SystemTime, String)> = None;
        let Ok(projects) = std::fs::read_dir(&query.homes.claude_code) else {
            return Ok(None);
        };
        for project in projects.flatten() {
            let Ok(files) = std::fs::read_dir(project.path()) else {
                continue;
            };
            for file in files.flatten() {
                let path = file.path();
                if path.extension().and_then(|value| value.to_str()) != Some("jsonl") {
                    continue;
                }
                let Ok(bytes) = std::fs::read(&path) else {
                    continue;
                };
                if !bytes
                    .windows(needle.len())
                    .any(|window| window == needle.as_bytes())
                {
                    continue;
                }
                let modified = file
                    .metadata()
                    .and_then(|meta| meta.modified())
                    .unwrap_or(std::time::UNIX_EPOCH);
                let Some(id) = path
                    .file_stem()
                    .map(|stem| stem.to_string_lossy().into_owned())
                else {
                    continue;
                };
                if best.as_ref().is_none_or(|(time, _)| modified > *time) {
                    best = Some((modified, id));
                }
            }
        }
        let Some((_, id)) = best else {
            return Ok(None);
        };
        let mut unnamed = query.clone();
        unnamed.query = None;
        unnamed.harnesses = vec![HarnessId::new(HarnessId::CLAUDE_CODE)];
        unnamed.limit = None;
        let found = self
            .project_index_page(
                &unnamed,
                self.scan_descriptors(&unnamed, unnamed.include_child_sessions),
            )?
            .sessions
            .into_iter()
            .find(|descriptor| descriptor.locator.session_id == id);
        Ok(found.map(|mut descriptor| {
            descriptor.title = Some(name.to_string());
            descriptor
        }))
    }

    fn discover_preview_page(&self, query: &DiscoveryQuery) -> Result<DiscoveryPage> {
        let search = query
            .query
            .as_deref()
            .map(str::trim)
            .filter(|text| !text.is_empty())
            .ok_or_else(|| Error::Other("search_previews requires a nonempty query".into()))?
            .to_lowercase();
        if query.limit == Some(0) {
            return Err(Error::Other("preview search limit must be positive".into()));
        }
        let cursor = query.cursor.as_deref().map(decode_cursor).transpose()?;
        let mut eligible_query = query.clone();
        eligible_query.query = None;
        eligible_query.search_previews = false;
        eligible_query.include_topic_candidates = true;
        let mut eligible = self.scan_descriptors(query, query.include_child_sessions);
        project_descriptors(&eligible_query, &mut eligible);
        // Read Codex's first history topic once per scan, never once per row.
        let topics = codex_history_topics(&query.homes.codex, &eligible).unwrap_or_default();
        let mut sessions = Vec::new();
        let mut total_matched = 0;
        let mut cursor_seen = cursor.is_none();
        let mut more = false;
        for mut descriptor in eligible {
            let metadata_match = descriptor_matches(&descriptor, &search);
            if !metadata_match {
                let topic = topics
                    .get(&descriptor.locator.session_id)
                    .filter(|_| descriptor.locator.harness.as_str() == HarnessId::CODEX);
                enrich_descriptor(&eligible_query, &mut descriptor, topic);
                if !descriptor
                    .preview_candidates
                    .iter()
                    .chain(&descriptor.latest_message_candidates)
                    .any(|candidate| candidate.content.to_lowercase().contains(&search))
                {
                    continue;
                }
            }
            total_matched += 1;
            if !cursor_seen {
                cursor_seen = cursor.as_ref() == Some(&descriptor_cursor_key(&descriptor));
                continue;
            }
            if sessions.len() < query.limit.unwrap_or(usize::MAX) {
                if metadata_match {
                    let topic = topics
                        .get(&descriptor.locator.session_id)
                        .filter(|_| descriptor.locator.harness.as_str() == HarnessId::CODEX);
                    enrich_descriptor(&eligible_query, &mut descriptor, topic);
                }
                sessions.push(descriptor);
            } else {
                more = true;
            }
        }
        if !cursor_seen {
            return Err(Error::Other("discovery cursor is stale or invalid".into()));
        }
        let next_cursor = more.then(|| sessions.last().map(encode_cursor)).flatten();
        let receipt = DiscoveryReceipt {
            searched_previews: true,
            requested_after_ms: query.updated_after_ms,
            requested_before_ms: query.updated_before_ms,
            requested_limit: query.limit,
            oldest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).min(),
            newest_returned_ms: sessions.iter().filter_map(|s| s.updated_at_ms).max(),
            returned: sessions.len(),
            total_matched,
            truncated: more,
        };
        Ok(DiscoveryPage {
            sessions,
            next_cursor,
            receipt,
        })
    }

    fn scan_descriptors(
        &self,
        query: &DiscoveryQuery,
        include_child_sessions: bool,
    ) -> Vec<SessionDescriptor> {
        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,
                HarnessId::OPENCLAW,
                HarnessId::HERMES,
                HarnessId::ORCHESTRATOR,
            ]
            .into_iter()
            .collect()
        } else {
            query.harnesses.iter().map(HarnessId::as_str).collect()
        };
        // Built once per scan: a subtree scope canonicalizes its root once and
        // remembers each recorded cwd it has already judged.
        let scope = WorkspaceScope::of(query);
        let mut found = Vec::new();
        if selected.contains(HarnessId::CLAUDE_CODE) {
            discover_jsonl(
                &query.homes.claude_code,
                HarnessId::CLAUDE_CODE,
                scope.as_ref(),
                include_child_sessions,
                &mut found,
            );
        }
        if selected.contains(HarnessId::CODEX) {
            discover_jsonl(
                &query.homes.codex,
                HarnessId::CODEX,
                scope.as_ref(),
                include_child_sessions,
                &mut found,
            );
        }
        if selected.contains(HarnessId::PI) {
            discover_jsonl(
                &query.homes.pi,
                HarnessId::PI,
                scope.as_ref(),
                include_child_sessions,
                &mut found,
            );
        }
        if selected.contains(HarnessId::OPENCODE) {
            discover_opencode(&query.homes.opencode, scope.as_ref(), &mut found);
        }
        if selected.contains(HarnessId::GROK) {
            discover_grok(&query.homes.grok, scope.as_ref(), &mut found);
        }
        if selected.contains(HarnessId::GEMINI) {
            discover_gemini(&query.homes.gemini, scope.as_ref(), &mut found);
        }
        if selected.contains(HarnessId::GOOSE) {
            discover_goose(&query.homes.goose, scope.as_ref(), &mut found);
        }
        if selected.contains(HarnessId::OPENCLAW) {
            discover_openclaw(&query.homes.openclaw, scope.as_ref(), &mut found);
        }
        if selected.contains(HarnessId::HERMES) {
            for store in hermes_session_stores(&query.homes.hermes) {
                discover_hermes(&store, scope.as_ref(), &selected, &mut found);
            }
        }
        if selected.contains(HarnessId::ORCHESTRATOR) {
            discover_orchestrator(&query.homes.orchestrator, scope.as_ref(), &mut found);
        }
        if selected.contains(HarnessId::SUPERCODE) {
            discover_supercode(&query.homes.supercode, scope.as_ref(), &mut found);
        }
        for descriptor in &mut found {
            finalize_nouns(descriptor);
        }
        found
    }

    /// 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 Some(mut descriptor) = self.refresh_file_index_descriptor(locator, workspace)? else {
            return Ok(None);
        };
        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))
    }

    /// Refresh only stable list metadata for one file-backed descriptor.
    /// This avoids reading conversation preview windows for background index
    /// maintenance.
    pub fn refresh_file_index_descriptor(
        &self,
        locator: &SessionLocator,
        workspace: Option<&Path>,
    ) -> Result<Option<SessionDescriptor>> {
        let scope = workspace.map(WorkspaceScope::exact);
        self.refresh_file_index_descriptor_scoped(locator, scope.as_ref())
    }

    /// [`Self::refresh_file_index_descriptor`] under a whole query's workspace
    /// filter, so a retained index honours `workspace_subtree` exactly as
    /// discovery does.
    pub fn refresh_file_index_descriptor_for(
        &self,
        locator: &SessionLocator,
        query: &DiscoveryQuery,
    ) -> Result<Option<SessionDescriptor>> {
        let scope = WorkspaceScope::of(query);
        self.refresh_file_index_descriptor_scoped(locator, scope.as_ref())
    }

    fn refresh_file_index_descriptor_scoped(
        &self,
        locator: &SessionLocator,
        workspace: Option<&WorkspaceScope>,
    ) -> 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| !wanted.admits(cwd)))
        {
            return Ok(None);
        }
        let parent_session_id = meta.parent_session_id.or_else(|| {
            (locator.harness.as_str() == HarnessId::CLAUDE_CODE)
                .then(|| claude_subagent_parent_id(path))
                .flatten()
        });
        let tail = tail_facts(path, locator.harness.as_str());
        let 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: tail.last_turn_ms.or_else(|| modified_ms(path)),
            message_count: None,
            model: tail.model.or(meta.model),
            parent_session_id,
            child_session_count: 0,
            nouns: OrchestrationNouns::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> {
        if let Some(session) = load_hermes_locator(locator) {
            return 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> {
        if let Some(session) = load_hermes_locator(locator) {
            return 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> {
        if let Some(session) = load_hermes_locator(locator) {
            let mut session = session?;
            if session.messages.len() > message_limit.max(1) {
                session
                    .messages
                    .drain(..session.messages.len() - message_limit.max(1));
            }
            return Ok(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,
        )
    }
}

/// ORCH-6: a Hermes row's durable address is the pair `{state.db, session
/// id}` — one store holds every session — so the generic file loader (which
/// opens the store's most recent session) would answer with the WRONG
/// conversation for every discovered locator but one, and its nouns with it.
/// Route by the locator's own session id instead.
fn load_hermes_locator(locator: &SessionLocator) -> Option<Result<Session>> {
    if locator.harness.as_str() != HarnessId::HERMES {
        return None;
    }
    let StorageLocator::File { path } = &locator.storage else {
        return None;
    };
    Some(Session::from_hermes_sqlite(path, Some(&locator.session_id)))
}

/// ORCH-6: resolve every discovered row's `trigger` and `workspace` through
/// ORCH-3's own derivation. Harness scanners fill only what their native store
/// states (`surface`, `profile`, `recurrence`, `cross_surface`, and an explicit
/// `trigger` when the source says); this pass rebuilds a `SessionMeta` from
/// those facts plus `cwd` and re-reads the nouns off it, so a discovered row
/// and a loaded session can never disagree about the same session.
fn finalize_nouns(descriptor: &mut SessionDescriptor) {
    let mut meta = SessionMeta::new(SessionSource::Native);
    meta.cwd = descriptor.cwd.clone();
    meta.trigger = descriptor.nouns.trigger;
    meta.surface = descriptor.nouns.surface.clone();
    meta.profile = descriptor.nouns.profile.clone();
    meta.recurrence = descriptor.nouns.recurrence.clone();
    meta.cross_surface = descriptor.nouns.cross_surface.clone();
    descriptor.nouns = OrchestrationNouns::from_meta(&meta);
}

fn project_descriptors(query: &DiscoveryQuery, found: &mut Vec<SessionDescriptor>) {
    roll_up_session_children(found, query.include_child_sessions);
    if let Some(root_session_id) = query.root_session_id.as_deref() {
        retain_session_family(found, root_session_id);
    }
    if let Some(family_path) = query.workspace_family.as_deref() {
        let family = RepoFamily::of(family_path);
        let mut cache: HashMap<PathBuf, bool> = HashMap::new();
        found.retain(|descriptor| {
            let Some(cwd) = descriptor.cwd.as_deref() else {
                return false;
            };
            *cache
                .entry(cwd.to_path_buf())
                .or_insert_with(|| RepoFamily::of(cwd).joins(&family))
        });
    }
    if let Some(profile) = query
        .profile
        .as_deref()
        .map(str::trim)
        .filter(|p| !p.is_empty())
    {
        found.retain(|descriptor| descriptor.nouns.profile.as_deref() == Some(profile));
    }
    if let Some(after) = query.updated_after_ms {
        found.retain(|descriptor| descriptor.updated_at_ms.is_some_and(|at| at >= after));
    }
    if let Some(before) = query.updated_before_ms {
        found.retain(|descriptor| descriptor.updated_at_ms.is_some_and(|at| at <= before));
    }
    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(|query| !query.is_empty())
    {
        let search = search.to_lowercase();
        found.retain(|descriptor| descriptor_matches(descriptor, &search));
    }
}

fn paginate_descriptors(
    query: &DiscoveryQuery,
    found: Vec<SessionDescriptor>,
) -> Result<(Vec<SessionDescriptor>, Option<String>)> {
    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 sessions = found[start.min(found.len())..end].to_vec();
    let next_cursor = (end < found.len())
        .then(|| sessions.last().map(encode_cursor))
        .flatten();
    Ok((sessions, next_cursor))
}

fn enrich_descriptors(
    query: &DiscoveryQuery,
    sessions: &mut [SessionDescriptor],
    codex_history: Option<&CodexHistoryTopicIndex>,
) -> Result<()> {
    let codex_topics = if query.include_topic_candidates && codex_history.is_none() {
        codex_history_topics(&query.homes.codex, sessions).unwrap_or_default()
    } else {
        HashMap::new()
    };
    for descriptor in sessions {
        let topic = (descriptor.locator.harness.as_str() == HarnessId::CODEX)
            .then(|| {
                codex_history
                    .and_then(|history| history.topics.get(&descriptor.locator.session_id))
                    .or_else(|| codex_topics.get(&descriptor.locator.session_id))
            })
            .flatten();
        enrich_descriptor(query, descriptor, topic);
    }
    Ok(())
}

fn enrich_descriptor(
    query: &DiscoveryQuery,
    descriptor: &mut SessionDescriptor,
    codex_topic: Option<&Vec<SessionPreviewCandidate>>,
) {
    if query.include_topic_candidates {
        descriptor.preview_candidates = codex_topic
            .cloned()
            .unwrap_or_else(|| topic_message_candidates(&descriptor.locator).unwrap_or_default());
    }
    descriptor.latest_message_candidates =
        latest_message_candidates(&descriptor.locator).unwrap_or_default();
}

fn is_false(value: &bool) -> bool {
    !value
}

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<&WorkspaceScope>,
    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| !wanted.admits(cwd)))
        {
            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()
        });
        let tail = tail_facts(&path, harness);
        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: tail.last_turn_ms.or_else(|| modified_ms(&path)),
            message_count: None,
            model: tail.model.or(meta.model),
            parent_session_id,
            child_session_count: if harness == HarnessId::CLAUDE_CODE && !include_child_sessions {
                count_claude_subagents(&path)
            } else {
                0
            },
            nouns: OrchestrationNouns::default(),
        });
    }
}

fn collect_jsonl(root: &Path, harness: &str, include_child_sessions: bool, out: &mut Vec<PathBuf>) {
    let mut walked = HashSet::new();
    collect_jsonl_in(root, harness, include_child_sessions, out, &mut walked);
}

/// A symlinked directory is walked like a real one: a harness home whose project directories
/// were moved to another volume and linked back (a box offloading its disk) still holds every
/// one of those sessions. `walked` holds each directory's canonical path, so a link cycle ends.
fn collect_jsonl_in(
    root: &Path,
    harness: &str,
    include_child_sessions: bool,
    out: &mut Vec<PathBuf>,
    walked: &mut HashSet<PathBuf>,
) {
    if !walked.insert(fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf())) {
        return;
    }
    let Ok(entries) = fs::read_dir(root) else {
        return;
    };
    for entry in entries.flatten() {
        let Ok(mut kind) = entry.file_type() else {
            continue;
        };
        let path = entry.path();
        if kind.is_symlink() {
            let Ok(target) = fs::metadata(&path) else {
                continue;
            };
            kind = target.file_type();
        }
        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_in(&path, harness, include_child_sessions, out, walked);
        } else if kind.is_file()
            && (path.extension().and_then(|v| v.to_str()) == Some("jsonl")
                || (harness == HarnessId::CODEX
                    && path
                        .file_name()
                        .and_then(|v| v.to_str())
                        .is_some_and(|name| name.ends_with(".jsonl.zst"))))
        {
            out.push(path);
        }
    }
}

fn read_header(path: &Path, harness: &str) -> Result<HeaderMeta> {
    let file = crate::session::open_session_reader(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"));
            // The session's name (`claude -n`, `/rename`) is its title.
            fill_string(&mut result.title, value.get("customTitle"));
            fill_string(&mut result.title, value.get("agentName"));
            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<&WorkspaceScope>,
    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<&WorkspaceScope>,
) -> 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| !wanted.admits(actual))) {
        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: tail_facts(path, HarnessId::GEMINI)
            .last_turn_ms
            .or_else(|| modified_ms(path)),
        message_count: None,
        model,
        parent_session_id: None,
        child_session_count: 0,
        nouns: OrchestrationNouns::default(),
    })
}

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,
    }
}

/// Hermes (UNI-15): enumerate sessions from the single `state.db` SQLite
/// store, strictly read-only (the store is a live, shared, WAL,
/// single-writer database owned by a running Hermes install). A missing or
/// non-Hermes file is skipped silently, like every other absent home.
fn discover_hermes(
    db_path: &Path,
    workspace: Option<&WorkspaceScope>,
    selected: &HashSet<&str>,
    found: &mut Vec<SessionDescriptor>,
) {
    if !db_path.is_file() {
        return;
    }
    let Ok(conn) = Connection::open_with_flags(
        db_path,
        rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
    ) else {
        return;
    };
    let fingerprint_ok = ["sessions", "messages", "schema_version"].iter().all(|t| {
        conn.query_row(
            "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?1",
            [t],
            |_| Ok(()),
        )
        .is_ok()
    });
    if !fingerprint_ok {
        return;
    }
    let Ok(mut statement) = conn.prepare(
        "SELECT id, cwd, title, model, message_count, started_at, ended_at, parent_session_id, \
         source, model_config FROM sessions ORDER BY started_at DESC",
    ) else {
        return;
    };
    let Ok(rows) = statement.query_map([], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, Option<String>>(1)?,
            row.get::<_, Option<String>>(2)?,
            row.get::<_, Option<String>>(3)?,
            row.get::<_, Option<i64>>(4)?,
            row.get::<_, Option<f64>>(5)?,
            row.get::<_, Option<f64>>(6)?,
            row.get::<_, Option<String>>(7)?,
            row.get::<_, Option<String>>(8)?,
            row.get::<_, Option<String>>(9)?,
        ))
    }) else {
        return;
    };
    for row in rows.flatten() {
        let (
            id,
            cwd,
            title,
            model,
            message_count,
            started_at,
            ended_at,
            parent,
            source,
            model_config,
        ) = row;
        // a worker session the orchestrator mirrored into this store (hermes-compat row 11) is listed once:
        // as the worker's own session when this query also reads that worker's store, else here, as Hermes
        // lists it; one Hermes has carried past its mirror is Hermes's conversation too, and listed here
        let mirror = model_config
            .as_deref()
            .and_then(|c| serde_json::from_str::<serde_json::Value>(c).ok())
            .and_then(|c| c.get("_supercode_mirror").cloned());
        if let Some(mirror) = mirror {
            let worker_read = mirror
                .get("harness")
                .and_then(serde_json::Value::as_str)
                .is_some_and(|h| selected.contains(h));
            let mirrored = mirror.get("messages").and_then(serde_json::Value::as_i64);
            if worker_read && mirrored.is_none_or(|m| message_count.unwrap_or(0) <= m) {
                continue;
            }
            if mirror.get("continued_as").is_some()
                && mirrored.is_none_or(|m| message_count.unwrap_or(0) <= m)
            {
                continue;
            }
        }
        let cwd = cwd.map(PathBuf::from);
        if let Some(filter) = workspace {
            if !filter.admits_literal(cwd.as_deref()) {
                continue;
            }
        }
        let updated_at_ms = ended_at
            .or(started_at)
            .map(|seconds| (seconds * 1000.0) as u64);
        // ORCH-6: the same derivation `Session::from_hermes_sqlite` runs, over
        // the same row — discovery reads the gateway columns it already has
        // open instead of guessing a lighter-weight variant.
        let mut meta = SessionMeta::new(SessionSource::Hermes);
        meta.cwd = cwd.clone();
        if let Some(hermes_source) = source.filter(|value| !value.is_empty()) {
            meta.lineage
                .insert("hermes_source".to_string(), hermes_source);
        }
        if let Some(parent_id) = parent.as_deref() {
            meta.lineage.insert(
                "hermes_lineage_kind".to_string(),
                crate::session::hermes_lineage_kind(
                    &conn,
                    parent_id,
                    model_config.as_deref(),
                    started_at,
                )
                .to_string(),
            );
        }
        hermes_capture_nouns(&conn, &id, &mut meta);
        found.push(SessionDescriptor {
            locator: SessionLocator {
                harness: HarnessId::new(HarnessId::HERMES),
                session_id: id,
                storage: StorageLocator::File {
                    path: db_path.to_path_buf(),
                },
            },
            cwd,
            title: title.filter(|t| !t.is_empty()),
            preview_candidates: Vec::new(),
            latest_message_candidates: Vec::new(),
            updated_at_ms,
            message_count: message_count.map(|count| count.max(0) as usize),
            model,
            parent_session_id: parent,
            child_session_count: 0,
            nouns: OrchestrationNouns::from_meta(&meta),
        });
    }
}

/// The orchestrator (ORC-7): every profile folder's `state.db` `bindings`
/// table is one row per conversation the orchestrator holds. A binding is not
/// a transcript — the transcript belongs to the WORKER harness it points at —
/// so the row carries the surface, trigger, profile and worker identity, and
/// its locator addresses the worker's own storage when the binding recorded
/// one. Strictly read-only, like every other store here.
fn discover_orchestrator(
    root: &Path,
    workspace: Option<&WorkspaceScope>,
    found: &mut Vec<SessionDescriptor>,
) {
    // A binding has no cwd of its own; a workspace filter can only exclude it.
    if workspace.is_some() {
        return;
    }
    for (profile, dir) in orchestrator_profile_dirs(root) {
        let db_path = dir.join("state.db");
        if !db_path.is_file() {
            continue;
        }
        let Ok(conn) = Connection::open_with_flags(
            &db_path,
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY | rusqlite::OpenFlags::SQLITE_OPEN_NO_MUTEX,
        ) else {
            continue;
        };
        // the session entries the orchestrator wrote into Hermes's `gateway_routing` (hermes-compat row 10),
        // each binding whole in `metadata.supercode`; the rest are Hermes's own sessions, listed by Hermes's reader
        let sessions = dir.join("sessions");
        let scope = std::fs::canonicalize(&sessions)
            .unwrap_or(sessions)
            .display()
            .to_string();
        let Ok(mut statement) = conn.prepare(
            "SELECT json_extract(entry_json, '$.metadata.supercode.binding'), \
             CAST(strftime('%s', json_extract(entry_json, '$.metadata.supercode.binding.last_activity_at')) AS INTEGER) \
             FROM gateway_routing WHERE scope = ?1 AND json_extract(entry_json, '$.metadata.supercode') IS NOT NULL \
             ORDER BY json_extract(entry_json, '$.metadata.supercode.binding.last_activity_at') DESC",
        ) else {
            continue;
        };
        let Ok(rows) = statement.query_map([&scope], |row| {
            Ok((
                row.get::<_, Option<String>>(0)?,
                row.get::<_, Option<i64>>(1)?,
            ))
        }) else {
            continue;
        };
        let rows = rows.flatten().filter_map(|(json, epoch)| {
            let b: Binding = serde_json::from_str(&json?).ok()?;
            let text = |v: &Option<String>| v.clone().filter(|s| !s.is_empty());
            Some((
                OrchestratorBindingRow {
                    platform: b.key.platform.clone().unwrap_or_default(),
                    chat_type: b.key.kind.clone().unwrap_or_default(),
                    chat_id: text(&b.key.chat_id),
                    thread_id: text(&b.key.thread_id),
                    participant_id: text(&b.key.participant_id),
                    worker_harness: b.worker.harness.as_str().to_string(),
                    worker_session_id: text(&b.worker.session_id),
                    worker_locator: text(&b.worker.locator),
                    started_at: b.started_at.clone(),
                    last_activity_at: b.last_activity_at.clone(),
                    ended_at: b.ended_at.clone(),
                    end_reason: b.end_reason.map(|r| r.as_str().to_string()),
                    handoff_to: b.handoff.as_ref().and_then(|h| h.to.clone()),
                    handoff_state: b.handoff.as_ref().map(|h| h.state.clone()),
                    handoff_error: b.handoff.as_ref().and_then(|h| h.error.clone()),
                    recurrence_job_id: b.recurrence.as_ref().map(|r| r.job_id.clone()),
                },
                epoch,
            ))
        });
        for (row, last_activity_epoch) in rows {
            found.push(orchestrator_descriptor(
                &db_path,
                &profile,
                &row,
                last_activity_epoch,
            ));
        }
    }
}

fn orchestrator_descriptor(
    db_path: &Path,
    profile: &str,
    row: &OrchestratorBindingRow,
    last_activity_epoch: Option<i64>,
) -> SessionDescriptor {
    let binding = Binding::from_orchestrator_row(profile, row);
    let nouns = binding.nouns();
    // The row's title names the worker the binding points at: that pair is
    // the only address from which the conversation itself can be read.
    let mut title = format!(
        "{} {}",
        row.worker_harness,
        row.worker_session_id
            .as_deref()
            .unwrap_or("(no worker session yet)")
    );
    if let Some(reason) = row.end_reason.as_deref().filter(|_| row.ended_at.is_some()) {
        title.push_str(&format!(" (ended: {reason})"));
    }
    SessionDescriptor {
        locator: SessionLocator {
            harness: HarnessId::new(HarnessId::ORCHESTRATOR),
            session_id: row.worker_session_id.clone().unwrap_or_default(),
            storage: StorageLocator::File {
                path: row
                    .worker_locator
                    .clone()
                    .map_or_else(|| db_path.to_path_buf(), PathBuf::from),
            },
        },
        cwd: None,
        title: Some(title),
        preview_candidates: Vec::new(),
        latest_message_candidates: Vec::new(),
        updated_at_ms: last_activity_epoch.map(|seconds| (seconds.max(0) as u64) * 1000),
        message_count: None,
        model: None,
        parent_session_id: None,
        child_session_count: 0,
        nouns,
    }
}

/// OpenClaw >= 2026.7: `<home>/agents/<agentId>/sessions/<uuid>.jsonl` are
/// plain pi-v3 dialect session files. `.trajectory.jsonl` runtime traces and
/// `.trajectory-path.json` pointers live in the SAME directory and are
/// excluded by suffix plus a header check (their first line carries
/// `traceSchema`, never `type:"session"`). Read-only discovery (UNI-16).
fn discover_openclaw(
    root: &Path,
    workspace: Option<&WorkspaceScope>,
    found: &mut Vec<SessionDescriptor>,
) {
    let agents = root.join("agents");
    let Ok(agent_dirs) = std::fs::read_dir(&agents) else {
        return;
    };
    for agent_dir in agent_dirs.flatten() {
        let sessions = agent_dir.path().join("sessions");
        let Ok(files) = std::fs::read_dir(&sessions) else {
            continue;
        };
        for file in files.flatten() {
            let path = file.path();
            let name = file.file_name();
            let name = name.to_string_lossy();
            if !name.ends_with(".jsonl") || name.ends_with(".trajectory.jsonl") {
                continue;
            }
            let Ok(text) = std::fs::read_to_string(&path) else {
                continue;
            };
            let Some(header_line) = text.lines().find(|line| !line.trim().is_empty()) else {
                continue;
            };
            let Ok(header) = serde_json::from_str::<serde_json::Value>(header_line) else {
                continue;
            };
            if header.get("type").and_then(serde_json::Value::as_str) != Some("session") {
                continue;
            }
            let session_id = header
                .get("id")
                .and_then(serde_json::Value::as_str)
                .unwrap_or_else(|| name.trim_end_matches(".jsonl"))
                .to_string();
            let cwd = header
                .get("cwd")
                .and_then(serde_json::Value::as_str)
                .map(PathBuf::from);
            if let Some(filter) = workspace {
                if !filter.admits_literal(cwd.as_deref()) {
                    continue;
                }
            }
            let updated_at_ms = tail_facts(&path, HarnessId::OPENCLAW)
                .last_turn_ms
                .or_else(|| {
                    file.metadata()
                        .ok()
                        .and_then(|metadata| metadata.modified().ok())
                        .and_then(|modified| modified.duration_since(std::time::UNIX_EPOCH).ok())
                        .map(|elapsed| elapsed.as_millis() as u64)
                });
            let message_count = text
                .lines()
                .filter(|line| line.contains("\"type\":\"message\""))
                .count();
            // ORCH-6: the same two facts `Session::load` reads for an OpenClaw
            // file — the gateway `sessionKey` in the header, and the agent id
            // in `agents/<id>/`.
            let mut meta = SessionMeta::new(SessionSource::OpenClaw);
            meta.cwd = cwd.clone();
            openclaw_capture_header_nouns(&header, &mut meta);
            if meta.profile.is_none() {
                meta.profile = openclaw_agent_id_from_path(&path);
            }
            found.push(SessionDescriptor {
                locator: SessionLocator {
                    harness: HarnessId::new(HarnessId::OPENCLAW),
                    session_id,
                    storage: StorageLocator::File { path },
                },
                cwd,
                title: None,
                preview_candidates: Vec::new(),
                latest_message_candidates: Vec::new(),
                updated_at_ms,
                message_count: Some(message_count),
                model: None,
                parent_session_id: None,
                child_session_count: 0,
                nouns: OrchestrationNouns::from_meta(&meta),
            });
        }
    }
}

fn discover_supercode(
    root: &Path,
    workspace: Option<&WorkspaceScope>,
    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))
        };
        // The sidecar is the store's authoritative content whenever it exists
        // (see `read_native_store_header`), and a reduced session can outlive its
        // working transcript entirely. Address the file that IS there: a locator
        // naming a deleted `<name>.jsonl` is one discovery's own loader rejects.
        let sidecar = path.with_extension("sidecar.jsonl");
        let path = if path.is_file() {
            path
        } else {
            sidecar.clone()
        };
        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| !wanted.admits(cwd))
        }) {
            continue;
        }
        let title = (!info.title.trim().is_empty()).then_some(info.title);
        // The native store's own records carry no clock; the sidecar beside it
        // stamps every turn. Prefer that, and keep mtime as the last resort.
        let updated_at_ms = tail_facts(&sidecar, HarnessId::SUPERCODE)
            .last_turn_ms
            .or_else(|| tail_facts(&path, HarnessId::SUPERCODE).last_turn_ms)
            .or_else(|| modified_ms(&path))
            .or_else(|| modified_ms(&sidecar));
        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,
            nouns: OrchestrationNouns::default(),
        });
    }
}

/// 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<&WorkspaceScope>,
    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| !wanted.admits(&cwd)) {
            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,
                nouns: OrchestrationNouns::default(),
            });
        }
    }
}

fn discover_opencode(
    root: &Path,
    workspace: Option<&WorkspaceScope>,
    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| !wanted.admits(&cwd)) {
                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,
                nouns: OrchestrationNouns::default(),
            });
        }
    }
}

fn discover_goose(
    root: &Path,
    workspace: Option<&WorkspaceScope>,
    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| !wanted.admits(&cwd)) {
            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,
            nouns: OrchestrationNouns::default(),
        });
    }
}

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 {
        // A Hermes locator addresses one session inside the whole-store `state.db`; the file's
        // tail is SQLite pages, not transcript lines, so the preview comes from the store by id.
        StorageLocator::File { path } | StorageLocator::Sqlite { path, .. }
            if locator.harness.as_str() == HarnessId::HERMES =>
        {
            latest_hermes_message_candidates(path, &locator.session_id)
        }
        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()
        ))
    })?;
    if harness == HarnessId::CODEX {
        return Ok(codex_preview_candidates(text.lines(), false));
    }
    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()
        ))
    })?;
    if harness == HarnessId::CODEX {
        return Ok(codex_preview_candidates(text.lines().rev(), true));
    }
    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::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_with_cursor(
            &mut candidates,
            role,
            content,
            metadata,
            Some(message_candidate_cursor(harness, &value)),
        );
        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
            break;
        }
    }
    Ok(candidates)
}

struct CodexPreviewRecord {
    native: Value,
    role: String,
    text: String,
}

// Pair adjacent conversational event/response mirrors one-to-one, within the
// bytes already read. This is intentionally narrower than the full codec's
// global assistant-text dedup: repeated same-kind records and separate pairs
// remain separate turns. Compare full display text BEFORE the 4096-character
// cap; prefer the canonical response's cursor/timestamp in either scan direction.
fn codex_preview_candidates<'a>(
    lines: impl Iterator<Item = &'a str>,
    latest: bool,
) -> Vec<SessionPreviewCandidate> {
    let mut candidates = Vec::new();
    let mut pending: Option<CodexPreviewRecord> = None;
    for line in lines {
        let Ok(native) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        let Some((role, content)) = codex_preview_message(&native) else {
            continue;
        };
        let Some(text) = display_text(content).filter(|text| !text.trim().is_empty()) else {
            continue;
        };
        let current = CodexPreviewRecord {
            role: role.to_string(),
            text,
            native,
        };
        if let Some(previous) = pending.take() {
            if previous.role == current.role
                && previous.text == current.text
                && previous.native.get("type") != current.native.get("type")
            {
                let canonical = if previous.native.get("type").and_then(Value::as_str)
                    == Some("response_item")
                {
                    previous
                } else {
                    current
                };
                push_codex_preview_candidate(&mut candidates, canonical, latest);
            } else {
                push_codex_preview_candidate(&mut candidates, previous, latest);
                pending = Some(current);
            }
        } else {
            pending = Some(current);
        }
        if candidates.len() >= LATEST_PREVIEW_CANDIDATES {
            break;
        }
    }
    if let Some(last) = pending {
        push_codex_preview_candidate(&mut candidates, last, latest);
    }
    candidates
}

fn push_codex_preview_candidate(
    candidates: &mut Vec<SessionPreviewCandidate>,
    record: CodexPreviewRecord,
    latest: bool,
) {
    let mut metadata = HashMap::new();
    if latest {
        if let Some(timestamp) = record.native.get("timestamp").and_then(Value::as_str) {
            metadata.insert("timestamp".to_string(), timestamp.to_string());
        }
    }
    let cursor = latest.then(|| message_candidate_cursor(HarnessId::CODEX, &record.native));
    push_message_candidate_with_cursor(
        candidates,
        &record.role,
        Some(&Value::String(record.text)),
        metadata,
        cursor,
    );
}

// Codex collab rollouts can carry narration only as event_msg records.
fn codex_preview_message(value: &Value) -> Option<(&str, Option<&Value>)> {
    let payload = value.get("payload")?;
    match (
        value.get("type").and_then(Value::as_str)?,
        payload.get("type").and_then(Value::as_str)?,
    ) {
        ("response_item", "message") => {
            let role @ ("user" | "assistant") = payload.get("role").and_then(Value::as_str)? else {
                return None;
            };
            Some((role, payload.get("content")))
        }
        ("event_msg", "user_message") => Some(("user", payload.get("message"))),
        ("event_msg", "agent_message") => Some(("assistant", payload.get("message"))),
        _ => None,
    }
}

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,
            )
        }
        _ => 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_hermes_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 FROM messages WHERE session_id = ?1 AND active = 1 \
             AND role IN ('user', 'assistant') AND content IS NOT NULL AND content != '' \
             ORDER BY timestamp DESC, id 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 (role, content) in rows.flatten() {
        push_message_candidate(
            &mut candidates,
            &role,
            Some(&Value::String(content)),
            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>,
) {
    push_message_candidate_with_cursor(candidates, role, content, metadata, None);
}

fn push_message_candidate_with_cursor(
    candidates: &mut Vec<SessionPreviewCandidate>,
    role: &str,
    content: Option<&Value>,
    metadata: HashMap<String, String>,
    cursor: Option<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 {
        cursor,
        role: role.to_string(),
        content: text.chars().take(MAX_CHARS).collect(),
        metadata,
    });
}

fn message_candidate_cursor(harness: &str, value: &Value) -> String {
    let native_identity = value
        .get("uuid")
        .or_else(|| value.get("id"))
        .or_else(|| value.pointer("/message/id"))
        .or_else(|| value.pointer("/payload/id"))
        .and_then(Value::as_str)
        .or_else(|| value.get("timestamp").and_then(Value::as_str));
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"supercode.session-preview-cursor.v1\0");
    hasher.update(harness.as_bytes());
    hasher.update(b"\0");
    if let Some(identity) = native_identity {
        hasher.update(identity.as_bytes());
    } else {
        // Some formats do not publish message ids. Hashing the complete native
        // record is still stable across discovery refreshes and reveals none
        // of the record itself to an untrusted presentation surface.
        hasher.update(value.to_string().as_bytes());
    }
    format!("v1:{}", &hasher.finalize().to_hex()[..24])
}

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);
    }
}

/// What the END of a JSONL transcript says about a session: when it last took
/// a turn, and which model took it.
#[derive(Default)]
struct TailFacts {
    /// Newest `timestamp` among the last records, as Unix epoch milliseconds.
    last_turn_ms: Option<u64>,
    /// Model on the LAST turn that named one.
    model: Option<String>,
}

/// Read [`TailFacts`] from the last records of a JSONL transcript.
///
/// Both facts have to come from the conversation rather than from cheaper
/// stand-ins. Recency is not the file's mtime: harnesses touch a transcript
/// without saying anything (see [`SessionDescriptor::updated_at_ms`]). The model
/// is not the one in the header either — every dialect here records the model
/// per turn, so a mid-session switch (`/model`) leaves the opening record naming
/// a model the session has not used for hours.
///
/// Reading whole files is not an option — one catalog holds thousands of
/// sessions and a single transcript runs to tens of megabytes — so this seeks to
/// the end and walks backwards over a bounded window, which is where an
/// append-only log keeps its newest records. The window grows only while no
/// timestamp has been found, and gives up at [`TAIL_SCAN_LIMIT`]; a model the
/// window does not reach stays `None` and the caller keeps the header's.
fn tail_facts(path: &Path, harness: &str) -> TailFacts {
    let mut facts = TailFacts::default();
    let Ok(mut file) = File::open(path) else {
        return facts;
    };
    let Ok(len) = file.metadata().map(|meta| meta.len()) else {
        return facts;
    };
    let mut window = TAIL_SCAN_START.min(len);
    loop {
        if file.seek(SeekFrom::Start(len - window)).is_err() {
            return facts;
        }
        let Ok(size) = usize::try_from(window) else {
            return facts;
        };
        let mut buf = vec![0u8; size];
        if file.read_exact(&mut buf).is_err() {
            return facts;
        }
        // A transcript is append-only, so the newest record is the LAST one:
        // walk line boundaries backwards from the end and decode one record at a
        // time, stopping as soon as both facts are in hand. Decoding the whole
        // window instead would put a UTF-8 validation of every byte of every
        // transcript in the catalog on the path of one `discover`.
        //
        // `end` is the exclusive end of the line under inspection; the scan stops
        // at `floor`, because a window that starts mid-file almost certainly
        // starts mid-record and that partial first line belongs to the next,
        // wider window.
        let floor = if window < len {
            buf.iter().position(|byte| *byte == b'\n').map(|at| at + 1)
        } else {
            Some(0)
        };
        if let Some(floor) = floor {
            let mut end = buf.len();
            while end > floor && !(facts.last_turn_ms.is_some() && facts.model.is_some()) {
                let start = buf[floor..end]
                    .iter()
                    .rposition(|byte| *byte == b'\n')
                    .map_or(floor, |at| floor + at + 1);
                if let Ok(record) = std::str::from_utf8(&buf[start..end])
                    .map_err(|_| ())
                    .and_then(|line| serde_json::from_str::<Value>(line).map_err(|_| ()))
                {
                    if facts.last_turn_ms.is_none() {
                        facts.last_turn_ms = record_timestamp(&record, harness)
                            .and_then(crate::sidecar::rfc3339_to_ms)
                            .and_then(|millis| u64::try_from(millis).ok());
                    }
                    if facts.model.is_none() {
                        facts.model = record_model(&record, harness).map(str::to_owned);
                    }
                }
                end = start.saturating_sub(1);
            }
        }
        if facts.last_turn_ms.is_some() || window >= len || window >= TAIL_SCAN_LIMIT {
            return facts;
        }
        window = (window * 2).min(len).min(TAIL_SCAN_LIMIT);
    }
}

/// When one transcript record was written, in the dialect that wrote it.
/// Every dialect discovery reads stamps its records with an RFC3339 UTC string;
/// only the key differs.
fn record_timestamp<'a>(record: &'a Value, harness: &str) -> Option<&'a str> {
    let key = match harness {
        HarnessId::SUPERCODE => "ts",
        _ => "timestamp",
    };
    record.get(key)?.as_str()
}

/// The model one transcript record names, in the dialect that wrote it.
/// Mirrors the model arms of [`update_header_meta`], read newest-first.
fn record_model<'a>(record: &'a Value, harness: &str) -> Option<&'a str> {
    match harness {
        HarnessId::CLAUDE_CODE | HarnessId::PI => record.get("message")?.get("model")?.as_str(),
        HarnessId::CODEX => {
            if record.get("type")?.as_str()? != "turn_context" {
                return None;
            }
            record.get("payload")?.get("model")?.as_str()
        }
        _ => None,
    }
}

/// Bytes read from a transcript's tail on the first attempt: comfortably more
/// than one record in every dialect discovery reads.
const TAIL_SCAN_START: u64 = 16 * 1024;

/// Where the widening tail scan stops. A transcript whose last mebibyte holds
/// no timestamped record is not one whose recency this can honestly report.
const TAIL_SCAN_LIMIT: u64 = 1024 * 1024;

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 discovery's workspace filter: the exact recorded folder (the default), or
/// the folder and every folder under it (`workspace_subtree`).
pub struct WorkspaceScope {
    wanted: PathBuf,
    /// Comparison keys of `wanted` (canonical and lexical) for subtree mode.
    roots: Vec<PathBuf>,
    subtree: bool,
    /// Subtree verdicts per recorded cwd: many sessions share one folder.
    judged: std::sync::Mutex<HashMap<PathBuf, bool>>,
}

impl WorkspaceScope {
    /// The exact-folder filter every caller had before `workspace_subtree`.
    pub fn exact(wanted: &Path) -> Self {
        Self {
            wanted: wanted.to_path_buf(),
            roots: Vec::new(),
            subtree: false,
            judged: std::sync::Mutex::new(HashMap::new()),
        }
    }

    /// The folder and every folder under it.
    pub fn subtree(root: &Path) -> Self {
        Self {
            wanted: root.to_path_buf(),
            roots: path_comparison_keys(root),
            subtree: true,
            judged: std::sync::Mutex::new(HashMap::new()),
        }
    }

    /// The scope a query asks for, if it filters by workspace at all.
    pub fn of(query: &DiscoveryQuery) -> Option<Self> {
        query.workspace.as_deref().map(|wanted| {
            if query.workspace_subtree {
                Self::subtree(wanted)
            } else {
                Self::exact(wanted)
            }
        })
    }

    /// Whether a session whose recorded working directory is `recorded`
    /// belongs to this scope. A relative recorded cwd never does, nor (on
    /// Windows) one without a drive or UNC prefix: neither says where the
    /// session ran.
    pub fn admits(&self, recorded: &Path) -> bool {
        if !recorded.is_absolute() {
            return false;
        }
        if !self.subtree {
            return same_path(recorded, &self.wanted);
        }
        let mut judged = self
            .judged
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        *judged.entry(recorded.to_path_buf()).or_insert_with(|| {
            path_comparison_keys(recorded)
                .iter()
                .any(|key| self.roots.iter().any(|root| key.starts_with(root)))
        })
    }

    /// Stores that historically compared the recorded cwd literally (Hermes,
    /// OpenClaw) keep that exact comparison; a subtree scope applies
    /// [`Self::admits`].
    fn admits_literal(&self, recorded: Option<&Path>) -> bool {
        if self.subtree {
            recorded.is_some_and(|cwd| self.admits(cwd))
        } else {
            recorded == Some(self.wanted.as_path())
        }
    }
}

/// The forms of `path` a subtree comparison accepts: its canonical form when
/// it exists on this machine (resolving symlinks and aliases) and its lexical
/// absolute form (a deleted folder still names where a session ran). Windows
/// verbatim prefixes are dropped and case is folded there, as the filesystem
/// does.
fn path_comparison_keys(path: &Path) -> Vec<PathBuf> {
    let mut keys = Vec::with_capacity(2);
    if let Ok(canonical) = fs::canonicalize(path) {
        keys.push(comparison_key(&canonical));
    }
    let lexical = comparison_key(&normalize_path(path));
    if !keys.contains(&lexical) {
        keys.push(lexical);
    }
    keys
}

#[cfg(windows)]
fn comparison_key(path: &Path) -> PathBuf {
    let text = path.to_string_lossy();
    let text = if let Some(unc) = text.strip_prefix(r"\\?\UNC\") {
        format!(r"\\{unc}")
    } else if let Some(local) = text.strip_prefix(r"\\?\") {
        local.to_string()
    } else {
        text.into_owned()
    };
    PathBuf::from(text.to_lowercase())
}

#[cfg(not(windows))]
fn comparison_key(path: &Path) -> PathBuf {
    path.to_path_buf()
}

/// 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::io::Write;
    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 codex_history_index_reads_appends_and_repairs_replacements() {
        let root = temp_dir("codex-history-index");
        let sessions = root.join("sessions");
        fs::create_dir_all(&sessions).unwrap();
        let history = root.join("history.jsonl");
        fs::write(
            &history,
            "{\"session_id\":\"alpha\",\"text\":\"first topic\"}\n",
        )
        .unwrap();

        let mut index = CodexHistoryTopicIndex::new(&sessions);
        assert_eq!(index.refresh().unwrap(), BTreeSet::from(["alpha".into()]));
        assert_eq!(index.topics["alpha"][0].content, "first topic");
        assert!(index.refresh().unwrap().is_empty());

        let mut file = fs::OpenOptions::new().append(true).open(&history).unwrap();
        write!(
            file,
            "{{\"session_id\":\"alpha\",\"text\":\"later topic\"}}\n\
             {{\"session_id\":\"beta\",\"text\":\"second topic\"}}\n"
        )
        .unwrap();
        file.flush().unwrap();
        assert_eq!(index.refresh().unwrap(), BTreeSet::from(["beta".into()]));
        assert_eq!(index.topics["alpha"][0].content, "first topic");
        assert_eq!(index.topics["beta"][0].content, "second topic");

        fs::write(
            &history,
            "{\"session_id\":\"gamma\",\"text\":\"replacement\"}\n",
        )
        .unwrap();
        assert_eq!(
            index.refresh().unwrap(),
            BTreeSet::from(["alpha".into(), "beta".into(), "gamma".into()])
        );
        assert!(!index.topics.contains_key("alpha"));
        assert_eq!(index.topics["gamma"][0].content, "replacement");

        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn codex_history_index_retains_an_incomplete_appended_record() {
        let root = temp_dir("codex-history-partial");
        let sessions = root.join("sessions");
        fs::create_dir_all(&sessions).unwrap();
        let history = root.join("history.jsonl");
        fs::write(&history, "{\"session_id\":\"partial\",\"text\":\"hel").unwrap();

        let mut index = CodexHistoryTopicIndex::new(&sessions);
        assert!(index.refresh().unwrap().is_empty());
        let mut file = fs::OpenOptions::new().append(true).open(&history).unwrap();
        writeln!(file, "lo\"}}").unwrap();
        file.flush().unwrap();

        assert_eq!(index.refresh().unwrap(), BTreeSet::from(["partial".into()]));
        assert_eq!(index.topics["partial"][0].content, "hello");
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn cached_codex_history_enrichment_matches_stateless_discovery() {
        let root = temp_dir("codex-history-parity");
        let sessions = root.join("sessions");
        let workspace = root.join("workspace");
        fs::create_dir_all(&sessions).unwrap();
        fs::create_dir_all(&workspace).unwrap();
        fs::write(
            sessions.join("rollout.jsonl"),
            format!(
                "{{\"type\":\"session_meta\",\"payload\":{{\"id\":\"alpha\",\"cwd\":{}}}}}\n{{\"type\":\"turn_context\",\"payload\":{{\"cwd\":{},\"model\":\"gpt-test\"}}}}\n{{\"type\":\"response_item\",\"payload\":{{\"type\":\"message\",\"role\":\"user\",\"content\":[{{\"type\":\"input_text\",\"text\":\"transcript fallback\"}}]}}}}\n",
                serde_json::to_string(&workspace.to_string_lossy()).unwrap(),
                serde_json::to_string(&workspace.to_string_lossy()).unwrap(),
            ),
        )
        .unwrap();
        fs::write(
            root.join("history.jsonl"),
            "{\"session_id\":\"alpha\",\"text\":\"history topic\"}\n",
        )
        .unwrap();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::from(HarnessId::CODEX)],
            homes: HarnessHomes {
                codex: sessions.clone(),
                ..HarnessHomes::default()
            },
            include_topic_candidates: true,
            ..DiscoveryQuery::default()
        };
        let catalog = HarnessCatalog::new();
        let projected = catalog
            .project_index(&query, catalog.discover_raw_index(&query))
            .unwrap();
        let expected = catalog
            .enrich_index_page(&query, projected.clone())
            .unwrap();
        let mut history = CodexHistoryTopicIndex::new(&sessions);
        history.refresh().unwrap();
        let actual = catalog
            .enrich_index_page_with_codex_history(&query, projected, &history)
            .unwrap();

        assert_eq!(actual, expected);
        assert_eq!(actual[0].preview_candidates[0].content, "history topic");
        fs::remove_dir_all(root).ok();
    }

    #[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"),
                openclaw: root.join("missing-openclaw"),
                hermes: root.join("missing-hermes"),
                orchestrator: root.join("missing-orchestrator"),
            },
            ..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");
            assert!(descriptor.latest_message_candidates[0].cursor.is_some());
            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_event_messages_supply_bounded_native_order_previews() {
        // The list reader must handle the native event-only narration emitted by
        // collab sessions, without loading the transcript or widening its budgets.
        let root = temp_dir("codex-event-previews");
        let path = root.join("rollout.jsonl");
        let user = serde_json::json!({
            "timestamp": "2026-01-01T00:00:01Z", "type": "event_msg",
            "payload": {"type": "user_message", "message": "Investigate the worker"}
        });
        let answer = serde_json::json!({
            "timestamp": "2026-01-01T00:00:02Z", "type": "event_msg",
            "payload": {"type": "agent_message", "message": "Worker findings"}
        });
        let noise = serde_json::json!({
            "type": "event_msg", "payload": {"type": "token_count", "message": "not a message"}
        });
        fs::write(&path, format!("{user}\n{answer}\n{noise}\n{{partial")).unwrap();
        let latest = latest_file_message_candidates(&path, HarnessId::CODEX).unwrap();
        assert_eq!(latest.len(), 2);
        assert_eq!(latest[0].role, "assistant");
        assert_eq!(latest[0].content, "Worker findings");
        assert_eq!(latest[1].role, "user");
        assert_eq!(latest[1].content, "Investigate the worker");
        assert_eq!(
            latest[0].metadata.get("timestamp").map(String::as_str),
            Some("2026-01-01T00:00:02Z")
        );
        assert_eq!(
            latest[0].cursor.as_deref(),
            Some(message_candidate_cursor(HarnessId::CODEX, &answer).as_str())
        );
        let topics = topic_file_message_candidates(&path, HarnessId::CODEX).unwrap();
        assert_eq!(topics.len(), 2);
        assert_eq!(topics[0].content, "Investigate the worker");

        let mut context_pairs = String::new();
        for index in 0..5 {
            let content = if index < 4 {
                format!("# AGENTS.md instructions for /work/{index}\n\n<INSTRUCTIONS>Context</INSTRUCTIONS>")
            } else {
                "The actual user request".to_string()
            };
            let event = serde_json::json!({
                "type": "event_msg", "payload": {"type": "user_message", "message": content}
            });
            let response = serde_json::json!({
                "type": "response_item", "payload": {"type": "message", "role": "user", "content": content}
            });
            context_pairs.push_str(&format!("{event}\n{response}\n"));
        }
        fs::write(&path, context_pairs).unwrap();
        let topics = topic_file_message_candidates(&path, HarnessId::CODEX).unwrap();
        assert!(topics
            .iter()
            .any(|candidate| candidate.content == "The actual user request"));

        // Mirrors must not halve the previously visible response-item window.
        // Either physical order keeps the response's original native cursor.
        let mut paired = String::new();
        for index in 0..12 {
            let content = format!("answer {index}");
            let event = serde_json::json!({
                "type": "event_msg", "payload": {"type": "agent_message", "message": content}
            });
            let response = serde_json::json!({
                "id": format!("response-{index}"), "type": "response_item",
                "payload": {"type": "message", "role": "assistant", "content": content}
            });
            if index % 2 == 0 {
                paired.push_str(&format!("{event}\n{response}\n"));
            } else {
                paired.push_str(&format!("{response}\n{event}\n"));
            }
        }
        fs::write(&path, paired).unwrap();
        let latest = latest_file_message_candidates(&path, HarnessId::CODEX).unwrap();
        assert_eq!(latest.len(), LATEST_PREVIEW_CANDIDATES);
        assert_eq!(latest[0].content, "answer 11");
        assert_eq!(latest[1].content, "answer 10");
        assert_eq!(latest[7].content, "answer 4");
        for (offset, candidate) in latest.iter().enumerate() {
            let response = serde_json::json!({"id": format!("response-{}", 11 - offset)});
            assert_eq!(
                candidate.cursor.as_deref(),
                Some(message_candidate_cursor(HarnessId::CODEX, &response).as_str())
            );
        }
        let topics = topic_file_message_candidates(&path, HarnessId::CODEX).unwrap();
        assert_eq!(topics.len(), LATEST_PREVIEW_CANDIDATES);
        assert_eq!(topics[7].content, "answer 7");

        let event = serde_json::json!({
            "type": "event_msg", "payload": {"type": "agent_message", "message": "again"}
        });
        let response = serde_json::json!({
            "type": "response_item", "payload": {"type": "message", "role": "assistant", "content": "again"}
        });
        for records in [
            format!("{event}\n{event}\n"),
            format!("{response}\n{response}\n"),
            format!("{event}\n{response}\n{event}\n{response}\n"),
            format!("{response}\n{event}\n{response}\n{event}\n"),
        ] {
            fs::write(&path, records).unwrap();
            assert_eq!(
                latest_file_message_candidates(&path, HarnessId::CODEX)
                    .unwrap()
                    .len(),
                2
            );
            assert_eq!(
                topic_file_message_candidates(&path, HarnessId::CODEX)
                    .unwrap()
                    .len(),
                2
            );
        }

        // Equal truncated prefixes alone are not evidence of a mirrored turn.
        let prefix = "x".repeat(4096);
        let distinct_event = serde_json::json!({
            "type": "event_msg", "payload": {"type": "agent_message", "message": format!("{prefix}A")}
        });
        let distinct_response = serde_json::json!({
            "type": "response_item", "payload": {"type": "message", "role": "assistant", "content": format!("{prefix}B")}
        });
        fs::write(&path, format!("{distinct_event}\n{distinct_response}\n")).unwrap();
        assert_eq!(
            latest_file_message_candidates(&path, HarnessId::CODEX)
                .unwrap()
                .len(),
            2
        );
        assert_eq!(
            topic_file_message_candidates(&path, HarnessId::CODEX)
                .unwrap()
                .len(),
            2
        );
        let long = serde_json::json!({
            "type": "event_msg", "payload": {"type": "agent_message", "message": "x".repeat(5000)}
        });
        fs::write(&path, format!("{long}\n")).unwrap();
        let latest = latest_file_message_candidates(&path, HarnessId::CODEX).unwrap();
        assert_eq!(latest[0].content.len(), 4096);
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn preview_cursor_tracks_native_boundary_not_growing_text() {
        let first = serde_json::json!({
            "timestamp": "2026-01-01T00:00:02Z",
            "type": "response_item",
            "payload": {"type": "message", "role": "assistant", "content": "partial"}
        });
        let grown = serde_json::json!({
            "timestamp": "2026-01-01T00:00:02Z",
            "type": "response_item",
            "payload": {"type": "message", "role": "assistant", "content": "partial and complete"}
        });
        let next = serde_json::json!({
            "timestamp": "2026-01-01T00:00:03Z",
            "type": "response_item",
            "payload": {"type": "message", "role": "assistant", "content": "next"}
        });

        assert_eq!(
            message_candidate_cursor(HarnessId::CODEX, &first),
            message_candidate_cursor(HarnessId::CODEX, &grown)
        );
        assert_ne!(
            message_candidate_cursor(HarnessId::CODEX, &first),
            message_candidate_cursor(HarnessId::CODEX, &next)
        );
    }

    #[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_hermes_sqlite_by_session() {
        // A copy of the committed Hermes fixture store, so the test may append to it.
        let root = temp_dir("hermes-follow");
        fs::create_dir_all(&root).unwrap();
        let db = root.join("state.db");
        fs::copy(
            PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                .join("../harness/tests/fixtures/hermes_home/state.db"),
            &db,
        )
        .unwrap();
        let catalog = HarnessCatalog::new();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::from(HarnessId::HERMES)],
            homes: HarnessHomes {
                hermes: db.clone(),
                ..HarnessHomes::default()
            },
            ..DiscoveryQuery::default()
        };
        let found = catalog.discover(&query).unwrap();
        assert!(found.len() >= 2, "{found:#?}");
        // Every discovered locator loads AND follows as ITS OWN session (not the store's newest),
        // and its list preview is that session's own latest message rather than an empty read of
        // the store file's tail.
        for descriptor in &found {
            assert_eq!(descriptor.locator.harness.as_str(), HarnessId::HERMES);
            let loaded = catalog.load(&descriptor.locator).unwrap();
            let last_text = loaded.messages.iter().rev().find_map(|message| {
                (matches!(message.role, crate::Role::User | crate::Role::Assistant))
                    .then(|| message.content.clone())
                    .flatten()
            });
            // Lineage-only fixture rows have no messages and therefore no preview.
            assert_eq!(
                descriptor
                    .latest_message_candidates
                    .first()
                    .map(|c| c.content.as_str()),
                last_text.as_deref(),
                "{}",
                descriptor.locator.session_id
            );
            assert_eq!(
                catalog.load(&descriptor.locator).unwrap().meta.session_id,
                Some(descriptor.locator.session_id.clone())
            );
            let mut follower = catalog.follow(&descriptor.locator).unwrap();
            match follower.poll().unwrap() {
                Some(crate::watch::SessionWatchEvent::SessionSnapshot { session, .. }) => {
                    assert_eq!(
                        session.meta.session_id,
                        Some(descriptor.locator.session_id.clone())
                    );
                }
                other => panic!("expected an initial snapshot, got {other:?}"),
            }
        }
        // Append a message to ONE session: only that session's follower wakes, with exactly the new
        // message, while a sibling's follower stays quiet.
        let target = &found[0].locator;
        let sibling = &found[1].locator;
        let mut target_follower = catalog.follow(target).unwrap();
        let mut sibling_follower = catalog.follow(sibling).unwrap();
        target_follower.poll().unwrap();
        sibling_follower.poll().unwrap();
        std::thread::sleep(std::time::Duration::from_millis(20));
        {
            let conn = rusqlite::Connection::open(&db).unwrap();
            conn.execute(
                "INSERT INTO messages (session_id, role, content, timestamp, active) VALUES (?1, 'assistant', 'appended by the follow test', ?2, 1)",
                rusqlite::params![target.session_id, 1_800_000_000.0_f64],
            )
            .unwrap();
        }
        match target_follower.poll().unwrap() {
            Some(crate::watch::SessionWatchEvent::MessagesAppended {
                session_id,
                messages,
                ..
            }) => {
                assert_eq!(session_id, Some(target.session_id.clone()));
                assert_eq!(messages.len(), 1);
                assert_eq!(
                    messages[0].content.as_deref(),
                    Some("appended by the follow test")
                );
            }
            other => panic!("expected messages_appended for the target session, got {other:?}"),
        }
        assert!(
            sibling_follower.poll().unwrap().is_none(),
            "the sibling session must not wake"
        );
        fs::remove_dir_all(&root).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 preview_search_filters_before_pagination_without_changing_metadata_search() {
        // Search/pagination must compose: filtering only the returned page loses
        // matches and gives a false total. Exercise the public catalog door.
        let root = temp_dir("preview-search");
        for (id, first, last) in [
            ("topic-hit", "NEBULA opening", "Finished"),
            ("latest-hit", "Ordinary opening", "Found the nebula"),
            ("no-hit", "Unrelated opening", "Finished"),
        ] {
            fs::write(root.join(format!("{id}.jsonl")), format!("{}\n{}\n",
                serde_json::json!({"sessionId": id, "cwd": "/work", "type": "user", "message": {"role": "user", "content": first}}),
                serde_json::json!({"sessionId": id, "type": "assistant", "message": {"role": "assistant", "content": last}}),
            )).unwrap();
        }
        let query: DiscoveryQuery = serde_json::from_value(serde_json::json!({
            "harnesses": ["claude-code"], "homes": {"claude_code": root},
            "query": "  nebula  ", "search_previews": true, "limit": 1
        }))
        .unwrap();
        let catalog = HarnessCatalog::new();
        let first = catalog.discover_page(&query).unwrap();
        assert!(first.receipt.searched_previews);
        assert_eq!(first.receipt.total_matched, 2);
        assert_eq!(first.sessions.len(), 1);
        assert!(first.receipt.truncated);
        let second = catalog
            .discover_page(&DiscoveryQuery {
                cursor: first.next_cursor.clone(),
                ..query.clone()
            })
            .unwrap();
        assert_eq!(second.receipt.total_matched, 2);
        assert_eq!(second.sessions.len(), 1);
        assert_ne!(first.sessions[0].locator, second.sessions[0].locator);
        assert!(!second.receipt.truncated);
        let mut metadata = serde_json::to_value(&query).unwrap();
        metadata["search_previews"] = false.into();
        let metadata_page = catalog
            .discover_page(&serde_json::from_value(metadata).unwrap())
            .unwrap();
        assert!(metadata_page.sessions.is_empty());
        assert!(serde_json::to_value(&metadata_page.receipt)
            .unwrap()
            .get("searched_previews")
            .is_none());

        // Metadata remains part of the union; matching multiple candidates must
        // still yield one session, not one row per message.
        let all = catalog
            .discover_page(&DiscoveryQuery {
                query: Some("hit".into()),
                limit: None,
                ..query.clone()
            })
            .unwrap();
        assert_eq!(all.sessions.len(), 3);
        assert_eq!(all.receipt.total_matched, 3);
        let elsewhere = catalog
            .discover_page(&DiscoveryQuery {
                workspace: Some("/elsewhere".into()),
                ..query.clone()
            })
            .unwrap();
        assert_eq!(elsewhere.receipt.total_matched, 0);
        let excluded_by_time = catalog
            .discover_page(&DiscoveryQuery {
                updated_after_ms: Some(u64::MAX),
                ..query.clone()
            })
            .unwrap();
        assert_eq!(excluded_by_time.receipt.total_matched, 0);
        for invalid in [
            DiscoveryQuery {
                query: None,
                ..query.clone()
            },
            DiscoveryQuery {
                query: Some("  ".into()),
                ..query.clone()
            },
            DiscoveryQuery {
                limit: Some(0),
                ..query.clone()
            },
            DiscoveryQuery {
                cursor: Some("bad-cursor".into()),
                ..query.clone()
            },
            DiscoveryQuery {
                cursor: first.next_cursor,
                query: Some("absent".into()),
                ..query.clone()
            },
        ] {
            assert!(catalog.discover_page(&invalid).is_err());
        }
        assert!(catalog.project_index_page(&query, Vec::new()).is_err());
        fs::remove_dir_all(root).ok();
    }

    #[test]
    fn preview_search_uses_codex_first_history_topic_and_bounded_candidates() {
        let root = temp_dir("preview-search-codex");
        let sessions = root.join("sessions");
        fs::create_dir_all(&sessions).unwrap();
        fs::write(
            root.join("history.jsonl"),
            format!(
                "{}\n{}\n",
                serde_json::json!({"session_id": "history-hit", "text": "Original nebula topic"}),
                serde_json::json!({"session_id": "history-hit", "text": "laterhistoryonly"}),
            ),
        )
        .unwrap();
        for id in ["history-hit", "latest-hit", "bounded"] {
            let mut content = format!(
                "{}\n",
                serde_json::json!({
                    "type": "session_meta", "payload": {"id": id, "cwd": "/work"}
                })
            );
            for index in 0..20 {
                let message = if id == "latest-hit" && index == 19 {
                    "Found NEBULA".to_string()
                } else if index == 10 {
                    "middlehistoryonly".to_string()
                } else {
                    format!("{}beyondtextcap", "x".repeat(4096))
                };
                content.push_str(&format!("{}\n", serde_json::json!({
                    "type": "event_msg", "payload": {"type": "agent_message", "message": message}
                })));
            }
            fs::write(sessions.join(format!("{id}.jsonl")), content).unwrap();
        }
        let catalog = HarnessCatalog::new();
        let query: DiscoveryQuery = serde_json::from_value(serde_json::json!({
            "harnesses": ["codex"], "homes": {"codex": sessions},
            "query": "nebula", "search_previews": true
        }))
        .unwrap();
        let page = catalog.discover_page(&query).unwrap();
        assert_eq!(page.receipt.total_matched, 2);
        for row in &page.sessions {
            assert!(row.preview_candidates.len() <= 8);
            assert!(row.latest_message_candidates.len() <= 8);
            assert!(row
                .preview_candidates
                .iter()
                .chain(&row.latest_message_candidates)
                .all(|candidate| candidate.content.chars().count() <= 4096));
        }
        for text in ["middlehistoryonly", "laterhistoryonly", "beyondtextcap"] {
            assert!(
                catalog
                    .discover_page(&DiscoveryQuery {
                        query: Some(text.into()),
                        ..query.clone()
                    })
                    .unwrap()
                    .sessions
                    .is_empty(),
                "not a full-history search: {text}"
            );
        }
        fs::remove_dir_all(root).unwrap();
    }

    #[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();
    }

    #[test]
    fn subtree_scope_admits_the_root_and_every_folder_under_it_only() {
        let base = temp_dir("subtree-scope");
        let work = base.join("work");
        let nested = work.join("repo").join("pkg");
        let sibling = base.join("workshop");
        fs::create_dir_all(&nested).unwrap();
        fs::create_dir_all(&sibling).unwrap();
        let scope = WorkspaceScope::subtree(&work);
        assert!(scope.admits(&work));
        assert!(scope.admits(&work.join("repo")));
        assert!(scope.admits(&nested));
        // A folder that no longer exists still names where its session ran.
        assert!(scope.admits(&work.join("deleted-since")));
        // `..` is resolved before containment is judged.
        assert!(!scope.admits(&work.join("..").join("workshop")));
        assert!(!scope.admits(&sibling), "/work must not admit /workshop");
        assert!(!scope.admits(&base));
        assert!(!scope.admits(Path::new(".")));
        assert!(!scope.admits(Path::new("repo/pkg")));
        let exact = WorkspaceScope::exact(&work);
        assert!(exact.admits(&work));
        assert!(!exact.admits(&nested), "the default stays the exact folder");
        fs::remove_dir_all(base).ok();
    }

    #[cfg(windows)]
    #[test]
    fn subtree_scope_on_windows_ignores_verbatim_prefix_and_case_and_rejects_driveless_cwds() {
        let base = temp_dir("subtree-windows");
        let work = base.join("Work");
        fs::create_dir_all(work.join("repo")).unwrap();
        let scope = WorkspaceScope::subtree(&work);
        let verbatim = fs::canonicalize(work.join("repo")).unwrap();
        assert!(verbatim.to_string_lossy().starts_with(r"\\?\"));
        assert!(scope.admits(&verbatim));
        let folded = PathBuf::from(work.join("repo").to_string_lossy().to_uppercase());
        assert!(scope.admits(&folded));
        // `\Users\...` has no drive: Rust does not call it absolute, and it
        // says nothing about which volume the session ran on.
        let text = work.to_string_lossy().into_owned();
        let driveless = PathBuf::from(&text[2..]);
        assert!(driveless.has_root() && !driveless.is_absolute());
        assert!(!scope.admits(&driveless));
        assert!(!scope.admits(&driveless.join("repo")));
        fs::remove_dir_all(base).ok();
    }

    #[test]
    fn subtree_scope_matches_through_a_symlinked_root() {
        let base = temp_dir("subtree-symlink");
        let real = base.join("real");
        fs::create_dir_all(real.join("repo")).unwrap();
        let alias = base.join("alias");
        #[cfg(unix)]
        let linked = std::os::unix::fs::symlink(&real, &alias).is_ok();
        #[cfg(windows)]
        let linked = std::os::windows::fs::symlink_dir(&real, &alias).is_ok()
            // Windows without Developer Mode refuses symlinks; a junction is
            // the unprivileged directory alias there.
            || std::process::Command::new("cmd")
                .arg("/C")
                .arg("mklink")
                .arg("/J")
                .arg(&alias)
                .arg(&real)
                .output()
                .is_ok_and(|output| output.status.success());
        if !linked {
            fs::remove_dir_all(base).ok();
            return;
        }
        // Root given through the alias, cwd recorded as the real path, and
        // the other way round (macOS records /private/var for /var).
        assert!(WorkspaceScope::subtree(&alias).admits(&real.join("repo")));
        assert!(WorkspaceScope::subtree(&real).admits(&alias.join("repo")));
        assert!(!WorkspaceScope::subtree(&alias).admits(&base));
        fs::remove_dir_all(base).ok();
    }

    #[test]
    fn workspace_subtree_discovery_filters_before_enrichment_and_defaults_to_exact() {
        let base = temp_dir("subtree-discovery");
        let home = base.join("projects");
        let work = base.join("work");
        let nested = work.join("repo");
        let sibling = base.join("workshop");
        for dir in [&home, &nested, &sibling] {
            fs::create_dir_all(dir).unwrap();
        }
        let write = |id: &str, cwd: &Path| {
            let project = home.join(id);
            fs::create_dir_all(&project).unwrap();
            let line = serde_json::json!({
                "type": "user",
                "sessionId": id,
                "cwd": cwd,
                "message": {"role": "user", "content": format!("hello from {id}")}
            });
            fs::write(project.join(format!("{id}.jsonl")), format!("{line}\n")).unwrap();
        };
        write("in-root", &work);
        write("in-repo", &nested);
        write("sibling", &sibling);
        write("relative", Path::new("repo"));
        let query = |subtree: bool| DiscoveryQuery {
            workspace: Some(work.clone()),
            workspace_subtree: subtree,
            harnesses: vec![HarnessId::from(HarnessId::CLAUDE_CODE)],
            homes: HarnessHomes {
                claude_code: home.clone(),
                ..HarnessHomes::default()
            },
            ..DiscoveryQuery::default()
        };
        let ids = |subtree: bool| {
            let mut ids: Vec<String> = HarnessCatalog::new()
                .discover(&query(subtree))
                .unwrap()
                .into_iter()
                .map(|descriptor| descriptor.locator.session_id)
                .collect();
            ids.sort();
            ids
        };
        assert_eq!(ids(false), vec!["in-root"]);
        assert_eq!(ids(true), vec!["in-repo", "in-root"]);

        // Wire: an old client never sends the field and gets the exact match;
        // the default is not serialized, so old servers see the same query.
        let old: DiscoveryQuery =
            serde_json::from_value(serde_json::json!({"workspace": work})).unwrap();
        assert!(!old.workspace_subtree);
        assert!(serde_json::to_value(&old)
            .unwrap()
            .get("workspace_subtree")
            .is_none());
        let new: DiscoveryQuery = serde_json::from_value(
            serde_json::json!({"workspace": work, "workspace_subtree": true}),
        )
        .unwrap();
        assert!(new.workspace_subtree);
        fs::remove_dir_all(base).ok();
    }
}