agentmux_srv/backend/history/

index.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! In-memory session index built from adapter discovery.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use super::adapter::*;

/// Format version of the saved metadata cache (`SessionIndex::with_cache_file`).
/// Bump when `SessionMeta` or what `extract_meta` derives changes meaning, so
/// a stale file is rebuilt rather than trusted.
pub(crate) const META_CACHE_VERSION: u32 = 1;

/// The saved metadata cache is rewritten at most this often (when changed).
pub(crate) const META_CACHE_WRITE_INTERVAL_MS: i64 = 10 * 60 * 1000;

#[derive(serde::Deserialize)]
struct MetaCacheFile {
    version: u32,
    entries: HashMap<String, (i64, Option<SessionMeta>)>,
}

#[derive(serde::Serialize)]
struct MetaCacheFileRef<'a> {
    version: u32,
    entries: &'a HashMap<String, (i64, Option<SessionMeta>)>,
}

/// Max characters of matched text returned per hit.
///
/// A single history message can be hundreds of KB, and this tool must never be
/// the reason a caller blows its own context window — the problem it exists to
/// solve is an agent not being able to audit itself, which is made worse, not
/// better, by an answer too large to read. Not theoretical: the
/// `GetAgentTranscript` call that motivated
/// `SPEC_AGENT_HISTORY_SEARCH_2026_09_17.md` returned 112 KB and had to be
/// spilled to a file.
const SNIPPET_MAX_CHARS: usize = 400;
/// Characters of leading context kept before the match inside a snippet.
const SNIPPET_LEAD_CHARS: usize = 120;

/// Query parameters for [`SessionIndex::search_sessions`].
#[derive(Debug, Clone, Default)]
pub struct HistorySearchOptions {
    /// Case-insensitive substring. Matched against message content, tool
    /// names, and tool argument summaries.
    pub query: String,
    /// Restrict to `"user"` or `"assistant"` messages.
    pub role: Option<String>,
    /// Only match tool calls whose name equals this (case-insensitive).
    ///
    /// Its own parameter rather than folding into `query` because "find where
    /// I *called* SendMessage" is a different question from "find where I
    /// wrote the word SendMessage", and only the structural answer settles an
    /// audit.
    pub tool: Option<String>,
    /// Max hits to return before reporting `truncated`.
    pub limit: usize,
    /// Only messages at or after this instant (Unix ms).
    pub since_ms: Option<i64>,
    /// Only messages at or before this instant (Unix ms).
    pub until_ms: Option<i64>,
    /// Open at most this many of the candidates, in the order given. The rest
    /// are counted in `sessions_skipped`, never silently dropped.
    pub max_sessions: Option<usize>,
    /// Also search sessions found only by working directory or account, not
    /// in the agent's own record (see `HistoryService::search_for_agent`).
    pub include_inferred: bool,
}

/// One match.
#[derive(Debug, Clone, serde::Serialize)]
pub struct HistorySearchHit {
    pub session_id: String,
    pub file_path: String,
    pub timestamp: i64,
    pub role: String,
    pub snippet: String,
    /// `Some(name)` when the hit is a tool call rather than message prose.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub tool_name: Option<String>,
    /// Where the text was read: `"provider_transcript"` or `"agentmux_record"`
    /// (the transcript is gone; AgentMux's own record of the session).
    pub source: &'static str,
    /// Why the session is the caller's: `"agent_record"` (its id is in the
    /// agent's own record or segment log) or `"inferred"` (found by working
    /// directory or account only).
    pub attribution: &'static str,
}

/// Where a searched session's content comes from.
#[derive(Debug, Clone)]
pub(crate) enum Source {
    Transcript,
    Record(super::record::RecordSession),
}

/// Why a session is the caller's.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum Attribution {
    /// Its id is in the agent's own record or segment log.
    AgentRecord,
    /// Found by working directory or account only.
    Inferred,
}

/// A session a search may open.
#[derive(Debug, Clone)]
pub(crate) struct SearchCandidate {
    pub meta: SessionMeta,
    pub source: Source,
    pub attribution: Attribution,
}

impl SearchCandidate {
    fn source_label(&self) -> &'static str {
        match self.source {
            Source::Transcript => "provider_transcript",
            Source::Record(_) => "agentmux_record",
        }
    }
    fn attribution_label(&self) -> &'static str {
        match self.attribution {
            Attribution::AgentRecord => "agent_record",
            Attribution::Inferred => "inferred",
        }
    }
}

/// Result of a bounded search.
#[derive(Debug, Clone, serde::Serialize)]
pub struct HistorySearchOutcome {
    pub hits: Vec<HistorySearchHit>,
    /// Sessions actually opened and read.
    pub sessions_scanned: u32,
    /// Candidates in the caller's window, counted before `max_sessions`.
    pub total_sessions: u32,
    /// True when the scan stopped on `limit` rather than exhausting every
    /// candidate — see `search_sessions`' doc comment for why this must never
    /// be conflated with "no matches".
    pub truncated: bool,
    /// True only when every candidate was read to the end: nothing cut by
    /// `max_sessions`, nothing unreadable, no hit limit reached. An empty
    /// `hits` is evidence that something did not happen only when this is
    /// true; otherwise the answer is "unknown", and `incomplete_reasons`
    /// says why.
    pub complete: bool,
    /// `"hit_limit"`, `"max_sessions"`, `"unreadable_sessions"`,
    /// `"unreadable_records"`, `"discovery_errors"`.
    pub incomplete_reasons: Vec<&'static str>,
    /// Candidates `max_sessions` left unopened.
    pub sessions_skipped: u32,
    /// Candidates that could not be read, and why.
    pub sessions_unreadable: Vec<UnreadableSession>,
    /// Candidates searched, but with records that could not be read.
    pub sessions_partly_read: Vec<UnreadableSession>,
    /// What the index refresh behind this search couldn't read. A session
    /// there may be missing from the candidates.
    pub discovery_errors: Vec<String>,
    /// Sessions the agent's own record or segment log names.
    pub ledger_sessions: u32,
    /// Of those, sessions whose provider transcript is gone, searched in
    /// AgentMux's record instead.
    pub record_only_sessions: u32,
    /// Sessions found only by working directory or account and left out,
    /// because the agent's own record doesn't name them. `include_inferred`
    /// searches them too.
    pub inferred_sessions_excluded: u32,
}

/// At most this many discovery errors are listed in a search's answer.
const DISCOVERY_ERRORS_LISTED: usize = 20;

impl HistorySearchOutcome {
    /// Record that the snapshot this search read couldn't see everything.
    pub fn add_discovery_errors(&mut self, problems: Vec<String>) {
        if problems.is_empty() {
            return;
        }
        let total = problems.len();
        self.discovery_errors = problems.into_iter().take(DISCOVERY_ERRORS_LISTED).collect();
        if total > DISCOVERY_ERRORS_LISTED {
            self.discovery_errors.push(format!("… and {} more", total - DISCOVERY_ERRORS_LISTED));
        }
        self.incomplete_reasons.push("discovery_errors");
        self.complete = false;
    }
}

/// A candidate session the search could not read.
#[derive(Debug, Clone, serde::Serialize)]
pub struct UnreadableSession {
    pub session_id: String,
    pub reason: String,
}

/// Whether a recorded tool name is the tool the caller named. Claude records
/// an MCP tool with its server prefix (`mcp__agentmux__SendMessage`) while
/// callers — and SearchHistory's own schema — name it bare (`SendMessage`), so
/// both forms match. Case-insensitive.
fn tool_name_matches(recorded: &str, wanted: &str) -> bool {
    let recorded = recorded.to_ascii_lowercase();
    let wanted = wanted.to_ascii_lowercase();
    recorded == wanted
        || (recorded.starts_with("mcp__") && recorded.ends_with(&format!("__{wanted}")))
}

/// Whether a message falls inside the caller's time window. A message with
/// no timestamp (0) can't be placed outside it, and dropping it could only
/// hide evidence, so it's kept.
fn within_window(timestamp_ms: i64, opts: &HistorySearchOptions) -> bool {
    timestamp_ms == 0
        || (opts.since_ms.is_none_or(|since| timestamp_ms >= since)
            && opts.until_ms.is_none_or(|until| timestamp_ms <= until))
}

/// Case-insensitive substring search returning a byte offset into the
/// **original** haystack, not into a lowercased copy of it.
///
/// The naive version of this (`haystack.to_lowercase().find(needle)`, then
/// slice `haystack` at the result) is wrong, and wrong in a way that can
/// panic: lowercasing is not length-preserving. Turkish `İ` (U+0130, 2 bytes)
/// lowercases to `i̇` (3 bytes), so every such character before the match
/// shifts the two strings out of alignment — slicing the original at an offset
/// derived from the lowercased copy can land mid-character and panic, or
/// silently return a misaligned snippet. Caught by reagentx P1 on PR #3321.
///
/// So the lowercase mapping is built explicitly, recording which original byte
/// each lowercased byte came from.
fn find_case_insensitive(haystack: &str, needle_lower: &str) -> Option<usize> {
    if needle_lower.is_empty() {
        return None;
    }
    let mut lowered = String::with_capacity(haystack.len());
    // lowered-byte-offset -> original-byte-offset
    let mut origin: Vec<usize> = Vec::with_capacity(haystack.len() + 1);
    for (orig_idx, ch) in haystack.char_indices() {
        for lc in ch.to_lowercase() {
            let mut buf = [0u8; 4];
            let encoded = lc.encode_utf8(&mut buf);
            for _ in 0..encoded.len() {
                origin.push(orig_idx);
            }
            lowered.push_str(encoded);
        }
    }
    origin.push(haystack.len());
    let at = lowered.find(needle_lower)?;
    // Always a char boundary in the original: every entry is the start index
    // of the char that produced it.
    origin.get(at).copied()
}

/// Bound a matched region to something a caller can actually read, keeping a
/// little context before the match so the hit is interpretable.
fn snippet_around(haystack: &str, match_at: usize) -> String {
    let start = haystack[..match_at]
        .char_indices()
        .rev()
        .take(SNIPPET_LEAD_CHARS)
        .last()
        .map(|(i, _)| i)
        .unwrap_or(match_at);
    let tail: String = haystack[start..].chars().take(SNIPPET_MAX_CHARS).collect();
    let mut out = String::new();
    if start > 0 {
        out.push('…');
    }
    out.push_str(&tail);
    if start + tail.len() < haystack.len() {
        out.push('…');
    }
    out
}

