"""Search pipeline: BM25 + Dense + RRF fusion.

Stage order (keyword path, fixed — see ``CLAUDE.md`` invariants):
expansion → BM25 + dense (parallel, with always-on scope-context filter
per ADR-0011 §6) → RRF fusion → cross-encoder rerank (optional) →
source/tag filter → validity filter → time-decay → MMR → access-freq
boost → importance boost → context-window expansion.

The scope-context filter is applied **at the BM25 + dense storage
calls** (not as a post-fusion stage) because the SQL fragment is the
single chokepoint that prevents cross-project leak; running it at the
storage layer rather than after fusion guarantees no candidate ever
reaches the pipeline that the project-context boundary would have
excluded. Tie-break ordering ``project_local > project_shared > user``
applies at the same site (storage ORDER BY) so same-relevance ranks
surface freshest-context-first.

Empty-query path (``query=""`` / ``None`` with ``tag_filter`` /
``source_filter`` set, #750) routes through ``_filter_only_search``
and skips expansion / BM25 / dense / RRF / rerank / MMR — none of
those have a meaningful signal without a query. Validity → decay →
access → importance → context-window still apply so the rank
reflects recency × access × importance. The scope-context filter
applies in this branch too via ``recall_chunks``.
"""

from __future__ import annotations

import asyncio
import logging
from fnmatch import fnmatch
from pathlib import Path
from typing import TYPE_CHECKING

from dataclasses import dataclass

from dataclasses import replace as dataclass_replace
from uuid import UUID

from memtomem.config import (
    MAX_CONTEXT_WINDOW_CHUNKS,
    AccessConfig,
    ContextWindowConfig,
    DecayConfig,
    MMRConfig,
    RerankConfig,
    SearchConfig,
    SessionSummaryConfig,
)
from memtomem.models import ContextInfo, NamespaceFilter, ScopeFilter, SearchResult
from memtomem.search.fusion import reciprocal_rank_fusion

logger = logging.getLogger(__name__)


def _bg_task_error_cb(task: asyncio.Task) -> None:
    """Log errors from fire-and-forget background tasks."""
    if task.cancelled():
        return
    exc = task.exception()
    if exc:
        logger.warning("Background task %s failed: %s", task.get_name(), exc)


def match_source_filter(filter_str: str, source_path: str) -> bool:
    """Match source_filter: glob when pattern chars present, substring otherwise.

    Both sides are folded to forward-slash form before comparing so a
    user-typed ``/tmp/keep/`` matches a Windows-stored
    ``\\tmp\\keep\\file.md`` (#720, sibling of #647). On POSIX
    ``str.replace("\\", "/")`` is a no-op for the typical case where
    paths don't contain backslashes.

    The ``source_filter`` parameter is shared across the search pipeline
    and several MCP tools (``mem_list``, ``mem_consolidate``); this
    helper is the canonical matcher for callers whose contract is
    "substring or glob, autodetected by pattern chars". Callers with a
    stricter contract use one of the two siblings below
    (``match_source_filter_substring``, ``match_source_filter_glob``)
    so the separator-fold rule still lives in one module per contract
    while the contract differences stay explicit at the call site.

    POSIX edge case: backslash is a legal filename character on POSIX, so
    a chunk indexed under ``foo\\bar.md`` would match a filter
    ``foo/bar`` after the fold. This is rare in practice (most tools and
    users avoid backslashes in POSIX filenames) and is the trade-off for
    a Windows-portable comparison.
    """
    norm_filter = filter_str.replace("\\", "/")
    norm_source = source_path.replace("\\", "/")
    if any(c in norm_filter for c in ("*", "?", "[")):
        return fnmatch(norm_source, norm_filter)
    return norm_filter in norm_source


def match_source_filter_substring(filter_str: str, source_path: str) -> bool:
    """Substring-only variant of :func:`match_source_filter`.

    Same separator-fold rule (#720), no glob fallback. Used by callers
    whose contract is substring-only (``mem_decay`` /
    :func:`~memtomem.search.decay.expire_chunks`, ``mem_auto_tag``,
    ``mem_export_chunks``). Sharing the substring + glob auto-detecting
    helper would silently broaden their behaviour to glob, which is a
    contract change. The fold lives in this single helper so a future
    "tidy-up" that strips the ``.replace("\\", "/")`` calls fails the
    POSIX-runnable pin in ``TestMatchSourceFilterSubstring`` instead of
    only the Windows CI leg.
    """
    return filter_str.replace("\\", "/") in source_path.replace("\\", "/")


