"""Core orchestration: corpus, rerank, LLM labels, live/batch evaluation."""
  
  from __future__ import annotations
  
  import json
  import time
  from pathlib import Path
  from typing import Any, Dict, List, Sequence, Tuple
  
  import requests
  from elasticsearch.helpers import scan
  
  from api.app import get_app_config, get_es_client, get_query_parser, init_service
  from indexer.mapping_generator import get_tenant_index_name
  
  from .clients import DashScopeLabelClient, RerankServiceClient, SearchServiceClient
  from .constants import (
      DEFAULT_ARTIFACT_ROOT,
      DEFAULT_LABELER_MODE,
      JUDGE_PROMPT_VERSION_COMPLEX,
      RELEVANCE_EXACT,
      RELEVANCE_IRRELEVANT,
      RELEVANCE_PARTIAL,
      VALID_LABELS,
  )
  from .metrics import aggregate_metrics, compute_query_metrics, label_distribution
  from .reports import render_batch_report_markdown
  from .store import EvalStore, QueryBuildResult
  from .utils import (
      build_display_title,
      build_rerank_doc,
      compact_option_values,
      compact_product_payload,
      ensure_dir,
      normalize_text,
      pick_text,
      sha1_text,
      utc_now_iso,
      utc_timestamp,
  )
  
  
  class SearchEvaluationFramework:
      def __init__(
          self,
          tenant_id: str,
          artifact_root: Path = DEFAULT_ARTIFACT_ROOT,
          search_base_url: str = "http://localhost:6002",
          labeler_mode: str = DEFAULT_LABELER_MODE,
      ):
          init_service(get_app_config().infrastructure.elasticsearch.host)
          self.tenant_id = str(tenant_id)
          self.artifact_root = ensure_dir(artifact_root)
          self.labeler_mode = str(labeler_mode).strip().lower() or DEFAULT_LABELER_MODE
          self.store = EvalStore(self.artifact_root / "search_eval.sqlite3")
          self.search_client = SearchServiceClient(search_base_url, self.tenant_id)
          app_cfg = get_app_config()
          rerank_service_url = str(
              app_cfg.services.rerank.providers["http"]["instances"]["default"]["service_url"]
          )
          self.rerank_client = RerankServiceClient(rerank_service_url)
          llm_cfg = app_cfg.services.translation.capabilities["llm"]
          api_key = app_cfg.infrastructure.secrets.dashscope_api_key
          if not api_key:
              raise RuntimeError("dashscope_api_key is required for search evaluation annotation")
          self.label_client = DashScopeLabelClient(
              model=str(llm_cfg["model"]),
              base_url=str(llm_cfg["base_url"]),
              api_key=str(api_key),
          )
          self.query_parser = None
  
      def _get_query_parser(self):
          if self.query_parser is None:
              self.query_parser = get_query_parser()
          return self.query_parser
  
      def build_query_parser_hints(self, query: str) -> Dict[str, Any]:
          parsed = self._get_query_parser().parse(query, generate_vector=False, target_languages=["en", "zh"])
          payload = parsed.to_dict()
          payload["text_for_rerank"] = parsed.text_for_rerank()
          return payload
  
      def get_query_profile(self, query: str, force_refresh: bool = False) -> Dict[str, Any]:
          if self.labeler_mode != "complex":
              raise RuntimeError("query profiles are only used in complex labeler mode")
          if not force_refresh:
              cached = self.store.get_query_profile(self.tenant_id, query, JUDGE_PROMPT_VERSION_COMPLEX)
              if cached is not None:
                  return cached
          parser_hints = self.build_query_parser_hints(query)
          profile, raw_response = self.label_client.extract_query_profile(query, parser_hints)
          profile["parser_hints"] = parser_hints
          self.store.upsert_query_profile(
              self.tenant_id,
              query,
              JUDGE_PROMPT_VERSION_COMPLEX,
              self.label_client.model,
              profile,
              raw_response,
          )
          return profile
  
      @staticmethod
      def _doc_evidence_text(doc: Dict[str, Any]) -> str:
          pieces: List[str] = [
              build_display_title(doc),
              pick_text(doc.get("vendor"), "en"),
              pick_text(doc.get("category_path"), "en"),
              pick_text(doc.get("category_name"), "en"),
          ]
          for sku in doc.get("skus") or []:
              pieces.extend(
                  [
                      str(sku.get("option1_value") or ""),
                      str(sku.get("option2_value") or ""),
                      str(sku.get("option3_value") or ""),
                  ]
              )
          for tag in doc.get("tags") or []:
              pieces.append(str(tag))
          return normalize_text(" | ".join(piece for piece in pieces if piece))
  
      def _apply_rule_based_label_guardrails(
          self,
          label: str,
          query_profile: Dict[str, Any],
          doc: Dict[str, Any],
      ) -> str:
          if label not in VALID_LABELS:
              return label
          evidence = self._doc_evidence_text(doc)
          category = normalize_text(query_profile.get("primary_category"))
          allowed_categories = [normalize_text(item) for item in query_profile.get("allowed_categories") or [] if str(item).strip()]
  
          primary_category_match = True
          if category:
              primary_category_match = category in evidence
          allowed_category_match = True
          if allowed_categories:
              allowed_category_match = any(signal in evidence for signal in allowed_categories)
  
          if label == RELEVANCE_EXACT and not primary_category_match:
              if allowed_category_match:
                  label = RELEVANCE_PARTIAL
              else:
                  return RELEVANCE_IRRELEVANT
  
          for attr in query_profile.get("required_attributes") or []:
              if not isinstance(attr, dict):
                  continue
              attr_name = normalize_text(attr.get("name"))
              if attr_name not in {"color", "fit", "length", "type", "product_type", "material", "size", "gender", "style", "waist_style", "rise"}:
                  continue
              required_terms = [normalize_text(item) for item in attr.get("required_terms") or [] if normalize_text(item)]
              conflicting_terms = [normalize_text(item) for item in attr.get("conflicting_terms") or [] if normalize_text(item)]
              if attr_name == "fit":
                  if any(term in {"oversized", "oversize"} for term in required_terms):
                      conflicting_terms.extend(["slim", "slimming", "fitted", "tight", "close-fitting"])
                  if any(term in {"fitted", "slim fit", "tight"} for term in required_terms):
                      conflicting_terms.extend(["oversized", "oversize", "loose", "relaxed"])
              has_required = any(term in evidence for term in required_terms) if required_terms else True
              has_conflict = any(term in evidence for term in conflicting_terms)
  
              if has_conflict:
                  return RELEVANCE_IRRELEVANT
              if label == RELEVANCE_EXACT and not has_required:
                  label = RELEVANCE_PARTIAL
  
          if label == RELEVANCE_PARTIAL and not primary_category_match and not allowed_category_match:
              return RELEVANCE_IRRELEVANT
  
          return label
  
      @staticmethod
      def _result_item_to_doc(item: Dict[str, Any]) -> Dict[str, Any]:
          option_values = list(item.get("option_values") or [])
          while len(option_values) < 3:
              option_values.append("")
          product = dict(item.get("product") or {})
          return {
              "spu_id": item.get("spu_id"),
              "title": product.get("title") or item.get("title"),
              "vendor": product.get("vendor"),
              "category_path": product.get("category"),
              "category_name": product.get("category"),
              "image_url": item.get("image_url") or product.get("image_url"),
              "tags": product.get("tags") or [],
              "skus": [
                  {
                      "option1_value": option_values[0],
                      "option2_value": option_values[1],
                      "option3_value": option_values[2],
                  }
              ],
          }
  
      def _collect_label_issues(
          self,
          label: str,
          query_profile: Dict[str, Any],
          doc: Dict[str, Any],
      ) -> List[str]:
          evidence = self._doc_evidence_text(doc)
          issues: List[str] = []
          category = normalize_text(query_profile.get("primary_category"))
          allowed_categories = [
              normalize_text(item)
              for item in query_profile.get("allowed_categories") or []
              if str(item).strip()
          ]
  
          primary_category_match = True if not category else category in evidence
          allowed_category_match = False if allowed_categories else primary_category_match
          if allowed_categories:
              allowed_category_match = any(signal in evidence for signal in allowed_categories)
  
          if label == RELEVANCE_EXACT and not primary_category_match:
              if allowed_category_match:
                  issues.append("Exact missing primary category evidence")
              else:
                  issues.append("Exact has category mismatch")
  
          if label == RELEVANCE_PARTIAL and not primary_category_match and not allowed_category_match:
              issues.append("Partial has category mismatch")
  
          for attr in query_profile.get("required_attributes") or []:
              if not isinstance(attr, dict):
                  continue
              attr_name = normalize_text(attr.get("name"))
              if attr_name not in {"color", "fit", "length", "type", "product_type", "material", "size", "gender", "style"}:
                  continue
              required_terms = [normalize_text(item) for item in attr.get("required_terms") or [] if normalize_text(item)]
              conflicting_terms = [normalize_text(item) for item in attr.get("conflicting_terms") or [] if normalize_text(item)]
              has_required = any(term in evidence for term in required_terms) if required_terms else True
              has_conflict = any(term in evidence for term in conflicting_terms)
  
              if has_conflict and label != RELEVANCE_IRRELEVANT:
                  issues.append(f"{label} conflicts on {attr_name}")
              if label == RELEVANCE_EXACT and not has_required:
                  issues.append(f"Exact missing {attr_name}")
          return issues
  
      def audit_live_query(
          self,
          query: str,
          *,
          top_k: int = 100,
          language: str = "en",
          auto_annotate: bool = False,
      ) -> Dict[str, Any]:
          live = self.evaluate_live_query(query=query, top_k=top_k, auto_annotate=auto_annotate, language=language)
          if self.labeler_mode != "complex":
              labels = [
                  item["label"] if item["label"] in VALID_LABELS else RELEVANCE_IRRELEVANT
                  for item in live["results"]
              ]
              return {
                  "query": query,
                  "tenant_id": self.tenant_id,
                  "top_k": top_k,
                  "metrics": live["metrics"],
                  "distribution": label_distribution(labels),
                  "query_profile": None,
                  "suspicious": [],
                  "results": live["results"],
              }
          query_profile = self.get_query_profile(query, force_refresh=False)
          suspicious: List[Dict[str, Any]] = []
  
          for item in live["results"]:
              doc = self._result_item_to_doc(item)
              issues = self._collect_label_issues(item["label"] or "", query_profile, doc)
              suggested_label = self._apply_rule_based_label_guardrails(item["label"] or "", query_profile, doc)
              if suggested_label != (item["label"] or ""):
                  issues = list(issues) + [f"Suggested relabel: {item['label']} -> {suggested_label}"]
              if issues:
                  suspicious.append(
                      {
                          "rank": item["rank"],
                          "spu_id": item["spu_id"],
                          "title": item["title"],
                          "label": item["label"],
                          "suggested_label": suggested_label,
                          "issues": issues,
                      }
                  )
  
          labels = [
              item["label"] if item["label"] in VALID_LABELS else RELEVANCE_IRRELEVANT
              for item in live["results"]
          ]
          return {
              "query": query,
              "tenant_id": self.tenant_id,
              "top_k": top_k,
              "metrics": live["metrics"],
              "distribution": label_distribution(labels),
              "query_profile": query_profile,
              "suspicious": suspicious,
              "results": live["results"],
          }
  
      def queries_from_file(self, path: Path) -> List[str]:
          return [
              line.strip()
              for line in path.read_text(encoding="utf-8").splitlines()
              if line.strip() and not line.strip().startswith("#")
          ]
  
      def corpus_docs(self, refresh: bool = False) -> List[Dict[str, Any]]:
          if not refresh and self.store.has_corpus(self.tenant_id):
              return self.store.get_corpus_docs(self.tenant_id)
  
          es_client = get_es_client().client
          index_name = get_tenant_index_name(self.tenant_id)
          docs: List[Dict[str, Any]] = []
          for hit in scan(
              client=es_client,
              index=index_name,
              query={
                  "_source": [
                      "spu_id",
                      "title",
                      "vendor",
                      "category_path",
                      "category_name",
                      "image_url",
                      "skus",
                      "tags",
                  ],
                  "query": {"match_all": {}},
              },
              size=500,
              preserve_order=False,
              clear_scroll=True,
          ):
              source = dict(hit.get("_source") or {})
              source["spu_id"] = str(source.get("spu_id") or hit.get("_id") or "")
              docs.append(source)
          self.store.upsert_corpus_docs(self.tenant_id, docs)
          return docs
  
      def full_corpus_rerank(
          self,
          query: str,
          docs: Sequence[Dict[str, Any]],
          batch_size: int = 24,
          force_refresh: bool = False,
      ) -> List[Dict[str, Any]]:
          cached = {} if force_refresh else self.store.get_rerank_scores(self.tenant_id, query)
          pending: List[Dict[str, Any]] = [doc for doc in docs if str(doc.get("spu_id")) not in cached]
          if pending:
              new_scores: Dict[str, float] = {}
              for start in range(0, len(pending), batch_size):
                  batch = pending[start : start + batch_size]
                  scores = self._rerank_batch_with_retry(query=query, docs=batch)
                  if len(scores) != len(batch):
                      raise RuntimeError(f"rerank returned {len(scores)} scores for {len(batch)} docs")
                  for doc, score in zip(batch, scores):
                      new_scores[str(doc.get("spu_id"))] = float(score)
              self.store.upsert_rerank_scores(
                  self.tenant_id,
                  query,
                  new_scores,
                  model_name="qwen3_vllm_score",
              )
              cached.update(new_scores)
  
          ranked = []
          for doc in docs:
              spu_id = str(doc.get("spu_id"))
              ranked.append({"spu_id": spu_id, "score": float(cached.get(spu_id, float("-inf"))), "doc": doc})
          ranked.sort(key=lambda item: item["score"], reverse=True)
          return ranked
  
      def _rerank_batch_with_retry(self, query: str, docs: Sequence[Dict[str, Any]]) -> List[float]:
          if not docs:
              return []
          doc_texts = [build_rerank_doc(doc) for doc in docs]
          try:
              scores, _meta = self.rerank_client.rerank(query=query, docs=doc_texts, normalize=False)
              return scores
          except Exception:
              if len(docs) == 1:
                  return [-1.0]
              if len(docs) <= 6:
                  scores: List[float] = []
                  for doc in docs:
                      scores.extend(self._rerank_batch_with_retry(query, [doc]))
                  return scores
              mid = len(docs) // 2
              left = self._rerank_batch_with_retry(query, docs[:mid])
              right = self._rerank_batch_with_retry(query, docs[mid:])
              return left + right
  
      def annotate_missing_labels(
          self,
          query: str,
          docs: Sequence[Dict[str, Any]],
          force_refresh: bool = False,
      ) -> Dict[str, str]:
          labels = {} if force_refresh else self.store.get_labels(self.tenant_id, query)
          missing_docs = [doc for doc in docs if str(doc.get("spu_id")) not in labels]
          if not missing_docs:
              return labels
  
          for start in range(0, len(missing_docs), self.label_client.batch_size):
              batch = missing_docs[start : start + self.label_client.batch_size]
              batch_pairs = self._classify_with_retry(query, batch, force_refresh=force_refresh)
              for sub_labels, raw_response, sub_batch in batch_pairs:
                  to_store = {str(doc.get("spu_id")): label for doc, label in zip(sub_batch, sub_labels)}
                  self.store.upsert_labels(
                      self.tenant_id,
                      query,
                      to_store,
                      judge_model=self.label_client.model,
                      raw_response=raw_response,
                  )
                  labels.update(to_store)
              time.sleep(0.1)
          return labels
  
      def _classify_with_retry(
          self,
          query: str,
          docs: Sequence[Dict[str, Any]],
          *,
          force_refresh: bool = False,
      ) -> List[Tuple[List[str], str, Sequence[Dict[str, Any]]]]:
          if not docs:
              return []
          try:
              if self.labeler_mode == "complex":
                  query_profile = self.get_query_profile(query, force_refresh=force_refresh)
                  labels, raw_response = self.label_client.classify_batch_complex(query, query_profile, docs)
                  labels = [
                      self._apply_rule_based_label_guardrails(label, query_profile, doc)
                      for doc, label in zip(docs, labels)
                  ]
              else:
                  labels, raw_response = self.label_client.classify_batch_simple(query, docs)
              return [(labels, raw_response, docs)]
          except Exception:
              if len(docs) == 1:
                  raise
              mid = len(docs) // 2
              return self._classify_with_retry(query, docs[:mid], force_refresh=force_refresh) + self._classify_with_retry(query, docs[mid:], force_refresh=force_refresh)
  
      def build_query_annotation_set(
          self,
          query: str,
          *,
          search_depth: int = 1000,
          rerank_depth: int = 10000,
          annotate_search_top_k: int = 120,
          annotate_rerank_top_k: int = 200,
          language: str = "en",
          force_refresh_rerank: bool = False,
          force_refresh_labels: bool = False,
      ) -> QueryBuildResult:
          search_payload = self.search_client.search(query=query, size=search_depth, from_=0, language=language)
          search_results = list(search_payload.get("results") or [])
          corpus = self.corpus_docs(refresh=False)
          full_rerank = self.full_corpus_rerank(
              query=query,
              docs=corpus,
              force_refresh=force_refresh_rerank,
          )
          rerank_depth_effective = min(rerank_depth, len(full_rerank))
  
          pool_docs: Dict[str, Dict[str, Any]] = {}
          for doc in search_results[:annotate_search_top_k]:
              pool_docs[str(doc.get("spu_id"))] = doc
          for item in full_rerank[:annotate_rerank_top_k]:
              pool_docs[str(item["spu_id"])] = item["doc"]
  
          labels = self.annotate_missing_labels(
              query=query,
              docs=list(pool_docs.values()),
              force_refresh=force_refresh_labels,
          )
  
          search_labeled_results: List[Dict[str, Any]] = []
          for rank, doc in enumerate(search_results, start=1):
              spu_id = str(doc.get("spu_id"))
              label = labels.get(spu_id)
              search_labeled_results.append(
                  {
                      "rank": rank,
                      "spu_id": spu_id,
                      "title": build_display_title(doc),
                      "image_url": doc.get("image_url"),
                      "rerank_score": None,
                      "label": label,
                      "option_values": list(compact_option_values(doc.get("skus") or [])),
                      "product": compact_product_payload(doc),
                  }
              )
  
          rerank_top_results: List[Dict[str, Any]] = []
          for rank, item in enumerate(full_rerank[:rerank_depth_effective], start=1):
              doc = item["doc"]
              spu_id = str(item["spu_id"])
              rerank_top_results.append(
                  {
                      "rank": rank,
                      "spu_id": spu_id,
                      "title": build_display_title(doc),
                      "image_url": doc.get("image_url"),
                      "rerank_score": round(float(item["score"]), 8),
                      "label": labels.get(spu_id),
                      "option_values": list(compact_option_values(doc.get("skus") or [])),
                      "product": compact_product_payload(doc),
                  }
              )
  
          top100_labels = [
              item["label"] if item["label"] in VALID_LABELS else RELEVANCE_IRRELEVANT
              for item in search_labeled_results[:100]
          ]
          metrics = compute_query_metrics(top100_labels)
          output_dir = ensure_dir(self.artifact_root / "query_builds")
          run_id = f"{utc_timestamp()}_{sha1_text(self.tenant_id + '|' + query)[:10]}"
          output_json_path = output_dir / f"{run_id}.json"
          payload = {
              "run_id": run_id,
              "created_at": utc_now_iso(),
              "tenant_id": self.tenant_id,
              "query": query,
              "config_meta": requests.get("http://localhost:6002/admin/config/meta", timeout=20).json(),
              "search_total": int(search_payload.get("total") or 0),
              "search_depth_requested": search_depth,
              "search_depth_effective": len(search_results),
              "rerank_depth_requested": rerank_depth,
              "rerank_depth_effective": rerank_depth_effective,
              "corpus_size": len(corpus),
              "annotation_pool": {
                  "annotate_search_top_k": annotate_search_top_k,
                  "annotate_rerank_top_k": annotate_rerank_top_k,
                  "pool_size": len(pool_docs),
              },
              "labeler_mode": self.labeler_mode,
              "query_profile": self.get_query_profile(query, force_refresh=force_refresh_labels) if self.labeler_mode == "complex" else None,
              "metrics_top100": metrics,
              "search_results": search_labeled_results,
              "full_rerank_top": rerank_top_results,
          }
          output_json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
          self.store.insert_build_run(run_id, self.tenant_id, query, output_json_path, payload["metrics_top100"])
          return QueryBuildResult(
              query=query,
              tenant_id=self.tenant_id,
              search_total=int(search_payload.get("total") or 0),
              search_depth=len(search_results),
              rerank_corpus_size=len(corpus),
              annotated_count=len(pool_docs),
              output_json_path=output_json_path,
          )
  
      def evaluate_live_query(
          self,
          query: str,
          top_k: int = 100,
          auto_annotate: bool = False,
          language: str = "en",
          force_refresh_labels: bool = False,
      ) -> Dict[str, Any]:
          search_payload = self.search_client.search(query=query, size=max(top_k, 100), from_=0, language=language)
          results = list(search_payload.get("results") or [])
          if auto_annotate:
              self.annotate_missing_labels(query=query, docs=results[:top_k], force_refresh=force_refresh_labels)
          labels = self.store.get_labels(self.tenant_id, query)
          recalled_spu_ids = {str(doc.get("spu_id")) for doc in results[:top_k]}
          labeled = []
          unlabeled_hits = 0
          for rank, doc in enumerate(results[:top_k], start=1):
              spu_id = str(doc.get("spu_id"))
              label = labels.get(spu_id)
              if label not in VALID_LABELS:
                  unlabeled_hits += 1
              labeled.append(
                  {
                      "rank": rank,
                      "spu_id": spu_id,
                      "title": build_display_title(doc),
                      "image_url": doc.get("image_url"),
                      "label": label,
                      "option_values": list(compact_option_values(doc.get("skus") or [])),
                      "product": compact_product_payload(doc),
                  }
              )
          metric_labels = [
              item["label"] if item["label"] in VALID_LABELS else RELEVANCE_IRRELEVANT
              for item in labeled
          ]
          label_stats = self.store.get_query_label_stats(self.tenant_id, query)
          rerank_scores = self.store.get_rerank_scores(self.tenant_id, query)
          relevant_missing_ids = [
              spu_id
              for spu_id, label in labels.items()
              if label in {RELEVANCE_EXACT, RELEVANCE_PARTIAL} and spu_id not in recalled_spu_ids
          ]
          missing_docs_map = self.store.get_corpus_docs_by_spu_ids(self.tenant_id, relevant_missing_ids)
          missing_relevant = []
          for spu_id in relevant_missing_ids:
              doc = missing_docs_map.get(spu_id)
              if not doc:
                  continue
              missing_relevant.append(
                  {
                      "spu_id": spu_id,
                      "label": labels[spu_id],
                      "rerank_score": rerank_scores.get(spu_id),
                      "title": build_display_title(doc),
                      "image_url": doc.get("image_url"),
                      "option_values": list(compact_option_values(doc.get("skus") or [])),
                      "product": compact_product_payload(doc),
                  }
              )
          label_order = {RELEVANCE_EXACT: 0, RELEVANCE_PARTIAL: 1, RELEVANCE_IRRELEVANT: 2}
          missing_relevant.sort(
              key=lambda item: (
                  label_order.get(str(item.get("label")), 9),
                  -(float(item.get("rerank_score")) if item.get("rerank_score") is not None else float("-inf")),
                  str(item.get("title") or ""),
              )
          )
          tips: List[str] = []
          if auto_annotate:
              tips.append("Single-query evaluation used cached labels and refreshed missing labels for recalled results.")
          else:
              tips.append("Single-query evaluation used the offline annotation cache only; recalled SPUs without cached labels were treated as Irrelevant.")
          if label_stats["total"] == 0:
              tips.append("This query has no offline annotation set yet. Build or refresh labels first if you want stable evaluation.")
          if unlabeled_hits:
              tips.append(f"{unlabeled_hits} recalled results were not in the annotation set and were counted as Irrelevant.")
          if not missing_relevant:
              tips.append("No cached Exact/Partial products were missed by this recall set.")
          return {
              "query": query,
              "tenant_id": self.tenant_id,
              "top_k": top_k,
              "metrics": compute_query_metrics(metric_labels),
              "results": labeled,
              "missing_relevant": missing_relevant,
              "label_stats": {
                  **label_stats,
                  "unlabeled_hits_treated_irrelevant": unlabeled_hits,
                  "recalled_hits": len(labeled),
                  "missing_relevant_count": len(missing_relevant),
                  "missing_exact_count": sum(1 for item in missing_relevant if item["label"] == RELEVANCE_EXACT),
                  "missing_partial_count": sum(1 for item in missing_relevant if item["label"] == RELEVANCE_PARTIAL),
              },
              "tips": tips,
              "total": int(search_payload.get("total") or 0),
          }
  
      def batch_evaluate(
          self,
          queries: Sequence[str],
          *,
          top_k: int = 100,
          auto_annotate: bool = True,
          language: str = "en",
          force_refresh_labels: bool = False,
      ) -> Dict[str, Any]:
          per_query = []
          for query in queries:
              live = self.evaluate_live_query(
                  query,
                  top_k=top_k,
                  auto_annotate=auto_annotate,
                  language=language,
                  force_refresh_labels=force_refresh_labels,
              )
              labels = [
                  item["label"] if item["label"] in VALID_LABELS else RELEVANCE_IRRELEVANT
                  for item in live["results"]
              ]
              per_query.append(
                  {
                      "query": live["query"],
                      "tenant_id": live["tenant_id"],
                      "top_k": live["top_k"],
                      "metrics": live["metrics"],
                      "distribution": label_distribution(labels),
                      "total": live["total"],
                  }
              )
          aggregate = aggregate_metrics([item["metrics"] for item in per_query])
          aggregate_distribution = {
              RELEVANCE_EXACT: sum(item["distribution"][RELEVANCE_EXACT] for item in per_query),
              RELEVANCE_PARTIAL: sum(item["distribution"][RELEVANCE_PARTIAL] for item in per_query),
              RELEVANCE_IRRELEVANT: sum(item["distribution"][RELEVANCE_IRRELEVANT] for item in per_query),
          }
          batch_id = f"batch_{utc_timestamp()}_{sha1_text(self.tenant_id + '|' + '|'.join(queries))[:10]}"
          report_dir = ensure_dir(self.artifact_root / "batch_reports")
          config_snapshot_path = report_dir / f"{batch_id}_config.json"
          config_snapshot = requests.get("http://localhost:6002/admin/config", timeout=20).json()
          config_snapshot_path.write_text(json.dumps(config_snapshot, ensure_ascii=False, indent=2), encoding="utf-8")
          output_json_path = report_dir / f"{batch_id}.json"
          report_md_path = report_dir / f"{batch_id}.md"
          payload = {
              "batch_id": batch_id,
              "created_at": utc_now_iso(),
              "tenant_id": self.tenant_id,
              "queries": list(queries),
              "top_k": top_k,
              "aggregate_metrics": aggregate,
              "aggregate_distribution": aggregate_distribution,
              "per_query": per_query,
              "config_snapshot_path": str(config_snapshot_path),
          }
          output_json_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
          report_md_path.write_text(render_batch_report_markdown(payload), encoding="utf-8")
          self.store.insert_batch_run(batch_id, self.tenant_id, output_json_path, report_md_path, config_snapshot_path, payload)
          return payload