"""Score saved runtime traces against evaluator-private reference chains.""" from __future__ import annotations import argparse import asyncio import hashlib import json import statistics from pathlib import Path from typing import Any, Dict, Iterable, List, Mapping from eviscene.orchestrator.llm_backend import APIBackend from eviscene.trajectory.reference_chain import ( LLMReferenceEvidenceMatcher, score_reference_chain_trace, ) def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Evaluate how completely saved traces recover the frozen decisive " "reference chain." ) ) parser.add_argument("--run-dir", required=True) parser.add_argument( "--gold", default=None, help="Optional evaluator_private/gold.jsonl override.", ) parser.add_argument( "--output", default=None, help="Defaults to /reference_chain_metrics.jsonl.", ) parser.add_argument( "--semantic-judge", action="store_true", help=( "Use an LLM only for qualified evidence unresolved by deterministic " "matching." ), ) parser.add_argument("--provider", default=None) parser.add_argument("--model", default=None) parser.add_argument( "--wire-api", default=None, choices=["interactions", "responses", "chat_completions"], ) parser.add_argument("--timeout", type=float, default=120.0) parser.add_argument( "--max-llm-candidates-per-fact", type=int, default=3, ) return parser.parse_args() def _load_json(path: Path) -> Dict[str, Any]: payload = json.loads(path.read_text(encoding="utf-8")) if not isinstance(payload, dict): raise TypeError(f"expected JSON object: {path}") return payload def _load_jsonl(path: Path) -> List[Dict[str, Any]]: rows: List[Dict[str, Any]] = [] for line in path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue payload = json.loads(line) if not isinstance(payload, dict): raise TypeError(f"expected JSONL object row: {path}") rows.append(payload) return rows def _write_jsonl(path: Path, rows: Iterable[Mapping[str, Any]]) -> None: rendered = "\n".join( json.dumps(row, ensure_ascii=False, separators=(",", ":")) for row in rows ) pending = path.with_name(f".{path.name}.tmp") path.parent.mkdir(parents=True, exist_ok=True) pending.write_text(rendered + ("\n" if rendered else ""), encoding="utf-8") pending.replace(path) def _sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _resolve_gold_path( run_dir: Path, manifest: Mapping[str, Any], override: str | None, ) -> Path: if override: path = Path(override).expanduser().resolve() if path.is_file(): return path raise FileNotFoundError(path) benchmark = manifest.get("benchmark") if isinstance(benchmark, Mapping): descriptor = benchmark.get("evaluation_gold") if isinstance(descriptor, Mapping): configured = str(descriptor.get("path", "")).strip() if configured and Path(configured).is_file(): return Path(configured).resolve() benchmark_path = str(benchmark.get("path", "")).strip() if benchmark_path: candidate = ( Path(benchmark_path).resolve().parent.parent / "evaluator_private" / "gold.jsonl" ) if candidate.is_file(): return candidate raise FileNotFoundError( "could not resolve evaluator-private gold from run manifest; pass --gold" ) def _private_index( rows: Iterable[Mapping[str, Any]], ) -> Dict[str, Dict[str, Any]]: indexed: Dict[str, Dict[str, Any]] = {} for row in rows: case_id = str(row.get("case_id", "")).strip() if not case_id: raise ValueError("gold row lacks case_id") if case_id in indexed: raise ValueError(f"duplicate gold case_id: {case_id}") indexed[case_id] = dict(row) return indexed def _mean_metrics(rows: Iterable[Mapping[str, Any]]) -> Dict[str, float]: collected: Dict[str, List[float]] = {} for row in rows: metrics = row.get("metrics") if not isinstance(metrics, Mapping): continue for key, value in metrics.items(): if isinstance(value, (int, float)): collected.setdefault(str(key), []).append(float(value)) return { key: round(statistics.mean(values), 6) for key, values in sorted(collected.items()) if values } async def _run(args: argparse.Namespace) -> Dict[str, Any]: run_dir = Path(args.run_dir).expanduser().resolve() manifest = _load_json(run_dir / "run_manifest.json") gold_path = _resolve_gold_path(run_dir, manifest, args.gold) gold_index = _private_index(_load_jsonl(gold_path)) output_path = ( Path(args.output).expanduser().resolve() if args.output else run_dir / "reference_chain_metrics.jsonl" ) agent_config = manifest.get("agent") if not isinstance(agent_config, Mapping): agent_config = {} provider = str(args.provider or agent_config.get("provider") or "gemini") model = str(args.model or agent_config.get("model") or "gemini-3.7-flash") wire_api = args.wire_api or agent_config.get("llm_wire_api") backend: APIBackend | None = None matcher: LLMReferenceEvidenceMatcher | None = None if args.semantic_judge: backend = APIBackend( provider=provider, model_name=model, wire_api=wire_api, temperature=0.0, max_tokens=500, timeout=args.timeout, ) matcher = LLMReferenceEvidenceMatcher( backend, provider=provider, model=model, ) score_metadata = { "evaluation_gold": { "path": str(gold_path), "sha256": _sha256(gold_path), }, "source_run_id": manifest.get("run_id"), "source_runtime_commit": manifest.get("git_commit"), } rows: List[Dict[str, Any]] = [] try: for case_id, gold in gold_index.items(): trace_path = run_dir / "traces" / f"{case_id}.json" if not trace_path.is_file(): rows.append( { "schema_version": "ifv-reference-chain-metrics-v2", "case_id": case_id, "score_metadata": score_metadata, "engineering_error": True, "error": "canonical trace is missing", "metrics": { "fact_recovery_recall": 0.0, "chain_recovery_recall": 0.0, "evidence_recovery_recall": 0.0, "basis_reference_precision": 0.0, }, } ) continue trace = _load_json(trace_path) rows.append( await score_reference_chain_trace( trace, gold, semantic_matcher=matcher, score_metadata=score_metadata, max_llm_candidates_per_fact=( args.max_llm_candidates_per_fact ), ) ) finally: if backend is not None: await backend.aclose() _write_jsonl(output_path, rows) summary = { "run_dir": str(run_dir), "output": str(output_path), "case_count": len(rows), "semantic_judge_enabled": bool(args.semantic_judge), "mean_metrics": _mean_metrics(rows), "semantic_judge": dict(matcher.metadata) if matcher else None, "matcher_error_count": sum( len( row.get("semantic_matcher", {}).get("errors", []) if isinstance(row.get("semantic_matcher"), Mapping) else [] ) for row in rows ), } return summary def main() -> None: args = _parse_args() summary = asyncio.run(_run(args)) print(json.dumps(summary, ensure_ascii=False, indent=2)) if int(summary.get("matcher_error_count", 0) or 0): raise SystemExit(2) if __name__ == "__main__": main()