ai-session-search 1.0.0-rc.4

// SPDX-FileCopyrightText: 2026 Andrew Hundt
// SPDX-License-Identifier: Apache-2.0

//! Typed application services shared by CLI, MCP, and language bindings.
//!
//! Services own operation boundaries, while [`Db`] remains the storage layer.
//! Adapters must not duplicate SQL, filtering, pagination, or lifecycle policy.

use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::fs;
use std::num::NonZeroUsize;
use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use anyhow::{anyhow, bail, Context, Result};

use crate::config::{Config, IndexRefresh, ScoringConfig};
use crate::db::{
    Db, MessageBatchControl, QueryCancellation, SchemaState, MIN_READABLE_SCHEMA_VERSION,
};
use crate::indexer::{self, AutoReindexOutcome, IndexCoordinator};
use crate::message_search::{
    apply_message_presentation_cancellable, attach_match_evidence_cancellable, ContextWindow,
    DetailLevel, ExecutionOrder, FieldViewBudget, LineWindow, MatchViewBudget, MatchWindow,
    MessageResponsePlan, MessageRetrievalPlan, MessageSearchError, MessageSearchInclude,
    MessageSearchIncludedData, MessageSearchOrderedDigest, MessageSearchOrigins, MessageSearchPlan,
    MessageSearchRequest, MessageSearchResponse, MessageSearchRuntimeDiagnostics,
    MessageSearchSpecification, MessageTarget, PageInfo, ReceiptLevel, ResolvedExtent,
    ResolvedMessagePredicates, ResolvedMessagePresentation, ResolvedMessageSearchRequest,
    SearchSurface, ValueOrigin, DEFAULT_MATCH_EVIDENCE_MAX_CHARS,
};
use crate::message_search_batches::{
    ensure_message_search_active, MessageSearchBatch, MessageSearchBatches,
};
use crate::models::{
    DiagnosticStatus, FileCrossRef, FileEditSummary, FileQuery, FileVersion, IndexStatus,
    MessageFilters, MessageHit, SearchExplain, SearchFilters, SearchHit, SessionMeta,
    SessionRecord,
};
use crate::runtime::ExecutionRuntime;
use crate::search_scope::{EffectiveAccessScope, TrustedAccessInputs};

/// RAII application root shared by native frontends and language bindings.
///
/// Opening an instance applies the configured SQLite contention and performance
/// policy exactly once. Dropping it closes the owned database connection.
pub struct SessionSearch {
    config: Config,
    access: EffectiveAccessScope,
    db: Db,
}

#[cfg(test)]
mod analysis_service_tests {
    use super::*;
    use crate::models::{
        Message, MessageAuthorship, MessageKind, MessageProvenance, MessageRecordRelation,
        Provider, Role,
    };
    use crate::util::minimal_record;

    fn human_original_provenance() -> MessageProvenance {
        MessageProvenance {
            authorship: MessageAuthorship::Human,
            record_relation: MessageRecordRelation::Original,
            ..MessageProvenance::default()
        }
    }

    /// Index one user message per string and return the opened app, so a selection test can state
    /// its corpus in one line instead of rebuilding `ParsedSession` per case.
    fn app_with_user_messages(
        dir: &std::path::Path,
        config: Config,
        texts: &[&str],
    ) -> SessionSearch {
        let mut config = config;
        config.index.db_path = Some(dir.join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(dir.join("cache").to_string_lossy().into_owned());
        let app = SessionSearch::open(config).unwrap();
        let mut parsed = minimal_record(
            Provider::Claude,
            std::path::Path::new("/fixture/selection.jsonl"),
            String::new(),
        );
        parsed.session.id = "claude:selection".into();
        parsed.session.provider_session_id = "selection".into();
        parsed.messages = texts
            .iter()
            .enumerate()
            .map(|(seq, text)| Message {
                seq: seq as i64,
                role: Role::User,
                ts: None,
                tool_name: None,
                kind: MessageKind::Conversation,
                tool_call_id: None,
                is_compaction: false,
                content: (*text).to_string(),
                provenance: human_original_provenance(),
            })
            .collect();
        app.database().upsert_session(&parsed, 0, 0).unwrap();
        app
    }

    /// Write a minimal standard-shaped skill package with one classification category.
    fn write_skill(search_root: &std::path::Path, name: &str, category: &str, pattern: &str) {
        let root = search_root.join(name);
        std::fs::create_dir_all(&root).unwrap();
        std::fs::write(
            root.join("SKILL.md"),
            format!(
                "---\nname: {name}\ndescription: test skill\nmetadata:\n  version: 2.1.0\n---\n\nbody\n"
            ),
        )
        .unwrap();
        std::fs::write(
            root.join("aise-capability.toml"),
            format!(
                "schema_version = 1\nkind = \"message-classification\"\n\n\
                 [[categories]]\nname = \"{category}\"\npatterns = [\'\'\'{pattern}\'\'\']\n"
            ),
        )
        .unwrap();
    }

    fn classify_messages(
        analysis: &AnalysisService<'_>,
        skill: crate::skill_catalog::SkillSelector,
        filters: MessageFilters,
    ) -> crate::corrections::MessageClassificationReport {
        let report = analysis
            .run_skill(&crate::skill_run::SkillRunQuery {
                skill,
                definition: None,
                input: crate::skill_run::SkillCapabilityInput::MessageClassification(
                    crate::skill_run::MessageClassificationQuery {
                        filters,
                        additional_skills: Vec::new(),
                    },
                ),
            })
            .unwrap();
        let crate::skill_run::SkillCapabilityOutput::MessageClassification(output) = report.output;
        output.report
    }

    #[test]
    fn typed_skill_run_resolves_one_descriptor_and_returns_tagged_provenance() {
        let dir = tempfile::tempdir().unwrap();
        let skills_root = dir.path().join("skills");
        let skill_root = skills_root.join("my-review");
        std::fs::create_dir_all(&skill_root).unwrap();
        std::fs::write(
            skill_root.join("SKILL.md"),
            "---\nname: my-review\ndescription: test classification\nmetadata:\n  version: \
             2.1.0\n---\n\ninstructions\n",
        )
        .unwrap();
        std::fs::write(
            skill_root.join("aise-capability.toml"),
            "schema_version = 1\nkind = \"message-classification\"\n\n\
             [[categories]]\nname = \"clobber\"\npatterns = ['''\\byou overwrote\\b''']\n",
        )
        .unwrap();

        let mut config = Config::default();
        config.skills.search_paths = vec![skills_root.to_string_lossy().into_owned()];
        let app = app_with_user_messages(dir.path(), config, &["you overwrote the notes"]);
        let report = app
            .analysis()
            .run_skill(&crate::skill_run::SkillRunQuery {
                skill: crate::skill_catalog::SkillSelector::Name(
                    crate::skill_catalog::SkillNameSelector {
                        name: crate::skill_catalog::SkillName::try_from("my-review".to_string())
                            .unwrap(),
                    },
                ),
                definition: None,
                input: crate::skill_run::SkillCapabilityInput::MessageClassification(
                    crate::skill_run::MessageClassificationQuery {
                        filters: MessageFilters::default(),
                        additional_skills: Vec::new(),
                    },
                ),
            })
            .unwrap();

        let value = serde_json::to_value(&report).unwrap();
        assert_eq!(value["requested_selector"]["name"], "my-review");
        assert_eq!(value["resolved_skill"]["name"], "my-review");
        assert_eq!(value["output"]["capability"], "message-classification");
        assert_eq!(
            value["output"]["result"]["report"]["matches"][0]["category"],
            "clobber"
        );
        assert_eq!(
            value["output"]["result"]["receipt"]["sha256"]
                .as_str()
                .unwrap()
                .len(),
            64
        );
        assert_eq!(
            value["output"]["result"]["receipt"],
            value["output"]["result"]["report"]["policies"][0],
            "the primary envelope receipt is derived from the first evaluated policy"
        );

        let duplicate = app
            .analysis()
            .run_skill(&crate::skill_run::SkillRunQuery {
                skill: crate::skill_catalog::SkillSelector::Name(
                    crate::skill_catalog::SkillNameSelector {
                        name: crate::skill_catalog::SkillName::try_from("my-review".to_string())
                            .unwrap(),
                    },
                ),
                definition: None,
                input: crate::skill_run::SkillCapabilityInput::MessageClassification(
                    crate::skill_run::MessageClassificationQuery {
                        filters: MessageFilters::default(),
                        additional_skills: vec![crate::skill_catalog::SkillSelector::Path(
                            crate::skill_catalog::SkillPathSelector { path: skill_root },
                        )],
                    },
                ),
            })
            .expect_err("the same canonical skill selected by name and path must fail");
        assert!(
            duplicate.to_string().contains("selected more than once"),
            "{duplicate:#}"
        );
    }

    /// `--skill NAME` must REPLACE the built-in rules, not merge with them, and the report must
    /// say which policy produced each match.
    ///
    /// The two fixture messages are chosen so each policy matches exactly one and misses the
    /// other: `you overwrote` hits no built-in category, and `you forgot` hits no category the
    /// external skill defines. A merge, a silent fallback, or an ignored selection each produce a
    /// different count here, so this fails loudly rather than looking plausible.
    #[test]
    fn a_named_skill_replaces_the_built_in_rules_and_names_itself_in_the_report() {
        let dir = tempfile::tempdir().unwrap();
        let skills_root = dir.path().join("skills");
        write_skill(
            &skills_root,
            "my-corrections",
            "clobber",
            r"\byou overwrote\b",
        );

        let mut config = Config::default();
        config.skills.search_paths = vec![skills_root.to_string_lossy().into_owned()];
        let app = app_with_user_messages(
            dir.path(),
            config,
            &["you overwrote my notes", "you forgot the tests"],
        );
        let analysis = app.analysis();

        let selected = classify_messages(
            &analysis,
            crate::skill_catalog::SkillSelector::name("my-corrections").unwrap(),
            MessageFilters::default(),
        );
        assert_eq!(
            selected
                .matches
                .iter()
                .map(|hit| (
                    hit.policy_name.as_str(),
                    hit.category.as_str(),
                    hit.matched_text.as_str()
                ))
                .collect::<Vec<_>>(),
            vec![("my-corrections", "clobber", "you overwrote")],
            "only the selected skill's categories may apply"
        );
        assert_eq!(
            selected
                .policies
                .iter()
                .map(|receipt| (receipt.name.as_str(), receipt.version.as_str()))
                .collect::<Vec<_>>(),
            vec![("my-corrections", "2.1.0")],
            "the receipt reports the policy's own version, not the aise version"
        );
        assert_eq!(
            selected.policies[0].sha256.len(),
            64,
            "a receipt without a digest cannot reproduce a run"
        );

        let defaulted = classify_messages(
            &analysis,
            crate::skill_catalog::SkillSelector::name("corrections").unwrap(),
            MessageFilters::default(),
        );
        assert_eq!(
            defaulted
                .matches
                .iter()
                .map(|hit| (hit.policy_name.as_str(), hit.category.as_str()))
                .collect::<Vec<_>>(),
            vec![(crate::corrections::EMBEDDED_POLICY_NAME, "skip_step")],
            "omitting --skill must evaluate the embedded policy, and ONLY it"
        );
    }

    /// A direct definition replaces only the selected package's executable rules. The package
    /// continues to own identity and authorization, while the canonical in-memory digest makes
    /// repeated calls with the same ordered definition reproducible.
    #[test]
    fn a_direct_definition_replaces_file_rules_without_replacing_skill_identity() {
        let dir = tempfile::tempdir().unwrap();
        let skills_root = dir.path().join("skills");
        write_skill(
            &skills_root,
            "my-corrections",
            "file-rule",
            r"\byou overwrote\b",
        );

        let mut config = Config::default();
        config.skills.search_paths = vec![skills_root.to_string_lossy().into_owned()];
        let app = app_with_user_messages(
            dir.path(),
            config,
            &["you overwrote my notes", "that answer is incorrect"],
        );
        let query = crate::skill_run::SkillRunQuery {
            skill: crate::skill_catalog::SkillSelector::name("my-corrections").unwrap(),
            definition: Some(crate::skill_run::MessageClassificationDefinition {
                categories: vec![crate::corrections::CorrectionCategorySpec {
                    name: "direct-rule".to_string(),
                    patterns: vec![r"\bincorrect\b".to_string()],
                }],
            }),
            input: crate::skill_run::SkillCapabilityInput::MessageClassification(
                crate::skill_run::MessageClassificationQuery::default(),
            ),
        };

        let first = app.analysis().run_skill(&query).unwrap();
        let second = app.analysis().run_skill(&query).unwrap();
        assert_eq!(first.resolved_skill.name.as_str(), "my-corrections");
        assert_eq!(
            first.resolved_skill.package_version.as_deref(),
            Some("2.1.0")
        );
        assert!(matches!(
            first.resolved_skill.selected_location,
            crate::skill_run::SelectedSkillLocation::Path { .. }
        ));
        assert!(matches!(
            first.resolved_skill.execution_source,
            crate::skill_run::CapabilityExecutionSource::Inline
        ));
        let crate::skill_run::SkillCapabilityOutput::MessageClassification(first_output) =
            first.output;
        let crate::skill_run::SkillCapabilityOutput::MessageClassification(second_output) =
            second.output;
        assert_eq!(
            first_output
                .report
                .matches
                .iter()
                .map(|matched| (
                    matched.policy_name.as_str(),
                    matched.category.as_str(),
                    matched.matched_text.as_str()
                ))
                .collect::<Vec<_>>(),
            vec![("my-corrections", "direct-rule", "incorrect")],
            "the direct definition must replace, rather than merge with, the file rules"
        );
        assert_eq!(first_output.receipt, first_output.report.policies[0]);
        assert_eq!(
            first_output.receipt.sha256, second_output.receipt.sha256,
            "the same ordered definition must produce the same canonical digest"
        );

        let invalid = crate::skill_run::SkillRunQuery {
            definition: Some(crate::skill_run::MessageClassificationDefinition {
                categories: Vec::new(),
            }),
            ..query
        };
        let error = app
            .analysis()
            .run_skill(&invalid)
            .expect_err("an empty direct definition must use the shared policy validator");
        assert!(
            error.to_string().contains("defines no categories"),
            "the validation error must say how the definition is invalid: {error:#}"
        );
    }

    #[test]
    fn a_direct_definition_and_additional_package_share_one_byte_budget() {
        let dir = tempfile::tempdir().unwrap();
        let skills_root = dir.path().join("skills");
        write_skill(&skills_root, "primary", "unused", r"\bunused\b");
        write_skill(&skills_root, "additional", "file-rule", r"\bwrong\b");
        let additional_capability = skills_root.join("additional/aise-capability.toml");
        let mut file = std::fs::read_to_string(&additional_capability).unwrap();
        file.push_str(&format!("\n# {}\n", "x".repeat(500 * 1024)));
        std::fs::write(&additional_capability, file).unwrap();

        let mut config = Config::default();
        config.skills.search_paths = vec![skills_root.to_string_lossy().into_owned()];
        let app = app_with_user_messages(dir.path(), config, &["wrong"]);
        let error = app
            .analysis()
            .run_skill(&crate::skill_run::SkillRunQuery {
                skill: crate::skill_catalog::SkillSelector::name("primary").unwrap(),
                definition: Some(crate::skill_run::MessageClassificationDefinition {
                    categories: vec![crate::corrections::CorrectionCategorySpec {
                        name: "inline".to_string(),
                        patterns: vec![format!("(?x){}needle", " ".repeat(600 * 1024))],
                    }],
                }),
                input: crate::skill_run::SkillCapabilityInput::MessageClassification(
                    crate::skill_run::MessageClassificationQuery {
                        filters: MessageFilters::default(),
                        additional_skills: vec![crate::skill_catalog::SkillSelector::name(
                            "additional",
                        )
                        .unwrap()],
                    },
                ),
            })
            .expect_err("direct and additional package definitions must share one budget");
        let message = error.to_string();
        assert!(
            message.contains("aggregate")
                && message.contains("additional")
                && message.contains("aise-capability.toml")
                && message.contains("bytes were consumed by earlier selections"),
            "{message}"
        );
    }

    /// An unknown `--skill` name must fail, naming the value and where to look up valid ones.
    ///
    /// The dangerous alternative is not a bad message: it is returning `Ok` with default-policy
    /// results, which looks like a successful run against rules the caller never selected.
    #[test]
    fn an_unknown_skill_name_fails_instead_of_silently_using_the_defaults() {
        let dir = tempfile::tempdir().unwrap();
        let app = app_with_user_messages(dir.path(), Config::default(), &["you forgot the tests"]);
        let error = app
            .analysis()
            .run_skill(&crate::skill_run::SkillRunQuery {
                skill: crate::skill_catalog::SkillSelector::name("not-installed").unwrap(),
                definition: None,
                input: crate::skill_run::SkillCapabilityInput::MessageClassification(
                    crate::skill_run::MessageClassificationQuery::default(),
                ),
            })
            .expect_err("an unknown skill must not resolve to the defaults");
        let message = format!("{error:#}");
        assert!(
            message.contains("not-installed") && message.contains("aise skills list"),
            "the error must name the value and how to find valid ones: {message}"
        );
    }

    #[test]
    fn analysis_service_reuses_indexed_correction_planning_and_role_queries() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().into_owned());
        let app = SessionSearch::open(config).unwrap();
        let mut parsed = minimal_record(
            Provider::Claude,
            std::path::Path::new("/fixture/session.jsonl"),
            String::new(),
        );
        parsed.session.id = "claude:analysis".into();
        parsed.session.provider_session_id = "analysis".into();
        parsed.messages = vec![
            Message {
                seq: 0,
                role: Role::User,
                ts: None,
                tool_name: None,
                kind: MessageKind::Conversation,
                tool_call_id: None,
                is_compaction: false,
                content: "actually, that is wrong".into(),
                provenance: human_original_provenance(),
            },
            Message {
                seq: 1,
                role: Role::Slash,
                ts: None,
                tool_name: None,
                kind: MessageKind::Conversation,
                tool_call_id: None,
                is_compaction: false,
                content: "/plan verify migration".into(),
                provenance: Default::default(),
            },
        ];
        app.database().upsert_session(&parsed, 0, 0).unwrap();
        let mut other = minimal_record(
            Provider::Codex,
            std::path::Path::new("/fixture/codex.jsonl"),
            String::new(),
        );
        other.session.id = "codex:analysis".into();
        other.session.provider_session_id = "analysis".into();
        other.messages = vec![Message {
            seq: 0,
            role: Role::User,
            ts: None,
            tool_name: None,
            kind: MessageKind::Conversation,
            tool_call_id: None,
            is_compaction: false,
            content: "unrelated provider message".into(),
            provenance: human_original_provenance(),
        }];
        app.database().upsert_session(&other, 0, 0).unwrap();

