"""Semantic search manager.

Indexes codebase content into embeddings and provides natural language
search over code. Gracefully degrades to BM25 keyword matching when no
embedding provider is available.
"""

from __future__ import annotations

import hashlib
import json
import logging
import math
import os
import queue
import re
import sqlite3
import subprocess
import threading
import time
from dataclasses import dataclass, field, replace
from itertools import islice
from typing import Any

from attocode_intel._internal.integrations.context.index_probe import (
    IndexProbe,
    activate,
    count,
    instrument,
    measure,
    measured_call,
    measured_lock,
)
from attocode_intel.query_ranking import demote_tests, hit_evidence, rerank_broad_candidates
from attocode_intel.search_evidence import search_snippet

logger = logging.getLogger(__name__)


def _sql_batches(rows):
    """Bound SQL parameters and temporary row materialization, retaining order."""
    rows = iter(rows)
    while batch := list(islice(rows, 64)):
        yield batch

# ---------------------------------------------------------------------------
# BM25 keyword search support types
# ---------------------------------------------------------------------------

_STOP_WORDS = frozenset({
    "self", "cls", "the", "is", "in", "of", "and", "or", "to", "a", "an",
    "for", "not", "on", "with", "as", "at", "by", "from", "it", "be",
    "this", "that", "if", "else", "def", "class", "import", "return",
    "none", "true", "false", "pass", "str", "int", "float", "bool", "list",
    "dict", "set", "tuple", "any", "type", "optional",
})

# Prose and data formats: down-ranked in keyword search, and kept out of
# whole-file BM25, where a long prose issue would match docs before code.
_NON_CODE_EXTS = frozenset({
    ".md", ".txt", ".rst", ".cfg", ".ini", ".yml", ".yaml", ".json", ".toml", ".xml", ".csv",
})

