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 supercode_interchange::catalog::{hermes_session_stores, CodexHistoryTopicIndex};

use crate::{
    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 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)?;
        }
        // Nothing that feeds the index moved: whatever woke this poll (a reader's -wal close or attribute change, an
        // unrelated file beside a store) changed no session, and the rebuild would reproduce the index it has. On a
        // Hermes store its gateway's every read woke it, and rebuilding the whole index each time held a core.
        if content_dirty.is_empty() {
            return Ok(None);
        }
        let before = self.current.clone();
        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;
                    // A conversation's name can arrive after its header was read: Codex names (and renames) a thread
                    // in its own index, read on every write since its header never changes; Claude Code writes
                    // `ai-title` within the header's bounds, read while an untitled transcript is still within them.
                    if let StorageLocator::File { path } = &descriptor.locator.storage {
                        let harness = descriptor.locator.harness.as_str();
                        if harness == HarnessId::CODEX
                            || (descriptor.title.is_none() && old.len < TOPIC_HEADER_BYTES)
                        {
                            if let Some(name) = supercode_interchange::catalog::native_topic(
                                harness,
                                path,
                                &descriptor.locator.session_id,
                            ) {
                                descriptor.title = Some(name);
                            }
                        }
                    }
                    Some(descriptor)
                } else {
                    HarnessCatalog::new()
                        .refresh_file_index_descriptor_for(&locator, &self.query)
                        .map_err(|error| error.to_string())?
                }
            } else {
                HarnessCatalog::new()
                    .refresh_file_index_descriptor_for(&locator, &self.query)
                    .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 mut walked = BTreeSet::new();
    collect_jsonl_paths_in(root, paths, &mut walked);
}

/// Follows symlinked directories, as the catalog's walker does: a project directory moved to
/// another volume and linked back is still this home's. `walked` ends a link cycle.
fn collect_jsonl_paths_in(root: &Path, paths: &mut Vec<PathBuf>, walked: &mut BTreeSet<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 file_type) = entry.file_type() else {
            continue;
        };
        let path = entry.path();
        if file_type.is_symlink() {
            let Ok(target) = fs::metadata(&path) else {
                continue;
            };
            file_type = target.file_type();
        }
        if file_type.is_dir() {
            collect_jsonl_paths_in(&path, paths, walked);
        } 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,
    })
}

/// The bytes a header read covers (`read_header`'s bound): a Claude Code transcript past them without a topic will not
/// gain one there.
const TOPIC_HEADER_BYTES: u64 = 256 * 1024;

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
}