supercode_harness/

session_index.rs

//! Revisioned session-list subscriptions for latency-sensitive frontends.
//!
//! Native filesystem events are treated as invalidation hints, never as the
//! session record itself. Each hint causes a bounded re-read of the affected
//! Claude Code or Codex transcript; a slow periodic catalog reconciliation
//! repairs dropped/coalesced platform events and fills a page after removals.

use std::collections::{BTreeMap, BTreeSet};
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{mpsc, Arc};
use std::time::{Duration, Instant, UNIX_EPOCH};

use notify::event::{AccessKind, AccessMode};
use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use serde::Serialize;
use tokio::sync::Notify;

use crate::{
    catalog::{hermes_session_stores, CodexHistoryTopicIndex},
    DiscoveryPage, DiscoveryQuery, HarnessCatalog, HarnessId, SessionDescriptor, SessionLocator,
    StorageLocator,
};

const RECONCILE_INTERVAL: Duration = Duration::from_secs(60);
const MAX_SUBSCRIPTION_ROWS: usize = 2_048;
const INVALIDATION_QUEUE_CAPACITY: usize = 1_024;

/// Stable public identity for a session-index change. Persistence paths remain
/// inside the trusted host and are sent only as part of complete descriptors.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize)]
pub struct SessionIndexKey {
    /// Owning harness id.
    pub harness: String,
    /// Harness-native durable session id.
    pub session_id: String,
}

impl SessionIndexKey {
    fn from_locator(locator: &SessionLocator) -> Self {
        Self {
            harness: locator.harness.as_str().to_string(),
            session_id: locator.session_id.clone(),
        }
    }
}

/// One complete replacement in a revisioned index delta.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum SessionIndexChange {
    /// A session entered the bounded result page.
    Added {
        /// Complete current descriptor.
        descriptor: SessionDescriptor,
    },
    /// A visible session's descriptor changed.
    Updated {
        /// Complete replacement descriptor.
        descriptor: SessionDescriptor,
    },
    /// A session disappeared from the bounded result page.
    Removed {
        /// Stable identity of the removed descriptor.
        key: SessionIndexKey,
    },
}

/// One subscription poll result. Revisions start at one for the initial
/// snapshot and increase by exactly one for each non-empty delta batch or
/// committed window resize. They describe the visible window, not changes to
/// out-of-window inventory totals returned by a same-limit read.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct SessionIndexDelta {
    /// Monotonic subscription-local revision.
    pub revision: u64,
    /// Complete replacement changes in deterministic identity order.
    pub changes: Vec<SessionIndexChange>,
}

/// Filesystem-backed index subscription. Dropping it drops the platform
/// watcher and callback channel, so unsubscribe has deterministic cleanup.
pub(crate) struct SessionIndexSubscription {
    query: DiscoveryQuery,
    raw: BTreeMap<SessionIndexKey, SessionDescriptor>,
    paths: BTreeMap<PathBuf, SessionIndexKey>,
    current: BTreeMap<SessionIndexKey, SessionDescriptor>,
    fingerprints: BTreeMap<PathBuf, FileFingerprint>,
    /// Whole-store SQLite files (Hermes `state.db` + its `-wal`/`-shm`), keyed by path. A store
    /// holds every session in one file, so a stamp change means "re-enumerate this store", not
    /// "this one path is one session". `None` = the file is absent.
    store_fingerprints: BTreeMap<PathBuf, Option<FileFingerprint>>,
    codex_history: Option<CodexHistoryTopicIndex>,
    revision: u64,
    receiver: mpsc::Receiver<notify::Result<Event>>,
    overflowed: Arc<AtomicBool>,
    _watcher: RecommendedWatcher,
    last_reconcile: Instant,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct FileFingerprint {
    len: u64,
    modified_ns: u128,
    modified_ms: Option<u64>,
    identity: u128,
}

/// A complete replacement prepared without changing the subscription. The
/// transport can serialize it before committing, so a failed response leaves
/// the old window and revision usable.
pub(crate) struct PreparedIndexResize {
    limit: usize,
    pub(crate) revision: u64,
    pub(crate) page: DiscoveryPage,
}

impl SessionIndexSubscription {
    pub(crate) fn homes(&self) -> &crate::HarnessHomes {
        &self.query.homes
    }

    pub(crate) fn prepare_resize(&self, limit: usize) -> Result<PreparedIndexResize, String> {
        let mut query = self.query.clone();
        query.limit = Some(limit);
        validate_query(&query)?;
        // This is a snapshot of the known index, not a filesystem barrier.
        // Pending invalidations stay queued and produce the next normal delta.
        let page = self.project_current(&query, &BTreeSet::new())?;
        Ok(PreparedIndexResize {
            limit,
            revision: if self.query.limit == Some(limit) {
                self.revision
            } else {
                self.revision.saturating_add(1)
            },
            page,
        })
    }

    pub(crate) fn commit_resize(&mut self, prepared: PreparedIndexResize) {
        self.query.limit = Some(prepared.limit);
        self.revision = prepared.revision;
        self.current = descriptor_map(prepared.page.sessions);
    }