        let analysis = app.analysis();
        let filters = MessageFilters::default();
        let corrections = classify_messages(
            &analysis,
            crate::skill_catalog::SkillSelector::name("corrections").unwrap(),
            filters.clone(),
        );
        assert_eq!(corrections.matches.len(), 1);
        // A default run evaluates exactly the embedded policy, and says so in its receipt.
        assert_eq!(
            corrections
                .policies
                .iter()
                .map(|receipt| receipt.name.as_str())
                .collect::<Vec<_>>(),
            vec![crate::corrections::EMBEDDED_POLICY_NAME]
        );
        assert_eq!(
            corrections.matches[0].policy_name,
            crate::corrections::EMBEDDED_POLICY_NAME
        );
        let planning = analysis.planning(&filters, &["^/plan$".into()]).unwrap();
        assert_eq!(planning.len(), 1);
        assert_eq!(planning[0].command, "/plan");
        let roles = analysis.role_statistics(&filters).unwrap();
        assert_eq!(roles.len(), 2);
        assert_eq!(roles.iter().map(|row| row.count).sum::<i64>(), 3);
        let provider_roles = analysis
            .role_statistics(&MessageFilters {
                providers: Some(vec![Provider::Claude]),
                ..Default::default()
            })
            .unwrap();
        assert_eq!(provider_roles.iter().map(|row| row.count).sum::<i64>(), 2);
    }

    #[test]
    fn analysis_run_matches_paged_documents_reference_on_indexed_sessions() {
        use crate::analysis_pipeline::{
            AnalysisPolicySpec, ClassificationRuleSpec, ClassificationTarget, PhraseTextMode,
            PhraseVocabularyPolicySpec,
        };

        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().into_owned());
        let app = SessionSearch::open(config).unwrap();
        // Multi-message sessions with junction-spanning phrases and a code fence that
        // opens in one message and closes in the next.
        for (id, provider, texts) in [
            (
                "claude:multi",
                Provider::Claude,
                vec!["use tdd across chunks", "tdd again across chunks"],
            ),
            (
                "codex:fence",
                Provider::Codex,
                vec!["prose one\n```", "code hidden\n```\nprose two"],
            ),
            ("gemini-cli:empty", Provider::GeminiCli, vec![]),
        ] {
            let mut parsed = minimal_record(
                Provider::Claude,
                std::path::Path::new("/fixture/session.jsonl"),
                String::new(),
            );
            parsed.session.provider = provider;
            parsed.session.id = id.into();
            parsed.session.provider_session_id = id.into();
            parsed.messages = texts
                .iter()
                .enumerate()
                .map(|(seq, text)| Message {
                    seq: seq as i64,
                    role: Role::User,
                    ts: None,
                    tool_name: None,
                    kind: MessageKind::Conversation,
                    tool_call_id: None,
                    is_compaction: false,
                    content: (*text).into(),
                    provenance: human_original_provenance(),
                })
                .collect();
            app.database().upsert_session(&parsed, 0, 0).unwrap();
        }
        let policy = AnalysisPolicySpec {
            classification_rules: vec![ClassificationRuleSpec {
                dimension: "technique".into(),
                label: "tdd".into(),
                target: ClassificationTarget::UserText,
                pattern: "(?i)\\btdd\\b".into(),
                weight: 1,
            }],
            relationship_rules: vec![],
            phrase_vocabulary: Some(PhraseVocabularyPolicySpec {
                widths: vec![2],
                max_unique_phrases: 1000,
                min_document_tokens: 0,
                excluded_tokens: vec![],
                exclude_numeric_tokens: true,
                text_mode: PhraseTextMode::ProseOnly,
            }),
            max_classification_chars: None,
        }
        .compile()
        .unwrap();
        let filters = SearchFilters::default();

        let request = crate::models::AnalysisRequest::new(
            filters.clone(),
            crate::models::AnalysisSessionSelection::AllEligible,
        )
        .unwrap();
        let streamed = app.analysis().run(&request, &policy).unwrap();

        // Reference: the public paged-documents API still materializes joined text; the
        // streaming run must be indistinguishable from analyzing those documents.
        let limited = SearchFilters {
            limit: 2,
            ..filters
        };
        let mut documents = Vec::new();
        let mut cursor = None;
        loop {
            let page = app.analysis().documents(&limited, cursor.as_ref()).unwrap();
            documents.extend(page.documents);
            match page.next_cursor {
                Some(next) => cursor = Some(next),
                None => break,
            }
        }
        let reference = policy.analyze(documents).unwrap();

        assert_eq!(
            serde_json::to_value(&streamed.result).unwrap(),
            serde_json::to_value(&reference).unwrap()
        );
        assert!(!streamed.result.vocabulary.is_empty());
        assert!(streamed.result.sessions["claude:multi"].has_user_text);
        assert!(!streamed.result.sessions["gemini-cli:empty"].has_user_text);
    }

    #[test]
    fn analysis_run_streams_one_snapshot_and_resolves_across_pages() {
        use crate::analysis_pipeline::{
            AnalysisPolicy, ClassificationRuleSpec, ClassificationTarget, RelationshipKind,
            RelationshipResolution, RelationshipRuleSpec,
        };

        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().into_owned());
        let app = SessionSearch::open(config).unwrap();
        for (id, provider, title, text) in [
            ("claude:root", Provider::Claude, "Root", "use TDD"),
            ("codex:root", Provider::Codex, "Root", ""),
            (
                "gemini-cli:child",
                Provider::GeminiCli,
                "Branch of Root",
                "use TDD",
            ),
        ] {
            let mut parsed = minimal_record(
                provider,
                std::path::Path::new("/fixture/session.jsonl"),
                String::new(),
            );
            parsed.session.id = id.into();
            parsed.session.provider_session_id = id.into();
            parsed.session.title = Some(title.into());
            if !text.is_empty() {
                parsed.messages.push(Message {
                    seq: 0,
                    role: Role::User,
                    ts: None,
                    tool_name: None,
                    kind: MessageKind::Conversation,
                    tool_call_id: None,
                    is_compaction: false,
                    content: text.into(),
                    provenance: human_original_provenance(),
                });
            }
            app.database().upsert_session(&parsed, 0, 0).unwrap();
        }
        let policy = AnalysisPolicy::compile(
            vec![ClassificationRuleSpec {
                dimension: "technique".into(),
                label: "tdd".into(),
                target: ClassificationTarget::UserText,
                pattern: "(?i)\\btdd\\b".into(),
                weight: 1,
            }],
            vec![RelationshipRuleSpec {
                id: "branch_of".into(),
                kind: RelationshipKind::Branch,
                pattern: "^Branch of (?P<parent>.+)$".into(),
            }],
        )
        .unwrap();
        let request = crate::models::AnalysisRequest::new(
            SearchFilters::default(),
            crate::models::AnalysisSessionSelection::AllEligible,
        )
        .unwrap();
        let automatic = app.analysis().run(&request, &policy).unwrap();
        for batch_size in [1, 2, 50] {
            let batched = app
                .analysis()
                .run_with_session_batch_size(
                    &request,
                    std::num::NonZeroUsize::new(batch_size).unwrap(),
                    &policy,
                )
                .unwrap();
            assert_eq!(
                serde_json::to_value(&batched).unwrap(),
                serde_json::to_value(&automatic).unwrap(),
                "internal batch size {batch_size} must not alter results, counts, or digests"
            );
        }
        let run = app
            .analysis()
            .run_with_session_batch_size(&request, std::num::NonZeroUsize::new(1).unwrap(), &policy)
            .unwrap();

        assert_eq!(run.result.sessions.len(), 3);
        assert_eq!(run.result.sessions["gemini-cli:child"].score, 1);
        assert_eq!(
            run.result.sessions["gemini-cli:child"].relationship_hints[0].resolution,
            RelationshipResolution::Ambiguous {
                session_ids: vec!["claude:root".into(), "codex:root".into()]
            }
        );

        let limited = crate::models::AnalysisRequest::new(
            SearchFilters::default(),
            crate::models::AnalysisSessionSelection::FirstCanonicalSessions {
                max_sessions: std::num::NonZeroUsize::new(2).unwrap(),
            },
        )
        .unwrap();
        assert_eq!(
            app.analysis()
                .run_with_session_batch_size(
                    &limited,
                    std::num::NonZeroUsize::new(1).unwrap(),
                    &policy,
                )
                .unwrap()
                .result
                .sessions
                .len(),
            2
        );
    }

    #[test]
    fn receipted_analysis_names_canonical_session_selection_and_exact_count_units() {
        use std::num::NonZeroUsize;

        use crate::analysis_pipeline::AnalysisPolicySpec;
        use crate::models::{AnalysisRequest, AnalysisSessionSelection};

        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().into_owned());
        let app = SessionSearch::open(config).unwrap();
        for (id, updated_at) in [
            ("claude:a-newer", "2026-03-01T00:00:00Z"),
            ("claude:z-older", "2026-01-01T00:00:00Z"),
        ] {
            let mut parsed = minimal_record(
                Provider::Claude,
                std::path::Path::new(if id == "claude:a-newer" {
                    "/fixture/a-newer.jsonl"
                } else {
                    "/fixture/z-older.jsonl"
                }),
                String::new(),
            );
            parsed.session.id = id.into();
            parsed.session.provider_session_id = id.into();
            parsed.session.updated_at = Some(updated_at.parse().unwrap());
            parsed.messages = vec![
                Message {
                    seq: 0,
                    role: Role::User,
                    ts: None,
                    tool_name: None,
                    kind: MessageKind::Conversation,
                    tool_call_id: None,
                    is_compaction: false,
                    content: "human input".into(),
                    provenance: human_original_provenance(),
                },
                Message {
                    seq: 1,
                    role: Role::Assistant,
                    ts: None,
                    tool_name: None,
                    kind: MessageKind::Conversation,
                    tool_call_id: None,
                    is_compaction: false,
                    content: "assistant output".into(),
                    provenance: Default::default(),
                },
                Message {
                    seq: 2,
                    role: Role::Tool,
                    ts: None,
                    tool_name: Some("test".into()),
                    kind: MessageKind::ToolResult,
                    tool_call_id: None,
                    is_compaction: false,
                    content: "tool output".into(),
                    provenance: Default::default(),
                },
            ];
            app.database().upsert_session(&parsed, 0, 0).unwrap();
        }
        let policy = AnalysisPolicySpec::default().compile().unwrap();
        let request = AnalysisRequest::new(
            SearchFilters::default(),
            AnalysisSessionSelection::FirstCanonicalSessions {
                max_sessions: NonZeroUsize::new(1).unwrap(),
            },
        )
        .unwrap();

        let run = app.analysis().run(&request, &policy).unwrap();
        assert_eq!(
            run.result.sessions.keys().cloned().collect::<Vec<_>>(),
            ["claude:a-newer"]
        );
        assert_eq!(run.receipt.selected_sessions, 1);
        assert_eq!(run.receipt.messages_in_selected_sessions, 3);
        assert_eq!(run.receipt.analyzed_user_messages, 1);
        assert!(run.receipt.has_more);
        assert_eq!(
            run.receipt.last_selected_session_id.as_deref(),
            Some("claude:a-newer")
        );
        assert!(run.receipt.policy_digest.starts_with("sha256:"));
        assert!(run.receipt.corpus_digest.starts_with("sha256:"));
        assert!(run.receipt.result_digest.starts_with("sha256:"));
        assert_ne!(run.receipt.policy_digest, run.receipt.corpus_digest);
        assert_ne!(run.receipt.corpus_digest, run.receipt.result_digest);
        assert_eq!(run.receipt.policy_digest, policy.policy_digest());
        assert_eq!(run.receipt.result_digest, run.result.result_digest());
        assert_eq!(
            serde_json::to_value(run.receipt.selection).unwrap(),
            serde_json::json!({
                "kind": "first_canonical_sessions",
                "max_sessions": 1
            })
        );

        let all = app
            .analysis()
            .run(
                &AnalysisRequest::new(
                    SearchFilters::default(),
                    AnalysisSessionSelection::AllEligible,
                )
                .unwrap(),
                &policy,
            )
            .unwrap();
        assert_eq!(all.receipt.selected_sessions, 2);
        assert_eq!(all.receipt.messages_in_selected_sessions, 6);
        assert_eq!(all.receipt.analyzed_user_messages, 2);
        assert!(!all.receipt.has_more);
        assert_ne!(all.receipt.corpus_digest, run.receipt.corpus_digest);
    }

    #[test]
    fn analysis_request_rejects_generic_session_limit_with_actionable_replacement() {
        use crate::models::{AnalysisRequest, AnalysisSessionSelection};

        let error = AnalysisRequest::new(
            SearchFilters {
                limit: 5,
                ..SearchFilters::default()
            },
            AnalysisSessionSelection::AllEligible,
        )
        .unwrap_err();

        assert!(error.to_string().contains("SearchFilters.limit"));
        assert!(error.to_string().contains("FirstCanonicalSessions"));
    }
}

impl SessionSearch {
    /// Load platform configuration and open the configured index.
    pub fn load() -> Result<Self> {
        Self::open(Config::load()?)
    }

    /// Open an index using an explicit configuration.
    ///
    /// `existing-only` opens a SQLite-enforced read-only handle and never creates directories,
    /// installs schema objects, refreshes sources, or changes journal policy. SQLite may create or
    /// update empty `-wal`/`-shm` coordination sidecars when opening a WAL database whose sidecars
    /// are absent; it does not modify the durable database contents. Other refresh modes
    /// may create a new current-schema database and later perform the configured incremental
    /// refresh. Newer, hybrid, and incomplete current schemas fail before an ordinary query.
    pub fn open(config: Config) -> Result<Self> {
        let inputs = TrustedAccessInputs::capture(&config.search.scope, Vec::new())?;
        Self::open_with_access_inputs(config, inputs)
    }

    /// Open an index with trusted runtime roots supplied by a harness integration.
    pub fn open_with_access_inputs(
        config: Config,
        access_inputs: TrustedAccessInputs,
    ) -> Result<Self> {
        Self::open_with_access_inputs_and_runtime(config, access_inputs, None)
    }

    /// Open an independent prepared-index reader sharing one fixed data-parallel worker budget.
    ///
    /// The caller must select `existing-only`; this constructor reuses all normal schema and
    /// authority validation while preventing per-reader Rayon pools. Each returned application
    /// owns one SQLite read connection, so a bounded set can read a WAL database concurrently.
    pub(crate) fn open_prepared_reader(
        config: Config,
        access_inputs: TrustedAccessInputs,
        runtime: Arc<ExecutionRuntime>,
    ) -> Result<Self> {
        if config.index.refresh != IndexRefresh::ExistingOnly {
            anyhow::bail!(
                "prepared readers require index.refresh=existing-only after the freshness coordinator completes"
            );
        }
        Self::open_with_access_inputs_and_runtime(config, access_inputs, Some(runtime))
    }

    fn open_with_access_inputs_and_runtime(
        config: Config,
        access_inputs: TrustedAccessInputs,
        runtime: Option<Arc<ExecutionRuntime>>,
    ) -> Result<Self> {
        let access = EffectiveAccessScope::resolve(&config.search.scope, access_inputs)?;
        let schema_state = IndexCoordinator::new(&config).inspect_schema()?;
        match schema_state {
            SchemaState::Missing if config.index.refresh == IndexRefresh::ExistingOnly => {
                anyhow::bail!(
                    "existing-only index {} does not exist; run `aise reindex --full` without --index-refresh existing-only",
                    config.db_path().display()
                )
            }
            SchemaState::Older { current, .. }
                if config.index.refresh == IndexRefresh::ExistingOnly
                    && current < MIN_READABLE_SCHEMA_VERSION =>
            {
                anyhow::bail!(
                    "existing-only index {} uses unreadable schema generation {current}; run `aise reindex --full` without --index-refresh existing-only",
                    config.db_path().display()
                )
            }
            SchemaState::Missing | SchemaState::Current | SchemaState::Older { .. } => {}
            SchemaState::Newer { current, supported } => anyhow::bail!(
                "index {} uses schema generation {current}, newer than this aise build supports ({supported}); upgrade aise before opening it",
                config.db_path().display()
            ),
            SchemaState::UnknownProviders { providers } => anyhow::bail!(
                "{}",
                SchemaState::unknown_providers_message(&config.db_path(), &providers)
            ),
            SchemaState::RepairableLayout { reason }
                if config.index.refresh == IndexRefresh::ExistingOnly =>
            {
                anyhow::bail!(
                    "existing-only index {} needs a one-time message-search index rebuild ({reason}); run a writable aise command (without --index-refresh existing-only) to self-heal it in place",
                    config.db_path().display()
                )
            }
            SchemaState::RepairableLayout { .. } => {
                // Elect one cross-process writer, re-inspect under the lock (a peer process may have
                // already healed it), and rebuild the derived message-search indexes online from the
                // intact base rows via the atomic exclusive migration. Then fall through to the
                // normal open below against the now-current schema.
                let heal_threads = NonZeroUsize::new(config.resolve_threads())
                    .expect("Config::resolve_threads always returns at least one");
                let heal_busy_timeout = config.index.busy_timeout_ms;
                crate::indexer::with_index_update_lock(&config, || {
                    if matches!(
                        IndexCoordinator::new(&config).inspect_schema()?,
                        SchemaState::RepairableLayout { .. }
                    ) {
                        let db = Db::open_with_threads(
                            &config.db_path(),
                            heal_busy_timeout,
                            heal_threads,
                        )?;
                        db.migrate_message_search_schema_exclusive()?;
                    }
                    Ok(())
                })?;
            }
            SchemaState::RecoveryRequired { reason } => anyhow::bail!(
                "index {} requires offline recovery: {reason}; stop AISE processes, then run `aise reindex --full`",
                config.db_path().display()
            ),
        }
        if config.index.refresh != IndexRefresh::ExistingOnly {
            fs::create_dir_all(config.cache_dir())?;
        }
        let worker_threads = NonZeroUsize::new(config.resolve_threads())
            .expect("Config::resolve_threads always returns at least one");
        let mut db = if config.index.refresh == IndexRefresh::ExistingOnly {
            if let Some(runtime) = runtime {
                Db::open_existing_read_only_with_runtime(
                    &config.db_path(),
                    config.index.busy_timeout_ms,
                    runtime,
                )?
            } else {
                Db::open_existing_read_only_with_threads(
                    &config.db_path(),
                    config.index.busy_timeout_ms,
                    worker_threads,
                )?
            }
        } else {
            Db::open_with_threads(
                &config.db_path(),
                config.index.busy_timeout_ms,
                worker_threads,
            )?
        };
        db.set_access_scope(access.clone());
        db.set_implicit_index_maintenance(config.index.refresh != IndexRefresh::ExistingOnly);
        Ok(Self { config, access, db })
    }

    /// Open with explicit maintenance authority even when implicit refresh is configured as
    /// `existing-only`. CLI `reindex` and `compact` are explicit write requests; ordinary query
    /// commands continue to receive a SQLite-enforced read-only handle. The stored configuration
    /// retains the user's selected refresh policy.
    pub(crate) fn open_for_maintenance(mut config: Config) -> Result<Self> {
        let refresh = config.index.refresh;
        config.index.refresh = IndexRefresh::BeforeQuery;
        let mut app = Self::open(config)?;
        app.config.index.refresh = refresh;
        Ok(app)
    }

    /// Effective configuration used by every service handle.
    pub const fn config(&self) -> &Config {
        &self.config
    }

    /// Immutable access authority used by every read service.
    pub const fn access_scope(&self) -> &EffectiveAccessScope {
        &self.access
    }