_CAMEL_RE = re.compile(r"(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
_ASCII_WORD_RE = re.compile(r"[a-zA-Z0-9]+")

# Mapping from common query terms to code construct types for query expansion
_CONSTRUCT_HINTS: dict[str, list[str]] = {
    "function": ["def", "func", "fn"],
    "class": ["class", "struct", "type"],
    "method": ["def", "func", "method"],
    "import": ["import", "require", "from", "use"],
    "test": ["test", "spec", "assert", "expect"],
    "error": ["error", "exception", "raise", "throw", "catch"],
    "config": ["config", "settings", "env", "option"],
    "api": ["route", "endpoint", "handler", "controller"],
    "auth": ["auth", "login", "password", "token", "session"],
    "database": ["query", "model", "schema", "migration", "table"],
}


def _source_spans(ast):
    symbols = [(f.start_line, f.end_line, "function", f.name) for f in ast.functions]
    classes = []
    for cls in ast.classes:
        classes.append((cls.start_line, cls.end_line, "class", cls.name))
        symbols.extend((m.start_line, m.end_line, "method", f"{cls.name}.{m.name}") for m in cls.methods)
    return symbols, classes


class _BodySpanCache:
    """One warm-up's bounded source-hash-bound spans, never retained ASTs or source."""

    def __init__(self, budget_bytes=16 * 2**20):
        self.remaining = budget_bytes
        self.rows = {}

    def put(self, path, content, ast):
        symbols, classes = _source_spans(ast)
        # Conservative allowance for Python containers, integers and names.
        size = 512 + len(path.encode()) + sum(256 + len(span[3].encode()) for span in [*symbols, *classes])
        if size <= self.remaining and len(content.encode()) <= 512 * 1024:
            self.rows[path] = (hashlib.sha256(content.encode()).digest(), (symbols, classes))
            self.remaining -= size

    def take(self, path, content):
        row = self.rows.pop(path, None)
        if row and row[0] == hashlib.sha256(content.encode()).digest():
            return row[1]
        return None


def _tokenize(text: str) -> list[str]:
    """Tokenize text: split camelCase/snake_case, lowercase, remove stop words."""
    # Camel boundaries use adjacent ASCII letters, so they cannot cross any
    # separator the old per-word loop split on. One substitution over the input
    # avoids a regex call and temporary split list for every source word.
    # Extract ASCII words BEFORE lowercasing: Unicode case folds may add ASCII.
    parts = (part.lower() for part in _ASCII_WORD_RE.findall(_CAMEL_RE.sub("_", text)))
    return [part for part in parts if len(part) >= 2 and part not in _STOP_WORDS]


def _expand_query(query: str, language: str = "") -> str:
    """Expand a search query with language and construct hints.

    Prepends language context and adds related terms for known constructs,
    improving recall for code-specific embedding models.

    Examples:
        "auth middleware" + "python" -> "python auth middleware login token session"
        "parse config" -> "parse config settings env option"
    """
    query_lower = query.lower()
    tokens = set(_tokenize(query))
    expansions: list[str] = []

    # Add language hint if known
    if language:
        expansions.append(language)

    # Add construct-related terms for each matching concept
    for concept, hints in _CONSTRUCT_HINTS.items():
        if concept in tokens or concept + "s" in tokens:
            for hint in hints:
                if hint not in query_lower and hint not in expansions:
                    expansions.append(hint)

    if not expansions:
        return query

    return f"{query} {' '.join(expansions)}"


def _summarize_code_to_nl(name: str, chunk_type: str, text: str) -> str:
    """Generate a heuristic NL summary from code structure.

    Converts code identifiers into natural language fragments to improve
    BM25 matching between NL queries and code symbols.

    Examples:
        "parseConfigFile" -> "parse config file"
        "UserAuthMiddleware" -> "user auth middleware"
    """
    # Tokenize the name into natural language words
    name_words = _CAMEL_RE.sub(" ", name).replace("_", " ").lower().strip()

    if chunk_type == "function":
        return f"function {name_words} {text[:200]}"
    elif chunk_type == "class":
        return f"class {name_words} {text[:200]}"
    elif chunk_type == "method":
        return f"method {name_words} {text[:200]}"
    else:
        return f"{name_words} {text[:200]}"


@dataclass(slots=True)
class _KeywordDoc:
    """A document in the BM25 keyword index."""

    id: str              # "func:path:name" or "file:path"
    file_path: str
    chunk_type: str      # file, function, class, method
    name: str
    text: str            # preview text
    is_config: bool
    is_test: bool
    term_freqs: dict[str, int] = field(default_factory=dict)
    doc_len: int = 0
    start_line: int = 0
    end_line: int = 0


@dataclass(slots=True)
class IndexProgress:
    """Progress of background embedding indexing."""

    total_files: int = 0
    indexed_files: int = 0
    failed_files: int = 0
    status: str = "idle"  # idle, running, paused, completed, error
    coverage: float = 0.0  # indexed / total
    started_at: float = 0.0
    elapsed_seconds: float = 0.0
    last_error: str = ""
    degraded_reason: str = ""


@dataclass(slots=True)
class ContextAssemblyConfig:
    """Tunable parameters for bootstrap() and relevant_context().

    Controls token budget allocation, size tier thresholds, and
    symbol display limits during context assembly.
    """

    # bootstrap() size tier thresholds (file count)
    small_repo_threshold: int = 100
    large_repo_threshold: int = 5000

    # bootstrap() budget ratios (with task_hint)
    summary_ratio: float = 0.20
    structure_ratio: float = 0.20
    conventions_ratio: float = 0.05
    search_ratio: float = 0.50

    # bootstrap() budget ratios (without task_hint)
    summary_ratio_no_hint: float = 0.40
    structure_ratio_no_hint: float = 0.44
    conventions_ratio_no_hint: float = 0.16

    # bootstrap() medium repo structure split
    medium_structure_map_ratio: float = 0.7

    # bootstrap() explorer params (large repos)
    explore_max_items: int = 20
    explore_importance_threshold: float = 0.3

    # bootstrap() hotspots count
    bootstrap_hotspots_n: int = 10

    # bootstrap() conventions sample size
    conventions_sample_size: int = 25

    # bootstrap() search top_k
    bootstrap_search_top_k: int = 5

    # relevant_context() max BFS depth
    max_depth: int = 2

    # relevant_context() symbol caps per file
    center_symbol_cap: int = 8
    neighbor_symbol_cap: int = 5

    # relevant_context() preview limits
    param_preview_limit: int = 4
    base_preview_limit: int = 3
    method_preview_limit: int = 4


@dataclass(slots=True)
class SearchScoringConfig:
    """Tunable scoring parameters for BM25 keyword search and two-stage retrieval.

    Default values match the original hardcoded constants. Adjusting these
    allows the meta-harness optimization loop to explore better configurations.
    """

    # BM25 base parameters (optimized via meta-harness, round 5)
    bm25_k1: float = 2.2      # was 1.5 → 2.8 (kw-only) → 2.2 (final with adaptive fusion)
    bm25_b: float = 0.3       # was 0.75 → 0.05 (kw-only) → 0.3 (final with adaptive fusion)

    # Graduated symbol name match boosts
    name_exact_boost: float = 5.0
    name_substring_boost: float = 3.0
    name_token_boost: float = 2.2

    # Definition-type boosts
    class_boost: float = 1.8
    function_boost: float = 1.4
    method_boost: float = 1.3

    # Path relevance: source directory boost
    src_dir_boost: float = 1.7

    # Multi-term coverage bonuses
    multi_term_high_bonus: float = 2.5
    multi_term_med_bonus: float = 1.8
    multi_term_high_threshold: float = 0.7
    multi_term_med_threshold: float = 0.4

    # File type penalties
    non_code_penalty: float = 0.3
    config_penalty: float = 0.15
    test_penalty: float = 0.6

    # Exact phrase bonus
    exact_phrase_bonus: float = 3.0

    # Deduplication: more diversity per file when vectors return semantic neighbors
    max_chunks_per_file: int = 8

    # Two-stage retrieval — wide candidates + balanced default RRF
    wide_k_multiplier: int = 12
    wide_k_min: int = 150
    rrf_k: int = 60               # default; adaptive fusion overrides per-list

    # Algorithmic signals (Phase 1 improvements)
    # Ablation-validated defaults: importance helps (+0.4%), rerank hurts (-1.4%) with aggressive threshold
    importance_weight: float = 0.5   # file importance boost (0 = disabled)
    frecency_weight: float = 0.2     # frecency boost (0 = disabled, currently no tracking data)
    rerank_confidence_threshold: float = 0.0  # disabled by default — ablation showed it hurts on code queries; set >0 to enable

    # Phase 2: Dependency proximity — boost results near high-ranking files in the import graph
    dep_proximity_weight: float = 0.3   # 0 = disabled
    dep_proximity_seed_count: int = 5    # top-N results used as "seeds" for neighborhood discovery

    # Phase 3a: Adaptive fusion — vary keyword/vector weighting by keyword confidence
    # When keyword search produces a confident top result, downweight vectors (avoid semantic noise)
    # When keyword is uncertain, use balanced fusion to let vectors fill the gap
    adaptive_fusion: bool = True               # True = adapt rrf_k by keyword confidence
    kw_dominance_threshold: float = 1.5        # top-1/top-2 score ratio for "dominant" classification
    rrf_k_keyword_high_conf: int = 10          # sharp k_keyword when keyword is confident
    rrf_k_vector_low_conf: int = 250           # smooth k_vector when keyword is confident (downweight)

    # Lexical candidates (search_candidates): RRF of keyword and body chunks.
    # A long query then also fuses the file order with whole-file BM25.
    chunk_rrf_k: int = 20
    body_weight: float = 1.15          # the keyword list has weight 1.0
    body_max_tokens: int = 20          # rarest query words that body search sends
    file_bm25_min_tokens: int = 20     # whole-file BM25 needs more unique words than this
    file_rrf_k: int = 60
    file_fusion_depth: int = 48        # files per list in whole-file fusion


@dataclass(slots=True)
class SemanticSearchResult:
    """A single semantic search result."""

    file_path: str
    chunk_type: str  # "file", "function", "class"
    name: str
    text: str
    score: float
    start_line: int = 0
    end_line: int = 0
    # Definition identity for metadata hits; start/end above describe source text.
    definition_start_line: int = 0
    definition_end_line: int = 0


def _file_order(results: list[SemanticSearchResult]) -> list[str]:
    """Each file path once, in the order of its first result."""
    return list(dict.fromkeys(result.file_path for result in results))


@dataclass(slots=True)
class SemanticSearchManager:
    """Manages semantic search over a codebase.

    Usage::

        mgr = SemanticSearchManager(root_dir="/path/to/repo")
        mgr.index()  # Build embeddings (once)
        results = mgr.search("authentication middleware", top_k=10)
    """

    root_dir: str
    nl_mode: str = ""  # "none" or "heuristic"; "" = read ATTOCODE_NL_EMBEDDING_MODE env var
    scoring_config: SearchScoringConfig = field(default_factory=SearchScoringConfig)
    ast_lookup: Any = field(default=None, repr=False)
    discovery_snapshot: Any = field(default=None, repr=False)
    index_probe: IndexProbe | None = field(default=None, repr=False)
    _warmup_reused_asts: int = field(default=0, repr=False)
    _provider: Any = field(default=None, repr=False)
    _store: Any = field(default=None, repr=False)
    _indexed: bool = field(default=False, repr=False)
    _keyword_fallback: bool = field(default=False, repr=False)
    _reindex_queue: queue.Queue[str | None] = field(default_factory=queue.Queue, repr=False)
    _reindex_thread: threading.Thread | None = field(default=None, repr=False)
    _closed: bool = field(default=False, repr=False)
    _reindex_pending: set[str] = field(default_factory=set, repr=False)
    _reindex_lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
    _reindex_worker_started: bool = field(default=False, repr=False)
    _kw_docs: list[_KeywordDoc] = field(default_factory=list, repr=False)
    _kw_df: dict[str, int] = field(default_factory=dict, repr=False)
    _kw_avg_dl: float = field(default=0.0, repr=False)
    _kw_index_built: bool = field(default=False, repr=False)
    _kw_cache_db_path: str = field(default="", repr=False)
    _kw_cache_lock: threading.RLock = field(default_factory=threading.RLock, repr=False)
    _body_index_built: bool = field(default=False, repr=False)
    _body_index_available: bool = field(default=True, repr=False)
    _body_revision: str = field(default="", repr=False)
    _body_generation: int = field(default=0, repr=False)
    _body_failed_at: float = field(default=0.0, repr=False)
    _body_thread: threading.Thread | None = field(default=None, repr=False)
    _body_state_lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
    _warmup_stage: str = field(default="idle", repr=False)
    _warmup_started: float = field(default=0.0, repr=False)
    _warmup_finished: float = field(default=0.0, repr=False)
    _warmup_seconds: dict[str, float] = field(default_factory=dict, repr=False)
    _trigram_index: Any = field(default=None, repr=False)
    _bg_indexer: Any = field(default=None, repr=False)
    _bg_thread: Any = field(default=None, repr=False)
    _index_progress: IndexProgress = field(default_factory=IndexProgress, repr=False)
    _summarizer: Any = field(default=None, repr=False)
    _importance_scores: dict[str, float] = field(default_factory=dict, repr=False)
    _frecency_tracker: Any = field(default=None, repr=False)
    _dep_graph: Any = field(default=None, repr=False)

    def __post_init__(self) -> None:
        if self.index_probe is None and os.environ.get("ATTOCODE_INTEL_INDEX_PROBE") == "1":
            self.index_probe = IndexProbe()
        # Defer provider creation to first use (_ensure_provider) to avoid
        # 5-15s model-load latency on construction.
        self._provider = None
        self._keyword_fallback = True  # assume keyword-only until provider loads
        self._kw_cache_db_path = os.path.join(
            self.root_dir, ".attocode", "index", "kw_index.db",
        )
        if not self.nl_mode:
            self.nl_mode = os.environ.get("ATTOCODE_NL_EMBEDDING_MODE", "heuristic")

    def _get_importance(self, file_path: str) -> float:
        """Get file importance score (0.0-1.0), lazily loaded."""
        if not self._importance_scores:
            self._load_importance_scores()
        return self._importance_scores.get(file_path, 0.0)

    def _load_importance_scores(self, context_manager: Any = None) -> None:
        """Load file importance from CodebaseContextManager if available."""
        try:
            from attocode_intel._internal.integrations.context.codebase_context import (
                CodebaseContextManager,
            )
            ctx = context_manager or CodebaseContextManager(self.root_dir)
            ctx._ensure_fresh()
            # Publish one complete generation, dropping deleted paths. A failed
            # inventory read must not leave a mixture of old and new weights.
            scores = {fi.relative_path: fi.importance for fi in ctx._files}
            self._importance_scores = scores or {"__empty__": 0.0}
        except Exception:
            logger.debug("Failed to load importance scores", exc_info=True)

    def _get_frecency_score(self, file_path: str) -> float:
        """Get frecency score for a file, lazily loading the tracker."""
        if self._frecency_tracker is None:
            try:
                from attocode_intel._internal.integrations.context.frecency import FrecencyTracker
                self._frecency_tracker = FrecencyTracker(self.root_dir, ai_mode=True)
            except Exception:
                self._frecency_tracker = False  # sentinel: don't retry
                return 0.0
        if self._frecency_tracker is False:
            return 0.0
        try:
            result = self._frecency_tracker.get_score(file_path)
            return float(result.score) if result else 0.0
        except Exception:
            return 0.0

    def _get_dep_graph(self) -> Any:
        """Lazily load the dependency graph from CodebaseContextManager."""
        if self._dep_graph is False:
            return None
        if self._dep_graph is None:
            try:
                from attocode_intel._internal.integrations.context.codebase_context import (
                    CodebaseContextManager,
                )
                ctx = CodebaseContextManager(self.root_dir)
                ctx._ensure_fresh()
                self._dep_graph = ctx.dependency_graph
                if self._dep_graph is None:
                    self._dep_graph = False
                    return None
            except Exception:
                logger.debug("Failed to load dependency graph", exc_info=True)
                self._dep_graph = False
                return None
        return self._dep_graph

    @staticmethod
    def _path_rank_penalty(file_path: str) -> float:
        """Multiplicative rank penalty for non-source paths (tests, benchmarks,
        docs) so genuine source outranks them. 1.0 = no penalty."""
        p = file_path.replace("\\", "/")
        for frag in (
            "/tests/", "tests/", "/test/", "asv_bench/",
            "/docs_src/", "docs_src/", "/benchmarks/",
        ):
            if frag in p or p.startswith(frag):
                return 0.6
        return 1.0

    def _apply_dependency_boost(
        self,
        fused: list[tuple[str, float]],
        result_map: dict[str, SemanticSearchResult],
        weight: float,
        seed_count: int,
    ) -> list[tuple[str, float]]:
        """Boost results that are in the dependency neighborhood of top-N seed results.

        For each of the top seed_count results, find files that import them or are
        imported by them. Any result file in this neighborhood gets a score boost.
        """
        graph = self._get_dep_graph()
        if graph is None or not fused:
            return fused

        # Collect neighborhood files from top-N seeds
        neighborhood: set[str] = set()
        for item_id, _ in fused[:seed_count]:
            result = result_map.get(item_id)
            if not result:
                continue
            fp = result.file_path
            neighborhood.update(graph.get_imports(fp))
            neighborhood.update(graph.get_importers(fp))

        if not neighborhood:
            return fused

        # Apply boost to any result in the neighborhood
        boosted: list[tuple[str, float]] = []
        for item_id, score in fused:
            result = result_map.get(item_id)
            if result and result.file_path in neighborhood:
                score *= (1.0 + weight)
            boosted.append((item_id, score))
        return sorted(boosted, key=lambda x: x[1], reverse=True)

    def _ensure_provider(self) -> None:
        """Initialize embedding provider on first use."""
        if self._provider is not None:
            return
        from attocode_intel._internal.integrations.context.embeddings import (
            NullEmbeddingProvider,
            create_embedding_provider,
        )
        self._provider = create_embedding_provider()
        self._keyword_fallback = isinstance(self._provider, NullEmbeddingProvider)
        if self._keyword_fallback:
            self._index_progress.degraded_reason = "embedding_provider_unavailable"
            if not self._index_progress.last_error:
                self._index_progress.last_error = (
                    "Embedding provider unavailable; semantic search is running in keyword-only mode."
                )

        if not self._keyword_fallback:
            from attocode_intel._internal.integrations.context.vector_store import (
                VectorStore,
                VectorStoreDimensionMismatchError,
            )
            db_dir = os.path.join(self.root_dir, ".attocode", "vectors")
            os.makedirs(db_dir, exist_ok=True)
            # strict_dimension=False: catch the mismatch as degraded mode
            # instead of crashing the whole MCP server. The provider + store
            # both stay available for diagnostic tools
            # (`embeddings_status`, etc.) but writes refuse to land until
            # the user resolves the mismatch.
            provider_name = getattr(self._provider, "name", lambda: "unknown")
            provider_name_str = provider_name() if callable(provider_name) else str(provider_name)
            provider_version = getattr(self._provider, "version", lambda: "")
            provider_version_str = (
                provider_version() if callable(provider_version) else str(provider_version)
            )
            try:
                self._store = VectorStore(
                    db_path=os.path.join(db_dir, "embeddings.db"),
                    dimension=self._provider.dimension(),
                    model_name=provider_name_str,
                    model_version=provider_version_str,
                    strict_dimension=False,
                )
            except VectorStoreDimensionMismatchError as exc:
                logger.warning(
                    "semantic_search: vector store unusable (%s); "
                    "falling back to keyword-only mode",
                    exc,
                )
                self._store = None
                self._keyword_fallback = True
                self._index_progress.degraded_reason = "dimension_mismatch"
                self._index_progress.last_error = str(exc)
            else:
                if self._store.degraded:
                    self._index_progress.degraded_reason = self._store.degraded_reason
                    self._index_progress.last_error = (
                        f"Vector store at {self._store.db_path} is in degraded mode: "
                        f"{self._store.degraded_reason}. Existing vectors preserved; "
                        f"writes disabled until resolved."
                    )

        if not self.nl_mode:
            self.nl_mode = os.environ.get("ATTOCODE_NL_EMBEDDING_MODE", "none")
        if self.nl_mode == "heuristic" and self._summarizer is None:
            from attocode_intel.indexing.summarizer import get_summarizer
            self._summarizer = get_summarizer()

    def _start_reindex_worker(self) -> None:
        """Start the background reindex worker once."""
        with self._reindex_lock:
            if self._reindex_worker_started or self._closed:
                return
            self._reindex_worker_started = True
            self._reindex_thread = threading.Thread(
                target=self._reindex_worker_loop, daemon=True,
                name="semantic-reindex-worker",
            )
            self._reindex_thread.start()

    def _reindex_worker_loop(self) -> None:
        """Background worker for bounded, queued file reindexing."""
        while True:
            rel_path = self._reindex_queue.get()
            try:
                if rel_path is None:
                    return
                self.reindex_file(rel_path)
            except Exception:
                logger.debug("queue_reindex_failed: %s", rel_path, exc_info=True)
            finally:
                with self._reindex_lock:
                    if rel_path is not None:
                        self._reindex_pending.discard(rel_path)
                self._reindex_queue.task_done()

    def queue_reindex(self, file_path: str) -> None:
        """Queue changes only for an already initialized embedding index.

        File notifications must never initialize or download an embedding model.
        Explicit indexing or search initializes the provider before enqueueing.
        """
        if self._keyword_fallback or not self._store:
            return
        try:
            rel = os.path.relpath(file_path, self.root_dir)
        except ValueError:
            rel = file_path

        self._start_reindex_worker()
        with self._reindex_lock:
            if self._closed or rel in self._reindex_pending:
                return
            self._reindex_pending.add(rel)
            self._reindex_queue.put(rel)

    @property
    def is_available(self) -> bool:
        """Check if semantic search is fully available."""
        self._ensure_provider()
        return not self._keyword_fallback

    @property
    def provider_name(self) -> str:
        self._ensure_provider()
        return self._provider.name if self._provider else "none"

    @staticmethod
    def _supported_languages() -> set[str]:
        """Languages the chunker actually embeds: Python/JS/TS always, plus
        any tree-sitter languages (Go, C, Rust, ...) when available.

        Coverage/readiness must use this set, not a hardcoded py/js/ts subset,
        or non-Python repos report 0 coverage despite having a built index.
        """
        _ts_langs: set[str] = set()
        try:
            from attocode_intel._internal.integrations.context.ts_parser import supported_languages
            _ts_langs = set(supported_languages())
        except ImportError:
            pass
        return {"python", "javascript", "typescript"} | _ts_langs

    def index(self, context_manager: Any = None) -> int:
        """Build or update the vector index.

        Args:
            context_manager: Optional CodebaseContextManager for file discovery.

        Returns:
            Number of chunks indexed.
        """
        self._ensure_provider()
        if self._keyword_fallback:
            logger.info("Semantic search: no provider, skipping indexing")
            return 0

        from attocode_intel._internal.integrations.context.codebase_context import (
            CodebaseContextManager,
        )
        from attocode_intel._internal.integrations.context.vector_store import VectorEntry

        ctx = context_manager or CodebaseContextManager(root_dir=self.root_dir)
        ctx._ensure_fresh()
        if not ctx._files:
            ctx.discover_files()

        # Supported languages: Python, JS, TS always; others when tree-sitter available
        _supported = self._supported_languages()

        chunks: list[tuple[str, str, str, str]] = []

        for f in ctx._files:
            if f.language not in _supported:
                continue
            file_chunks = self._chunk_single_file(f.relative_path, f.path)
            chunks.extend(file_chunks)

        if not chunks:
            return 0

        # Batch embed (use NL summaries when heuristic mode is enabled)
        texts = self._get_embedding_texts(chunks)
        batch_size = 64
        all_vectors: list[list[float]] = []

        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            try:
                vectors = self._provider.embed(batch)
                all_vectors.extend(vectors)
            except Exception:
                logger.warning("Embedding batch failed at offset %d", i, exc_info=True)
                all_vectors.extend([[] for _ in batch])

        # Store (display signature at index 3 is preserved, not body/NL summary)
        entries = []
        for chunk, vec in zip(chunks, all_vectors, strict=False):
            if not vec:
                continue
            cid = chunk[0]
            fpath = chunk[1]
            ctype = chunk[2]
            text = chunk[3]
            entries.append(VectorEntry(
                id=cid,
                file_path=fpath,
                chunk_type=ctype,
                name=cid.split(":")[-1] if ":" in cid else fpath,
                text=text,
                vector=vec,
            ))

        if entries:
            self._store.upsert_batch(entries)

        # Update metadata for indexed files
        indexed_files: dict[str, int] = {}
        for e in entries:
            indexed_files[e.file_path] = indexed_files.get(e.file_path, 0) + 1
        for f in ctx._files:
            if f.relative_path in indexed_files:
                try:
                    mtime = os.path.getmtime(f.path)
                except OSError:
                    mtime = 0.0
                self._store.set_file_metadata(
                    f.relative_path, mtime, indexed_files[f.relative_path],
                )

        self._indexed = True
        logger.info("Semantic search: indexed %d chunks from %d files",
                     len(entries), len(ctx._files))
        return len(entries)

    def search(
        self,
        query: str,
        top_k: int = 10,
        file_filter: str = "",
        two_stage: bool = True,
        rerank: bool = False,
        expand_query: bool = True,
    ) -> list[SemanticSearchResult]:
        """Search the codebase by natural language query.

        Uses two-stage retrieval when vector search is available:
        1. Wide recall: vector top-50 + keyword top-50
        2. Score normalization (min-max to [0,1])
        3. Merge with Reciprocal Rank Fusion (RRF)
        4. Optional cross-encoder reranking

        This approach outperforms single-stage search on code retrieval
        benchmarks by combining semantic similarity with keyword matching.

        Args:
            query: Natural language search query.
            top_k: Number of results to return.
            file_filter: Optional glob pattern (e.g. "*.py").
            two_stage: Whether to use two-stage retrieval (default True).
            rerank: Whether to apply cross-encoder reranking after fusion.
            expand_query: Whether to expand query with code construct hints.

        Returns:
            List of search results sorted by relevance.
        """
        # Detect language from file_filter for query expansion
        _lang = ""
        if file_filter:
            _ext_map = {".py": "python", ".js": "javascript", ".ts": "typescript",
                        ".go": "go", ".rs": "rust", ".java": "java", ".rb": "ruby"}
            for ext, lang in _ext_map.items():
                if file_filter.endswith(ext):
                    _lang = lang
                    break

        # Apply query expansion for better recall
        expanded_query = _expand_query(query, _lang) if expand_query else query

        # Search never initializes an optional embedding provider: doing so can
        # load or download weights on an interactive request. Explicit indexing
        # activates vectors; until then, lexical candidates remain available.
        if self._provider is None:
            return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)

        # An explicitly initialized vector path retains keyword fusion.
        if not self._kw_index_built:
            self._build_keyword_index()

        self._ensure_provider()
        if self._keyword_fallback:
            return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)

        # Refuse vectors produced by a different embedding model (manual-reindex
        # migration): serving them returns semantically garbage results.
        store_model = getattr(self._store, "model_name", "") if self._store else ""
        active_model = getattr(self._provider, "name", "")
        if store_model and active_model and store_model != active_model:
            logger.warning(
                "Vector index built with %s but active model is %s; serving "
                "keyword results only. Run reindex to rebuild vectors.",
                store_model, active_model,
            )
            return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)

        # Coverage-based switchover: use keyword fallback while indexing
        if not self._indexed and self._store:
            count = self._store.count()
            if count == 0 and self._bg_indexer is None:
                # No embeddings and no background indexer — use keyword fallback
                return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)
            elif not self.is_index_ready() and self._bg_indexer is not None:
                # Indexer running but coverage < 80% — use keyword fallback
                return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)

        # Embed query — use expanded query for better recall
        try:
            query_vectors = self._provider.embed_query([expanded_query])
            if not query_vectors or not query_vectors[0]:
                self._index_progress.degraded_reason = "query_embedding_failed"
                self._index_progress.last_error = "Query embedding returned no vector."
                return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)
            query_vec = query_vectors[0]
        except Exception as exc:
            logger.warning("Query embedding failed, falling back to keyword", exc_info=True)
            self._index_progress.degraded_reason = "query_embedding_failed"
            self._index_progress.last_error = f"Query embedding failed: {type(exc).__name__}: {exc}"
            return self.search_candidates(expanded_query, top_k, file_filter) or self._scoped_literal_results(query, file_filter)

        # Build set of files that exist on disk to filter out stale branch data.
        # In local mode the filesystem is the source of truth — vectors from
        # files that no longer exist (e.g. after a branch switch) are skipped.
        existing_files: set[str] | None = None
        try:
            indexed_paths = self._store.get_all_indexed_files()
            if indexed_paths:
                existing_files = {
                    p for p in indexed_paths
                    if os.path.exists(os.path.join(self.root_dir, p))
                }
        except Exception:
            logger.debug("Failed to build existing_files set", exc_info=True)

        # Stage 1a: Vector search (wide recall)
        cfg = self.scoring_config
        wide_k = max(top_k * cfg.wide_k_multiplier, cfg.wide_k_min) if two_stage else top_k
        raw_results = self._store.search(
            query_vec, top_k=wide_k, file_filter=file_filter,
            existing_files=existing_files,
        )

        if not two_stage:
            return [
                SemanticSearchResult(
                    file_path=r.file_path,
                    chunk_type=r.chunk_type,
                    name=r.name,
                    text=r.text,
                    score=r.score,
                )
                for r in raw_results[:top_k]
            ]

        # Stage 1b: Keyword search (complementary recall) — always run
        keyword_results = self.search_candidates(expanded_query, top_k=wide_k, file_filter=file_filter)

        # If both pipelines returned nothing, bail out early
        if not raw_results and not keyword_results:
            return []

        # Stage 2: Score normalization + Reciprocal Rank Fusion
        from attocode_intel._internal.integrations.context.ast_chunker import reciprocal_rank_fusion

        # Map chunk_type to vector ID prefix for consistent key space
        _type_prefix = {"function": "func", "class": "cls", "method": "method", "file": "file"}

        vector_ranked = self._normalize_scores(
            [(r.id, r.score) for r in raw_results],
        )
        def _result_id(result: SemanticSearchResult) -> str:
            prefix = _type_prefix.get(result.chunk_type, result.chunk_type)
            if result.chunk_type == "file":
                return f"file:{result.file_path}"
            return f"{prefix}:{result.file_path}:{result.name}"

        # Use composite IDs for keyword results matching vector key space
        keyword_ranked = self._normalize_scores([
            (_result_id(r), r.score)
            for r in keyword_results
        ])

        # Adaptive fusion: trust keyword more only when BOTH lists agree on the top result.
        # Agreement = vector and keyword identify the same top file → keyword is genuinely correct.
        # Disagreement = treat both as uncertain, use balanced fusion (vectors may have semantic insight).
        # This avoids over-trusting keyword when its top match is high-dominance but wrong (Redis case).
        if cfg.adaptive_fusion and keyword_ranked and vector_ranked:
            kw_top = keyword_ranked[0][1] if keyword_ranked else 0.0
            kw_second = keyword_ranked[1][1] if len(keyword_ranked) > 1 else 0.0
            dominance = (kw_top / kw_second) if kw_second > 0 else float("inf")

            # Extract file path from top results (id format: "type:path:name")
            def _file_of(item_id: str) -> str:
                parts = item_id.split(":", 2)
                return parts[1] if len(parts) >= 2 else item_id

            kw_top_file = _file_of(keyword_ranked[0][0])
            # Cross-modal agreement: keyword top-1 file appears in vector top-3
            # Strict agreement protects redis-style queries (where vectors are correct
            # and keyword is wrong) — keyword shouldn't be trusted unless vectors confirm.
            vec_top_files = {_file_of(r[0]) for r in vector_ranked[:3]}
            agreement = kw_top_file in vec_top_files
            high_confidence = dominance >= cfg.kw_dominance_threshold and agreement

            if high_confidence:
                k_keyword = cfg.rrf_k_keyword_high_conf
                k_vector = cfg.rrf_k_vector_low_conf
            else:
                k_keyword = cfg.rrf_k
                k_vector = cfg.rrf_k

            # Per-list weighted RRF (inline to support per-list k)
            fused_dict: dict[str, float] = {}
            for rank, (item_id, _) in enumerate(vector_ranked):
                fused_dict[item_id] = fused_dict.get(item_id, 0.0) + 1.0 / (k_vector + rank + 1)
            for rank, (item_id, _) in enumerate(keyword_ranked):
                fused_dict[item_id] = fused_dict.get(item_id, 0.0) + 1.0 / (k_keyword + rank + 1)
            fused = sorted(fused_dict.items(), key=lambda x: x[1], reverse=True)
        else:
            fused = reciprocal_rank_fusion(vector_ranked, keyword_ranked, k=cfg.rrf_k)

        # Build result lookup for fast access
        result_map: dict[str, SemanticSearchResult] = {}
        for r in raw_results:
            result_map[r.id] = SemanticSearchResult(
                file_path=r.file_path,
                chunk_type=r.chunk_type,
                name=r.name,
                text=r.text,
                score=r.score,
            )
        for r in keyword_results:
            kw_id = _result_id(r)
            if kw_id not in result_map:
                result_map[kw_id] = r

        # Cross-encoder reranking: explicit or lazy (auto-trigger on low confidence)
        if rerank:
            rerank_k = min(len(fused), 3 * top_k)
            fused = self._rerank(query, fused[:rerank_k], result_map, top_k)
        elif cfg.rerank_confidence_threshold > 0 and fused:
            top_score = fused[0][1] if fused else 0
            if top_score < cfg.rerank_confidence_threshold:
                rerank_k = min(len(fused), 3 * top_k)
                fused = self._rerank(query, fused[:rerank_k], result_map, top_k)

        # Dependency proximity boost: files structurally close to top seeds get a bump
        if cfg.dep_proximity_weight > 0 and fused:
            fused = self._apply_dependency_boost(
                fused, result_map,
                weight=cfg.dep_proximity_weight,
                seed_count=cfg.dep_proximity_seed_count,
            )

        # Frecency boost: recently-accessed files get a score bump
        if cfg.frecency_weight > 0:
            boosted: list[tuple[str, float]] = []
            for item_id, fused_score in fused:
                result = result_map.get(item_id)
                if result:
                    frec = self._get_frecency_score(result.file_path)
                    if frec > 0:
                        boost = min(frec / 100.0, 0.3) * cfg.frecency_weight
                        fused_score *= (1.0 + boost)
                boosted.append((item_id, fused_score))
            fused = sorted(boosted, key=lambda x: x[1], reverse=True)

        # Path penalty: down-weight tests/benchmarks/docs so source ranks above them
        penalized: list[tuple[str, float]] = []
        for item_id, fused_score in fused:
            result = result_map.get(item_id)
            if result:
                fused_score *= self._path_rank_penalty(result.file_path)
            penalized.append((item_id, fused_score))
        fused = sorted(penalized, key=lambda x: x[1], reverse=True)

        # Return top-k by fused score
        merged: list[SemanticSearchResult] = []
        for item_id, fused_score in fused[:top_k]:
            result = result_map.get(item_id)
            if result:
                result.score = round(fused_score, 4)
                merged.append(result)

        return merged

    def _scoped_literal_results(self, query, file_filter):
        if not file_filter or not self._kw_index_built:
            return []
        from attocode_intel.search_evidence import exact_scoped_source
        hit = exact_scoped_source(self.root_dir, query, file_filter,
                                  {doc.file_path for doc in self._kw_docs})
        return [hit] if hit is not None else []

    def search_candidates(
        self, query: str, top_k: int = 50, file_filter: str = "",
        trace: dict[str, list[str]] | None = None, tests_last: bool = True,
    ) -> list[SemanticSearchResult]:
        """Return local lexical candidates without initializing an embedding model.

        Name/path/docstring BM25 and source-body FTS5 each contribute a ranked
        list. Explicit exclusions only down-rank matches; they never remove a
        candidate. This is also the provider-free entry point for bootstrap.

        A ``trace`` dict receives the file order of each stage. The keys are
        ``keyword``, ``body``, ``file_bm25``, ``chunk_fused`` (before whole-file
        fusion) and ``fused`` (before the broad-query rerank).

        ``tests_last`` applies the source/test intent partition (``demote_tests``):
        source first normally, tests first for short explicit test requests.
        Task scores for test suggestions turn it off.
        """
        if top_k <= 0:
            return []
        from attocode_intel.focused_evidence import scope_exclusion_terms, task_terms

        positive, _ = task_terms(query)
        negative = scope_exclusion_terms(query)
        # The indexes keep plurals and match whole words: send "settings" as written too.
        positive |= task_terms(query, fold=False)[0]
        terms = sorted(positive) if positive else _tokenize(query)
        if not terms:
            if not self._kw_index_built:
                self._schedule_body_index()
            return self._scoped_literal_results(query, file_filter)
        if not self._kw_index_built:
            self._schedule_body_index()
            return []
        cfg = self.scoring_config
        lexical_query = " ".join(terms)
        wide_k = max(top_k * 3, 60)
        keyword = self._keyword_search(lexical_query, wide_k, file_filter)
        body = self._body_search(terms, wide_k, file_filter)
        files = self._file_search(terms, cfg.file_fusion_depth, file_filter)
        if not self._kw_index_built:
            # A body hit failed freshness validation while keyword results
            # were being scored. Do not expose potentially stale candidates.
            self._schedule_body_index()
            return []
        if trace is not None:
            trace.update(keyword=_file_order(keyword), body=_file_order(body), file_bm25=files)
        if not body and not files:
            if not keyword:
                keyword = self._scoped_literal_results(query, file_filter)
            keyword = self._penalize_exclusions(keyword, negative)
            if trace is not None:
                trace["chunk_fused"] = _file_order(keyword)
            if tests_last:
                keyword = demote_tests(query, keyword)
            if trace is not None:
                trace["fused"] = _file_order(keyword)
            return rerank_broad_candidates(query, keyword, top_k, file_filter)

        def key(result: SemanticSearchResult) -> tuple[str, str, str, int, int]:
            return (result.file_path, result.chunk_type, result.name,
                    result.start_line or result.definition_start_line,
                    result.end_line or result.definition_end_line)

        # A metadata definition contributes once to its best-ranked body window.
        # Other windows remain distinct; overloads can match only their own span.
        body_by_owner: dict[tuple[str, str, str], list[SemanticSearchResult]] = {}
        for hit in body:
            body_by_owner.setdefault((hit.file_path, hit.chunk_type, hit.name), []).append(hit)
        candidates: dict[tuple[str, str, str, int, int], SemanticSearchResult] = {}
        scores: dict[tuple[str, str, str, int, int], float] = {}
        for weight, ranked in ((1.0, keyword), (cfg.body_weight, body)):
            seen = set()
            for rank, result in enumerate(ranked):
                item = key(result)
                if ranked is keyword and result.definition_start_line:
                    windows = body_by_owner.get((result.file_path, result.chunk_type, result.name), [])
                    match = next((h for h in windows if result.definition_start_line <= h.start_line
                                  <= h.end_line <= result.definition_end_line), None)
                    if match is not None:
                        item = key(match)
                if item in seen:
                    continue
                seen.add(item)
                scores[item] = scores.get(item, 0.0) + weight / (cfg.chunk_rrf_k + rank + 1)
                if item not in candidates or (ranked is body and result.start_line):
                    candidates[item] = result

        ordered = sorted(scores, key=lambda item: (-scores[item], item))
        # Preserve the established navigation order of named owners. Distinct
        # windows are evidence within an owner, not extra votes for its rank.
        # A definition's metadata can still support its best body window above.
        owner_scores: dict[tuple[str, str, str], float] = {}
        representatives = {}
        for weight, ranked in ((1.0, keyword), (cfg.body_weight, body)):
            seen_chunks, seen_owners = set(), set()
            for rank, result in enumerate(ranked):
                owner = (result.file_path, result.chunk_type, result.name)
                item = key(result)
                if item in seen_chunks or (ranked is body and owner in seen_owners):
                    continue
                seen_chunks.add(item)
                seen_owners.add(owner)
                owner_scores[owner] = owner_scores.get(owner, 0.0) + weight / (cfg.chunk_rrf_k + rank + 1)
                if ranked is keyword or owner not in representatives:
                    representatives[owner] = item
                if ranked is body:
                    representatives[owner] = item
        heads = []
        for owner in sorted(owner_scores, key=lambda item: (-owner_scores[item], item)):
            item = representatives[owner]
            # A keyword identity may have fused into its corresponding window.
            if item not in candidates:
                item = next(k for k in ordered if k[:3] == owner)
            heads.append(replace(candidates[item], score=round(owner_scores[owner], 6)))
            representatives[owner] = item
        selected = set(representatives.values())
        floor = heads[-1].score if heads else 0.0
        fused = heads + [replace(candidates[item], score=min(round(scores[item], 6), floor))
                         for item in ordered if item not in selected]
        fused = self._penalize_exclusions(fused, negative)
        if trace is not None:
            trace["chunk_fused"] = _file_order(fused)
        if files:
            fused = self._fuse_file_order(fused, files)
        if tests_last:
            fused = demote_tests(query, fused)
        if trace is not None:
            trace["fused"] = _file_order(fused)
        return rerank_broad_candidates(query, fused, top_k, file_filter)

    def _fuse_file_order(
        self, results: list[SemanticSearchResult], files: list[str],
    ) -> list[SemanticSearchResult]:
        """Fuse the file order of ``results`` with whole-file BM25.

        This is the Loc-Bench eval arm: equal-weight RRF (k=60) of the two
        file orders, each cut to 48 files. Each fused file's first entry moves
        to the front in fused order, and a file that only BM25 found gets an
        entry. The other entries follow in their old order, scored no higher.
        """
        cfg = self.scoring_config
        first: dict[str, SemanticSearchResult] = {}
        for result in results:
            first.setdefault(result.file_path, result)
        fused: dict[str, float] = {}
        for order in (list(first)[:cfg.file_fusion_depth], files[:cfg.file_fusion_depth]):
            for rank, path in enumerate(order):
                fused[path] = fused.get(path, 0.0) + 1.0 / (cfg.file_rrf_k + rank + 1)
        heads = []
        for path in sorted(fused, key=lambda p: (-fused[p], p)):
            head = first.get(path) or self._file_entry(path)
            if head is not None:
                heads.append(replace(head, score=round(fused[path], 6)))
        moved = {id(first[path]) for path in fused if path in first}
        floor = heads[-1].score if heads else 0.0
        return heads + [
            replace(result, score=min(result.score, floor))
            for result in results if id(result) not in moved
        ]

    @staticmethod
    def _penalize_exclusions(
        results: list[SemanticSearchResult], negative: set[str],
    ) -> list[SemanticSearchResult]:
        if not negative:
            return results
        penalized: list[SemanticSearchResult] = []
        for result in results:
            identity = set(_tokenize(f"{result.file_path} {result.name}"))
            source = set(_tokenize(result.text))
            # Explicit negatives such as "without swarm" also cover compound
            # path segments like ``attoswarm``; this is only a soft penalty.
            identity_hits = sum(
                any(term == token or (len(term) >= 4 and term in token) for token in identity)
                for term in negative
            )
            source_hits = len(source & negative)
            factor = 0.25 ** identity_hits * 0.8 ** source_hits
            penalized.append(replace(result, score=round(result.score * factor, 6)))
        return sorted(penalized, key=lambda result: -result.score)

    def _source_revision(self) -> str:
        """Detect checkouts across branches without trusting stale cached rows."""
        try:
            result = subprocess.run(
                ["git", "rev-parse", "HEAD"], cwd=self.root_dir,
                capture_output=True, text=True, timeout=1, check=False,
            )
            return result.stdout.strip() if result.returncode == 0 else ""
        except (OSError, subprocess.TimeoutExpired):
            return ""

    def _open_body_db(self) -> sqlite3.Connection | None:
        """Open the optional FTS5 index; classic BM25 remains the fallback."""
        # Schema setup and invalidation share the same lock as index writes.
        # Otherwise initial schema creation can race a background warm-up.
        with self._kw_cache_lock:
            return self._open_body_db_unlocked()

    def _open_body_db_unlocked(self) -> sqlite3.Connection | None:
        if not self._body_index_available:
            return None
        conn = self._open_kw_cache_db()
        if conn is None:
            return None
        try:
            version_row = conn.execute(
                "SELECT value FROM metadata WHERE key = 'body_schema_version'",
            ).fetchone()
            if version_row and (not version_row[0].isdigit() or int(version_row[0]) > 5):
                conn.close()
                return None
            if version_row is None or version_row[0] != "5":
                conn.execute("DROP TABLE IF EXISTS body_fts")
                conn.execute("DROP TABLE IF EXISTS body_file_fts")
                conn.execute("DROP TABLE IF EXISTS body_files")
            conn.execute("""
                CREATE TABLE IF NOT EXISTS body_files (
                    file_path TEXT PRIMARY KEY, mtime_ns INTEGER NOT NULL,
                    size INTEGER NOT NULL
                )
            """)
            conn.execute("""
                CREATE VIRTUAL TABLE IF NOT EXISTS body_fts USING fts5(
                    file_path UNINDEXED, chunk_type UNINDEXED, name UNINDEXED,
                    start_line UNINDEXED, end_line UNINDEXED,
                    body UNINDEXED, terms
                )
            """)
            # One row per file: BM25 over a whole file counts matches that
            # are spread over many chunks.
            conn.execute("""
                CREATE VIRTUAL TABLE IF NOT EXISTS body_file_fts USING fts5(
                    file_path UNINDEXED, terms
                )
            """)
            if version_row is None or version_row[0] != "5":
                conn.execute(
                    "INSERT OR REPLACE INTO metadata (key, value) "
                    "VALUES ('body_schema_version', '5')",
                )
                conn.commit()
            return conn
        except sqlite3.OperationalError as exc:
            logger.debug("Source-body cache unavailable", exc_info=True)
            # Busy/locked databases are transient, not evidence that SQLite
            # lacks FTS5. Keep warm-up retryable after those failures.
            if "no such module: fts5" in str(exc).lower():
                self._body_index_available = False
            conn.close()
            return None

    @staticmethod
    def _body_chunks(rel_path: str, abs_path: str, body_spans=None) -> list[tuple[str, str, int, int, str, str]]:
        """Produce bounded, line-accurate source windows with enclosing symbols."""
        return SemanticSearchManager._body_document(rel_path, abs_path, body_spans)[0]

    @staticmethod
    def _body_document(rel_path: str, abs_path: str, body_spans=None, *, ast_lookup=None,
                       include_file_terms=False) -> tuple[list[tuple[str, str, int, int, str, str]], str]:
        """Share one file read and line tokens across overlapping windows and whole-file FTS."""
        from attocode_intel._internal.integrations.context.codebase_ast import parse_file

        tokenize = instrument("body.tokenize", _tokenize)
        try:
            with measure('body.read'), open(abs_path, encoding="utf-8", errors="replace") as source:
                content = source.read()
                lines = content.splitlines()
        except OSError:
            return [], ""
        count("body.read_chars", len(content))
        if not lines:
            return [], ""

        with measure('body.spans'):
            symbol_spans, class_spans = [], []
            try:
                cached = body_spans.take(rel_path, content) if body_spans is not None else None
                if cached is not None:
                    symbol_spans, class_spans = cached
                else:
                    ast = None
                    if ast_lookup is not None:
                        try:
                            ast = ast_lookup(rel_path, hashlib.sha256(content.encode()).hexdigest())
                        except Exception:
                            logger.debug("Existing AST lookup failed for %s", rel_path, exc_info=True)
                    symbol_spans, class_spans = _source_spans(ast or parse_file(abs_path, content))
            except Exception:
                logger.debug("Source window AST parse failed for %s", rel_path, exc_info=True)

        chunks: list[tuple[str, str, int, int, str, str]] = []
        # Bound retention to this one source file (body discovery caps it at
        # 512 KiB). Truncated lines must be tokenized separately: cutting an
        # identifier can change its final token and camel-case boundaries.
        tokenized_lines: list[str | None] = [None] * len(lines)

        # Generated source repeats identical lines across different windows/spans.
        # Reuse exact displayed text only, within one file. Bound both entries and
        # retained characters; long/unique lines continue through the original path.
        line_terms: dict[str, str] = {}
        remaining_chars = 32_768

        def tokenize_line(shown: str) -> str:
            nonlocal remaining_chars
            cached = line_terms.get(shown)
            if cached is not None:
                return cached
            terms = " ".join(tokenize(shown))
            size = len(shown) + len(terms)
            if len(line_terms) < 1024 and len(shown) <= 512 and size <= remaining_chars:
                line_terms[shown] = terms
                remaining_chars -= size
            return terms

        def terms_for_line(index: int, shown: str) -> str:
            if shown != lines[index]:
                return tokenize_line(shown)
            terms = tokenized_lines[index]
            if terms is None:
                terms = tokenize_line(shown)
                tokenized_lines[index] = terms
            return terms

        def emit(first: int, last: int, chunk_type: str, name: str) -> None:
            """Emit windows within one owner span (1-based inclusive)."""
            start = max(0, first - 1)
            stop = min(len(lines), last)
            while start < stop:
                shown: list[str] = []
                chars = 0
                for line in lines[start:min(start + 32, stop)]:
                    remaining = 2400 - chars
                    if remaining <= 0:
                        break
                    shown.append(line[:remaining])
                    chars += min(len(line), remaining) + 1
                    if len(line) >= remaining:
                        break
                if not shown:
                    break
                end = start + len(shown)
                body = "\n".join(shown)
                terms = " ".join(filter(None, (terms_for_line(start + i, line) for i, line in enumerate(shown))))
                if terms:
                    chunks.append((chunk_type, name, start + 1, end, body, terms))
                # Four-line overlap catches matches near a window boundary.
                start = max(start + 1, end - 4) if end < stop else end

        covered = [False] * len(lines)
        for first, last, chunk_type, name in symbol_spans:
            if first <= 0 or last < first:
                continue
            emit(first, last, chunk_type, name)
            for index in range(first - 1, min(last, len(lines))):
                covered[index] = True

        # Imports, constants, class attributes, and parser-unsupported files
        # still have searchable source windows, but cannot impersonate methods.
        cursor = 0
        while cursor < len(lines):
            if covered[cursor]:
                cursor += 1
                continue
            first = cursor
            while cursor < len(lines) and not covered[cursor]:
                cursor += 1
            middle = (first + cursor + 1) // 2
            enclosing = [span for span in class_spans if span[0] <= middle <= span[1]]
            if enclosing:
                _first, _last, chunk_type, name = min(
                    enclosing, key=lambda span: (span[1] - span[0], span[0]),
                )
            else:
                chunk_type, name = "file", os.path.basename(rel_path)
            emit(first + 1, cursor, chunk_type, name)
        file_terms = " ".join(filter(None, (terms_for_line(i, line) for i, line in enumerate(lines)))) if include_file_terms else ""
        return chunks, file_terms

    def _sync_body_index(self, context_manager: Any = None, body_spans=None) -> None:
        """Refresh changed/deleted discovered files, including branch changes."""
        if self._closed:
            return
        generation = self._body_generation
        revision = measured_call("body.revision", self._source_revision)
        if self._body_index_built and revision == self._body_revision:
            return
        conn = measured_call("body.open", self._open_body_db)
        if conn is None:
            return
        from attocode_intel._internal.integrations.context.codebase_context import (
            EXTENSION_LANGUAGES,
            CodebaseContextManager,
        )

        execute = instrument("body.sql", conn.execute)
        try:
            ctx = context_manager or CodebaseContextManager(root_dir=self.root_dir)
            with measure('body.discovery'):
                ctx._ensure_fresh()
                if not ctx._files:
                    ctx.discover_files()
            current: dict[str, tuple[str, int, int]] = {}
            root_real = os.path.realpath(self.root_dir)
            with measure('body.stat'):
                for file_info in ctx._files:
                    if self._closed:
                        return
                    try:
                        # Discovery can include symlinks. Never read or persist
                        # source from outside the selected workspace.
                        if os.path.commonpath((root_real, os.path.realpath(file_info.path))) != root_real:
                            continue
                        stat = os.stat(file_info.path)
                    except (OSError, ValueError):
                        continue
                    # Avoid loading generated or giant files into the interactive index.
                    if stat.st_size <= 512 * 1024:
                        current[file_info.relative_path] = (
                            file_info.path, stat.st_mtime_ns, stat.st_size,
                        )
            with measured_lock(self._kw_cache_lock, "body.lock"):
                with measure('body.cache_load'):
                    cached = {
                        path: (mtime, size)
                        for path, mtime, size in execute(
                            "SELECT file_path, mtime_ns, size FROM body_files",
                        )
                    }
                deleted = set(cached) - set(current)
                stale = {
                    path for path, (_abs, mtime, size) in current.items()
                    if cached.get(path) != (mtime, size)
                }
                with measure('body.delete'):
                    for paths in _sql_batches(sorted(deleted | stale)):
                        if self._closed:
                            return
                        placeholders = ",".join(["?"] * len(paths))
                        for table in ("body_fts", "body_file_fts", "body_files"):
                            execute(f"DELETE FROM {table} WHERE file_path IN ({placeholders})", paths)
                with measure('body.files'):
                    for path in sorted(stale):
                        if self._closed:
                            return
                        count("body.files_started")
                        abs_path, mtime, size = current[path]
                        ext = os.path.splitext(path)[1].lower()
                        with measure('body.document', path):
                            chunks, file_terms = self._body_document(
                                path, abs_path, body_spans, ast_lookup=self.ast_lookup,
                                include_file_terms=ext in EXTENSION_LANGUAGES and ext not in _NON_CODE_EXTS,
                            )
                        count("body.windows", len(chunks))
                        # One SQLite step per bounded batch avoids releasing and
                        # reacquiring the GIL for every window during AST hydration.
                        # Keep row order and the surrounding atomic transaction.
                        for offset in range(0, len(chunks), 64):
                            if self._closed:
                                return
                            batch = chunks[offset:offset + 64]
                            execute(
                                "INSERT INTO body_fts "
                                "(file_path, chunk_type, name, start_line, end_line, body, terms) "
                                "VALUES " + ",".join(["(?, ?, ?, ?, ?, ?, ?)"] * len(batch)),
                                tuple(value for chunk in batch for value in (path, *chunk)),
                            )
                        # Source files only: docs and data stay in the chunk search.
                        if file_terms:
                            execute(
                                "INSERT INTO body_file_fts (file_path, terms) VALUES (?, ?)",
                                (path, file_terms),
                            )
                        execute(
                            "INSERT INTO body_files (file_path, mtime_ns, size) VALUES (?, ?, ?)",
                            (path, mtime, size),
                        )
                        count("body.files_completed")
                if self._closed:
                    return  # Closing the connection rolls back this partial refresh.
                measured_call("body.commit", conn.commit)
                count("body.commits")
            self._body_index_built = generation == self._body_generation
            self._body_revision = revision
        except Exception:
            logger.debug("Failed to refresh source-body index", exc_info=True)
        finally:
            conn.close()

    def _schedule_body_index(self) -> None:
        """Warm both lexical indexes once without delaying navigation."""
        if self._kw_index_built and not self._body_index_available:
            return
        with self._body_state_lock:
            if self._closed:
                return
            if self._body_thread is not None and self._body_thread.is_alive():
                return
            if self._body_failed_at and time.monotonic() - self._body_failed_at < 5.0:
                return
            worker = threading.Thread(
                target=self._warm_candidate_indexes, daemon=True,
                name="lexical-candidate-indexer",
            )
            self._body_thread = worker
            worker.start()

    def warm(self) -> None:
        """Start the lexical warm-up in the background, unless both indexes are ready."""
        if not (self._kw_index_built and (self._body_index_built or not self._body_index_available)):
            self._schedule_body_index()

    def _warm_candidate_indexes(self) -> None:
        # One fresh profile per attempt; ordinary clients retain no diagnostics.
        if self.index_probe is not None:
            self.index_probe = IndexProbe()
        with activate(self.index_probe):
            self._warm_candidate_indexes_impl()

    def _warm_candidate_indexes_impl(self) -> None:
        """Build AST and optional body indexes, honoring mid-build invalidation."""
        generation = self._body_generation
        self._warmup_finished = 0.0
        self._warmup_started = time.monotonic()
        self._warmup_seconds = {}
        self._warmup_reused_asts = 0

        def timed(stage, operation):
            self._warmup_stage = stage
            started = time.monotonic()
            try:
                operation()
            finally:
                elapsed = time.monotonic() - started
                if stage == "keyword":
                    # Importance is now published inside the keyword builder;
                    # keep the reported phase times exclusive as before.
                    elapsed -= self._warmup_seconds.get("importance", 0.0)
                self._warmup_seconds = {**self._warmup_seconds, stage: max(0.0, elapsed)}

        try:
            from attocode_intel._internal.integrations.context.codebase_context import (
                CodebaseContextManager,
            )

            # One file discovery serves all three steps. A new context per step
            # discovered the files and built the dependency graph three times.
            ctx = CodebaseContextManager(root_dir=self.root_dir, cancelled=lambda: self._closed)
            if self.discovery_snapshot is not None:
                with measure("keyword.discovery_reuse"):
                    try:
                        reused = ctx.reuse_discovery(self.discovery_snapshot())
                    except InterruptedError:
                        raise
                    except Exception:
                        logger.debug("Discovery handoff unavailable; using fresh discovery", exc_info=True)
                        reused = False
                    if reused:
                        count("keyword.discovery_reused")
            if self._closed:
                return
            body_spans = _BodySpanCache()
            if not self._kw_index_built:
                timed("keyword", lambda: self._build_keyword_index(ctx, body_spans))
                if self._closed or generation != self._body_generation:
                    self._kw_index_built = False
                    return
            else:
                timed("importance", lambda: self._load_importance_scores(ctx))
            if self._closed:
                return
            if self._body_index_available and not self._body_index_built:
                timed("body", lambda: self._sync_body_index(ctx, body_spans))
        except InterruptedError:
            return
        except Exception:
            logger.warning("Lexical candidate warm-up failed", exc_info=True)
        finally:
            self._warmup_finished = time.monotonic()
            if self._closed:
                self._warmup_stage = "cancelled"
            elif generation != self._body_generation:
                self._warmup_stage = "invalidated"
            elif self._kw_index_built and (self._body_index_built or not self._body_index_available):
                self._warmup_stage = "ready"
            if not self._closed and generation == self._body_generation and not (
                self._kw_index_built
                and (self._body_index_built or not self._body_index_available)
            ):
                self._body_failed_at = time.monotonic()

    def wait_for_body_index(self, timeout: float = 30.0) -> bool:
        """Wait for background warm-up in tests and explicit offline evaluations."""
        worker = self._body_thread
        if worker is not None:
            worker.join(timeout=timeout)
        return self._body_index_built

    def _body_search(
        self, query_terms: list[str], top_k: int, file_filter: str,
    ) -> list[SemanticSearchResult]:
        """Find body-only implementations; all MATCH terms are tokenizer-made."""
        if top_k <= 0:
            return []
        if not self._body_index_built or self._source_revision() != self._body_revision:
            self._body_index_built = False
            self._schedule_body_index()
            return []
        conn = self._open_body_db()
        if conn is None:
            return []
        import fnmatch

        tokens = list(dict.fromkeys(t for term in query_terms for t in _tokenize(term)))
        if not tokens:
            conn.close()
            return []
        try:
            # Filter before the SQL limit. Applying a glob only to the fetched
            # rows can hide all matches from a requested language or subtree.
            where = "WHERE terms MATCH ?"
            parameters: list[Any] = []
            if file_filter:
                where += " AND body_fts.file_path GLOB ?"
                parameters.append(file_filter.replace("[!", "[^"))
            parameters.append(max(top_k * 8, 120))
            limit = self.scoring_config.body_max_tokens
            with self._kw_cache_lock:
                if len(tokens) > limit:
                    tokens = self._rarest_body_tokens(conn, tokens, limit)
                if not tokens:
                    return []
                parameters.insert(0, " OR ".join(f'"{token}"' for token in tokens))
                rows = conn.execute(
                    "SELECT body_fts.file_path, chunk_type, name, start_line, end_line, "
                    "body, bm25(body_fts), body_files.mtime_ns, body_files.size "
                    "FROM body_fts JOIN body_files "
                    "ON body_fts.file_path = body_files.file_path "
                    + where + " ORDER BY bm25(body_fts), body_fts.file_path, "
                    "body_fts.start_line, body_fts.name LIMIT ?",
                    parameters,
                ).fetchall()
        except sqlite3.Error:
            logger.debug("Source-body query failed; using AST keyword results", exc_info=True)
            return []
        finally:
            conn.close()

        selected: dict[tuple[str, str, str, int, int], SemanticSearchResult] = {}
        files: dict[str, int] = {}
        valid_paths: dict[str, bool] = {}
        stale_paths: set[str] = set()
        # Keep the previous first-hit owner coverage under the per-file cap.
        # Repeated windows fill spare capacity instead of hiding other methods.
        owners = set()
        primary, additional = [], []
        for row in rows:
            owner = tuple(row[:3])
            (additional if owner in owners else primary).append(row)
            owners.add(owner)
        for path, chunk_type, name, start, end, body, rank, mtime_ns, size in [*primary, *additional]:
            if file_filter and not fnmatch.fnmatch(path, file_filter):
                continue
            if path not in valid_paths:
                try:
                    stat = os.stat(os.path.join(self.root_dir, path))
                    valid_paths[path] = (stat.st_mtime_ns, stat.st_size) == (mtime_ns, size)
                except OSError:
                    valid_paths[path] = False
            if not valid_paths[path]:
                stale_paths.add(path)
                continue
            key = (path, chunk_type, name, int(start), int(end))
            if key in selected or files.get(path, 0) >= self.scoring_config.max_chunks_per_file:
                continue
            files[path] = files.get(path, 0) + 1
            selected[key] = SemanticSearchResult(
                file_path=path, chunk_type=chunk_type, name=name,
                text=f"{path}:{start}-{end}\n{body}",
                score=max(0.0, -float(rank)),
                start_line=int(start), end_line=int(end),
            )
            if len(selected) >= top_k:
                break
        if stale_paths:
            for path in stale_paths:
                self.invalidate_file(os.path.join(self.root_dir, path))
            return []
        return list(selected.values())

    def _file_search(self, query_terms: list[str], top_k: int, file_filter: str) -> list[str]:
        """Rank whole source files by BM25 over their terms; fresh files only.

        Only for a query of more than 20 words, where the chunk search keeps
        only part of the query. On a short query, BM25 length normalization
        prefers tiny files that mention a name over the file that defines it.
        """
        tokens = list(dict.fromkeys(t for term in query_terms for t in _tokenize(term)))
        if (top_k <= 0 or len(tokens) <= self.scoring_config.file_bm25_min_tokens
                or not self._body_index_built):
            return []
        if self._source_revision() != self._body_revision:
            # As in body search: report the index as stale, not a full result.
            self._body_index_built = False
            self._schedule_body_index()
            return []
        conn = self._open_body_db()
        if conn is None:
            return []
        import fnmatch

        try:
            where = "WHERE terms MATCH ?"
            parameters: list[Any] = []
            if file_filter:
                where += " AND body_file_fts.file_path GLOB ?"
                parameters.append(file_filter.replace("[!", "[^"))
            parameters.append(top_k)
            with self._kw_cache_lock:
                if len(tokens) > 64:
                    tokens = self._rarest_body_tokens(conn, tokens, 64)
                if not tokens:
                    return []
                parameters.insert(0, " OR ".join(f'"{token}"' for token in tokens))
                rows = conn.execute(
                    "SELECT body_file_fts.file_path, body_files.mtime_ns, body_files.size "
                    "FROM body_file_fts JOIN body_files "
                    "ON body_file_fts.file_path = body_files.file_path "
                    + where + " ORDER BY bm25(body_file_fts), body_file_fts.file_path LIMIT ?",
                    parameters,
                ).fetchall()
        except sqlite3.Error:
            logger.debug("Whole-file query failed; using chunk results", exc_info=True)
            return []
        finally:
            conn.close()
        files = []
        for path, mtime_ns, size in rows:
            if file_filter and not fnmatch.fnmatch(path, file_filter):
                continue
            try:
                stat = os.stat(os.path.join(self.root_dir, path))
            except OSError:
                continue
            # A changed file is skipped here; the body search schedules its refresh.
            if (stat.st_mtime_ns, stat.st_size) == (mtime_ns, size):
                files.append(path)
        return files

    def _file_entry(self, path: str) -> SemanticSearchResult | None:
        """A result for a file that only whole-file BM25 found: its first lines."""
        try:
            with open(os.path.join(self.root_dir, path), encoding="utf-8", errors="replace") as source:
                lines = [line.rstrip("\n") for _, line in zip(range(30), source, strict=False)]
        except OSError:
            return None
        return SemanticSearchResult(
            file_path=path, chunk_type="file", name=os.path.basename(path),
            text=f"{path}:1-{len(lines)}\n" + "\n".join(lines), score=0.0,
            start_line=1, end_line=max(len(lines), 1),
        )

    @staticmethod
    def _rarest_body_tokens(conn: sqlite3.Connection, tokens: list[str], limit: int) -> list[str]:
        """Keep the words of a long task that the fewest source chunks contain.

        The MATCH stays bounded. Without document counts, a long issue sent the
        words that came first, and those were often common ones.
        """
        try:
            conn.execute("CREATE VIRTUAL TABLE IF NOT EXISTS temp.body_vocab "
                         "USING fts5vocab(main, body_fts, row)")
            marks = ",".join("?" * len(tokens))
            counts = dict(conn.execute(
                f"SELECT term, doc FROM temp.body_vocab WHERE term IN ({marks})", tokens,
            ).fetchall())
        except sqlite3.Error:
            logger.debug("Source-body term counts unavailable", exc_info=True)
            return tokens[:limit]
        # A word that no chunk holds cannot match. Ties keep the query order.
        return sorted((t for t in tokens if t in counts), key=counts.__getitem__)[:limit]

    def candidate_diagnostics(self) -> dict[str, int | float | bool | str]:
        """Read-only coverage snapshot for compact search provenance."""
        ready = self._kw_index_built and (
            self._body_index_built or not self._body_index_available
        )
        snapshot: dict[str, int | float | bool | str] = {
            "status": "ready" if ready else "warming",
            "keyword_index_ready": self._kw_index_built,
            "body_index_available": self._body_index_available,
            "body_index_ready": self._body_index_built,
            "body_indexed_files": 0,
            "keyword_indexed_files": 0,
            "body_index_coverage": 0.0,
            "vector_status": self._index_progress.status,
            "vector_coverage": self._index_progress.coverage,
            "warmup_stage": self._warmup_stage,
            "warmup_reused_asts": self._warmup_reused_asts,
            "warmup_seconds": round((self._warmup_finished or time.monotonic()) - self._warmup_started, 3)
                if self._warmup_started else 0.0,
        }
        snapshot.update({f"{stage}_seconds": round(seconds, 3)
                         for stage, seconds in self._warmup_seconds.items()})
        if not self._body_index_built:
            return snapshot
        conn = self._open_body_db()
        if conn is None:
            snapshot["body_index_available"] = False
            return snapshot
        try:
            body_count = int(conn.execute("SELECT count(*) FROM body_files").fetchone()[0])
            keyword_count = int(conn.execute("SELECT count(*) FROM kw_files").fetchone()[0])
            snapshot.update(
                body_indexed_files=body_count,
                keyword_indexed_files=keyword_count,
                body_index_coverage=round(body_count / keyword_count, 3)
                if keyword_count else 0.0,
            )
        except sqlite3.Error:
            logger.debug("Failed to read candidate coverage", exc_info=True)
        finally:
            conn.close()
        return snapshot

    def _normalize_scores(
        self,
        ranked: list[tuple[str, float]],
        method: str = "min_max",
    ) -> list[tuple[str, float]]:
        """Normalize scores to [0, 1] range for fair fusion.

        Ensures that vector similarity scores and BM25 scores are on
        comparable scales before Reciprocal Rank Fusion.

        Args:
            ranked: List of (id, score) tuples.
            method: Normalization method (currently only "min_max").

        Returns:
            List of (id, normalized_score) tuples in the same order.
        """
        if not ranked or len(ranked) < 2:
            return ranked
        scores = [s for _, s in ranked]
        min_s, max_s = min(scores), max(scores)
        if max_s == min_s:
            return [(key, 1.0) for key, _ in ranked]
        return [(key, (s - min_s) / (max_s - min_s)) for key, s in ranked]

    def _rerank(
        self,
        query: str,
        fused: list[tuple[str, float]],
        result_map: dict[str, SemanticSearchResult],
        top_k: int,
    ) -> list[tuple[str, float]]:
        """Apply cross-encoder reranking to fused results.

        Args:
            query: The search query.
            fused: List of (id, fused_score) from RRF.
            result_map: Lookup from item ID to SemanticSearchResult.
            top_k: Number of final results desired.

        Returns:
            Reranked list of (id, score) tuples.
        """
        from attocode_intel._internal.integrations.context.reranker import get_reranker

        reranker = get_reranker()

        # Build candidates: (id, text, fused_score)
        candidates: list[tuple[str, str, float]] = []
        for item_id, score in fused:
            result = result_map.get(item_id)
            if result:
                candidates.append((item_id, result.text, score))

        if not candidates:
            return fused[:top_k]

        reranked = reranker.rerank(query, candidates, top_k=top_k)
        return [(cid, score) for cid, _text, score in reranked]

    # ------------------------------------------------------------------
    # Trigram pre-filtering for keyword search
    # ------------------------------------------------------------------

    def _get_trigram_index(self) -> Any | None:
        """Get a loaded TrigramIndex instance, or None if unavailable."""
        if self._trigram_index is not None:
            return self._trigram_index
        try:
            from attocode_intel._internal.integrations.context.trigram_index import TrigramIndex

            index_dir = os.path.join(self.root_dir, ".attocode", "index")
            if not os.path.isdir(index_dir):
                return None
            idx = TrigramIndex(index_dir=index_dir)
            if idx.load():
                self._trigram_index = idx
                return idx
        except Exception:
            logger.debug("Failed to load trigram index for kw pre-filter", exc_info=True)
        return None

    def _trigram_prefilter(self, query_tokens: list[str]) -> set[str] | None:
        """Use trigram index to find candidate files containing query terms.

        Returns set of candidate file paths, or None to fall back to full scan.
        Uses UNION semantics: a file matching ANY query term is included.
        """
        idx = self._get_trigram_index()
        if idx is None or not idx.is_ready():
            return None

        filterable = [t for t in query_tokens if len(t) >= 3]
        if not filterable:
            return None

        candidates: set[str] = set()
        any_definitive = False

        for token in filterable:
            escaped = re.escape(token)
            try:
                result = idx.query(escaped, selectivity_threshold=0.5)
            except Exception:
                logger.debug("Trigram query failed for '%s'", token, exc_info=True)
                return None

            if result is None:
                continue  # too common or no trigrams — skip this token
            any_definitive = True
            candidates.update(result)

        if not any_definitive:
            return None  # no token could be filtered, full scan needed

        return candidates

    def _keyword_search(
        self,
        query: str,
        top_k: int,
        file_filter: str,
    ) -> list[SemanticSearchResult]:
        """BM25 content-aware keyword search using AST-extracted data."""
        if not self._kw_index_built:
            self._build_keyword_index()

        if not self._kw_docs:
            return []

        import fnmatch

        query_tokens = _tokenize(query)
        if not query_tokens:
            return []

        # Trigram pre-filter: narrow to files containing query terms
        candidate_files = self._trigram_prefilter(query_tokens)
        if candidate_files is not None:
            docs_to_score = [
                d for d in self._kw_docs if d.file_path in candidate_files
            ]
            logger.debug(
                "Trigram pre-filter: %d/%d docs from %d candidate files",
                len(docs_to_score), len(self._kw_docs), len(candidate_files),
            )
        else:
            docs_to_score = self._kw_docs

        query_lower = query.lower()
        cfg = self.scoring_config

        # BM25 parameters — N and avg_dl use FULL corpus for correct IDF
        k1 = cfg.bm25_k1
        b = cfg.bm25_b
        N = len(self._kw_docs)  # noqa: N806
        avg_dl = self._kw_avg_dl or 1.0

        scored: list[tuple[float, _KeywordDoc]] = []
        for doc in docs_to_score:
            if file_filter and not fnmatch.fnmatch(doc.file_path, file_filter):
                continue

            # BM25 score
            score = 0.0
            for term in query_tokens:
                tf = doc.term_freqs.get(term, 0)
                if tf == 0:
                    continue
                df = self._kw_df.get(term, 0)
                if df == 0:
                    continue
                idf = math.log((N - df + 0.5) / (df + 0.5) + 1.0)
                dl = doc.doc_len or 1
                tf_norm = (tf * (k1 + 1)) / (tf + k1 * (1 - b + b * dl / avg_dl))
                score += idf * tf_norm

            if score <= 0:
                continue

            # Post-hoc boosts
            _name_lower = doc.name.lower()
            name_tokens = _tokenize(doc.name)

            # Graduated symbol name match boost
            _name_boost_applied = False
            for term in query_tokens:
                if term == _name_lower:
                    score *= cfg.name_exact_boost
                    _name_boost_applied = True
                    break
                if term in _name_lower:
                    score *= cfg.name_substring_boost
                    _name_boost_applied = True
                    break
            if not _name_boost_applied:
                for term in query_tokens:
                    if term in name_tokens:
                        score *= cfg.name_token_boost
                        break

            # Definition-type boost: classes and functions rank above file-level
            if doc.chunk_type == "class":
                score *= cfg.class_boost
            elif doc.chunk_type == "function":
                score *= cfg.function_boost
            elif doc.chunk_type == "method":
                score *= cfg.method_boost

            # Path relevance boost: source directories rank higher
            _SRC_DIRS = {"src", "lib", "pkg", "core", "internal", "app", "main"}  # noqa: N806
            _first_dir = doc.file_path.split("/")[0] if "/" in doc.file_path else ""
            if _first_dir.lower() in _SRC_DIRS:
                score *= cfg.src_dir_boost

            # Multi-term coverage bonus
            if len(query_tokens) > 1:
                _matched_terms = sum(1 for t in query_tokens if doc.term_freqs.get(t, 0) > 0)
                _coverage = _matched_terms / len(query_tokens)
                if _coverage >= cfg.multi_term_high_threshold:
                    score *= cfg.multi_term_high_bonus
                elif _coverage >= cfg.multi_term_med_threshold:
                    score *= cfg.multi_term_med_bonus

            # Non-code file penalty (markdown, text, config formats)
            _ext = os.path.splitext(doc.file_path)[1].lower()
            if _ext in _NON_CODE_EXTS:
                score *= cfg.non_code_penalty

            # Config/doc file penalty (stacks with non-code ext penalty)
            if doc.is_config:
                score *= cfg.config_penalty

            # Test file mild penalty
            if doc.is_test:
                score *= cfg.test_penalty

            # Exact phrase bonus: multi-word query substring match in text
            if len(query_tokens) > 1 and query_lower in doc.text.lower():
                score *= cfg.exact_phrase_bonus

            # File importance boost (PageRank + hub score from dependency graph)
            if cfg.importance_weight > 0:
                imp = self._get_importance(doc.file_path)
                score *= (1.0 + imp * cfg.importance_weight)

            scored.append((score, doc))

        if not scored:
            return []

        # Equal BM25 scores must not inherit filesystem/discovery insertion
        # order. Otherwise a cold rebuild can change ranks without code edits.
        # Span-qualified persistence IDs must not reorder ordinary lexical ties.
        scored.sort(key=lambda x: (-x[0], x[1].file_path,
                                  x[1].id.rsplit(":", 2)[0] if x[1].start_line else x[1].id))

        # Normalize to 0-1
        max_score = scored[0][0] if scored else 1.0
        if max_score <= 0:
            max_score = 1.0

        # Deduplicate: max N chunks per file
        file_counts: dict[str, int] = {}
        results: list[SemanticSearchResult] = []
        for raw_score, doc in scored:
            if file_counts.get(doc.file_path, 0) >= cfg.max_chunks_per_file:
                continue
            file_counts[doc.file_path] = file_counts.get(doc.file_path, 0) + 1
            results.append(SemanticSearchResult(
                file_path=doc.file_path,
                chunk_type=doc.chunk_type,
                name=doc.name,
                text=doc.text,
                score=round(raw_score / max_score, 4),
                definition_start_line=doc.start_line, definition_end_line=doc.end_line,
            ))
            if len(results) >= top_k:
                break

        return results

    # ------------------------------------------------------------------
    # BM25 keyword index disk cache
    # ------------------------------------------------------------------

    def _open_kw_cache_db(self) -> sqlite3.Connection | None:
        """Open the keyword index cache database. Returns None on failure."""
        with self._kw_cache_lock:
            return self._open_kw_cache_db_unlocked()

    def _open_kw_cache_db_unlocked(self) -> sqlite3.Connection | None:
        conn = None
        try:
            cache_dir = os.path.dirname(self._kw_cache_db_path)
            root_real = os.path.realpath(self.root_dir)
            if os.path.commonpath((root_real, os.path.realpath(os.path.dirname(cache_dir)))) != root_real:
                return None
            os.makedirs(cache_dir, mode=0o700, exist_ok=True)
            if os.path.commonpath((root_real, os.path.realpath(cache_dir))) != root_real:
                return None
            os.chmod(cache_dir, 0o700)
            flags = os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0)
            fd = os.open(self._kw_cache_db_path, flags, 0o600)
            try:
                os.fchmod(fd, 0o600)
            finally:
                os.close(fd)
            conn = sqlite3.connect(self._kw_cache_db_path, check_same_thread=False)
            conn.execute("PRAGMA journal_mode=WAL")
            conn.execute("PRAGMA synchronous=NORMAL")
            conn.execute("CREATE TABLE IF NOT EXISTS metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
            row = conn.execute("SELECT value FROM metadata WHERE key='schema_version'").fetchone()
            if row is not None and (not row[0].isdigit() or int(row[0]) > 4):
                conn.close()
                return None
            if row is None or row[0] != "4":
                # v2 IDs collapsed overloads and stored no definition spans; v3 docs
                # lacked Python docstrings. Reparse rather than patching persisted rows.
                conn.executescript("DROP TABLE IF EXISTS kw_files; DROP TABLE IF EXISTS kw_docs;")
            conn.executescript("""
                CREATE TABLE IF NOT EXISTS kw_files (
                    file_path TEXT PRIMARY KEY, mtime REAL NOT NULL
                );
                CREATE TABLE IF NOT EXISTS kw_docs (
                    id TEXT PRIMARY KEY, file_path TEXT NOT NULL,
                    chunk_type TEXT NOT NULL, name TEXT NOT NULL,
                    text TEXT NOT NULL,
                    is_config INTEGER NOT NULL DEFAULT 0,
                    is_test INTEGER NOT NULL DEFAULT 0,
                    term_freqs TEXT NOT NULL, doc_len INTEGER NOT NULL DEFAULT 0,
                    start_line INTEGER NOT NULL DEFAULT 0, end_line INTEGER NOT NULL DEFAULT 0
                );
                CREATE INDEX IF NOT EXISTS ix_kw_docs_file ON kw_docs(file_path);
            """)
            conn.execute("INSERT OR REPLACE INTO metadata (key, value) VALUES ('schema_version', '4')")
            conn.commit()
            return conn
        except Exception:
            logger.debug("Failed to open kw cache db", exc_info=True)
            if conn is not None:
                conn.close()
            return None

    def _save_kw_cache(
        self,
        docs: list[_KeywordDoc],
        file_mtimes: dict[str, float],
    ) -> None:
        """Persist keyword index docs to disk cache."""
        if self._closed:
            return
        conn = self._open_kw_cache_db()
        if conn is None:
            return
        execute = instrument("keyword.sql", conn.execute)
        encode = instrument("keyword.serialize", json.dumps)
        try:
            with self._kw_cache_lock:
                execute("DELETE FROM kw_files")
                execute("DELETE FROM kw_docs")
                for batch in _sql_batches(file_mtimes.items()):
                    if self._closed:
                        return
                    execute(
                        "INSERT INTO kw_files (file_path, mtime) VALUES "
                        + ",".join(["(?, ?)"] * len(batch)),
                        tuple(value for row in batch for value in row),
                    )
                rows = (
                    (doc.id, doc.file_path, doc.chunk_type, doc.name,
                     doc.text, int(doc.is_config), int(doc.is_test),
                     encode(doc.term_freqs), doc.doc_len, doc.start_line, doc.end_line)
                    for doc in docs
                )
                for batch in _sql_batches(rows):
                    if self._closed:
                        return
                    execute(
                        "INSERT OR REPLACE INTO kw_docs "
                        "(id, file_path, chunk_type, name, text, is_config, "
                        "is_test, term_freqs, doc_len, start_line, end_line) VALUES "
                        + ",".join(["(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"] * len(batch)),
                        tuple(value for row in batch for value in row),
                    )
                if self._closed:
                    return
                measured_call("keyword.commit", conn.commit)
                count("keyword.commits")
        except Exception:
            logger.debug("Failed to save kw cache", exc_info=True)
        finally:
            conn.close()

    def _load_kw_cache(
        self,
        current_files: dict[str, tuple[str, float]],
    ) -> tuple[list[_KeywordDoc], set[str]] | None:
        """Load cached keyword docs, identify files that need re-parsing.

        Args:
            current_files: dict of rel_path -> (abs_path, mtime)

        Returns:
            (cached_docs, files_to_parse) or None if cache unavailable.
        """
        conn = self._open_kw_cache_db()
        if conn is None:
            return None
        try:
            with self._kw_cache_lock:
                rows = conn.execute(
                    "SELECT file_path, mtime FROM kw_files",
                ).fetchall()
                if not rows:
                    conn.close()
                    return None

                cached_mtimes = {r[0]: r[1] for r in rows}

                # Identify stale (modified/new) and deleted files
                stale: set[str] = set()
                for fpath, (_, current_mtime) in current_files.items():
                    cached_mtime = cached_mtimes.get(fpath)
                    if cached_mtime is None or current_mtime > cached_mtime:
                        stale.add(fpath)
                deleted = set(cached_mtimes.keys()) - set(current_files.keys())
                exclude = stale | deleted

                # Load docs for unchanged files
                doc_rows = conn.execute(
                    "SELECT id, file_path, chunk_type, name, text, "
                    "is_config, is_test, term_freqs, doc_len, start_line, end_line FROM kw_docs",
                ).fetchall()

                docs: list[_KeywordDoc] = []
                for row in doc_rows:
                    if row[1] in exclude:
                        continue
                    docs.append(_KeywordDoc(
                        id=row[0], file_path=row[1], chunk_type=row[2],
                        name=row[3], text=row[4], is_config=bool(row[5]),
                        is_test=bool(row[6]),
                        term_freqs=json.loads(row[7]), doc_len=row[8],
                        start_line=row[9], end_line=row[10],
                    ))

            conn.close()
            return (docs, stale)
        except Exception:
            logger.debug("Failed to load kw cache", exc_info=True)
            try:
                conn.close()
            except Exception:
                pass
            return None

    # ------------------------------------------------------------------
    # BM25 keyword index builder (incremental with cache)
    # ------------------------------------------------------------------

    def _build_keyword_index(self, context_manager: Any = None, body_spans=None) -> None:
        """Build BM25 inverted index from AST-extracted data.

        Uses disk cache to avoid re-parsing unchanged files.
        """
        if self._closed:
            return
        from attocode_intel._internal.integrations.context.codebase_ast import parse_file
        from attocode_intel._internal.integrations.context.codebase_context import (
            CodebaseContextManager,
        )

        ctx = context_manager or CodebaseContextManager(root_dir=self.root_dir)
        tokenize = instrument("keyword.tokenize", _tokenize)
        with measure('keyword.discovery'):
            ctx._ensure_fresh()
            if not ctx._files:
                ctx.discover_files()

        _CONFIG_EXTS = frozenset({".toml", ".json", ".yaml", ".yml", ".cfg", ".ini", ".md", ".rst", ".txt"})  # noqa: N806
        _CONFIG_NAMES = frozenset({  # noqa: N806
            "pyproject.toml", "package.json", "tsconfig.json", "setup.cfg",
            "setup.py", "readme.md", "readme.rst", "changelog.md", "license",
        })

        # Discover current files with mtimes
        current_files: dict[str, tuple[str, float]] = {}
        file_meta: dict[str, tuple[bool, bool]] = {}  # rel -> (is_config, is_test)
        with measure('keyword.stat'):
            for f in ctx._files:
                if self._closed:
                    return
                try:
                    mtime = os.path.getmtime(f.path)
                except OSError:
                    continue
                current_files[f.relative_path] = (f.path, mtime)
                basename = os.path.basename(f.relative_path).lower()
                ext = os.path.splitext(basename)[1]
                is_config = ext in _CONFIG_EXTS or basename in _CONFIG_NAMES
                file_meta[f.relative_path] = (is_config, f.is_test)

        # Try incremental update from cache
        cache_result = measured_call("keyword.cache_load", self._load_kw_cache, current_files)

        if cache_result is not None:
            docs, files_to_parse = cache_result
            logger.debug(
                "kw_index incremental: %d cached docs, %d files to parse",
                len(docs), len(files_to_parse),
            )
        else:
            docs = []
            files_to_parse = set(current_files.keys())
            logger.debug("kw_index full rebuild: %d files", len(files_to_parse))

        # Parse only files that need updating
        with measure('keyword.files'):
            for rel in files_to_parse:
                if self._closed:
                    return
                count("keyword.files_started")
                entry = current_files.get(rel)
                if entry is None:
                    continue
                abs_path, _ = entry
                is_config, is_test = file_meta.get(rel, (False, False))

                try:
                    if body_spans is None and self.ast_lookup is None:
                        ast = measured_call("keyword.parse", parse_file, abs_path)
                    else:
                        with measure('keyword.read'), open(abs_path, encoding="utf-8", errors="replace") as source:
                            content = source.read()
                        count("keyword.read_chars", len(content))
                        with measure("keyword.hash_lookup"):
                            ast = self.ast_lookup(rel, hashlib.sha256(content.encode()).hexdigest()) if self.ast_lookup else None
                        if ast is None:
                            ast = measured_call("keyword.parse", parse_file, abs_path, content)
                        else:
                            self._warmup_reused_asts += 1
                        if body_spans is not None:
                            measured_call("keyword.spans", body_spans.put, rel, content, ast)
                except Exception:
                    continue

                with measure('keyword.documents', rel):
                    # File-level doc
                    file_text_parts = list(re.split(r"[/\\]", rel))
                    file_text_parts.extend(imp.module for imp in ast.imports[:30])
                    file_text_parts.extend(ast.get_symbols()[:30])
                    file_text = " ".join(file_text_parts)
                    file_tokens = tokenize(file_text)

                    file_tf: dict[str, int] = {}
                    for t in file_tokens:
                        file_tf[t] = file_tf.get(t, 0) + 1

                    docs.append(_KeywordDoc(
                        id=f"file:{rel}",
                        file_path=rel,
                        chunk_type="file",
                        name=os.path.basename(rel),
                        text=file_text[:200],
                        is_config=is_config,
                        is_test=is_test,
                        term_freqs=file_tf,
                        doc_len=len(file_tokens),
                    ))

                    # Function-level docs
                    for func in ast.functions:
                        parts: list[str] = []
                        parts.extend([func.name] * 3)
                        # Add NL summary of the function name for better query matching
                        nl_summary = _summarize_code_to_nl(func.name, "function", "")
                        parts.append(nl_summary)
                        if func.docstring:
                            parts.append(func.docstring[:300])
                        parts.extend(p.name for p in func.parameters[:10])
                        parts.extend(p.type_annotation for p in func.parameters[:10] if p.type_annotation)
                        if func.return_type:
                            parts.append(func.return_type)
                        parts.extend(func.decorators[:5])
                        text = " ".join(parts)
                        tokens = tokenize(text)

                        tf: dict[str, int] = {}
                        for t in tokens:
                            tf[t] = tf.get(t, 0) + 1

                        preview_parts = [f"function {func.name}"]
                        if func.docstring:
                            preview_parts.append(func.docstring[:100])
                        params_str = ", ".join(p.name for p in func.parameters[:6])
                        if params_str:
                            preview_parts.append(f"({params_str})")

                        docs.append(_KeywordDoc(
                            id=f"func:{rel}:{func.name}:{func.start_line}:{func.end_line}",
                            start_line=func.start_line, end_line=func.end_line,
                            file_path=rel,
                            chunk_type="function",
                            name=func.name,
                            text=" | ".join(preview_parts),
                            is_config=False,
                            is_test=is_test,
                            term_freqs=tf,
                            doc_len=len(tokens),
                        ))

                    # Class + method-level docs
                    for cls in ast.classes:
                        cls_parts: list[str] = [cls.name] * 3
                        # Add NL summary of the class name
                        cls_nl = _summarize_code_to_nl(cls.name, "class", "")
                        cls_parts.append(cls_nl)
                        if cls.bases:
                            cls_parts.extend(cls.bases)
                        if cls.docstring:
                            cls_parts.append(cls.docstring[:300])
                        cls_parts.extend(m.name for m in cls.methods[:15])
                        cls_text = " ".join(cls_parts)
                        cls_tokens = tokenize(cls_text)

                        cls_tf: dict[str, int] = {}
                        for t in cls_tokens:
                            cls_tf[t] = cls_tf.get(t, 0) + 1

                        cls_preview = f"class {cls.name}"
                        if cls.bases:
                            cls_preview += f"({', '.join(cls.bases[:3])})"
                        if cls.docstring:
                            cls_preview += f" | {cls.docstring[:100]}"

                        docs.append(_KeywordDoc(
                            id=f"cls:{rel}:{cls.name}:{cls.start_line}:{cls.end_line}",
                            start_line=cls.start_line, end_line=cls.end_line,
                            file_path=rel,
                            chunk_type="class",
                            name=cls.name,
                            text=cls_preview,
                            is_config=False,
                            is_test=is_test,
                            term_freqs=cls_tf,
                            doc_len=len(cls_tokens),
                        ))

                        for method in cls.methods:
                            m_parts: list[str] = [method.name] * 3
                            m_parts.append(cls.name)
                            if method.docstring:
                                m_parts.append(method.docstring[:300])
                            m_parts.extend(p.name for p in method.parameters[:10])
                            m_parts.extend(p.type_annotation for p in method.parameters[:10] if p.type_annotation)
                            if method.return_type:
                                m_parts.append(method.return_type)
                            m_text = " ".join(m_parts)
                            m_tokens = tokenize(m_text)

                            m_tf: dict[str, int] = {}
                            for t in m_tokens:
                                m_tf[t] = m_tf.get(t, 0) + 1

                            m_preview = f"method {cls.name}.{method.name}"
                            if method.docstring:
                                m_preview += f" | {method.docstring[:100]}"

                            docs.append(_KeywordDoc(
                                id=f"method:{rel}:{cls.name}.{method.name}:{method.start_line}:{method.end_line}",
                                start_line=method.start_line, end_line=method.end_line,
                                file_path=rel,
                                chunk_type="method",
                                name=f"{cls.name}.{method.name}",
                                text=m_preview,
                                is_config=False,
                                is_test=is_test,
                                term_freqs=m_tf,
                                doc_len=len(m_tokens),
                            ))
                count("keyword.files_completed")

        # Build document frequencies (always recomputed from full doc list)
        with measure('keyword.frequencies'):
            df: dict[str, int] = {}
            for doc in docs:
                if self._closed:
                    return
                for term in doc.term_freqs:
                    df[term] = df.get(term, 0) + 1

            total_len = sum(doc.doc_len for doc in docs)
            avg_dl = total_len / len(docs) if docs else 1.0

        self._kw_docs = docs
        self._kw_df = df
        self._kw_avg_dl = avg_dl
        # Keyword queries can run while persistence/body indexing continues.
        # Their weights must already exist, or _get_importance creates another
        # context and rediscovers/reparses the workspace on the request thread.
        # Warmup subtracts this subphase from the reported keyword duration.
        importance_started = time.monotonic()
        self._load_importance_scores(ctx)
        self._warmup_seconds = {**self._warmup_seconds,
                                "importance": time.monotonic() - importance_started}
        self._kw_index_built = True

        # Persist to cache
        file_mtimes = {rel: mt for rel, (_, mt) in current_files.items()}
        count("keyword.documents", len(docs))
        measured_call("keyword.cache_save", self._save_kw_cache, docs, file_mtimes)

        logger.debug(
            "BM25 keyword index built: %d docs, %d terms (%d files parsed)",
            len(docs), len(df), len(files_to_parse),
        )

    def _slice_body(
        self, file_lines: list[str], start_line: int, end_line: int, max_tokens: int,
    ) -> str:
        """Return the symbol's body text (lines after the signature line),
        trimmed to an approximate token budget.

        start_line/end_line are 1-based and inclusive (codebase_ast convention).
        The first line is the signature; the body is start_line+1..end_line.
        Returns "" when there is no body (one-line def) or budget <= 0.
        """
        from attocode_intel._internal.integrations.utilities.token_estimate import estimate_tokens

        if max_tokens <= 0:
            return ""
        n = len(file_lines)
        body_start = start_line  # 0-based index of first body line == 1-based start_line
        body_end = min(end_line, n)  # clamp past EOF; slice upper bound
        if body_start >= body_end:
            return ""
        body = file_lines[body_start:body_end]
        text = "\n".join(line.rstrip() for line in body).strip()
        if not text:
            return ""
        if estimate_tokens(text) <= max_tokens:
            return text
        # Trim from the head, keeping whole lines, until under budget.
        kept: list[str] = []
        for line in body:
            kept.append(line.rstrip())
            if estimate_tokens("\n".join(kept)) > max_tokens:
                kept.pop()
                break
        return "\n".join(kept).strip()

    def _chunk_single_file(
        self, rel_path: str, abs_path: str,
    ) -> list[tuple]:
        """Extract chunks from a single file.

        Returns a list of chunk tuples.  File- and class-level chunks are
        4-tuples ``(id, file_path, chunk_type, text)``.  Function- and
        method-level chunks are 5-tuples
        ``(id, file_path, chunk_type, display_text, embed_text)`` where
        ``display_text`` is the concise signature (stored for display) and
        ``embed_text`` additionally includes a token-budgeted body so the
        embedding model sees the implementation.
        """
        from attocode_intel._internal.integrations.context.codebase_ast import parse_file

        try:
            ast = parse_file(abs_path)
        except Exception:
            return []

        import os
        try:
            _body_budget = int(os.environ.get("ATTOCODE_BODY_TOKEN_BUDGET", "400"))
        except ValueError:
            _body_budget = 400
        try:
            with open(abs_path, encoding="utf-8", errors="ignore") as _fh:
                _file_lines = _fh.read().splitlines()
        except OSError:
            _file_lines = []

        chunks: list[tuple] = []

        # File-level summary
        imports = [imp.module for imp in ast.imports[:20]]
        symbols = ast.get_symbols()[:20]
        summary_parts = []
        file_doc = getattr(ast, "docstring", None)
        if file_doc:
            summary_parts.append(file_doc[:200])
        if imports:
            summary_parts.append(f"imports: {', '.join(imports)}")
        if symbols:
            summary_parts.append(f"defines: {', '.join(symbols)}")
        if summary_parts:
            chunks.append((
                f"file:{rel_path}",
                rel_path,
                "file",
                " | ".join(summary_parts),
            ))

        # Function-level chunks
        for func in ast.functions:
            text_parts = [f"function {func.name}"]
            if func.docstring:
                text_parts.append(func.docstring[:150])
            params = ", ".join(p.name for p in func.parameters[:8])
            if params:
                text_parts.append(f"params: {params}")
            if func.return_type:
                text_parts.append(f"returns: {func.return_type}")
            sig_text = " | ".join(text_parts)
            body = (
                self._slice_body(_file_lines, func.start_line, func.end_line, _body_budget)
                if _file_lines else ""
            )
            embed_text = f"{sig_text}\n{body}" if body else sig_text
            chunks.append((
                f"func:{rel_path}:{func.name}",
                rel_path,
                "function",
                sig_text,
                embed_text,
            ))

        # Class-level chunks (including method-level)
        for cls in ast.classes:
            text_parts = [f"class {cls.name}"]
            if cls.bases:
                text_parts.append(f"extends: {', '.join(cls.bases)}")
            if cls.docstring:
                text_parts.append(cls.docstring[:150])
            methods = [m.name for m in cls.methods[:8]]
            if methods:
                text_parts.append(f"methods: {', '.join(methods)}")
            chunks.append((
                f"cls:{rel_path}:{cls.name}",
                rel_path,
                "class",
                " | ".join(text_parts),
            ))

            # Method-level chunks for richer search
            for method in cls.methods:
                m_parts = [f"method {cls.name}.{method.name}"]
                if method.docstring:
                    m_parts.append(method.docstring[:150])
                m_params = ", ".join(p.name for p in method.parameters[:8])
                if m_params:
                    m_parts.append(f"params: {m_params}")
                if method.return_type:
                    m_parts.append(f"returns: {method.return_type}")
                m_sig = " | ".join(m_parts)
                m_body = (
                    self._slice_body(
                        _file_lines, method.start_line, method.end_line, _body_budget,
                    )
                    if _file_lines else ""
                )
                m_embed = f"{m_sig}\n{m_body}" if m_body else m_sig
                chunks.append((
                    f"method:{rel_path}:{cls.name}.{method.name}",
                    rel_path,
                    "method",
                    m_sig,
                    m_embed,
                ))

        return chunks

    def _get_embedding_texts(
        self, chunks: list[tuple], language: str = "python",
    ) -> list[str]:
        """Return texts to feed to the embedding model.

        When ``nl_mode`` is ``"heuristic"``, each chunk's text is transformed
        into a natural language summary via :class:`CodeSummarizer` before
        embedding.  The original chunk text (index 3) is still stored in
        ``VectorEntry.text`` for display.

        Args:
            chunks: List of ``(id, file_path, chunk_type, text)`` tuples as
                returned by :meth:`_chunk_single_file`.
            language: Programming language hint for the summarizer.

        Returns:
            A list of strings, one per chunk, suitable for embedding.
        """
        if self._summarizer is None:
            # No NL mode — embed body-augmented text (index 4) when present,
            # else the display text (index 3).
            return [c[4] if len(c) > 4 else c[3] for c in chunks]

        result: list[str] = []
        for chunk in chunks:
            cid = chunk[0]
            _fpath = chunk[1]
            ctype = chunk[2]
            text = chunk[3]
            # Derive the symbol name from the chunk ID
            name = cid.split(":")[-1] if ":" in cid else _fpath
            summary = self._summarizer.summarize(text, ctype, name, language)
            result.append(summary)
        return result

    def reindex_file(self, file_path: str) -> int:
        """Re-index a single file: delete old vectors and re-embed.

        Args:
            file_path: Absolute or relative path to the file.

        Returns:
            Number of chunks indexed (0 if skipped).
        """
        self._ensure_provider()
        if self._keyword_fallback or not self._store:
            return 0

        from attocode_intel._internal.integrations.context.vector_store import VectorEntry

        try:
            rel = os.path.relpath(file_path, self.root_dir)
        except ValueError:
            rel = file_path

        abs_path = os.path.join(self.root_dir, rel) if not os.path.isabs(file_path) else file_path

        if not os.path.isfile(abs_path):
            # File was deleted — just remove vectors
            self._store.delete_by_file(rel)
            self._store.delete_file_metadata(rel)
            return 0

        # Re-chunk
        chunks = self._chunk_single_file(rel, abs_path)
        if not chunks:
            self._store.delete_by_file(rel)
            self._store.delete_file_metadata(rel)
            return 0

        # Embed FIRST (before deleting old vectors) — if embed fails,
        # we keep existing vectors rather than losing them
        texts = self._get_embedding_texts(chunks)
        try:
            vectors = self._provider.embed(texts)
        except Exception:
            logger.warning(
                "reindex_file embed failed for %s, keeping old vectors",
                rel, exc_info=True,
            )
            return 0

        # Build entries
        entries = []
        for chunk, vec in zip(chunks, vectors, strict=False):
            if not vec:
                continue
            cid = chunk[0]
            fpath = chunk[1]
            ctype = chunk[2]
            text = chunk[3]
            entries.append(VectorEntry(
                id=cid,
                file_path=fpath,
                chunk_type=ctype,
                name=cid.split(":")[-1] if ":" in cid else fpath,
                text=text,
                vector=vec,
            ))

        # Now safe to delete old + upsert new
        self._store.delete_by_file(rel)
        if entries:
            self._store.upsert_batch(entries)

        # Update metadata
        try:
            mtime = os.path.getmtime(abs_path)
        except OSError:
            mtime = 0.0
        self._store.set_file_metadata(rel, mtime, len(entries))

        logger.debug("reindex_file: %s -> %d chunks", rel, len(entries))
        self._indexed = True
        return len(entries)

    def invalidate_file(self, file_path: str) -> None:
        """Remove embeddings for a changed file."""
        self._kw_index_built = False  # Force rebuild on next keyword search
        self._body_index_built = False  # Recheck this file in the source-body index
        self._body_generation += 1
        self._body_failed_at = 0.0
        if self._trigram_index is not None:
            self._trigram_index.close()
        self._trigram_index = None  # Reload on next use
        try:
            rel = os.path.relpath(file_path, self.root_dir)
        except ValueError:
            rel = file_path
        # Mark stale in keyword cache
        try:
            conn = self._open_kw_cache_db()
            if conn:
                with self._kw_cache_lock:
                    conn.execute(
                        "UPDATE kw_files SET mtime = 0 WHERE file_path = ?",
                        (rel,),
                    )
                    if conn.execute(
                        "SELECT 1 FROM sqlite_master WHERE name = 'body_files'",
                    ).fetchone():
                        conn.execute(
                            "UPDATE body_files SET mtime_ns = -1 WHERE file_path = ?",
                            (rel,),
                        )
                    conn.commit()
                conn.close()
        except Exception:
            pass
        if self._store:
            self._store.delete_by_file(rel)

    def reindex_stale_files(self, context_manager: Any = None) -> int:
        """Re-index only files whose mtime has changed since last indexing.

        Designed to be called on agent startup for incremental freshness.

        Returns:
            Number of chunks re-indexed.
        """
        self._ensure_provider()
        if self._keyword_fallback or not self._store:
            return 0

        from attocode_intel._internal.integrations.context.codebase_context import (
            CodebaseContextManager,
        )

        ctx = context_manager or CodebaseContextManager(root_dir=self.root_dir)
        ctx._ensure_fresh()
        if not ctx._files:
            ctx.discover_files()

        # Build mtime map for all discovered files
        file_mtimes: dict[str, float] = {}
        file_paths: dict[str, str] = {}  # rel -> abs
        for f in ctx._files:
            try:
                file_mtimes[f.relative_path] = os.path.getmtime(f.path)
                file_paths[f.relative_path] = f.path
            except OSError:
                continue

        stale = self._store.get_stale_files(file_mtimes)

        # Clean up vectors for files that have been deleted from disk
        all_indexed = self._store.get_all_indexed_files()
        current_files = set(file_mtimes.keys())
        deleted_count = 0
        for indexed_path in all_indexed:
            if indexed_path not in current_files:
                self._store.delete_by_file(indexed_path)
                self._store.delete_file_metadata(indexed_path)
                deleted_count += 1
                logger.debug("Cleaned up vectors for deleted file: %s", indexed_path)

        if not stale and not deleted_count:
            logger.debug("reindex_stale_files: all files up to date")
            return 0

        if deleted_count:
            logger.info("reindex_stale_files: cleaned up %d deleted files", deleted_count)

        logger.info("reindex_stale_files: %d stale files to re-index", len(stale))
        total_chunks = 0
        for rel_path in stale:
            abs_path = file_paths.get(rel_path)
            if abs_path:
                total_chunks += self.reindex_file(abs_path)

        self._indexed = True
        return total_chunks

    # ------------------------------------------------------------------
    # Background indexing (P3)
    # ------------------------------------------------------------------

    def is_index_ready(self) -> bool:
        """Check if the vector index has sufficient coverage (>=80%)."""
        self._ensure_provider()
        if self._keyword_fallback or not self._store:
            return False
        progress = self.get_index_progress()
        return progress.coverage >= 0.8

    def get_index_progress(self) -> IndexProgress:
        """Get current indexing progress (thread-safe snapshot)."""
        with self._reindex_lock:
            if self._bg_indexer is not None:
                # Return a snapshot to avoid races
                p = self._index_progress
                return IndexProgress(
                    total_files=p.total_files,
                    indexed_files=p.indexed_files,
                    failed_files=p.failed_files,
                    status=p.status,
                    coverage=p.coverage,
                    started_at=p.started_at,
                    elapsed_seconds=p.elapsed_seconds,
                )
        # If no background indexer, check store directly
        if self._store:
            # Use cached total if available to avoid expensive filesystem scan
            cached_total = self._index_progress.total_files
            if cached_total > 0:
                total = cached_total
            else:
                from attocode_intel._internal.integrations.context.codebase_context import (
                    CodebaseContextManager,
                )
                ctx = CodebaseContextManager(root_dir=self.root_dir)
                ctx._ensure_fresh()
                if not ctx._files:
                    ctx.discover_files()
                _supported = self._supported_languages()
                total = len([f for f in ctx._files if f.language in _supported])
                self._index_progress.total_files = total
            indexed = len(self._store.get_all_indexed_files()) if total > 0 else 0
            coverage = indexed / total if total > 0 else 0.0
            return IndexProgress(
                total_files=total,
                indexed_files=indexed,
                coverage=round(coverage, 3),
                status="completed" if coverage >= 0.99 else "idle",
            )
        return self._index_progress

    def start_background_indexing(self) -> IndexProgress:
        """Start background progressive embedding indexing."""
        import time

        self._ensure_provider()
        if self._keyword_fallback or not self._store:
            self._index_progress.status = "error"
            self._index_progress.degraded_reason = "embedding_provider_unavailable"
            self._index_progress.last_error = (
                "Background indexing unavailable because no embedding provider is configured."
            )
            return self._index_progress

        with self._reindex_lock:
            if self._bg_indexer is not None:
                return self._index_progress

            self._index_progress = IndexProgress(status="running", started_at=time.time())
            stop_event = threading.Event()

            def _worker() -> None:
                try:
                    self._run_background_indexing(stop_event)
                except Exception as exc:
                    logger.error("Background indexer crashed", exc_info=True)
                    with self._reindex_lock:
                        self._index_progress.status = "error"
                        self._index_progress.degraded_reason = "background_indexer_crashed"
                        self._index_progress.last_error = (
                            f"Background indexer crashed: {type(exc).__name__}: {exc}"
                        )
                finally:
                    with self._reindex_lock:
                        self._bg_indexer = None
                        self._bg_thread = None

            worker = threading.Thread(target=_worker, daemon=True, name="bg-embedding-indexer")
            self._bg_indexer = stop_event
            self._bg_thread = worker
            worker.start()
        return self._index_progress

    def stop_background_indexing(self) -> None:
        """Stop background indexing gracefully."""
        if self._bg_indexer is not None:
            self._bg_indexer.set()

    def _run_background_indexing(self, stop_event: threading.Event) -> None:
        """Background indexer loop: process files in batches."""
        import time

        from attocode_intel._internal.integrations.context.codebase_context import (
            CodebaseContextManager,
        )
        from attocode_intel._internal.integrations.context.vector_store import VectorEntry

        ctx = CodebaseContextManager(root_dir=self.root_dir)
        ctx._ensure_fresh()
        if not ctx._files:
            ctx.discover_files()

        _supported = self._supported_languages()
        if _supported == {"python", "javascript", "typescript"}:
            self._index_progress.degraded_reason = "tree_sitter_languages_unavailable"

        indexable = [f for f in ctx._files if f.language in _supported]
        self._index_progress.total_files = len(indexable)

        # Determine which files need indexing
        file_mtimes: dict[str, float] = {}
        for f in indexable:
            try:
                file_mtimes[f.relative_path] = os.path.getmtime(f.path)
            except OSError:
                pass

        stale = set(self._store.get_stale_files(file_mtimes))
        # Also include files not yet indexed at all
        all_indexed = set(self._store.get_all_indexed_files())
        to_index = [f for f in indexable if f.relative_path in stale or f.relative_path not in all_indexed]

        self._index_progress.indexed_files = len(indexable) - len(to_index)
        batch_size = 50

        for i in range(0, len(to_index), batch_size):
            if stop_event.is_set():
                self._index_progress.status = "paused"
                return

            batch = to_index[i:i + batch_size]
            for f in batch:
                if stop_event.is_set():
                    self._index_progress.status = "paused"
                    return
                try:
                    chunks = self._chunk_single_file(f.relative_path, f.path)
                    if not chunks:
                        self._index_progress.failed_files += 1
                        self._index_progress.last_error = (
                            f"No indexable chunks generated for {f.relative_path}"
                        )
                        continue

                    texts = self._get_embedding_texts(chunks)
                    vectors = self._provider.embed(texts)
                    entries = []
                    for chunk, vec in zip(chunks, vectors, strict=False):
                        if not vec:
                            continue
                        cid = chunk[0]
                        fpath = chunk[1]
                        ctype = chunk[2]
                        text = chunk[3]
                        entries.append(VectorEntry(
                            id=cid,
                            file_path=fpath,
                            chunk_type=ctype,
                            name=cid.split(":")[-1] if ":" in cid else fpath,
                            text=text,
                            vector=vec,
                        ))

                    if entries:
                        self._store.delete_by_file(f.relative_path)
                        self._store.upsert_batch(entries)
                        try:
                            mtime = os.path.getmtime(f.path)
                        except OSError:
                            mtime = 0.0
                        self._store.set_file_metadata(f.relative_path, mtime, len(entries))
                        with self._reindex_lock:
                            self._index_progress.indexed_files += 1
                    else:
                        with self._reindex_lock:
                            self._index_progress.failed_files += 1
                            self._index_progress.last_error = (
                                f"No vectors generated for {f.relative_path}"
                            )
                            self._index_progress.degraded_reason = "partial_index_failure"
                except Exception as exc:
                    logger.debug("bg_index_failed: %s", f.relative_path, exc_info=True)
                    with self._reindex_lock:
                        self._index_progress.failed_files += 1
                        self._index_progress.last_error = (
                            f"Failed to index {f.relative_path}: {type(exc).__name__}: {exc}"
                        )
                        self._index_progress.degraded_reason = "partial_index_failure"

            # Update coverage
            with self._reindex_lock:
                total = self._index_progress.total_files or 1
                self._index_progress.coverage = round(self._index_progress.indexed_files / total, 3)
                self._index_progress.elapsed_seconds = round(time.time() - self._index_progress.started_at, 1)

            # Yield CPU between batches
            time.sleep(0.1)

        with self._reindex_lock:
            self._index_progress.status = "completed"
            total = self._index_progress.total_files or 1
            self._index_progress.coverage = round(self._index_progress.indexed_files / total, 3)
            self._index_progress.elapsed_seconds = round(time.time() - self._index_progress.started_at, 1)
            if self._index_progress.failed_files > 0 and not self._index_progress.degraded_reason:
                self._index_progress.degraded_reason = "partial_index_failure"
        self._indexed = True
        logger.info(
            "Background indexing complete: %d/%d files (%.0f%% coverage)",
            self._index_progress.indexed_files,
            self._index_progress.total_files,
            self._index_progress.coverage * 100,
        )

    def close(self) -> None:
        """Close the vector store and stop background indexer."""
        with self._reindex_lock:
            if not self._closed:
                self._closed = True
                if self._reindex_thread is not None:
                    self._reindex_queue.put(None)
        self.stop_background_indexing()
        # Synchronize with scheduling so a worker cannot start after this join.
        with self._body_state_lock:
            body_thread = self._body_thread
        if body_thread is not None:
            body_thread.join(timeout=5.0)
            if body_thread.is_alive():
                raise RuntimeError("Source-body indexer did not stop; resources kept open")
        if self._reindex_thread is not None:
            self._reindex_thread.join(timeout=5.0)
            if self._reindex_thread.is_alive():
                raise RuntimeError("Reindex worker did not stop; store kept open")
        thread = self._bg_thread
        if thread is not None:
            thread.join(timeout=5.0)
            if thread.is_alive():
                raise RuntimeError("Background indexer did not stop; store kept open")
        if self._store:
            self._store.close()
        if self._trigram_index is not None:
            self._trigram_index.close()
            self._trigram_index = None

    def format_results(self, results: list[SemanticSearchResult], query: str = "") -> str:
        """Format search results as human-readable text."""
        if not results:
            return "No results found."

        lines = [f"Semantic search results ({len(results)}):"]
        for i, r in enumerate(results, 1):
            location = r.file_path
            if r.start_line > 0:
                location += f":{r.start_line}-{r.end_line or r.start_line}"
            lines.append(
                f"  {i}. [{r.chunk_type}] {location}"
                f" — {r.name} (score: {r.score:.3f})"
            )
            if query:
                evidence = hit_evidence(query, r)
                if evidence["matched_terms"]:
                    lines.append(
                        "     matches: " + ", ".join(evidence["matched_terms"])
                        + " (" + ", ".join(evidence["match_fields"]) + ")"
                    )
            if r.text:
                evidence = search_snippet(r, query, 120)
                preview = evidence["snippet"]
                if evidence["snippet_start_line"] is not None:
                    preview = f"snippet {evidence['snippet_start_line']}-{evidence['snippet_end_line']}: {preview}"
                if evidence["snippet_truncated"]:
                    preview += "..."
                lines.append(f"     {preview}")
        return "\n".join(lines)