/// Every match within one message: its prose, plus each tool call.
fn match_message(
    candidate: &SearchCandidate,
    msg: &HistoryMessage,
    needle: &str,
    opts: &HistorySearchOptions,
) -> Vec<HistorySearchHit> {
    let meta = &candidate.meta;
    let mut hits = Vec::new();

    // Tool-call matches. When `tool` is set the search is structural: the
    // filter is the tool NAME, and an empty query then means "every call to
    // it" rather than "no matches".
    //
    // Every snippet offset below is resolved against the exact string the
    // snippet is cut from. Deriving an offset from one string and slicing
    // another — e.g. finding in `"{name} {args}"` and slicing `args` — puts
    // the window in the wrong place, and for a long argument summary silently
    // returns evidence that does not contain the match. A hit whose snippet
    // omits the match is worse than no hit: it looks like verification.
    // Both were reagentx P1s on PR #3321.
    for tu in &msg.tool_uses {
        if let Some(want) = opts.tool.as_deref() {
            if !tool_name_matches(&tu.name, want) {
                continue;
            }
            // Empty query + tool filter = "every call to this tool", so the
            // start of the arguments is the right window.
            let at = if needle.is_empty() {
                Some(0)
            } else {
                find_case_insensitive(&tu.argument_summary, needle)
            };
            if let Some(at) = at {
                hits.push(HistorySearchHit {
                    session_id: meta.session_id.clone(),
                    file_path: meta.file_path.clone(),
                    timestamp: msg.timestamp,
                    role: msg.role.clone(),
                    snippet: snippet_around(&tu.argument_summary, at),
                    tool_name: Some(tu.name.clone()),
                    source: candidate.source_label(),
                    attribution: candidate.attribution_label(),
                });
            }
            continue;
        }
        if needle.is_empty() {
            continue;
        }
        // No tool filter: the query may match the tool's NAME or its
        // arguments. Searched separately so the snippet offset always belongs
        // to `argument_summary`, the string actually being cut.
        let at = if find_case_insensitive(&tu.name, needle).is_some() {
            Some(0)
        } else {
            find_case_insensitive(&tu.argument_summary, needle)
        };
        if let Some(at) = at {
            hits.push(HistorySearchHit {
                session_id: meta.session_id.clone(),
                file_path: meta.file_path.clone(),
                timestamp: msg.timestamp,
                role: msg.role.clone(),
                snippet: snippet_around(&tu.argument_summary, at),
                tool_name: Some(tu.name.clone()),
                source: candidate.source_label(),
                attribution: candidate.attribution_label(),
            });
        }
    }

    // Prose match — skipped entirely when the caller asked a structural
    // tool-only question.
    if opts.tool.is_none() && !needle.is_empty() {
        if let Some(at) = find_case_insensitive(&msg.content, needle) {
            hits.push(HistorySearchHit {
                session_id: meta.session_id.clone(),
                file_path: meta.file_path.clone(),
                timestamp: msg.timestamp,
                role: msg.role.clone(),
                snippet: snippet_around(&msg.content, at),
                tool_name: None,
                source: candidate.source_label(),
                attribution: candidate.attribution_label(),
            });
        }
    }

    hits
}

/// Search message content and tool calls across a chosen set of sessions,
/// whose content `load` reads — from a provider transcript or from AgentMux's
/// record. See [`SessionIndex::search_sessions`] for why stopping early is
/// always reported.
pub(crate) fn search_candidates(
    candidates: &[SearchCandidate],
    opts: &HistorySearchOptions,
    load: &dyn Fn(&SearchCandidate) -> Result<Option<HistorySession>, HistoryError>,
) -> HistorySearchOutcome {
    let needle = opts.query.to_lowercase();
    let mut hits: Vec<HistorySearchHit> = Vec::new();
    let mut sessions_scanned = 0u32;
    let mut sessions_unreadable: Vec<UnreadableSession> = Vec::new();
    let mut sessions_partly_read: Vec<UnreadableSession> = Vec::new();
    let mut truncated = false;

    let opened = opts.max_sessions.map_or(candidates.len(), |max| max.min(candidates.len()));
    let sessions_skipped = (candidates.len() - opened) as u32;

    for candidate in &candidates[..opened] {
        let meta = &candidate.meta;
        if hits.len() >= opts.limit {
            // More candidates remained that were never opened.
            truncated = true;
            break;
        }
        // A session that fails to parse is skipped, not fatal: one
        // malformed file must not make the whole audit unanswerable. But
        // it's reported, never counted as read.
        let session = match load(candidate) {
            Ok(Some(session)) => session,
            Ok(None) => {
                sessions_unreadable.push(UnreadableSession {
                    session_id: meta.session_id.clone(),
                    reason: "no longer readable (moved, deleted or emptied)".into(),
                });
                continue;
            }
            Err(e) => {
                sessions_unreadable.push(UnreadableSession {
                    session_id: meta.session_id.clone(),
                    reason: e.to_string(),
                });
                continue;
            }
        };
        sessions_scanned += 1;
        if session.skipped_records > 0 {
            sessions_partly_read.push(UnreadableSession {
                session_id: meta.session_id.clone(),
                reason: format!(
                    "{} record(s) could not be read and were not searched",
                    session.skipped_records
                ),
            });
        }
        for msg in &session.messages {
            if let Some(role) = opts.role.as_deref() {
                if msg.role != role {
                    continue;
                }
            }
            if !within_window(msg.timestamp, opts) {
                continue;
            }
            if hits.len() >= opts.limit {
                truncated = true;
                break;
            }
            for hit in match_message(candidate, msg, &needle, opts) {
                if hits.len() >= opts.limit {
                    truncated = true;
                    break;
                }
                hits.push(hit);
            }
        }
    }

    let mut incomplete_reasons = Vec::new();
    if truncated {
        incomplete_reasons.push("hit_limit");
    }
    if sessions_skipped > 0 {
        incomplete_reasons.push("max_sessions");
    }
    if !sessions_unreadable.is_empty() {
        incomplete_reasons.push("unreadable_sessions");
    }
    if !sessions_partly_read.is_empty() {
        incomplete_reasons.push("unreadable_records");
    }
    HistorySearchOutcome {
        hits,
        sessions_scanned,
        total_sessions: candidates.len() as u32,
        truncated,
        complete: incomplete_reasons.is_empty(),
        incomplete_reasons,
        sessions_skipped,
        sessions_unreadable,
        sessions_partly_read,
        discovery_errors: Vec::new(),
        ledger_sessions: 0,
        record_only_sessions: 0,
        inferred_sessions_excluded: 0,
    }
}

/// AgentMux-ISOLATED provider-home roots under which delete/clear is permitted:
/// `<shared>/providers/` and `<shared>/identities/`. Anything outside these
/// (the user's personal `~/.claude` / `~/.config/claude-*`) is OFF-LIMITS so a
/// "clear all" can never nuke transcripts AgentMux didn't create.
fn default_isolated_roots() -> Vec<PathBuf> {
    let shared = std::env::var_os("AGENTMUX_SHARED_DIR")
        .map(PathBuf::from)
        .or_else(|| agentmux_common::data_paths::agentmux_root().ok().map(|root| root.join("shared")));
    match shared {
        Some(s) => vec![s.join("providers"), s.join("identities")],
        None => Vec::new(),
    }
}

/// Normalize a working directory for use as an index key.
///
/// Separators are unified and trailing ones dropped so the same directory
/// written `C:\x\y`, `C:/x/y` and `C:/x/y/` all collide on one key. Case is
/// folded on Windows only: its filesystem is case-insensitive, so `C:\Users`
/// and `c:\users` are genuinely the same directory — but on Linux they are
/// two different directories, and folding case there would merge the history
/// of two unrelated agents.
pub(crate) fn normalize_working_dir(dir: &str) -> String {
    let unified = dir.replace('\\', "/");
    let trimmed = unified.trim_end_matches('/');
    if cfg!(windows) {
        trimmed.to_lowercase()
    } else {
        trimmed.to_string()
    }
}

/// Outcome of [`SessionIndex::try_first_build`].
#[derive(Debug, PartialEq, Eq)]
pub enum FirstBuild {
    /// This call ran the first build.
    Built { discovered: u32 },
    /// Another caller already finished it.
    AlreadyBuilt,
    /// A refresh (possibly the first build) is running right now.
    Busy,
}

/// In-memory index of discovered sessions.
pub struct SessionIndex {
    /// session_id -> SessionMeta
    sessions: Mutex<HashMap<String, SessionMeta>>,
    /// normalized working directory -> session_ids.
    ///
    /// The identity index below can only find sessions for an agent that has
    /// a `db_agent_identity_links` row binding it to the account its
    /// transcripts were written under. An agent running on ambient
    /// credentials has no such row (and its registry record's `identity_id`
    /// reads `"default"`), so for that — the common — case the identity index
    /// resolves to nothing at all while the transcripts sit on disk perfectly
    /// intact. The working directory is recorded both in the transcript
    /// itself (`cwd`) and on the agent's own row, which makes it the join key
    /// that actually holds for those agents. See
    /// `docs/specs/SPEC_CROSS_CHANNEL_AGENT_HISTORY_RESOLUTION_2026_09_21.md`.
    /// **LOCK ORDER — `sessions` is always acquired BEFORE this map, and
    /// before `by_identity`.** Never the reverse, anywhere.
    ///
    /// `refresh()` holds the `sessions` guard while it swaps both index maps
    /// in, so it takes `sessions -> by_working_dir`. A reader that took
    /// `by_working_dir -> sessions` would be a textbook AB-BA deadlock the
    /// moment a `refresh()` (triggered by any `list`/`sessions_for_agent`
    /// call) overlapped it on the shared `Arc<SessionIndex>` — thread A
    /// holding `sessions` and blocking on the map, thread B holding the map
    /// and blocking on `sessions`. Both readers below therefore take
    /// `sessions` first, even though reading the map first would read more
    /// naturally. reagentx P1 on PR #3480.
    by_working_dir: Mutex<HashMap<String, Vec<String>>>,
    /// identity_id -> session_ids (only non-empty `SessionMeta::identity_id`
    /// values are indexed here). Maintained alongside `sessions` so "this
    /// identity's sessions" is an O(k) HashMap lookup + small per-identity
    /// sort, not an O(total sessions) scan of everything on disk — see
    /// `list_for_identity` and
    /// `docs/specs/SPEC_AGENT_IDENTITY_HISTORY_PERSISTENCE_PROTOCOL_2026_08_16.md`
    /// §4.4.
    by_identity: Mutex<HashMap<String, Vec<String>>>,
    /// file path -> (mtime it was parsed at, what it parsed to). A cold
    /// refresh JSON-parses every line of every transcript on the machine
    /// (42k files / 3.8 GB on one real host), so later refreshes reuse this
    /// for files whose mtime hasn't moved. Only touched under `refresh_lock`.
    meta_cache: Mutex<HashMap<String, (i64, Option<SessionMeta>)>>,
    /// Where `meta_cache` is kept between launches (`with_cache_file`), so the
    /// first refresh after a start parses only transcripts that changed since,
    /// instead of every transcript on the machine (63k sessions took 92–154 s
    /// per launch; SPEC_AGENT_OPEN_LATENCY_2026_09_27.md §4.3). `None`: kept in
    /// memory only.
    cache_file: Option<PathBuf>,
    /// The cache file has been read (once, at the first refresh).
    cache_loaded: std::sync::atomic::AtomicBool,
    /// Unix ms when the cache file was last written; 0 = never.
    cache_written_at_ms: std::sync::atomic::AtomicI64,
    /// Serializes refreshes. Held for a whole refresh, never while taking any
    /// lock above, so it adds no ordering constraint.
    pub(super) refresh_lock: Mutex<()>,
    /// Unix ms when the last refresh finished; 0 = never.
    refreshed_at_ms: std::sync::atomic::AtomicI64,
    /// Refreshes started so far; each takes the next number, under
    /// `refresh_lock`, before it lists anything.
    refreshes_started: std::sync::atomic::AtomicU64,
    /// The number of the last refresh that finished. A request that finds it
    /// higher than `refreshes_started` was when the request began knows a
    /// refresh listed the disk after it began — see [`Self::refresh_covering_now`].
    last_finished_refresh: std::sync::atomic::AtomicU64,
    /// What the last refresh couldn't read (see [`Discovery::problems`]),
    /// kept with the snapshot it produced.
    discovery_problems: Mutex<Vec<String>>,
    /// Adapters for all registered providers
    adapters: Vec<Box<dyn HistoryAdapter>>,
    /// Roots under which destructive ops (delete/clear) are allowed.
    isolated_roots: Vec<PathBuf>,
}