    /// Session catalog operations.
    pub const fn catalog(&self) -> CatalogService<'_> {
        CatalogService::new(&self.db)
    }

    /// Message search and context operations.
    pub const fn messages(&self) -> MessageService<'_> {
        MessageService::new(&self.config, &self.db, SearchSurface::Rust)
    }

    /// Construct message search for a language or protocol adapter's documented defaults.
    #[doc(hidden)]
    pub const fn messages_for_surface(&self, surface: SearchSurface) -> MessageService<'_> {
        MessageService::new(&self.config, &self.db, surface)
    }

    /// Open an exhaustive literal, regex, or queryless search as fully enriched batches on a
    /// dedicated read-only connection.
    ///
    /// Ordinary callers should prefer [`MessageService::search`], which returns one materialized
    /// response. `batch_rows` trades active memory for handoff and enrichment-query frequency; it
    /// does not change result membership, ordering, context, the request-wide union of included
    /// data, or terminal metadata. It accepts any positive `usize`; an allocation failure returns
    /// an error that asks for a smaller value.
    ///
    /// # Errors
    ///
    /// Returns an error when the request selects a finite page or fuzzy matching. Use
    /// [`MessageService::search`] for those requests.
    pub fn message_search_batches(
        &self,
        request: MessageSearchRequest,
        batch_rows: NonZeroUsize,
    ) -> Result<MessageSearchBatches> {
        self.message_search_batches_for_surface(SearchSurface::Rust, request, batch_rows)
    }

    /// Construct exhaustive batches with one output surface's documented default policy.
    #[doc(hidden)]
    pub fn message_search_batches_for_surface(
        &self,
        surface: SearchSurface,
        request: MessageSearchRequest,
        batch_rows: NonZeroUsize,
    ) -> Result<MessageSearchBatches> {
        MessageSearchBatches::spawn(
            self.config.clone(),
            self.access.clone(),
            surface,
            request,
            batch_rows,
        )
    }

    /// File history operations.
    pub const fn files(&self) -> FileService<'_> {
        FileService::new(&self.db)
    }

    /// Destination-independent session export operations.
    pub const fn exports(&self) -> ExportService<'_> {
        ExportService::new(&self.db)
    }

    /// Index lifecycle operations.
    pub const fn index(&self) -> IndexService<'_> {
        IndexService::new(&self.config, &self.db)
    }

    /// Index diagnostics and destructive maintenance operations.
    pub const fn maintenance(&self) -> MaintenanceService<'_> {
        MaintenanceService::new(&self.config, &self.db)
    }

    /// Effective provider roots and discovery status.
    pub const fn sources(&self) -> SourceService<'_> {
        SourceService::new(&self.config)
    }

    /// Indexed correction, planning, and statistics operations.
    pub const fn analysis(&self) -> AnalysisService<'_> {
        AnalysisService::new(&self.config, &self.db)
    }

    /// Crate-internal storage access for operations not yet represented by a service.
    ///
    /// External consumers (Python bindings, integration tests, downstream crates) must use
    /// the typed services above; raw [`Db`] access is not part of the supported API.
    pub(crate) const fn database(&self) -> &Db {
        &self.db
    }

    /// Install a frontend-specific progress sink.
    pub fn set_progress_reporter(&mut self, reporter: impl Fn(&str) + Send + Sync + 'static) {
        self.db.set_progress_reporter(reporter);
    }
}

#[cfg(test)]
mod execution_runtime_tests {
    use super::*;
    use crate::config::{SearchScopeConfig, SearchScopeMode};
    use crate::message_search::{MessageQuery, MessageTarget};
    use crate::models::{
        FileEdit, FileQuery, Message, MessageFilters, MessageKind, MessageSearchMode, Provider,
        Role, SearchFilters,
    };
    use crate::util::minimal_record;
    use std::sync::atomic::{AtomicU32, Ordering};
    use std::sync::Arc;

    fn all_session_filters(limit: usize) -> SearchFilters {
        SearchFilters {
            limit,
            ..SearchFilters::default()
        }
    }

    #[test]
    fn ordinary_open_refuses_newer_schema_before_initialization_can_mutate_it() {
        let root = tempfile::tempdir().unwrap();
        let db_path = root.path().join("index.db");
        drop(Db::open(&db_path).unwrap());
        let conn = rusqlite::Connection::open(&db_path).unwrap();
        conn.pragma_update(None, "user_version", crate::db::SCHEMA_VERSION + 1)
            .unwrap();
        let schema_before: String = conn
            .query_row(
                "select group_concat(coalesce(sql, ''), char(10))
                   from (select sql from sqlite_schema order by type, name)",
                [],
                |row| row.get(0),
            )
            .unwrap();
        drop(conn);

        let mut config = Config::default();
        config.index.db_path = Some(db_path.to_string_lossy().into_owned());
        let error = SessionSearch::open(config)
            .err()
            .expect("newer schema must be rejected")
            .to_string();
        assert!(error.contains("newer than this aise build"), "{error}");

        let conn = rusqlite::Connection::open_with_flags(
            &db_path,
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
        )
        .unwrap();
        let schema_after: String = conn
            .query_row(
                "select group_concat(coalesce(sql, ''), char(10))
                   from (select sql from sqlite_schema order by type, name)",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(schema_after, schema_before);
    }

    /// A provider variant can be added without a schema-generation bump (Prime Agent was, at
    /// generation 5), so an index a newer aise wrote can carry a `sessions.provider` value this
    /// build cannot decode. Before this guard the older build failed deep inside whichever query
    /// touched such a row with "unrecognized provider ... run `aise reindex --full`", and that
    /// command failed the same way while enumerating retained sessions -- circular advice that
    /// locked the older build out of every command (observed from a Pi session on 2026-08-13).
    /// The open must refuse once, name the provider, and say the fix: upgrade aise.
    #[test]
    fn ordinary_open_refuses_an_index_carrying_a_provider_this_build_does_not_know() {
        let root = tempfile::tempdir().unwrap();
        let db_path = root.path().join("index.db");
        {
            let db = Db::open(&db_path).unwrap();
            let parsed = minimal_record(
                Provider::Codex,
                std::path::Path::new("/tmp/known-session.jsonl"),
                String::new(),
            );
            db.replace_session(&parsed, 0, 0).unwrap();
        }
        let conn = rusqlite::Connection::open(&db_path).unwrap();
        conn.execute(
            "update sessions set provider = 'future-provider' where id = 'codex:known-session'",
            [],
        )
        .unwrap();
        drop(conn);

        let mut config = Config::default();
        config.index.db_path = Some(db_path.to_string_lossy().into_owned());
        for refresh in [IndexRefresh::Auto, IndexRefresh::ExistingOnly] {
            let mut config = config.clone();
            config.index.refresh = refresh;
            let error = SessionSearch::open(config)
                .err()
                .unwrap_or_else(|| panic!("{refresh:?}: unknown provider must be rejected"))
                .to_string();
            assert!(error.contains("future-provider"), "{refresh:?}: {error}");
            assert!(error.contains("upgrade aise"), "{refresh:?}: {error}");
            assert!(
                error.contains(env!("CARGO_PKG_VERSION")),
                "{refresh:?}: names this build so the reader can compare: {error}"
            );
            assert!(
                !error.contains("reindex --full"),
                "{refresh:?}: reindex is the command that fails; advising it is circular: {error}"
            );
        }
    }

    #[test]
    fn restricted_open_without_authoritative_roots_fails_before_database_creation() {
        let root = tempfile::tempdir().unwrap();
        let db_path = root.path().join("must-not-exist.db");
        let mut config = Config::default();
        config.index.db_path = Some(db_path.to_string_lossy().into_owned());
        config.search.scope = SearchScopeConfig {
            mode: SearchScopeMode::AllowedRoots,
            roots: Vec::new(),
            include_invocation_directory: false,
        };

        let error = SessionSearch::open_with_access_inputs(config, TrustedAccessInputs::default())
            .err()
            .expect("restricted open must fail")
            .to_string();

        assert!(error.contains("resolved no authoritative roots"), "{error}");
        assert!(!db_path.exists());
    }

    #[test]
    fn allowed_roots_scope_is_consistent_across_read_services_and_exact_ids() {
        let root = tempfile::tempdir().unwrap();
        let allowed = root.path().join("allowed");
        let outside = root.path().join("outside");
        let sibling = root.path().join("allowed-sibling");
        fs::create_dir_all(&allowed).unwrap();
        fs::create_dir_all(&outside).unwrap();
        fs::create_dir_all(&sibling).unwrap();

        let mut config = Config::default();
        config.index.db_path = Some(root.path().join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(root.path().join("cache").to_string_lossy().into_owned());
        config.search.scope = SearchScopeConfig {
            mode: SearchScopeMode::AllowedRoots,
            roots: vec![allowed.to_string_lossy().into_owned()],
            include_invocation_directory: false,
        };
        let app =
            SessionSearch::open_with_access_inputs(config, TrustedAccessInputs::default()).unwrap();

        let insert = |id: &str, workspace: &std::path::Path, transcript: &std::path::Path| {
            let mut parsed = minimal_record(Provider::Claude, transcript, String::new());
            parsed.session.id = id.into();
            parsed.session.provider_session_id = id.replace(':', "-");
            parsed.session.cwd = Some(workspace.to_string_lossy().into_owned());
            parsed.session.repo_root = Some(workspace.join("repo").to_string_lossy().into_owned());
            if id != "claude:allowed" {
                parsed.session.title = Some("scope needle scope needle scope needle".into());
            }
            parsed.session.preview_text = "scope needle".into();
            parsed.transcript_text = "scope needle transcript".into();
            parsed.messages = vec![Message {
                seq: 0,
                role: Role::User,
                ts: None,
                tool_name: None,
                kind: MessageKind::Conversation,
                tool_call_id: None,
                is_compaction: false,
                content: "scope needle message".into(),
                provenance: Default::default(),
            }];
            parsed.file_edits = vec![FileEdit {
                seq: 0,
                ts: None,
                tool: "Write".into(),
                file_path: workspace.join("scope.txt").to_string_lossy().into_owned(),
                file_name: "scope.txt".into(),
                new_content: Some(format!("content from {id}")),
                edits: Vec::new(),
            }];
            app.database().upsert_session(&parsed, 0, 0).unwrap();
        };

        insert(
            "claude:allowed",
            &allowed.join("project"),
            &outside.join("allowed-transcript.jsonl"),
        );
        // A transcript below an allowed root must not grant authority to an unrelated workspace.
        insert(
            "claude:hidden",
            &outside.join("project"),
            &allowed.join("hidden-transcript.jsonl"),
        );
        // Prefix matching must respect path components rather than string prefixes.
        insert(
            "claude:sibling",
            &sibling.join("project"),
            &outside.join("sibling-transcript.jsonl"),
        );

        let sessions = app
            .catalog()
            .list_sessions(&all_session_filters(0))
            .unwrap();
        assert_eq!(sessions.len(), 1);
        assert_eq!(sessions[0].id, "claude:allowed");
        let session_hits = app
            .catalog()
            .search_sessions(
                "scope needle",
                &all_session_filters(0),
                None,
                &ScoringConfig::default(),
            )
            .unwrap();
        assert_eq!(session_hits.len(), 1);
        assert_eq!(session_hits[0].session.id, "claude:allowed");
        let narrow_scoring = ScoringConfig::default();
        let bounded_hits = app
            .catalog()
            .search_sessions(
                "scope needle",
                &all_session_filters(1),
                None,
                &narrow_scoring,
            )
            .unwrap();
        assert_eq!(bounded_hits.len(), 1);
        assert_eq!(bounded_hits[0].session.id, "claude:allowed");

        let message_response = app
            .messages()
            .search(
                MessageSearchRequest::builder(
                    MessageQuery::literal("scope needle").unwrap(),
                    MessageTarget::content(),
                )
                .build()
                .unwrap(),
            )
            .unwrap();
        assert_eq!(message_response.hits().len(), 1);
        assert_eq!(message_response.hits()[0].session_id, "claude:allowed");
        assert!(app
            .messages()
            .context("claude:hidden", 0, 1, 1)
            .unwrap()
            .is_empty());
        assert!(app
            .messages()
            .session_metadata(&["claude:hidden".into()])
            .unwrap()
            .is_empty());

        let files = app.files().search(&FileQuery::default()).unwrap();
        assert_eq!(files.len(), 1);
        assert!(std::path::Path::new(&files[0].file_path).starts_with(&allowed));
        let cross_reference = app.files().cross_reference(&FileQuery::default()).unwrap();
        assert_eq!(cross_reference.len(), 1);
        assert_eq!(cross_reference[0].session_id, "claude:allowed");
        let history = app
            .files()
            .history("scope.txt", &FileQuery::default())
            .unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].session_id, "claude:allowed");
        let reconstructed = app
            .files()
            .reconstruct("scope.txt", &FileQuery::default(), None)
            .unwrap();
        assert_eq!(reconstructed.session_id, "claude:allowed");
        assert_eq!(reconstructed.content, "content from claude:allowed");
        let analysis = app
            .analysis()
            .documents(&all_session_filters(10), None)
            .unwrap();
        assert_eq!(analysis.documents.len(), 1);
        assert_eq!(analysis.documents[0].session.id, "claude:allowed");

        assert!(app
            .exports()
            .render_full("claude:allowed", crate::export::ExportFormat::Json)
            .is_ok());
        let hidden_error = app
            .catalog()
            .resolve_session("claude:hidden")
            .unwrap_err()
            .to_string();
        assert!(hidden_error.contains("no session matches"));
        assert!(!hidden_error.contains("allowed-sibling"));
        assert!(!hidden_error.contains("outside"));
        assert!(app
            .exports()
            .render_full("claude:hidden", crate::export::ExportFormat::Json)
            .unwrap_err()
            .to_string()
            .contains("no session matches"));
    }

    #[test]
    fn applications_own_independent_worker_counts() {
        let root = tempfile::tempdir().unwrap();
        let mut one_config = Config::default();
        one_config.index.db_path = Some(root.path().join("one.db").to_string_lossy().into_owned());
        one_config.performance.threads = 1;
        let mut two_config = Config::default();
        two_config.index.db_path = Some(root.path().join("two.db").to_string_lossy().into_owned());
        two_config.performance.threads = 2;

        let one = SessionSearch::open(one_config).unwrap();
        let two = SessionSearch::open(two_config).unwrap();

        assert_eq!(one.database().worker_threads(), 1);
        assert_eq!(two.database().worker_threads(), 2);
    }

    #[test]
    fn prepared_readers_require_existing_only_and_reuse_the_shared_worker_budget() {
        let root = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(root.path().join("index.db").to_string_lossy().into_owned());
        drop(SessionSearch::open(config.clone()).unwrap());
        let runtime = Arc::new(ExecutionRuntime::new(
            NonZeroUsize::new(2).expect("nonzero worker count"),
        ));

        let error = SessionSearch::open_prepared_reader(
            config.clone(),
            TrustedAccessInputs::default(),
            Arc::clone(&runtime),
        )
        .err()
        .expect("writable policy must be rejected")
        .to_string();
        assert!(error.contains("existing-only"), "{error}");

        config.index.refresh = IndexRefresh::ExistingOnly;
        let first = SessionSearch::open_prepared_reader(
            config.clone(),
            TrustedAccessInputs::default(),
            Arc::clone(&runtime),
        )
        .unwrap();
        let second =
            SessionSearch::open_prepared_reader(config, TrustedAccessInputs::default(), runtime)
                .unwrap();
        assert_eq!(first.database().worker_threads(), 2);
        assert_eq!(second.database().worker_threads(), 2);
    }

    #[test]
    fn existing_only_search_never_builds_the_lazy_trigram_index() {
        let root = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(root.path().join("index.db").to_string_lossy().into_owned());
        config.index.cache_dir = Some(root.path().join("cache").to_string_lossy().into_owned());
        let mut parsed = minimal_record(
            Provider::Claude,
            std::path::Path::new("/fixture/session.jsonl"),
            String::new(),
        );
        parsed.session.id = "claude:existing-only".into();
        parsed.session.provider_session_id = "existing-only".into();
        parsed.messages = vec![Message {
            seq: 0,
            role: Role::User,
            ts: None,
            tool_name: None,
            kind: MessageKind::Conversation,
            tool_call_id: None,
            is_compaction: false,
            content: "request failed with ECONNRESET".into(),
            provenance: Default::default(),
        }];
        let writer = SessionSearch::open(config.clone()).unwrap();
        writer.database().upsert_session(&parsed, 0, 0).unwrap();
        drop(writer);
        let conn = rusqlite::Connection::open_with_flags(
            config.db_path(),
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
        )
        .unwrap();
        let schema_before: String = conn
            .query_row(
                "select group_concat(coalesce(sql, ''), char(10))
                   from (select sql from sqlite_schema order by type, name)",
                [],
                |row| row.get(0),
            )
            .unwrap();
        drop(conn);

        config.index.refresh = IndexRefresh::ExistingOnly;
        let mut app = SessionSearch::open(config).unwrap();
        let write_error = app
            .database()
            .upsert_session(&parsed, 1, 1)
            .expect_err("existing-only database authority must reject writes")
            .to_string();
        assert!(
            write_error.contains("readonly") || write_error.contains("read-only"),
            "{write_error}"
        );

        let progress_calls = Arc::new(AtomicU32::new(0));
        let observed_calls = Arc::clone(&progress_calls);
        app.set_progress_reporter(move |_message| {
            observed_calls.fetch_add(1, Ordering::Relaxed);
        });
        let hits = app
            .messages()
            .search_legacy(
                "ECONNRESET",
                &MessageFilters {
                    match_mode: MessageSearchMode::Regex,
                    ..Default::default()
                },
            )
            .unwrap();

        assert_eq!(
            hits.len(),
            1,
            "existing-only still scans the unindexed delta"
        );
        assert_eq!(
            progress_calls.load(Ordering::Relaxed),
            0,
            "existing-only must not start implicit trigram maintenance"
        );
        assert!(
            app.database()
                .vocabulary(true, 0, None)
                .unwrap()
                .iter()
                .any(|(term, _, _)| term == "eco" || term == "con" || term == "res"),
            "v4 trigram vocabulary remains readable without legacy maintenance"
        );
        drop(app);
        let conn = rusqlite::Connection::open_with_flags(
            root.path().join("index.db"),
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
        )
        .unwrap();
        let schema_after: String = conn
            .query_row(
                "select group_concat(coalesce(sql, ''), char(10))
                   from (select sql from sqlite_schema order by type, name)",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(schema_after, schema_before);
    }

    #[test]
    fn existing_only_open_refuses_missing_index_without_creating_paths() {
        let root = tempfile::tempdir().unwrap();
        let db_path = root.path().join("missing/index.db");
        let cache_path = root.path().join("missing/cache");
        let mut config = Config::default();
        config.index.db_path = Some(db_path.to_string_lossy().into_owned());
        config.index.cache_dir = Some(cache_path.to_string_lossy().into_owned());
        config.index.refresh = IndexRefresh::ExistingOnly;

        let error = SessionSearch::open(config)
            .err()
            .expect("missing existing-only index must fail")
            .to_string();

        assert!(error.contains("does not exist"), "{error}");
        assert!(error.contains("aise reindex --full"), "{error}");
        assert!(!db_path.exists());
        assert!(!cache_path.exists());
        assert!(!root.path().join("missing").exists());
    }
}

/// Typed indexed analysis shared by native adapters.
///
/// This service returns data and performs no terminal or filesystem I/O. It deliberately keeps
/// legacy recovery-directory counters out of the canonical API: statistics describe the shared
/// index and honor the same structural message filters as CLI search.
#[derive(Clone, Copy)]
pub struct AnalysisService<'app> {
    config: &'app Config,
    db: &'app Db,
}

