| """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, |
| } |
|
|
