"""RetrievalQueryEvaluator — validates intent-to-query translation. Selector / includes / excludes match against ``queryString`` only. ``filterExpression`` is captured in normalized telemetry (for diagnostics) but explicitly excluded from selector matching in V1 — see ``contracts/retrieval-query-evaluator.md`` "filterExpression scope (V1)" for the telemetry-driven rationale. """ from __future__ import annotations from dataclasses import dataclass from typing import Any, Dict, List, Optional from common import STATUS_FAIL, STATUS_PASS from .utils import ( FAIL_SCORE, PASS_SCORE, THRESHOLD, get_retrieval_executions, has_retrieval_failure, normalized_contains, ) # The diagnostic_code values. _DIAG_PASS = "pass" _DIAG_NO_MATCHING_QUERY = "no_matching_query" _DIAG_REQUIRED_TERMS_MISSING = "required_terms_missing" _DIAG_EXCLUDED_TERMS_FOUND = "excluded_terms_found" _DIAG_MIXED_TERM_FAILURE = "mixed_term_failure" _DIAG_RETRIEVAL_FAILURE = "retrieval_failure" @dataclass(frozen=True) class _TermFailure: includes_missing: List[str] excludes_found: List[str] class RetrievalQueryEvaluator: """Validate that Copilot's retrieval queries match expected patterns. Configuration parameters (``capability``, ``selector``, ``includes``, ``excludes``) are set via the constructor from evaluator config and are not accepted at call time. """ def __init__( self, capability: str, selector: str, includes: Optional[List[str]] = None, excludes: Optional[List[str]] = None, ) -> None: if not isinstance(capability, str) or not capability: raise ValueError( "RetrievalQueryEvaluator: 'capability' is required and must be a non-empty string." ) if not isinstance(selector, str) or not selector: raise ValueError( "RetrievalQueryEvaluator: 'selector' is required and must be a non-empty string." ) if includes is not None and ( not isinstance(includes, list) or not all(isinstance(x, str) for x in includes) ): raise ValueError( "RetrievalQueryEvaluator: 'includes' must be a list of strings." ) if excludes is not None and ( not isinstance(excludes, list) or not all(isinstance(x, str) for x in excludes) ): raise ValueError( "RetrievalQueryEvaluator: 'excludes' must be a list of strings." ) self.capability = capability self.selector = selector self.includes = list(includes) if includes else [] self.excludes = list(excludes) if excludes else [] def __call__( self, *, retrieval_telemetry: Optional[Dict[str, Any]] = None, response: Optional[str] = None, **kwargs: Any, ) -> Dict[str, Any]: """Evaluate retrieval telemetry against the configured assertions. Returns the score dict described in ``contracts/retrieval-query-evaluator.md`` Return Value Schema. """ includes = self.includes excludes = self.excludes executions = get_retrieval_executions( retrieval_telemetry, capability=self.capability ) if has_retrieval_failure( retrieval_telemetry, executions, capability=self.capability ): return _build_result( score=FAIL_SCORE, status=STATUS_FAIL, diagnostic_code=_DIAG_RETRIEVAL_FAILURE, reason="Retrieval telemetry unavailable or selected execution failed", includes_missing=list(includes), ) matched_queries = _find_query_texts_by_selector(executions, self.selector) if not matched_queries: return _build_result( score=FAIL_SCORE, status=STATUS_FAIL, diagnostic_code=_DIAG_NO_MATCHING_QUERY, reason=( f"No query matching selector '{self.selector}' found in retrieval executions" ), includes_missing=list(includes), ) # Pass if any matched queryString satisfies includes and excludes. # Materialize once: needed both for the any() check and the failure # aggregation below. failures = [ _evaluate_terms(query_text, includes, excludes) for query_text in matched_queries ] for query_text, failure in zip(matched_queries, failures): if not failure.includes_missing and not failure.excludes_found: return _build_result( score=PASS_SCORE, status=STATUS_PASS, diagnostic_code=_DIAG_PASS, reason=_format_pass_reason( query_text, selector=self.selector, includes=includes, excludes=excludes, ), matched_queries=matched_queries, ) includes_missing = sorted({ term for f in failures for term in f.includes_missing }) excludes_found = sorted({ term for f in failures for term in f.excludes_found }) if excludes_found and includes_missing: diag_code = _DIAG_MIXED_TERM_FAILURE reason = ( f"Missing required: {includes_missing}; found excluded: {excludes_found}" ) elif excludes_found: diag_code = _DIAG_EXCLUDED_TERMS_FOUND reason = f"Found excluded terms: {excludes_found}" else: diag_code = _DIAG_REQUIRED_TERMS_MISSING reason = f"Missing required terms: {includes_missing}" return _build_result( score=FAIL_SCORE, status=STATUS_FAIL, diagnostic_code=diag_code, reason=reason, matched_queries=matched_queries, includes_missing=includes_missing, excludes_found=excludes_found, ) def _build_result( *, score: float, status: str, diagnostic_code: str, reason: str, matched_queries: Optional[List[str]] = None, includes_missing: Optional[List[str]] = None, excludes_found: Optional[List[str]] = None, ) -> Dict[str, Any]: """Assemble the eight-key Return Value Schema dict with sensible defaults. Single source of truth for the output shape — every return path in ``__call__`` routes through here, so adding a field to the schema is a one-line change. """ return { "retrieval_query": score, "result": status, "threshold": THRESHOLD, "diagnostic_code": diagnostic_code, "reason": reason, "matched_queries": matched_queries if matched_queries is not None else [], "includes_missing": includes_missing if includes_missing is not None else [], "excludes_found": excludes_found if excludes_found is not None else [], } def _format_pass_reason( query_text: str, *, selector: str, includes: List[str], excludes: List[str], ) -> str: if includes: formatted_includes = ", ".join(repr(term) for term in includes) base = ( f"queryString {query_text!r} has all inclusion terms: " f"{formatted_includes}" ) else: base = f"queryString {query_text!r} matches selector {selector!r}" if excludes: formatted_excludes = ", ".join(repr(term) for term in excludes) return f"{base} and avoids all exclusion terms: {formatted_excludes}" return base def _find_query_texts_by_selector( executions: List[Dict[str, Any]], selector: str, ) -> List[str]: """Return ``queryString`` values containing the selector (case-insensitive). Iterates ``execution.queries[].queryString`` only. ``filterExpression`` is intentionally excluded from selector matching in V1. """ matched: List[str] = [] for execution in executions: queries = execution.get("queries") or [] if not isinstance(queries, list): continue for query in queries: if not isinstance(query, dict): continue query_string = query.get("queryString") if not isinstance(query_string, str) or not query_string: continue if normalized_contains(query_string, selector): matched.append(query_string) return matched def _evaluate_terms( query_text: str, includes: List[str], excludes: List[str], ) -> _TermFailure: """Return the includes-missing and excludes-found terms for a query.""" missing = [t for t in includes if not normalized_contains(query_text, t)] found = [t for t in excludes if normalized_contains(query_text, t)] return _TermFailure(includes_missing=missing, excludes_found=found)