/// Immutable analytical provenance produced with one result under the same SQLite snapshot.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AnalysisReceipt {
    /// Population strategy applied after structural session filtering.
    pub selection: crate::models::AnalysisSessionSelection,
    /// SQLite schema version observed by this run.
    pub database_schema_version: i64,
    /// Sessions whose metadata and attributable human-authored text entered the analysis.
    pub selected_sessions: u64,
    /// All-role message rows belonging to the selected sessions.
    pub messages_in_selected_sessions: u64,
    /// Original messages with attributable human-authored text that entered analysis.
    pub analyzed_user_messages: u64,
    /// Whether the named bounded selection left at least one matching session unselected.
    pub has_more: bool,
    /// Last selected canonical session ID, or `None` for an empty selection.
    pub last_selected_session_id: Option<String>,
    /// Greatest selected session update/create timestamp, or `None` when unavailable.
    pub max_selected_session_updated_at: Option<String>,
    /// Identity of normalized compiled policy semantics.
    pub policy_digest: String,
    /// Identity of selected, ordered session metadata and attributable human-authored input.
    pub corpus_digest: String,
    /// Identity of the canonical in-memory [`AnalysisResult`](crate::AnalysisResult).
    pub result_digest: String,
}

/// Pure analysis result paired with the exact selection, snapshot input, policy, and result facts.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ReceiptedAnalysis {
    /// Canonical classifications, relationships, vocabulary, and graph inputs.
    pub result: crate::analysis_pipeline::AnalysisResult,
    /// Same-snapshot selection, count, and digest facts for `result`.
    pub receipt: AnalysisReceipt,
}

const DEFAULT_ANALYSIS_SESSION_BATCH_SIZE: std::num::NonZeroUsize =
    std::num::NonZeroUsize::new(50).expect("analysis session batch constant is nonzero");

impl<'app> AnalysisService<'app> {
    /// Create an analysis service with configuration-backed capability and planning policy.
    pub const fn new(config: &'app Config, db: &'app Db) -> Self {
        Self { config, db }
    }

    pub(crate) fn corrections_with_resolved_policies(
        &self,
        filters: &crate::models::MessageFilters,
        policies: &crate::corrections::ResolvedCorrectionPolicySet,
    ) -> Result<crate::corrections::MessageClassificationReport> {
        let matches = self.db.find_corrections(policies, filters)?;
        Ok(crate::corrections::MessageClassificationReport {
            policies: policies.receipts(),
            matches,
        })
    }

    /// Resolve and execute one read-only deterministic skill capability.
    ///
    /// For `K` selected packages and `N` catalog entries, descriptor lookup is `O(K * N)`;
    /// canonical duplicate detection is expected `O(K)`. Only selected capability documents are
    /// compiled, under one aggregate byte budget. Message classification then uses the existing
    /// indexed correction query exactly once.
    pub fn run_skill(
        &self,
        query: &crate::skill_run::SkillRunQuery,
    ) -> Result<crate::skill_run::SkillRunReport> {
        let crate::skill_run::SkillCapabilityInput::MessageClassification(arguments) = &query.input;
        let embedded_primary = matches!(
            &query.skill,
            crate::skill_catalog::SkillSelector::Name(selector)
                if selector.name.as_str() == crate::corrections::EMBEDDED_POLICY_NAME
        );
        let roots = || {
            self.config
                .skills
                .search_paths
                .iter()
                .map(|root| crate::util::expand_tilde(root))
                .collect::<Vec<_>>()
        };

        let (resolved_skill, policies) = if embedded_primary {
            if arguments.additional_skills.iter().any(|selector| {
                matches!(
                    selector,
                    crate::skill_catalog::SkillSelector::Name(selector)
                        if selector.name.as_str() == crate::corrections::EMBEDDED_POLICY_NAME
                )
            }) {
                bail!("skill \"corrections\" was selected more than once");
            }
            let mut capability_budget = crate::message_classification::CapabilityLoadBudget::new();
            let mut compiled = vec![match &query.definition {
                Some(definition) => crate::message_classification::compile_inline_definition(
                    definition.clone(),
                    crate::corrections::EMBEDDED_POLICY_NAME.to_string(),
                    env!("CARGO_PKG_VERSION").to_string(),
                    &mut capability_budget,
                )?,
                None => crate::corrections::embedded_policy()?,
            }];
            if !arguments.additional_skills.is_empty() {
                let catalog = crate::skill_catalog::load_skill_catalog(&roots());
                let descriptors = crate::skill_catalog::resolve_skill_selectors(
                    &arguments.additional_skills,
                    &catalog,
                )?;
                if descriptors.iter().any(|descriptor| {
                    descriptor.frontmatter.as_ref().is_some_and(|frontmatter| {
                        frontmatter.name == crate::corrections::EMBEDDED_POLICY_NAME
                    })
                }) {
                    bail!(
                        "the embedded corrections skill and an installed corrections package \
                         identify the same capability; remove the duplicate --skill selector"
                    );
                }
                let additional = crate::message_classification::compile_skill_descriptors(
                    descriptors,
                    &mut capability_budget,
                )?;
                compiled.extend(additional.policies().iter().cloned());
            }
            (
                crate::skill_run::ResolvedSkillReceipt {
                    name: crate::skill_catalog::SkillName::try_from(
                        crate::corrections::EMBEDDED_POLICY_NAME.to_string(),
                    )
                    .map_err(anyhow::Error::msg)?,
                    package_version: Some(env!("CARGO_PKG_VERSION").to_string()),
                    selected_location: crate::skill_run::SelectedSkillLocation::Embedded,
                    execution_source: if query.definition.is_some() {
                        crate::skill_run::CapabilityExecutionSource::Inline
                    } else {
                        crate::skill_run::CapabilityExecutionSource::Embedded
                    },
                },
                crate::corrections::ResolvedCorrectionPolicySet::from_policies(compiled),
            )
        } else {
            let catalog = crate::skill_catalog::load_skill_catalog(&roots());
            let selectors = std::iter::once(query.skill.clone())
                .chain(arguments.additional_skills.iter().cloned())
                .collect::<Vec<_>>();
            let descriptors = crate::skill_catalog::resolve_skill_selectors(&selectors, &catalog)?;
            let primary = descriptors
                .first()
                .context("a skill run requires one primary resolved skill")?;
            let frontmatter = primary
                .frontmatter
                .as_ref()
                .context("resolved skill has no valid frontmatter")?;
            let package_version = frontmatter
                .metadata
                .get("version")
                .cloned()
                .context("runnable skill must declare metadata.version in SKILL.md")?;
            let resolved_skill = crate::skill_run::ResolvedSkillReceipt {
                name: crate::skill_catalog::SkillName::try_from(frontmatter.name.clone())
                    .map_err(anyhow::Error::msg)?,
                package_version: Some(package_version.clone()),
                selected_location: crate::skill_run::SelectedSkillLocation::Path {
                    canonical_skill_md: primary.root.join("SKILL.md"),
                },
                execution_source: if query.definition.is_some() {
                    crate::skill_run::CapabilityExecutionSource::Inline
                } else {
                    crate::skill_run::CapabilityExecutionSource::Path {
                        canonical_capability_toml: primary
                            .capability
                            .clone()
                            .require_path(&frontmatter.name)?,
                    }
                },
            };
            let policies = match &query.definition {
                None => crate::message_classification::compile_skill_descriptors(
                    descriptors,
                    &mut crate::message_classification::CapabilityLoadBudget::new(),
                )?,
                Some(definition) => {
                    let mut capability_budget =
                        crate::message_classification::CapabilityLoadBudget::new();
                    let mut compiled =
                        vec![crate::message_classification::compile_inline_definition(
                            definition.clone(),
                            frontmatter.name.clone(),
                            package_version,
                            &mut capability_budget,
                        )?];
                    let additional = crate::message_classification::compile_skill_descriptors(
                        descriptors.into_iter().skip(1).collect(),
                        &mut capability_budget,
                    )?;
                    compiled.extend(additional.policies().iter().cloned());
                    crate::corrections::ResolvedCorrectionPolicySet::from_policies(compiled)
                }
            };
            (resolved_skill, policies)
        };
        let report = self.corrections_with_resolved_policies(&arguments.filters, &policies)?;
        let receipt = report
            .policies
            .first()
            .cloned()
            .context("a primary capability produced no policy receipt")?;
        Ok(crate::skill_run::SkillRunReport {
            requested_selector: query.skill.clone(),
            resolved_skill,
            output: crate::skill_run::SkillCapabilityOutput::MessageClassification(
                crate::skill_run::MessageClassificationResult { receipt, report },
            ),
        })
    }

    /// Aggregate slash-command usage with configured and request-specific token patterns.
    ///
    /// # Errors
    ///
    /// Returns an error when a configured/request pattern is invalid or the index query fails.
    ///
    /// # Complexity
    ///
    /// Time is proportional to filtered user-message rows times the combined command patterns.
    /// Memory is proportional to distinct matched commands, sessions, and projects.
    pub fn planning(
        &self,
        filters: &MessageFilters,
        command_patterns: &[String],
    ) -> Result<Vec<crate::models::PlanningCount>> {
        let command_filters =
            crate::analytics::compile_planning_filters(self.config, command_patterns)?;
        self.db.planning_usage(filters, &command_filters)
    }

    /// Count indexed messages by normalized role, ordered by role.
    ///
    /// # Errors
    ///
    /// Returns an error when the index query fails.
    pub fn role_statistics(
        &self,
        filters: &MessageFilters,
    ) -> Result<Vec<crate::analytics::RoleStat>> {
        let rows: Vec<_> = self
            .db
            .message_role_counts(filters)?
            .into_iter()
            .map(|(role, count)| crate::analytics::RoleStat { role, count })
            .collect();
        let limit = if filters.limit == 0 {
            usize::MAX
        } else {
            filters.limit
        };
        Ok(rows.into_iter().skip(filters.offset).take(limit).collect())
    }

    /// Return one bounded keyset page of provider-normalized session text for outward analysis.
    ///
    /// # Complexity
    ///
    /// Time and returned memory are proportional to the page's selected sessions plus their joined
    /// attributable human-authored text. Keyset traversal avoids work proportional to prior page
    /// offsets.
    pub fn documents(
        &self,
        filters: &SearchFilters,
        cursor: Option<&crate::models::AnalysisCursor>,
    ) -> Result<crate::models::AnalysisDocumentPage> {
        self.db.analysis_documents(filters, cursor)
    }

    /// Run a compiled provider-neutral policy over one snapshot of the indexed corpus.
    ///
    /// [`AnalysisRequest`](crate::models::AnalysisRequest) names either every eligible session or
    /// a reproducible canonical-session-ID prefix. Internal keyset traversal is automatic and
    /// does not alter analysis, receipts, or publication results. Attributable human-authored text
    /// streams per message, so memory is bounded by the policy's explicit bounds plus one message
    /// — a single session's aggregate user text is never materialized (except when a
    /// `user_text`/`any` classification rule runs without `max_classification_chars`).
    ///
    /// # Complexity
    ///
    /// Time is proportional to streamed human-authored bytes times applicable policy rules, plus
    /// phrase aggregation. Memory is bounded by policy limits plus one message, except an
    /// explicitly unbounded `user_text`/`any` classification rule retains its matched text.
    pub fn run(
        &self,
        request: &crate::models::AnalysisRequest,
        policy: &crate::analysis_pipeline::AnalysisPolicy,
    ) -> Result<ReceiptedAnalysis> {
        self.run_with_session_batch_size(request, DEFAULT_ANALYSIS_SESSION_BATCH_SIZE, policy)
    }

    /// Run with an explicit internal session batch size for invariant tests and internal tuning.
    /// This value controls database traversal only and must never change returned results.
    pub(crate) fn run_with_session_batch_size(
        &self,
        request: &crate::models::AnalysisRequest,
        session_batch_size: std::num::NonZeroUsize,
        policy: &crate::analysis_pipeline::AnalysisPolicy,
    ) -> Result<ReceiptedAnalysis> {
        let mut accumulator = policy.accumulator();
        let traversal = self.db.visit_analysis_sessions(
            request,
            session_batch_size,
            |session, message_count, user_message_count, chunks| {
                let mut content_chunks = chunks.map(|chunk| chunk.map(|(_seq, content)| content));
                accumulator.push_session_text_stream(
                    session,
                    message_count,
                    user_message_count,
                    &mut content_chunks,
                )
            },
        )?;
        let result = accumulator.finish()?;
        let receipt = AnalysisReceipt {
            selection: request.selection(),
            database_schema_version: crate::db::SCHEMA_VERSION,
            selected_sessions: traversal.selected_sessions,
            messages_in_selected_sessions: traversal.messages_in_selected_sessions,
            analyzed_user_messages: traversal.analyzed_user_messages,
            has_more: traversal.has_more,
            last_selected_session_id: traversal.last_selected_session_id,
            max_selected_session_updated_at: traversal
                .max_selected_session_updated_at
                .map(|value| value.to_rfc3339()),
            policy_digest: policy.policy_digest(),
            corpus_digest: traversal.corpus_digest,
            result_digest: result.result_digest(),
        };
        Ok(ReceiptedAnalysis { result, receipt })
    }
}

/// Read-only session rendering over the shared catalog database.
#[derive(Clone, Copy)]
pub struct ExportService<'db> {
    db: &'db Db,
}

impl<'db> ExportService<'db> {
    /// Create an export service over an existing database connection.
    pub const fn new(db: &'db Db) -> Self {
        Self { db }
    }

    /// Resolve an exact session ID or unambiguous prefix and render its complete transcript.
    ///
    /// The returned document is held in memory and performs no terminal or filesystem I/O.
    /// Resolution errors include candidate context for ambiguous prefixes.
    ///
    /// # Errors
    ///
    /// Returns an error when the identifier is missing or ambiguous, database access fails, or
    /// structured serialization fails.
    pub fn render_full(
        &self,
        id_or_prefix: &str,
        format: crate::export::ExportFormat,
    ) -> Result<crate::export::ExportDocument> {
        let session = self.db.resolve_session(id_or_prefix)?;
        crate::export::render_full(&session, format)
    }

    /// Select sessions and stream their rendered documents into one immutable bundle.
    ///
    /// This retains no full-corpus transcript collection: memory is bounded by the selected
    /// session metadata plus the largest individual rendered document.
    /// Time and output bytes are proportional to the selected transcripts; `filters.limit = 0`
    /// intentionally selects the complete filtered corpus.
    pub fn publish_bundle(
        &self,
        filters: &SearchFilters,
        plan: &crate::export::ExportPublicationPlan,
    ) -> Result<crate::export::ExportPublicationReceipt> {
        let sessions = self.db.list_recent(filters)?;
        let format = plan.format();
        let documents = sessions.into_iter().map(|session| {
            self.render_full(&session.id, format)
                .map(|document| (session.id, document))
        });
        plan.publish(documents)
    }
}

#[derive(Clone, Copy)]
/// Read-only discovery operations bound to a validated configuration.
pub struct SourceService<'config> {
    config: &'config Config,
}

impl<'config> SourceService<'config> {
    /// Create a source service over `config`.
    pub const fn new(config: &'config Config) -> Self {
        Self { config }
    }

    /// Discover enabled sources and return status for every supported provider.
    pub fn inventory(&self) -> Vec<crate::source::ProviderSourceStatus> {
        crate::source::inventory(self.config)
    }
}

/// Measurements from one completed compaction transaction.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CompactOutcome {
    pub before_bytes: u64,
    pub after_bytes: u64,
}

impl CompactOutcome {
    pub const fn reclaimed_bytes(self) -> u64 {
        self.before_bytes.saturating_sub(self.after_bytes)
    }
}

/// Diagnostics and writer-exclusive index maintenance.
#[derive(Clone, Copy)]
pub struct MaintenanceService<'app> {
    config: &'app Config,
    db: &'app Db,
}

impl<'app> MaintenanceService<'app> {
    pub const fn new(config: &'app Config, db: &'app Db) -> Self {
        Self { config, db }
    }

    pub fn diagnostics(&self) -> Result<DiagnosticStatus> {
        crate::diagnostics::collect(self.config, self.db)
    }

    /// Merge FTS segments, rebuild the database, and checkpoint the WAL while
    /// holding the same process lock used by index writers.
    pub fn compact(&self) -> Result<CompactOutcome> {
        indexer::with_index_update_lock(self.config, || {
            let before_bytes = self.db.storage_allocation()?.total_bytes;
            self.db.optimize_fts()?;
            self.db.vacuum()?;
            self.db.checkpoint_truncate()?;
            Ok(CompactOutcome {
                before_bytes,
                after_bytes: self.db.storage_allocation()?.total_bytes,
            })
        })
    }
}

#[derive(Clone, Copy)]
/// Explicit refresh, reindex, and status operations sharing the application's elected writer.
pub struct IndexService<'app> {
    config: &'app Config,
    db: &'app Db,
}

impl<'app> IndexService<'app> {
    pub const fn new(config: &'app Config, db: &'app Db) -> Self {
        Self { config, db }
    }

    pub fn refresh(&self) -> Result<AutoReindexOutcome> {
        indexer::refresh_index_opportunistically(self.config, self.db, None)
    }

    pub fn reindex(&self, full: bool) -> Result<(usize, usize)> {
        let outcome = self.reindex_report(full)?;
        Ok((outcome.files_seen, outcome.sessions_updated))
    }

    pub fn reindex_report(&self, full: bool) -> Result<indexer::ExplicitReindexOutcome> {
        indexer::explicit_reindex_and_migrate(self.config, self.db, full, None)
    }

    /// Report parser/schema freshness and only repairs applicable to discoverable sources.
    pub fn status(&self) -> Result<IndexStatus> {
        crate::diagnostics::index_status(self.config, self.db)
    }
}

#[derive(Clone, Copy)]
/// Typed session listing, search, resolution, and compact inspection operations.
pub struct CatalogService<'db> {
    db: &'db Db,
}

impl<'db> CatalogService<'db> {
    pub const fn new(db: &'db Db) -> Self {
        Self { db }
    }

    /// List filtered sessions in database order.
    ///
    /// # Complexity
    ///
    /// Current list paging uses SQL `OFFSET`: with `S` rows considered, `N` eligible rows,
    /// offset `O`, and positive limit `K`, favorable indexed work is `O(log S + O + K)`; a
    /// planner-selected temporary sort is conservatively `O(S log S + O + K)`. Session-span
    /// overlap adds `O(S)` constant-time comparisons and no message scan. Returned memory is
    /// `O(K + D_K)` for selected row/text bytes. A zero limit intentionally returns the complete
    /// filtered corpus, so `K=N`; SQLite page-cache or temporary-sort memory is additional.
    pub fn list_sessions(&self, filters: &SearchFilters) -> Result<Vec<SessionRecord>> {
        filters.validate()?;
        self.db.list_recent(filters)
    }

    /// List a numeric-offset page in the same `(updated_at DESC, id ASC)` order as
    /// [`CatalogService::list_sessions`]. Intended for bounded protocol adapters and has the same
    /// time/I/O bounds; application-retained output is `O(K + D_K)` for this page.
    pub fn list_sessions_page(
        &self,
        filters: &SearchFilters,
        offset: usize,
    ) -> Result<Vec<SessionRecord>> {
        filters.validate()?;
        self.db.list_recent_page(filters, offset)
    }