impl SessionIndex {
    pub fn new(adapters: Vec<Box<dyn HistoryAdapter>>) -> Self {
        Self::with_isolated_roots(adapters, default_isolated_roots())
    }

    /// Construct with explicit isolated roots (used by new() and tests).
    pub fn with_isolated_roots(
        adapters: Vec<Box<dyn HistoryAdapter>>,
        isolated_roots: Vec<PathBuf>,
    ) -> Self {
        SessionIndex {
            sessions: Mutex::new(HashMap::new()),
            by_working_dir: Mutex::new(HashMap::new()),
            by_identity: Mutex::new(HashMap::new()),
            meta_cache: Mutex::new(HashMap::new()),
            cache_file: None,
            cache_loaded: std::sync::atomic::AtomicBool::new(false),
            cache_written_at_ms: std::sync::atomic::AtomicI64::new(0),
            refresh_lock: Mutex::new(()),
            refreshed_at_ms: std::sync::atomic::AtomicI64::new(0),
            refreshes_started: std::sync::atomic::AtomicU64::new(0),
            last_finished_refresh: std::sync::atomic::AtomicU64::new(0),
            discovery_problems: Mutex::new(Vec::new()),
            adapters,
            isolated_roots,
        }
    }

    /// Keep the parsed-metadata cache in `path` between launches. Read at the
    /// first refresh (not here: the file can be tens of MB, and construction
    /// runs at srv startup), written after the first build and then at most
    /// every [`META_CACHE_WRITE_INTERVAL_MS`], only when something changed.
    pub fn with_cache_file(mut self, path: PathBuf) -> Self {
        self.cache_file = Some(path);
        self
    }

    /// Unix ms when the last refresh finished; 0 if none has.
    pub fn refreshed_at_ms(&self) -> i64 {
        self.refreshed_at_ms.load(std::sync::atomic::Ordering::Acquire)
    }

    /// [`Self::refresh`], unless another refresh is already running — then
    /// `None` at once instead of waiting it out.
    pub fn try_refresh(&self) -> Option<(u32, u32, u32)> {
        let guard = match self.refresh_lock.try_lock() {
            Ok(g) => g,
            Err(std::sync::TryLockError::Poisoned(p)) => p.into_inner(),
            Err(std::sync::TryLockError::WouldBlock) => return None,
        };
        Some(self.refresh_holding(guard))
    }

    /// The first build, unless one isn't needed or can't start now: never
    /// waits on the refresh lock, and checks "already built" while holding it
    /// (`refreshed_at_ms` is stored under the lock), so it can't follow
    /// another caller's first build with a second full pass.
    pub fn try_first_build(&self) -> FirstBuild {
        let guard = match self.refresh_lock.try_lock() {
            Ok(g) => g,
            Err(std::sync::TryLockError::Poisoned(p)) => p.into_inner(),
            Err(std::sync::TryLockError::WouldBlock) => return FirstBuild::Busy,
        };
        if self.refreshed_at_ms() > 0 {
            return FirstBuild::AlreadyBuilt;
        }
        let (discovered, _, _) = self.refresh_holding(guard);
        FirstBuild::Built { discovered }
    }

    /// Fill `meta_cache` from the cache file, once per index, if it is still
    /// empty. A missing, unreadable or other-version file is ignored: the
    /// refresh then parses everything, as it always did.
    fn load_cache_file_once(&self) {
        use std::sync::atomic::Ordering;
        let Some(path) = &self.cache_file else { return };
        if self.cache_loaded.swap(true, Ordering::AcqRel) {
            return;
        }
        let Ok(bytes) = std::fs::read(path) else { return };
        match serde_json::from_slice::<MetaCacheFile>(&bytes) {
            Ok(file) if file.version == META_CACHE_VERSION => {
                let mut cache = self.meta_cache.lock().unwrap();
                if cache.is_empty() {
                    tracing::info!(entries = file.entries.len(), "history: loaded the saved metadata cache");
                    *cache = file.entries;
                }
            }
            Ok(file) => tracing::info!(version = file.version, "history: saved metadata cache is another version; rebuilding"),
            Err(e) => tracing::warn!(error = %e, "history: saved metadata cache unreadable; rebuilding"),
        }
    }

    /// Save `cache` to the cache file when it changed: always after the first
    /// build, then at most every `META_CACHE_WRITE_INTERVAL_MS` (searches also
    /// refresh, and the file is tens of MB). Written to a temporary file and
    /// renamed over the old one, so a reader (another srv instance) never sees
    /// half a file. Failures are logged, never fatal: the cache only saves time.
    fn write_cache_file(&self, cache: &HashMap<String, (i64, Option<SessionMeta>)>, changed: bool) {
        use std::sync::atomic::Ordering;
        let Some(path) = &self.cache_file else { return };
        let now = chrono::Utc::now().timestamp_millis();
        let last = self.cache_written_at_ms.load(Ordering::Acquire);
        if !changed && last > 0 {
            return;
        }
        if last > 0 && now - last < META_CACHE_WRITE_INTERVAL_MS {
            return;
        }
        let tmp = path.with_extension(format!("tmp-{}", std::process::id()));
        let write = || -> std::io::Result<()> {
            if let Some(dir) = path.parent() {
                std::fs::create_dir_all(dir)?;
            }
            let body = serde_json::to_vec(&MetaCacheFileRef { version: META_CACHE_VERSION, entries: cache })
                .map_err(std::io::Error::other)?;
            std::fs::write(&tmp, body)?;
            std::fs::rename(&tmp, path)
        };
        // Recorded whether or not the write works: an unwritable cache (a
        // read-only shared dir, a full disk) then waits out the interval too,
        // instead of being serialized and retried on every refresh — and
        // searches refresh.
        self.cache_written_at_ms.store(now.max(1), Ordering::Release);
        match write() {
            Ok(()) => tracing::debug!(entries = cache.len(), "history: saved the metadata cache"),
            Err(e) => {
                // Never leave a partial or orphaned temporary file (tens of MB,
                // one per srv process) in the shared dir.
                let _ = std::fs::remove_file(&tmp);
                tracing::warn!(error = %e, path = %path.display(), "history: couldn't save the metadata cache");
            }
        }
    }

    /// True if `path` lives under an AgentMux-isolated provider home and is
    /// therefore safe to delete. Personal global homes are never isolated.
    fn is_isolated(&self, path: &Path) -> bool {
        self.isolated_roots.iter().any(|r| path.starts_with(r))
    }

    /// Full scan: discover all files and extract metadata.
    /// Returns (discovered, updated, new) counts.
    pub fn refresh(&self) -> (u32, u32, u32) {
        let guard = self.refresh_lock.lock().unwrap_or_else(|p| p.into_inner());
        self.refresh_holding(guard)
    }

    /// Make the snapshot include every session on disk when this call began,
    /// sharing the work with concurrent callers. A refresh that started after
    /// this call began already listed everything this call needs, so once it
    /// has finished there's nothing left to do; otherwise this runs its own.
    /// However many searches arrive together, that's at most two walks —
    /// every search running its own, one after another, put the fourth past
    /// the MCP's 10 s timeout on a heavy host (Codex P1 on #3693).
    pub fn refresh_covering_now(&self) {
        use std::sync::atomic::Ordering;
        let started_before_this_call = self.refreshes_started.load(Ordering::Acquire);
        let guard = self.refresh_lock.lock().unwrap_or_else(|p| p.into_inner());
        if self.last_finished_refresh.load(Ordering::Acquire) > started_before_this_call {
            return;
        }
        self.refresh_holding(guard);
    }

    /// What the last refresh couldn't read — a session there may be missing
    /// from the snapshot.
    pub fn discovery_problems(&self) -> Vec<String> {
        self.discovery_problems.lock().unwrap().clone()
    }

    fn refresh_holding(&self, _refreshing: std::sync::MutexGuard<'_, ()>) -> (u32, u32, u32) {
        use std::sync::atomic::Ordering;
        let this_refresh = self.refreshes_started.fetch_add(1, Ordering::AcqRel) + 1;
        let mut problems: Vec<String> = Vec::new();
        let mut discovered: u32 = 0;
        let mut updated: u32 = 0;
        let mut new_count: u32 = 0;
        self.load_cache_file_once();
        let old_cache = std::mem::take(&mut *self.meta_cache.lock().unwrap());
        let mut parsed: u32 = 0;
        let mut reused: usize = 0;
        let mut new_cache: HashMap<String, (i64, Option<SessionMeta>)> = HashMap::with_capacity(old_cache.len());

        let mut new_sessions: HashMap<String, SessionMeta> = HashMap::new();
        let mut new_by_identity: HashMap<String, Vec<String>> = HashMap::new();
        let mut new_by_working_dir: HashMap<String, Vec<String>> = HashMap::new();

        for adapter in &self.adapters {
            let files = match adapter.discover() {
                Ok(found) => {
                    problems.extend(found.problems);
                    found.files
                }
                Err(e) => {
                    tracing::warn!(
                        "history: failed to discover {} files: {}",
                        adapter.provider(),
                        e
                    );
                    problems.push(format!("{} discovery failed: {e}", adapter.provider()));
                    continue;
                }
            };

            discovered += files.len() as u32;

            for file in &files {
                let extracted = match old_cache.get(&file.file_path) {
                    Some((mtime, cached)) if *mtime == file.mtime_ms => {
                        reused += 1;
                        Ok(cached.clone())
                    }
                    _ => {
                        parsed += 1;
                        adapter.extract_meta(&file.file_path)
                    }
                };
                if let Ok(ref meta) = extracted {
                    new_cache.insert(file.file_path.clone(), (file.mtime_ms, meta.clone()));
                }
                match extracted {
                    Ok(Some(meta)) => {
                        if !meta.identity_id.is_empty() {
                            new_by_identity
                                .entry(meta.identity_id.clone())
                                .or_default()
                                .push(meta.session_id.clone());
                        }
                        if !meta.working_directory.is_empty() {
                            new_by_working_dir
                                .entry(normalize_working_dir(&meta.working_directory))
                                .or_default()
                                .push(meta.session_id.clone());
                        }
                        new_sessions.insert(meta.session_id.clone(), meta);
                    }
                    Ok(None) => {} // empty/invalid session
                    Err(e) => {
                        tracing::debug!(
                            "history: failed to extract meta from {}: {}",
                            file.file_path,
                            e
                        );
                        // A transcript that couldn't be indexed is missing
                        // from the snapshot, whoever it belongs to.
                        problems.push(format!("{}: {e}", file.file_path));
                    }
                }
            }
        }

        // Compare with existing index
        let mut sessions = self.sessions.lock().unwrap();
        for (id, _meta) in &new_sessions {
            if sessions.contains_key(id) {
                updated += 1;
            } else {
                new_count += 1;
            }
        }

        *sessions = new_sessions;
        *self.by_identity.lock().unwrap() = new_by_identity;
        *self.by_working_dir.lock().unwrap() = new_by_working_dir;
        drop(sessions);
        // Something to save: a transcript parsed afresh, or one gone since.
        let changed = parsed > 0 || reused < old_cache.len();
        self.write_cache_file(&new_cache, changed);
        tracing::debug!(parsed, reused, "history: refresh reused cached metadata");
        *self.meta_cache.lock().unwrap() = new_cache;
        if !problems.is_empty() {
            tracing::warn!(count = problems.len(), first = %problems[0], "history: discovery couldn't read everything");
        }
        *self.discovery_problems.lock().unwrap() = problems;
        self.refreshed_at_ms.store(
            chrono::Utc::now().timestamp_millis().max(1),
            Ordering::Release,
        );
        self.last_finished_refresh.store(this_refresh, Ordering::Release);

        (discovered, updated, new_count)
    }