    pub(crate) fn open(
        mut query: DiscoveryQuery,
        notifier: Arc<Notify>,
    ) -> Result<(Self, Vec<SessionDescriptor>), String> {
        validate_query(&query)?;
        query.cursor = None;
        query.limit = Some(query.limit.unwrap_or(100));

        let catalog = HarnessCatalog::new();
        let raw = descriptor_map(catalog.discover_raw_index(&query));
        let projected = catalog
            .project_index(&query, raw.values().cloned())
            .map_err(|error| error.to_string())?;
        let mut codex_history = (query.include_topic_candidates
            && query
                .harnesses
                .iter()
                .any(|harness| harness.as_str() == HarnessId::CODEX))
        .then(|| CodexHistoryTopicIndex::new(&query.homes.codex));
        if let Some(history) = &mut codex_history {
            // Discovery has always treated an unavailable history file as a
            // soft fallback to transcript topics. Preserve that behavior.
            let _ = history.refresh();
        }
        let initial = match &codex_history {
            Some(history) => {
                catalog.enrich_index_page_with_codex_history(&query, projected, history)
            }
            None => catalog.enrich_index_page(&query, projected),
        }
        .map_err(|error| error.to_string())?;
        let paths = descriptor_path_map(&raw);
        let current = descriptor_map(initial.iter().cloned());
        let fingerprints = scan_file_fingerprints(&query);
        let store_fingerprints = scan_store_fingerprints(&query);
        let (sender, receiver) = mpsc::sync_channel(INVALIDATION_QUEUE_CAPACITY);
        let overflowed = Arc::new(AtomicBool::new(false));
        let callback_overflowed = Arc::clone(&overflowed);
        let callback_notifier = Arc::clone(&notifier);
        let mut watcher = notify::recommended_watcher(move |event| {
            if sender.try_send(event).is_err() {
                callback_overflowed.store(true, Ordering::Release);
            }
            callback_notifier.notify_one();
        })
        .map_err(|error| error.to_string())?;
        for root in watch_roots(&query) {
            if let Some(watched) = existing_watch_root(&root) {
                watcher
                    .watch(&watched, RecursiveMode::Recursive)
                    .map_err(|error| format!("cannot watch {}: {error}", watched.display()))?;
            }
        }
        for store in store_paths(&query) {
            // The store's directory, not the store file: a WAL-mode writer creates and removes the
            // `-wal`/`-shm` siblings, and a first run creates the store itself.
            let Some(dir) = store.parent() else { continue };
            if let Some(watched) = existing_watch_root(dir) {
                watcher
                    .watch(&watched, RecursiveMode::NonRecursive)
                    .map_err(|error| format!("cannot watch {}: {error}", watched.display()))?;
            }
        }
        if let Some(history) = &codex_history {
            let target = if history.path().is_file() {
                history.path()
            } else {
                history.path().parent().unwrap_or(history.path())
            };
            if target.exists() {
                watcher
                    .watch(target, RecursiveMode::NonRecursive)
                    .map_err(|error| format!("cannot watch {}: {error}", target.display()))?;
            }
        }

        Ok((
            Self {
                query,
                raw,
                paths,
                current,
                fingerprints,
                store_fingerprints,
                codex_history,
                revision: 1,
                receiver,
                overflowed,
                _watcher: watcher,
                last_reconcile: Instant::now(),
            },
            initial,
        ))
    }

    /// Drain and coalesce native invalidations once. No events means no I/O
    /// until the minute-scale metadata-only recovery sweep becomes due.
    pub(crate) fn poll(&mut self) -> Result<Option<SessionIndexDelta>, String> {
        let mut paths = BTreeSet::new();
        let mut sweep = self.overflowed.swap(false, Ordering::AcqRel);
        let mut stores = false;
        while let Ok(event) = self.receiver.try_recv() {
            match event {
                // an open, a read or a close without writing changes nothing: on Linux, notify reports every
                // open (IN_OPEN) in a watched directory, so a Hermes gateway opening its store per request drove
                // a full re-index per open (a core at 100% in Open Autonomy's reporter); a close after writing stays
                Ok(event) if matches!(event.kind, EventKind::Access(access) if access != AccessKind::Close(AccessMode::Write)) =>
                    {}
                Ok(event) => {
                    if event.paths.is_empty() {
                        sweep = true;
                    }
                    for path in event.paths {
                        if is_store_shm(&self.store_fingerprints, &path) {
                            continue;
                        }
                        if path.extension().and_then(|value| value.to_str()) == Some("jsonl") {
                            paths.insert(path);
                        } else if self
                            .store_fingerprints
                            .contains_key(&normalized_store_path(&path))
                        {
                            stores = true;
                        } else {
                            sweep = true;
                        }
                    }
                }
                Err(_) => sweep = true,
            }
        }
        if self.last_reconcile.elapsed() >= RECONCILE_INTERVAL {
            sweep = true;
        }
        if paths.is_empty() && !sweep && !stores {
            return Ok(None);
        }

        let before = self.current.clone();
        let mut content_dirty = BTreeSet::new();
        let history_path = self
            .codex_history
            .as_ref()
            .map(|history| normalized_path(history.path()));
        if let Some(history) = &mut self.codex_history {
            if let Ok(changed) = history.refresh() {
                content_dirty.extend(changed.into_iter().map(|session_id| SessionIndexKey {
                    harness: HarnessId::CODEX.to_string(),
                    session_id,
                }));
            }
        }
        if sweep {
            self.reconcile_filesystem(&mut content_dirty)?;
        }
        if sweep || stores {
            self.reconcile_stores(&mut content_dirty)?;
        }
        for path in paths {
            if history_path
                .as_ref()
                .is_some_and(|history_path| normalized_path(&path) == *history_path)
            {
                continue;
            }
            self.refresh_path(&path, &mut content_dirty)?;
        }
        self.rebuild_current(&content_dirty)?;
        let changes = diff_descriptors(&before, &self.current);
        if changes.is_empty() {
            return Ok(None);
        }
        self.revision = self.revision.saturating_add(1);
        Ok(Some(SessionIndexDelta {
            revision: self.revision,
            changes,
        }))
    }