def match_source_filter_glob(filter_str: str, source_path: str) -> bool:
    """Glob-only variant of :func:`match_source_filter`.

    Same separator-fold rule (#720), no substring fallback. Used by
    callers whose contract is glob-only (``mem_entity_scan``); a
    substring filter that lacks ``*?[`` characters returns ``False``
    here unless it happens to be an exact-match glob, mirroring the
    pre-fix behaviour. The fold lives in this single helper so the
    POSIX-runnable pin in ``TestMatchSourceFilterGlob`` catches a
    revert without depending on the Windows CI leg.
    """
    return fnmatch(source_path.replace("\\", "/"), filter_str.replace("\\", "/"))


def _apply_validity_filter(results: list[SearchResult], as_of_unix: int) -> list[SearchResult]:
    """Drop chunks whose temporal-validity window excludes ``as_of_unix``.

    Inclusive on both ends (``valid_from <= as_of <= valid_to``); ``None`` on
    a bound means unbounded on that side; both ``None`` means always-valid
    (RFC §Comparison semantics — opt-in default for chunks without a window).

    Order is preserved. The function returns a new list so callers can
    chain it after source/tag filter without mutating the input.

    Granularity note: when the pipeline falls back to the default
    ``int(time.time())``, results may be served from the search-result TTL
    cache for up to ``cache_ttl`` seconds. A chunk whose window expires at
    midnight can therefore continue to surface for the cache-TTL window
    after expiry. This is acceptable because the RFC's date-only bounds
    already operate at 24h granularity; sub-minute drift is invisible at
    that resolution.
    """
    filtered: list[SearchResult] = []
    for r in results:
        vfrom = r.chunk.metadata.valid_from_unix
        vto = r.chunk.metadata.valid_to_unix
        if vfrom is None and vto is None:
            filtered.append(r)
            continue
        lower = vfrom if vfrom is not None else float("-inf")
        upper = vto if vto is not None else float("inf")
        if lower <= as_of_unix <= upper:
            filtered.append(r)
    return filtered


@dataclass
class RetrievalStats:
    bm25_candidates: int = 0
    dense_candidates: int = 0
    fused_total: int = 0
    final_total: int = 0
    bm25_error: str | None = None
    dense_error: str | None = None
    # Chunks that live in namespaces matching ``system_namespace_prefixes``
    # (e.g. ``archive:*``) and were therefore excluded from the default,
    # namespace=None search. Non-zero only when the caller did not pick an
    # explicit namespace — surfaces as a hint in mem_search's output so
    # users know their archived memories still exist.
    hidden_system_ns: int = 0


if TYPE_CHECKING:
    from memtomem.embedding.base import EmbeddingProvider
    from memtomem.llm.base import LLMProvider
    from memtomem.storage.base import StorageBackend


_EXPANSION_CACHE_MAX = 100