    /// Every session recorded as having run in `working_directory`.
    ///
    /// Matching is on the normalized full path ([`normalize_working_dir`]),
    /// never a substring: `list`'s `project` filter uses `contains`, which
    /// would make an agent in `/agents/foo` also claim every session from
    /// `/agents/foo-2`.
    pub fn list_for_working_directory(&self, working_directory: &str) -> Vec<SessionMeta> {
        if working_directory.is_empty() {
            return Vec::new();
        }
        let key = normalize_working_dir(working_directory);
        // `sessions` FIRST — see the lock-order invariant on `by_working_dir`.
        let sessions = self.sessions.lock().unwrap();
        let by_working_dir = self.by_working_dir.lock().unwrap();
        let Some(session_ids) = by_working_dir.get(&key) else {
            return Vec::new();
        };
        session_ids.iter().filter_map(|id| sessions.get(id)).cloned().collect()
    }

    /// This identity's sessions, sorted/paginated the same way `list` is —
    /// but via the `by_identity` index instead of scanning every session on
    /// disk. `identity_id` is a bundle/account id (`SessionMeta::identity_id`
    /// — see its own doc comment); resolving an `agent_id` to one is the
    /// caller's job (`Store::agent_identity_list_for_agent`), not this
    /// index's — a bundle is a filesystem fact, which agent(s) claim it is
    /// a database fact, and this module has no Store access by design (it's
    /// pure discovery/indexing over the filesystem).
    pub fn list_for_identity(
        &self,
        identity_id: &str,
        offset: usize,
        limit: usize,
        sort_by: &str,
        sort_dir: &str,
    ) -> (Vec<SessionMeta>, u32, bool) {
        // `sessions` FIRST — see the lock-order invariant on `by_working_dir`.
        // This ordering was inverted here BEFORE PR #3480 (`refresh()` has
        // always held `sessions` while swapping `by_identity` in), so the same
        // AB-BA deadlock was already latent on `main` for the identity index.
        // Fixed alongside the `by_working_dir` one rather than leaving a known
        // instance of the bug in place next to its fix.
        let sessions = self.sessions.lock().unwrap();
        let by_identity = self.by_identity.lock().unwrap();
        let Some(session_ids) = by_identity.get(identity_id) else {
            return (Vec::new(), 0, false);
        };
        let mut filtered: Vec<&SessionMeta> =
            session_ids.iter().filter_map(|id| sessions.get(id)).collect();
        Self::sort_sessions(&mut filtered, sort_by, sort_dir);

        let total = filtered.len() as u32;
        let has_more = offset + limit < filtered.len();
        let page: Vec<SessionMeta> = filtered.into_iter().skip(offset).take(limit).cloned().collect();
        (page, total, has_more)
    }

    /// Shared sort logic between `list` (scans everything) and
    /// `list_for_identity` (scans one identity's sessions) — extracted so
    /// the two can't silently diverge on sort semantics.
    pub(crate) fn sort_sessions(filtered: &mut [&SessionMeta], sort_by: &str, sort_dir: &str) {
        let desc = sort_dir != "asc";
        match sort_by {
            "created_at" | "created" => {
                filtered.sort_by(|a, b| {
                    if desc {
                        b.created_at.cmp(&a.created_at)
                    } else {
                        a.created_at.cmp(&b.created_at)
                    }
                });
            }
            "messages" => {
                filtered.sort_by(|a, b| {
                    if desc {
                        b.message_count.cmp(&a.message_count)
                    } else {
                        a.message_count.cmp(&b.message_count)
                    }
                });
            }
            "tokens" => {
                filtered.sort_by(|a, b| {
                    if desc {
                        b.total_tokens.cmp(&a.total_tokens)
                    } else {
                        a.total_tokens.cmp(&b.total_tokens)
                    }
                });
            }
            _ => {
                // Default: modified_at desc
                filtered.sort_by(|a, b| {
                    if desc {
                        b.modified_at.cmp(&a.modified_at)
                    } else {
                        a.modified_at.cmp(&b.modified_at)
                    }
                });
            }
        }
    }

    /// List sessions with pagination and optional filters.
    pub fn list(
        &self,
        provider: Option<&str>,
        project: Option<&str>,
        offset: usize,
        limit: usize,
        sort_by: &str,
        sort_dir: &str,
    ) -> (Vec<SessionMeta>, u32, bool) {
        let sessions = self.sessions.lock().unwrap();

        let mut filtered: Vec<&SessionMeta> = sessions
            .values()
            .filter(|s| {
                if let Some(p) = provider {
                    if s.provider != p {
                        return false;
                    }
                }
                if let Some(proj) = project {
                    if !s.working_directory.contains(proj) {
                        return false;
                    }
                }
                true
            })
            .collect();

        Self::sort_sessions(&mut filtered, sort_by, sort_dir);

        let total = filtered.len() as u32;
        let has_more = offset + limit < filtered.len();
        let page: Vec<SessionMeta> = filtered
            .into_iter()
            .skip(offset)
            .take(limit)
            .cloned()
            .collect();

        (page, total, has_more)
    }

    /// Get a session by ID — returns just the meta from index.
    pub fn get_meta(&self, session_id: &str) -> Option<SessionMeta> {
        let sessions = self.sessions.lock().unwrap();
        sessions.get(session_id).cloned()
    }

    /// Full parse of a session by ID.
    pub fn get_full(&self, session_id: &str) -> Result<Option<HistorySession>, HistoryError> {
        let meta = match self.get_meta(session_id) {
            Some(m) => m,
            None => return Ok(None),
        };

        // Find the adapter for this provider
        for adapter in &self.adapters {
            if adapter.provider() == meta.provider {
                return adapter.parse_file(&meta.file_path);
            }
        }

        Err(HistoryError::Other(format!(
            "no adapter for provider: {}",
            meta.provider
        )))
    }

    /// Search message content and tool calls across a specific, already-chosen
    /// set of sessions.
    ///
    /// `SPEC_AGENT_HISTORY_SEARCH_2026_09_17.md` §3.1/§3.2. The caller picks
    /// the candidate sessions (normally `HistoryService::search_for_agent`,
    /// which resolves them from the agent's linked identities and applies the
    /// `since`/`until` window) — this function only scans what it is handed,
    /// so the expensive `get_full` parse never touches a session the caller
    /// already ruled out on cheap indexed metadata.
    ///
    /// Bounded deliberately: it stops at `limit` hits and reports whether it
    /// stopped early. **That signal is load-bearing.** A search that silently
    /// truncates and returns nothing reproduces the exact failure this whole
    /// feature exists to prevent — an agent concluding "I never did that" from
    /// an incomplete audit — so "no matches" and "ran out of budget" must never
    /// look the same to a caller. The same goes for every other way a
    /// candidate goes unread (a `max_sessions` cut, a file that won't parse):
    /// each is counted, and any of them makes the answer not `complete`.
    #[cfg(test)]
    pub fn search_sessions(
        &self,
        candidates: &[SessionMeta],
        opts: &HistorySearchOptions,
    ) -> HistorySearchOutcome {
        let candidates: Vec<SearchCandidate> = candidates
            .iter()
            .map(|meta| SearchCandidate {
                meta: meta.clone(),
                source: Source::Transcript,
                attribution: Attribution::AgentRecord,
            })
            .collect();
        search_candidates(&candidates, opts, &|c| self.get_full(&c.meta.session_id))
    }

    /// Check if the index has been populated.
    #[cfg(test)]
    pub fn is_empty(&self) -> bool {
        self.sessions.lock().unwrap().is_empty()
    }

    /// Delete a session: remove its on-disk transcript (and the sibling
    /// `<session_id>/` subagents dir Claude keeps next to it) and drop it from
    /// the index. Returns Ok(true) if a file was removed, Ok(false) if the
    /// session id is unknown.
    pub fn delete(&self, session_id: &str) -> Result<bool, HistoryError> {
        let meta = match self.get_meta(session_id) {
            Some(m) => m,
            None => return Ok(false),
        };
        let path = Path::new(&meta.file_path);
        // Safety: only delete inside AgentMux-isolated homes — never the user's
        // personal Claude/Codex/... transcripts that the browse also surfaces.
        if !self.is_isolated(path) {
            return Err(HistoryError::Other(format!(
                "refusing to delete '{}' — it lives in your personal {} home, not AgentMux's",
                meta.file_path, meta.provider
            )));
        }
        // Treat an already-removed file (GC'd out-of-band) as success so the
        // stale index entry still gets dropped (otherwise clear-all can't
        // converge and the index stays non-empty).
        match std::fs::remove_file(path) {
            Ok(()) => {}
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
            Err(e) => return Err(e.into()),
        }
        // Claude stores subagent transcripts in a sibling `<session_id>/` dir;
        // remove it too so the clear actually frees the space. Best-effort.
        if let Some(parent) = path.parent() {
            let sidecar = parent.join(session_id);
            if sidecar.is_dir() {
                let _ = std::fs::remove_dir_all(&sidecar);
            }
        }
        self.sessions.lock().unwrap().remove(session_id);
        Ok(true)
    }

