#!/usr/bin/env python3 """Run the repository-native Meeting Lab benchmark pipeline. The runner intentionally orchestrates the existing production stages without changing their prompts or semantic behavior. It adds pre-flight checks, artifact preservation and benchmark metadata around those stages. """ from __future__ import annotations import argparse import json import os import platform import re import shutil import subprocess import sys import time from collections import Counter from dataclasses import asdict from datetime import datetime from pathlib import Path from typing import Any, Callable import requests REPO_ROOT = Path(__file__).resolve().parents[1] if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) from src.meeting_lab.chunking.chunk_transcript import ( # noqa: E402 build_chunks, read_transcript_blocks, write_chunks, ) from src.meeting_lab.consolidation.canonicalize import ( # noqa: E402 canonicalize_extractions, write_canonicalized, ) from src.meeting_lab.consolidation.consolidate_facts import ( # noqa: E402 build_consolidated_output, build_consolidation_prompt, call_ollama as call_consolidator_ollama, fact_items, parse_model_json, repair_model_group_coverage, resolve_num_predict, validate_consolidated_output, validate_group_shapes, validate_model_groups, write_json, write_report as write_consolidator_report, ) from src.meeting_lab.extraction.extract_chunks import ( # noqa: E402 DEFAULT_ENDPOINT, build_prompt as build_extraction_prompt, call_ollama as call_extraction_ollama, find_normalized_chunks, normalize_current_schema, parse_json_response, ) from src.meeting_lab.models.meeting_context import ( # noqa: E402 MeetingContext, load_meeting_context, ) from src.meeting_lab.normalization.normalize_transcript import ( # noqa: E402 normalize_block, split_blocks, ) from src.meeting_lab.protocol.render_working_protocol import ( # noqa: E402 DEFAULT_NUM_PREDICT as RENDERER_NUM_PREDICT, render_working_protocol, ) DEFAULT_MODEL = "qwen3.5:9b" DEFAULT_TARGET_CHARS = 4500 DEFAULT_MAX_CHARS = 5500 DEFAULT_MIN_CHARS = 2500 DEFAULT_OVERLAP_BLOCKS = 0 DEFAULT_NUM_CTX = 32768 DEFAULT_TEMPERATURE = 0.0 DEFAULT_TIMEOUT = 1800 DEFAULT_PROGRESS_INTERVAL = 30 BENCHMARK_ROOT = Path("samples") / "benchmarks" class RunnerError(RuntimeError): """Raised for actionable runner failures.""" def __init__(self, message: str, details: dict[str, Any] | None = None) -> None: super().__init__(message) self.details = details or {} def parse_args(argv: list[str] | None = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Run the Meeting Lab production benchmark pipeline." ) parser.add_argument("--input", required=True, type=Path, help="Normalized Whisper JSON input.") parser.add_argument("--context", required=True, type=Path, help="Meeting Context YAML.") parser.add_argument("--model", default=DEFAULT_MODEL, help=f"Ollama model tag (default: {DEFAULT_MODEL}).") parser.add_argument( "--benchmark-label", default="meeting_run", help="Output directory label prefix under samples/benchmarks.", ) parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT, help=f"Ollama generate endpoint (default: {DEFAULT_ENDPOINT}).") parser.add_argument("--target-chars", type=int, default=DEFAULT_TARGET_CHARS) parser.add_argument("--max-chars", type=int, default=DEFAULT_MAX_CHARS) parser.add_argument("--min-chars", type=int, default=DEFAULT_MIN_CHARS) parser.add_argument("--overlap-blocks", type=int, default=DEFAULT_OVERLAP_BLOCKS) parser.add_argument("--temperature", type=float, default=DEFAULT_TEMPERATURE) parser.add_argument("--num-ctx", type=int, default=DEFAULT_NUM_CTX) parser.add_argument( "--num-predict", type=int, default=None, help="Extraction num_predict. Default uses the committed extraction default.", ) parser.add_argument( "--consolidator-num-predict", type=int, default=None, help="Semantic Consolidator num_predict. Default uses committed adaptive sizing.", ) parser.add_argument( "--renderer-num-predict", type=int, default=RENDERER_NUM_PREDICT, help=f"Renderer num_predict (default: {RENDERER_NUM_PREDICT}).", ) parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT) parser.add_argument("--progress-interval", type=int, default=DEFAULT_PROGRESS_INTERVAL) parser.add_argument( "--allow-loaded-models", action="store_true", help="Allow execution when `ollama ps` reports already loaded models.", ) parser.add_argument( "--output-root", type=Path, default=BENCHMARK_ROOT, help="Directory where benchmark run directories are created.", ) return parser.parse_args(argv) def sanitize_label(label: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "_", label.strip()) return cleaned.strip("._-") or "meeting_run" def create_unique_output_dir(output_root: Path, label: str, now: Callable[[], datetime] = datetime.now) -> Path: timestamp = now().strftime("%Y%m%d_%H%M%S") base = output_root / f"{sanitize_label(label)}_{timestamp}" candidate = base suffix = 1 while candidate.exists(): candidate = output_root / f"{base.name}_{suffix:02d}" suffix += 1 candidate.mkdir(parents=True) return candidate def read_json_object(path: Path) -> dict[str, Any]: try: data = json.loads(path.read_text(encoding="utf-8-sig")) except json.JSONDecodeError as exc: raise RunnerError(f"Input is not valid JSON: {path}: {exc}") from exc if not isinstance(data, dict): raise RunnerError(f"Input JSON must contain an object: {path}") return data def validate_whisper_input(path: Path) -> dict[str, Any]: if not path.is_file(): raise RunnerError(f"Input file does not exist: {path}") data = read_json_object(path) segments = data.get("segments") if not isinstance(segments, list) or not segments: raise RunnerError("Input JSON must contain a non-empty 'segments' list.") bad = [ index for index, segment in enumerate(segments) if not isinstance(segment, dict) or not isinstance(segment.get("text"), str) ] if bad: raise RunnerError(f"Input segments must be objects with string text; first bad index: {bad[0]}") return data def ollama_base_url(endpoint: str) -> str: return endpoint.rsplit("/api/", 1)[0] if "/api/" in endpoint else endpoint.rstrip("/") def require_ollama( endpoint: str, model: str, timeout: int = 10, allow_loaded_models: bool = False, ) -> dict[str, Any]: base = ollama_base_url(endpoint) try: tags_response = requests.get(f"{base}/api/tags", timeout=timeout) tags_response.raise_for_status() except requests.RequestException as exc: raise RunnerError(f"Ollama endpoint is not reachable at {base}: {exc}") from exc tags = tags_response.json() models = tags.get("models") if not isinstance(models, list): raise RunnerError(f"Ollama /api/tags returned an unexpected payload from {base}.") installed = {item.get("name") for item in models if isinstance(item, dict)} if model not in installed: raise RunnerError(f"Requested model is not installed in Ollama: {model}") active_models: list[str] = [] try: ps_response = requests.get(f"{base}/api/ps", timeout=timeout) ps_response.raise_for_status() ps_models = ps_response.json().get("models", []) if isinstance(ps_models, list): active_models = [ str(item.get("name")) for item in ps_models if isinstance(item, dict) and item.get("name") ] except requests.RequestException: active_models = ["unavailable"] result = { "base_url": base, "installed_models": sorted(installed), "active_models": active_models, "allow_loaded_models": allow_loaded_models, } if active_models and active_models != ["unavailable"] and not allow_loaded_models: raise RunnerError( "Ollama has loaded model(s). Stop them before a reference benchmark " "or pass --allow-loaded-models to record and allow this condition: " + ", ".join(active_models), details={"preflight": {"ollama": result}}, ) return result def system_metadata(model: str, endpoint: str) -> dict[str, Any]: return { "branch": run_command_text(["git", "rev-parse", "--abbrev-ref", "HEAD"]), "commit": run_command_text(["git", "rev-parse", "HEAD"]), "operating_system": platform.platform(), "cpu": platform.processor() or os.environ.get("PROCESSOR_IDENTIFIER"), "ram_bytes": total_ram_bytes(), "python_version": sys.version.replace("\n", " "), "ollama_version": run_command_text(["ollama", "--version"]), "model": model, "endpoint": endpoint, } def run_command_text(command: list[str]) -> str | None: try: completed = subprocess.run( command, check=False, capture_output=True, text=True, encoding="utf-8", errors="replace", ) except OSError: return None text = (completed.stdout or completed.stderr).strip() return text or None def total_ram_bytes() -> int | None: try: import psutil # type: ignore[import-not-found] except ModuleNotFoundError: return None return int(psutil.virtual_memory().total) def run_stage(name: str, timings: dict[str, Any], func: Callable[[], Any]) -> Any: print(f"[{datetime.now().isoformat(timespec='seconds')}] Starting {name}", flush=True) start = time.perf_counter() try: result = func() except Exception: elapsed = time.perf_counter() - start timings[name] = {"runtime_seconds": round(elapsed, 3), "status": "failed"} print(f"[{datetime.now().isoformat(timespec='seconds')}] FAILED {name} after {elapsed:.3f}s", flush=True) raise elapsed = time.perf_counter() - start timings[name] = {"runtime_seconds": round(elapsed, 3), "status": "passed"} print(f"[{datetime.now().isoformat(timespec='seconds')}] Finished {name} in {elapsed:.3f}s", flush=True) return result def write_normalized_chunk(input_path: Path, output_path: Path, changes_path: Path) -> dict[str, Any]: source = input_path.read_text(encoding="utf-8-sig") blocks = split_blocks(source) normalized_blocks = [] changes = [] for block_number, block in enumerate(blocks, start=1): normalized, change = normalize_block(block, block_number) normalized_blocks.append(normalized) if change is not None: changes.append(change) output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text("\n\n".join(normalized_blocks).strip() + "\n", encoding="utf-8") report = { "source_file": input_path.name, "output_file": output_path.name, "blocks_total": len(blocks), "blocks_changed": len(changes), "changes": [asdict(change) for change in changes], } changes_path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return report def extraction_output_path(chunk_path: Path, output_dir: Path) -> Path: name = chunk_path.name.replace("_normalized.txt", "_extraction.json") return output_dir / name def run_extraction_stage( normalized_dir: Path, output_dir: Path, model: str, endpoint: str, timeout: int, temperature: float, num_predict: int | None, num_ctx: int, meeting_context: MeetingContext, ) -> list[Path]: output_dir.mkdir(parents=True, exist_ok=True) output_paths = [] for chunk_path in find_normalized_chunks(normalized_dir): output_path = extraction_output_path(chunk_path, output_dir) raw_path = output_path.with_suffix(".raw.txt") metadata_path = output_path.with_name(output_path.stem + "_metadata.json") print(f"Extracting {chunk_path.name} -> {output_path.name}", flush=True) transcript = chunk_path.read_text(encoding="utf-8-sig").strip() if not transcript: raise RunnerError(f"Normalized chunk is empty: {chunk_path}") prompt = build_extraction_prompt(chunk_path.name, transcript, meeting_context=meeting_context) raw_text, metadata = call_extraction_ollama( endpoint=endpoint, model=model, prompt=prompt, timeout=timeout, temperature=temperature, num_predict=num_predict, num_ctx=num_ctx, ) raw_path.write_text(raw_text + "\n", encoding="utf-8") metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") parsed = parse_json_response(raw_text) extraction = normalize_current_schema(parsed) extraction["context"] = meeting_context.provenance() output_path.write_text(json.dumps(extraction, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") output_paths.append(output_path) return output_paths def source_coverage_report_from_groups( groups: Any, expected_source_ids: set[str], ) -> dict[str, Any]: seen = [] violations = [] if not isinstance(groups, list): return { "valid": False, "expected_source_fact_count": len(expected_source_ids), "observed_source_fact_occurrence_count": 0, "group_count": None, "violations": [{"type": "groups_not_list"}], } for group_index, group in enumerate(groups): source_ids = group.get("source_item_ids") if isinstance(group, dict) else None if not isinstance(source_ids, list) or not source_ids: violations.append({"type": "invalid_group_source_ids", "group_index": group_index}) continue string_ids = [item_id for item_id in source_ids if isinstance(item_id, str)] seen.extend(string_ids) for unknown in sorted(set(string_ids) - expected_source_ids): violations.append({"type": "unknown_id", "id": unknown, "group_index": group_index}) counts = Counter(seen) for item_id, count in sorted(counts.items()): if count > 1: violations.append({"type": "duplicate_id", "id": item_id, "count": count}) for missing in sorted(expected_source_ids - set(seen)): violations.append({"type": "missing_id", "id": missing}) return { "valid": not violations, "expected_source_fact_count": len(expected_source_ids), "observed_source_fact_occurrence_count": len(seen), "group_count": len(groups), "violations": violations, } def run_semantic_consolidator_stage( canonicalized_path: Path, output_dir: Path, model: str, endpoint: str, timeout: int, num_ctx: int, num_predict: int | None, progress_interval: int, ) -> dict[str, Any]: output_dir.mkdir(parents=True, exist_ok=True) canonicalized = read_json_object(canonicalized_path) facts = fact_items(canonicalized) prompt = build_consolidation_prompt(facts) prompt_token_estimate = (len(prompt) + 3) // 4 resolved_num_predict = resolve_num_predict( requested_num_predict=num_predict, facts=facts, prompt_token_estimate=prompt_token_estimate, num_ctx=num_ctx, ) raw_text, response_data, runtime = call_consolidator_ollama( endpoint=endpoint, model=model, prompt=prompt, timeout=timeout, num_ctx=num_ctx, num_predict=resolved_num_predict, think=False, progress_interval=progress_interval, ) raw_response_path = output_dir / "raw_model_response.txt" raw_json_path = output_dir / "raw_ollama_response.json" output_path = output_dir / "consolidated_extractions.json" report_path = output_dir / "report.md" repair_metadata_path = output_dir / "repair_metadata.json" raw_response_path.write_text(raw_text + "\n", encoding="utf-8") write_json(raw_json_path, response_data) model_output = parse_model_json(raw_text) expected_fact_ids = {item["item_id"] for item in facts} validator_before = source_coverage_report_from_groups( model_output.get("groups"), expected_fact_ids, ) write_json(output_dir / "validator_before.json", validator_before) model_output, repair_changes = repair_model_group_coverage(model_output, facts) if repair_changes: write_json( repair_metadata_path, { "scope": "semantic_consolidator_v0_source_coverage", "llm_used": False, "repair_count": len(repair_changes), "repairs": repair_changes, }, ) else: write_json( repair_metadata_path, { "scope": "semantic_consolidator_v0_source_coverage", "llm_used": False, "repair_count": 0, "repairs": [], }, ) write_json(output_dir / "repaired_model_groups.json", model_output) groups = validate_model_groups(model_output, expected_fact_ids) validate_group_shapes(groups) output = build_consolidated_output(canonicalized, groups) validate_consolidated_output(canonicalized, output) write_json(output_path, output) validator_after = source_coverage_report_from_groups( [item for item in output["items"] if item.get("category") == "fact"], expected_fact_ids, ) # Output fact items use source_item_ids and should be covered by the same checker. write_json(output_dir / "validator_after.json", validator_after) write_consolidator_report( path=report_path, model=model, runtime=runtime, prompt_chars=len(prompt), prompt_token_estimate=prompt_token_estimate, num_predict=resolved_num_predict, fact_count=len(facts), groups=groups, output_path=output_path, repair_changes=repair_changes, ) return { "runtime_seconds": runtime, "fact_count": len(facts), "num_predict": resolved_num_predict, "validator_before": validator_before, "validator_after": validator_after, "repair_count": len(repair_changes), "response_metadata": response_data, } def category_counts_from_extractions(extractions_dir: Path) -> dict[str, int]: categories = ("facts", "decisions", "todos", "questions", "positions", "technical") counts = {category: 0 for category in categories} for path in sorted(extractions_dir.glob("chunk_*_extraction.json")): data = read_json_object(path) for category in categories: values = data.get(category, []) if isinstance(values, list): counts[category] += len(values) return counts def chunk_distribution(chunks_dir: Path) -> dict[str, Any]: manifest = read_json_object(chunks_dir / "manifest.json") chunks = manifest.get("chunks", []) sizes = [chunk["chars"] for chunk in chunks if isinstance(chunk, dict) and isinstance(chunk.get("chars"), int)] return { "chunk_count": manifest.get("chunk_count"), "min_chars": min(sizes) if sizes else None, "max_chars": max(sizes) if sizes else None, "avg_chars": round(sum(sizes) / len(sizes), 1) if sizes else None, "chunks": sizes, } def write_benchmark_report(path: Path, metadata: dict[str, Any]) -> None: renderer = metadata.get("renderer", {}) semantic = metadata.get("semantic_consolidator", {}) timings = metadata.get("runtime_seconds_by_stage", {}) ollama = metadata.get("preflight", {}).get("ollama", {}) lines = [ "# Meeting Lab Benchmark Report", "", f"- Status: `{metadata.get('status')}`", f"- Input: `{metadata.get('input')}`", f"- Meeting Context: `{metadata.get('context')}`", f"- Model: `{metadata.get('model')}`", f"- Output directory: `{metadata.get('output_dir')}`", "", "## Configuration", "", f"- Chunking: target={metadata['configuration']['target_chars']}, max={metadata['configuration']['max_chars']}, min={metadata['configuration']['min_chars']}, overlap={metadata['configuration']['overlap_blocks']}", f"- LLM: think=false, temperature={metadata['configuration']['temperature']}, num_ctx={metadata['configuration']['num_ctx']}", f"- Allow loaded Ollama models: {metadata['configuration'].get('allow_loaded_models')}", "", "## Ollama Cleanliness", "", f"- Detected loaded models: `{ollama.get('active_models')}`", f"- Loaded-model override: `{ollama.get('allow_loaded_models', metadata['configuration'].get('allow_loaded_models'))}`", "", "## Runtime", "", "| Stage | Seconds | Status |", "|---|---:|---|", ] for stage, data in timings.items(): lines.append(f"| {stage} | {data.get('runtime_seconds')} | {data.get('status')} |") lines.extend( [ "", f"- Total wall-clock runtime: {metadata.get('total_runtime_seconds')} seconds", f"- Average extraction time per chunk: {metadata.get('average_extraction_seconds_per_chunk')} seconds", "", "## Counts", "", f"- Chunk distribution: `{metadata.get('chunk_distribution')}`", f"- Extraction counts: `{metadata.get('extraction_counts')}`", f"- Canonicalizer stats: `{metadata.get('canonicalizer_stats')}`", "", "## Validation", "", f"- Validator before: valid={semantic.get('validator_before', {}).get('valid')}, violations={len(semantic.get('validator_before', {}).get('violations', []))}", f"- Validator after: valid={semantic.get('validator_after', {}).get('valid')}, violations={len(semantic.get('validator_after', {}).get('violations', []))}", f"- Deterministic repair count: {semantic.get('repair_count')}", f"- Renderer contract valid: {renderer.get('valid')}", f"- Renderer violations: {len(renderer.get('violations', [])) if isinstance(renderer.get('violations'), list) else 'n/a'}", ] ) if metadata.get("failure"): lines.extend(["", "## Failure", "", str(metadata["failure"])]) path.write_text("\n".join(lines).rstrip() + "\n", encoding="utf-8") def preflight(args: argparse.Namespace, output_dir: Path) -> tuple[dict[str, Any], MeetingContext]: input_data = validate_whisper_input(args.input) if not args.context.is_file(): raise RunnerError(f"Meeting Context file does not exist: {args.context}") meeting_context = load_meeting_context(args.context) ollama = require_ollama( args.endpoint, args.model, allow_loaded_models=args.allow_loaded_models, ) if args.target_chars <= 0 or args.max_chars <= 0 or args.min_chars < 0: raise RunnerError("Chunk size parameters must be positive.") if args.min_chars > args.target_chars: raise RunnerError("--min-chars must not exceed --target-chars.") if args.target_chars > args.max_chars: raise RunnerError("--target-chars must not exceed --max-chars.") if args.overlap_blocks < 0: raise RunnerError("--overlap-blocks must not be negative.") probe = output_dir / ".write_test" probe.write_text("ok", encoding="utf-8") probe.unlink() return { "input_segments": len(input_data["segments"]), "input_segment_text_chars": sum(len(segment.get("text", "")) for segment in input_data["segments"]), "input_text_chars": len(str(input_data.get("text", ""))), "ollama": ollama, }, meeting_context def run_pipeline(args: argparse.Namespace) -> tuple[int, Path, Path, Path | None]: output_dir = create_unique_output_dir(args.output_root, args.benchmark_label) report_path = output_dir / "benchmark_report.md" metadata_path = output_dir / "run_metadata.json" protocol_path: Path | None = None timings: dict[str, Any] = {} start = time.perf_counter() metadata: dict[str, Any] = { "status": "running", "input": str(args.input), "context": str(args.context), "model": args.model, "output_dir": str(output_dir), "configuration": { "target_chars": args.target_chars, "max_chars": args.max_chars, "min_chars": args.min_chars, "overlap_blocks": args.overlap_blocks, "temperature": args.temperature, "num_ctx": args.num_ctx, "num_predict": args.num_predict, "consolidator_num_predict": args.consolidator_num_predict, "renderer_num_predict": args.renderer_num_predict, "think": False, "semantic_retries": 0, "manual_intervention": False, "allow_loaded_models": args.allow_loaded_models, }, "system": system_metadata(args.model, args.endpoint), "runtime_seconds_by_stage": timings, } try: preflight_data, meeting_context = run_stage( "preflight", timings, lambda: preflight(args, output_dir), ) metadata["preflight"] = preflight_data chunks_dir = output_dir / "chunks" run_stage( "chunking", timings, lambda: write_chunks( build_chunks( read_transcript_blocks(args.input), target_chars=args.target_chars, max_chars=args.max_chars, min_chars=args.min_chars, overlap_blocks=args.overlap_blocks, ), chunks_dir, args.input.name, ), ) metadata["chunk_distribution"] = chunk_distribution(chunks_dir) normalized_dir = output_dir / "normalized_chunks" def normalize_all() -> None: normalized_dir.mkdir(parents=True, exist_ok=True) for chunk_path in sorted(chunks_dir.glob("chunk_*.txt")): base = chunk_path.stem write_normalized_chunk( chunk_path, normalized_dir / f"{base}_normalized.txt", normalized_dir / f"{base}_changes.json", ) shutil.copyfile(chunks_dir / "manifest.json", normalized_dir / "manifest.json") run_stage("normalization", timings, normalize_all) extractions_dir = output_dir / "extractions" extraction_paths = run_stage( "extraction", timings, lambda: run_extraction_stage( normalized_dir=normalized_dir, output_dir=extractions_dir, model=args.model, endpoint=args.endpoint, timeout=args.timeout, temperature=args.temperature, num_predict=args.num_predict, num_ctx=args.num_ctx, meeting_context=meeting_context, ), ) metadata["extraction_files"] = len(extraction_paths) metadata["extraction_counts"] = category_counts_from_extractions(extractions_dir) canonicalized_path = output_dir / "canonicalizer" / "canonicalized_extractions.json" canonicalized = run_stage( "canonicalizer", timings, lambda: canonicalize_extractions( extractions_dir, merge_duplicates=True, meeting_context_path=args.context, ), ) write_canonicalized(canonicalized, canonicalized_path) metadata["canonicalizer_stats"] = canonicalized.get("stats") semantic_metadata = run_stage( "semantic_consolidator", timings, lambda: run_semantic_consolidator_stage( canonicalized_path=canonicalized_path, output_dir=output_dir / "semantic_consolidator", model=args.model, endpoint=args.endpoint, timeout=args.timeout, num_ctx=args.num_ctx, num_predict=args.consolidator_num_predict, progress_interval=args.progress_interval, ), ) metadata["semantic_consolidator"] = semantic_metadata renderer_metadata = run_stage( "renderer", timings, lambda: render_working_protocol( input_path=output_dir / "semantic_consolidator" / "consolidated_extractions.json", output_dir=output_dir / "working_protocol", model=args.model, endpoint=args.endpoint, timeout=args.timeout, num_ctx=args.num_ctx, num_predict=args.renderer_num_predict, think=False, ), ) metadata["renderer"] = { "valid": renderer_metadata.get("valid"), "runtime_seconds": renderer_metadata.get("runtime_seconds"), "violations": read_json_object(output_dir / "working_protocol" / "validation_report.json").get("violations", []), "metadata": renderer_metadata, } if renderer_metadata.get("output_path"): protocol_path = Path(str(renderer_metadata["output_path"])) metadata["status"] = "completed" # Renderer contract invalid is a completed benchmark with failed final protocol. exit_code = 0 except Exception as exc: # noqa: BLE001 - preserve artifacts and report. metadata["status"] = "failed" metadata["failure"] = f"{type(exc).__name__}: {exc}" if isinstance(exc, RunnerError) and exc.details: metadata.update(exc.details) exit_code = 2 print(f"Error: {metadata['failure']}", file=sys.stderr) total_runtime = time.perf_counter() - start metadata["total_runtime_seconds"] = round(total_runtime, 3) chunk_count = metadata.get("chunk_distribution", {}).get("chunk_count") extraction_runtime = timings.get("extraction", {}).get("runtime_seconds") metadata["average_extraction_seconds_per_chunk"] = ( round(extraction_runtime / chunk_count, 3) if isinstance(extraction_runtime, (int, float)) and isinstance(chunk_count, int) and chunk_count else None ) metadata_path.write_text(json.dumps(metadata, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") write_benchmark_report(report_path, metadata) return exit_code, output_dir, report_path, protocol_path def main(argv: list[str] | None = None) -> int: args = parse_args(argv) try: exit_code, output_dir, report_path, protocol_path = run_pipeline(args) except RunnerError as exc: print(f"Error: {exc}", file=sys.stderr) return 2 print(f"Output directory: {output_dir}") print(f"Benchmark report: {report_path}") if protocol_path is not None and protocol_path.exists(): print(f"Protocol: {protocol_path}") else: print("Protocol: not written") return exit_code if __name__ == "__main__": raise SystemExit(main())