    /// Search all structurally eligible session fields with configured relevance scoring.
    ///
    /// # Complexity
    ///
    /// Let `S` be session rows considered, `B` total eligible field/transcript bytes, `N`
    /// eligible sessions, `K` a positive result limit, `D_K` bytes retained in the top-K records,
    /// and `D_max` the largest current
    /// record plus transcript, `A` query tokens, and `P` configured workers. Batched scoring plus
    /// amortized linear top-K compaction does `O(B*(A+1) + N + K log K)` aggregate work; the final
    /// term is the retained-page sort. Peak processing memory is
    /// `O(K + D_K + batch_budget + D_max)`. Session filtering adds an `O(S)` scan bound but
    /// rejects non-overlapping rows before transcript scoring/materialization. Ideal scoring latency is `O(B*(A+1)/P + N/P)` plus
    /// serial SQLite traversal, batch joins, retained-set compaction, and final `O(K log K)` sort.
    /// A zero limit intentionally retains every match, increasing memory to `O(N + B)` and final
    /// sorting to `O(N log N)`; parallel scoring does not change result membership or rank.
    pub fn search_sessions(
        &self,
        query: &str,
        filters: &SearchFilters,
        current_repo: Option<&str>,
        scoring: &ScoringConfig,
    ) -> Result<Vec<SearchHit>> {
        filters.validate()?;
        self.db.search(query, filters, current_repo, scoring)
    }

    /// Crate-internal session-search path with cooperative cancellation between scoring batches
    /// and before retained-set compaction/sort. Public callers keep the stable method above.
    pub(crate) fn search_sessions_cancellable(
        &self,
        query: &str,
        filters: &SearchFilters,
        current_repo: Option<&str>,
        scoring: &ScoringConfig,
        cancellation: &QueryCancellation,
    ) -> Result<Vec<SearchHit>> {
        filters.validate()?;
        self.db
            .search_cancellable(query, filters, current_repo, scoring, cancellation)
    }

    /// Resolve one canonical session ID or unique prefix; ambiguous errors list candidates.
    pub fn resolve_session(&self, id_or_prefix: &str) -> Result<SessionRecord> {
        self.db.resolve_session_record(id_or_prefix)
    }

    /// Build a bounded evidence summary for one resolved session without filesystem writes.
    pub fn inspect(
        &self,
        id_or_prefix: &str,
        options: crate::inspect::InspectionOptions,
    ) -> Result<crate::inspect::SessionInspection> {
        crate::inspect::inspect_session(self.db, id_or_prefix, options)
    }
}

#[derive(Clone, Copy)]
/// Canonical message search, explain, metadata, and context operations used by all adapters.
pub struct MessageService<'db> {
    config: &'db Config,
    db: &'db Db,
    surface: SearchSurface,
}

#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub(crate) enum MessageSearchBatchControl {
    Continue,
    Stop,
}

/// Terminal state from a batched traversal. `page` and `ordered_digest` exist only after natural
/// exhaustion; a stopped consumer has not received a complete requested result set.
pub(crate) struct MessageSearchBatchVisitOutcome {
    pub(crate) exhausted: bool,
    pub(crate) page: Option<PageInfo>,
    pub(crate) planner: Option<SearchExplain>,
    pub(crate) origins: Option<MessageSearchOrigins>,
    pub(crate) ordered_digest: Option<String>,
}

impl<'db> MessageService<'db> {
    pub const fn new(config: &'db Config, db: &'db Db, surface: SearchSurface) -> Self {
        Self {
            config,
            db,
            surface,
        }
    }

    /// Describe the final input catalogue and this service's configured queryless/content defaults.
    ///
    /// The configured overlay is resolved by [`Self::plan`], not by a parallel documentation
    /// algorithm. It performs no session scan or message query and retains only one small request.
    pub fn message_search_spec(&self) -> Result<MessageSearchSpecification> {
        Self::message_search_spec_for_config(self.config, self.surface)
    }

    /// Resolve configured defaults without opening or querying an index.
    ///
    /// MCP tool discovery uses this path so initialize/tools-list remain available while a fresh
    /// index is missing or being prepared. The canonical default request has no session selector;
    /// reaching the resolver is therefore an internal contract error rather than a database read.
    pub(crate) fn message_search_spec_for_config(
        config: &Config,
        surface: SearchSurface,
    ) -> Result<MessageSearchSpecification> {
        let request = MessageSearchRequest::builder(
            crate::message_search::MessageQuery::All,
            MessageTarget::content(),
        )
        .build()?;
        let plan = Self::plan_with_session_resolver(config, surface, request, |_| {
            bail!(
                "configured message-search specification unexpectedly requires session resolution"
            )
        })?;
        Ok(MessageSearchSpecification::new(
            ResolvedMessageSearchRequest::from_plan(&plan)?,
        ))
    }

    /// Open exhaustive batches using this service's configuration, access scope, and surface.
    ///
    /// This is the adapter seam for CLI and other in-process consumers that already own a
    /// configured database handle. It preserves the same access authority on the dedicated
    /// read-only snapshot instead of reopening an unrestricted application.
    pub(crate) fn search_batches(
        &self,
        request: MessageSearchRequest,
        batch_rows: NonZeroUsize,
    ) -> Result<MessageSearchBatches> {
        MessageSearchBatches::spawn(
            self.config.clone(),
            self.db.access_scope().clone(),
            self.surface,
            request,
            batch_rows,
        )
    }

    /// Search individual messages using the selected exact, fuzzy, or regex mode.
    ///
    /// # Complexity
    ///
    /// Let `F` be structurally filtered rows, `C <= F` safe-trigram candidates, `B_F`/`B_C`
    /// their selected-field bytes, `Q` query length, `P_Q` total postings entries read for the
    /// required grams, and `K` the requested page window. Schema-v4 content and tool-argument
    /// literal/regex searches do `O(Q + P_Q + C + B_C)` work including authoritative
    /// verification; worst case remains `O(F + B_F)` when no safe/selective anchor exists.
    /// Tool-argument candidates come from raw JSON, then RFC 6901 projection remains authoritative.
    /// Peak Rust result memory is `O(K + bytes(K))` for finite non-fuzzy pages and `O(F + B_F)`
    /// when `limit == 0`; SQLite's implementation-owned statement/postings state is conservatively
    /// `O(P_Q + C)`. These paths are serial, so latency has the same critical-path bound as work.
    /// Fuzzy search scores every filtered row with parallel batches, does
    /// `O(B_F + F + W log W)` aggregate work for retained window `W=offset+limit`, and has ideal
    /// scoring latency `O(B_F/P + W log W)` over `P` workers plus serial SQLite traversal and
    /// scheduling overhead. Content and derived-field fuzzy paths both use bounded parallel
    /// batches; tool-argument fuzzy projection remains exhaustive but parallelizes JSON parsing
    /// and scoring. Peak Rust memory is `O(W + bytes(W) + batch_budget + largest_row + P*Q)`;
    /// validation rejects an unbounded fuzzy page.
    pub fn search_legacy(&self, query: &str, filters: &MessageFilters) -> Result<Vec<MessageHit>> {
        self.db.search_messages(query, filters)
    }

    pub fn plan(&self, request: MessageSearchRequest) -> Result<MessageSearchPlan> {
        Self::plan_with_session_resolver(self.config, self.surface, request, |value| {
            self.catalog().resolve_session(value)
        })
    }

    fn plan_with_session_resolver(
        config: &Config,
        surface: SearchSurface,
        request: MessageSearchRequest,
        resolve_session: impl FnOnce(&str) -> Result<SessionRecord>,
    ) -> Result<MessageSearchPlan> {
        let purpose = request
            .purpose()
            .map(|selection| -> Result<_> {
                let definition = config
                    .search
                    .purposes
                    .get(selection.name())
                    .ok_or_else(|| MessageSearchError::InvalidParameter {
                        parameter: "purpose",
                        reason: format!("unknown message-search purpose {:?}", selection.name()),
                    })?;
                if selection
                    .version()
                    .is_some_and(|version| version != definition.version)
                {
                    return Err(MessageSearchError::InvalidParameter {
                        parameter: "purpose_version",
                        reason: format!(
                            "purpose {:?} version {} is unavailable; configured version is {}",
                            selection.name(),
                            selection.version().unwrap(),
                            definition.version
                        ),
                    }
                    .into());
                }
                Ok((selection.name().to_string(), definition))
            })
            .transpose()?;
        let purpose_origin = || {
            purpose
                .as_ref()
                .map(|(name, definition)| ValueOrigin::Purpose {
                    name: name.clone(),
                    version: definition.version,
                })
        };
        let purpose_preferences = purpose
            .as_ref()
            .map(|(_, definition)| &definition.preferences);

        let (requested_limit, offset, explicit_all) = match request.extent() {
            crate::message_search::RequestedExtent::Page { limit, offset } => {
                (limit, offset, false)
            }
            crate::message_search::RequestedExtent::AllResults { offset } => (None, offset, true),
        };
        let (limit, limit_origin) = if explicit_all {
            (None, ValueOrigin::Explicit)
        } else if let Some(limit) = requested_limit {
            (Some(limit), ValueOrigin::Explicit)
        } else if let Some(limit) = purpose_preferences.and_then(|value| value.default_limit) {
            (Some(limit), purpose_origin().unwrap())
        } else if let Some(limit) = config.search.message_search.default_limit {
            (Some(limit), ValueOrigin::OperationConfig)
        } else {
            let surface_limit = match surface {
                SearchSurface::Mcp => NonZeroUsize::new(config.mcp.search_messages_limit),
                // Native programmatic/interactive surfaces preserve the complete selected corpus
                // when no operation/purpose/call limit was supplied. MCP alone supplies an
                // implicit finite page because its response is injected directly into model
                // context. Fuzzy validation below still rejects an unbounded resolved extent.
                SearchSurface::Rust | SearchSurface::Cli | SearchSurface::Python => None,
            };
            match surface_limit {
                Some(limit) => (Some(limit), ValueOrigin::SurfaceConfig { surface }),
                None => (None, ValueOrigin::TypedDefault),
            }
        };
        if let (Some(current), Some(maximum)) = (limit, config.search.budgets.max_hits_per_page) {
            if current > maximum {
                return Err(MessageSearchError::InvalidParameter {
                    parameter: "limit",
                    reason: format!(
                        "resolved message-search limit {} exceeds search.budgets.max_hits_per_page {}; lower the request, purpose, operation default, or MCP default",
                        current, maximum
                    ),
                }
                .into());
            }
        }
        let extent = if explicit_all {
            ResolvedExtent::AllResults { offset }
        } else if let Some(limit) = limit {
            ResolvedExtent::Page { limit, offset }
        } else if matches!(
            request.query(),
            crate::message_search::MessageQuery::Fuzzy(_)
        ) {
            return Err(MessageSearchError::InvalidParameter {
                parameter: "limit",
                reason:
                    "fuzzy search requires a positive page size from the request or configuration"
                        .into(),
            }
            .into());
        } else {
            ResolvedExtent::AllResults { offset }
        };

        let (context, context_before_origin, context_after_origin) =
            if let Some(context) = request.context() {
                (context, ValueOrigin::Explicit, ValueOrigin::Explicit)
            } else {
                let (before, before_origin) = if let Some(value) =
                    purpose_preferences.and_then(|preferences| preferences.context_before)
                {
                    (value, purpose_origin().unwrap())
                } else if let Some(value) = config.search.message_search.context.context_before {
                    (value, ValueOrigin::OperationConfig)
                } else {
                    (0, ValueOrigin::TypedDefault)
                };
                let (after, after_origin) = if let Some(value) =
                    purpose_preferences.and_then(|preferences| preferences.context_after)
                {
                    (value, purpose_origin().unwrap())
                } else if let Some(value) = config.search.message_search.context.context_after {
                    (value, ValueOrigin::OperationConfig)
                } else {
                    (0, ValueOrigin::TypedDefault)
                };
                (
                    ContextWindow::new(before, after),
                    before_origin,
                    after_origin,
                )
            };
        if let Some(maximum) = config.search.budgets.max_context_neighbors_per_hit {
            let total = context
                .messages_before()
                .checked_add(context.messages_after())
                .ok_or_else(|| MessageSearchError::InvalidParameter {
                    parameter: "context",
                    reason: "resolved message-search context total overflows".to_string(),
                })?;
            if total > maximum.get() {
                return Err(MessageSearchError::InvalidParameter {
                    parameter: "context",
                    reason: format!(
                        "resolved message-search context total {} exceeds search.budgets.max_context_neighbors_per_hit {}; lower context_before or context_after",
                        total, maximum
                    ),
                }
                .into());
            }
        }

        let (includes, includes_origin) = if let Some(includes) = request.includes() {
            (includes.to_vec(), ValueOrigin::Explicit)
        } else if let Some(value) =
            purpose_preferences.and_then(|preferences| preferences.include_refs)
        {
            (
                value
                    .then_some(MessageSearchInclude::ParsedReferences)
                    .into_iter()
                    .collect(),
                purpose_origin().unwrap(),
            )
        } else if surface == SearchSurface::Mcp {
            (
                vec![MessageSearchInclude::NormalizedSessionMetadata],
                ValueOrigin::SurfaceConfig { surface },
            )
        } else {
            (Vec::new(), ValueOrigin::TypedDefault)
        };
        let include_refs = includes.contains(&MessageSearchInclude::ParsedReferences);
        let detail = request.presentation().detail();
        let detail_origin = if detail.is_some() {
            ValueOrigin::Explicit
        } else {
            ValueOrigin::TypedDefault
        };
        let (message_lines, message_lines_origin) =
            if let Some(value) = request.presentation().message_lines() {
                (value, ValueOrigin::Explicit)
            } else if detail == Some(DetailLevel::Full) {
                (
                    LineWindow::Full,
                    ValueOrigin::DetailPreset {
                        detail: DetailLevel::Full,
                    },
                )
            } else if let Some(value) =
                purpose_preferences.and_then(|preferences| preferences.lines_per_message)
            {
                (LineWindow::from_signed(value)?, purpose_origin().unwrap())
            } else {
                let value = match surface {
                    SearchSurface::Cli => config.cli.lines_per_message,
                    SearchSurface::Mcp => config.mcp.lines_per_message,
                    SearchSurface::Rust | SearchSurface::Python => 0,
                };
                (
                    LineWindow::from_signed(value)?,
                    ValueOrigin::SurfaceConfig { surface },
                )
            };
        let (match_evidence_max_chars, match_evidence_max_chars_origin) = if let Some(value) =
            purpose_preferences.and_then(|preferences| preferences.match_evidence_max_chars)
        {
            (value, purpose_origin().unwrap())
        } else if let Some(value) = config.search.message_search.match_evidence_max_chars {
            (value, ValueOrigin::OperationConfig)
        } else {
            (
                NonZeroUsize::new(DEFAULT_MATCH_EVIDENCE_MAX_CHARS)
                    .expect("typed match evidence default is positive"),
                ValueOrigin::TypedDefault,
            )
        };
        let compact_boundary_chars = match surface {
            SearchSurface::Mcp => config.mcp.preview_chars,
            SearchSurface::Cli => config.cli.evidence_preview_chars,
            SearchSurface::Rust | SearchSurface::Python => DEFAULT_MATCH_EVIDENCE_MAX_CHARS,
        };
        let (field_view, field_view_origin) =
            if let Some(value) = request.presentation().field_view() {
                (value, ValueOrigin::Explicit)
            } else {
                match detail {
                    Some(DetailLevel::Compact) => (
                        FieldViewBudget::MaxChars {
                            max_chars: NonZeroUsize::new(compact_boundary_chars)
                                .expect("validated presentation defaults are positive"),
                        },
                        ValueOrigin::DetailPreset {
                            detail: DetailLevel::Compact,
                        },
                    ),
                    Some(DetailLevel::Full) => (
                        FieldViewBudget::NoCharLimit,
                        ValueOrigin::DetailPreset {
                            detail: DetailLevel::Full,
                        },
                    ),
                    None if surface == SearchSurface::Mcp => (
                        FieldViewBudget::MaxChars {
                            max_chars: NonZeroUsize::new(config.mcp.preview_chars)
                                .expect("validated MCP preview_chars is positive"),
                        },
                        ValueOrigin::SurfaceConfig { surface },
                    ),
                    None => (FieldViewBudget::NoCharLimit, ValueOrigin::TypedDefault),
                }
            };
        let (match_view, match_view_origin) =
            if let Some(value) = request.presentation().match_view() {
                (value, ValueOrigin::Explicit)
            } else {
                match detail {
                    Some(DetailLevel::Full) => (
                        MatchViewBudget::MinimalSpan,
                        ValueOrigin::DetailPreset {
                            detail: DetailLevel::Full,
                        },
                    ),
                    Some(DetailLevel::Compact) => (
                        MatchViewBudget::MaxChars {
                            max_chars: match_evidence_max_chars,
                        },
                        ValueOrigin::DetailPreset {
                            detail: DetailLevel::Compact,
                        },
                    ),
                    None => (
                        MatchViewBudget::MaxChars {
                            max_chars: match_evidence_max_chars,
                        },
                        match_evidence_max_chars_origin.clone(),
                    ),
                }
            };
        let effective_match_evidence_max_chars = match match_view {
            MatchViewBudget::MinimalSpan => match_evidence_max_chars,
            MatchViewBudget::MaxChars { max_chars } => max_chars,
        };
        let (receipt, receipt_origin) = if let Some(value) = request.receipt_level() {
            (value, ValueOrigin::Explicit)
        } else if let Some(value) =
            purpose_preferences.and_then(|preferences| preferences.receipt_level)
        {
            (value, purpose_origin().unwrap())
        } else {
            (ReceiptLevel::None, ValueOrigin::TypedDefault)
        };

        let predicates = request.predicates();
        let resolved_session = predicates.session().map(resolve_session).transpose()?;
        if let (Some(session), Some(providers)) = (&resolved_session, predicates.providers()) {
            if !providers.contains(&session.provider) {
                let selected = providers
                    .iter()
                    .map(|provider| provider.as_str())
                    .collect::<Vec<_>>()
                    .join(", ");
                return Err(MessageSearchError::Conflict(format!(
                    "session_id {:?} resolves to provider {}, outside providers=[{}]; include {} \
                     in providers or remove session_id",
                    predicates.session().unwrap_or_default(),
                    session.provider,
                    selected,
                    session.provider
                ))
                .into());
            }
        }
        let session_id = resolved_session.map(|session| session.id);
        let normalize = |value: &str| crate::util::normalize_path_prefix(value);
        let resolved_predicates = ResolvedMessagePredicates {
            role: predicates.role(),
            kinds: predicates.kinds().map(<[_]>::to_vec),
            providers: predicates.providers().map(<[_]>::to_vec),
            session_id,
            workspace_path_prefix: predicates.workspace_path_prefix().map(normalize),
            transcript_path_prefix: predicates.transcript_path_prefix().map(normalize),
            exclude_workspace_path_prefixes: predicates
                .exclude_workspace_path_prefixes()
                .map(normalize)
                .collect(),
            exclude_transcript_path_prefixes: predicates
                .exclude_transcript_path_prefixes()
                .map(normalize)
                .collect(),
            exclude_session_ids: predicates
                .exclude_session_ids()
                .map(str::to_string)
                .collect(),
            time: predicates.time(),
            sequence: predicates.sequence(),
            tool_name_contains: predicates.tool_name_contains().map(str::to_string),
            include_compaction: predicates.include_compaction(),
        };
        let ordering = if matches!(
            request.query(),
            crate::message_search::MessageQuery::Fuzzy(_)
        ) {
            ExecutionOrder::FuzzyRelevance
        } else {
            ExecutionOrder::SessionSequence
        };
        Ok(MessageSearchPlan {
            retrieval: MessageRetrievalPlan {
                query: request.query().clone(),
                target: request.target().clone(),
                predicates: resolved_predicates,
                match_window: request.match_window(),
                ordering,
                extent,
            },
            response: MessageResponsePlan {
                context,
                includes,
                presentation: ResolvedMessagePresentation {
                    include_refs,
                    message_lines,
                    match_evidence_max_chars: effective_match_evidence_max_chars,
                    detail,
                    field_view,
                    match_view,
                },
            },
            receipt,
            origins: MessageSearchOrigins {
                result_extent: limit_origin,
                context_messages_before: context_before_origin,
                context_messages_after: context_after_origin,
                includes: includes_origin,
                detail: detail_origin,
                lines_per_message: message_lines_origin,
                field_view: field_view_origin,
                match_view: match_view_origin,
                receipt_level: receipt_origin,
                result_order: ValueOrigin::Derived,
            },
        })
    }