    /// Bulk-delete all indexed sessions matching the optional provider/project
    /// filter (no filter = clear everything). Returns the number removed.
    pub fn clear(&self, provider: Option<&str>, project: Option<&str>) -> u32 {
        // Snapshot matching ids while holding the lock, then do fs ops without
        // it (delete() re-locks per id).
        let ids: Vec<String> = {
            let sessions = self.sessions.lock().unwrap();
            sessions
                .values()
                .filter(|s| provider.map_or(true, |p| s.provider == p))
                .filter(|s| project.map_or(true, |proj| s.working_directory.contains(proj)))
                // Never bulk-delete the user's personal global transcripts.
                .filter(|s| self.is_isolated(Path::new(&s.file_path)))
                .map(|s| s.session_id.clone())
                .collect()
        };
        let mut deleted = 0;
        for id in ids {
            if matches!(self.delete(&id), Ok(true)) {
                deleted += 1;
            }
        }
        deleted
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use std::path::PathBuf;

    /// Adapter that "discovers" caller-supplied files and derives the session id
    /// from each file stem. Just enough to populate the index for delete tests.
    struct MockAdapter {
        files: Vec<DiscoveredFile>,
        provider: String,
        working_directory: String,
        identity_id: String,
    }
    impl HistoryAdapter for MockAdapter {
        fn provider(&self) -> &str {
            &self.provider
        }
        fn discover_files(&self) -> Result<Vec<DiscoveredFile>, HistoryError> {
            Ok(self
                .files
                .iter()
                .map(|f| DiscoveredFile { file_path: f.file_path.clone(), mtime_ms: f.mtime_ms })
                .collect())
        }
        fn extract_meta(&self, file_path: &str) -> Result<Option<SessionMeta>, HistoryError> {
            let id = PathBuf::from(file_path)
                .file_stem()
                .map(|s| s.to_string_lossy().into_owned())
                .unwrap_or_default();
            Ok(Some(SessionMeta {
                session_id: id,
                file_path: file_path.to_string(),
                provider: self.provider.clone(),
                model: String::new(),
                slug: String::new(),
                working_directory: self.working_directory.clone(),
                created_at: 0,
                modified_at: 0,
                message_count: 0,
                first_user_message: String::new(),
                file_size_bytes: 0,
                git_branch: String::new(),
                total_tokens: 0,
                subagent_count: 0,
                identity_id: self.identity_id.clone(),
                starts_undated: false,
            }))
        }
        fn parse_file(&self, _: &str) -> Result<Option<HistorySession>, HistoryError> {
            Ok(None)
        }
    }

    fn write_session(dir: &std::path::Path, id: &str) -> String {
        let f = dir.join(format!("{id}.jsonl"));
        fs::write(&f, b"{}").unwrap();
        f.to_string_lossy().into_owned()
    }

    #[test]
    fn delete_removes_file_and_index_entry() {
        let dir = std::env::temp_dir().join(format!("amux-hist-del-{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        let fp = write_session(&dir, "sess-abc");
        // A sibling subagents dir should be removed too.
        let sidecar = dir.join("sess-abc");
        fs::create_dir_all(sidecar.join("subagents")).unwrap();

        let idx = SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files: vec![DiscoveredFile { file_path: fp.clone(), mtime_ms: 0 }],
                provider: "mock".into(),
                working_directory: "/proj".into(),
                identity_id: String::new(),
            })],
            vec![dir.clone()],
        );
        idx.refresh();
        assert!(idx.get_meta("sess-abc").is_some());

