"""
  Main Searcher module - executes search queries against Elasticsearch.
  

  Handles query parsing, ranking, and result formatting.

  """
  

  from typing import Dict, Any, List, Optional
  import json

  import logging

  import hashlib

  from string import Formatter

  from utils.es_client import ESClient
  from query import QueryParser, ParsedQuery

  from query.style_intent import StyleIntentRegistry

  from embeddings.image_encoder import CLIPImageEncoder

  from .es_query_builder import ESQueryBuilder

  from .sku_intent_selector import SkuSelectionDecision, StyleSkuSelector

  from config import SearchConfig

  from config.tenant_config_loader import get_tenant_config_loader

  from context.request_context import RequestContext, RequestContextStage

  from api.models import FacetResult, FacetConfig

  from api.result_formatter import ResultFormatter

  from indexer.mapping_generator import get_tenant_index_name

  logger = logging.getLogger(__name__)

  backend_verbose_logger = logging.getLogger("backend.verbose")
  
  
  def _log_backend_verbose(payload: Dict[str, Any]) -> None:
      if not backend_verbose_logger.handlers:
          return
      backend_verbose_logger.info(
          json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
      )

  def _index_debug_rows_by_doc(rows: Any) -> Dict[str, Dict[str, Any]]:
      indexed: Dict[str, Dict[str, Any]] = {}
      if not isinstance(rows, list):
          return indexed
      for item in rows:
          if not isinstance(item, dict):
              continue
          doc_id = item.get("doc_id")
          if doc_id is None:
              continue
          indexed[str(doc_id)] = item
      return indexed
  
  

  def _summarize_ltr_features(per_result_debug: List[Dict[str, Any]], top_n: int = 20) -> Dict[str, Any]:
      rows = list(per_result_debug[:top_n])
      if not rows:
          return {"top_n": 0, "counts": {}, "averages": {}, "top_docs": []}
  
      def _feature(row: Dict[str, Any], key: str) -> Any:

          funnel = row.get("ranking_funnel", {})
          for stage_name in ("rerank", "fine_rank", "coarse_rank"):
              stage_features = funnel.get(stage_name, {}).get("ltr_features")
              if isinstance(stage_features, dict) and key in stage_features:
                  return stage_features.get(key)

          return None
  
      def _count(flag: str) -> int:
          return sum(1 for row in rows if bool(_feature(row, flag)))
  
      def _avg(name: str) -> float | None:
          values = [float(v) for row in rows if (v := _feature(row, name)) is not None]
          if not values:
              return None
          return round(sum(values) / len(values), 6)
  
      top_docs = []
      for row in rows[:10]:
          top_docs.append(
              {
                  "spu_id": row.get("spu_id"),
                  "final_rank": row.get("final_rank"),
                  "title_zh": row.get("title_multilingual", {}).get("zh")
                  if isinstance(row.get("title_multilingual"), dict)
                  else None,
                  "es_score": _feature(row, "es_score"),
                  "text_score": _feature(row, "text_score"),
                  "knn_score": _feature(row, "knn_score"),
                  "rerank_score": _feature(row, "rerank_score"),
                  "fine_score": _feature(row, "fine_score"),
                  "has_translation_match": _feature(row, "has_translation_match"),
                  "has_text_knn": _feature(row, "has_text_knn"),
                  "has_image_knn": _feature(row, "has_image_knn"),
                  "has_style_boost": _feature(row, "has_style_boost"),
              }
          )
  
      return {
          "top_n": len(rows),
          "counts": {
              "translation_match_docs": _count("has_translation_match"),
              "text_knn_docs": _count("has_text_knn"),
              "image_knn_docs": _count("has_image_knn"),
              "style_boost_docs": _count("has_style_boost"),
              "text_fallback_to_es_docs": _count("text_score_fallback_to_es"),
          },
          "averages": {
              "es_score": _avg("es_score"),
              "text_score": _avg("text_score"),
              "knn_score": _avg("knn_score"),
              "rerank_score": _avg("rerank_score"),
              "fine_score": _avg("fine_score"),
              "source_score": _avg("source_score"),
              "translation_score": _avg("translation_score"),
              "text_knn_score": _avg("text_knn_score"),
              "image_knn_score": _avg("image_knn_score"),
          },
          "top_docs": top_docs,
      }
  
  

  class SearchResult:

      """Container for search results (外部友好格式)."""

  
      def __init__(
          self,

          results: List[Any],  # List[SpuResult]

          total: int,
          max_score: float,
          took_ms: int,

          facets: Optional[List[FacetResult]] = None,

          query_info: Optional[Dict[str, Any]] = None,

          suggestions: Optional[List[str]] = None,
          related_searches: Optional[List[str]] = None,

          debug_info: Optional[Dict[str, Any]] = None

      ):

          self.results = results

          self.total = total
          self.max_score = max_score
          self.took_ms = took_ms

          self.facets = facets

          self.query_info = query_info or {}

          self.suggestions = suggestions or []
          self.related_searches = related_searches or []

          self.debug_info = debug_info

      def to_dict(self) -> Dict[str, Any]:
          """Convert to dictionary representation."""

          result = {

              "results": [r.model_dump() if hasattr(r, 'model_dump') else r for r in self.results],

              "total": self.total,
              "max_score": self.max_score,
              "took_ms": self.took_ms,

              "facets": [f.model_dump() for f in self.facets] if self.facets else None,

              "query_info": self.query_info,
              "suggestions": self.suggestions,
              "related_searches": self.related_searches

          }

          if self.debug_info is not None:
              result["debug_info"] = self.debug_info
          return result

  
  
  class Searcher:
      """
      Main search engine class.
  
      Handles:
      - Query parsing and translation

      - Dynamic multi-language text recall planning

      - ES query building
      - Result ranking and formatting
      """
  
      def __init__(
          self,

          es_client: ESClient,

          config: SearchConfig,

          query_parser: Optional[QueryParser] = None,
          image_encoder: Optional[CLIPImageEncoder] = None,

      ):
          """
          Initialize searcher.
  
          Args:

              es_client: Elasticsearch client

              config: SearchConfig instance

              query_parser: Query parser (created if not provided)

              image_encoder: Optional pre-initialized image encoder

          """

          self.es_client = es_client

          self.config = config

          self.text_embedding_field = config.query_config.text_embedding_field or "title_embedding"

          self.image_embedding_field = config.query_config.image_embedding_field
          if self.image_embedding_field and image_encoder is None:
              self.image_encoder = CLIPImageEncoder()
          else:
              self.image_encoder = image_encoder

          # Index name is now generated dynamically per tenant, no longer stored here
          self.query_parser = query_parser or QueryParser(config, image_encoder=self.image_encoder)

          self.source_fields = config.query_config.source_fields

          self.style_intent_registry = StyleIntentRegistry.from_query_config(self.config.query_config)
          self.style_sku_selector = StyleSkuSelector(
              self.style_intent_registry,
              text_encoder_getter=lambda: getattr(self.query_parser, "text_encoder", None),

          )

          # Query builder - simplified single-layer architecture

          self.query_builder = ESQueryBuilder(

              match_fields=[],
              field_boosts=self.config.field_boosts,
              multilingual_fields=self.config.query_config.multilingual_fields,
              shared_fields=self.config.query_config.shared_fields,
              core_multilingual_fields=self.config.query_config.core_multilingual_fields,

              text_embedding_field=self.text_embedding_field,

              image_embedding_field=self.image_embedding_field,

              source_fields=self.source_fields,

              function_score_config=self.config.function_score,

              default_language=self.config.query_config.default_language,

              knn_text_boost=self.config.query_config.knn_text_boost,
              knn_image_boost=self.config.query_config.knn_image_boost,
              knn_text_k=self.config.query_config.knn_text_k,
              knn_text_num_candidates=self.config.query_config.knn_text_num_candidates,
              knn_text_k_long=self.config.query_config.knn_text_k_long,
              knn_text_num_candidates_long=self.config.query_config.knn_text_num_candidates_long,
              knn_image_k=self.config.query_config.knn_image_k,
              knn_image_num_candidates=self.config.query_config.knn_image_num_candidates,

              base_minimum_should_match=self.config.query_config.base_minimum_should_match,
              translation_minimum_should_match=self.config.query_config.translation_minimum_should_match,
              translation_boost=self.config.query_config.translation_boost,

              tie_breaker_base_query=self.config.query_config.tie_breaker_base_query,

              best_fields_boosts=self.config.query_config.best_fields,
              best_fields_clause_boost=self.config.query_config.best_fields_boost,
              phrase_field_boosts=self.config.query_config.phrase_fields,
              phrase_match_boost=self.config.query_config.phrase_match_boost,

          )
  

      def _apply_source_filter(self, es_query: Dict[str, Any]) -> None:
          """
          Apply tri-state _source semantics:
          - None: do not set _source (return full source)
          - []: _source=false (return no source fields)
          - [..]: _source.includes=[..]
          """
          if self.source_fields is None:
              return
          if not isinstance(self.source_fields, list):
              raise ValueError("query_config.source_fields must be null or list[str]")
          if len(self.source_fields) == 0:
              es_query["_source"] = False
              return
          es_query["_source"] = {"includes": self.source_fields}
  

      def _resolve_exact_knn_rescore_window(self) -> int:
          configured = int(self.config.rerank.exact_knn_rescore_window)
          if configured > 0:
              return configured
          return int(self.config.rerank.rerank_window)
  

      def _build_exact_knn_rescore(
          self,
          *,
          query_vector: Any,
          image_query_vector: Any,

          parsed_query: Optional[ParsedQuery] = None,

      ) -> Optional[Dict[str, Any]]:
          clauses: List[Dict[str, Any]] = []
  

          text_clause = self.query_builder.build_exact_text_knn_rescore_clause(
              query_vector,
              parsed_query=parsed_query,
              query_name="exact_text_knn_query",
          )
          if text_clause:
              clauses.append(text_clause)

          image_clause = self.query_builder.build_exact_image_knn_rescore_clause(
              image_query_vector,
              query_name="exact_image_knn_query",
          )
          if image_clause:
              clauses.append(image_clause)

  
          if not clauses:
              return None
  
          return {
              "window_size": self._resolve_exact_knn_rescore_window(),
              "query": {
                  # Phase 1: only compute exact vector scores and expose them in matched_queries.
                  "score_mode": "total",
                  "query_weight": 1.0,
                  "rescore_query_weight": 0.0,
                  "rescore_query": {
                      "bool": {
                          "should": clauses,
                          "minimum_should_match": 1,
                      }
                  },
              },
          }
  
      def _attach_exact_knn_rescore(
          self,
          es_query: Dict[str, Any],
          *,
          in_rank_window: bool,
          query_vector: Any,
          image_query_vector: Any,

          parsed_query: Optional[ParsedQuery] = None,

      ) -> None:
          if not in_rank_window or not self.config.rerank.exact_knn_rescore_enabled:
              return
          rescore = self._build_exact_knn_rescore(
              query_vector=query_vector,
              image_query_vector=image_query_vector,

              parsed_query=parsed_query,

          )
          if not rescore:
              return
          existing = es_query.get("rescore")
          if existing is None:
              es_query["rescore"] = rescore
          elif isinstance(existing, list):
              es_query["rescore"] = [*existing, rescore]
          else:
              es_query["rescore"] = [existing, rescore]
  

      def _resolve_rerank_source_filter(
          self,
          doc_template: str,
          parsed_query: Optional[ParsedQuery] = None,
      ) -> Dict[str, Any]:

          """
          Build a lightweight _source filter for rerank prefetch.
  
          Only fetch fields required by rerank doc template to reduce ES payload.
          """
          field_map = {
              "title": "title",
              "brief": "brief",
              "vendor": "vendor",
              "description": "description",
              "category_path": "category_path",
          }
          includes: set[str] = set()
          template = str(doc_template or "{title}")
          for _, field_name, _, _ in Formatter().parse(template):
              if not field_name:
                  continue
              key = field_name.split(".", 1)[0].split("!", 1)[0].split(":", 1)[0]
              mapped = field_map.get(key)
              if mapped:
                  includes.add(mapped)
  
          # Fallback to title-only to keep rerank docs usable.
          if not includes:
              includes.add("title")
  

          if self._should_run_sku_selection(parsed_query):
              # SKU-level fields are needed both by text matching (optionN_value) and
              # by the image pick (image_src) of the unified SKU selector.

              includes.update(
                  {
                      "skus",
                      "option1_name",
                      "option2_name",
                      "option3_name",
                  }
              )

          if self._has_style_intent(parsed_query):
              # Treated as an additional value source for attribute matching
              # (on the same dimension as optionN).
              includes.add("enriched_taxonomy_attributes")

          # Needed when inner_hits url string differs from sku.image_src but ES exposes
          # _nested.offset — we re-resolve the winning url from image_embedding[offset].
          if self._has_image_signal(parsed_query):
              includes.add("image_embedding")
  

          return {"includes": sorted(includes)}
  
      def _fetch_hits_by_ids(
          self,
          index_name: str,
          doc_ids: List[str],
          source_spec: Optional[Any],
      ) -> tuple[Dict[str, Dict[str, Any]], int]:
          """
          Fetch page documents by IDs for final response fill.
  
          Returns:
              (hits_by_id, es_took_ms)
          """
          if not doc_ids:
              return {}, 0
  
          body: Dict[str, Any] = {
              "query": {
                  "ids": {
                      "values": doc_ids,
                  }
              }
          }
          if source_spec is not None:
              body["_source"] = source_spec
  
          resp = self.es_client.search(
              index_name=index_name,
              body=body,
              size=len(doc_ids),
              from_=0,
          )
          hits = resp.get("hits", {}).get("hits") or []
          hits_by_id: Dict[str, Dict[str, Any]] = {}
          for hit in hits:
              hid = hit.get("_id")
              if hid is None:
                  continue
              hits_by_id[str(hid)] = hit
          return hits_by_id, int(resp.get("took", 0) or 0)
  

      @staticmethod

      def _restore_hits_in_doc_order(
          doc_ids: List[str],
          hits_by_id: Dict[str, Dict[str, Any]],
      ) -> List[Dict[str, Any]]:
          ordered_hits: List[Dict[str, Any]] = []
          for doc_id in doc_ids:
              hit = hits_by_id.get(str(doc_id))
              if hit is not None:
                  ordered_hits.append(hit)
          return ordered_hits
  
      @staticmethod
      def _merge_source_specs(*source_specs: Any) -> Optional[Dict[str, Any]]:
          includes: set[str] = set()
          for source_spec in source_specs:
              if not isinstance(source_spec, dict):
                  continue
              for field_name in source_spec.get("includes") or []:
                  includes.add(str(field_name))
          if not includes:
              return None
          return {"includes": sorted(includes)}
  
      @staticmethod

      def _has_style_intent(parsed_query: Optional[ParsedQuery]) -> bool:
          profile = getattr(parsed_query, "style_intent_profile", None)
          return bool(getattr(profile, "is_active", False))

      def _has_image_signal(self, parsed_query: Optional[ParsedQuery]) -> bool:
          """True when the query carries an image vector that can drive an image-based SKU pick."""
          if parsed_query is None:
              return False
          if not getattr(self.config.query_config, "image_embedding_field", None):
              return False
          return getattr(parsed_query, "image_query_vector", None) is not None
  
      def _should_run_sku_selection(self, parsed_query: Optional[ParsedQuery]) -> bool:
          """Trigger unified SKU selection when either signal is present.
  
          Text-intent alone drives attribute-value matching; image signal alone drives
          image-nearest SKU promotion; together, image is a visual tie-breaker inside
          the text-matched set.
          """
          return self._has_style_intent(parsed_query) or self._has_image_signal(parsed_query)
  

      def _apply_style_intent_to_hits(

          self,
          es_hits: List[Dict[str, Any]],
          parsed_query: ParsedQuery,
          context: Optional[RequestContext] = None,

      ) -> Dict[str, SkuSelectionDecision]:

          if context is not None:
              context.start_stage(RequestContextStage.STYLE_SKU_PREPARE_HITS)
          try:

              return self.style_sku_selector.prepare_hits(es_hits, parsed_query)

          finally:
              if context is not None:
                  context.end_stage(RequestContextStage.STYLE_SKU_PREPARE_HITS)

      def search(
          self,
          query: str,

          tenant_id: str,

          size: int = 10,
          from_: int = 0,
          filters: Optional[Dict[str, Any]] = None,

          range_filters: Optional[Dict[str, Any]] = None,

          facets: Optional[List[FacetConfig]] = None,

          min_score: Optional[float] = None,

          context: Optional[RequestContext] = None,

          sort_by: Optional[str] = None,

          sort_order: Optional[str] = "desc",

          debug: bool = False,

          language: str = "en",

          sku_filter_dimension: Optional[List[str]] = None,

          enable_rerank: Optional[bool] = None,

          rerank_query_template: Optional[str] = None,
          rerank_doc_template: Optional[str] = None,

      ) -> SearchResult:
          """

          Execute search query (外部友好格式).

  
          Args:
              query: Search query string

              tenant_id: Tenant ID (required for filtering)

              size: Number of results to return
              from_: Offset for pagination

              filters: Exact match filters
              range_filters: Range filters for numeric fields
              facets: Facet configurations for faceted search

              min_score: Minimum score threshold

              context: Request context for tracking (required)

              sort_by: Field name for sorting
              sort_order: Sort order: 'asc' or 'desc'

              debug: Enable debug information output

              language: Response / field selection language hint (e.g. zh, en)
              sku_filter_dimension: SKU grouping dimensions for per-SPU variant pick
              enable_rerank: If None, use ``config.rerank.enabled``; if set, overrides

                  whether the final rerank provider is invoked (subject to rank window).
                  When false, the ranking pipeline still runs and rerank stage becomes
                  pass-through.

              rerank_query_template: Override for rerank query text template; None uses
                  ``config.rerank.rerank_query_template`` (e.g. ``"{query}"``).
              rerank_doc_template: Override for per-hit document text passed to rerank;
                  None uses ``config.rerank.rerank_doc_template``. Placeholders are
                  resolved in ``search/rerank_client.py``.

  
          Returns:

              SearchResult object with formatted results

          """

          if context is None:

              raise ValueError("context is required")

          # 根据租户配置决定翻译开关（离线/在线统一）
          tenant_loader = get_tenant_config_loader()
          tenant_cfg = tenant_loader.get_tenant_config(tenant_id)

          index_langs = tenant_cfg.get("index_languages") or []
          enable_translation = len(index_langs) > 0

          enable_embedding = self.config.query_config.enable_text_embedding

          coarse_cfg = self.config.coarse_rank
          fine_cfg = self.config.fine_rank

          rc = self.config.rerank
          effective_query_template = rerank_query_template or rc.rerank_query_template
          effective_doc_template = rerank_doc_template or rc.rerank_doc_template

          fine_query_template = fine_cfg.rerank_query_template or effective_query_template
          fine_doc_template = fine_cfg.rerank_doc_template or effective_doc_template

          # 重排开关优先级：请求参数显式传值 > 服务端配置（默认开启）
          rerank_enabled_by_config = bool(rc.enabled)
          do_rerank = rerank_enabled_by_config if enable_rerank is None else bool(enable_rerank)

          fine_enabled = bool(fine_cfg.enabled)

          rerank_window = rc.rerank_window

          coarse_input_window = max(rerank_window, int(coarse_cfg.input_window))
          coarse_output_window = max(rerank_window, int(coarse_cfg.output_window))
          fine_input_window = max(rerank_window, int(fine_cfg.input_window))
          fine_output_window = max(rerank_window, int(fine_cfg.output_window))

          # 多阶段排序窗口独立于最终 rerank 开关：即使关闭最终 rerank，也保留 coarse/fine 流程。
          in_rank_window = (from_ + size) <= rerank_window
          es_fetch_from = 0 if in_rank_window else from_
          es_fetch_size = coarse_input_window if in_rank_window else size

  
          es_score_normalization_factor: Optional[float] = None
          initial_ranks_by_doc: Dict[str, int] = {}

          coarse_ranks_by_doc: Dict[str, int] = {}
          fine_ranks_by_doc: Dict[str, int] = {}
          rerank_ranks_by_doc: Dict[str, int] = {}

          coarse_debug_info: Optional[Dict[str, Any]] = None
          fine_debug_info: Optional[Dict[str, Any]] = None

          rerank_debug_info: Optional[Dict[str, Any]] = None

  
          # Start timing
          context.start_stage(RequestContextStage.TOTAL)
  
          context.logger.info(
              f"开始搜索请求 | 查询: '{query}' | 参数: size={size}, from_={from_}, "

              f"enable_rerank(request)={enable_rerank}, enable_rerank(config)={rerank_enabled_by_config}, "

              f"fine_enabled(config)={fine_enabled}, "
              f"enable_rerank(effective)={do_rerank}, in_rank_window={in_rank_window}, "

              f"es_fetch=({es_fetch_from},{es_fetch_size}) | "

              f"index_languages={index_langs} | "

              f"enable_translation={enable_translation}, enable_embedding={enable_embedding}, min_score={min_score}",

              extra={'reqid': context.reqid, 'uid': context.uid}
          )
  
          # Store search parameters in context
          context.metadata['search_params'] = {
              'size': size,
              'from_': from_,

              'es_fetch_from': es_fetch_from,
              'es_fetch_size': es_fetch_size,

              'in_rank_window': in_rank_window,

              'rerank_enabled_by_config': rerank_enabled_by_config,

              'fine_enabled': fine_enabled,

              'enable_rerank_request': enable_rerank,
              'rerank_query_template': effective_query_template,
              'rerank_doc_template': effective_doc_template,

              'fine_query_template': fine_query_template,
              'fine_doc_template': fine_doc_template,

              'filters': filters,

              'range_filters': range_filters,
              'facets': facets,

              'enable_translation': enable_translation,
              'enable_embedding': enable_embedding,

              'enable_rerank': do_rerank,

              'coarse_input_window': coarse_input_window,
              'coarse_output_window': coarse_output_window,
              'fine_input_window': fine_input_window,
              'fine_output_window': fine_output_window,
              'rerank_window': rerank_window,

              'min_score': min_score,

              'sort_by': sort_by,
              'sort_order': sort_order

          }

          context.metadata['feature_flags'] = {
              'translation_enabled': enable_translation,
              'embedding_enabled': enable_embedding,

              'fine_enabled': fine_enabled,

              'rerank_enabled': do_rerank,
              'style_intent_enabled': bool(self.style_intent_registry.enabled),

          }

  
          # Step 1: Parse query

          context.start_stage(RequestContextStage.QUERY_PARSING)
          try:
              parsed_query = self.query_parser.parse(
                  query,

                  generate_vector=enable_embedding,

                  tenant_id=tenant_id,

                  context=context,
                  target_languages=index_langs if enable_translation else [],

              )
              # Store query analysis results in context
              context.store_query_analysis(
                  original_query=parsed_query.original_query,

                  query_normalized=parsed_query.query_normalized,

                  rewritten_query=parsed_query.rewritten_query,
                  detected_language=parsed_query.detected_language,
                  translations=parsed_query.translations,

                  keywords_queries=parsed_query.keywords_queries,

                  query_vector=parsed_query.query_vector.tolist() if parsed_query.query_vector is not None else None,

              )

              context.metadata["feature_flags"]["style_intent_active"] = self._has_style_intent(parsed_query)

              context.logger.info(
                  f"查询解析完成 | 原查询: '{parsed_query.original_query}' | "
                  f"重写后: '{parsed_query.rewritten_query}' | "
                  f"语言: {parsed_query.detected_language} | "

                  f"关键词: {parsed_query.keywords_queries} | "

                  f"文本向量: {'是' if parsed_query.query_vector is not None else '否'} | "

                  f"图片向量: {'是' if parsed_query.image_query_vector is not None else '否'}",

                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
          except Exception as e:
              context.set_error(e)
              context.logger.error(
                  f"查询解析失败 | 错误: {str(e)}",
                  extra={'reqid': context.reqid, 'uid': context.uid}
              )
              raise
          finally:
              context.end_stage(RequestContextStage.QUERY_PARSING)
  

          # Step 2: Query building

          context.start_stage(RequestContextStage.QUERY_BUILDING)
          try:

              # Generate tenant-specific index name
              index_name = get_tenant_index_name(tenant_id)

              # index_name = "search_products"

              # No longer need to add tenant_id to filters since each tenant has its own index

              image_query_vector = None
              if enable_embedding:
                  image_query_vector = parsed_query.image_query_vector

              es_query = self.query_builder.build_query(

                  query_text=parsed_query.rewritten_query or parsed_query.query_normalized,

                  query_vector=parsed_query.query_vector if enable_embedding else None,

                  image_query_vector=image_query_vector,

                  filters=filters,

                  range_filters=range_filters,

                  facet_configs=facets,

                  size=es_fetch_size,
                  from_=es_fetch_from,

                  enable_knn=enable_embedding and (
                      parsed_query.query_vector is not None

                      or image_query_vector is not None

                  ),

                  min_score=min_score,

                  parsed_query=parsed_query,

              )

              self._attach_exact_knn_rescore(
                  es_query,
                  in_rank_window=in_rank_window,
                  query_vector=parsed_query.query_vector if enable_embedding else None,
                  image_query_vector=image_query_vector,

                  parsed_query=parsed_query,

              )

              # Add facets for faceted search
              if facets:
                  facet_aggs = self.query_builder.build_facets(facets)
                  if facet_aggs:
                      if "aggs" not in es_query:
                          es_query["aggs"] = {}
                      es_query["aggs"].update(facet_aggs)

              # Add sorting if specified
              if sort_by:
                  es_query = self.query_builder.add_sorting(es_query, sort_by, sort_order)

                  es_query["track_scores"] = True

              # Keep requested response _source semantics for the final response fill.
              response_source_spec = es_query.get("_source")
  

              # In multi-stage rank window, first pass only needs score signals for coarse rank.

              es_query_for_fetch = es_query

              if in_rank_window:

                  es_query_for_fetch = dict(es_query)

                  es_query_for_fetch["_source"] = False

              # Extract size and from from body for ES client parameters

              body_for_es = {k: v for k, v in es_query_for_fetch.items() if k not in ['size', 'from']}

  
              # Store ES query in context
              context.store_intermediate_result('es_query', es_query)

              # Serialize ES query to compute a compact size + stable digest for correlation