    fn reconcile_filesystem(
        &mut self,
        content_dirty: &mut BTreeSet<SessionIndexKey>,
    ) -> Result<(), String> {
        self.last_reconcile = Instant::now();
        let next = scan_file_fingerprints(&self.query);
        let changed = self
            .fingerprints
            .keys()
            .chain(next.keys())
            .filter(|path| self.fingerprints.get(*path) != next.get(*path))
            .cloned()
            .collect::<BTreeSet<_>>();
        for path in changed {
            self.refresh_path(&path, content_dirty)?;
        }
        self.fingerprints = next;
        Ok(())
    }

    /// Re-enumerate every whole-store harness whose store stamps moved. Rows are diffed by value:
    /// a store keeps its sessions' `message_count`/`ended_at` current on every append, so a
    /// descriptor that compares equal is unchanged and one that differs is content-dirty.
    fn reconcile_stores(
        &mut self,
        content_dirty: &mut BTreeSet<SessionIndexKey>,
    ) -> Result<(), String> {
        let next = scan_store_fingerprints(&self.query);
        if next == self.store_fingerprints {
            return Ok(());
        }
        self.store_fingerprints = next;
        let mut query = self.query.clone();
        query
            .harnesses
            .retain(|harness| harness.as_str() == HarnessId::HERMES);
        if query.harnesses.is_empty() {
            return Ok(());
        }
        let fresh = descriptor_map(HarnessCatalog::new().discover_raw_index(&query));
        let stale = self
            .raw
            .keys()
            .filter(|key| key.harness == HarnessId::HERMES)
            .cloned()
            .collect::<Vec<_>>();
        for key in stale {
            if !fresh.contains_key(&key) {
                self.raw.remove(&key);
                content_dirty.insert(key);
            }
        }
        for (key, descriptor) in fresh {
            if self.raw.get(&key) != Some(&descriptor) {
                self.raw.insert(key.clone(), descriptor);
                content_dirty.insert(key);
            }
        }
        Ok(())
    }

    fn refresh_path(
        &mut self,
        path: &Path,
        content_dirty: &mut BTreeSet<SessionIndexKey>,
    ) -> Result<(), String> {
        if path.extension().and_then(|value| value.to_str()) != Some("jsonl") {
            return Ok(());
        }
        let event_path = normalized_path(path);
        let previous_key = self.paths.get(&event_path).cloned();
        let previous = previous_key
            .as_ref()
            .and_then(|key| self.raw.get(key))
            .cloned();
        let previous_fingerprint = self.fingerprints.get(&event_path).copied();
        let fingerprint = file_fingerprint(&event_path);

        let Some(fingerprint) = fingerprint else {
            self.fingerprints.remove(&event_path);
            if let Some(key) = previous_key {
                self.paths.remove(&event_path);
                self.raw.remove(&key);
                content_dirty.insert(key);
            }
            return Ok(());
        };
        self.fingerprints.insert(event_path.clone(), fingerprint);

        let locator = previous
            .as_ref()
            .map(|descriptor| descriptor.locator.clone())
            .or_else(|| locator_for_path(&self.query, &event_path));
        let Some(locator) = locator else {
            return Ok(());
        };
        let refreshed =
            if let (Some(descriptor), Some(old)) = (previous.as_ref(), previous_fingerprint) {
                if can_reuse_header(descriptor, old, fingerprint) {
                    let mut descriptor = descriptor.clone();
                    descriptor.updated_at_ms = fingerprint.modified_ms;
                    Some(descriptor)
                } else {
                    HarnessCatalog::new()
                        .refresh_file_index_descriptor(&locator, self.query.workspace.as_deref())
                        .map_err(|error| error.to_string())?
                }
            } else {
                HarnessCatalog::new()
                    .refresh_file_index_descriptor(&locator, self.query.workspace.as_deref())
                    .map_err(|error| error.to_string())?
            };
        let Some(descriptor) = refreshed else {
            return Ok(());
        };
        let key = SessionIndexKey::from_locator(&descriptor.locator);
        if let Some(previous_key) = previous_key {
            if previous_key != key {
                self.raw.remove(&previous_key);
                content_dirty.insert(previous_key);
            }
        }
        self.paths.insert(event_path, key.clone());
        self.raw.insert(key.clone(), descriptor);
        content_dirty.insert(key);
        Ok(())
    }