        assert!(matches!(idx.delete("sess-abc"), Ok(true)));
        assert!(!std::path::Path::new(&fp).exists(), "transcript removed");
        assert!(!sidecar.exists(), "subagents sidecar removed");
        assert!(idx.get_meta("sess-abc").is_none(), "dropped from index");
        assert!(matches!(idx.delete("sess-abc"), Ok(false)), "second delete is a no-op");

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn clear_respects_provider_filter() {
        let dir = std::env::temp_dir().join(format!("amux-hist-clr-{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        let a = write_session(&dir, "a");
        let b = write_session(&dir, "b");

        let idx = SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files: vec![
                    DiscoveredFile { file_path: a.clone(), mtime_ms: 0 },
                    DiscoveredFile { file_path: b.clone(), mtime_ms: 0 },
                ],
                provider: "mock".into(),
                working_directory: "/proj".into(),
                identity_id: String::new(),
            })],
            vec![dir.clone()],
        );
        idx.refresh();

        // Non-matching provider clears nothing; matching clears both.
        assert_eq!(idx.clear(Some("other"), None), 0);
        assert!(std::path::Path::new(&a).exists());
        assert_eq!(idx.clear(Some("mock"), None), 2);
        assert!(!std::path::Path::new(&a).exists() && !std::path::Path::new(&b).exists());
        assert!(idx.is_empty());

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn refuses_to_delete_outside_isolated_roots() {
        // A session that lives OUTSIDE the isolated roots (i.e. the user's
        // personal global home) must never be deletable.
        let dir = std::env::temp_dir().join(format!("amux-hist-guard-{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        let fp = write_session(&dir, "personal-sess");

        let idx = SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files: vec![DiscoveredFile { file_path: fp.clone(), mtime_ms: 0 }],
                provider: "claude".into(),
                working_directory: "/home/me/.claude".into(),
                identity_id: String::new(),
            })],
            // isolated roots deliberately do NOT include `dir`.
            vec![std::env::temp_dir().join("amux-some-other-isolated-root")],
        );
        idx.refresh();

        assert!(idx.delete("personal-sess").is_err(), "single delete must refuse");
        assert!(std::path::Path::new(&fp).exists(), "file must be intact");
        assert_eq!(idx.clear(None, None), 0, "clear-all must skip it");
        assert!(std::path::Path::new(&fp).exists(), "still intact after clear");

        let _ = fs::remove_dir_all(&dir);
    }

    // docs/specs/SPEC_AGENT_IDENTITY_HISTORY_PERSISTENCE_PROTOCOL_2026_08_16.md
    // §4.4: list_for_identity is the actual "fast lookup" this whole step
    // exists for — an O(sessions for this identity) HashMap-backed lookup
    // instead of list()'s O(total sessions on disk) scan.

    #[test]
    fn list_for_identity_returns_only_that_identitys_sessions() {
        let dir = std::env::temp_dir().join(format!("amux-hist-idx-{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        let a1 = write_session(&dir, "a1");
        let a2 = write_session(&dir, "a2");
        let b1 = write_session(&dir, "b1");
        let unattributed = write_session(&dir, "unattributed");

        let idx = SessionIndex::with_isolated_roots(
            vec![
                Box::new(MockAdapter {
                    files: vec![
                        DiscoveredFile { file_path: a1, mtime_ms: 1 },
                        DiscoveredFile { file_path: a2, mtime_ms: 2 },
                    ],
                    provider: "mock".into(),
                    working_directory: "/proj".into(),
                    identity_id: "identity-a".into(),
                }),
                Box::new(MockAdapter {
                    files: vec![DiscoveredFile { file_path: b1, mtime_ms: 1 }],
                    provider: "mock".into(),
                    working_directory: "/proj".into(),
                    identity_id: "identity-b".into(),
                }),
                Box::new(MockAdapter {
                    files: vec![DiscoveredFile { file_path: unattributed, mtime_ms: 1 }],
                    provider: "mock".into(),
                    working_directory: "/proj".into(),
                    identity_id: String::new(),
                }),
            ],
            vec![dir.clone()],
        );
        idx.refresh();

        let (sessions_a, total_a, _) = idx.list_for_identity("identity-a", 0, 10, "created_at", "desc");
        assert_eq!(total_a, 2);
        let ids_a: std::collections::HashSet<_> = sessions_a.iter().map(|s| s.session_id.clone()).collect();
        assert_eq!(ids_a, std::collections::HashSet::from(["a1".to_string(), "a2".to_string()]));

        let (sessions_b, total_b, _) = idx.list_for_identity("identity-b", 0, 10, "created_at", "desc");
        assert_eq!(total_b, 1);
        assert_eq!(sessions_b[0].session_id, "b1");

        let _ = fs::remove_dir_all(&dir);
    }

    #[test]
    fn list_for_identity_returns_empty_for_an_unknown_identity() {
        let idx = SessionIndex::with_isolated_roots(vec![], vec![]);
        idx.refresh();
        let (sessions, total, has_more) = idx.list_for_identity("no-such-identity", 0, 10, "created_at", "desc");
        assert!(sessions.is_empty());
        assert_eq!(total, 0);
        assert!(!has_more);
    }

    #[test]
    fn list_for_identity_paginates_and_respects_sort_direction() {
        let dir = std::env::temp_dir().join(format!("amux-hist-idx-page-{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        let files: Vec<DiscoveredFile> = (0..3)
            .map(|i| DiscoveredFile { file_path: write_session(&dir, &format!("s{i}")), mtime_ms: i })
            .collect();

        let idx = SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files,
                provider: "mock".into(),
                working_directory: "/proj".into(),
                identity_id: "identity-a".into(),
            })],
            vec![dir.clone()],
        );
        idx.refresh();

        let (page1, total, has_more) = idx.list_for_identity("identity-a", 0, 2, "created_at", "desc");
        assert_eq!(total, 3);
        assert_eq!(page1.len(), 2);
        assert!(has_more);
        let (page2, _, has_more2) = idx.list_for_identity("identity-a", 2, 2, "created_at", "desc");
        assert_eq!(page2.len(), 1);
        assert!(!has_more2);

        let _ = fs::remove_dir_all(&dir);
    }

    /// Hammers the two conflicting lock orders against each other on the same
    /// `Arc<SessionIndex>` a real deployment shares between requests.
    ///
    /// A deadlock does not *fail* a test, it hangs it — and a hung test hangs
    /// CI rather than reporting anything useful. So the workers signal a
    /// channel and the assertion is a bounded `recv_timeout`: a deadlock
    /// becomes a normal test failure with a readable message, and the threads
    /// are deliberately not joined (joining a deadlocked thread would hang
    /// exactly as badly as the bug).
    ///
    /// Against the pre-fix ordering — `refresh()` taking `sessions ->
    /// by_working_dir` while `list_for_working_directory()` took
    /// `by_working_dir -> sessions` — this wedges. reagentx P1 on PR #3480.
    #[test]
    fn concurrent_refresh_and_lookups_do_not_deadlock() {
        use std::sync::mpsc;
        use std::sync::Arc;
        use std::time::Duration;

        let dir = std::env::temp_dir().join(format!("amux-hist-lockorder-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let f = write_session(&dir, "lock-order-session");

        let index = Arc::new(SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files: vec![DiscoveredFile { file_path: f, mtime_ms: 1 }],
                provider: "mock".into(),
                working_directory: "/proj".into(),
                identity_id: "acct-lock".into(),
            })],
            vec![dir.clone()],
        ));

        const ITERATIONS: usize = 300;
        let (tx, rx) = mpsc::channel();

        for worker in 0..3 {
            let idx = Arc::clone(&index);
            let tx = tx.clone();
            std::thread::spawn(move || {
                for _ in 0..ITERATIONS {
                    match worker {
                        0 => {
                            idx.refresh();
                        }
                        1 => {
                            idx.list_for_working_directory("/proj");
                        }
                        _ => {
                            idx.list_for_identity("acct-lock", 0, 10, "created_at", "desc");
                        }
                    }
                }
                let _ = tx.send(worker);
            });
        }
        drop(tx);

        for _ in 0..3 {
            rx.recv_timeout(Duration::from_secs(30)).expect(
                "deadlock: a worker never finished. `refresh()` holds `sessions` while \
                 swapping the index maps, so every reader must take `sessions` FIRST — \
                 see the lock-order invariant on `SessionIndex::by_working_dir`",
            );
        }

        std::fs::remove_dir_all(&dir).ok();
    }

    #[test]
    fn refresh_rebuilds_the_identity_index_from_scratch_each_time() {
        // A session that's re-discovered under a DIFFERENT identity on a
        // second refresh (e.g. its bundle got re-linked) must not leave a
        // stale entry under the old identity_id.
        let dir = std::env::temp_dir().join(format!("amux-hist-idx-rebuild-{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        let fp = write_session(&dir, "moved");

        let idx = SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files: vec![DiscoveredFile { file_path: fp.clone(), mtime_ms: 1 }],
                provider: "mock".into(),
                working_directory: "/proj".into(),
                identity_id: "identity-old".into(),
            })],
            vec![dir.clone()],
        );
        idx.refresh();
        assert_eq!(idx.list_for_identity("identity-old", 0, 10, "created_at", "desc").1, 1);

        let idx2 = SessionIndex::with_isolated_roots(
            vec![Box::new(MockAdapter {
                files: vec![DiscoveredFile { file_path: fp, mtime_ms: 1 }],
                provider: "mock".into(),
                working_directory: "/proj".into(),
                identity_id: "identity-new".into(),
            })],
            vec![dir.clone()],
        );
        idx2.refresh();
        assert_eq!(idx2.list_for_identity("identity-old", 0, 10, "created_at", "desc").1, 0, "stale identity must not linger");
        assert_eq!(idx2.list_for_identity("identity-new", 0, 10, "created_at", "desc").1, 1);

        let _ = fs::remove_dir_all(&dir);
    }

    // ── History search (SPEC_AGENT_HISTORY_SEARCH_2026_09_17.md) ──

    /// Adapter serving canned messages, recording which files were actually
    /// opened — the recording is what lets a test prove the search never
    /// parses a session the caller already excluded.
    struct SearchMockAdapter {
        sessions: HashMap<String, Vec<HistoryMessage>>,
        parsed: Mutex<Vec<String>>,
    }

    impl HistoryAdapter for SearchMockAdapter {
        fn provider(&self) -> &str {
            "mock"
        }
        fn discover_files(&self) -> Result<Vec<DiscoveredFile>, HistoryError> {
            Ok(self
                .sessions
                .keys()
                .map(|id| DiscoveredFile { file_path: format!("/tmp/{id}.jsonl"), mtime_ms: 0 })
                .collect())
        }
        fn extract_meta(&self, file_path: &str) -> Result<Option<SessionMeta>, HistoryError> {
            let id = PathBuf::from(file_path)
                .file_stem()
                .map(|s| s.to_string_lossy().into_owned())
                .unwrap_or_default();
            Ok(Some(SessionMeta {
                session_id: id,
                file_path: file_path.to_string(),
                provider: "mock".into(),
                model: String::new(),
                slug: String::new(),
                working_directory: "/proj".into(),
                created_at: 0,
                modified_at: 0,
                message_count: 0,
                first_user_message: String::new(),
                file_size_bytes: 0,
                git_branch: String::new(),
                total_tokens: 0,
                subagent_count: 0,
                identity_id: String::new(),
                starts_undated: false,
            }))
        }
        fn parse_file(&self, file_path: &str) -> Result<Option<HistorySession>, HistoryError> {
            self.parsed.lock().unwrap().push(file_path.to_string());
            let id = PathBuf::from(file_path)
                .file_stem()
                .map(|s| s.to_string_lossy().into_owned())
                .unwrap_or_default();
            if id.starts_with("corrupt") {
                return Err(HistoryError::Other("unexpected end of JSON".into()));
            }
            let Some(messages) = self.sessions.get(&id) else { return Ok(None) };
            let meta = self.extract_meta(file_path)?.unwrap();
            let skipped_records = if id.starts_with("partial") { 2 } else { 0 };
            Ok(Some(HistorySession { meta, messages: messages.clone(), skipped_records }))
        }
    }

    fn msg(role: &str, content: &str, tools: Vec<(&str, &str)>) -> HistoryMessage {
        HistoryMessage {
            role: role.to_string(),
            content: content.to_string(),
            timestamp: 100,
            tool_uses: tools
                .into_iter()
                .map(|(n, a)| ToolUseSummary { name: n.to_string(), argument_summary: a.to_string() })
                .collect(),
        }
    }

    fn search_index(sessions: Vec<(&str, Vec<HistoryMessage>)>) -> SessionIndex {
        let map: HashMap<String, Vec<HistoryMessage>> =
            sessions.into_iter().map(|(id, m)| (id.to_string(), m)).collect();
        let idx = SessionIndex::new(vec![Box::new(SearchMockAdapter {
            sessions: map,
            parsed: Mutex::new(Vec::new()),
        })]);
        idx.refresh();
        idx
    }

    fn opts(query: &str) -> HistorySearchOptions {
        HistorySearchOptions { query: query.into(), limit: 50, ..Default::default() }
    }

    fn at(timestamp: i64, content: &str) -> HistoryMessage {
        HistoryMessage { timestamp, ..msg("assistant", content, vec![]) }
    }

    #[test]
    fn a_bare_tool_name_matches_the_mcp_prefixed_call() {
        // Claude records an MCP tool with its server prefix, while callers —
        // and SearchHistory's own schema — name it bare. "Did I call
        // SendMessage" must find the AgentMux tool, not only the built-in one.
        let idx = search_index(vec![(
            "s1",
            vec![msg("assistant", "sent", vec![("mcp__agentmux__SendMessage", r#"{"to":"Agent4"}"#)])],
        )]);
        let cands = vec![idx.get_meta("s1").unwrap()];
        for wanted in ["SendMessage", "sendmessage", "mcp__agentmux__SendMessage"] {
            let mut o = opts("");
            o.tool = Some(wanted.into());
            let out = idx.search_sessions(&cands, &o);
            assert_eq!(out.hits.len(), 1, "tool: {wanted}");
            assert_eq!(out.hits[0].tool_name.as_deref(), Some("mcp__agentmux__SendMessage"));
        }
        let mut other = opts("");
        other.tool = Some("Message".into());
        assert!(idx.search_sessions(&cands, &other).hits.is_empty(), "a suffix alone is not a name");
    }

    #[test]
    fn messages_outside_the_time_window_are_not_hits() {
        let idx = search_index(vec![(
            "s1",
            vec![at(1_000, "deploy early"), at(5_000, "deploy late"), at(0, "deploy undated")],
        )]);
        let cands = vec![idx.get_meta("s1").unwrap()];
        let mut o = opts("deploy");
        o.since_ms = Some(2_000);
        let snippets: Vec<String> = idx.search_sessions(&cands, &o).hits.into_iter().map(|h| h.snippet).collect();
        // An undated message can't be placed outside the window, and dropping
        // it could only hide evidence.
        assert_eq!(snippets, vec!["deploy late", "deploy undated"]);

        let mut o = opts("deploy");
        o.until_ms = Some(2_000);
        let snippets: Vec<String> = idx.search_sessions(&cands, &o).hits.into_iter().map(|h| h.snippet).collect();
        assert_eq!(snippets, vec!["deploy early", "deploy undated"]);
    }

    #[test]
    fn a_full_search_says_it_is_complete() {
        let idx = search_index(vec![("s1", vec![msg("assistant", "nothing relevant", vec![])])]);
        let cands = vec![idx.get_meta("s1").unwrap()];
        let out = idx.search_sessions(&cands, &opts("never said"));
        assert!(out.hits.is_empty());
        assert!(out.complete, "every candidate was read to the end");
        assert!(out.incomplete_reasons.is_empty());
    }

    #[test]
    fn a_max_sessions_cut_is_counted_and_makes_the_answer_incomplete() {
        let idx = search_index(vec![
            ("a", vec![msg("assistant", "x", vec![])]),
            ("b", vec![msg("assistant", "x", vec![])]),
            ("c", vec![msg("assistant", "x", vec![])]),
        ]);
        let cands: Vec<SessionMeta> = ["a", "b", "c"].iter().map(|id| idx.get_meta(id).unwrap()).collect();
        let mut o = opts("absent");
        o.max_sessions = Some(1);
        let out = idx.search_sessions(&cands, &o);
        assert_eq!(out.total_sessions, 3, "counted before the cut");
        assert_eq!(out.sessions_scanned, 1);
        assert_eq!(out.sessions_skipped, 2);
        assert!(!out.complete);
        assert_eq!(out.incomplete_reasons, vec!["max_sessions"]);
    }

    #[test]
    fn a_session_with_unreadable_records_makes_the_answer_incomplete() {
        // What the session's readable records hold is still found; that it
        // had records the search couldn't read is reported, so "no hit" for
        // anything else isn't taken as proof (Codex P1 on #3693).
        let idx = search_index(vec![("partial-1", vec![msg("assistant", "deploy done", vec![])])]);
        let cands = vec![idx.get_meta("partial-1").unwrap()];
        let out = idx.search_sessions(&cands, &opts("deploy"));
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.sessions_scanned, 1, "it was read, partly");
        assert_eq!(out.sessions_partly_read.len(), 1);
        assert_eq!(out.sessions_partly_read[0].session_id, "partial-1");
        assert!(out.sessions_partly_read[0].reason.contains("2 record"), "{}", out.sessions_partly_read[0].reason);
        assert!(!out.complete);
        assert_eq!(out.incomplete_reasons, vec!["unreadable_records"]);
    }

    #[test]
    fn an_unreadable_session_is_reported_not_counted_as_scanned() {
        let idx = search_index(vec![
            ("good", vec![msg("assistant", "x", vec![])]),
            ("corrupt-1", vec![]),
        ]);
        let cands: Vec<SessionMeta> =
            ["good", "corrupt-1"].iter().map(|id| idx.get_meta(id).unwrap()).collect();
        let out = idx.search_sessions(&cands, &opts("absent"));
        assert_eq!(out.sessions_scanned, 1, "only the session actually read");
        assert_eq!(out.sessions_unreadable.len(), 1);
        assert_eq!(out.sessions_unreadable[0].session_id, "corrupt-1");
        assert!(out.sessions_unreadable[0].reason.contains("unexpected end of JSON"));
        assert!(!out.complete);
        assert_eq!(out.incomplete_reasons, vec!["unreadable_sessions"]);
    }

    #[test]
    fn search_matches_message_content() {
        let idx = search_index(vec![("s1", vec![msg("assistant", "I merged the WAN signing PR", vec![])])]);
        let cands = vec![idx.get_meta("s1").unwrap()];
        let out = idx.search_sessions(&cands, &opts("wan signing"));
        assert_eq!(out.hits.len(), 1);
        assert!(out.hits[0].snippet.contains("WAN signing"), "snippet carries the match");
        assert!(!out.truncated);
    }

    #[test]
    fn search_matches_a_tool_call_structurally_not_just_prose() {
        // The question that motivated this whole feature: "did I call
        // SendMessage to Agent4" must be answerable even when no prose in the
        // transcript mentions it.
        let args = r#"{"to":"Agent4","message":"following up"}"#;
        let idx = search_index(vec![(
            "s1",
            vec![msg("assistant", "sure, sending now", vec![("SendMessage", args)])],
        )]);
        let cands = vec![idx.get_meta("s1").unwrap()];

        let mut o = opts("");
        o.tool = Some("sendmessage".into()); // case-insensitive
        let out = idx.search_sessions(&cands, &o);
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.hits[0].tool_name.as_deref(), Some("SendMessage"));
        assert!(out.hits[0].snippet.contains("Agent4"));

        // And filtering by argument content within that tool.
        let mut o2 = opts("agent4");
        o2.tool = Some("SendMessage".into());
        assert_eq!(idx.search_sessions(&cands, &o2).hits.len(), 1);
        let mut o3 = opts("agent9");
        o3.tool = Some("SendMessage".into());
        assert_eq!(idx.search_sessions(&cands, &o3).hits.len(), 0);
    }

    #[test]
    fn an_exhausted_budget_is_distinguishable_from_no_matches() {
        // The single most important property here. A search that silently
        // stops early and returns nothing recreates the exact failure this
        // feature exists to prevent: an agent concluding "I never did that"
        // from an incomplete audit.
        let many: Vec<HistoryMessage> =
            (0..10).map(|i| msg("assistant", &format!("hit number {i}"), vec![])).collect();
        let idx = search_index(vec![("s1", many)]);
        let cands = vec![idx.get_meta("s1").unwrap()];

        let mut limited = opts("hit number");
        limited.limit = 3;
        let out = idx.search_sessions(&cands, &limited);
        assert_eq!(out.hits.len(), 3);
        assert!(out.truncated, "stopping on budget must be reported");
        assert!(!out.complete);
        assert_eq!(out.incomplete_reasons, vec!["hit_limit"]);

        let absent = idx.search_sessions(&cands, &opts("never appears anywhere"));
        assert!(absent.hits.is_empty());
        assert!(!absent.truncated, "genuinely-no-matches must NOT look truncated");
        assert!(absent.complete);
    }

    #[test]
    fn search_only_opens_the_sessions_it_is_handed() {
        // What makes the since/until window cheap: excluded candidates are
        // never parsed, because the caller filters on indexed metadata first.
        let idx = search_index(vec![
            ("keep", vec![msg("assistant", "findme", vec![])]),
            ("skip", vec![msg("assistant", "findme", vec![])]),
        ]);
        let cands = vec![idx.get_meta("keep").unwrap()];
        let out = idx.search_sessions(&cands, &opts("findme"));
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.sessions_scanned, 1);
        assert_eq!(out.total_sessions, 1);
        assert_eq!(out.hits[0].session_id, "keep");
    }

    #[test]
    fn role_filter_narrows_to_one_side_of_the_conversation() {
        let idx = search_index(vec![(
            "s1",
            vec![msg("user", "deploy it", vec![]), msg("assistant", "deploy it? confirming first", vec![])],
        )]);
        let cands = vec![idx.get_meta("s1").unwrap()];
        let mut o = opts("deploy it");
        o.role = Some("user".into());
        let out = idx.search_sessions(&cands, &o);
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.hits[0].role, "user");
    }

    #[test]
    fn a_snippet_is_bounded_even_for_a_pathologically_large_message() {
        // This tool must never be the reason a caller blows its own context.
        let huge = format!("{}NEEDLE{}", "x".repeat(50_000), "y".repeat(50_000));
        let idx = search_index(vec![("s1", vec![msg("assistant", &huge, vec![])])]);
        let cands = vec![idx.get_meta("s1").unwrap()];
        let out = idx.search_sessions(&cands, &opts("needle"));
        assert_eq!(out.hits.len(), 1);
        assert!(out.hits[0].snippet.chars().count() <= SNIPPET_MAX_CHARS + 2, "snippet stays bounded");
        assert!(out.hits[0].snippet.contains("NEEDLE"), "and still shows the match");
    }

    // ── Snippet/offset correctness (reagentx P1s on PR #3321) ──
    //
    // All three of these fail against the original implementation. They are
    // about the same underlying mistake in three places: an offset computed
    // against one string and then used to slice a different one.

    #[test]
    fn a_tool_hit_snippet_contains_the_match_even_far_into_a_long_argument() {
        // The snippet used to be cut from offset 0 whenever a `tool` filter
        // was set, so a match past SNIPPET_MAX_CHARS was reported as a hit
        // whose evidence did not contain it. A hit whose snippet omits the
        // match is worse than no hit — it looks like verification.
        let args = format!("{}NEEDLE_TARGET{}", "p".repeat(3_000), "q".repeat(100));
        let idx = search_index(vec![("s1", vec![msg("assistant", "sent", vec![("SendMessage", &args)])])]);
        let cands = vec![idx.get_meta("s1").unwrap()];

        let mut o = opts("needle_target");
        o.tool = Some("SendMessage".into());
        let out = idx.search_sessions(&cands, &o);
        assert_eq!(out.hits.len(), 1);
        assert!(
            out.hits[0].snippet.contains("NEEDLE_TARGET"),
            "snippet must show the match, not the start of the arguments: {}",
            out.hits[0].snippet
        );
    }

    #[test]
    fn an_untooled_tool_hit_offsets_against_the_arguments_not_name_plus_arguments() {
        // Previously the offset came from `format!("{name} {args}")` but was
        // used to slice `args` alone, so the window was shifted by
        // name.len()+1 — and for a long name effectively clamped to the end.
        let args = format!("{}FIND_ME{}", "z".repeat(1_000), "w".repeat(1_000));
        let idx = search_index(vec![(
            "s1",
            vec![msg("assistant", "no prose match here", vec![("SomeVeryLongToolNameIndeed", &args)])],
        )]);
        let cands = vec![idx.get_meta("s1").unwrap()];

        let out = idx.search_sessions(&cands, &opts("find_me"));
        assert_eq!(out.hits.len(), 1);
        assert_eq!(out.hits[0].tool_name.as_deref(), Some("SomeVeryLongToolNameIndeed"));
        assert!(
            out.hits[0].snippet.contains("FIND_ME"),
            "snippet window must be centred on the real match: {}",
            out.hits[0].snippet
        );
    }

    #[test]
    fn matching_is_safe_when_lowercasing_changes_byte_length() {
        // Lowercasing is not length-preserving: 'İ' (U+0130, 2 bytes) becomes
        // 'i̇' (3 bytes). Finding in a lowercased copy and slicing the
        // original therefore drifts by one byte per such character, which can
        // slice mid-character and panic. This must neither panic nor
        // mis-window.
        let content = format!("{}NEEDLE tail", "İ".repeat(200));
        let idx = search_index(vec![("s1", vec![msg("assistant", &content, vec![])])]);
        let cands = vec![idx.get_meta("s1").unwrap()];

        let out = idx.search_sessions(&cands, &opts("needle"));
        assert_eq!(out.hits.len(), 1);
        assert!(
            out.hits[0].snippet.contains("NEEDLE"),
            "snippet must still contain the match: {}",
            out.hits[0].snippet
        );
    }

    #[test]
    fn case_insensitive_find_reports_offsets_into_the_original_string() {
        // Direct unit coverage of the helper the three fixes above rely on.
        assert_eq!(find_case_insensitive("Hello World", "world"), Some(6));
        assert_eq!(find_case_insensitive("abc", "zzz"), None);
        assert_eq!(find_case_insensitive("abc", ""), None);

        // 'İ' is 2 bytes and lowercases to 3, so a naive lowercased-offset
        // would report 3 here instead of the correct 2.
        let s = "İX";
        assert_eq!(find_case_insensitive(s, "x"), Some(2));
        assert!(s.is_char_boundary(find_case_insensitive(s, "x").unwrap()));
    }
}

