diff --git a/.gitignore b/.gitignore index 56facb4..796fb5d 100644 --- a/.gitignore +++ b/.gitignore @@ -23,6 +23,7 @@ dist/ .pytest_cache/ .test-tmp/ .test-tmp-root/ +tmp_run_meeting_tests/ .coverage htmlcov/ diff --git a/README.md b/README.md index aed136b..8d9966e 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,33 @@ PYTHONPATH=src .venv/bin/python -m meeting_lab.consolidation.consolidate_facts \ Die eigentliche Ausgabeerzeugung ist bewusst der letzte Verarbeitungsschritt. +Ein kompletter repository-nativer Referenzlauf fuer das Progeo-Benchmark kann +auf dem Linux AI-PC ohne Codex mit einem Befehl gestartet werden: + +```text +python scripts/run_meeting.py \ + --input samples/real_live/progeo_meeting/progeo_whispercpp_vulkan_turbo_trimmed_converted.json \ + --context samples/real_live/progeo_meeting/meeting_context.yaml \ + --model qwen3.5:9b \ + --benchmark-label progeo_ai_pc +``` + +Die Linux-Einrichtung ist: + +```text +python3 -m venv .venv +source .venv/bin/activate +python -m pip install --upgrade pip +python -m pip install -e . +ollama pull qwen3.5:9b +ollama list +ollama ps +``` + +`ollama ps` muss fuer den Referenzlauf leer sein. Details zu Voraussetzungen, +Artefakten, Windows-Beispiel und Vergleich mit dem AI-PC stehen in +`docs/run-meeting.md`. + --- ## Lizenz diff --git a/docs/run-meeting.md b/docs/run-meeting.md new file mode 100644 index 0000000..b907ada --- /dev/null +++ b/docs/run-meeting.md @@ -0,0 +1,142 @@ +# Running A Meeting Benchmark + +This page documents the repository-native runner for the current Meeting Lab +production benchmark on the Linux AI-PC. It does not run Whisper. It starts +from a converted, normalized Whisper JSON file. + +## Prerequisites + +- Git +- Python 3.11 or newer +- Ollama running locally +- Required Ollama model installed +- Meeting Lab dependencies installed in a virtual environment + +## Setup + +From the repository root on Linux: + +```bash +python3 -m venv .venv +source .venv/bin/activate +python -m pip install --upgrade pip +python -m pip install -e . +``` + +Install the reference model in Ollama if it is not already present: + +```bash +ollama pull qwen3.5:9b +``` + +Confirm Ollama sees the model: + +```bash +ollama list +ollama ps +``` + +`ollama ps` must show no loaded model before a reference benchmark run. By +default, the runner fails pre-flight if Ollama reports a loaded model. Use +`--allow-loaded-models` only for a deliberately non-clean run; the default +reference command does not use that override. + +## Progeo Reference Command + +Run from the repository root: + +```bash +python scripts/run_meeting.py \ + --input samples/real_live/progeo_meeting/progeo_whispercpp_vulkan_turbo_trimmed_converted.json \ + --context samples/real_live/progeo_meeting/meeting_context.yaml \ + --model qwen3.5:9b \ + --benchmark-label progeo_ai_pc +``` + +## Windows / PowerShell Example + +For local Windows development, activate the virtual environment through +PowerShell and run the same benchmark: + +```powershell +python -m venv .venv +.\.venv\Scripts\Activate.ps1 +python -m pip install --upgrade pip +python -m pip install -e . +ollama pull qwen3.5:9b +python scripts\run_meeting.py ` + --input samples\real_live\progeo_meeting\progeo_whispercpp_vulkan_turbo_trimmed_converted.json ` + --context samples\real_live\progeo_meeting\meeting_context.yaml ` + --model qwen3.5:9b ` + --benchmark-label progeo_ai_pc +``` + +## Production Defaults + +The Progeo reference benchmark defaults match the latest controlled experiment +selection: + +- Chunking: `target_chars=4500`, `max_chars=5500`, `min_chars=2500`, + `overlap_blocks=0` +- LLM: `qwen3.5:9b`, `think=false`, `temperature=0`, `num_ctx=32768` +- Extraction: context-aware chunk extraction with current committed prompts +- Consolidation: Canonicalizer V1, Semantic Consolidator V0 adaptive sizing, + strict source-coverage validation and deterministic source-coverage repair +- Rendering: Working Protocol Renderer V2 +- No semantic retries and no manual intervention + +Use `--help` to see explicit overrides. + +## Output + +The runner creates a unique directory: + +```text +samples/benchmarks/_/ +``` + +It never overwrites an existing run. Artifacts include: + +- `chunks/` +- `normalized_chunks/` +- `extractions/`, including parsed extraction JSON and raw successful model + responses +- `canonicalizer/canonicalized_extractions.json` +- `semantic_consolidator/`, including raw response, repaired groups, + validator reports, repair metadata and consolidated output +- `working_protocol/`, including raw renderer response and validation report +- `working_protocol.md` only when the renderer contract is valid +- `run_metadata.json` +- `benchmark_report.md` + +If a stage fails, artifacts produced up to the failure are preserved and the +report records the failure. + +## Comparing Benchmarks + +For AI-PC comparison, compare the new `progeo_ai_pc_` directory +against: + +```text +samples/benchmarks/progeo_qwen35_9b_20260805_133942/ +``` + +Key fields are in `run_metadata.json` and summarized in `benchmark_report.md`: + +- branch and commit +- operating system, CPU, RAM, Python version and Ollama version +- model tag, observed Ollama active models and whether + `--allow-loaded-models` was used +- chunk count and chunk-size distribution +- runtime per stage and total wall-clock runtime +- average extraction time per chunk +- validator violations and deterministic repair count +- renderer contract result + +## Known Limitation + +The current Working Protocol Renderer V2 often produces contract-invalid output +that does not start with `# Working Protocol`. The runner preserves the raw +renderer response and validation report, and writes `working_protocol.md` only +when the contract validator passes. This is the known BUG-011 renderer +limitation, not a runner failure by itself. diff --git a/scripts/run_meeting.py b/scripts/run_meeting.py new file mode 100644 index 0000000..d33d70a --- /dev/null +++ b/scripts/run_meeting.py @@ -0,0 +1,808 @@ +#!/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()) diff --git a/tests/test_run_meeting.py b/tests/test_run_meeting.py new file mode 100644 index 0000000..86e8c0e --- /dev/null +++ b/tests/test_run_meeting.py @@ -0,0 +1,489 @@ +import json +import unittest +import uuid +from datetime import datetime +from pathlib import Path +from unittest.mock import Mock, patch + +from scripts import run_meeting + + +TEST_TMP = Path("tmp_run_meeting_tests") + + +VALID_CONTEXT = """schema_version: "1" + +meeting: + meeting_id: "test-meeting" + title: "Test Meeting" + language: "de" + +participants: + - participant_id: "martin" + display_name: "Martin" + aliases: [] + attendance_status: "present" + +mentioned_people: [] +organization: + departments: [] +known_entities: {} +""" + + +def write_whisper_json(path: Path) -> None: + path.write_text( + json.dumps( + { + "text": "Martin: Hallo.\nJames: Weiter.", + "segments": [ + {"text": "Martin: Hallo."}, + {"text": "James: Weiter."}, + ], + } + ), + encoding="utf-8", + ) + + +def write_context(path: Path, text: str = VALID_CONTEXT) -> None: + path.write_text(text, encoding="utf-8") + + +def response_json(payload): + response = Mock() + response.raise_for_status.return_value = None + response.json.return_value = payload + return response + + +def workspace_tempdir(): + TEST_TMP.mkdir(exist_ok=True) + return ManualTempDirectory(TEST_TMP / f"run-meeting-{uuid.uuid4().hex}") + + +class ManualTempDirectory: + def __init__(self, path: Path) -> None: + self.path = path + + def __enter__(self) -> str: + self.path.mkdir(parents=True) + return str(self.path) + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + +class RunMeetingTests(unittest.TestCase): + def test_cli_argument_parsing_defaults(self) -> None: + args = run_meeting.parse_args( + [ + "--input", + "input.json", + "--context", + "context.yaml", + ] + ) + + self.assertEqual(args.model, "qwen3.5:9b") + self.assertEqual(args.target_chars, 4500) + self.assertEqual(args.max_chars, 5500) + self.assertEqual(args.min_chars, 2500) + self.assertEqual(args.overlap_blocks, 0) + self.assertEqual(args.temperature, 0.0) + self.assertEqual(args.num_ctx, 32768) + self.assertFalse(args.allow_loaded_models) + + def test_explicit_parameter_overrides(self) -> None: + args = run_meeting.parse_args( + [ + "--input", + "input.json", + "--context", + "context.yaml", + "--model", + "other:model", + "--target-chars", + "3000", + "--max-chars", + "4000", + "--min-chars", + "1000", + "--overlap-blocks", + "1", + "--num-ctx", + "8192", + "--temperature", + "0.2", + ] + ) + + self.assertEqual(args.model, "other:model") + self.assertEqual(args.target_chars, 3000) + self.assertEqual(args.max_chars, 4000) + self.assertEqual(args.min_chars, 1000) + self.assertEqual(args.overlap_blocks, 1) + self.assertEqual(args.num_ctx, 8192) + self.assertEqual(args.temperature, 0.2) + + def test_unique_output_directory_creation(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + fixed = datetime(2026, 8, 5, 12, 0, 0) + first = run_meeting.create_unique_output_dir(root, "progeo test", now=lambda: fixed) + second = run_meeting.create_unique_output_dir(root, "progeo test", now=lambda: fixed) + + self.assertEqual(first.name, "progeo_test_20260805_120000") + self.assertEqual(second.name, "progeo_test_20260805_120000_01") + + def test_invalid_input_path_fails_early(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + context = root / "context.yaml" + write_context(context) + args = run_meeting.parse_args( + [ + "--input", + str(root / "missing.json"), + "--context", + str(context), + "--output-root", + str(root / "benchmarks"), + ] + ) + code, output_dir, _report, _protocol = run_meeting.run_pipeline(args) + + self.assertEqual(code, 2) + self.assertTrue((output_dir / "run_metadata.json").exists()) + + def test_invalid_meeting_context_fails(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + input_path = root / "input.json" + context = root / "context.yaml" + write_whisper_json(input_path) + write_context(context, "schema_version: '['\n") + args = run_meeting.parse_args( + [ + "--input", + str(input_path), + "--context", + str(context), + "--output-root", + str(root / "benchmarks"), + ] + ) + code, _output_dir, _report, _protocol = run_meeting.run_pipeline(args) + + self.assertEqual(code, 2) + + def test_unavailable_ollama_endpoint_fails(self) -> None: + with patch("scripts.run_meeting.requests.get", side_effect=run_meeting.requests.ConnectionError("down")): + with self.assertRaisesRegex(run_meeting.RunnerError, "not reachable"): + run_meeting.require_ollama("http://127.0.0.1:11434/api/generate", "qwen3.5:9b") + + def test_missing_model_fails(self) -> None: + response = response_json({"models": [{"name": "other:model"}]}) + with patch("scripts.run_meeting.requests.get", return_value=response): + with self.assertRaisesRegex(run_meeting.RunnerError, "not installed"): + run_meeting.require_ollama("http://127.0.0.1:11434/api/generate", "qwen3.5:9b") + + def test_loaded_model_fails_preflight_by_default(self) -> None: + responses = [ + response_json({"models": [{"name": "qwen3.5:9b"}]}), + response_json({"models": [{"name": "other:model"}]}), + ] + with patch("scripts.run_meeting.requests.get", side_effect=responses): + with self.assertRaisesRegex(run_meeting.RunnerError, "loaded model"): + run_meeting.require_ollama("http://127.0.0.1:11434/api/generate", "qwen3.5:9b") + + def test_loaded_model_override_permits_execution(self) -> None: + responses = [ + response_json({"models": [{"name": "qwen3.5:9b"}]}), + response_json({"models": [{"name": "other:model"}]}), + ] + with patch("scripts.run_meeting.requests.get", side_effect=responses): + data = run_meeting.require_ollama( + "http://127.0.0.1:11434/api/generate", + "qwen3.5:9b", + allow_loaded_models=True, + ) + + self.assertEqual(data["active_models"], ["other:model"]) + self.assertTrue(data["allow_loaded_models"]) + + def test_loaded_model_preflight_failure_records_metadata(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + input_path = root / "input.json" + context = root / "context.yaml" + write_whisper_json(input_path) + write_context(context) + args = run_meeting.parse_args( + [ + "--input", + str(input_path), + "--context", + str(context), + "--output-root", + str(root / "benchmarks"), + ] + ) + error = run_meeting.RunnerError( + "Ollama has loaded model(s).", + details={ + "preflight": { + "ollama": { + "active_models": ["other:model"], + "allow_loaded_models": False, + } + } + }, + ) + + with patch("scripts.run_meeting.require_ollama", side_effect=error): + code, output_dir, report_path, _protocol = run_meeting.run_pipeline(args) + + metadata = json.loads((output_dir / "run_metadata.json").read_text(encoding="utf-8")) + report = report_path.read_text(encoding="utf-8") + self.assertEqual(code, 2) + self.assertEqual(metadata["preflight"]["ollama"]["active_models"], ["other:model"]) + self.assertFalse(metadata["preflight"]["ollama"]["allow_loaded_models"]) + self.assertIn("Detected loaded models: `['other:model']`", report) + self.assertIn("Loaded-model override: `False`", report) + + def test_stage_failure_preserves_artifacts(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + input_path = root / "input.json" + context = root / "context.yaml" + write_whisper_json(input_path) + write_context(context) + args = run_meeting.parse_args( + [ + "--input", + str(input_path), + "--context", + str(context), + "--output-root", + str(root / "benchmarks"), + ] + ) + + def fail_extraction(**kwargs): + marker = kwargs["output_dir"] / "partial.raw.txt" + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_text("partial", encoding="utf-8") + raise ValueError("model failed") + + with patch("scripts.run_meeting.require_ollama", return_value={"active_models": []}), patch( + "scripts.run_meeting.run_extraction_stage", + side_effect=fail_extraction, + ): + code, output_dir, report_path, _protocol = run_meeting.run_pipeline(args) + + self.assertEqual(code, 2) + self.assertTrue((output_dir / "extractions" / "partial.raw.txt").exists()) + self.assertTrue(report_path.exists()) + self.assertIn("model failed", (output_dir / "run_metadata.json").read_text(encoding="utf-8")) + + def test_benchmark_report_generation(self) -> None: + with workspace_tempdir() as directory: + path = Path(directory) / "report.md" + run_meeting.write_benchmark_report( + path, + { + "status": "completed", + "input": "input.json", + "context": "context.yaml", + "model": "qwen3.5:9b", + "output_dir": "out", + "configuration": { + "target_chars": 4500, + "max_chars": 5500, + "min_chars": 2500, + "overlap_blocks": 0, + "temperature": 0.0, + "num_ctx": 32768, + }, + "runtime_seconds_by_stage": {"chunking": {"runtime_seconds": 1.0, "status": "passed"}}, + "preflight": { + "ollama": {"active_models": ["other:model"], "allow_loaded_models": True} + }, + "total_runtime_seconds": 1.0, + "average_extraction_seconds_per_chunk": None, + "chunk_distribution": {}, + "extraction_counts": {}, + "canonicalizer_stats": {}, + "semantic_consolidator": { + "validator_before": {"valid": True, "violations": []}, + "validator_after": {"valid": True, "violations": []}, + "repair_count": 0, + }, + "renderer": {"valid": False, "violations": [{"type": "x"}]}, + }, + ) + + text = path.read_text(encoding="utf-8") + + self.assertIn("# Meeting Lab Benchmark Report", text) + self.assertIn("Renderer contract valid: False", text) + self.assertIn("Detected loaded models: `['other:model']`", text) + self.assertIn("Loaded-model override: `True`", text) + + def test_metadata_records_loaded_model_condition_and_override(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + input_path = root / "input.json" + context = root / "context.yaml" + write_whisper_json(input_path) + write_context(context) + args = run_meeting.parse_args( + [ + "--input", + str(input_path), + "--context", + str(context), + "--output-root", + str(root / "benchmarks"), + "--allow-loaded-models", + ] + ) + + def fake_extraction(**kwargs): + output = kwargs["output_dir"] / "chunk_01_extraction.json" + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text( + json.dumps( + { + "facts": ["Martin | Test fact | clear | evidence"], + "decisions": [], + "todos": [], + "questions": [], + "positions": [], + "technical": [], + "context": {"source_file": str(context)}, + } + ), + encoding="utf-8", + ) + return [output] + + def fake_semantic(**_kwargs): + return { + "validator_before": {"valid": True, "violations": []}, + "validator_after": {"valid": True, "violations": []}, + "repair_count": 0, + "response_metadata": {}, + } + + def fake_renderer(**kwargs): + output_dir = kwargs["output_dir"] + output_dir.mkdir(parents=True, exist_ok=True) + (output_dir / "validation_report.json").write_text( + json.dumps({"valid": False, "violations": []}), + encoding="utf-8", + ) + return {"valid": False, "runtime_seconds": 0.1, "output_path": None} + + with patch( + "scripts.run_meeting.require_ollama", + return_value={"active_models": ["other:model"], "allow_loaded_models": True}, + ), patch( + "scripts.run_meeting.run_extraction_stage", + side_effect=fake_extraction, + ), patch( + "scripts.run_meeting.run_semantic_consolidator_stage", + side_effect=fake_semantic, + ), patch( + "scripts.run_meeting.render_working_protocol", + side_effect=fake_renderer, + ): + code, output_dir, report_path, _protocol_path = run_meeting.run_pipeline(args) + + metadata = json.loads((output_dir / "run_metadata.json").read_text(encoding="utf-8")) + report = report_path.read_text(encoding="utf-8") + self.assertEqual(code, 0) + self.assertTrue(metadata["configuration"]["allow_loaded_models"]) + self.assertEqual(metadata["preflight"]["ollama"]["active_models"], ["other:model"]) + self.assertTrue(metadata["preflight"]["ollama"]["allow_loaded_models"]) + self.assertIn("Detected loaded models: `['other:model']`", report) + self.assertIn("Loaded-model override: `True`", report) + + def test_successful_run_uses_mocked_llm_stages(self) -> None: + with workspace_tempdir() as directory: + root = Path(directory) + input_path = root / "input.json" + context = root / "context.yaml" + write_whisper_json(input_path) + write_context(context) + args = run_meeting.parse_args( + [ + "--input", + str(input_path), + "--context", + str(context), + "--output-root", + str(root / "benchmarks"), + ] + ) + + def fake_extraction(**kwargs): + output = kwargs["output_dir"] / "chunk_01_extraction.json" + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text( + json.dumps( + { + "facts": ["Martin | Test fact | clear | evidence"], + "decisions": [], + "todos": [], + "questions": [], + "positions": [], + "technical": [], + "context": {"source_file": str(context)}, + } + ), + encoding="utf-8", + ) + output.with_suffix(".raw.txt").write_text("{}", encoding="utf-8") + return [output] + + def fake_semantic(**_kwargs): + out = root / "placeholder" + return { + "validator_before": {"valid": True, "violations": []}, + "validator_after": {"valid": True, "violations": []}, + "repair_count": 0, + "response_metadata": {}, + "output": str(out), + } + + def fake_renderer(**kwargs): + output_dir = kwargs["output_dir"] + output_dir.mkdir(parents=True, exist_ok=True) + (output_dir / "validation_report.json").write_text( + json.dumps({"valid": False, "violations": [{"type": "missing_required_heading"}]}), + encoding="utf-8", + ) + return {"valid": False, "runtime_seconds": 0.1, "output_path": None} + + with patch("scripts.run_meeting.require_ollama", return_value={"active_models": []}), patch( + "scripts.run_meeting.run_extraction_stage", + side_effect=fake_extraction, + ), patch( + "scripts.run_meeting.run_semantic_consolidator_stage", + side_effect=fake_semantic, + ), patch( + "scripts.run_meeting.render_working_protocol", + side_effect=fake_renderer, + ): + code, output_dir, report_path, protocol_path = run_meeting.run_pipeline(args) + + self.assertEqual(code, 0) + self.assertTrue((output_dir / "canonicalizer" / "canonicalized_extractions.json").exists()) + self.assertTrue(report_path.exists()) + self.assertIsNone(protocol_path) + + +if __name__ == "__main__": + unittest.main()