    fn rebuild_current(&mut self, content_dirty: &BTreeSet<SessionIndexKey>) -> Result<(), String> {
        let page = self.project_current(&self.query, content_dirty)?;
        self.current = descriptor_map(page.sessions);
        Ok(())
    }

    fn project_current(
        &self,
        query: &DiscoveryQuery,
        content_dirty: &BTreeSet<SessionIndexKey>,
    ) -> Result<DiscoveryPage, String> {
        let catalog = HarnessCatalog::new();
        let mut page = catalog
            .project_index_page(query, self.raw.values().cloned())
            .map_err(|error| error.to_string())?;
        let mut next = Vec::with_capacity(page.sessions.len());
        for mut descriptor in page.sessions {
            let key = SessionIndexKey::from_locator(&descriptor.locator);
            if let Some(previous) = self.current.get(&key) {
                descriptor.preview_candidates = previous.preview_candidates.clone();
                descriptor.latest_message_candidates = previous.latest_message_candidates.clone();
            }
            if !self.current.contains_key(&key) || content_dirty.contains(&key) {
                let enriched = match &self.codex_history {
                    Some(history) => catalog.enrich_index_page_with_codex_history(
                        query,
                        vec![descriptor],
                        history,
                    ),
                    None => catalog.enrich_index_page(query, vec![descriptor]),
                };
                descriptor = enriched
                    .map_err(|error| error.to_string())?
                    .pop()
                    .expect("one descriptor remains one descriptor");
            }
            next.push(descriptor);
        }
        page.sessions = next;
        Ok(page)
    }
}

pub(crate) fn validate_query(query: &DiscoveryQuery) -> Result<(), String> {
    if query.search_previews {
        return Err(
            "sessions.index.subscribe does not support preview search; use sessions.discover"
                .into(),
        );
    }
    if query.cursor.is_some() {
        return Err("sessions.index.subscribe does not accept a cursor".into());
    }
    validate_limit(query.limit.unwrap_or(100))?;
    if query.harnesses.is_empty()
        || query.harnesses.iter().any(|harness| {
            !matches!(
                harness.as_str(),
                HarnessId::CLAUDE_CODE | HarnessId::CODEX | HarnessId::HERMES
            )
        })
    {
        return Err(
            "sessions.index.subscribe currently requires explicit claude-code, codex and/or hermes harnesses"
                .into(),
        );
    }
    Ok(())
}

pub(crate) fn validate_limit(limit: usize) -> Result<(), String> {
    if limit == 0 || limit > MAX_SUBSCRIPTION_ROWS {
        return Err(format!(
            "session index limit must be between 1 and {MAX_SUBSCRIPTION_ROWS}"
        ));
    }
    Ok(())
}

fn watch_roots(query: &DiscoveryQuery) -> BTreeSet<PathBuf> {
    query
        .harnesses
        .iter()
        .filter_map(|harness| match harness.as_str() {
            HarnessId::CLAUDE_CODE => Some(query.homes.claude_code.clone()),
            HarnessId::CODEX => Some(query.homes.codex.clone()),
            _ => None,
        })
        .collect()
}

fn existing_watch_root(root: &Path) -> Option<PathBuf> {
    if root.is_dir() {
        return Some(root.to_path_buf());
    }
    // Watching an entire home directory because a harness has never created
    // its store is disproportionate. One parent level catches the ordinary
    // first-run mkdir; the recovery reconciliation handles rarer deeper gaps.
    root.parent()
        .filter(|parent| parent.is_dir())
        .map(Path::to_path_buf)
}

/// Whole-store SQLite files named by the query (one file = every session of that harness).
fn store_paths(query: &DiscoveryQuery) -> BTreeSet<PathBuf> {
    query
        .harnesses
        .iter()
        .flat_map(|harness| match harness.as_str() {
            HarnessId::HERMES => hermes_session_stores(&query.homes.hermes),
            _ => Vec::new(),
        })
        .collect()
}

/// A store's stamp set: the file itself and its WAL-mode siblings, which is where a live writer's
/// appends land until a checkpoint. Absent files are kept as `None` so their creation is a change.
fn scan_store_fingerprints(query: &DiscoveryQuery) -> BTreeMap<PathBuf, Option<FileFingerprint>> {
    let mut stamps = BTreeMap::new();
    for store in store_paths(query) {
        for path in store_sibling_paths(&store) {
            let stamp = file_fingerprint(&path);
            stamps.insert(normalized_store_path(&path), stamp);
        }
    }
    stamps
}

/// The files that hold a store's committed content: the database and its write-ahead log. The
/// `-shm` sibling is SQLite's shared-memory index, which every reader writes (this index's own
/// read-only enumeration included): stamping it made each enumeration trigger the next one.
fn store_sibling_paths(store: &Path) -> [PathBuf; 2] {
    let name = store
        .file_name()
        .and_then(|value| value.to_str())
        .unwrap_or("state.db");
    [
        store.to_path_buf(),
        store.with_file_name(format!("{name}-wal")),
    ]
}

/// A watched store's `-shm` sibling: its changes are readers' bookkeeping, not content.
fn is_store_shm(stores: &BTreeMap<PathBuf, Option<FileFingerprint>>, path: &Path) -> bool {
    let path = normalized_store_path(path);
    path.to_str()
        .and_then(|value| value.strip_suffix("-shm"))
        .is_some_and(|store| stores.contains_key(Path::new(store)))
}

/// Store siblings come and go, so canonicalize through the (stable) directory rather than the file.
fn normalized_store_path(path: &Path) -> PathBuf {
    match (path.parent(), path.file_name()) {
        (Some(dir), Some(name)) => normalized_path(dir).join(name),
        _ => path.to_path_buf(),
    }
}

fn locator_for_path(query: &DiscoveryQuery, path: &Path) -> Option<SessionLocator> {
    let claude_root = normalized_path(&query.homes.claude_code);
    let codex_root = normalized_path(&query.homes.codex);
    let harness = if query
        .harnesses
        .iter()
        .any(|harness| harness.as_str() == HarnessId::CLAUDE_CODE)
        && path.starts_with(&claude_root)
    {
        HarnessId::CLAUDE_CODE
    } else if query
        .harnesses
        .iter()
        .any(|harness| harness.as_str() == HarnessId::CODEX)
        && path.starts_with(&codex_root)
    {
        HarnessId::CODEX
    } else {
        return None;
    };
    Some(SessionLocator {
        harness: HarnessId::new(harness),
        session_id: path
            .file_stem()
            .and_then(|value| value.to_str())
            .unwrap_or("unknown")
            .to_string(),
        storage: StorageLocator::File {
            path: path.to_path_buf(),
        },
    })
}

fn normalized_path(path: &Path) -> PathBuf {
    fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf())
}