#[cfg(test)]
mod refresh_tests {
    //! Incremental refresh (SPEC_DURABLE_CONVERSATION_MEMORY_2026_09_23.md
    //! §4.5): a cold refresh parses every transcript on the machine, so a
    //! refresh must only re-parse what changed.
    use super::*;
    use std::sync::Arc;

    /// Files whose mtimes a test can change between refreshes, counting every
    /// `extract_meta` call per path.
    struct ChangingAdapter {
        files: Mutex<Vec<DiscoveredFile>>,
        extracted: Mutex<Vec<String>>,
    }

    impl ChangingAdapter {
        fn new(files: &[(&str, i64)]) -> Arc<Self> {
            Arc::new(Self {
                files: Mutex::new(
                    files
                        .iter()
                        .map(|(p, m)| DiscoveredFile { file_path: p.to_string(), mtime_ms: *m })
                        .collect(),
                ),
                extracted: Mutex::new(Vec::new()),
            })
        }
        fn set(&self, files: &[(&str, i64)]) {
            *self.files.lock().unwrap() = files
                .iter()
                .map(|(p, m)| DiscoveredFile { file_path: p.to_string(), mtime_ms: *m })
                .collect();
        }
        fn take_extracted(&self) -> Vec<String> {
            let mut v = std::mem::take(&mut *self.extracted.lock().unwrap());
            v.sort();
            v
        }
    }

    impl HistoryAdapter for Arc<ChangingAdapter> {
        fn provider(&self) -> &str {
            "mock"
        }
        fn discover_files(&self) -> Result<Vec<DiscoveredFile>, HistoryError> {
            Ok(self
                .files
                .lock()
                .unwrap()
                .iter()
                .map(|f| DiscoveredFile { file_path: f.file_path.clone(), mtime_ms: f.mtime_ms })
                .collect())
        }
        fn extract_meta(&self, file_path: &str) -> Result<Option<SessionMeta>, HistoryError> {
            self.extracted.lock().unwrap().push(file_path.to_string());
            let id = PathBuf::from(file_path)
                .file_stem()
                .map(|s| s.to_string_lossy().into_owned())
                .unwrap_or_default();
            Ok(Some(SessionMeta {
                session_id: id,
                file_path: file_path.to_string(),
                provider: "mock".into(),
                model: String::new(),
                slug: String::new(),
                working_directory: "/proj".into(),
                created_at: 0,
                modified_at: 0,
                message_count: 0,
                first_user_message: String::new(),
                file_size_bytes: 0,
                git_branch: String::new(),
                total_tokens: 0,
                subagent_count: 0,
                identity_id: String::new(),
                starts_undated: false,
            }))
        }
        fn parse_file(&self, _: &str) -> Result<Option<HistorySession>, HistoryError> {
            Ok(None)
        }
    }

