"""Integrity audit and descriptive summaries for immutable E00 artifacts.""" from __future__ import annotations from collections import defaultdict import json from pathlib import Path from statistics import mean, median from typing import Any, Sequence from .telemetry import ALLOWED_EVENT_TYPES class AnalysisError(RuntimeError): """Raised when raw artifacts fail integrity checks.""" SUMMARY_METRICS = ( "file_recall_at_1", "file_recall_at_5", "file_recall_at_10", "mrr", "ndcg_at_10", "query_seconds", ) def load_json(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise AnalysisError(f"Cannot read JSON artifact {path}: {exc}") from exc if not isinstance(value, dict): raise AnalysisError(f"Expected a JSON object in {path}") return value def audit_pilot_artifacts(root: Path, report: dict[str, Any]) -> dict[str, Any]: errors: list[str] = [] run_count = 0 event_count = 0 for row in report.get("runs", []): run_id = str(row["run_id"]) directory = ( root / "results" / "raw" / str(report["experiment_id"]) / str(row["harness_id"]) / str(row["task_id"]) / run_id ) required = ( "run_manifest.json", "trajectory.jsonl", "ranking.json", "final_metrics.json", ) missing = [name for name in required if not (directory / name).is_file()] if missing: errors.append(f"{run_id} missing artifacts {missing}") continue manifest = load_json(directory / "run_manifest.json") metrics = load_json(directory / "final_metrics.json") if manifest.get("run_id") != run_id: errors.append(f"{run_id} manifest identity mismatch") for metric in SUMMARY_METRICS: if metrics.get(metric) != row.get(metric): errors.append(f"{run_id} report differs from final_metrics for {metric}") try: ranking = json.loads((directory / "ranking.json").read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: errors.append(f"{run_id} invalid ranking: {exc}") ranking = None if not isinstance(ranking, list): errors.append(f"{run_id} ranking must be an array") events: list[dict[str, Any]] = [] for line_number, line in enumerate( (directory / "trajectory.jsonl").read_text(encoding="utf-8").splitlines(), start=1, ): try: event = json.loads(line) except json.JSONDecodeError as exc: errors.append(f"{run_id} invalid trajectory line {line_number}: {exc}") continue if not isinstance(event, dict): errors.append(f"{run_id} trajectory line {line_number} is not an object") continue events.append(event) if [event.get("sequence") for event in events] != list(range(len(events))): errors.append(f"{run_id} trajectory sequence is not contiguous") if any(event.get("run_id") != run_id for event in events): errors.append(f"{run_id} trajectory contains a foreign run_id") unknown = { str(event.get("event_type")) for event in events if event.get("event_type") not in ALLOWED_EVENT_TYPES } if unknown: errors.append(f"{run_id} trajectory contains unknown events {sorted(unknown)}") if not events or events[0].get("event_type") != "run_started": errors.append(f"{run_id} trajectory does not start with run_started") if not events or events[-1].get("event_type") != "run_finished": errors.append(f"{run_id} trajectory does not end with run_finished") run_count += 1 event_count += len(events) expected_runs = int(report.get("run_count", -1)) if run_count != expected_runs: errors.append(f"audited {run_count} runs but report declares {expected_runs}") if errors: raise AnalysisError("Artifact audit failed: " + " | ".join(errors)) return { "status": "passed", "run_count": run_count, "event_count": event_count, "required_artifacts_per_run": 4, } def summarize_rows(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) by_task_harness: dict[tuple[str, str], dict[str, Any]] = {} for row in rows: harness_id = str(row["harness_id"]) grouped[harness_id].append(row) by_task_harness[(str(row["task_id"]), harness_id)] = row treatments: dict[str, Any] = {} for harness_id, harness_rows in sorted(grouped.items()): treatment: dict[str, Any] = { "task_count": len(harness_rows), "all_gold_in_top_10_count": sum(bool(row["all_gold_in_top_10"]) for row in harness_rows), } for metric in SUMMARY_METRICS: values = [float(row[metric]) for row in harness_rows] treatment[f"mean_{metric}"] = mean(values) treatment[f"median_{metric}"] = median(values) if harness_id == "H003": treatment["total_embedded_chunks"] = sum(int(row["embedded_chunks"]) for row in harness_rows) treatment["total_cached_chunks"] = sum(int(row["cached_chunks"]) for row in harness_rows) treatment["total_index_build_seconds"] = sum(float(row["build_seconds"]) for row in harness_rows) treatments[harness_id] = treatment tasks = sorted({str(row["task_id"]) for row in rows}) paired: dict[str, Any] = {} for treatment, baseline in (("H001", "H000"), ("H003", "H000")): comparison = f"{treatment}_minus_{baseline}" comparison_metrics: dict[str, Any] = {} for metric in ("file_recall_at_5", "file_recall_at_10", "mrr", "ndcg_at_10"): differences = [ float(by_task_harness[(task, treatment)][metric]) - float(by_task_harness[(task, baseline)][metric]) for task in tasks if (task, treatment) in by_task_harness and (task, baseline) in by_task_harness ] if differences: comparison_metrics[f"mean_delta_{metric}"] = mean(differences) comparison_metrics[f"task_deltas_{metric}"] = differences if comparison_metrics: paired[comparison] = comparison_metrics return {"treatments": treatments, "paired_differences": paired} def analyze_pilot_report(root: Path, report_path: Path, analysis_revision: str) -> dict[str, Any]: report = load_json(report_path) audit = audit_pilot_artifacts(root, report) summary = summarize_rows(report["runs"]) return { "schema_version": 1, "experiment_id": report["experiment_id"], "development_only": True, "raw_report": str(report_path.resolve()), "experiment_code_revision": report["code_revision"], "analysis_code_revision": analysis_revision, "audit": audit, **summary, }