    /// Execute the canonical message-search request with surface-specific omitted-limit semantics.
    ///
    /// With no explicit, purpose, or operation limit, Rust/CLI/Python preserve every literal,
    /// regex, or no-text match; MCP uses its configured finite page. Fuzzy search always requires
    /// a finite resolved page. Presentation limits are resolved separately and never affect hit
    /// membership. Planning, membership, evidence, includes, and context share one SQLite read
    /// snapshot, so a concurrent WAL refresh becomes visible only to a later search.
    pub fn search(&self, request: MessageSearchRequest) -> Result<MessageSearchResponse> {
        self.search_cancellable(request, None)
    }

    pub(crate) fn search_cancellable(
        &self,
        request: MessageSearchRequest,
        cancellation: Option<&AtomicBool>,
    ) -> Result<MessageSearchResponse> {
        self.db
            .with_read_snapshot(|| self.search_in_snapshot(request, cancellation, || {}))
    }

    #[cfg(test)]
    fn search_with_after_retrieval_for_test(
        &self,
        request: MessageSearchRequest,
        after_retrieval: impl FnOnce(),
    ) -> Result<MessageSearchResponse> {
        self.db
            .with_read_snapshot(|| self.search_in_snapshot(request, None, after_retrieval))
    }

    fn search_in_snapshot(
        &self,
        request: MessageSearchRequest,
        cancellation: Option<&AtomicBool>,
        after_retrieval: impl FnOnce(),
    ) -> Result<MessageSearchResponse> {
        ensure_message_search_active(cancellation)?;
        let plan = self.plan(request)?;
        let resolved_request = ResolvedMessageSearchRequest::from_plan(&plan)?;
        let include_explain = plan.receipt != ReceiptLevel::None;
        let (mut hits, planner) = self
            .db
            .search_message_plan(&plan.retrieval, include_explain)?;
        let (next_offset, extent) = match plan.retrieval.extent {
            ResolvedExtent::Page { limit, offset } => {
                let has_more = hits.len() > limit.get();
                hits.truncate(limit.get());
                let next_offset = if has_more {
                    Some(
                        offset
                            .checked_add(limit.get())
                            .ok_or_else(|| anyhow!("message page next offset overflows"))?,
                    )
                } else {
                    None
                };
                (next_offset, plan.retrieval.extent)
            }
            ResolvedExtent::AllResults { .. } => (None, plan.retrieval.extent),
        };
        if plan.retrieval.match_window == Some(MatchWindow::Latest) {
            hits.reverse();
        }
        after_retrieval();
        ensure_message_search_active(cancellation)?;
        let runtime_diagnostics = self.runtime_diagnostics_for_plan(&plan)?;
        let (hits, context_windows, included) =
            self.enrich_hits_cancellable(&plan, hits, cancellation, runtime_diagnostics, None)?;
        let origins = (plan.receipt == ReceiptLevel::Full).then(|| plan.origins.clone());
        let match_mode = plan.retrieval.query.mode();
        let match_details = match_mode.map(|mode| (plan.retrieval.target.clone(), mode));
        let response_query = plan.retrieval.query.text().map(str::to_owned);
        let returned = hits.len();
        Ok(
            MessageSearchResponse::new(crate::message_search::MessageSearchResponseParts {
                request: resolved_request,
                match_details,
                hits,
                context_windows,
                page: PageInfo::new(extent, returned, next_offset, plan.retrieval.ordering),
                response: plan.response,
                planner,
                origins,
                included,
            })
            .with_query(response_query),
        )
    }

    /// Visit an exhaustive non-fuzzy request in fully enriched, bounded batches.
    ///
    /// This callback seam is the shared implementation for streaming adapters. The materialized
    /// [`MessageService::search`] API remains the simple default for ordinary callers. This method
    /// retains only one raw/enriched batch plus its active selected-field, context, include, and
    /// presentation bytes; stopping or returning an error drops the SQLite snapshot without
    /// draining unread results.
    #[cfg(test)]
    pub(crate) fn visit_search_batches(
        &self,
        request: MessageSearchRequest,
        batch_rows: NonZeroUsize,
        visitor: impl FnMut(MessageSearchBatch) -> Result<MessageSearchBatchControl>,
    ) -> Result<MessageSearchBatchVisitOutcome> {
        let plan = self.plan(request)?;
        self.visit_search_plan_batches(plan, batch_rows, None, visitor)
    }

    pub(crate) fn visit_search_plan_batches(
        &self,
        plan: MessageSearchPlan,
        batch_rows: NonZeroUsize,
        cancellation: Option<&AtomicBool>,
        mut visitor: impl FnMut(MessageSearchBatch) -> Result<MessageSearchBatchControl>,
    ) -> Result<MessageSearchBatchVisitOutcome> {
        Self::validate_batch_plan(&plan)?;
        ensure_message_search_active(cancellation)?;

        let include_explain = plan.receipt != ReceiptLevel::None;
        let match_mode = plan.retrieval.query.mode();
        let mut digest = MessageSearchOrderedDigest::new(plan.retrieval.target.clone(), match_mode);
        let mut emitted = 0_usize;
        let mut emitted_session_metadata = BTreeSet::new();
        let visit = self.db.visit_message_plan_batches(
            &plan.retrieval,
            include_explain,
            batch_rows,
            |raw_hits| {
                ensure_message_search_active(cancellation)?;
                let (results, context_windows, included) = self.enrich_hits_cancellable(
                    &plan,
                    raw_hits,
                    cancellation,
                    None,
                    Some(&mut emitted_session_metadata),
                )?;
                for result in &results {
                    digest.update(result);
                }
                emitted = emitted
                    .checked_add(results.len())
                    .ok_or_else(|| anyhow!("batched message-search emitted count overflows"))?;
                let control = visitor(MessageSearchBatch {
                    results,
                    context_windows,
                    included,
                })?;
                Ok(match control {
                    MessageSearchBatchControl::Continue => MessageBatchControl::Continue,
                    MessageSearchBatchControl::Stop => MessageBatchControl::Stop,
                })
            },
        )?;
        debug_assert_eq!(visit.rows_visited, emitted);
        let exhausted = visit.exhausted;
        let page = exhausted.then(|| {
            PageInfo::new(
                plan.retrieval.extent,
                emitted,
                None,
                plan.retrieval.ordering,
            )
        });
        let origins =
            (exhausted && plan.receipt == ReceiptLevel::Full).then(|| plan.origins.clone());
        let ordered_digest =
            (exhausted && plan.receipt == ReceiptLevel::Full).then(|| digest.finish());
        Ok(MessageSearchBatchVisitOutcome {
            exhausted,
            page,
            planner: visit.explain,
            origins,
            ordered_digest,
        })
    }

    fn validate_batch_selection(
        all_results: bool,
        fuzzy: bool,
    ) -> std::result::Result<(), MessageSearchError> {
        if !all_results {
            return Err(MessageSearchError::InvalidParameter {
                parameter: "all_results",
                reason: "batched message-search traversal requires all_results; use search() for a finite materialized page".into(),
            });
        }
        if fuzzy {
            return Err(MessageSearchError::InvalidParameter {
                parameter: "query_mode",
                reason: "batched message-search traversal supports literal, regex, and queryless all-results requests; pass a positive limit to search() for fuzzy results".into(),
            });
        }
        Ok(())
    }

    pub(crate) fn validate_batch_request(
        surface: SearchSurface,
        request: &MessageSearchRequest,
    ) -> std::result::Result<(), MessageSearchError> {
        let resolves_exhaustively_without_a_surface_limit =
            matches!(
                request.extent(),
                crate::message_search::RequestedExtent::Page { limit: None, .. }
            ) && !matches!(surface, SearchSurface::Mcp);
        Self::validate_batch_selection(
            matches!(
                request.extent(),
                crate::message_search::RequestedExtent::AllResults { .. }
            ) || resolves_exhaustively_without_a_surface_limit,
            matches!(
                request.query(),
                crate::message_search::MessageQuery::Fuzzy(_)
            ),
        )
    }

    pub(crate) fn validate_batch_plan(
        plan: &MessageSearchPlan,
    ) -> std::result::Result<(), MessageSearchError> {
        Self::validate_batch_selection(
            matches!(plan.retrieval.extent, ResolvedExtent::AllResults { .. }),
            matches!(
                plan.retrieval.query,
                crate::message_search::MessageQuery::Fuzzy(_)
            ),
        )
    }

    pub(crate) fn runtime_diagnostics_for_plan(
        &self,
        plan: &MessageSearchPlan,
    ) -> Result<Option<MessageSearchRuntimeDiagnostics>> {
        plan.response
            .includes
            .contains(&MessageSearchInclude::RuntimeDiagnostics)
            .then(|| {
                serde_json::to_vec(self.config).map(|bytes| {
                    MessageSearchRuntimeDiagnostics::new(
                        self.surface,
                        format!("sha256:{}", crate::hashing::sha256(&bytes)),
                    )
                })
            })
            .transpose()
            .map_err(Into::into)
    }

    fn enrich_hits_cancellable(
        &self,
        plan: &MessageSearchPlan,
        hits: Vec<MessageHit>,
        cancellation: Option<&AtomicBool>,
        runtime_diagnostics: Option<MessageSearchRuntimeDiagnostics>,
        emitted_session_metadata: Option<&mut BTreeSet<String>>,
    ) -> Result<(
        Vec<crate::message_search::MessageSearchHit>,
        Vec<Vec<MessageHit>>,
        MessageSearchIncludedData,
    )> {
        ensure_message_search_active(cancellation)?;
        let mut hits = attach_match_evidence_cancellable(
            &plan.retrieval.query,
            &plan.retrieval.target,
            plan.response.presentation.match_view,
            hits,
            || ensure_message_search_active(cancellation),
        )?;
        ensure_message_search_active(cancellation)?;
        apply_message_presentation_cancellable(
            &plan.retrieval.target,
            plan.response.presentation,
            &mut hits,
            || ensure_message_search_active(cancellation),
        )?;
        if plan
            .response
            .includes
            .contains(&MessageSearchInclude::ParsedReferences)
        {
            for hit in &mut hits {
                ensure_message_search_active(cancellation)?;
                hit.set_parsed_references(crate::refs::extract_refs_from_text(
                    &hit.message().content,
                    hit.message().tool_name.as_deref(),
                ));
            }
        }
        ensure_message_search_active(cancellation)?;
        let include_normalized_metadata = plan
            .response
            .includes
            .contains(&MessageSearchInclude::NormalizedSessionMetadata);
        let include_raw_metadata = plan
            .response
            .includes
            .contains(&MessageSearchInclude::RawProviderMetadata);
        let deduplicate_session_metadata = emitted_session_metadata.is_some();
        let session_ids = if include_normalized_metadata || include_raw_metadata {
            let already_emitted = emitted_session_metadata.as_deref();
            hits.iter()
                .map(|hit| hit.message().session_id.clone())
                .filter(|session_id| {
                    already_emitted.is_none_or(|emitted| !emitted.contains(session_id))
                })
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect::<Vec<_>>()
        } else {
            Vec::new()
        };
        let emit_session_metadata = !deduplicate_session_metadata || !session_ids.is_empty();
        let normalized_session_metadata = (include_normalized_metadata && emit_session_metadata)
            .then(|| {
                self.db
                    .session_metadata(&session_ids)
                    .map(|metadata| metadata.into_iter().collect::<BTreeMap<_, _>>())
            })
            .transpose()?;
        ensure_message_search_active(cancellation)?;
        let raw_provider_metadata = (include_raw_metadata && emit_session_metadata)
            .then(|| self.db.session_raw_provider_metadata(&session_ids))
            .transpose()?;
        ensure_message_search_active(cancellation)?;
        let included = MessageSearchIncludedData::new(
            normalized_session_metadata,
            raw_provider_metadata,
            runtime_diagnostics,
        );
        if let Some(emitted) = emitted_session_metadata {
            emitted.extend(session_ids);
        }
        let context_windows = if plan.response.context.messages_before() == 0
            && plan.response.context.messages_after() == 0
        {
            Vec::new()
        } else {
            let before = i64::try_from(plan.response.context.messages_before())
                .map_err(|_| anyhow!("resolved context_before exceeds SQLite's signed range"))?;
            let after = i64::try_from(plan.response.context.messages_after())
                .map_err(|_| anyhow!("resolved context_after exceeds SQLite's signed range"))?;
            let anchors = hits
                .iter()
                .map(|hit| (hit.message.session_id.clone(), hit.message.seq))
                .collect::<Vec<_>>();
            self.db.message_context_windows(&anchors, before, after)?
        };
        ensure_message_search_active(cancellation)?;
        Ok((hits, context_windows, included))
    }

    fn catalog(&self) -> CatalogService<'db> {
        CatalogService::new(self.db)
    }

    /// Run the same search as [`MessageService::search`] and optionally return its actual planner
    /// receipt, including corpus size and any exact/regex prefilter decision.
    pub fn search_with_explain(
        &self,
        query: &str,
        filters: &MessageFilters,
        include_explain: bool,
    ) -> Result<(Vec<MessageHit>, Option<SearchExplain>)> {
        self.db
            .search_messages_with_explain(query, filters, include_explain)
    }

    /// Load compact metadata for the requested canonical IDs; absent IDs are omitted.
    pub fn session_metadata(&self, session_ids: &[String]) -> Result<HashMap<String, SessionMeta>> {
        self.db.session_metadata(session_ids)
    }

    /// Return a bounded sequence window around one message.
    ///
    /// # Complexity
    ///
    /// The `(session_id, seq)` index makes database work and memory proportional to
    /// `before + after + 1`, subject to messages available in the session.
    pub fn context(
        &self,
        session_id: &str,
        seq: i64,
        before: i64,
        after: i64,
    ) -> Result<Vec<MessageHit>> {
        self.db.message_context(session_id, seq, before, after)
    }

    /// Read one session's messages, selecting the oldest or newest `filters.limit` by `order`,
    /// always returned in chronological (seq-ascending) order. Wraps
    /// [`Db::read_session_messages`]; `filters.session_id` must be set.
    pub fn read_session(
        &self,
        filters: &MessageFilters,
        order: crate::db::MessageOrder,
    ) -> Result<Vec<MessageHit>> {
        self.db.read_session_messages(filters, order)
    }
}

#[derive(Clone, Copy)]
/// Read-only file activity, causal history, reconstruction, and safe publication operations.
pub struct FileService<'db> {
    db: &'db Db,
}

impl<'db> FileService<'db> {
    pub const fn new(db: &'db Db) -> Self {
        Self { db }
    }

    /// Aggregate matching file-edit rows.
    ///
    /// # Complexity
    ///
    /// Work is proportional to filtered edit rows; returned memory is proportional to grouped
    /// files and is unbounded only when the query explicitly uses a zero limit.
    pub fn search(&self, query: &FileQuery) -> Result<Vec<FileEditSummary>> {
        self.db.file_search(query)
    }

    /// Return deterministic file/session edit-count pages using `query.limit` and `query.offset`.
    pub fn cross_reference(&self, query: &FileQuery) -> Result<Vec<FileCrossRef>> {
        self.db.file_cross_ref(query)
    }

    /// Return causally ordered edits for one selected file.
    ///
    /// # Complexity
    ///
    /// Reconstruction work is proportional to matching versions and their stored edit payloads;
    /// returned memory is bounded by `query.limit` unless it is explicitly zero. `query.offset`
    /// skips versions after stable session/version ordering.
    pub fn history(&self, file: &str, query: &FileQuery) -> Result<Vec<FileVersion>> {
        crate::files::history(self.db, file, query)
    }

    /// Replay edits through one requested version.
    ///
    /// # Complexity
    ///
    /// Time is proportional to replayed edit operations and content bytes; peak memory includes
    /// the reconstructed file plus the selected edit history.
    pub fn reconstruct(
        &self,
        file: &str,
        query: &FileQuery,
        version: Option<usize>,
    ) -> Result<crate::files::ReconstructedFile> {
        crate::files::reconstruct_query(self.db, file, query, version)
    }

    /// Lazily reconstruct every causally ordered version with a complete replay path.
    pub fn reconstruct_versions(
        &self,
        file: &str,
        query: &FileQuery,
    ) -> Result<crate::files::ReconstructedFileVersions> {
        crate::files::reconstruct_versions_query(self.db, file, query)
    }

    /// Atomically publish every reconstructable version to a new non-replacing directory.
    pub fn publish_versions(
        &self,
        file: &str,
        query: &FileQuery,
        destination: &std::path::Path,
    ) -> Result<crate::files::RecoveryPublicationReceipt> {
        let versions = self.reconstruct_versions(file, query)?;
        crate::files::publish_reconstructed_versions(versions, destination)
    }

    /// Collision-safely write a reconstructed file through the shared recovery policy.
    pub fn restore(
        &self,
        reconstructed: &crate::files::ReconstructedFile,
        output_dir: Option<&std::path::Path>,
    ) -> Result<std::path::PathBuf> {
        crate::files::restore_reconstructed(reconstructed, output_dir)
    }
}

#[cfg(test)]
mod message_search_service_tests {
    use std::num::{NonZeroU32, NonZeroUsize};
    use std::sync::{Arc, Barrier};

    use super::*;
    use crate::config::{MessagePurposePreferences, PurposeDefinition, SearchOperation};
    use crate::message_search::{
        MessageQuery, MessageSearchRequestBuilder, MessageTarget, PurposeSelection,
        RequestedExtent, ResolvedExtent,
    };
    use crate::models::{Message, MessageKind, Provider, Role};
    use crate::util::minimal_record;

    fn disposable_db() -> (tempfile::TempDir, Db) {
        let directory = tempfile::tempdir().unwrap();
        let db = Db::open(&directory.path().join("message-search.db")).unwrap();
        db.mark_schema_current().unwrap();
        (directory, db)
    }

    fn literal_request() -> MessageSearchRequestBuilder {
        MessageSearchRequest::builder(
            MessageQuery::literal("needle").unwrap(),
            MessageTarget::content(),
        )
    }

    #[test]
    fn message_search_spec_uses_the_real_surface_and_operation_default_planner() {
        let (_directory, db) = disposable_db();
        let mut config = Config::default();
        config.mcp.search_messages_limit = 9;

        for surface in [
            SearchSurface::Rust,
            SearchSurface::Cli,
            SearchSurface::Python,
        ] {
            let spec = MessageService::new(&config, &db, surface)
                .message_search_spec()
                .unwrap();
            assert_eq!(
                spec.configured_default().extent(),
                crate::message_search::ResolvedRequestExtent::AllResults { offset: 0 }
            );
            assert_eq!(
                spec.registry(),
                crate::message_search::MessageSearchParameterRegistry::current()
            );
        }
        let mcp = MessageService::new(&config, &db, SearchSurface::Mcp)
            .message_search_spec()
            .unwrap();
        assert_eq!(
            mcp.configured_default().extent(),
            crate::message_search::ResolvedRequestExtent::Page {
                limit: NonZeroUsize::new(9).unwrap(),
                offset: 0,
            }
        );

        config.search.message_search.default_limit = NonZeroUsize::new(4);
        config.search.message_search.context.context_before = Some(2);
        config.search.message_search.context.context_after = Some(3);
        for surface in SearchSurface::ALL {
            let spec = MessageService::new(&config, &db, *surface)
                .message_search_spec()
                .unwrap();
            assert_eq!(
                spec.configured_default().extent(),
                crate::message_search::ResolvedRequestExtent::Page {
                    limit: NonZeroUsize::new(4).unwrap(),
                    offset: 0,
                }
            );
            assert_eq!(
                spec.configured_default().context(),
                ContextWindow::new(2, 3)
            );
        }
    }