class SearchPipeline:
    def __init__(
        self,
        storage: StorageBackend,
        embedder: EmbeddingProvider,
        config: SearchConfig,
        decay_config: DecayConfig | None = None,
        mmr_config: MMRConfig | None = None,
        access_config: AccessConfig | None = None,
        reranker: object | None = None,
        rerank_config: RerankConfig | None = None,
        expansion_config: object | None = None,
        importance_config: object | None = None,
        context_window_config: ContextWindowConfig | None = None,
        llm_provider: LLMProvider | None = None,
        session_summary_config: SessionSummaryConfig | None = None,
    ) -> None:
        self._storage = storage
        self._embedder = embedder
        self._config = config
        self._decay_config = decay_config or DecayConfig()
        self._mmr_config = mmr_config or MMRConfig()
        self._access_config = access_config or AccessConfig()
        self._reranker = reranker
        self._rerank_config = rerank_config
        self._expansion_config = expansion_config
        self._importance_config = importance_config
        self._context_window_config = context_window_config
        self._llm_provider = llm_provider
        self._session_summary_config = session_summary_config

        # Search result TTL cache (per-instance) with version counter
        self._search_cache: dict[str, tuple[float, int, list[SearchResult], RetrievalStats]] = {}
        self._cache_ttl = config.cache_ttl
        self._cache_version = 0
        self._bg_tasks: set[asyncio.Task] = set()

        # LLM query expansion cache (cleared on invalidate_cache)
        self._expansion_cache: dict[str, str] = {}

    def _cache_key(
        self,
        query: str,
        top_k: int,
        source_filter: str | None,
        tag_filter: str | None,
        namespace: str | list[str] | None,
        context_window: int | None = None,
        scope: str | list[str] | None = None,
        project_context_root: Path | None = None,
    ) -> str:
        import hashlib

        ctx_win = self._resolve_context_window(context_window)
        if self._reranker is not None and self._rerank_config is not None:
            rcfg = self._rerank_config
            rerank_signal = f"on:{rcfg.oversample}:{rcfg.min_pool}:{rcfg.max_pool}"
        else:
            rerank_signal = "off"
        # ADR-0011: scope + project_context_root MUST participate in the cache
        # key. Two callers from different projects must not share a cache slot
        # — the always-on context-boundary fragment differs.
        scope_signal = f"{scope}|{project_context_root}"
        raw = (
            f"{query}|{top_k}|{source_filter}|{tag_filter}|{namespace}"
            f"|bm25={self._config.enable_bm25}:{self._config.bm25_candidates}"
            f"|dense={self._config.enable_dense}:{self._config.dense_candidates}"
            f"|rrf_k={self._config.rrf_k}|w={tuple(self._config.rrf_weights)}"
            f"|decay={self._decay_config.enabled}:{self._decay_config.half_life_days}"
            f"|mmr={self._mmr_config.enabled}:{self._mmr_config.lambda_param}"
            f"|ctx_win={ctx_win}"
            f"|rerank={rerank_signal}"
            f"|scope={scope_signal}"
        )
        return hashlib.md5(raw.encode()).hexdigest()

    def invalidate_cache(self) -> None:
        """Clear the search result TTL cache (call after data/config changes)."""
        self._cache_version += 1
        self._search_cache.clear()
        self._expansion_cache.clear()

    def _resolve_context_window(self, override: int | None) -> int:
        """Return the effective context window size (0 = disabled)."""
        if override is not None:
            return max(0, min(override, MAX_CONTEXT_WINDOW_CHUNKS))
        cfg = self._context_window_config
        if cfg and cfg.enabled:
            return max(0, min(cfg.window_size, MAX_CONTEXT_WINDOW_CHUNKS))
        return 0

    async def _session_summary_boost_sources(
        self,
        query: str,
        scope_filter: ScopeFilter | None = None,
        project_context_root: Path | None = None,
    ) -> set[str]:
        """Stage-1 enrichment lookup against ``archive:session:*``.

        Runs a small BM25 lookup against the session-summary namespace
        (top-k = ``expansion_lookup_top_k``) and, for each hit above
        ``expansion_score_threshold``, follows ``chunk_links`` of type
        ``"summarizes"`` from the summary chunk back to the source
        chunks it summarized. The set of source files spanned by those
        chunks becomes the boost-sources list.

        ADR-0011 PR-D review: ``scope_filter`` / ``project_context_root``
        are threaded through so the rescue lookup honors the same
        always-on scope-context fragment as the primary retrieval.
        Without these, the new storage default ("no project context →
        user only") would silently exclude project_shared /
        project_local summaries from the rescue path even when the
        outer search is pinned to a project.

        Returns an empty set when the feature is disabled, the lookup
        fails, no summary scores above threshold, or no surviving
        ``summarizes`` link points at any source. The caller treats an
        empty set as "skip the rescue leg".
        """
        cfg = self._session_summary_config
        if cfg is None or cfg.expansion_lookup_top_k <= 0:
            return set()

        archive_filter = NamespaceFilter(pattern="archive:session:*")
        try:
            summary_hits = await self._storage.bm25_search(
                query,
                top_k=cfg.expansion_lookup_top_k,
                namespace_filter=archive_filter,
                scope_filter=scope_filter,
                project_context_root=project_context_root,
            )
        except Exception:
            logger.debug("session-summary lookup failed; skipping rescue", exc_info=True)
            return set()

        threshold = cfg.expansion_score_threshold
        candidate_summaries = [r for r in summary_hits if r.score >= threshold]
        if not candidate_summaries:
            return set()

        # For each above-threshold summary, walk chunk_links(summarizes)
        # to reach the original source chunks, then collect their
        # source_file paths. Failures on any one summary are logged and
        # skipped — we never let a single bad summary mute the rescue.
        target_chunk_ids: list[UUID] = []
        for r in candidate_summaries:
            try:
                links = await self._storage.get_chunks_shared_from(
                    r.chunk.id, link_type="summarizes"
                )
            except Exception:
                logger.debug(
                    "get_chunks_shared_from failed for summary %s", r.chunk.id, exc_info=True
                )
                continue
            for link in links:
                target_chunk_ids.append(link.target_id)

        if not target_chunk_ids:
            return set()

        try:
            chunks_map = await self._storage.get_chunks_batch(target_chunk_ids)
        except Exception:
            logger.debug("get_chunks_batch failed for rescue targets", exc_info=True)
            return set()

        return {str(c.metadata.source_file) for c in chunks_map.values()}

    async def _rescue_retrieval(
        self,
        query: str,
        query_embedding: list[float],
        top_k: int,
        boost_sources: set[str],
        use_bm25: bool,
        use_dense: bool,
        scope_filter: ScopeFilter | None = None,
        project_context_root: Path | None = None,
    ) -> list[SearchResult]:
        """Parallel BM25+dense retrieval restricted to boost_sources.

        Runs unrestricted retrievals and post-filters by source_file
        membership rather than threading a new parameter through the
        storage primitives — keeps ``bm25_search`` / ``dense_search``
        signatures clean. The leg is merged into RRF as a third input
        list, weighted by ``expansion_rescue_weight``.

        ADR-0011 PR-D review: scope context is threaded through so the
        rescue legs honor the same always-on scope filter the primary
        retrieval uses. The default ("no project context → user only")
        would otherwise drop project_shared / project_local rescue
        candidates whenever the outer search was pinned to a project.
        """
        if not boost_sources:
            return []
        # Cast a slightly wider net so source filtering still leaves a
        # useful candidate pool. Bounded by existing candidate caps.
        oversample = max(top_k, self._config.bm25_candidates)

        async def _bm25_leg() -> list[SearchResult]:
            if not use_bm25:
                return []
            try:
                hits = await self._storage.bm25_search(
                    query,
                    top_k=oversample,
                    namespace_filter=None,
                    scope_filter=scope_filter,
                    project_context_root=project_context_root,
                )
            except Exception:
                logger.debug("rescue bm25 leg failed", exc_info=True)
                return []
            return [r for r in hits if str(r.chunk.metadata.source_file) in boost_sources]

        async def _dense_leg() -> list[SearchResult]:
            if not use_dense or not query_embedding:
                return []
            try:
                hits = await self._storage.dense_search(
                    query_embedding,
                    top_k=oversample,
                    namespace_filter=None,
                    scope_filter=scope_filter,
                    project_context_root=project_context_root,
                )
            except Exception:
                logger.debug("rescue dense leg failed", exc_info=True)
                return []
            return [r for r in hits if str(r.chunk.metadata.source_file) in boost_sources]

        bm25_rescue, dense_rescue = await asyncio.gather(_bm25_leg(), _dense_leg())

        # Merge the rescue leg's BM25 + dense candidates into a single
        # ranked list (keep best rank across the two legs) so we feed
        # one rescue list into the outer RRF rather than two — this
        # matches the RFC's "third input list" framing and avoids
        # over-weighting rescue when it dominates both legs.
        seen: dict[UUID, SearchResult] = {}
        for r in bm25_rescue + dense_rescue:
            existing = seen.get(r.chunk.id)
            if existing is None or r.score > existing.score:
                seen[r.chunk.id] = r
        return sorted(seen.values(), key=lambda r: r.score, reverse=True)[:oversample]

    async def _expand_context(self, results: list[SearchResult], window: int) -> list[SearchResult]:
        """Attach ±window adjacent chunks to each result (batch, single DB call)."""
        if not results or window <= 0:
            return results

        source_files = list({r.chunk.metadata.source_file for r in results})
        chunks_by_source = await self._storage.list_chunks_by_sources(source_files)

        # Build per-file index: {chunk_id -> position}
        file_indexes: dict[str, dict[str, int]] = {}
        for sf, chunks in chunks_by_source.items():
            file_indexes[str(sf)] = {str(c.id): i for i, c in enumerate(chunks)}

        expanded: list[SearchResult] = []
        for r in results:
            sf_key = str(r.chunk.metadata.source_file)
            idx_map = file_indexes.get(sf_key)
            if idx_map is None:
                expanded.append(r)
                continue
            pos = idx_map.get(str(r.chunk.id))
            if pos is None:
                expanded.append(r)
                continue

            file_chunks = chunks_by_source[r.chunk.metadata.source_file]
            before = file_chunks[max(0, pos - window) : pos]
            after = file_chunks[pos + 1 : pos + 1 + window]

            expanded.append(
                SearchResult(
                    chunk=r.chunk,
                    score=r.score,
                    rank=r.rank,
                    source=r.source,
                    context=ContextInfo(
                        window_before=tuple(before),
                        window_after=tuple(after),
                        chunk_position=pos + 1,
                        total_chunks_in_file=len(file_chunks),
                        context_tier_used="standard",
                    ),
                )
            )
        return expanded

    async def _filter_only_search(
        self,
        *,
        top_k: int | None,
        source_filter: str | None,
        tag_filter: str | None,
        namespace: str | list[str] | None,
        context_window: int | None,
        as_of_unix: int | None,
        scope_filter: ScopeFilter | None = None,
        project_context_root: Path | None = None,
    ) -> tuple[list[SearchResult], RetrievalStats]:
        """Empty-query path (#750): enumerate by filter, skip retrievers.

        Tag and/or source filter become the primary selectors via
        ``recall_chunks``. Each candidate enters with ``score=1.0`` so
        the post-filter stages (validity → decay → access → importance)
        produce a meaningful order even without a keyword to rank by.
        MMR and the reranker are skipped — both need a query/embedding
        signal that doesn't exist in this mode.
        """
        import time

        top_k = self._config.default_top_k if top_k is None else top_k
        ns_filter = NamespaceFilter.parse(
            namespace,
            system_prefixes=tuple(self._config.system_namespace_prefixes),
        )

        # Over-sample so the validity stage can prune without starving
        # the response. ``top_k * 5`` mirrors the rerank-pool floor logic
        # used in the keyword path; floored at 100 for tiny ``top_k``.
        # Worst case ≈ 2500 candidates given the route's ``top_k <= 500``
        # cap, which is the same order as ``bm25_candidates``.
        #
        # On a popular tag (chunk count > candidate_limit) older highly
        # accessed/important chunks beyond this recency cutoff never reach
        # the boost stages — ``recall_chunks`` orders ``created_at DESC``,
        # so this is a "browse newest by tag" UX, not "rank everything by
        # importance". The keyword path doesn't have this floor because
        # BM25/dense pre-filter on relevance; here recency is the only
        # signal we have to cap on. Acceptable for the click-the-pill
        # workflow; revisit if a "show me old-but-important by tag" need
        # surfaces.
        candidate_limit = max(top_k * 5, 100)
        chunks = await self._storage.recall_chunks(
            source_filter=source_filter,
            tag_filter=tag_filter,
            namespace_filter=ns_filter,
            scope_filter=scope_filter,
            project_context_root=project_context_root,
            limit=candidate_limit,
        )

        fused: list[SearchResult] = [
            SearchResult(chunk=c, score=1.0, rank=i + 1, source="recall")
            for i, c in enumerate(chunks)
        ]
        stats = RetrievalStats(fused_total=len(fused))

        effective_as_of = as_of_unix if as_of_unix is not None else int(time.time())
        if fused:
            fused = _apply_validity_filter(fused, effective_as_of)

        if self._decay_config.enabled and fused:
            from memtomem.search.decay import apply_score_decay

            fused = apply_score_decay(fused, half_life_days=self._decay_config.half_life_days)

        if self._access_config.enabled and fused:
            from memtomem.search.access import apply_access_boost

            access_chunk_ids = [r.chunk.id for r in fused]
            access_counts = await self._storage.get_access_counts(access_chunk_ids)
            fused = apply_access_boost(
                fused, access_counts, max_boost=self._access_config.max_boost
            )

        if self._importance_config and getattr(self._importance_config, "enabled", False) and fused:
            from memtomem.search.importance import apply_importance_boost

            chunk_ids_imp = [r.chunk.id for r in fused]
            imp_scores = await self._storage.get_importance_scores(chunk_ids_imp)
            fused = apply_importance_boost(
                fused,
                imp_scores,
                max_boost=getattr(self._importance_config, "max_boost", 1.5),
            )

        # Re-sort after boosts and trim to ``top_k``. Boost stages mutate
        # scores but preserve list order; the keyword path relies on the
        # reranker / RRF for final ordering, so this branch sorts here.
        fused.sort(key=lambda r: r.score, reverse=True)
        fused = fused[:top_k]

        ctx_win = self._resolve_context_window(context_window)
        if ctx_win > 0 and fused:
            fused = await self._expand_context(fused, ctx_win)

        stats.final_total = len(fused)

        if fused:

            async def _increment():
                await self._storage.increment_access([r.chunk.id for r in fused])

            t = asyncio.create_task(_increment())
            t.add_done_callback(_bg_task_error_cb)
            self._bg_tasks.add(t)
            t.add_done_callback(self._bg_tasks.discard)

        return fused, stats

    async def search(
        self,
        query: str,
        top_k: int | None = None,
        source_filter: str | None = None,
        tag_filter: str | None = None,
        namespace: str | list[str] | None = None,
        rrf_weights: list[float] | None = None,
        context_window: int | None = None,
        as_of_unix: int | None = None,
        scope: str | list[str] | None = None,
        project_context_root: Path | None = None,
    ) -> tuple[list[SearchResult], RetrievalStats]:
        # #750: tag/source-only branch — no keyword to rank by, so the
        # filter takes over as the primary selector. We enumerate via
        # ``recall_chunks`` and skip BM25/dense/expansion/rescue/rerank
        # entirely; post-filter stages (validity, decay, access,
        # importance, ctx-window) still apply so ranking reflects
        # recency × access × importance.
        scope_filter = ScopeFilter.parse(scope)
        query = (query or "").strip()
        if not query:
            if not (tag_filter or source_filter):
                return [], RetrievalStats()
            return await self._filter_only_search(
                top_k=top_k,
                source_filter=source_filter,
                tag_filter=tag_filter,
                namespace=namespace,
                context_window=context_window,
                as_of_unix=as_of_unix,
                scope_filter=scope_filter,
                project_context_root=project_context_root,
            )

        top_k = self._config.default_top_k if top_k is None else top_k
        effective_weights = rrf_weights or self._config.rrf_weights

        # Check TTL cache for identical queries
        import time

        # ``as_of_unix`` is intentionally excluded from ``cache_key`` and
        # bypasses the cache entirely when explicit. Default-path (None)
        # callers fall through to ``int(time.time())`` below and reuse
        # cached results within ``cache_ttl`` — accepting up-to-TTL
        # staleness near a date boundary, which the RFC's date-only
        # bounds already absorb.
        cache_key = self._cache_key(
            query,
            top_k,
            source_filter,
            tag_filter,
            namespace,
            context_window,
            scope=scope,
            project_context_root=project_context_root,
        )
        version_at_start = self._cache_version
        ttl_snapshot = self._cache_ttl
        if as_of_unix is None and cache_key in self._search_cache:
            ts, ver, cached_results, cached_stats = self._search_cache[cache_key]
            if ver == self._cache_version and time.time() - ts < ttl_snapshot:
                return cached_results, cached_stats
            self._search_cache.pop(cache_key, None)

        bm25_k = max(self._config.bm25_candidates, top_k)
        dense_k = max(self._config.dense_candidates, top_k)
        ns_filter = NamespaceFilter.parse(
            namespace,
            system_prefixes=tuple(self._config.system_namespace_prefixes),
        )

        # When the caller did not pin a namespace, count how many chunks sit
        # behind a system-namespace prefix (e.g. archive:*) so the tool layer
        # can hint "N hidden — pass namespace=... to include them".
        hidden_system_ns = 0
        if namespace is None and self._config.system_namespace_prefixes:
            try:
                hidden_system_ns = await self._storage.count_chunks_by_ns_prefix(
                    list(self._config.system_namespace_prefixes)
                )
            except Exception:
                logger.debug("count_chunks_by_ns_prefix failed; skipping hint", exc_info=True)

        use_bm25 = self._config.enable_bm25
        use_dense = self._config.enable_dense

        # Stage 0: Query expansion
        if self._expansion_config and getattr(self._expansion_config, "enabled", False):
            from memtomem.search.expansion import (
                expand_query_headings,
                expand_query_llm,
                expand_query_tags,
            )

            strategy = getattr(self._expansion_config, "strategy", "tags")
            max_terms = getattr(self._expansion_config, "max_terms", 3)
            if strategy in ("tags", "both"):
                query = await expand_query_tags(query, self._storage, max_terms)
            if strategy in ("headings", "both"):
                # ADR-0011 PR-D round 11: thread the outer search's
                # project context onto the heading-expansion's dense
                # probe so it samples from the same scope set the
                # primary retrieval is pinned to.
                query = await expand_query_headings(
                    query,
                    self._storage,
                    self._embedder,
                    max_terms,
                    project_context_root=project_context_root,
                )
            if strategy == "llm":
                if query in self._expansion_cache:
                    query = self._expansion_cache[query]
                elif self._llm_provider is not None:
                    try:
                        original = query
                        query = await expand_query_llm(
                            query,
                            self._llm_provider,
                            max_terms,  # type: ignore[arg-type]
                        )
                        if len(self._expansion_cache) >= _EXPANSION_CACHE_MAX:
                            self._expansion_cache.clear()
                        self._expansion_cache[original] = query
                    except Exception:
                        logger.warning(
                            "LLM query expansion failed, using original query",
                            exc_info=True,
                        )

        # Stage 1 + 2: run enabled retrievers concurrently
        bm25_results: list[SearchResult] = []
        dense_results: list[SearchResult] = []
        query_embedding: list[float] = []
        bm25_error: str | None = None

        if use_bm25:
            bm25_task = asyncio.create_task(
                self._storage.bm25_search(
                    query,
                    top_k=bm25_k,
                    namespace_filter=ns_filter,
                    scope_filter=scope_filter,
                    project_context_root=project_context_root,
                )
            )
        dense_error: str | None = None
        if use_dense:
            try:
                query_embedding = await self._embedder.embed_query(query)
                dense_results = await self._storage.dense_search(
                    query_embedding,
                    top_k=dense_k,
                    namespace_filter=ns_filter,
                    scope_filter=scope_filter,
                    project_context_root=project_context_root,
                )
            except Exception as exc:
                logger.warning("Dense search unavailable: %s", exc)
                dense_results = []
                dense_error = str(exc)
        if use_bm25:
            try:
                bm25_results = await bm25_task
            except Exception as exc:
                logger.warning("BM25 search failed: %s", exc)
                bm25_results = []
                bm25_error = str(exc)

        stats = RetrievalStats(
            bm25_candidates=len(bm25_results),
            dense_candidates=len(dense_results),
            bm25_error=bm25_error,
            dense_error=dense_error,
            hidden_system_ns=hidden_system_ns,
        )

        # Stage 1 enrichment (RFC P1 Phase C): session-summary rescue.
        # When an above-threshold past-session summary's chunk_links
        # point at source files relevant to this query, run a parallel
        # BM25+dense retrieval restricted to those files and merge the
        # result as a third RRF input. This brings past-session chunks
        # into ranking contention without changing the retrieval
        # primitives' signatures — keeping ``bm25_search`` /
        # ``dense_search`` archive-agnostic. (Pure post-fusion score
        # multiplier was rejected: it can only re-rank candidates that
        # already surfaced organically; "ranking contention" requires
        # injection.)
        rescue_results: list[SearchResult] = []
        rescue_chunk_ids: set[UUID] = set()
        if (
            namespace is None
            and self._session_summary_config is not None
            and self._session_summary_config.expansion_lookup_top_k > 0
        ):
            try:
                boost_sources = await self._session_summary_boost_sources(
                    query,
                    scope_filter=scope_filter,
                    project_context_root=project_context_root,
                )
                if boost_sources:
                    rescue_results = await self._rescue_retrieval(
                        query,
                        query_embedding,
                        top_k=top_k,
                        boost_sources=boost_sources,
                        use_bm25=use_bm25,
                        use_dense=use_dense,
                        scope_filter=scope_filter,
                        project_context_root=project_context_root,
                    )
                    rescue_chunk_ids = {r.chunk.id for r in rescue_results}
            except Exception:
                logger.debug("session-summary rescue leg failed", exc_info=True)

        # Stage 3: fusion (or single-retriever passthrough)
        # When reranking is active, widen the candidate pool so the
        # cross-encoder can rescue items RRF ranked just outside top_k.
        # pool = clamp(oversample * top_k, [min_pool, max_pool]) — scales
        # with the request and bounded by cost controls. Collapses to
        # top_k when reranking is disabled so single-retriever passthrough
        # size is unchanged.
        if self._reranker is not None and self._rerank_config is not None:
            rcfg = self._rerank_config
            rerank_pool = max(
                rcfg.min_pool,
                min(rcfg.max_pool, int(rcfg.oversample * top_k)),
            )
        else:
            rerank_pool = top_k

        # Mark rescue-leg results so fusion can OR-propagate the flag.
        if rescue_results:
            rescue_results = [
                dataclass_replace(r, via_session_summary=True) for r in rescue_results
            ]

        if use_bm25 and use_dense:
            fusion_lists = [bm25_results, dense_results]
            fusion_weights = list(effective_weights)
            fusion_labels = ["bm25", "dense"]
            if rescue_results:
                fusion_lists.append(rescue_results)
                rescue_w = (
                    self._session_summary_config.expansion_rescue_weight
                    if self._session_summary_config is not None
                    else 0.5
                )
                fusion_weights.append(rescue_w)
                fusion_labels.append("session_rescue")
            fused = reciprocal_rank_fusion(
                fusion_lists,
                k=self._config.rrf_k,
                top_k=rerank_pool,
                weights=fusion_weights,
                list_labels=fusion_labels,
            )
        elif use_bm25:
            if rescue_results:
                rescue_w = (
                    self._session_summary_config.expansion_rescue_weight
                    if self._session_summary_config is not None
                    else 0.5
                )
                fused = reciprocal_rank_fusion(
                    [bm25_results, rescue_results],
                    k=self._config.rrf_k,
                    top_k=rerank_pool,
                    weights=[effective_weights[0], rescue_w],
                    list_labels=["bm25", "session_rescue"],
                )
            else:
                fused = bm25_results[:rerank_pool]
        elif use_dense:
            if rescue_results:
                rescue_w = (
                    self._session_summary_config.expansion_rescue_weight
                    if self._session_summary_config is not None
                    else 0.5
                )
                fused = reciprocal_rank_fusion(
                    [dense_results, rescue_results],
                    k=self._config.rrf_k,
                    top_k=rerank_pool,
                    weights=[effective_weights[1] if len(effective_weights) > 1 else 1.0, rescue_w],
                    list_labels=["dense", "session_rescue"],
                )
            else:
                fused = dense_results[:rerank_pool]
        else:
            fused = []
        stats.fused_total = len(fused)

        # Stage 3b: Cross-encoder reranking
        if self._reranker is not None and fused:
            try:
                fused = await self._reranker.rerank(query, fused, top_k=top_k)
            except Exception as exc:
                logger.warning("Reranking failed, using original order: %s", exc)
                # Fallback must still honor the caller's response size —
                # fused is at rerank_pool (e.g. 20) right now, not top_k.
                fused = fused[:top_k]

        # Filter by source file if requested
        if source_filter:
            fused = [
                r
                for r in fused
                if match_source_filter(source_filter, str(r.chunk.metadata.source_file))
            ]

        # Filter by tag if requested (comma-separated = OR matching)
        if tag_filter:
            required = {t.strip() for t in tag_filter.split(",") if t.strip()}
            fused = [r for r in fused if required & set(r.chunk.metadata.tags)]

        # Stage β': temporal-validity filter (RFC §Pipeline integration).
        # AND-combined with source/tag filter via sequential application —
        # a chunk must pass both to survive. Default ``as_of`` is the
        # current wall-clock; explicit values bypass the result cache so
        # historical queries don't poison default-path cache slots.
        effective_as_of = as_of_unix if as_of_unix is not None else int(time.time())
        if fused:
            fused = _apply_validity_filter(fused, effective_as_of)

        # Stage 4: Time decay (re-score older chunks lower)
        if self._decay_config.enabled and fused:
            from memtomem.search.decay import apply_score_decay

            fused = apply_score_decay(fused, half_life_days=self._decay_config.half_life_days)

        # Stage 5: MMR diversity re-ranking
        if self._mmr_config.enabled and fused and use_dense:
            from memtomem.search.mmr import apply_mmr

            chunk_ids = [str(r.chunk.id) for r in fused]
            emb_dict_raw = await self._storage.get_embeddings_for_chunks(chunk_ids)
            if emb_dict_raw:
                from uuid import UUID

                emb_dict = {UUID(k): v for k, v in emb_dict_raw.items()}
                fused = apply_mmr(fused, emb_dict, lambda_param=self._mmr_config.lambda_param)

        # Stage 6: Access-frequency boost
        if self._access_config.enabled and fused:
            from memtomem.search.access import apply_access_boost

            access_chunk_ids = [r.chunk.id for r in fused]
            access_counts = await self._storage.get_access_counts(access_chunk_ids)
            fused = apply_access_boost(
                fused, access_counts, max_boost=self._access_config.max_boost
            )

        # Stage 7: Importance boost
        if self._importance_config and getattr(self._importance_config, "enabled", False) and fused:
            from memtomem.search.importance import apply_importance_boost

            chunk_ids_imp = [r.chunk.id for r in fused]
            imp_scores = await self._storage.get_importance_scores(chunk_ids_imp)
            fused = apply_importance_boost(
                fused,
                imp_scores,
                max_boost=getattr(self._importance_config, "max_boost", 1.5),
            )

        # Stage 8: Context window expansion (post-scoring, does not affect ranking)
        ctx_win = self._resolve_context_window(context_window)
        if ctx_win > 0 and fused:
            fused = await self._expand_context(fused, ctx_win)

        # Re-stamp ``via_session_summary`` for any chunk that came in
        # via the rescue leg. Downstream stages (decay, MMR, access,
        # importance, reranker, context expansion) construct fresh
        # ``SearchResult`` instances with default field values and so
        # silently drop the flag — restoring here at the boundary keeps
        # propagation a single-source-of-truth concern of the pipeline
        # rather than leaking the obligation into every stage.
        if rescue_chunk_ids:
            fused = [
                dataclass_replace(r, via_session_summary=True)
                if r.chunk.id in rescue_chunk_ids
                else r
                for r in fused
            ]

        stats.final_total = len(fused)

        # Increment access counts for returned results (fire-and-forget)
        if fused:

            async def _increment():
                await self._storage.increment_access([r.chunk.id for r in fused])

            t = asyncio.create_task(_increment())
            t.add_done_callback(_bg_task_error_cb)
            self._bg_tasks.add(t)
            t.add_done_callback(self._bg_tasks.discard)

        # Save to query history (fire-and-forget)
        async def _save_history():
            emb = query_embedding if use_dense else []
            await self._storage.save_query_history(
                query,
                emb,
                [str(r.chunk.id) for r in fused[:top_k]],
                [r.score for r in fused[:top_k]],
            )

        t2 = asyncio.create_task(_save_history())
        t2.add_done_callback(_bg_task_error_cb)
        self._bg_tasks.add(t2)
        t2.add_done_callback(self._bg_tasks.discard)

        # Store in TTL cache only if version hasn't changed during search.
        # Skip the write when ``as_of_unix`` was explicit so that a
        # historical-query result never overwrites the default-path slot
        # (next default caller would otherwise be served a past-snapshot
        # filtering of the same query).
        if as_of_unix is None and self._cache_version == version_at_start:
            self._search_cache[cache_key] = (time.time(), version_at_start, fused, stats)
            # Evict old entries (keep max 50)
            if len(self._search_cache) > 50:
                try:
                    oldest_key = min(self._search_cache, key=lambda k: self._search_cache[k][0])
                    self._search_cache.pop(oldest_key, None)
                except ValueError:
                    pass  # cache emptied by concurrent invalidate_cache()

        return fused, stats

    async def close(self) -> None:
        """Release resources held by the pipeline (reranker client, etc.)."""
        if self._bg_tasks:
            await asyncio.gather(*self._bg_tasks, return_exceptions=True)
            self._bg_tasks.clear()
        if self._reranker is not None and hasattr(self._reranker, "close"):
            await self._reranker.close()