    fn index_over(adapter: &Arc<ChangingAdapter>) -> SessionIndex {
        SessionIndex::with_isolated_roots(vec![Box::new(adapter.clone())], vec![])
    }

    /// Across a restart: a new index over the same saved cache file parses
    /// only what changed since the last launch saved it.
    #[test]
    fn a_new_index_with_the_saved_cache_reparses_only_what_changed() {
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("history").join("session-meta-cache.json");
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1), ("/h/b.jsonl", 1)]);
        let first = index_over(&a).with_cache_file(cache.clone());
        first.refresh();
        assert_eq!(a.take_extracted(), vec!["/h/a.jsonl", "/h/b.jsonl"]);
        assert!(cache.exists(), "the first build saves the cache");

        // "Restart": a fresh index, b changed while AgentMux was closed.
        a.set(&[("/h/a.jsonl", 1), ("/h/b.jsonl", 2), ("/h/c.jsonl", 1)]);
        let second = index_over(&a).with_cache_file(cache.clone());
        second.refresh();
        assert_eq!(a.take_extracted(), vec!["/h/b.jsonl", "/h/c.jsonl"], "a came from the saved cache");
        assert!(second.get_meta("a").is_some() && second.get_meta("c").is_some());
    }

    /// An unwritable cache location is tried once per interval, not on every
    /// refresh, and leaves no temporary file behind.
    #[test]
    fn an_unwritable_cache_waits_out_the_interval_and_leaves_no_temp_file() {
        let dir = tempfile::tempdir().unwrap();
        let blocker = dir.path().join("not-a-dir");
        std::fs::write(&blocker, b"a file where the cache dir would be").unwrap();
        let cache = blocker.join("session-meta-cache.json");
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1)]);
        let idx = index_over(&a).with_cache_file(cache.clone());
        idx.refresh();
        let first_try = idx.cache_written_at_ms.load(std::sync::atomic::Ordering::Acquire);
        assert!(first_try > 0, "the failed attempt is recorded");
        a.set(&[("/h/a.jsonl", 2)]);
        idx.refresh();
        assert_eq!(idx.cache_written_at_ms.load(std::sync::atomic::Ordering::Acquire), first_try, "not retried within the interval");
        let leftovers: Vec<_> = std::fs::read_dir(dir.path()).unwrap().filter_map(|e| e.ok()).map(|e| e.file_name()).collect();
        assert_eq!(leftovers.len(), 1, "only the blocker file: {leftovers:?}");
    }

    #[test]
    fn a_saved_cache_of_another_version_is_rebuilt_not_trusted() {
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("session-meta-cache.json");
        std::fs::write(&cache, br#"{"version":999,"entries":{"/h/a.jsonl":[1,null]}}"#).unwrap();
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1)]);
        let idx = index_over(&a).with_cache_file(cache);
        idx.refresh();
        assert_eq!(a.take_extracted(), vec!["/h/a.jsonl"]);
        assert!(idx.get_meta("a").is_some(), "the stale entry (no meta) wasn't used");
    }

    #[test]
    fn an_unreadable_saved_cache_is_ignored() {
        let dir = tempfile::tempdir().unwrap();
        let cache = dir.path().join("session-meta-cache.json");
        std::fs::write(&cache, b"not json").unwrap();
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1)]);
        let idx = index_over(&a).with_cache_file(cache.clone());
        idx.refresh();
        assert_eq!(a.take_extracted(), vec!["/h/a.jsonl"]);
        let saved: serde_json::Value = serde_json::from_slice(&std::fs::read(&cache).unwrap()).unwrap();
        assert_eq!(saved["version"], serde_json::json!(META_CACHE_VERSION), "rewritten after the build");
    }

    #[test]
    fn a_second_refresh_reparses_only_what_changed() {
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1), ("/h/b.jsonl", 1)]);
        let idx = index_over(&a);
        idx.refresh();
        assert_eq!(a.take_extracted(), vec!["/h/a.jsonl", "/h/b.jsonl"]);

        idx.refresh();
        assert!(a.take_extracted().is_empty(), "unchanged files must not be re-parsed");

        a.set(&[("/h/a.jsonl", 2), ("/h/b.jsonl", 1), ("/h/c.jsonl", 1)]);
        idx.refresh();
        assert_eq!(a.take_extracted(), vec!["/h/a.jsonl", "/h/c.jsonl"]);
        assert!(idx.get_meta("c").is_some());
    }

    #[test]
    fn a_file_gone_from_disk_leaves_the_index() {
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1), ("/h/b.jsonl", 1)]);
        let idx = index_over(&a);
        idx.refresh();
        a.set(&[("/h/a.jsonl", 1)]);
        idx.refresh();
        assert!(idx.get_meta("b").is_none());
        assert!(idx.get_meta("a").is_some());
        assert!(a.take_extracted().len() == 2, "only the first refresh parsed");
    }

    #[test]
    fn refreshed_at_is_zero_until_the_first_refresh_completes() {
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1)]);
        let idx = index_over(&a);
        assert_eq!(idx.refreshed_at_ms(), 0);
        idx.refresh();
        assert!(idx.refreshed_at_ms() > 0);
    }

    /// A search must never wait out a cold build behind the MCP's 10 s
    /// timeout: while another refresh runs, `try_refresh` returns at once.
    #[test]
    fn try_refresh_does_not_wait_for_a_refresh_in_progress() {
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1)]);
        let idx = index_over(&a);
        let _running = idx.refresh_lock.lock().unwrap();
        assert!(idx.try_refresh().is_none());
        assert!(a.take_extracted().is_empty());
    }

    #[test]
    fn try_refresh_refreshes_when_nothing_else_is() {
        let a = ChangingAdapter::new(&[("/h/a.jsonl", 1)]);
        let idx = index_over(&a);
        assert!(idx.try_refresh().is_some());
        assert!(idx.get_meta("a").is_some());
    }

    /// Discovery that walks slowly and counts its walks.
    struct SlowAdapter {
        walks: std::sync::atomic::AtomicUsize,
    }

    impl HistoryAdapter for Arc<SlowAdapter> {
        fn provider(&self) -> &str {
            "mock"
        }
        fn discover_files(&self) -> Result<Vec<DiscoveredFile>, HistoryError> {
            self.walks.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            std::thread::sleep(std::time::Duration::from_millis(150));
            Ok(Vec::new())
        }
        fn extract_meta(&self, _: &str) -> Result<Option<SessionMeta>, HistoryError> {
            Ok(None)
        }
        fn parse_file(&self, _: &str) -> Result<Option<HistorySession>, HistoryError> {
            Ok(None)
        }
    }

    /// Searches arriving together share refreshes: a refresh that started
    /// after a request began already lists everything that existed when it
    /// began. Every search running its own full walk, one after another,
    /// put the fourth past the MCP's 10 s timeout on a heavy host (Codex P1
    /// on #3693).
    #[test]
    fn concurrent_searches_share_refreshes() {
        let a = Arc::new(SlowAdapter { walks: Default::default() });
        let idx = Arc::new(SessionIndex::with_isolated_roots(vec![Box::new(a.clone())], vec![]));
        let barrier = Arc::new(std::sync::Barrier::new(8));
        let threads: Vec<_> = (0..8)
            .map(|_| {
                let (idx, barrier) = (idx.clone(), barrier.clone());
                std::thread::spawn(move || {
                    barrier.wait();
                    idx.refresh_covering_now();
                })
            })
            .collect();
        for t in threads {
            t.join().unwrap();
        }
        let walks = a.walks.load(std::sync::atomic::Ordering::SeqCst);
        assert!((1..=2).contains(&walks), "8 simultaneous searches walked {walks} times");
    }

    /// …but a refresh that finished before the request began may have
    /// missed a session written since, so it's never reused.
    #[test]
    fn a_refresh_finished_before_the_request_is_not_reused() {
        let a = Arc::new(SlowAdapter { walks: Default::default() });
        let idx = SessionIndex::with_isolated_roots(vec![Box::new(a.clone())], vec![]);
        idx.refresh();
        idx.refresh_covering_now();
        assert_eq!(a.walks.load(std::sync::atomic::Ordering::SeqCst), 2);
    }

    /// Discovery that reports what it couldn't look at.
    struct ProblemAdapter {
        problems: Mutex<Vec<String>>,
    }

    impl HistoryAdapter for Arc<ProblemAdapter> {
        fn provider(&self) -> &str {
            "mock"
        }
        fn discover_files(&self) -> Result<Vec<DiscoveredFile>, HistoryError> {
            Ok(vec![DiscoveredFile { file_path: "/h/locked.jsonl".into(), mtime_ms: 1 }])
        }
        fn discover(&self) -> Result<Discovery, HistoryError> {
            Ok(Discovery { files: self.discover_files()?, problems: self.problems.lock().unwrap().clone() })
        }
        fn extract_meta(&self, _: &str) -> Result<Option<SessionMeta>, HistoryError> {
            Err(HistoryError::Other("sharing violation".into()))
        }
        fn parse_file(&self, _: &str) -> Result<Option<HistorySession>, HistoryError> {
            Ok(None)
        }
    }

    /// A directory the walk couldn't read, or a transcript it couldn't index,
    /// may hold the very session being searched for. Both are kept with the
    /// snapshot so a search over it can't claim to be complete (Codex P1 on
    /// #3693).
    #[test]
    fn what_discovery_could_not_read_is_kept_with_the_snapshot() {
        let a = Arc::new(ProblemAdapter { problems: Mutex::new(vec!["/h/private: Access is denied".into()]) });
        let idx = SessionIndex::with_isolated_roots(vec![Box::new(a.clone())], vec![]);
        idx.refresh();
        let problems = idx.discovery_problems();
        assert!(problems.iter().any(|p| p.contains("/h/private")), "{problems:?}");
        assert!(
            problems.iter().any(|p| p.contains("locked.jsonl") && p.contains("sharing violation")),
            "a transcript that couldn't be indexed is a problem too: {problems:?}"
        );

        a.problems.lock().unwrap().clear();
        idx.refresh();
        assert_eq!(idx.discovery_problems().len(), 1, "each refresh replaces the last one's problems");
    }

    /// The first build never waits on, or repeats, another caller's
    /// (Codex P2 on #3930).
    #[test]
    fn try_first_build_skips_when_busy_or_already_built() {
        let idx = SessionIndex::with_isolated_roots(vec![], vec![]);
        {
            let _running = idx.refresh_lock.lock().unwrap();
            assert_eq!(idx.try_first_build(), FirstBuild::Busy);
        }
        assert_eq!(idx.try_first_build(), FirstBuild::Built { discovered: 0 });
        let built_at = idx.refreshed_at_ms();
        assert!(built_at > 0);
        assert_eq!(idx.try_first_build(), FirstBuild::AlreadyBuilt);
        assert_eq!(idx.refreshed_at_ms(), built_at);
    }
}