#!/usr/bin/env python3 """Run labelled recall cases through the real MCP transport. The default suite is isolated, synthetic and offline. --live requires a custom case file and uses the current library/providers; it may incur provider costs. """ from __future__ import annotations import argparse from collections import Counter import json import math import os import re from pathlib import Path import statistics import tempfile import time from xkb_doctor import assess, probe from xkb_evidence import fields, identity_key, record_id DEFAULT_CASES = Path(__file__).resolve().parent.parent / "evals" / "recall-cases.json" def fixture_env(root: Path, fixtures: dict) -> dict[str, str]: # Do not inherit credentials, runtime files, paths or remote endpoints. env = {key: value for key, value in os.environ.items() if key.upper() in { "SYSTEMROOT", "SYSTEMDRIVE", "WINDIR", "COMSPEC", "PATHEXT", "TEMP", "TMP", "PATH", "NUMBER_OF_PROCESSORS", "PROCESSOR_ARCHITECTURE"}} data = root / "data" index = data / "bookmarks" / "search_index.json" index.parent.mkdir(parents=True) index.write_text(json.dumps({"items": fixtures.get("cards", [])}, ensure_ascii=True), encoding="utf-8") wiki = data / "x-knowledge-base" / "wiki" / "topics" wiki.mkdir(parents=True) for item in fixtures.get("wiki", []): if not item["id"].replace("-", "").replace("_", "").isalnum(): raise ValueError("invalid fixture wiki id") (wiki / (item["id"] + ".md")).write_text(item["content"], encoding="utf-8") env.update({"HOME": str(root), "USERPROFILE": str(root), "XKB_DATA_DIR": str(data), "XKB_CONFIG": str(root / "absent.json"), "XKB_SERVICE_DB": str(root / "knowledge.sqlite"), "XKB_EMBEDDING_TIMEOUT": "1", "XKB_XBRAIN_TIMEOUT": "1"}) return env def score_case(case: dict, packet: dict) -> dict: surface = case.get("surface", "records") if surface not in {"records", "delivery"}: raise ValueError("surface must be records or delivery") if surface == "delivery" and not isinstance(packet.get("delivery"), dict): raise ValueError("service did not return a delivery decision") records = packet["delivery"]["records"] if surface == "delivery" else packet["records"] record_ids = [record_id(r) for r in records] evidence_keys = [r.get("evidence_key") or identity_key(r) for r in records] label_unit = case.get("label_unit", "document") if label_unit not in {"document", "evidence"}: raise ValueError("label_unit must be document or evidence") actual = set(evidence_keys if label_unit == "evidence" else record_ids) expected = set(case["expected_ids"]) alternatives = set(case.get("expected_any_ids", [])) allowed = set(case.get("allowed_ids", expected | alternatives)) missing = sorted(expected - actual) unexpected = sorted(actual - allowed) errors = [] if surface == 'delivery': # A provenance note naming the source of this wiki line is not a # procedure for tracing the user's data source. Score the claim text. bodies = [re.sub(r'\*\((?:self-derived\b|source:)[^\n]*?\)\*', '', fields(record)[1]) for record in records] excerpts = '\n'.join(bodies) if any(not any(c.isalpha() for c in body) for body in bodies): errors.append('content-free delivered excerpt') for pattern in case.get('required_excerpt_patterns', []): if not re.search(pattern, excerpts, re.IGNORECASE): errors.append('missing excerpt requirement: ' + pattern) for pattern in case.get('forbidden_excerpt_patterns', []): if re.search(pattern, excerpts, re.IGNORECASE): errors.append('forbidden excerpt: ' + pattern) if surface == "delivery" and packet["delivery"].get("status") != "ready": errors.append("proactive delivery judgement incomplete") if len(records) > case.get("max_delivered", len(records)): errors.append("delivery budget exceeded") if missing: errors.append("missing: " + ", ".join(missing)) if alternatives and not alternatives.intersection(actual): errors.append("missed required evidence group") if unexpected: errors.append("unexpected: " + ", ".join(unexpected)) if case.get("expected_delivery") == "none" and records: errors.append("unnecessary interruption") if case.get("expected_delivery") == "evidence" and not records: errors.append("missed conversational need") if case.get("retrieval_mode") and packet["retrieval_mode"] != case["retrieval_mode"]: errors.append("wrong retrieval mode") identified = [key for key in evidence_keys if key] if len(identified) != len(set(identified)): errors.append("duplicate evidence") if len(identified) != len(evidence_keys): errors.append("unidentified evidence") return {"id": case["id"], "ok": not errors, "errors": errors, "surface": surface, "record_ids": record_ids, "evidence_keys": evidence_keys, "label_unit": label_unit, "recall_at_k": ((len(expected & actual) + bool(alternatives & actual)) / (len(expected) + bool(alternatives))) if expected or alternatives else None, "precision_at_k": len(allowed & actual) / len(actual) if actual else None, "no_answer": not expected and not alternatives, "false_positive_count": len(unexpected), "retrieval_mode": packet["retrieval_mode"], "judge_status": packet.get("judge", {}).get("status", "not_attempted"), "expected_delivery": case.get("expected_delivery"), "need": case.get("need", ""), "conversation_messages": packet.get("conversation_messages", 0)} def run_cases(cases: list[dict], env: dict | None = None, *, require_semantic: bool = False, require_judge: bool = False) -> dict: rows = [] for case in cases: started = time.monotonic() try: # A case namespace overrides only for this child, not future cases. from runtime_config import runtime_env child_env = dict(runtime_env() if env is None else env) if "namespace" in case: child_env["XKB_NAMESPACE"] = case["namespace"] packet = probe(case["query"], env=child_env, limit=case.get("limit", 10), **({"conversation": case["conversation"]} if "conversation" in case else {})) row = score_case(case, packet) readiness = assess(packet, require_semantic=require_semantic, require_judge=require_judge) row["errors"].extend(readiness["problems"]) row["ok"] = not row["errors"] except Exception as exc: row = {"id": case["id"], "ok": False, "errors": [str(exc)], "retrieval_mode": "failed", "judge_status": "not_attempted", "recall_at_k": 0.0 if case["expected_ids"] or case.get("expected_any_ids") else None, "precision_at_k": None, "no_answer": not case["expected_ids"] and not case.get("expected_any_ids")} row["elapsed_ms"] = round((time.monotonic() - started) * 1000) rows.append(row) return summarize(rows) def summarize(rows: list[dict]) -> dict: latencies = sorted(r["elapsed_ms"] for r in rows) recalls = [r["recall_at_k"] for r in rows if r.get("recall_at_k") is not None] precisions = [r["precision_at_k"] for r in rows if r.get("precision_at_k") is not None] negatives = [r for r in rows if r.get("no_answer")] return {"ok": all(r["ok"] for r in rows), "cases": len(rows), "passed": sum(r["ok"] for r in rows), "mean_recall_at_k": statistics.mean(recalls) if recalls else None, "mean_precision_at_k": statistics.mean(precisions) if precisions else None, "no_answer_passed": sum(r["ok"] for r in negatives), "no_answer_cases": len(negatives), "latency_ms": {"p50": statistics.median(latencies), "p95": latencies[math.ceil(len(latencies) * .95) - 1]}, "retrieval_modes": dict(Counter(r["retrieval_mode"] for r in rows)), "judge_statuses": dict(Counter(r["judge_status"] for r in rows)), "results": rows} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--cases", type=Path) parser.add_argument("--live", action="store_true") parser.add_argument("--require-semantic", action="store_true") parser.add_argument("--require-judge", action="store_true") parser.add_argument("--output", type=Path) args = parser.parse_args() if args.live and args.cases is None: parser.error("--live requires --cases with labels for your own library") try: suite = json.loads((args.cases or DEFAULT_CASES).read_text(encoding="utf-8")) cases = suite["cases"] if (not isinstance(cases, list) or not cases or len({c["id"] for c in cases}) != len(cases)): raise ValueError("cases must be a non-empty list with unique ids") for case in cases: for field in ('required_excerpt_patterns', 'forbidden_excerpt_patterns'): patterns = case.get(field, []) if not isinstance(patterns, list) or any(not isinstance(p, str) or not p for p in patterns): raise ValueError(field + ' must contain non-empty regular expressions') for pattern in patterns: re.compile(pattern) if (not isinstance(case["query"], str) or not case["query"].strip() or not isinstance(case["expected_ids"], list) or not all(isinstance(x, str) for x in case["expected_ids"])): raise ValueError("each case needs a query and expected_ids list") from xkb_recall import conversation_messages if "conversation" in case: conversation_messages(case["conversation"]) if case.get("expected_delivery") not in {None, "none", "evidence"}: raise ValueError("expected_delivery must be none or evidence") alternatives = case.get("expected_any_ids", []) if not isinstance(alternatives, list) or not all(isinstance(x, str) for x in alternatives): raise ValueError("expected_any_ids must be a list of equivalent evidence ids") if case.get("label_unit", "document") not in {"document", "evidence"}: raise ValueError("label_unit must be document or evidence") allowed = case.get("allowed_ids", case["expected_ids"] + alternatives) if (not isinstance(allowed, list) or not all(isinstance(x, str) for x in allowed) or not set(case["expected_ids"] + alternatives).issubset(allowed)): raise ValueError("allowed_ids must be a list containing all expected_ids") if ("max_delivered" in case and (type(case["max_delivered"]) is not int or case["max_delivered"] < 0)): raise ValueError("max_delivered must be a non-negative integer") if args.live: report = run_cases(cases, require_semantic=args.require_semantic, require_judge=args.require_judge) else: if "fixtures" not in suite: raise ValueError("offline mode needs fixtures; use --live for an existing library") with tempfile.TemporaryDirectory(prefix="xkb-eval-") as tmp: report = run_cases(cases, fixture_env(Path(tmp), suite["fixtures"]), require_semantic=args.require_semantic, require_judge=args.require_judge) report["scope"] = ("live library" if args.live else "synthetic offline keyword/ACL/transport regression; not a semantic quality benchmark") except Exception as exc: report = {"ok": False, "error": str(exc)} output = json.dumps(report, ensure_ascii=True, indent=2) if args.output: args.output.write_text(output + "\n", encoding="utf-8") print(output) return 0 if report["ok"] else 1 if __name__ == "__main__": import xkb_usage xkb_usage.record(__file__) raise SystemExit(main())