    fn limit_of(plan: &MessageSearchPlan) -> Option<usize> {
        match plan.extent() {
            ResolvedExtent::Page { limit, .. } => Some(limit.get()),
            ResolvedExtent::AllResults { .. } => None,
        }
    }

    fn insert_session(
        db: &Db,
        provider: Provider,
        provider_session_id: &str,
        workspace: &str,
        transcript: &str,
        contents: &[&str],
    ) {
        let mut parsed = minimal_record(provider, std::path::Path::new(transcript), String::new());
        parsed.session.id = format!("{}:{provider_session_id}", provider.as_str());
        parsed.session.provider_session_id = provider_session_id.to_string();
        parsed.session.cwd = Some(workspace.into());
        parsed.session.repo_root = Some(format!("{workspace}/repo"));
        parsed.session.source_path = transcript.into();
        parsed.session.raw_metadata_json =
            Some(format!(r#"{{"fixture_provider":"{}"}}"#, provider.as_str()));
        parsed.messages = contents
            .iter()
            .enumerate()
            .map(|(sequence, content)| Message {
                seq: sequence as i64,
                role: Role::User,
                ts: None,
                tool_name: None,
                kind: MessageKind::Conversation,
                tool_call_id: None,
                is_compaction: false,
                content: (*content).into(),
                provenance: Default::default(),
            })
            .collect();
        db.upsert_session(&parsed, 0, 0).unwrap();
    }

    fn existing_only_search_app(
        provider: Provider,
        provider_session_id: &str,
        contents: &[&str],
    ) -> (tempfile::TempDir, SessionSearch) {
        let directory = tempfile::tempdir().unwrap();
        let db_path = directory.path().join("message-search.db");
        let db = Db::open(&db_path).unwrap();
        db.mark_schema_current().unwrap();
        insert_session(
            &db,
            provider,
            provider_session_id,
            "/workspace/message-search",
            "/transcripts/message-search.jsonl",
            contents,
        );
        drop(db);
        let mut config = Config::default();
        config.index.db_path = Some(db_path.to_string_lossy().into_owned());
        config.index.cache_dir = Some(
            directory
                .path()
                .join("cache")
                .to_string_lossy()
                .into_owned(),
        );
        config.index.refresh = IndexRefresh::ExistingOnly;
        let app = SessionSearch::open(config).unwrap();
        (directory, app)
    }

    #[test]
    fn omitted_limits_keep_rust_cli_and_python_unbounded_while_mcp_stays_bounded() {
        let (_directory, db) = disposable_db();
        let config = Config::default();
        let request = literal_request()
            .extent(RequestedExtent::page(None, 7).unwrap())
            .build()
            .unwrap();

        for surface in [
            SearchSurface::Rust,
            SearchSurface::Cli,
            SearchSurface::Python,
        ] {
            let plan = MessageService::new(&config, &db, surface)
                .plan(request.clone())
                .unwrap();
            assert_eq!(plan.extent(), ResolvedExtent::AllResults { offset: 7 });
            assert_eq!(plan.origins().result_extent(), &ValueOrigin::TypedDefault);
        }

        let mcp = MessageService::new(&config, &db, SearchSurface::Mcp)
            .plan(request.clone())
            .unwrap();
        assert_eq!(limit_of(&mcp), Some(config.mcp.search_messages_limit));
        assert_eq!(
            mcp.origins().result_extent(),
            &ValueOrigin::SurfaceConfig {
                surface: SearchSurface::Mcp,
            }
        );

        let fuzzy = MessageSearchRequest::builder(
            MessageQuery::fuzzy("needle").unwrap(),
            MessageTarget::content(),
        )
        .build()
        .unwrap();
        for surface in [
            SearchSurface::Rust,
            SearchSurface::Cli,
            SearchSurface::Python,
        ] {
            assert!(
                MessageService::new(&config, &db, surface)
                    .plan(fuzzy.clone())
                    .unwrap_err()
                    .to_string()
                    .contains("requires a positive page size"),
                "{surface:?} fuzzy search must not acquire an implicit non-MCP page"
            );
        }
        assert_eq!(
            limit_of(
                &MessageService::new(&config, &db, SearchSurface::Mcp)
                    .plan(fuzzy)
                    .unwrap()
            ),
            Some(config.mcp.search_messages_limit)
        );
    }

    #[test]
    fn explicit_all_results_truthfully_bypasses_only_the_finite_page_ceiling() {
        let (_directory, db) = disposable_db();
        let mut config = Config::default();
        config.search.budgets.max_hits_per_page = NonZeroUsize::new(50);
        let request = literal_request()
            .extent(RequestedExtent::all_results())
            .build()
            .unwrap();

        let plan = MessageService::new(&config, &db, SearchSurface::Rust)
            .plan(request)
            .expect("an explicit all_results request is not a finite page");
        assert_eq!(plan.extent(), ResolvedExtent::AllResults { offset: 0 });
        assert_eq!(
            plan.origins().result_extent(),
            &ValueOrigin::Explicit,
            "the receipt must expose the explicit bypass, not an unused default page size"
        );
    }

    #[test]
    fn planner_precedence_is_explicit_and_policy_conflicts_are_rejected() {
        let (_directory, db) = disposable_db();
        let mut config = Config::default();
        config.search.message_search.default_limit = NonZeroUsize::new(8);
        config.search.message_search.context.context_before = Some(2);
        config.search.purposes.insert(
            "focused-review".into(),
            PurposeDefinition {
                version: NonZeroU32::new(1).unwrap(),
                operation: SearchOperation::MessageSearch,
                preferences: MessagePurposePreferences {
                    default_limit: NonZeroUsize::new(6),
                    context_before: Some(3),
                    context_after: Some(4),
                    receipt_level: Some(ReceiptLevel::Summary),
                    include_refs: Some(true),
                    lines_per_message: Some(-5),
                    match_evidence_max_chars: NonZeroUsize::new(90),
                },
            },
        );
        let purpose = PurposeSelection::new("focused-review", NonZeroU32::new(1)).unwrap();
        let service = MessageService::new(&config, &db, SearchSurface::Mcp);

        let purpose_plan = service
            .plan(literal_request().purpose(purpose.clone()).build().unwrap())
            .unwrap();
        assert_eq!(limit_of(&purpose_plan), Some(6));
        assert!(matches!(
            purpose_plan.origins().result_extent(),
            ValueOrigin::Purpose { name, version }
                if name == "focused-review" && version.get() == 1
        ));
        assert_eq!(purpose_plan.context(), ContextWindow::new(3, 4));
        assert!(purpose_plan.presentation().include_refs());
        assert_eq!(
            purpose_plan.presentation().message_lines(),
            LineWindow::Tail(NonZeroUsize::new(5).unwrap())
        );
        assert_eq!(
            purpose_plan.presentation().match_evidence_max_chars().get(),
            90
        );
        assert!(matches!(
            purpose_plan.origins().match_view(),
            ValueOrigin::Purpose { name, version }
                if name == "focused-review" && version.get() == 1
        ));
        assert_eq!(purpose_plan.receipt_level(), ReceiptLevel::Summary);

        let explicit = service
            .plan(
                literal_request()
                    .purpose(purpose)
                    .extent(RequestedExtent::page(Some(3), 0).unwrap())
                    .context(ContextWindow::new(9, 10))
                    .includes([])
                    .message_lines(LineWindow::Head(NonZeroUsize::new(2).unwrap()))
                    .match_view(MatchViewBudget::max_chars(44).unwrap())
                    .receipt_level(ReceiptLevel::Full)
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(limit_of(&explicit), Some(3));
        assert_eq!(explicit.origins().result_extent(), &ValueOrigin::Explicit);
        assert_eq!(
            explicit.origins().context_messages_before(),
            &ValueOrigin::Explicit
        );
        assert_eq!(explicit.origins().receipt_level(), &ValueOrigin::Explicit);
        assert_eq!(explicit.presentation().match_evidence_max_chars().get(), 44);
        assert_eq!(explicit.origins().match_view(), &ValueOrigin::Explicit);

        config.search.budgets.max_hits_per_page = NonZeroUsize::new(4);
        let limit_error = MessageService::new(&config, &db, SearchSurface::Mcp)
            .plan(literal_request().build().unwrap())
            .unwrap_err()
            .to_string();
        assert!(limit_error.contains("max_hits_per_page"), "{limit_error}");
        assert!(limit_error.contains("8"), "{limit_error}");
        assert!(limit_error.contains("4"), "{limit_error}");

        config.search.budgets.max_hits_per_page = None;
        config.search.budgets.max_context_neighbors_per_hit = NonZeroUsize::new(4);
        let context_error = MessageService::new(&config, &db, SearchSurface::Mcp)
            .plan(
                literal_request()
                    .context(ContextWindow::new(9, 10))
                    .build()
                    .unwrap(),
            )
            .unwrap_err()
            .to_string();
        assert!(
            context_error.contains("max_context_neighbors_per_hit"),
            "{context_error}"
        );
        assert!(context_error.contains("19"), "{context_error}");
        assert!(context_error.contains("4"), "{context_error}");
    }

    #[test]
    fn planner_resolves_unique_session_prefix_and_rejects_ambiguous_prefix() {
        let (_directory, db) = disposable_db();
        insert_session(
            &db,
            Provider::Claude,
            "abcdef",
            "/workspace/a",
            "/transcripts/a.jsonl",
            &["needle"],
        );
        insert_session(
            &db,
            Provider::Claude,
            "abcxyz",
            "/workspace/b",
            "/transcripts/b.jsonl",
            &["needle"],
        );
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);

        let unique = service
            .plan(
                literal_request()
                    .session_id("claude:abcd")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(
            unique.retrieval.predicates.session_id.as_deref(),
            Some("claude:abcdef")
        );

        assert!(service
            .plan(
                literal_request()
                    .session_id("claude:abc")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap_err()
            .to_string()
            .contains("ambiguous"));
    }

    #[test]
    fn provider_sets_filter_results_and_reject_a_session_outside_the_selected_sources() {
        let (_directory, db) = disposable_db();
        for (provider, id) in [
            (Provider::Claude, "provider-set-claude"),
            (Provider::Codex, "provider-set-codex"),
            (Provider::GeminiCli, "provider-set-gemini"),
        ] {
            insert_session(
                &db,
                provider,
                id,
                "/workspace/providers",
                &format!("/transcripts/{id}.jsonl"),
                &["shared provider-set needle"],
            );
        }
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);
        let request = literal_request()
            .providers(vec![Provider::Codex, Provider::Claude, Provider::Codex])
            .unwrap()
            .extent(RequestedExtent::all_results())
            .build()
            .unwrap();
        let response = service.search(request).unwrap();
        assert_eq!(
            response
                .hits()
                .iter()
                .map(|hit| hit.message().provider)
                .collect::<Vec<_>>(),
            [Provider::Claude, Provider::Codex]
        );
        assert_eq!(
            response.request().provider_scope(),
            &crate::message_search::ProviderScope::Selected {
                providers: vec![Provider::Claude, Provider::Codex]
            }
        );

        let error = service
            .plan(
                literal_request()
                    .providers(vec![Provider::Codex])
                    .unwrap()
                    .session_id("claude:provider-set-claude")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap_err()
            .to_string();
        assert!(error.contains("session_id"), "{error}");
        assert!(error.contains("provider claude"), "{error}");
        assert!(error.contains("providers=[codex]"), "{error}");
        assert!(error.contains("include claude"), "{error}");
    }

    #[test]
    fn bounded_batches_match_the_simple_materialized_response_and_stop_without_completion() {
        let (_directory, db) = disposable_db();
        insert_session(
            &db,
            Provider::Claude,
            "bounded-service",
            "/workspace/bounded",
            "/transcripts/bounded.jsonl",
            &[
                "needle zero https://example.com/zero",
                "context one",
                "needle two https://example.com/two",
                "needle three",
                "needle four",
            ],
        );
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);
        let request = literal_request()
            .session_id("claude:bounded-service")
            .unwrap()
            .extent(RequestedExtent::all_results())
            .context(ContextWindow::new(1, 1))
            .includes([
                MessageSearchInclude::NormalizedSessionMetadata,
                MessageSearchInclude::ParsedReferences,
            ])
            .receipt_level(ReceiptLevel::Full)
            .build()
            .unwrap();
        let materialized = service.search(request.clone()).unwrap();

        let mut batch_lengths = Vec::new();
        let mut sequences = Vec::new();
        let mut field_views = Vec::new();
        let mut context_sequences = Vec::new();
        let mut parsed_reference_counts = Vec::new();
        let mut included_sessions = BTreeSet::new();
        let outcome =
            service
                .visit_search_batches(request.clone(), NonZeroUsize::new(2).unwrap(), |batch| {
                    batch_lengths.push(batch.results.len());
                    sequences.extend(batch.results.iter().map(|result| result.seq));
                    field_views.extend(
                        batch
                            .results
                            .iter()
                            .map(|result| result.field_view().text().to_string()),
                    );
                    parsed_reference_counts.extend(
                        batch
                            .results
                            .iter()
                            .map(|result| result.parsed_references().map_or(0, <[_]>::len)),
                    );
                    context_sequences.extend(batch.context_windows.iter().map(|window| {
                        window.iter().map(|message| message.seq).collect::<Vec<_>>()
                    }));
                    if let Some(metadata) = batch.included.normalized_session_metadata() {
                        included_sessions.extend(metadata.keys().cloned());
                    }
                    Ok(MessageSearchBatchControl::Continue)
                })
                .unwrap();

        assert_eq!(batch_lengths, vec![2, 2]);
        assert_eq!(
            sequences,
            materialized
                .results()
                .iter()
                .map(|result| result.seq)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            field_views,
            materialized
                .results()
                .iter()
                .map(|result| result.field_view().text().to_string())
                .collect::<Vec<_>>()
        );
        assert_eq!(
            parsed_reference_counts,
            materialized
                .results()
                .iter()
                .map(|result| result.parsed_references().map_or(0, <[_]>::len))
                .collect::<Vec<_>>()
        );
        assert_eq!(
            context_sequences,
            materialized
                .context_windows()
                .iter()
                .map(|window| window.iter().map(|message| message.seq).collect::<Vec<_>>())
                .collect::<Vec<_>>()
        );
        assert_eq!(
            included_sessions,
            BTreeSet::from(["claude:bounded-service".to_string()])
        );
        assert!(outcome.exhausted);
        assert_eq!(
            outcome.page.expect("natural exhaustion has page metadata"),
            materialized.page()
        );
        assert_eq!(
            outcome.ordered_digest.as_deref(),
            Some(materialized.ordered_digest().as_str())
        );
        assert!(outcome.planner.is_some());
        assert!(outcome.origins.is_some());

        let mut stopped_sequences = Vec::new();
        let stopped = service
            .visit_search_batches(request, NonZeroUsize::new(2).unwrap(), |batch| {
                stopped_sequences.extend(batch.results.iter().map(|result| result.seq));
                Ok(MessageSearchBatchControl::Stop)
            })
            .unwrap();
        assert_eq!(stopped_sequences, vec![0, 2]);
        assert!(!stopped.exhausted);
        assert!(stopped.page.is_none());
        assert!(stopped.planner.is_none());
        assert!(stopped.origins.is_none());
        assert!(stopped.ordered_digest.is_none());
    }

    #[test]
    fn bounded_latest_offset_matches_the_materialized_chronological_result() {
        let (_directory, db) = disposable_db();
        insert_session(
            &db,
            Provider::Claude,
            "bounded-latest",
            "/workspace/bounded-latest",
            "/transcripts/bounded-latest.jsonl",
            &[
                "needle zero",
                "needle one",
                "needle two",
                "needle three",
                "needle four",
            ],
        );
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);
        let request = literal_request()
            .session_id("claude:bounded-latest")
            .unwrap()
            .match_window(MatchWindow::Latest)
            .extent(RequestedExtent::all_results_from(2))
            .receipt_level(ReceiptLevel::Full)
            .build()
            .unwrap();
        let materialized = service.search(request.clone()).unwrap();
        let mut streamed = Vec::new();
        let outcome = service
            .visit_search_batches(request, NonZeroUsize::new(1).unwrap(), |batch| {
                streamed.extend(batch.results.into_iter().map(|result| result.seq));
                Ok(MessageSearchBatchControl::Continue)
            })
            .unwrap();

        assert_eq!(streamed, vec![0, 1, 2]);
        assert_eq!(
            streamed,
            materialized
                .results()
                .iter()
                .map(|result| result.seq)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            outcome.ordered_digest.as_deref(),
            Some(materialized.ordered_digest().as_str())
        );
    }

    #[test]
    fn exhaustive_batches_match_the_simple_list_and_own_terminal_state() {
        let (_directory, app) = existing_only_search_app(
            Provider::Codex,
            "batch-fixture",
            &["needle zero", "context one", "needle two", "needle three"],
        );
        let request = literal_request()
            .session_id("codex:batch-fixture")
            .unwrap()
            .extent(RequestedExtent::all_results())
            .context(ContextWindow::new(1, 1))
            .includes([
                MessageSearchInclude::NormalizedSessionMetadata,
                MessageSearchInclude::RawProviderMetadata,
                MessageSearchInclude::RuntimeDiagnostics,
            ])
            .receipt_level(ReceiptLevel::Full)
            .build()
            .unwrap();
        let materialized = app.messages().search(request.clone()).unwrap();
        let mut batches = app
            .message_search_batches(request, NonZeroUsize::new(2).unwrap())
            .unwrap();
        assert_eq!(batches.request().receipt_level(), ReceiptLevel::Full);
        assert_eq!(
            batches
                .runtime_diagnostics()
                .expect("requested request-wide diagnostics")
                .config_digest(),
            materialized
                .included()
                .runtime_diagnostics()
                .expect("materialized response diagnostics")
                .config_digest()
        );
        assert!(batches
            .completion()
            .unwrap_err()
            .to_string()
            .contains("unread results"));
        let resolved_request = batches.request().clone();
        let mut sequences = Vec::new();
        let mut contexts = Vec::new();
        let mut serialized_results = Vec::new();
        let mut normalized_metadata_batches = 0;
        let mut raw_metadata_batches = 0;
        while let Some(batch) = batches.next_batch().unwrap() {
            assert!(batch.results().len() <= 2);
            assert!(batch.included().runtime_diagnostics().is_none());
            if let Some(metadata) = batch.included().normalized_session_metadata() {
                normalized_metadata_batches += 1;
                assert_eq!(metadata.len(), 1);
                assert!(metadata.contains_key("codex:batch-fixture"));
            }
            if let Some(metadata) = batch.included().raw_provider_metadata() {
                raw_metadata_batches += 1;
                assert_eq!(metadata.len(), 1);
                assert!(metadata.contains_key("codex:batch-fixture"));
            }
            serialized_results.extend((0..batch.results().len()).map(|index| {
                serde_json::to_value(
                    batch
                        .result_document(&resolved_request, index)
                        .expect("index comes from the batch result length"),
                )
                .unwrap()
            }));
            sequences.extend(batch.results().iter().map(|result| result.seq));
            contexts.extend(
                batch
                    .context_windows()
                    .iter()
                    .map(|window| window.iter().map(|message| message.seq).collect::<Vec<_>>()),
            );
        }
        let completion = batches.completion().unwrap();
        assert_eq!(normalized_metadata_batches, 1);
        assert_eq!(raw_metadata_batches, 1);
        assert_eq!(
            sequences,
            materialized
                .results()
                .iter()
                .map(|result| result.seq)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            contexts,
            materialized
                .context_windows()
                .iter()
                .map(|window| window.iter().map(|message| message.seq).collect::<Vec<_>>())
                .collect::<Vec<_>>()
        );
        assert_eq!(completion.page(), materialized.page());
        assert_eq!(
            serialized_results,
            (0..materialized.results().len())
                .map(|index| {
                    serde_json::to_value(materialized.result_document(index).unwrap()).unwrap()
                })
                .collect::<Vec<_>>()
        );
        assert_eq!(
            serde_json::to_value(completion.page_document()).unwrap(),
            serde_json::to_value(materialized.page_document()).unwrap()
        );
        assert_eq!(
            serde_json::to_value(completion.receipt_document(&resolved_request)).unwrap(),
            serde_json::to_value(materialized.receipt_document()).unwrap()
        );
        assert_eq!(
            completion.ordered_digest(),
            Some(materialized.ordered_digest().as_str())
        );
        assert!(completion.search_explanation().is_some());
        assert!(completion.parameter_origins().is_some());
        assert!(batches.next_batch().unwrap().is_none());
        batches.close().unwrap();
        batches.close().unwrap();
    }

    #[test]
    fn exhaustive_batches_preserve_the_rust_surface_unbounded_default() {
        let (_directory, app) = existing_only_search_app(
            Provider::Codex,
            "batch-default-fixture",
            &["needle zero", "needle one", "needle two"],
        );
        let request = literal_request()
            .session_id("codex:batch-default-fixture")
            .unwrap()
            .build()
            .unwrap();
        let materialized = app.messages().search(request.clone()).unwrap();
        let mut batches = app
            .message_search_batches(request, NonZeroUsize::new(2).unwrap())
            .unwrap();
        let mut sequences = Vec::new();
        while let Some(batch) = batches.next_batch().unwrap() {
            sequences.extend(batch.results().iter().map(|result| result.seq));
        }
        assert_eq!(
            sequences,
            materialized
                .results()
                .iter()
                .map(|result| result.seq)
                .collect::<Vec<_>>()
        );
        assert_eq!(
            batches.request().extent(),
            crate::message_search::ResolvedRequestExtent::AllResults { offset: 0 }
        );
        assert_eq!(batches.completion().unwrap().page(), materialized.page());
    }

    #[test]
    fn exhaustive_batches_reject_finite_pages_with_the_simple_api_recovery() {
        let (_directory, app) =
            existing_only_search_app(Provider::Antigravity, "batch-fixture", &["needle zero"]);
        let request = literal_request()
            .session_id("antigravity:batch-fixture")
            .unwrap()
            .extent(RequestedExtent::page(Some(1), 0).unwrap())
            .build()
            .unwrap();
        let error = app
            .message_search_batches(request, NonZeroUsize::new(1).unwrap())
            .err()
            .expect("finite page must be rejected")
            .to_string();
        assert!(error.contains("requires all_results"));
        assert!(error.contains("use search() for a finite materialized page"));
    }

    #[test]
    fn exhaustive_batches_keep_one_snapshot_and_close_does_not_drain() {
        let (directory, app) = existing_only_search_app(
            Provider::GeminiCli,
            "batch-fixture",
            &["needle zero", "needle one", "needle two"],
        );
        let request = literal_request()
            .session_id("gemini-cli:batch-fixture")
            .unwrap()
            .extent(RequestedExtent::all_results())
            .build()
            .unwrap();
        let mut batches = app
            .message_search_batches(request, NonZeroUsize::new(1).unwrap())
            .unwrap();
        let first = batches.next_batch().unwrap().unwrap();
        assert_eq!(
            first
                .results()
                .iter()
                .map(|result| result.seq)
                .collect::<Vec<_>>(),
            vec![0]
        );

        let writer = Db::open(&directory.path().join("message-search.db")).unwrap();
        insert_session(
            &writer,
            Provider::GeminiCli,
            "batch-fixture",
            "/workspace/message-search",
            "/transcripts/message-search.jsonl",
            &["needle zero", "needle one", "needle two", "needle late"],
        );
        let mut remaining = Vec::new();
        while let Some(batch) = batches.next_batch().unwrap() {
            remaining.extend(batch.results().iter().map(|result| result.seq));
        }
        assert_eq!(remaining, vec![1, 2]);
        assert_eq!(
            writer
                .search_messages("needle", &MessageFilters::default())
                .unwrap()
                .len(),
            4
        );

        let request = literal_request()
            .session_id("gemini-cli:batch-fixture")
            .unwrap()
            .extent(RequestedExtent::all_results())
            .build()
            .unwrap();
        for _ in 0..16 {
            let mut immediately_stopped = app
                .message_search_batches(request.clone(), NonZeroUsize::new(1).unwrap())
                .unwrap();
            immediately_stopped.close().unwrap();
        }
        assert_eq!(writer.wal_checkpoint_busy_for_test().unwrap(), 0);

        let mut stopped = app
            .message_search_batches(request, NonZeroUsize::new(1).unwrap())
            .unwrap();
        assert!(stopped.next_batch().unwrap().is_some());
        stopped.close().unwrap();
        stopped.close().unwrap();
        assert!(stopped
            .next_batch()
            .unwrap_err()
            .to_string()
            .contains("closed"));
        assert!(stopped
            .completion()
            .unwrap_err()
            .to_string()
            .contains("no terminal metadata"));
        assert_eq!(writer.wal_checkpoint_busy_for_test().unwrap(), 0);
    }

    #[test]
    fn materialized_search_keeps_hits_and_context_in_one_snapshot() {
        let (directory, app) = existing_only_search_app(
            Provider::GeminiCli,
            "materialized-snapshot",
            &["needle before refresh"],
        );
        let writer = Db::open(&directory.path().join("message-search.db")).unwrap();
        let request = literal_request()
            .session_id("gemini-cli:materialized-snapshot")
            .unwrap()
            .extent(RequestedExtent::all_results())
            .context(ContextWindow::new(0, 1))
            .build()
            .unwrap();

        let response = app
            .messages()
            .search_with_after_retrieval_for_test(request.clone(), || {
                insert_session(
                    &writer,
                    Provider::GeminiCli,
                    "materialized-snapshot",
                    "/workspace/message-search",
                    "/transcripts/message-search.jsonl",
                    &["needle before refresh", "context committed during refresh"],
                );
            })
            .unwrap();
        assert_eq!(response.results().len(), 1);
        assert!(response
            .context_windows()
            .iter()
            .flatten()
            .all(|message| message.content != "context committed during refresh"));

        let refreshed = app.messages().search(request).unwrap();
        assert!(refreshed
            .context_windows()
            .iter()
            .flatten()
            .any(|message| message.content == "context committed during refresh"));
    }

    #[test]
    fn exhaustive_batches_resolve_session_prefix_inside_the_result_snapshot() {
        let (directory, app) = existing_only_search_app(
            Provider::Pi,
            "snapshot-original",
            &["needle before replacement"],
        );
        let request = literal_request()
            .session_id("pi:snapshot")
            .unwrap()
            .extent(RequestedExtent::all_results())
            .build()
            .unwrap();
        let ready = Arc::new(Barrier::new(2));
        let worker_ready = Arc::clone(&ready);
        let mut batches = MessageSearchBatches::spawn_with_before_traversal(
            app.config.clone(),
            app.access.clone(),
            SearchSurface::Rust,
            request,
            NonZeroUsize::new(1).unwrap(),
            move || {
                worker_ready.wait();
            },
        )
        .unwrap();

        // The worker has resolved the prefix and anchored its transaction, but has not started
        // result traversal. Replacing the source identity now must affect only later searches.
        let writer = Db::open(&directory.path().join("message-search.db")).unwrap();
        insert_session(
            &writer,
            Provider::Pi,
            "snapshot-replacement",
            "/workspace/message-search",
            "/transcripts/message-search.jsonl",
            &["needle after replacement"],
        );
        ready.wait();

        let mut sessions = Vec::new();
        while let Some(batch) = batches.next_batch().unwrap() {
            sessions.extend(
                batch
                    .results()
                    .iter()
                    .map(|result| result.message.session_id.clone()),
            );
        }
        assert_eq!(sessions, vec!["pi:snapshot-original"]);
        assert_eq!(
            writer.resolve_session("pi:snapshot").unwrap().session.id,
            "pi:snapshot-replacement"
        );
    }

    #[test]
    fn latest_literal_and_regex_select_newest_matches_then_present_chronologically() {
        let (_directory, db) = disposable_db();
        insert_session(
            &db,
            Provider::Claude,
            "latest-window",
            "/workspace/latest",
            "/transcripts/latest.jsonl",
            &["needle zero", "unrelated", "needle two", "needle three"],
        );
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);

        for query in [
            MessageQuery::literal("needle").unwrap(),
            MessageQuery::regex("needle (zero|two|three)").unwrap(),
        ] {
            let response = service
                .search(
                    MessageSearchRequest::builder(query, MessageTarget::content())
                        .session_id("claude:latest")
                        .unwrap()
                        .match_window(MatchWindow::Latest)
                        .extent(RequestedExtent::page(Some(2), 0).unwrap())
                        .build()
                        .unwrap(),
                )
                .unwrap();
            assert_eq!(
                response
                    .hits()
                    .iter()
                    .map(|hit| hit.seq)
                    .collect::<Vec<_>>(),
                vec![2, 3]
            );
            assert_eq!(response.page().next_offset(), Some(2));
        }
    }

    #[test]
    fn latest_window_applies_to_tool_name_and_tool_argument_targets() {
        let (_directory, db) = disposable_db();
        let mut parsed = minimal_record(
            Provider::Claude,
            std::path::Path::new("/transcripts/latest-tools.jsonl"),
            String::new(),
        );
        parsed.session.id = "claude:latest-tools".into();
        parsed.session.provider_session_id = "latest-tools".into();
        parsed.session.cwd = Some("/workspace/latest-tools".into());
        parsed.messages = (0..4)
            .map(|seq| Message {
                seq,
                role: Role::Tool,
                ts: None,
                tool_name: Some(format!("exec-{seq}")),
                kind: MessageKind::ToolCall,
                tool_call_id: Some(format!("call-{seq}")),
                is_compaction: false,
                content: format!(r#"{{"args":{{"cmd":"needle {seq}"}}}}"#),
                provenance: Default::default(),
            })
            .collect();
        db.upsert_session(&parsed, 0, 0).unwrap();
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);

        for (query, target) in [
            (
                MessageQuery::literal("exec").unwrap(),
                MessageTarget::tool_name(),
            ),
            (
                MessageQuery::regex(r"exec-[0-3]").unwrap(),
                MessageTarget::tool_name(),
            ),
            (
                MessageQuery::literal("needle").unwrap(),
                MessageTarget::tool_argument("/cmd").unwrap(),
            ),
            (
                MessageQuery::regex(r"needle [0-3]").unwrap(),
                MessageTarget::tool_argument("/cmd").unwrap(),
            ),
        ] {
            let response = service
                .search(
                    MessageSearchRequest::builder(query, target)
                        .session_id("claude:latest-tools")
                        .unwrap()
                        .match_window(MatchWindow::Latest)
                        .extent(RequestedExtent::page(Some(2), 0).unwrap())
                        .build()
                        .unwrap(),
                )
                .unwrap();
            assert_eq!(
                response
                    .hits()
                    .iter()
                    .map(|hit| hit.seq)
                    .collect::<Vec<_>>(),
                vec![2, 3]
            );
        }
    }

    #[test]
    fn typed_path_domains_are_separate_while_legacy_path_remains_broad() {
        let (_directory, db) = disposable_db();
        insert_session(
            &db,
            Provider::Claude,
            "workspace-domain",
            "/domains/shared/workspace",
            "/elsewhere/workspace.jsonl",
            &["needle"],
        );
        insert_session(
            &db,
            Provider::Claude,
            "transcript-domain",
            "/elsewhere/transcript",
            "/domains/shared/transcript/session.jsonl",
            &["needle"],
        );
        let config = Config::default();
        let service = MessageService::new(&config, &db, SearchSurface::Rust);

        let workspace = service
            .search(
                literal_request()
                    .workspace_path_prefix("/domains/shared")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(workspace.hits().len(), 1);
        assert_eq!(workspace.hits()[0].session_id, "claude:workspace-domain");

        let transcript = service
            .search(
                literal_request()
                    .transcript_path_prefix("/domains/shared")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(transcript.hits().len(), 1);
        assert_eq!(transcript.hits()[0].session_id, "claude:transcript-domain");

        let excluded_workspace = service
            .search(
                literal_request()
                    .exclude_workspace_path_prefix("/domains/shared")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(excluded_workspace.hits().len(), 1);
        assert_eq!(
            excluded_workspace.hits()[0].session_id,
            "claude:transcript-domain"
        );

        let excluded_transcript = service
            .search(
                literal_request()
                    .exclude_transcript_path_prefix("/domains/shared")
                    .unwrap()
                    .build()
                    .unwrap(),
            )
            .unwrap();
        assert_eq!(excluded_transcript.hits().len(), 1);
        assert_eq!(
            excluded_transcript.hits()[0].session_id,
            "claude:workspace-domain"
        );

        let legacy = service
            .search_legacy(
                "needle",
                &MessageFilters {
                    path_prefix: Some("/domains/shared".into()),
                    ..Default::default()
                },
            )
            .unwrap();
        assert_eq!(legacy.len(), 2);
    }
}

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

    #[test]
    fn services_share_one_database_and_return_typed_empty_results() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open(&dir.path().join("index.db")).unwrap();
        db.mark_schema_current().unwrap();

        let config = Config::default();
        let messages = MessageService::new(&config, &db, SearchSurface::Rust)
            .search_legacy("missing", &MessageFilters::default())
            .unwrap();
        let files = FileService::new(&db).search(&FileQuery::default()).unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().to_string());
        let app = SessionSearch::open(config).unwrap();
        app.database().mark_schema_current().unwrap();
        let status = app.index().status().unwrap();

        assert!(messages.is_empty());
        assert!(files.is_empty());
        assert!(status.parser_health.schema_current);
        assert_eq!(status.parser_health.indexed_sessions, 0);
    }

    #[test]
    fn catalog_rejects_a_parent_filter_that_can_match_no_user_session() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open(&dir.path().join("index.db")).unwrap();
        db.mark_schema_current().unwrap();

        let error = CatalogService::new(&db)
            .list_sessions(&SearchFilters {
                session_kinds: Some(vec![crate::models::SessionKind::User]),
                parent_session_id: Some("claude:parent".to_string()),
                ..SearchFilters::default()
            })
            .expect_err("a user-started session cannot have a parent")
            .to_string();

        assert!(error.contains("session_kinds"), "{error}");
        assert!(error.contains("parent_session_id"), "{error}");
        assert!(error.contains("subagent"), "{error}");
        assert!(error.contains("omit session_kinds"), "{error}");
    }