fn descriptor_path_map(
    descriptors: &BTreeMap<SessionIndexKey, SessionDescriptor>,
) -> BTreeMap<PathBuf, SessionIndexKey> {
    descriptors
        .iter()
        .map(|(key, descriptor)| {
            (
                normalized_path(descriptor.locator.storage.path()),
                key.clone(),
            )
        })
        .collect()
}

fn scan_file_fingerprints(query: &DiscoveryQuery) -> BTreeMap<PathBuf, FileFingerprint> {
    let mut paths = Vec::new();
    for root in watch_roots(query) {
        collect_jsonl_paths(&root, &mut paths);
    }
    paths
        .into_iter()
        .filter_map(|path| {
            let path = normalized_path(&path);
            file_fingerprint(&path).map(|fingerprint| (path, fingerprint))
        })
        .collect()
}

fn collect_jsonl_paths(root: &Path, paths: &mut Vec<PathBuf>) {
    let Ok(entries) = fs::read_dir(root) else {
        return;
    };
    for entry in entries.flatten() {
        let Ok(file_type) = entry.file_type() else {
            continue;
        };
        let path = entry.path();
        if file_type.is_dir() {
            collect_jsonl_paths(&path, paths);
        } else if file_type.is_file()
            && path.extension().and_then(|value| value.to_str()) == Some("jsonl")
        {
            paths.push(path);
        }
    }
}

fn file_fingerprint(path: &Path) -> Option<FileFingerprint> {
    let metadata = fs::metadata(path).ok()?;
    let modified = metadata.modified().ok()?.duration_since(UNIX_EPOCH).ok()?;
    #[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;
    Some(FileFingerprint {
        len: metadata.len(),
        modified_ns: modified.as_nanos(),
        modified_ms: u64::try_from(modified.as_millis()).ok(),
        identity,
    })
}

fn can_reuse_header(
    descriptor: &SessionDescriptor,
    previous: FileFingerprint,
    current: FileFingerprint,
) -> bool {
    previous.identity == current.identity
        && previous.len <= current.len
        && descriptor.cwd.is_some()
        && descriptor.model.is_some()
        && !descriptor.locator.session_id.is_empty()
}

fn descriptor_map(
    descriptors: impl IntoIterator<Item = SessionDescriptor>,
) -> BTreeMap<SessionIndexKey, SessionDescriptor> {
    descriptors
        .into_iter()
        .map(|descriptor| {
            (
                SessionIndexKey::from_locator(&descriptor.locator),
                descriptor,
            )
        })
        .collect()
}

fn diff_descriptors(
    before: &BTreeMap<SessionIndexKey, SessionDescriptor>,
    after: &BTreeMap<SessionIndexKey, SessionDescriptor>,
) -> Vec<SessionIndexChange> {
    let mut changes = Vec::new();
    for (key, descriptor) in after {
        match before.get(key) {
            None => changes.push(SessionIndexChange::Added {
                descriptor: descriptor.clone(),
            }),
            Some(previous) if previous != descriptor => {
                changes.push(SessionIndexChange::Updated {
                    descriptor: descriptor.clone(),
                });
            }
            Some(_) => {}
        }
    }
    for key in before.keys() {
        if !after.contains_key(key) {
            changes.push(SessionIndexChange::Removed { key: key.clone() });
        }
    }
    changes
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn preview_search_is_refused_before_opening_a_retained_index() {
        let query = DiscoveryQuery {
            search_previews: true,
            query: Some("nebula".into()),
            ..DiscoveryQuery::default()
        };
        let error = match SessionIndexSubscription::open(query, Arc::new(Notify::new())) {
            Err(error) => error,
            Ok(_) => panic!("preview search must not open live watchers"),
        };
        assert!(error.contains("use sessions.discover"), "{error}");
    }

    fn descriptor(id: &str, updated_at_ms: u64) -> SessionDescriptor {
        SessionDescriptor {
            locator: SessionLocator {
                harness: HarnessId::new(HarnessId::CODEX),
                session_id: id.into(),
                storage: StorageLocator::File {
                    path: PathBuf::from(format!("/{id}.jsonl")),
                },
            },
            cwd: None,
            title: None,
            preview_candidates: Vec::new(),
            latest_message_candidates: Vec::new(),
            updated_at_ms: Some(updated_at_ms),
            message_count: None,
            model: None,
            parent_session_id: None,
            child_session_count: 0,
            nouns: Default::default(),
        }
    }

    #[test]
    fn resize_retains_index_watcher_and_cached_previews_until_commit() {
        let root = std::env::temp_dir().join(format!(
            "supercode-index-resize-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        fs::create_dir_all(&root).unwrap();
        let root = root.canonicalize().unwrap();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::new(HarnessId::CODEX)],
            homes: crate::HarnessHomes {
                codex: root.clone(),
                ..crate::HarnessHomes::default()
            },
            limit: Some(1),
            include_topic_candidates: true,
            ..DiscoveryQuery::default()
        };
        let (mut index, _) =
            SessionIndexSubscription::open(query, Arc::new(Notify::new())).unwrap();
        for (id, updated) in [("newest", 3), ("middle", 2), ("oldest", 1)] {
            let path = root.join(format!("{id}.jsonl"));
            fs::write(&path, format!(
                "{{\"type\":\"session_meta\",\"payload\":{{\"id\":\"{id}\",\"cwd\":\"/workspace\"}}}}\n{{\"type\":\"response_item\",\"payload\":{{\"type\":\"message\",\"role\":\"user\",\"content\":[{{\"type\":\"input_text\",\"text\":\"topic {id}\"}}]}}}}\n"
            )).unwrap();
            let mut row = descriptor(id, updated);
            row.locator.storage = StorageLocator::File { path };
            index
                .raw
                .insert(SessionIndexKey::from_locator(&row.locator), row);
        }
        index.paths = descriptor_path_map(&index.raw);
        index.rebuild_current(&BTreeSet::new()).unwrap();
        let original = index.current.clone();
        assert!(!original
            .values()
            .next()
            .unwrap()
            .preview_candidates
            .is_empty());
        // If resize re-read this already-visible transcript, its topic would disappear.
        fs::write(root.join("newest.jsonl"), "").unwrap();
        // A new file must not be enumerated by resize; its queued event belongs to poll.
        let queued = root.join("queued.jsonl");
        fs::write(
            &queued,
            "{\"type\":\"session_meta\",\"payload\":{\"id\":\"queued\",\"cwd\":\"/workspace\"}}\n",
        )
        .unwrap();
        let (sender, receiver) = mpsc::channel();
        // Keep the native channel alive while driving deterministic invalidations.
        let _native_receiver = std::mem::replace(&mut index.receiver, receiver);
        sender
            .send(Ok(Event::new(notify::EventKind::Any).add_path(queued)))
            .unwrap();
        let watcher = &index._watcher as *const _;
        let raw_row = index.raw.values().next().unwrap() as *const _;
        let raw = index.raw.clone();
        let reconcile = index.last_reconcile;
        let prepared = index.prepare_resize(2).unwrap();
        assert_eq!(prepared.revision, 2);
        assert_eq!(prepared.page.receipt.total_matched, 3);
        assert_eq!(prepared.page.sessions.len(), 2);
        assert_eq!(
            prepared.page.sessions[0],
            *original.values().next().unwrap()
        );
        assert!(!prepared.page.sessions[1].preview_candidates.is_empty());
        assert_eq!(index.current, original);
        assert_eq!(index.revision, 1);
        drop(prepared); // A response-construction failure must not commit the candidate.
        assert!(index.prepare_resize(0).is_err());
        assert!(index.prepare_resize(2049).is_err());
        assert_eq!(index.current, original);
        assert_eq!(index.revision, 1);
        let prepared = index.prepare_resize(2).unwrap();
        index.commit_resize(prepared);
        assert_eq!(index.raw, raw);
        assert_eq!(index.raw.values().next().unwrap() as *const _, raw_row);
        assert_eq!(&index._watcher as *const _, watcher);
        assert_eq!(index.last_reconcile, reconcile);
        assert_eq!(index.prepare_resize(2).unwrap().revision, 2);
        let delta = index.poll().unwrap().unwrap();
        assert_eq!(delta.revision, 3);
        let shrink = index.prepare_resize(1).unwrap();
        assert_eq!(shrink.revision, 4);
        index.commit_resize(shrink);
        assert_eq!(index.current.len(), 1);
        assert_eq!(index.prepare_resize(1).unwrap().revision, 4);
        drop(index);
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn same_limit_receipt_counts_out_of_window_changes_without_visible_revision() {
        let root = std::env::temp_dir().join(format!(
            "supercode-index-total-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        fs::create_dir_all(&root).unwrap();
        let root = root.canonicalize().unwrap();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::new(HarnessId::CODEX)],
            homes: crate::HarnessHomes {
                codex: root.clone(),
                ..crate::HarnessHomes::default()
            },
            limit: Some(1),
            ..DiscoveryQuery::default()
        };
        let (mut index, _) =
            SessionIndexSubscription::open(query, Arc::new(Notify::new())).unwrap();
        let visible = descriptor("visible", u64::MAX);
        index.raw = descriptor_map([visible]);
        index.rebuild_current(&BTreeSet::new()).unwrap();
        let (sender, receiver) = mpsc::channel();
        let _native_receiver = std::mem::replace(&mut index.receiver, receiver);
        let hidden = root.join("hidden.jsonl");
        fs::write(
            &hidden,
            "{\"type\":\"session_meta\",\"payload\":{\"id\":\"hidden\",\"cwd\":\"/workspace\"}}\n",
        )
        .unwrap();
        sender
            .send(Ok(
                Event::new(notify::EventKind::Any).add_path(hidden.clone())
            ))
            .unwrap();
        assert!(index.poll().unwrap().is_none());
        let added = index.prepare_resize(1).unwrap();
        assert_eq!(added.revision, 1);
        assert_eq!(added.page.receipt.total_matched, 2);
        fs::remove_file(&hidden).unwrap();
        sender
            .send(Ok(Event::new(notify::EventKind::Any).add_path(hidden)))
            .unwrap();
        assert!(index.poll().unwrap().is_none());
        let removed = index.prepare_resize(1).unwrap();
        assert_eq!(removed.revision, 1);
        assert_eq!(removed.page.receipt.total_matched, 1);
        drop(index);
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn hermes_store_appends_surface_as_index_updates() {
        // A private copy of the committed Hermes fixture store, in its own directory, so the
        // subscription watches exactly one store and the test may write to it.
        let root = std::env::temp_dir().join(format!(
            "supercode-index-hermes-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        fs::create_dir_all(&root).unwrap();
        let db = root.join("state.db");
        fs::copy(
            PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/hermes_home/state.db"),
            &db,
        )
        .unwrap();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::new(HarnessId::HERMES)],
            homes: crate::HarnessHomes {
                hermes: db.clone(),
                claude_code: root.join("missing-claude"),
                codex: root.join("missing-codex"),
                ..crate::HarnessHomes::default()
            },
            ..DiscoveryQuery::default()
        };
        let (mut subscription, initial) =
            SessionIndexSubscription::open(query, Arc::new(Notify::new())).unwrap();
        assert!(initial.len() >= 2, "{initial:#?}");
        assert!(initial
            .iter()
            .all(|descriptor| descriptor.locator.harness.as_str() == HarnessId::HERMES));
        assert!(
            subscription.poll().unwrap().is_none(),
            "quiet store, quiet index"
        );

        // Append one message the way Hermes does: a messages row plus the session's count bump.
        let target = initial[0].locator.session_id.clone();
        std::thread::sleep(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', 'index test append', ?2, 1)",
                rusqlite::params![target, 1_800_000_000.0_f64],
            )
            .unwrap();
            conn.execute(
                "UPDATE sessions SET message_count = message_count + 1, ended_at = ?2 WHERE id = ?1",
                rusqlite::params![target, 1_800_000_000.0_f64],
            )
            .unwrap();
        }
        // Wait for the native watcher (bounded), then poll: exactly the appended session changes.
        let deadline = Instant::now() + Duration::from_secs(5);
        let delta = loop {
            if let Some(delta) = subscription.poll().unwrap() {
                break delta;
            }
            assert!(
                Instant::now() < deadline,
                "no index delta after the store append"
            );
            std::thread::sleep(Duration::from_millis(50));
        };
        assert_eq!(delta.changes.len(), 1, "{delta:#?}");
        match &delta.changes[0] {
            SessionIndexChange::Updated { descriptor } => {
                assert_eq!(descriptor.locator.session_id, target);
                assert_eq!(
                    descriptor.message_count,
                    initial[0].message_count.map(|count| count + 1)
                );
            }
            other => panic!("expected an update for {target}, got {other:?}"),
        }
        assert!(
            subscription.poll().unwrap().is_none(),
            "one append, one delta"
        );
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn index_delta_is_a_complete_deterministic_replacement_set() {
        let before = descriptor_map([descriptor("removed", 1), descriptor("updated", 2)]);
        let after = descriptor_map([descriptor("updated", 3), descriptor("added", 4)]);
        let changes = diff_descriptors(&before, &after);
        assert!(matches!(
            &changes[0],
            SessionIndexChange::Added { descriptor } if descriptor.locator.session_id == "added"
        ));
        assert!(matches!(
            &changes[1],
            SessionIndexChange::Updated { descriptor } if descriptor.locator.session_id == "updated"
        ));
        assert!(matches!(
            &changes[2],
            SessionIndexChange::Removed { key } if key.session_id == "removed"
        ));
    }

    #[test]
    fn raw_index_projects_child_activity_into_one_root_row() {
        let root = descriptor("root", 10);
        let mut child = descriptor("child", 20);
        child.parent_session_id = Some("root".into());
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::new(HarnessId::CODEX)],
            limit: Some(100),
            ..DiscoveryQuery::default()
        };

        let projected = HarnessCatalog::new()
            .project_index(&query, [root, child])
            .unwrap();

        assert_eq!(projected.len(), 1);
        assert_eq!(projected[0].locator.session_id, "root");
        assert_eq!(projected[0].updated_at_ms, Some(20));
        assert_eq!(projected[0].child_session_count, 1);
    }

    #[test]
    fn complete_raw_index_backfills_a_bounded_page_without_discovery() {
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::new(HarnessId::CODEX)],
            limit: Some(2),
            ..DiscoveryQuery::default()
        };
        let catalog = HarnessCatalog::new();
        let mut raw = descriptor_map([
            descriptor("oldest", 1),
            descriptor("middle", 2),
            descriptor("newest", 3),
        ]);
        let initial = catalog
            .project_index(&query, raw.values().cloned())
            .unwrap();
        assert_eq!(
            initial
                .iter()
                .map(|descriptor| descriptor.locator.session_id.as_str())
                .collect::<Vec<_>>(),
            ["newest", "middle"]
        );

        raw.remove(&SessionIndexKey {
            harness: HarnessId::CODEX.into(),
            session_id: "newest".into(),
        });
        let after = catalog
            .project_index(&query, raw.values().cloned())
            .unwrap();
        assert_eq!(
            after
                .iter()
                .map(|descriptor| descriptor.locator.session_id.as_str())
                .collect::<Vec<_>>(),
            ["middle", "oldest"]
        );
    }

    #[test]
    fn append_reuses_an_immutable_header_but_replacement_does_not() {
        let mut existing = descriptor("session", 1);
        existing.cwd = Some(PathBuf::from("/workspace"));
        existing.model = Some("model".into());
        let before = FileFingerprint {
            len: 100,
            modified_ns: 1,
            modified_ms: Some(1),
            identity: 7,
        };
        let append = FileFingerprint {
            len: 200,
            modified_ns: 2,
            modified_ms: Some(2),
            identity: 7,
        };
        let replacement = FileFingerprint {
            identity: 8,
            ..append
        };

        assert!(can_reuse_header(&existing, before, append));
        assert!(!can_reuse_header(&existing, before, replacement));
    }

    #[tokio::test]
    async fn filesystem_event_wakes_index_without_a_poll_timer() {
        let nonce = std::time::SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let root = std::env::temp_dir().join(format!(
            "supercode-session-index-{}-{nonce}",
            std::process::id()
        ));
        let codex = root.join("codex");
        fs::create_dir_all(&codex).unwrap();
        let query = DiscoveryQuery {
            harnesses: vec![HarnessId::new(HarnessId::CODEX)],
            homes: crate::HarnessHomes {
                codex: codex.clone(),
                ..crate::HarnessHomes::default()
            },
            limit: Some(10),
            ..DiscoveryQuery::default()
        };
        let notifier = Arc::new(Notify::new());
        let (mut index, initial) =
            SessionIndexSubscription::open(query, Arc::clone(&notifier)).unwrap();
        assert!(initial.is_empty());

        let session = codex.join("new.jsonl");
        fs::write(
            &session,
            concat!(
                "{\"type\":\"session_meta\",\"payload\":{\"id\":\"new\",\"cwd\":\"/workspace\"}}\n",
                "{\"type\":\"turn_context\",\"payload\":{\"cwd\":\"/workspace\",\"model\":\"gpt-test\"}}\n"
            ),
        )
        .unwrap();

        tokio::time::timeout(Duration::from_secs(5), notifier.notified())
            .await
            .expect("filesystem invalidation should wake the index");
        let delta = index
            .poll()
            .unwrap()
            .expect("the filesystem event should produce a visible delta");
        assert!(matches!(
            &delta.changes[0],
            SessionIndexChange::Added { descriptor }
                if descriptor.locator.session_id == "new"
        ));

        drop(index);
        fs::remove_dir_all(root).unwrap();
    }
}