    #[test]
    fn application_owns_config_database_and_service_lifecycle() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().to_string());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().to_string());
        config.providers.claude.enabled = false;
        config.providers.claude_desktop.enabled = false;
        config.providers.codex.enabled = false;
        config.providers.cursor.enabled = false;
        config.providers.antigravity.enabled = false;
        config.providers.pi.enabled = false;
        config.providers.prime_agent.enabled = false;
        config.providers.aistudio.enabled = false;
        config.providers.gemini_cli.enabled = false;

        let app = SessionSearch::open(config).unwrap();
        let outcome = app.index().refresh().unwrap();

        assert!(matches!(outcome, AutoReindexOutcome::Updated { .. }));
        assert!(app.config().db_path().exists());
        assert!(app
            .catalog()
            .list_sessions(&SearchFilters {
                limit: 1,
                ..SearchFilters::default()
            })
            .unwrap()
            .is_empty());

        let diagnostics = app.maintenance().diagnostics().unwrap();
        assert_eq!(
            std::path::Path::new(&diagnostics.db_path),
            app.config().db_path()
        );
        let compacted = app.maintenance().compact().unwrap();
        assert!(compacted.after_bytes > 0);
        assert_eq!(
            compacted.reclaimed_bytes(),
            compacted.before_bytes.saturating_sub(compacted.after_bytes)
        );
    }

    #[test]
    fn explicit_reindex_owns_lock_on_current_schema() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().to_string());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().to_string());
        config.providers.claude.enabled = false;
        config.providers.claude_desktop.enabled = false;
        config.providers.codex.enabled = false;
        config.providers.cursor.enabled = false;
        config.providers.antigravity.enabled = false;
        config.providers.pi.enabled = false;
        config.providers.prime_agent.enabled = false;
        config.providers.aistudio.enabled = false;
        config.providers.gemini_cli.enabled = false;

        let app = SessionSearch::open(config).unwrap();
        assert!(!app.database().needs_backfill().unwrap());

        assert_eq!(app.index().reindex(false).unwrap(), (0, 0));

        assert!(!app.database().needs_backfill().unwrap());
        assert!(app.index().status().unwrap().parser_health.schema_current);
        assert!(indexer::index_update_lock_path(&app.config().db_path()).exists());
    }

    #[test]
    fn public_full_reindex_promotes_v3_to_current_search_schema() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.index.db_path = Some(dir.path().join("index.db").to_string_lossy().to_string());
        config.index.cache_dir = Some(dir.path().join("cache").to_string_lossy().to_string());
        config.providers.claude.enabled = false;
        config.providers.claude_desktop.enabled = false;
        config.providers.codex.enabled = false;
        config.providers.cursor.enabled = false;
        config.providers.antigravity.enabled = false;
        config.providers.pi.enabled = false;
        config.providers.prime_agent.enabled = false;
        config.providers.aistudio.enabled = false;
        config.providers.gemini_cli.enabled = false;

        drop(Db::open(&config.db_path()).unwrap());
        let conn = rusqlite::Connection::open(config.db_path()).unwrap();
        conn.execute_batch(
            "drop trigger messages_ai;
                 drop trigger messages_ad;
                 drop trigger messages_au;
                 drop table messages_trigram_terms;
                 drop table messages_trigram_vocab;
                 drop table messages_trigram;
                 pragma user_version=3;",
        )
        .unwrap();
        crate::fts::install_released_message_word_index(&conn).unwrap();
        crate::trigram_index::ensure_schema(&conn).unwrap();
        drop(conn);

        let app = SessionSearch::open(config).unwrap();

        assert_eq!(app.index().reindex(true).unwrap(), (0, 0));
        assert_eq!(
            app.database().schema_version().unwrap(),
            crate::db::SCHEMA_VERSION
        );
        let conn = rusqlite::Connection::open_with_flags(
            app.config().db_path(),
            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
        )
        .unwrap();
        let layout: (bool, bool, bool) = conn
            .query_row(
                "select exists(select 1 from sqlite_schema where name='messages_trigram'),
                        exists(select 1 from sqlite_schema where name='messages_trigram_vocab'),
                        not exists(select 1 from sqlite_schema where name='trigram_postings')",
                [],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!(layout, (true, true, true));
    }
}