feat: add post-diarization speaker mapping workflow

PROJECT_KNOWLEDGE.md

@@ -25,7 +25,28 @@ Implemented:
 - Prompt loading from `src/meeting_lab/llm/prompts.py`.
 - Meeting Context V1 loading, validation and optional extraction prompt
   injection with minimal extraction JSON provenance.
+- FFmpeg-backed WAV, FLAC and M4A preparation into a per-run canonical mono
+  16 kHz signed PCM16 WAV artifact before transcription or diarization. Audio
+  preparation always runs. Optional loudness normalization defaults to on and
+  currently uses the isolated FFmpeg filter
+  `loudnorm=I=-16:LRA=11:TP=-1.5`. This is a conservative speech-recording
+  default and may be revisited after empirical comparison without changing the
+  orchestration API.
 - Interim Markdown protocol generation in `src/meeting_lab/protocol/`.
+- Direct protocol prompt input protection: diarized transcripts are rendered as
+  compact adjacent-speaker blocks without per-segment timestamps. Every source
+  segment remains represented in order. A deterministic heuristic enforces a
+  configurable safe input budget, falls back to complete plain transcript text
+  when necessary, and fails before any Ollama request if even that input is too
+  large. Silent head/tail truncation is prohibited.
+- The `qwen3.8:27b` direct-protocol stage explicitly requests `num_ctx=32768`
+  and `think=false`; the practical prompt target is approximately 29,000 tokens.
+  A 31,038-token synthetic prompt passed, but larger prompts are not assumed safe
+  from the model's advertised 262,144-token native context alone.
+- `regenerate_mvp_protocol` updates the run's validated Meeting Context and
+  regenerates protocol artifacts from the existing diarized transcript when
+  available. It never reruns audio preparation, Whisper or Pyannote, and it
+  preserves anonymous speaker labels in the source transcript.
 - Non-LLM unit tests for chunking, extraction helpers, protocol rendering and
   gold-test runner validation.
 - Meeting Context V1 scaffold and documentation for manually maintained
@@ -139,6 +160,9 @@ departments only when they are explicitly supplied as metadata. It must not be
 used to infer responsibilities. In the current implementation this context can
 be injected into chunk extraction prompts as authoritative metadata, and only
 minimal provenance is written to extraction JSON.
+The implemented MVP statuses are exactly `present` and `mentioned_only`.
+Legacy entries without a status receive collection-appropriate defaults. Only
+present participants may be targets of explicit `SPEAKER_XX` mappings.
 
 A `responsible` or future `owner` / `assignee` value may be recorded only when
 source evidence explicitly assigns, accepts or confirms responsibility. If the

docs/diarization.md

@@ -0,0 +1,62 @@
+# Optional Speaker Diarization
+
+The direct-protocol MVP keeps speaker diarization disabled by default. Enable
+anonymous Community-1 speaker labels with `--diarization auto`, `gpu`, or
+`cpu`:
+
+```bash
+python3 scripts/run_mvp_meeting.py meeting.wav \
+  --whisper-model /path/to/ggml-model.bin \
+  --diarization auto
+```
+
+Native mode (the default runtime) requires a compatible local PyTorch and
+`pyannote.audio==4.0.7`. For isolated ROCm/CUDA environments, select the
+container runtime and provide its image and hardware arguments explicitly:
+
+```bash
+python3 scripts/run_mvp_meeting.py meeting.wav \
+  --whisper-model /path/to/ggml-model.bin \
+  --diarization gpu \
+  --diarization-runtime container \
+  --diarization-container-image IMAGE \
+  --diarization-container-arg=--device=/dev/kfd \
+  --diarization-container-arg=--device=/dev/dri \
+  --diarization-container-arg=--group-add \
+  --diarization-container-arg=video
+```
+
+The container receives `HF_TOKEN` by environment-variable name only. It mounts
+the source audio and repository read-only and writes diarization artifacts into
+the current run directory. Meeting Lab loads mono 16 kHz PCM16 WAV with
+Python's `wave` module and sends an in-memory tensor to pyannote, avoiding its
+torchcodec file decoder.
+
+Anonymous `SPEAKER_XX` labels are aligned to Whisper segments by maximum
+temporal overlap with Community-1 exclusive diarization. The original Whisper
+transcript is preserved; the derived transcript under `diarization/` is used as
+the direct-protocol generator's source input.
+
+The full diarized JSON and timestamped text remain immutable audit artifacts,
+but their per-segment formatting is too verbose for a full-meeting LLM prompt:
+timestamps and repeated speaker labels can more than double input size. For
+protocol generation, Meeting Lab deterministically groups only adjacent
+segments assigned to the same anonymous speaker and omits timestamps. A later
+return by the same speaker starts a new block, and unassigned segments remain
+under `SPEAKER_UNASSIGNED`. `protocol/transcript_input.txt` preserves the exact
+derived representation sent to prompt construction.
+
+Before contacting Ollama, Meeting Lab conservatively estimates prompt tokens
+from UTF-8 byte count without adding a model tokenizer dependency. The default safe
+budget is 29,000 estimated tokens within the explicitly configured 32,768-token
+Ollama context. The estimate is calibrated against the currently validated
+German BPD input and is configurable through
+`MvpMeetingConfig.protocol_safe_input_token_budget` or
+`--protocol-safe-input-token-budget`.
+
+If compact diarized input exceeds the budget, the generator deterministically
+uses the complete plain segment transcript and records the fallback. If that
+also exceeds the budget, generation fails before model lookup or generation;
+it never truncates, chunks, summarizes, retries, or makes multiple protocol
+calls implicitly. Full diarization artifacts are never overwritten by this
+selection.

docs/meeting-context.md

@@ -107,8 +107,10 @@ or aliases are corrected.
 
 `department`: Organizational unit. Optional and nullable.
 
-`attendance_status`: `present` for participants. This distinguishes attendees
-from mentioned people.
+`attendance_status`: exactly `present` for participants or `mentioned_only` for
+people who are relevant but did not attend. For backward compatibility, a
+missing status defaults to `present` in `participants` and `mentioned_only` in
+`mentioned_people`.
 
 `mentioned_people`: People discussed or referenced but not present. They are
 not participants and must not be treated as speakers.
@@ -153,7 +155,7 @@ mentioned_people:
     aliases: []
     role: null
     department: null
-    attendance_status: "not_present"
+    attendance_status: "mentioned_only"
     notes: "Wurde erwaehnt, war aber nicht anwesend."
 
 organization:
@@ -282,6 +284,8 @@ The current validator checks that:
 - participant ids and mentioned-person ids do not collide
 - referenced departments exist in `organization.departments`
 - `attendance_status` values are valid
+- speaker mappings reference present participants only; mentioned-only people
+  cannot be diarized speakers
 - participants are marked `present`
 - mentioned people are not marked `present`
 

samples/real_live/progeo_meeting/meeting_context.yaml

@@ -75,7 +75,7 @@ mentioned_people:
     aliases: []
     role: null
     department: null
-    attendance_status: "not_present"
+    attendance_status: "mentioned_only"
     notes: null
 
 organization:

samples/real_live/project_process_meeting/meeting_context.yaml

@@ -64,7 +64,7 @@ mentioned_people:
       - "Giovana"
     role: "Leiterin Business Development"
     department_id: "bd"
-    attendance_status: "not_present"
+    attendance_status: "mentioned_only"
     notes: null
 
 organization:

samples/templates/meeting_context.template.yaml

@@ -4,6 +4,8 @@
 schema_version: "1"
 
 meeting:
+  # Stable identifier used for provenance across corrections and later runs.
+  meeting_id: ""
   # Human-readable title for the meeting.
   title: ""
   # Dominant meeting language, for example "de" or "en".
@@ -28,6 +30,12 @@ participants:
     attendance_status: "present"
     notes: null
 
+# Optional authoritative mapping from diarization labels to actual participants.
+# Add entries only after a human or trusted external process confirms identity.
+# Never infer mappings from conversational context. Unmapped labels stay anonymous.
+speaker_mappings: {}
+#  SPEAKER_00: "participant-id"
+
 mentioned_people:
   # People discussed or referenced but not present in the meeting.
   # Mentioned people are not speakers and must not become responsible persons
@@ -37,7 +45,7 @@ mentioned_people:
     aliases: []
     role: null
     department: null
-    attendance_status: "not_present"
+    attendance_status: "mentioned_only"
     notes: null
 
 organization:

scripts/run_direct_protocol.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python3
+"""Run the one-call direct protocol MVP from compact Whisper JSON."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import shutil
+import sys
+import time
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable
+
+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.llm.ollama import DEFAULT_ENDPOINT  # noqa: E402
+from src.meeting_lab.protocol.generate_direct_protocol import (  # noqa: E402
+    DEFAULT_MODEL,
+    DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+    DirectProtocolResult,
+    generate_direct_protocol,
+)
+
+
+DEFAULT_OUTPUT_ROOT = Path("meeting_data/runs")
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Generate one direct protocol from compact Whisper JSON.")
+    parser.add_argument("transcript", type=Path)
+    parser.add_argument("--context", type=Path)
+    parser.add_argument("--output-root", type=Path, default=DEFAULT_OUTPUT_ROOT)
+    parser.add_argument("--model", default=DEFAULT_MODEL)
+    parser.add_argument("--ollama-endpoint", default=DEFAULT_ENDPOINT)
+    parser.add_argument(
+        "--safe-input-token-budget",
+        type=int,
+        default=DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+    )
+    return parser.parse_args(argv)
+
+
+def create_unique_run_dir(
+    output_root: Path,
+    transcript_stem: str,
+    now: Callable[[], datetime] = datetime.now,
+) -> Path:
+    safe_stem = re.sub(r"[^A-Za-z0-9_.-]+", "_", transcript_stem).strip("._-") or "meeting"
+    base = output_root / f"{safe_stem}_{now().strftime('%Y%m%d_%H%M%S')}"
+    candidate = base
+    suffix = 1
+    while candidate.exists():
+        candidate = output_root / f"{base.name}_{suffix:02d}"
+        suffix += 1
+    candidate.mkdir(parents=True)
+    return candidate
+
+
+def write_json(path: Path, data: Any) -> None:
+    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def persist_result(run_dir: Path, result: DirectProtocolResult) -> Path:
+    protocol_dir = run_dir / "protocol"
+    protocol_dir.mkdir()
+    (protocol_dir / "exact_prompt.txt").write_text(result.exact_prompt, encoding="utf-8")
+    write_json(protocol_dir / "raw_response.json", result.raw_response)
+    write_json(protocol_dir / "runtime_metadata.json", result.runtime_metadata)
+    transcript_input = getattr(result, "transcript_input", None)
+    if transcript_input is not None:
+        (protocol_dir / "transcript_input.txt").write_text(
+            transcript_input, encoding="utf-8"
+        )
+    protocol_path = run_dir / "protocol.md"
+    protocol_path.write_text(result.protocol_text, encoding="utf-8")
+    return protocol_path
+
+
+def run(args: argparse.Namespace) -> tuple[int, Path, Path | None]:
+    run_dir = create_unique_run_dir(args.output_root, args.transcript.stem)
+    timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
+    started = time.perf_counter()
+    protocol_path: Path | None = None
+    metadata: dict[str, Any] = {
+        "run_id": run_dir.name,
+        "timestamp": timestamp,
+        "transcript_path": str(args.transcript.resolve()),
+        "context_path": str(args.context.resolve()) if args.context else None,
+        "model": args.model,
+        "ollama_endpoint": args.ollama_endpoint,
+        "status": "running",
+        "total_runtime_seconds": None,
+        "final_protocol_path": None,
+    }
+    try:
+        transcript_dir = run_dir / "transcript"
+        transcript_dir.mkdir()
+        if not args.transcript.is_file():
+            raise FileNotFoundError(f"Transcript file does not exist: {args.transcript}")
+        preserved_transcript = transcript_dir / "transcript.json"
+        shutil.copy2(args.transcript, preserved_transcript)
+
+        preserved_context: Path | None = None
+        if args.context is not None:
+            if not args.context.is_file():
+                raise FileNotFoundError(f"Meeting Context file does not exist: {args.context}")
+            context_dir = run_dir / "context"
+            context_dir.mkdir()
+            preserved_context = context_dir / "meeting_context.yaml"
+            shutil.copy2(args.context, preserved_context)
+
+        write_json(
+            run_dir / "input_manifest.json",
+            {
+                "transcript_source": str(args.transcript.resolve()),
+                "transcript_copy": str(preserved_transcript.resolve()),
+                "context_source": str(args.context.resolve()) if args.context else None,
+                "context_copy": str(preserved_context.resolve()) if preserved_context else None,
+            },
+        )
+        result = generate_direct_protocol(
+            preserved_transcript,
+            preserved_context,
+            model=args.model,
+            endpoint=args.ollama_endpoint,
+            safe_input_token_budget=args.safe_input_token_budget,
+        )
+        protocol_path = persist_result(run_dir, result)
+        metadata["status"] = "completed"
+        metadata["final_protocol_path"] = str(protocol_path.resolve())
+    except Exception as exc:
+        metadata["status"] = "failed"
+        metadata["failure"] = f"{type(exc).__name__}: {exc}"
+        print(f"Error: {metadata['failure']}", file=sys.stderr)
+    finally:
+        metadata["total_runtime_seconds"] = round(time.perf_counter() - started, 3)
+        write_json(run_dir / "run_metadata.json", metadata)
+    return (0 if metadata["status"] == "completed" else 2), run_dir, protocol_path
+
+
+def main(argv: list[str] | None = None) -> int:
+    code, _run_dir, protocol_path = run(parse_args(argv))
+    if protocol_path is not None:
+        print(protocol_path)
+    return code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_mvp_meeting.py

@@ -0,0 +1,151 @@
+#!/usr/bin/env python3
+"""CLI adapter for the reusable Meeting Lab MVP orchestration API."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+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.llm.ollama import DEFAULT_ENDPOINT  # noqa: E402
+from src.meeting_lab.models.meeting_context import MeetingContext  # noqa: E402
+from src.meeting_lab.orchestration.mvp import (  # noqa: E402
+    DEFAULT_DIARIZATION_MODEL,
+    DEFAULT_MODEL,
+    DEFAULT_OUTPUT_ROOT,
+    DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+    MvpMeetingConfig,
+    create_unique_run_dir,
+    run_mvp_meeting,
+)
+from src.meeting_lab.progress import ProgressSink  # noqa: E402
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Transcribe one meeting and generate one direct protocol."
+    )
+    parser.add_argument("audio_file", type=Path)
+    parser.add_argument("--whisper-model", type=Path, required=True)
+    parser.add_argument("--whisper-executable", default="whisper-cli")
+    parser.add_argument("--ffmpeg-executable", default="ffmpeg")
+    parser.add_argument(
+        "--audio-normalization",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help=(
+            "Enable FFmpeg loudness normalization during canonical audio preparation "
+            "(default: enabled)."
+        ),
+    )
+    parser.add_argument("--context", type=Path)
+    parser.add_argument("--output-root", type=Path, default=DEFAULT_OUTPUT_ROOT)
+    parser.add_argument("--language", default="de")
+    parser.add_argument(
+        "--threads",
+        default="auto",
+        help="Thread count or 'auto' for physical CPU cores (default: auto).",
+    )
+    parser.add_argument("--model", default=DEFAULT_MODEL)
+    parser.add_argument("--ollama-endpoint", default=DEFAULT_ENDPOINT)
+    parser.add_argument(
+        "--protocol-safe-input-token-budget",
+        type=int,
+        default=DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+        help="Conservative estimated prompt-token limit before any Ollama request.",
+    )
+    parser.add_argument(
+        "--diarization",
+        choices=("auto", "gpu", "cpu", "off"),
+        default="off",
+        help="Optional Community-1 diarization device mode (default: off).",
+    )
+    parser.add_argument(
+        "--diarization-runtime",
+        choices=("native", "container"),
+        default="native",
+        help="Run pyannote in this Python environment or an explicit container.",
+    )
+    parser.add_argument(
+        "--diarization-container-image",
+        help="Container image required with --diarization-runtime container.",
+    )
+    parser.add_argument(
+        "--diarization-container-arg",
+        action="append",
+        default=[],
+        help="Additional docker argument; repeat and use = for values beginning with --.",
+    )
+    return parser.parse_args(argv)
+
+
+def config_from_args(args: argparse.Namespace) -> MvpMeetingConfig:
+    return MvpMeetingConfig(
+        audio_file=args.audio_file,
+        whisper_model=args.whisper_model,
+        whisper_executable=args.whisper_executable,
+        ffmpeg_executable=args.ffmpeg_executable,
+        audio_normalization=args.audio_normalization,
+        context_file=args.context,
+        output_root=args.output_root,
+        language=args.language,
+        threads=args.threads,
+        model=args.model,
+        ollama_endpoint=args.ollama_endpoint,
+        protocol_safe_input_token_budget=args.protocol_safe_input_token_budget,
+        diarization=args.diarization,
+        diarization_runtime=args.diarization_runtime,
+        diarization_container_image=args.diarization_container_image,
+        diarization_container_args=tuple(args.diarization_container_arg),
+    )
+
+
+def run(
+    args: argparse.Namespace,
+    *,
+    context_override: MeetingContext | dict[str, Any] | None = None,
+    progress_sink: ProgressSink | None = None,
+) -> tuple[int, Path | None, Path | None]:
+    """Compatibility wrapper for existing Python callers of the CLI module."""
+    result = run_mvp_meeting(
+        config_from_args(args),
+        meeting_context=context_override,
+        progress_sink=progress_sink,
+    )
+    return result.exit_code, result.run_dir, result.protocol_path
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    if args.diarization == "off":
+        print("Diarization: disabled")
+    else:
+        print(
+            f"Diarization: enabled; backend=pyannote.audio; "
+            f"model={DEFAULT_DIARIZATION_MODEL}; requested_device={args.diarization}; "
+            f"runtime={args.diarization_runtime}"
+        )
+    code, run_dir, protocol_path = run(args)
+    if run_dir is not None and args.diarization != "off":
+        metadata_path = run_dir / "diarization" / "metadata.json"
+        if metadata_path.is_file():
+            details = json.loads(metadata_path.read_text(encoding="utf-8"))
+            print(
+                f"Diarization result: device={details.get('actual_device')}; "
+                f"device_name={details.get('device_name') or 'n/a'}; "
+                f"runtime={details.get('runtime_seconds'):.3f}s; "
+                f"speakers={details.get('speaker_count')}; artifacts={metadata_path.parent}"
+            )
+    if protocol_path is not None:
+        print(protocol_path)
+    return code
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/transcribe_meeting.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+"""Transcribe one audio file with whisper.cpp; do not generate a protocol."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+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.transcription.whisper import (  # noqa: E402
+    TranscriptionError,
+    transcribe_audio,
+)
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Create a compact Meeting Lab transcript with whisper.cpp.")
+    parser.add_argument("audio_file", type=Path)
+    parser.add_argument("--model", type=Path, required=True, help="Path to a whisper.cpp GGML model.")
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--language", default="auto", help="Language code or 'auto' (default: auto).")
+    parser.add_argument("--threads", default="auto", help="Thread count or 'auto' for physical CPU cores (default: auto).")
+    parser.add_argument("--whisper-executable", default="whisper-cli", help="whisper.cpp CLI executable (default: whisper-cli).")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(argv)
+    try:
+        result = transcribe_audio(
+            args.audio_file,
+            args.model,
+            args.output_dir,
+            args.language,
+            executable=args.whisper_executable,
+            threads=args.threads,
+        )
+    except TranscriptionError as exc:
+        print(f"Error: {exc}")
+        return 1
+    print(f"Transcript: {result.transcript_json}")
+    print(f"Runtime: {result.runtime_seconds:.3f} seconds")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/meeting_lab/audio/__init__.py

@@ -0,0 +1,5 @@
+"""Canonical audio preparation boundary."""
+
+from .preparation import AudioPreparationError, PreparedAudio, prepare_audio
+
+__all__ = ["AudioPreparationError", "PreparedAudio", "prepare_audio"]

src/meeting_lab/audio/preparation.py

@@ -0,0 +1,190 @@
+"""Prepare supported recordings for deterministic downstream processing."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import wave
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+
+SUPPORTED_EXTENSIONS = {".wav", ".flac", ".m4a"}
+CANONICAL_SAMPLE_RATE = 16_000
+CANONICAL_CHANNELS = 1
+CANONICAL_SAMPLE_WIDTH_BYTES = 2
+CANONICAL_CODEC = "pcm_s16le"
+DEFAULT_NORMALIZATION_FILTER = "loudnorm=I=-16:LRA=11:TP=-1.5"
+DEFAULT_NORMALIZATION_METHOD = "ffmpeg_loudnorm"
+
+
+class AudioPreparationError(RuntimeError):
+    """Raised when source audio cannot be prepared as canonical WAV."""
+
+
+@dataclass(frozen=True)
+class PreparedAudio:
+    source_path: Path
+    source_format: str
+    prepared_path: Path
+    method: str
+    ffmpeg_executable: str
+    normalization_enabled: bool = True
+    normalization_method: str | None = DEFAULT_NORMALIZATION_METHOD
+    normalization_filter: str | None = DEFAULT_NORMALIZATION_FILTER
+
+    def metadata(self) -> dict[str, object]:
+        return {
+            "original_source_path": str(self.source_path.resolve()),
+            "original_source_name": self.source_path.name,
+            "original_format": self.source_format,
+            "prepared_audio_path": str(self.prepared_path.resolve()),
+            "preparation_method": self.method,
+            "ffmpeg_executable": self.ffmpeg_executable,
+            "normalization_enabled": self.normalization_enabled,
+            "normalization_method": self.normalization_method,
+            "normalization_filter": self.normalization_filter,
+            "canonical_output": {
+                "container": "wav",
+                "codec": CANONICAL_CODEC,
+                "channels": CANONICAL_CHANNELS,
+                "sample_rate_hz": CANONICAL_SAMPLE_RATE,
+                "bits_per_sample": CANONICAL_SAMPLE_WIDTH_BYTES * 8,
+            },
+        }
+
+
+Runner = Callable[..., subprocess.CompletedProcess[str]]
+
+
+def prepare_audio(
+    source_path: Path,
+    prepared_path: Path,
+    *,
+    ffmpeg_executable: str = "ffmpeg",
+    normalization_enabled: bool = True,
+    runner: Runner = subprocess.run,
+) -> PreparedAudio:
+    """Create and validate a canonical mono 16 kHz signed PCM16 WAV artifact."""
+    source_path = Path(source_path)
+    prepared_path = Path(prepared_path)
+    source_format = source_path.suffix.lower()
+    if not source_path.is_file():
+        raise AudioPreparationError(f"Source audio does not exist: {source_path}")
+    if source_format not in SUPPORTED_EXTENSIONS:
+        supported = ", ".join(sorted(SUPPORTED_EXTENSIONS))
+        raise AudioPreparationError(
+            f"Unsupported audio format {source_format or '<none>'!r}; supported: {supported}."
+        )
+
+    resolved_executable = _resolve_executable(ffmpeg_executable)
+    prepared_path.parent.mkdir(parents=True, exist_ok=True)
+    temporary_path = prepared_path.with_name(f".{prepared_path.name}.tmp.wav")
+    command_parts = [
+        resolved_executable,
+        "-nostdin",
+        "-hide_banner",
+        "-loglevel",
+        "error",
+        "-y",
+        "-i",
+        str(source_path),
+        "-map_metadata",
+        "-1",
+        "-vn",
+    ]
+    if normalization_enabled:
+        command_parts.extend(("-af", DEFAULT_NORMALIZATION_FILTER))
+    command_parts.extend(
+        (
+            "-ac",
+            str(CANONICAL_CHANNELS),
+            "-ar",
+            str(CANONICAL_SAMPLE_RATE),
+            "-c:a",
+            CANONICAL_CODEC,
+            "-fflags",
+            "+bitexact",
+            str(temporary_path),
+        )
+    )
+    command: Sequence[str] = tuple(command_parts)
+    try:
+        completed = runner(command, capture_output=True, text=True, check=False)
+    except OSError as exc:
+        raise AudioPreparationError(f"Could not run FFmpeg: {exc}") from exc
+    if completed.returncode != 0:
+        detail = (
+            completed.stderr or completed.stdout or "no diagnostic output"
+        ).strip()
+        raise AudioPreparationError(
+            f"FFmpeg failed to prepare {source_path.name} (exit {completed.returncode}): "
+            f"{detail}"
+        )
+    try:
+        _validate_canonical_wav(temporary_path)
+        os.replace(temporary_path, prepared_path)
+    except Exception:
+        temporary_path.unlink(missing_ok=True)
+        raise
+
+    return PreparedAudio(
+        source_path=source_path,
+        source_format=source_format.removeprefix("."),
+        prepared_path=prepared_path,
+        method="ffmpeg",
+        ffmpeg_executable=resolved_executable,
+        normalization_enabled=normalization_enabled,
+        normalization_method=(
+            DEFAULT_NORMALIZATION_METHOD if normalization_enabled else None
+        ),
+        normalization_filter=(
+            DEFAULT_NORMALIZATION_FILTER if normalization_enabled else None
+        ),
+    )
+
+
+def _resolve_executable(executable: str) -> str:
+    value = executable.strip()
+    if not value:
+        raise AudioPreparationError("FFmpeg executable must not be empty.")
+    if Path(value).parent != Path("."):
+        path = Path(value)
+        if path.is_file() and os.access(path, os.X_OK):
+            return str(path)
+        raise AudioPreparationError(f"FFmpeg executable is not available: {value}")
+    resolved = shutil.which(value)
+    if resolved is None:
+        raise AudioPreparationError(
+            f"FFmpeg executable {value!r} was not found on PATH. Install FFmpeg or "
+            "configure its executable path."
+        )
+    return resolved
+
+
+def _validate_canonical_wav(path: Path) -> None:
+    try:
+        with wave.open(str(path), "rb") as recording:
+            properties = (
+                recording.getnchannels(),
+                recording.getframerate(),
+                recording.getsampwidth(),
+                recording.getcomptype(),
+            )
+    except (OSError, EOFError, wave.Error) as exc:
+        raise AudioPreparationError(
+            f"FFmpeg did not produce a readable WAV file: {path}: {exc}"
+        ) from exc
+    expected = (
+        CANONICAL_CHANNELS,
+        CANONICAL_SAMPLE_RATE,
+        CANONICAL_SAMPLE_WIDTH_BYTES,
+        "NONE",
+    )
+    if properties != expected:
+        raise AudioPreparationError(
+            "Prepared audio is not canonical mono 16 kHz PCM16 WAV: "
+            f"channels={properties[0]}, sample_rate={properties[1]}, "
+            f"sample_width={properties[2]}, compression={properties[3]}."
+        )

src/meeting_lab/diarization/__init__.py

@@ -0,0 +1,20 @@
+"""Optional speaker diarization and transcript alignment."""
+
+from src.meeting_lab.diarization.alignment import align_transcript, write_diarized_transcript
+from src.meeting_lab.diarization.backend import (
+    DEFAULT_MODEL,
+    DiarizationError,
+    DiarizationResult,
+    diarize_audio,
+    select_device,
+)
+
+__all__ = [
+    "DEFAULT_MODEL",
+    "DiarizationError",
+    "DiarizationResult",
+    "align_transcript",
+    "diarize_audio",
+    "select_device",
+    "write_diarized_transcript",
+]

src/meeting_lab/diarization/alignment.py

@@ -0,0 +1,129 @@
+"""Deterministic Whisper-segment alignment to anonymous diarization turns."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+class AlignmentError(ValueError):
+    """Raised when transcript or diarization inputs are malformed."""
+
+
+def _number(value: Any, description: str) -> float:
+    if not isinstance(value, (int, float)) or isinstance(value, bool):
+        raise AlignmentError(f"{description} must be a number.")
+    return float(value)
+
+
+def _validated_turns(turns: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    validated = []
+    for index, turn in enumerate(turns):
+        if not isinstance(turn, dict):
+            raise AlignmentError(f"Diarization turn {index} must be an object.")
+        start = _number(turn.get("start"), f"Diarization turn {index} start")
+        end = _number(turn.get("end"), f"Diarization turn {index} end")
+        speaker = turn.get("speaker_id", turn.get("speaker"))
+        if end < start:
+            raise AlignmentError(f"Diarization turn {index} ends before it starts.")
+        if not isinstance(speaker, str) or not speaker.startswith("SPEAKER_"):
+            raise AlignmentError(
+                f"Diarization turn {index} must have an anonymous SPEAKER_ label."
+            )
+        validated.append({"start": start, "end": end, "speaker_id": speaker})
+    return validated
+
+
+def align_transcript(
+    transcript: dict[str, Any], exclusive_turns: list[dict[str, Any]]
+) -> dict[str, Any]:
+    """Return a derived transcript using maximum exclusive-turn overlap per segment."""
+    if not isinstance(transcript, dict) or not isinstance(transcript.get("segments"), list):
+        raise AlignmentError("Whisper transcript must contain a 'segments' list.")
+    turns = _validated_turns(exclusive_turns)
+    aligned_segments: list[dict[str, Any]] = []
+
+    for index, source in enumerate(transcript["segments"]):
+        if not isinstance(source, dict):
+            raise AlignmentError(f"Transcript segment {index} must be an object.")
+        start = _number(source.get("start"), f"Transcript segment {index} start")
+        end = _number(source.get("end"), f"Transcript segment {index} end")
+        if end < start:
+            raise AlignmentError(f"Transcript segment {index} ends before it starts.")
+        overlap_by_speaker: dict[str, float] = {}
+        for turn in turns:
+            overlap = max(0.0, min(end, turn["end"]) - max(start, turn["start"]))
+            if overlap:
+                speaker = turn["speaker_id"]
+                overlap_by_speaker[speaker] = overlap_by_speaker.get(speaker, 0.0) + overlap
+        speaker_id = None
+        overlap_seconds = 0.0
+        if overlap_by_speaker:
+            speaker_id, overlap_seconds = min(
+                overlap_by_speaker.items(), key=lambda item: (-item[1], item[0])
+            )
+        duration = end - start
+        aligned = dict(source)
+        aligned.update(
+            {
+                "speaker_id": speaker_id,
+                "speaker_overlap_seconds": round(overlap_seconds, 6),
+                "speaker_overlap_ratio": round(
+                    overlap_seconds / duration if duration > 0 else 0.0, 6
+                ),
+            }
+        )
+        aligned_segments.append(aligned)
+
+    return {
+        "text": diarized_transcript_text(aligned_segments, include_end=True),
+        "segments": aligned_segments,
+        "speaker_labels_anonymous": True,
+        "alignment_source": "exclusive_diarization",
+    }
+
+
+def _timestamp(seconds: float) -> str:
+    milliseconds = int(round(seconds * 1000))
+    hours, remainder = divmod(milliseconds, 3_600_000)
+    minutes, remainder = divmod(remainder, 60_000)
+    secs, millis = divmod(remainder, 1000)
+    return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}"
+
+
+def diarized_transcript_text(
+    segments: list[dict[str, Any]], *, include_end: bool = True
+) -> str:
+    lines = []
+    for segment in segments:
+        start = _timestamp(float(segment["start"]))
+        end = _timestamp(float(segment["end"]))
+        speaker = segment.get("speaker_id") or "SPEAKER_UNASSIGNED"
+        timestamp = f"[{start} - {end}]" if include_end else f"[{start}]"
+        lines.append(f"{timestamp} {speaker}: {str(segment.get('text', '')).strip()}")
+    return "\n".join(lines) + ("\n" if lines else "")
+
+
+def write_diarized_transcript(
+    transcript_path: Path,
+    exclusive_turns_path: Path,
+    output_dir: Path,
+) -> tuple[Path, Path]:
+    """Read source artifacts and write a separate speaker-aware transcript pair."""
+    transcript = json.loads(Path(transcript_path).read_text(encoding="utf-8-sig"))
+    turns = json.loads(Path(exclusive_turns_path).read_text(encoding="utf-8"))
+    if not isinstance(turns, list):
+        raise AlignmentError("Exclusive diarization turns must contain a JSON list.")
+    derived = align_transcript(transcript, turns)
+    output_dir = Path(output_dir)
+    output_dir.mkdir(parents=True, exist_ok=True)
+    json_path = output_dir / "transcript_diarized.json"
+    text_path = output_dir / "transcript_diarized.txt"
+    json_path.write_text(
+        json.dumps(derived, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
+    text_path.write_text(
+        diarized_transcript_text(derived["segments"], include_end=True), encoding="utf-8"
+    )
+    return json_path, text_path

src/meeting_lab/diarization/backend.py

@@ -0,0 +1,314 @@
+"""pyannote Community-1 backend with native and isolated-container runtimes."""
+
+from __future__ import annotations
+
+import importlib.metadata
+import json
+import os
+import subprocess
+import time
+import wave
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Literal, Sequence
+
+
+DEFAULT_MODEL = "pyannote/speaker-diarization-community-1"
+PYANNOTE_VERSION = "4.0.7"
+DeviceMode = Literal["auto", "gpu", "cpu"]
+RuntimeMode = Literal["native", "container"]
+
+
+class DiarizationError(RuntimeError):
+    """Raised when diarization configuration or execution fails."""
+
+
+@dataclass(frozen=True)
+class DiarizationResult:
+    output_dir: Path
+    metadata_path: Path
+    ordinary_rttm: Path
+    exclusive_rttm: Path
+    turns_json: Path
+    exclusive_turns_json: Path
+    metadata: dict[str, Any]
+
+
+def _host_uid() -> int:
+    getter = getattr(os, "getuid", None)
+    if getter is None:
+        raise DiarizationError("Container diarization requires host UID discovery.")
+    return int(getter())
+
+
+def _host_gid() -> int:
+    getter = getattr(os, "getgid", None)
+    if getter is None:
+        raise DiarizationError("Container diarization requires host GID discovery.")
+    return int(getter())
+
+
+def select_device(mode: DeviceMode, torch_module: Any) -> tuple[Any, str | None]:
+    """Resolve CPU/GPU without depending on the GPU vendor."""
+    if mode == "cpu":
+        return torch_module.device("cpu"), None
+    if mode not in ("auto", "gpu"):
+        raise DiarizationError(f"Unsupported diarization device mode: {mode}")
+    try:
+        available = bool(torch_module.cuda.is_available())
+        if available:
+            name = str(torch_module.cuda.get_device_name(0))
+            probe = torch_module.zeros(1, device="cuda")
+            del probe
+            return torch_module.device("cuda"), name
+    except Exception as exc:
+        if mode == "gpu":
+            raise DiarizationError(f"Requested PyTorch GPU is not usable: {exc}") from exc
+    if mode == "gpu":
+        raise DiarizationError("Requested PyTorch GPU is unavailable.")
+    return torch_module.device("cpu"), None
+
+
+def _load_pcm_wave(audio_path: Path, torch_module: Any) -> tuple[Any, int, float, dict[str, Any]]:
+    try:
+        with wave.open(str(audio_path), "rb") as source:
+            channels = source.getnchannels()
+            sample_rate = source.getframerate()
+            sample_width = source.getsampwidth()
+            frame_count = source.getnframes()
+            pcm = bytearray(source.readframes(frame_count))
+    except (OSError, wave.Error) as exc:
+        raise DiarizationError(f"Cannot read PCM WAV input {audio_path}: {exc}") from exc
+    if channels != 1 or sample_rate != 16000 or sample_width != 2:
+        raise DiarizationError(
+            "Diarization currently requires mono 16 kHz signed 16-bit PCM WAV; "
+            f"got channels={channels}, sample_rate={sample_rate}, sample_width={sample_width}."
+        )
+    waveform = torch_module.frombuffer(pcm, dtype=torch_module.int16).to(
+        torch_module.float32
+    )
+    waveform = (waveform / 32768.0).reshape(channels, frame_count)
+    duration = frame_count / sample_rate
+    validation = {
+        "waveform_dtype": str(waveform.dtype),
+        "waveform_shape": list(waveform.shape),
+        "sample_rate": sample_rate,
+        "sample_count": frame_count,
+        "duration_seconds": duration,
+        "min_sample_value": waveform.min().item(),
+        "max_sample_value": waveform.max().item(),
+        "audio_loading": "python_wave_pcm16",
+    }
+    return waveform, sample_rate, duration, validation
+
+
+def _turns(annotation: Any) -> list[dict[str, Any]]:
+    return [
+        {
+            "start": segment.start,
+            "end": segment.end,
+            "speaker_id": speaker,
+        }
+        for segment, _track, speaker in annotation.itertracks(yield_label=True)
+    ]
+
+
+def _write_json(path: Path, value: Any) -> None:
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def _result_from_output(output_dir: Path) -> DiarizationResult:
+    metadata_path = output_dir / "metadata.json"
+    try:
+        metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise DiarizationError(f"Cannot read diarization metadata: {exc}") from exc
+    return DiarizationResult(
+        output_dir=output_dir,
+        metadata_path=metadata_path,
+        ordinary_rttm=output_dir / "diarization.rttm",
+        exclusive_rttm=output_dir / "exclusive_diarization.rttm",
+        turns_json=output_dir / "turns.json",
+        exclusive_turns_json=output_dir / "exclusive_turns.json",
+        metadata=metadata,
+    )
+
+
+def _require_writable_output(output_dir: Path) -> None:
+    unwritable = [
+        path
+        for path in (output_dir, *output_dir.rglob("*"))
+        if not os.access(path, os.W_OK)
+    ]
+    if unwritable:
+        rendered = ", ".join(str(path) for path in unwritable[:3])
+        if len(unwritable) > 3:
+            rendered += f", and {len(unwritable) - 3} more"
+        raise DiarizationError(
+            f"Container diarization artifacts are not writable by the host user: {rendered}"
+        )
+
+
+def run_native_pyannote(
+    audio_path: Path,
+    output_dir: Path,
+    device_mode: DeviceMode,
+    *,
+    model: str = DEFAULT_MODEL,
+) -> DiarizationResult:
+    """Run one local pyannote inference using an in-memory waveform mapping."""
+    try:
+        import torch
+        from pyannote.audio import Pipeline
+    except ImportError as exc:
+        raise DiarizationError(
+            f"Native diarization requires pyannote.audio=={PYANNOTE_VERSION} and PyTorch."
+        ) from exc
+
+    token = os.environ.get("HF_TOKEN")
+    if not token:
+        raise DiarizationError("HF_TOKEN is required for the pyannote model.")
+    output_dir.mkdir(parents=True, exist_ok=True)
+    waveform, sample_rate, duration, audio_metadata = _load_pcm_wave(audio_path, torch)
+    device, device_name = select_device(device_mode, torch)
+
+    try:
+        pipeline = Pipeline.from_pretrained(model, token=token)
+        pipeline.to(device)
+        started = time.perf_counter()
+        output = pipeline(
+            {
+                "waveform": waveform,
+                "sample_rate": sample_rate,
+                "uri": audio_path.stem,
+            }
+        )
+        runtime = time.perf_counter() - started
+    except Exception as exc:
+        raise DiarizationError(f"pyannote diarization failed: {type(exc).__name__}: {exc}") from exc
+
+    ordinary = getattr(output, "speaker_diarization", output)
+    exclusive = getattr(output, "exclusive_speaker_diarization", None)
+    if exclusive is None:
+        raise DiarizationError("Community-1 did not return exclusive diarization.")
+    ordinary_turns = _turns(ordinary)
+    exclusive_turns = _turns(exclusive)
+    with (output_dir / "diarization.rttm").open("w", encoding="utf-8") as handle:
+        ordinary.write_rttm(handle)
+    with (output_dir / "exclusive_diarization.rttm").open(
+        "w", encoding="utf-8"
+    ) as handle:
+        exclusive.write_rttm(handle)
+    _write_json(output_dir / "turns.json", ordinary_turns)
+    _write_json(output_dir / "exclusive_turns.json", exclusive_turns)
+    speakers = sorted({turn["speaker_id"] for turn in ordinary_turns})
+    actual_device = str(device)
+    metadata = {
+        "backend": "pyannote.audio",
+        "model": model,
+        "pyannote_version": importlib.metadata.version("pyannote.audio"),
+        "torch_version": torch.__version__,
+        "hip_version": getattr(torch.version, "hip", None),
+        "cuda_version": getattr(torch.version, "cuda", None),
+        "runtime_adapter": "native",
+        "requested_device_mode": device_mode,
+        "actual_device": actual_device,
+        "device_name": device_name if actual_device == "cuda" else None,
+        "audio_duration_seconds": duration,
+        "runtime_seconds": runtime,
+        "rtf": runtime / duration,
+        "speaker_count": len(speakers),
+        "speaker_labels": speakers,
+        "turn_count": len(ordinary_turns),
+        "exclusive_turn_count": len(exclusive_turns),
+        "audio": audio_metadata,
+        "credentials_persisted": False,
+        "output_files": {
+            "ordinary_rttm": "diarization.rttm",
+            "exclusive_rttm": "exclusive_diarization.rttm",
+            "turns": "turns.json",
+            "exclusive_turns": "exclusive_turns.json",
+        },
+    }
+    _write_json(output_dir / "metadata.json", metadata)
+    return _result_from_output(output_dir)
+
+
+def run_container_pyannote(
+    audio_path: Path,
+    output_dir: Path,
+    device_mode: DeviceMode,
+    *,
+    image: str,
+    container_args: Sequence[str] = (),
+    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+    uid_getter: Callable[[], int] = _host_uid,
+    gid_getter: Callable[[], int] = _host_gid,
+) -> DiarizationResult:
+    """Run the same backend in an explicitly configured disposable container."""
+    if not image.strip():
+        raise DiarizationError("A diarization container image is required.")
+    output_dir.mkdir(parents=True, exist_ok=True)
+    host_uid = uid_getter()
+    host_gid = gid_getter()
+    if host_uid < 0 or host_gid < 0:
+        raise DiarizationError("Host UID and GID must be non-negative integers.")
+    command = [
+        "docker", "run", "--rm", "--ipc=host", "--shm-size=8g", "-e", "HF_TOKEN",
+        *container_args,
+        "-v", f"{Path(audio_path).resolve()}:/input/audio.wav:ro",
+        "-v", f"{output_dir.resolve()}:/output:rw",
+        "-v", f"{Path(__file__).resolve().parents[3]}:/work/meeting-lab:ro",
+        "-w", "/work/meeting-lab",
+        image,
+        "/bin/bash", "-lc",
+        (
+            "inference_status=0; "
+            f"python -m pip install --disable-pip-version-check pyannote.audio=={PYANNOTE_VERSION} "
+            "> /output/pip-install.log 2>&1 && "
+            "python -m src.meeting_lab.diarization.container_entry "
+            f"/input/audio.wav /output --device {device_mode} || inference_status=$?; "
+            f"chown -R {host_uid}:{host_gid} /output || exit $?; "
+            "chmod -R u+rwX /output || exit $?; "
+            'exit "$inference_status"'
+        ),
+    ]
+    try:
+        completed = runner(command, check=False, capture_output=True, text=True)
+    except OSError as exc:
+        raise DiarizationError(f"Could not start diarization container: {exc}") from exc
+    (output_dir / "container_stdout.log").write_text(completed.stdout, encoding="utf-8")
+    (output_dir / "container_stderr.log").write_text(completed.stderr, encoding="utf-8")
+    if completed.returncode != 0:
+        detail = completed.stderr.strip() or completed.stdout.strip() or "no diagnostic output"
+        raise DiarizationError(
+            f"Diarization container failed with exit code {completed.returncode}: {detail}"
+        )
+    _require_writable_output(output_dir)
+    result = _result_from_output(output_dir)
+    metadata = dict(result.metadata)
+    metadata["runtime_adapter"] = "container"
+    _write_json(result.metadata_path, metadata)
+    return _result_from_output(output_dir)
+
+
+def diarize_audio(
+    audio_path: Path,
+    output_dir: Path,
+    device_mode: DeviceMode,
+    *,
+    runtime: RuntimeMode = "native",
+    container_image: str | None = None,
+    container_args: Sequence[str] = (),
+) -> DiarizationResult:
+    if runtime == "native":
+        return run_native_pyannote(audio_path, output_dir, device_mode)
+    if runtime == "container":
+        return run_container_pyannote(
+            audio_path,
+            output_dir,
+            device_mode,
+            image=container_image or "",
+            container_args=container_args,
+        )
+    raise DiarizationError(f"Unsupported diarization runtime: {runtime}")

src/meeting_lab/diarization/container_entry.py

@@ -0,0 +1,22 @@
+"""Internal entry point for the isolated pyannote container adapter."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from src.meeting_lab.diarization.backend import run_native_pyannote
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("audio", type=Path)
+    parser.add_argument("output", type=Path)
+    parser.add_argument("--device", choices=("auto", "gpu", "cpu"), required=True)
+    args = parser.parse_args()
+    run_native_pyannote(args.audio, args.output, args.device)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/meeting_lab/llm/ollama.py

@@ -0,0 +1,95 @@
+"""Minimal Ollama client behavior used by the direct protocol MVP."""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Any
+
+import requests
+
+
+DEFAULT_ENDPOINT = "http://127.0.0.1:11434"
+
+
+class OllamaError(RuntimeError):
+    """Raised when Ollama cannot safely complete the requested operation."""
+
+
+@dataclass(frozen=True)
+class OllamaGeneration:
+    raw_response: dict[str, Any]
+    text: str
+    client_wall_time_seconds: float
+
+
+def ollama_base_url(endpoint: str) -> str:
+    endpoint = endpoint.rstrip("/")
+    return endpoint.rsplit("/api/", 1)[0] if "/api/" in endpoint else endpoint
+
+
+def generate_url(endpoint: str) -> str:
+    return f"{ollama_base_url(endpoint)}/api/generate"
+
+
+def require_model(endpoint: str, model: str, timeout: int = 10) -> dict[str, Any]:
+    base_url = ollama_base_url(endpoint)
+    try:
+        response = requests.get(f"{base_url}/api/tags", timeout=timeout)
+        response.raise_for_status()
+        data = response.json()
+    except (requests.RequestException, ValueError) as exc:
+        raise OllamaError(f"Ollama endpoint is not reachable at {base_url}: {exc}") from exc
+
+    models = data.get("models") if isinstance(data, dict) else None
+    if not isinstance(models, list):
+        raise OllamaError("Ollama /api/tags returned a malformed response.")
+    installed = {
+        item.get("name")
+        for item in models
+        if isinstance(item, dict) and isinstance(item.get("name"), str)
+    }
+    if model not in installed:
+        raise OllamaError(f"Requested model is not installed in Ollama: {model}")
+    return {"base_url": base_url, "model": model, "installed": True}
+
+
+def generate_once(
+    endpoint: str,
+    model: str,
+    prompt: str,
+    *,
+    timeout: int,
+    num_ctx: int,
+    num_predict: int,
+) -> OllamaGeneration:
+    payload = {
+        "model": model,
+        "prompt": prompt,
+        "think": False,
+        "stream": False,
+        "options": {
+            "temperature": 0.0,
+            "num_ctx": num_ctx,
+            "num_predict": num_predict,
+        },
+    }
+    started = time.perf_counter()
+    try:
+        response = requests.post(generate_url(endpoint), json=payload, timeout=timeout)
+        response.raise_for_status()
+        data = response.json()
+    except requests.RequestException as exc:
+        raise OllamaError(f"Ollama generation request failed: {exc}") from exc
+    except ValueError as exc:
+        raise OllamaError("Ollama generation response is not valid JSON.") from exc
+    wall_time = time.perf_counter() - started
+
+    if not isinstance(data, dict):
+        raise OllamaError("Ollama generation response must be a JSON object.")
+    text = data.get("response")
+    if not isinstance(text, str):
+        raise OllamaError("Ollama generation response has no string 'response' field.")
+    if not text.strip():
+        raise OllamaError("Ollama returned an empty protocol.")
+    return OllamaGeneration(data, text, wall_time)

src/meeting_lab/models/meeting_context.py

@@ -3,13 +3,15 @@
 from __future__ import annotations
 
 import ast
+import copy
+import re
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
 
 SUPPORTED_SCHEMA_VERSIONS = {"1"}
-VALID_ATTENDANCE_STATUSES = {"present", "not_present", "absent"}
+VALID_ATTENDANCE_STATUSES = {"present", "mentioned_only"}
 
 
 class MeetingContextValidationError(ValueError):
@@ -29,6 +31,21 @@ class MeetingContext:
     def meeting_id(self) -> str:
         return str(self.data["meeting"]["meeting_id"])
 
+    @property
+    def speaker_mappings(self) -> dict[str, str]:
+        mappings = self.data.get("speaker_mappings")
+        return dict(mappings) if isinstance(mappings, dict) else {}
+
+    def participant_for_speaker(self, speaker_label: str) -> dict[str, Any] | None:
+        """Resolve only an explicit authoritative mapping; never infer identity."""
+        participant_id = self.speaker_mappings.get(speaker_label)
+        if participant_id is None:
+            return None
+        for participant in self.data.get("participants", []):
+            if participant.get("participant_id") == participant_id:
+                return participant
+        return None
+
     def provenance(self) -> dict[str, str]:
         return {
             "meeting_id": self.meeting_id,
@@ -42,8 +59,43 @@ def load_meeting_context(path: Path) -> MeetingContext:
     if not isinstance(loaded, dict):
         raise MeetingContextValidationError("Meeting Context must be a YAML object.")
 
-    validate_meeting_context(loaded)
-    return MeetingContext(data=loaded, source_file=path)
+    normalized = _with_attendance_defaults(loaded)
+    validate_meeting_context(normalized)
+    return MeetingContext(data=normalized, source_file=path)
+
+
+def create_meeting_context(
+    data: dict[str, Any], *, source_file: Path = Path("<generated>")
+) -> MeetingContext:
+    """Validate structured data and return an immutable context boundary."""
+    validated = _with_attendance_defaults(data)
+    validate_meeting_context(validated)
+    return MeetingContext(data=validated, source_file=source_file)
+
+
+def serialize_meeting_context_yaml(context: MeetingContext) -> str:
+    """Serialize validated Meeting Context data deterministically as YAML."""
+    validate_meeting_context(context.data)
+    try:
+        import yaml  # type: ignore[import-not-found]
+    except ModuleNotFoundError as exc:
+        raise MeetingContextValidationError(
+            "PyYAML is required to write Meeting Context YAML."
+        ) from exc
+    return yaml.safe_dump(
+        context.data,
+        allow_unicode=True,
+        sort_keys=False,
+        default_flow_style=False,
+    )
+
+
+def write_meeting_context(context: MeetingContext, path: Path) -> Path:
+    """Persist a validated context without changing its schema or semantics."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(serialize_meeting_context_yaml(context), encoding="utf-8")
+    return path
 
 
 def validate_meeting_context(data: dict[str, Any]) -> None:
@@ -74,10 +126,26 @@ def validate_meeting_context(data: dict[str, Any]) -> None:
             + ", ".join(collisions)
         )
 
+    speaker_mappings = _optional_mapping(
+        data.get("speaker_mappings"), "speaker_mappings"
+    )
+    for speaker_label, participant_id in speaker_mappings.items():
+        if not isinstance(speaker_label, str) or re.fullmatch(
+            r"SPEAKER_\d+", speaker_label
+        ) is None:
+            raise MeetingContextValidationError(
+                f"Invalid diarization speaker label: {speaker_label!r}."
+            )
+        if not isinstance(participant_id, str) or participant_id not in participant_ids:
+            raise MeetingContextValidationError(
+                f"speaker_mappings.{speaker_label} references unknown participant: "
+                f"{participant_id!r}."
+            )
+
     for index, participant in enumerate(participants):
         item_path = f"participants[{index}]"
-        _validate_attendance(participant, item_path)
-        if participant.get("attendance_status") != "present":
+        status = _validate_attendance(participant, item_path, default="present")
+        if status != "present":
             raise MeetingContextValidationError(
                 f"{item_path}.attendance_status must be 'present'."
             )
@@ -85,10 +153,10 @@ def validate_meeting_context(data: dict[str, Any]) -> None:
 
     for index, person in enumerate(mentioned_people):
         item_path = f"mentioned_people[{index}]"
-        _validate_attendance(person, item_path)
-        if person.get("attendance_status") == "present":
+        status = _validate_attendance(person, item_path, default="mentioned_only")
+        if status != "mentioned_only":
             raise MeetingContextValidationError(
-                f"{item_path}.attendance_status must not be 'present'."
+                f"{item_path}.attendance_status must be 'mentioned_only'."
             )
         _validate_department_reference(person, item_path, department_ids)
 
@@ -131,6 +199,28 @@ def render_meeting_context_for_prompt(context: MeetingContext) -> str:
         for participant in participants:
             lines.append(_render_person_line(participant, "participant_id", departments_by_id))
 
+    speaker_mappings = context.speaker_mappings
+    if speaker_mappings:
+        participants_by_id = {
+            participant["participant_id"]: participant
+            for participant in participants
+            if isinstance(participant, dict) and participant.get("participant_id")
+        }
+        lines.extend(
+            [
+                "",
+                "Confirmed diarization speaker mappings (authoritative):",
+                "- Use only these explicit mappings. Never infer identities for other speaker labels.",
+                "- Unmapped SPEAKER_XX labels must remain anonymous.",
+            ]
+        )
+        for speaker_label, participant_id in sorted(speaker_mappings.items()):
+            participant = participants_by_id[participant_id]
+            lines.append(
+                f"- {speaker_label}: {_text(participant.get('display_name'))} "
+                f"(participant_id: {participant_id})"
+            )
+
     mentioned_people = _optional_list(data.get("mentioned_people"), "mentioned_people")
     if mentioned_people:
         lines.extend(["", "Mentioned but absent people:"])
@@ -245,12 +335,28 @@ def _collect_unique_ids(items: list[Any], key: str, path: str) -> set[str]:
     return ids
 
 
-def _validate_attendance(item: dict[str, Any], path: str) -> None:
-    status = item.get("attendance_status")
+def _validate_attendance(item: dict[str, Any], path: str, *, default: str) -> str:
+    status = item.get("attendance_status", default)
     if status not in VALID_ATTENDANCE_STATUSES:
         raise MeetingContextValidationError(
             f"{path}.attendance_status has invalid value: {status!r}."
         )
+    return status
+
+
+def _with_attendance_defaults(data: dict[str, Any]) -> dict[str, Any]:
+    normalized = copy.deepcopy(data)
+    participants = normalized.get("participants")
+    if isinstance(participants, list):
+        for participant in participants:
+            if isinstance(participant, dict):
+                participant.setdefault("attendance_status", "present")
+    mentioned_people = normalized.get("mentioned_people")
+    if isinstance(mentioned_people, list):
+        for person in mentioned_people:
+            if isinstance(person, dict):
+                person.setdefault("attendance_status", "mentioned_only")
+    return normalized
 
 
 def _validate_department_reference(

src/meeting_lab/orchestration/__init__.py

@@ -0,0 +1,17 @@
+"""Reusable orchestration APIs for Meeting Lab applications and CLIs."""
+
+from src.meeting_lab.orchestration.mvp import (
+    DEFAULT_OUTPUT_ROOT,
+    MvpMeetingConfig,
+    MvpRunResult,
+    create_unique_run_dir,
+    run_mvp_meeting,
+)
+
+__all__ = [
+    "DEFAULT_OUTPUT_ROOT",
+    "MvpMeetingConfig",
+    "MvpRunResult",
+    "create_unique_run_dir",
+    "run_mvp_meeting",
+]

src/meeting_lab/orchestration/mvp.py

@@ -0,0 +1,467 @@
+"""Reusable audio-to-direct-protocol MVP orchestration."""
+
+from __future__ import annotations
+
+import json
+import re
+import shutil
+import sys
+import time
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from src.meeting_lab.audio import prepare_audio
+from src.meeting_lab.diarization import (
+    DEFAULT_MODEL as DEFAULT_DIARIZATION_MODEL,
+    diarize_audio,
+    write_diarized_transcript,
+)
+from src.meeting_lab.llm.ollama import DEFAULT_ENDPOINT
+from src.meeting_lab.models.meeting_context import (
+    MeetingContext,
+    create_meeting_context,
+    load_meeting_context,
+    validate_meeting_context,
+    write_meeting_context,
+)
+from src.meeting_lab.progress import ProgressEvent, ProgressSink, ProgressStatus
+from src.meeting_lab.protocol.generate_direct_protocol import (
+    DEFAULT_MODEL,
+    DEFAULT_NUM_CTX,
+    DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+    DirectProtocolResult,
+    generate_direct_protocol,
+    load_compact_transcript,
+)
+from src.meeting_lab.transcription.whisper import transcribe_audio
+
+
+DEFAULT_OUTPUT_ROOT = Path("meeting_data/runs")
+ContextInput = MeetingContext | Mapping[str, Any]
+
+
+@dataclass(frozen=True)
+class MvpMeetingConfig:
+    audio_file: Path
+    whisper_model: Path
+    whisper_executable: str = "whisper-cli"
+    ffmpeg_executable: str = "ffmpeg"
+    audio_normalization: bool = True
+    context_file: Path | None = None
+    output_root: Path = DEFAULT_OUTPUT_ROOT
+    language: str = "de"
+    threads: str | int = "auto"
+    model: str = DEFAULT_MODEL
+    ollama_endpoint: str = DEFAULT_ENDPOINT
+    protocol_num_ctx: int = DEFAULT_NUM_CTX
+    protocol_safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET
+    diarization: str = "off"
+    diarization_runtime: str = "native"
+    diarization_container_image: str | None = None
+    diarization_container_args: Sequence[str] = ()
+
+
+@dataclass(frozen=True)
+class MvpRunResult:
+    exit_code: int
+    run_dir: Path | None
+    protocol_path: Path | None
+
+
+def regenerate_mvp_protocol(
+    run_dir: Path,
+    *,
+    meeting_context: ContextInput,
+    model: str = DEFAULT_MODEL,
+    ollama_endpoint: str = DEFAULT_ENDPOINT,
+    protocol_num_ctx: int = DEFAULT_NUM_CTX,
+    protocol_safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+    progress_sink: ProgressSink | None = None,
+) -> MvpRunResult:
+    """Regenerate only protocol artifacts from an existing completed run."""
+    started = time.perf_counter()
+    run_dir = Path(run_dir)
+    context = _effective_context(meeting_context)
+    if context is None:
+        raise ValueError("Meeting Context is required for protocol regeneration.")
+    if protocol_num_ctx <= 0:
+        raise ValueError("Protocol Ollama context size must be positive.")
+    if protocol_safe_input_token_budget <= 0:
+        raise ValueError("Protocol safe input token budget must be positive.")
+
+    diarized_transcript = run_dir / "diarization" / "transcript_diarized.json"
+    plain_transcript = run_dir / "transcript" / "transcript.json"
+    transcript_path = (
+        diarized_transcript if diarized_transcript.is_file() else plain_transcript
+    )
+    if not transcript_path.is_file():
+        raise FileNotFoundError(
+            f"Existing run has no protocol transcript artifact: {run_dir}"
+        )
+
+    context_path = run_dir / "context" / "meeting_context.yaml"
+    write_meeting_context(context, context_path)
+    _emit(progress_sink, "protocol_generation", "started", started)
+    try:
+        result = generate_direct_protocol(
+            transcript_path,
+            context_path,
+            model=model,
+            endpoint=ollama_endpoint,
+            num_ctx=protocol_num_ctx,
+            safe_input_token_budget=protocol_safe_input_token_budget,
+        )
+        protocol_path = _persist_protocol(run_dir, result)
+    except Exception as exc:
+        _emit(
+            progress_sink,
+            "failed",
+            "failed",
+            started,
+            message=f"protocol_generation: {type(exc).__name__}: {exc}",
+        )
+        raise
+    _emit(progress_sink, "protocol_generation", "completed", started)
+    _emit(progress_sink, "completed", "completed", started)
+    return MvpRunResult(0, run_dir, protocol_path)
+
+
+def create_unique_run_dir(
+    output_root: Path,
+    meeting_name: str,
+    now: Callable[[], datetime] = datetime.now,
+) -> Path:
+    safe_name = re.sub(r"[^A-Za-z0-9_.-]+", "_", meeting_name).strip("._-") or "meeting"
+    base = output_root / f"{safe_name}_{now().strftime('%Y%m%d_%H%M%S')}"
+    candidate = base
+    suffix = 1
+    while candidate.exists():
+        candidate = output_root / f"{base.name}_{suffix:02d}"
+        suffix += 1
+    candidate.mkdir(parents=True)
+    return candidate
+
+
+def _write_json(path: Path, value: Any) -> None:
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def _effective_context(value: ContextInput | None) -> MeetingContext | None:
+    if value is None:
+        return None
+    if isinstance(value, MeetingContext):
+        validate_meeting_context(value.data)
+        return value
+    if isinstance(value, Mapping):
+        return create_meeting_context(dict(value), source_file=Path("<programmatic>"))
+    raise TypeError("meeting_context must be MeetingContext, mapping, or None.")
+
+
+def _validate_inputs(
+    config: MvpMeetingConfig, meeting_context: MeetingContext | None
+) -> None:
+    if not config.audio_file.is_file():
+        raise FileNotFoundError(f"Audio file does not exist: {config.audio_file}")
+    if not config.whisper_model.is_file():
+        raise FileNotFoundError(f"Whisper model does not exist: {config.whisper_model}")
+    if meeting_context is not None and config.context_file is not None:
+        raise ValueError("Use either a context file or a programmatic Meeting Context, not both.")
+    if meeting_context is not None:
+        validate_meeting_context(meeting_context.data)
+    elif config.context_file is not None:
+        if not config.context_file.is_file():
+            raise FileNotFoundError(
+                f"Meeting Context file does not exist: {config.context_file}"
+            )
+        load_meeting_context(config.context_file)
+    if config.diarization not in ("off", "auto", "gpu", "cpu"):
+        raise ValueError(f"Unsupported diarization mode: {config.diarization}")
+    if config.diarization_runtime not in ("native", "container"):
+        raise ValueError(
+            f"Unsupported diarization runtime: {config.diarization_runtime}"
+        )
+    if (
+        config.diarization != "off"
+        and config.diarization_runtime == "container"
+        and not config.diarization_container_image
+    ):
+        raise ValueError("A diarization container image is required.")
+    if config.protocol_safe_input_token_budget <= 0:
+        raise ValueError("Protocol safe input token budget must be positive.")
+    if config.protocol_num_ctx <= 0:
+        raise ValueError("Protocol Ollama context size must be positive.")
+
+
+def _emit(
+    sink: ProgressSink | None,
+    stage: str,
+    status: ProgressStatus,
+    overall_started: float,
+    *,
+    message: str | None = None,
+) -> None:
+    if sink is not None:
+        sink(
+            ProgressEvent(
+                stage=stage,
+                status=status,
+                elapsed_seconds=time.perf_counter() - overall_started,
+                message=message,
+            )
+        )
+
+
+def _persist_protocol(run_dir: Path, result: DirectProtocolResult) -> Path:
+    protocol_dir = run_dir / "protocol"
+    protocol_dir.mkdir(exist_ok=True)
+    (protocol_dir / "exact_prompt.txt").write_text(result.exact_prompt, encoding="utf-8")
+    _write_json(protocol_dir / "raw_response.json", result.raw_response)
+    _write_json(protocol_dir / "runtime_metadata.json", result.runtime_metadata)
+    transcript_input = getattr(result, "transcript_input", None)
+    if transcript_input is not None:
+        (protocol_dir / "transcript_input.txt").write_text(
+            transcript_input, encoding="utf-8"
+        )
+    protocol_path = run_dir / "protocol.md"
+    protocol_path.write_text(result.protocol_text, encoding="utf-8")
+    return protocol_path
+
+
+def run_mvp_meeting(
+    config: MvpMeetingConfig,
+    *,
+    meeting_context: ContextInput | None = None,
+    progress_sink: ProgressSink | None = None,
+) -> MvpRunResult:
+    """Run the existing MVP directly, without subprocess or GUI dependencies."""
+    overall_started = time.perf_counter()
+    validation_started = time.perf_counter()
+    _emit(progress_sink, "preparing", "started", overall_started)
+    try:
+        effective_context = _effective_context(meeting_context)
+        _validate_inputs(config, effective_context)
+    except Exception as exc:
+        _emit(
+            progress_sink,
+            "failed",
+            "failed",
+            overall_started,
+            message=f"preparing: {type(exc).__name__}: {exc}",
+        )
+        print(f"Error: {type(exc).__name__}: {exc}", file=sys.stderr)
+        return MvpRunResult(2, None, None)
+
+    validation_runtime = time.perf_counter() - validation_started
+    run_dir = create_unique_run_dir(config.output_root, config.audio_file.stem)
+    timestamp = datetime.now().astimezone().isoformat(timespec="seconds")
+    transcript_path = run_dir / "transcript" / "transcript.json"
+    protocol_path = run_dir / "protocol.md"
+    stage_runtimes: dict[str, float | None] = {
+        "validation": round(validation_runtime, 3),
+        "setup": None,
+        "audio_preparation": None,
+        "whisper": None,
+        "transcript_validation": None,
+        "protocol": None,
+    }
+    if config.diarization != "off":
+        stage_runtimes["diarization"] = None
+        stage_runtimes["diarization_alignment"] = None
+    metadata: dict[str, Any] = {
+        "run_id": run_dir.name,
+        "timestamp": timestamp,
+        "input_audio": str(config.audio_file.resolve()),
+        "audio_preparation": None,
+        "transcript_output": str(transcript_path.resolve()),
+        "protocol_output": str(protocol_path.resolve()),
+        "whisper_model": str(config.whisper_model.resolve()),
+        "model": config.model,
+        "ollama_endpoint": config.ollama_endpoint,
+        "status": "running",
+        "stage_runtimes_seconds": stage_runtimes,
+        "total_runtime_seconds": None,
+        "failure": None,
+        "diarization": {
+            "enabled": config.diarization != "off",
+            "backend": "pyannote.audio" if config.diarization != "off" else None,
+            "model": DEFAULT_DIARIZATION_MODEL if config.diarization != "off" else None,
+            "requested_device_mode": config.diarization,
+            "runtime": config.diarization_runtime if config.diarization != "off" else None,
+            "metadata_path": None,
+            "transcript_diarized": None,
+        },
+    }
+    current_stage = "preparing"
+    stage_started = time.perf_counter()
+
+    try:
+        audio_dir = run_dir / "audio"
+        transcript_dir = run_dir / "transcript"
+        context_dir = run_dir / "context"
+        protocol_dir = run_dir / "protocol"
+        audio_dir.mkdir()
+        transcript_dir.mkdir()
+        context_dir.mkdir()
+        protocol_dir.mkdir()
+        _write_json(
+            audio_dir / "input_manifest.json",
+            {
+                "source_file": str(config.audio_file.resolve()),
+                "filename": config.audio_file.name,
+                "size_bytes": config.audio_file.stat().st_size,
+            },
+        )
+
+        preparation_started = time.perf_counter()
+        current_stage = "audio_preparation"
+        stage_started = preparation_started
+        prepared_audio = prepare_audio(
+            config.audio_file,
+            audio_dir / "prepared.wav",
+            ffmpeg_executable=config.ffmpeg_executable,
+            normalization_enabled=config.audio_normalization,
+        )
+        stage_runtimes["audio_preparation"] = round(
+            time.perf_counter() - preparation_started, 3
+        )
+        current_stage = "preparing"
+        preparation_metadata = prepared_audio.metadata()
+        metadata["audio_preparation"] = preparation_metadata
+        _write_json(audio_dir / "preparation_metadata.json", preparation_metadata)
+        _write_json(
+            audio_dir / "input_manifest.json",
+            {
+                "source_file": str(config.audio_file.resolve()),
+                "filename": config.audio_file.name,
+                "size_bytes": config.audio_file.stat().st_size,
+                "format": config.audio_file.suffix.lower().removeprefix("."),
+                "prepared_audio": preparation_metadata,
+            },
+        )
+
+        preserved_context: Path | None = None
+        if effective_context is not None:
+            preserved_context = context_dir / "meeting_context.yaml"
+            write_meeting_context(effective_context, preserved_context)
+        elif config.context_file is not None:
+            preserved_context = context_dir / "meeting_context.yaml"
+            shutil.copy2(config.context_file, preserved_context)
+        stage_runtimes["setup"] = round(time.perf_counter() - stage_started, 3)
+        _emit(progress_sink, "preparing", "completed", overall_started)
+
+        current_stage = "transcription"
+        stage_started = time.perf_counter()
+        _emit(progress_sink, "transcription", "started", overall_started)
+        transcription = transcribe_audio(
+            prepared_audio.prepared_path,
+            config.whisper_model,
+            transcript_dir,
+            config.language,
+            executable=config.whisper_executable,
+            threads=config.threads,
+        )
+        stage_runtimes["whisper"] = round(time.perf_counter() - stage_started, 3)
+        _emit(progress_sink, "transcription", "completed", overall_started)
+
+        stage_started = time.perf_counter()
+        load_compact_transcript(transcription.transcript_json)
+        stage_runtimes["transcript_validation"] = round(
+            time.perf_counter() - stage_started, 3
+        )
+
+        protocol_transcript = transcription.transcript_json
+        if config.diarization != "off":
+            current_stage = "diarization"
+            stage_started = time.perf_counter()
+            _emit(progress_sink, "diarization", "started", overall_started)
+            diarization_dir = run_dir / "diarization"
+            diarization = diarize_audio(
+                prepared_audio.prepared_path,
+                diarization_dir,
+                config.diarization,
+                runtime=config.diarization_runtime,
+                container_image=config.diarization_container_image,
+                container_args=config.diarization_container_args,
+            )
+            stage_runtimes["diarization"] = round(
+                time.perf_counter() - stage_started, 3
+            )
+            metadata["diarization"].update(
+                {
+                    "actual_device": diarization.metadata.get("actual_device"),
+                    "device_name": diarization.metadata.get("device_name"),
+                    "runtime_seconds": diarization.metadata.get("runtime_seconds"),
+                    "speaker_count": diarization.metadata.get("speaker_count"),
+                    "metadata_path": str(diarization.metadata_path.resolve()),
+                }
+            )
+
+            stage_started = time.perf_counter()
+            protocol_transcript, diarized_text = write_diarized_transcript(
+                transcription.transcript_json,
+                diarization.exclusive_turns_json,
+                diarization_dir,
+            )
+            load_compact_transcript(protocol_transcript)
+            stage_runtimes["diarization_alignment"] = round(
+                time.perf_counter() - stage_started, 3
+            )
+            metadata["diarization"].update(
+                {
+                    "transcript_diarized": str(protocol_transcript.resolve()),
+                    "transcript_diarized_text": str(diarized_text.resolve()),
+                }
+            )
+            _emit(progress_sink, "diarization", "completed", overall_started)
+
+        current_stage = "protocol_generation"
+        stage_started = time.perf_counter()
+        _emit(progress_sink, "protocol_generation", "started", overall_started)
+        result = generate_direct_protocol(
+            protocol_transcript,
+            preserved_context,
+            model=config.model,
+            endpoint=config.ollama_endpoint,
+            num_ctx=config.protocol_num_ctx,
+            safe_input_token_budget=config.protocol_safe_input_token_budget,
+        )
+        stage_runtimes["protocol"] = round(time.perf_counter() - stage_started, 3)
+        protocol_path = _persist_protocol(run_dir, result)
+        _emit(progress_sink, "protocol_generation", "completed", overall_started)
+        metadata["status"] = "completed"
+        _emit(progress_sink, "completed", "completed", overall_started)
+    except Exception as exc:
+        metadata_stage = {
+            "preparing": "setup",
+            "audio_preparation": "audio_preparation",
+            "transcription": "whisper",
+            "diarization": "diarization",
+            "protocol_generation": "protocol",
+        }.get(current_stage, current_stage)
+        runtime_key = metadata_stage
+        if runtime_key in stage_runtimes and stage_runtimes[runtime_key] is None:
+            stage_runtimes[runtime_key] = round(time.perf_counter() - stage_started, 3)
+        metadata["status"] = "failed"
+        metadata["failure"] = {
+            "stage": metadata_stage,
+            "type": type(exc).__name__,
+            "message": str(exc),
+        }
+        protocol_path = None
+        _emit(
+            progress_sink,
+            "failed",
+            "failed",
+            overall_started,
+            message=f"{current_stage}: {type(exc).__name__}: {exc}",
+        )
+        print(f"Error: {type(exc).__name__}: {exc}", file=sys.stderr)
+    finally:
+        metadata["total_runtime_seconds"] = round(time.perf_counter() - overall_started, 3)
+        _write_json(run_dir / "run_metadata.json", metadata)
+
+    exit_code = 0 if metadata["status"] == "completed" else 2
+    return MvpRunResult(exit_code, run_dir, protocol_path)

src/meeting_lab/progress.py

@@ -0,0 +1,21 @@
+"""Small observer boundary for long-running Meeting Lab operations."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Literal
+
+
+ProgressStatus = Literal["started", "completed", "failed"]
+
+
+@dataclass(frozen=True)
+class ProgressEvent:
+    stage: str
+    status: ProgressStatus
+    elapsed_seconds: float
+    progress: float | None = None
+    message: str | None = None
+
+
+ProgressSink = Callable[[ProgressEvent], None]

src/meeting_lab/protocol/direct_protocol_prompt.py

@@ -0,0 +1,32 @@
+"""Prompt construction for the direct transcript-to-protocol MVP."""
+
+from __future__ import annotations
+
+
+DIRECT_PROTOCOL_INSTRUCTION = """Erstelle aus dem vollständigen Transkript und dem Meeting-Kontext ein vollständiges, strukturiertes und professionelles internes Besprechungsprotokoll in deutscher Sprache.
+
+Das Protokoll muss themenorientiert sein, nicht chronologisch und nicht nach technischen Kategorien gegliedert. Beginne mit # Meeting Protocol. Verwende für jedes kohärente Thema eine Überschrift ## <Thema> und darunter eine strukturierte Synthese der Diskussion. Bewahre relevante Diskussionsverläufe, unterschiedliche Positionen, offene Punkte und Entscheidungsgrundlagen. Dokumentiere die wesentlichen Inhalte nachvollziehbar und fasse Themenblöcke so zusammen, dass auch Personen, die nicht am Meeting teilgenommen haben, den Kontext und die Entwicklung der Diskussion verstehen können. Nenne Entscheidungen oder abgestimmte Positionen nur, wenn sie tatsächlich belegt sind. Führe Maßnahmen nur auf, wenn eine konkrete zukünftige Handlung gestützt ist; nenne verantwortliche Personen und Fristen ausschließlich bei expliziter Zuweisung, Annahme oder Bestätigung im Transkript. Vorschläge, Einwände, Möglichkeiten und vorläufige Ideen sind keine Entscheidungen oder Verpflichtungen. Bewahre relevante Einschränkungen und ungelöste Meinungsverschiedenheiten. Nenne offene Punkte nur, wenn sie wirklich offen bleiben. Nicht jedes Thema benötigt Entscheidungen, Maßnahmen oder offene Punkte.
+
+Erzeuge keine reine Wiedergabe des Transkripts und verlängere das Protokoll nicht unnötig durch Wiederholungen. Synthetisiere zusammengehörige Aussagen, entferne Füllwörter und Gesprächsrauschen und erfinde keine Fakten, Entscheidungen, Zustimmungen, Verantwortlichen oder Fristen. Gib kein JSON, keine internen Labels und keine Analyse oder Denkprotokolle aus. Das Ergebnis soll als Markdown-Protokoll nach geringfügiger menschlicher Redaktion intern versendbar sein. Eine kompakte themenübergreifende Maßnahmenliste am Ende ist optional, wenn sie nützlich und vollständig belegt ist."""
+
+COMPACT_DIARIZED_PROTOCOL_INSTRUCTION = """Erstelle aus dem vollständigen Transkript und Meeting-Kontext ein vollständiges, professionelles internes Besprechungsprotokoll auf Deutsch. Das Transkript ist in aufeinanderfolgende anonyme Sprecherblöcke gegliedert.
+
+Beginne mit # Meeting Protocol. Gliedere themenorientiert mit ## <Thema> und synthetisiere je Thema den relevanten Diskussionsverlauf, Kontext, unterschiedliche Positionen, Entscheidungsgrundlagen, Einschränkungen und ungelöste Meinungsverschiedenheiten so, dass Dritte ihn nachvollziehen können. Nenne Entscheidungen nur bei Beleg. Nenne Maßnahmen, Verantwortliche und Fristen nur bei expliziter Zuweisung, Annahme oder Bestätigung; Vorschläge sind keine Verpflichtungen.
+
+Entferne nur Wiederholungen, Füllwörter und Gesprächsrauschen. Erfinde keine Fakten oder Identitäten. Gib kein JSON, keine Sprecherlabels und kein Denkprotokoll aus. Eine belegte themenübergreifende Maßnahmenliste am Ende ist optional."""
+
+MAPPED_SPEAKER_ATTRIBUTION_INSTRUCTION = """Nutze die autoritativen SPEAKER_XX-zu-Teilnehmer-Zuordnungen im Meeting-Kontext, um ausdrücklich belegte Aussagen, Positionen, Entscheidungen, Zuweisungen und angenommene persönliche Verpflichtungen namentlich zuzuordnen. Eine ausdrückliche Ich-Zusage eines zugeordneten Sprechers belegt persönliche Verantwortung. Unterscheide stets den Sprecher einer Aussage von darin nur erwähnten Personen. Leite für nicht zugeordnete Sprecher keine Identität ab und erfinde keine persönliche Verantwortung. Gib die technischen SPEAKER_XX-Bezeichnungen nicht im nutzerseitigen Protokoll aus."""
+
+
+def build_direct_protocol_prompt(
+    transcript: str,
+    meeting_context: str | None = None,
+    *,
+    instruction: str = DIRECT_PROTOCOL_INSTRUCTION,
+) -> str:
+    context = meeting_context.strip() if meeting_context else "Kein Meeting-Kontext bereitgestellt."
+    return (
+        f"{instruction}\n\n"
+        f"MEETING-KONTEXT:\n{context}\n\n"
+        f"VOLLSTAENDIGES TRANSKRIPT:\n{transcript.strip()}\n"
+    )

src/meeting_lab/protocol/generate_direct_protocol.py

@@ -0,0 +1,236 @@
+"""One-call direct protocol generation from a compact Whisper transcript."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable
+
+from src.meeting_lab.llm.ollama import (
+    DEFAULT_ENDPOINT,
+    OllamaGeneration,
+    generate_once,
+    require_model,
+)
+from src.meeting_lab.models.meeting_context import (
+    MeetingContext,
+    load_meeting_context,
+    render_meeting_context_for_prompt,
+)
+from src.meeting_lab.protocol.direct_protocol_prompt import (
+    COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
+    MAPPED_SPEAKER_ATTRIBUTION_INSTRUCTION,
+    build_direct_protocol_prompt,
+)
+from src.meeting_lab.protocol.transcript_input import (
+    TranscriptInputError,
+    compact_diarized_transcript,
+    plain_segment_transcript,
+)
+
+
+DEFAULT_MODEL = "qwen3.6:35B-A3B"
+DEFAULT_NUM_CTX = 32768
+DEFAULT_NUM_PREDICT = 8192
+DEFAULT_TIMEOUT = 1800
+DEFAULT_SAFE_INPUT_TOKEN_BUDGET = 29_000
+ESTIMATED_UTF8_BYTES_PER_TOKEN = 4.4
+
+
+class DirectProtocolError(ValueError):
+    """Raised for invalid direct-protocol inputs or model output."""
+
+
+@dataclass(frozen=True)
+class DirectProtocolResult:
+    protocol_text: str
+    exact_prompt: str
+    model_metadata: dict[str, Any]
+    runtime_metadata: dict[str, Any]
+    raw_response: dict[str, Any]
+    transcript_input: str | None = None
+
+
+@dataclass(frozen=True)
+class SelectedTranscriptInput:
+    text: str
+    prompt: str
+    representation: str
+    estimated_input_tokens: int
+    safe_input_token_budget: int
+    fallback_used: bool
+    diarization_enabled: bool
+
+
+def load_compact_transcript(path: Path) -> str:
+    data = _load_transcript_document(path)
+    text = data.get("text")
+    if not isinstance(text, str) or not text.strip():
+        raise DirectProtocolError("Transcript top-level 'text' must be a non-empty string.")
+    return text
+
+
+def _load_transcript_document(path: Path) -> dict[str, Any]:
+    if not path.is_file():
+        raise DirectProtocolError(f"Transcript file does not exist: {path}")
+    try:
+        data = json.loads(path.read_text(encoding="utf-8-sig"))
+    except json.JSONDecodeError as exc:
+        raise DirectProtocolError(f"Transcript is not valid JSON: {path}: {exc}") from exc
+    if not isinstance(data, dict):
+        raise DirectProtocolError("Transcript JSON must contain a top-level object.")
+    if "text" not in data:
+        raise DirectProtocolError("Transcript JSON must contain top-level 'text'.")
+    return data
+
+
+def estimate_input_tokens(prompt: str) -> int:
+    """Estimate tokens without adding a model-specific tokenizer dependency."""
+    byte_count = len(prompt.encode("utf-8"))
+    return max(1, int(byte_count / ESTIMATED_UTF8_BYTES_PER_TOKEN + 0.999999))
+
+
+def select_transcript_input(
+    transcript: dict[str, Any],
+    rendered_context: str | None,
+    *,
+    safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+) -> SelectedTranscriptInput:
+    """Select complete prompt input without allowing silent tail truncation."""
+    if safe_input_token_budget <= 0:
+        raise DirectProtocolError("Safe protocol input token budget must be positive.")
+    diarization_enabled = transcript.get("speaker_labels_anonymous") is True
+    if diarization_enabled:
+        try:
+            compact = compact_diarized_transcript(transcript.get("segments"))
+            plain_text = plain_segment_transcript(transcript.get("segments"))
+        except TranscriptInputError as exc:
+            raise DirectProtocolError(str(exc)) from exc
+        instruction = COMPACT_DIARIZED_PROTOCOL_INSTRUCTION
+        if (
+            rendered_context
+            and "Confirmed diarization speaker mappings" in rendered_context
+        ):
+            instruction = f"{instruction}\n\n{MAPPED_SPEAKER_ATTRIBUTION_INSTRUCTION}"
+        compact_prompt = build_direct_protocol_prompt(
+            compact.text,
+            rendered_context,
+            instruction=instruction,
+        )
+        compact_estimate = estimate_input_tokens(compact_prompt)
+        if compact_estimate <= safe_input_token_budget:
+            return SelectedTranscriptInput(
+                text=compact.text,
+                prompt=compact_prompt,
+                representation="diarized_compact",
+                estimated_input_tokens=compact_estimate,
+                safe_input_token_budget=safe_input_token_budget,
+                fallback_used=False,
+                diarization_enabled=True,
+            )
+        representation = "plain_transcript_fallback"
+        fallback_used = True
+    else:
+        plain_text = transcript.get("text")
+        if not isinstance(plain_text, str) or not plain_text.strip():
+            raise DirectProtocolError("Transcript top-level 'text' must be a non-empty string.")
+        representation = "plain_transcript"
+        fallback_used = False
+
+    plain_prompt = build_direct_protocol_prompt(plain_text, rendered_context)
+    plain_estimate = estimate_input_tokens(plain_prompt)
+    if plain_estimate > safe_input_token_budget:
+        raise DirectProtocolError(
+            "Protocol prompt/input is too large for the configured safe input budget "
+            f"({plain_estimate} estimated tokens > {safe_input_token_budget}). "
+            "No LLM request was made; silent truncation is not allowed."
+        )
+    return SelectedTranscriptInput(
+        text=plain_text,
+        prompt=plain_prompt,
+        representation=representation,
+        estimated_input_tokens=plain_estimate,
+        safe_input_token_budget=safe_input_token_budget,
+        fallback_used=fallback_used,
+        diarization_enabled=diarization_enabled,
+    )
+
+
+def generate_direct_protocol(
+    transcript_path: Path,
+    context_path: Path | None = None,
+    *,
+    model: str = DEFAULT_MODEL,
+    endpoint: str = DEFAULT_ENDPOINT,
+    timeout: int = DEFAULT_TIMEOUT,
+    num_ctx: int = DEFAULT_NUM_CTX,
+    num_predict: int = DEFAULT_NUM_PREDICT,
+    safe_input_token_budget: int = DEFAULT_SAFE_INPUT_TOKEN_BUDGET,
+    model_check: Callable[[str, str, int], dict[str, Any]] = require_model,
+    generation_call: Callable[..., OllamaGeneration] = generate_once,
+) -> DirectProtocolResult:
+    transcript = _load_transcript_document(transcript_path)
+    context: MeetingContext | None = (
+        load_meeting_context(context_path) if context_path is not None else None
+    )
+    rendered_context = render_meeting_context_for_prompt(context) if context else None
+    selected = select_transcript_input(
+        transcript,
+        rendered_context,
+        safe_input_token_budget=safe_input_token_budget,
+    )
+
+    model_metadata = model_check(endpoint, model, 10)
+    generation = generation_call(
+        endpoint,
+        model,
+        selected.prompt,
+        timeout=timeout,
+        num_ctx=num_ctx,
+        num_predict=num_predict,
+    )
+    data = generation.raw_response
+    runtime_metadata = {
+        "model": model,
+        "prompt_token_count": data.get("prompt_eval_count"),
+        "output_token_count": data.get("eval_count"),
+        "prompt_evaluation_duration_ns": data.get("prompt_eval_duration"),
+        "generation_duration_ns": data.get("eval_duration"),
+        "total_ollama_duration_ns": data.get("total_duration"),
+        "client_wall_time_seconds": generation.client_wall_time_seconds,
+        "completion_reason": data.get("done_reason"),
+        "done": data.get("done"),
+        "request_count": 1,
+        "temperature": 0.0,
+        "think": False,
+        "num_ctx": num_ctx,
+        "num_predict": num_predict,
+        "selected_transcript_representation": selected.representation,
+        "estimated_input_tokens": selected.estimated_input_tokens,
+        "safe_input_token_budget": selected.safe_input_token_budget,
+        "input_token_estimation_method": "utf8_bytes_divided_by_4.4",
+        "fallback_used": selected.fallback_used,
+        "diarization_enabled": selected.diarization_enabled,
+        "speaker_attribution_available": (
+            True
+            if selected.representation == "diarized_compact"
+            else False
+            if selected.representation == "plain_transcript_fallback"
+            else None
+        ),
+        "speaker_attribution_loss_reason": (
+            "plain_transcript_fallback"
+            if selected.representation == "plain_transcript_fallback"
+            else None
+        ),
+        "speaker_mapping_count": len(context.speaker_mappings) if context else 0,
+    }
+    return DirectProtocolResult(
+        protocol_text=generation.text,
+        exact_prompt=selected.prompt,
+        model_metadata=model_metadata,
+        runtime_metadata=runtime_metadata,
+        raw_response=data,
+        transcript_input=selected.text,
+    )

src/meeting_lab/protocol/transcript_input.py

@@ -0,0 +1,97 @@
+"""Deterministic transcript representations for one-call protocol prompts."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+
+class TranscriptInputError(ValueError):
+    """Raised when a transcript cannot be represented without content loss."""
+
+
+@dataclass(frozen=True)
+class SpeakerBlock:
+    """One contiguous run of transcript segments assigned to one speaker."""
+
+    speaker_id: str
+    segment_texts: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class CompactDiarizedTranscript:
+    """Compact prompt text plus structural evidence of segment preservation."""
+
+    text: str
+    blocks: tuple[SpeakerBlock, ...]
+    source_segment_count: int
+
+    @property
+    def represented_segment_count(self) -> int:
+        return sum(len(block.segment_texts) for block in self.blocks)
+
+    @property
+    def segment_texts(self) -> tuple[str, ...]:
+        return tuple(text for block in self.blocks for text in block.segment_texts)
+
+
+def normalize_segment_text(value: Any, index: int) -> str:
+    """Normalize formatting whitespace while retaining all semantic text."""
+    if not isinstance(value, str):
+        raise TranscriptInputError(f"Transcript segment {index} text must be a string.")
+    return " ".join(value.split())
+
+
+def compact_diarized_transcript(segments: Any) -> CompactDiarizedTranscript:
+    """Group only adjacent same-speaker segments and omit repeated timestamps."""
+    if not isinstance(segments, list) or not segments:
+        raise TranscriptInputError(
+            "Diarized transcript must contain a non-empty 'segments' list."
+        )
+
+    mutable_blocks: list[tuple[str, list[str]]] = []
+    source_texts: list[str] = []
+    for index, segment in enumerate(segments):
+        if not isinstance(segment, dict):
+            raise TranscriptInputError(f"Transcript segment {index} must be an object.")
+        speaker = segment.get("speaker_id") or "SPEAKER_UNASSIGNED"
+        if not isinstance(speaker, str) or not speaker.startswith("SPEAKER_"):
+            raise TranscriptInputError(
+                f"Transcript segment {index} must use an anonymous SPEAKER_ label."
+            )
+        text = normalize_segment_text(segment.get("text"), index)
+        source_texts.append(text)
+        if mutable_blocks and mutable_blocks[-1][0] == speaker:
+            mutable_blocks[-1][1].append(text)
+        else:
+            mutable_blocks.append((speaker, [text]))
+
+    blocks = tuple(
+        SpeakerBlock(speaker_id=speaker, segment_texts=tuple(texts))
+        for speaker, texts in mutable_blocks
+    )
+    rendered = "\n".join(
+        f"{block.speaker_id}: {' '.join(block.segment_texts)}" for block in blocks
+    )
+    result = CompactDiarizedTranscript(
+        text=rendered + "\n",
+        blocks=blocks,
+        source_segment_count=len(segments),
+    )
+    if result.represented_segment_count != len(segments):
+        raise TranscriptInputError("Compact diarized transcript lost source segments.")
+    if result.segment_texts != tuple(source_texts):
+        raise TranscriptInputError("Compact diarized transcript changed segment order or text.")
+    return result
+
+
+def plain_segment_transcript(segments: Any) -> str:
+    """Reconstruct plain transcript text from every segment in source order."""
+    if not isinstance(segments, list) or not segments:
+        raise TranscriptInputError("Transcript must contain a non-empty 'segments' list.")
+    texts = []
+    for index, segment in enumerate(segments):
+        if not isinstance(segment, dict):
+            raise TranscriptInputError(f"Transcript segment {index} must be an object.")
+        texts.append(normalize_segment_text(segment.get("text"), index))
+    return " ".join(texts)

src/meeting_lab/transcription/__init__.py

@@ -0,0 +1,5 @@
+"""Audio transcription support for the direct-protocol MVP."""
+
+from .whisper import TranscriptionError, TranscriptionResult, transcribe_audio
+
+__all__ = ["TranscriptionError", "TranscriptionResult", "transcribe_audio"]

src/meeting_lab/transcription/whisper.py

@@ -0,0 +1,290 @@
+"""Isolated whisper.cpp wrapper producing Meeting Lab compact transcripts."""
+
+from __future__ import annotations
+
+import json
+import os
+import platform
+import re
+import shutil
+import subprocess
+import tempfile
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable, Sequence
+
+
+BACKEND = "whisper.cpp"
+RAW_FILENAME = "whisper_raw.json"
+TRANSCRIPT_FILENAME = "transcript.json"
+TEXT_FILENAME = "transcript.txt"
+METADATA_FILENAME = "runtime_metadata.json"
+DEFAULT_THREADS = "auto"
+
+
+class TranscriptionError(RuntimeError):
+    """Raised when parameters, Whisper execution, or output are invalid."""
+
+
+@dataclass(frozen=True)
+class TranscriptionResult:
+    output_dir: Path
+    raw_output: Path
+    transcript_json: Path
+    transcript_text: Path
+    runtime_metadata: Path
+    runtime_seconds: float
+
+
+def _logical_cpu_count() -> int:
+    """Return the available logical CPU count as a last-resort fallback."""
+    if hasattr(os, "sched_getaffinity"):
+        try:
+            count = len(os.sched_getaffinity(0))
+            if count > 0:
+                return count
+        except OSError:
+            pass
+    return os.cpu_count() or 1
+
+
+def _linux_physical_core_count() -> int | None:
+    affinity = None
+    if hasattr(os, "sched_getaffinity"):
+        try:
+            affinity = os.sched_getaffinity(0)
+        except OSError:
+            pass
+    cores: set[tuple[str, str]] = set()
+    for cpu_dir in Path("/sys/devices/system/cpu").glob("cpu[0-9]*"):
+        try:
+            cpu_number = int(cpu_dir.name[3:])
+            if affinity is not None and cpu_number not in affinity:
+                continue
+            topology = cpu_dir / "topology"
+            package = (topology / "physical_package_id").read_text().strip()
+            core = (topology / "core_id").read_text().strip()
+            cores.add((package, core))
+        except (OSError, ValueError):
+            continue
+    return len(cores) or None
+
+
+def _darwin_physical_core_count() -> int | None:
+    try:
+        completed = subprocess.run(
+            ("sysctl", "-n", "hw.physicalcpu"),
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        count = int(completed.stdout.strip())
+        return count if completed.returncode == 0 and count > 0 else None
+    except (OSError, ValueError):
+        return None
+
+
+def physical_core_count() -> int:
+    """Detect physical cores where supported, falling back to available threads."""
+    system = platform.system()
+    detected = _linux_physical_core_count() if system == "Linux" else None
+    if system == "Darwin":
+        detected = _darwin_physical_core_count()
+    return detected or _logical_cpu_count()
+
+
+def resolve_threads(
+    threads: str | int,
+    detector: Callable[[], int] = physical_core_count,
+) -> int:
+    if isinstance(threads, bool):
+        raise TranscriptionError("Threads must be 'auto' or a positive integer.")
+    if threads == "auto":
+        count = detector()
+    else:
+        try:
+            count = int(threads)
+        except (TypeError, ValueError) as exc:
+            raise TranscriptionError("Threads must be 'auto' or a positive integer.") from exc
+    if count <= 0:
+        raise TranscriptionError("Threads must be 'auto' or a positive integer.")
+    return count
+
+
+def _vulkan_support(raw: dict[str, Any]) -> bool | None:
+    system_info = raw.get("systeminfo")
+    if not isinstance(system_info, str) or "VULKAN" not in system_info.upper():
+        return None
+    return re.search(r"VULKAN\s*=\s*1", system_info, re.IGNORECASE) is not None
+
+
+def _json_object(path: Path) -> dict[str, Any]:
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise TranscriptionError(f"Cannot read Whisper JSON output {path}: {exc}") from exc
+    if not isinstance(data, dict):
+        raise TranscriptionError("Whisper JSON output must contain a top-level object.")
+    return data
+
+
+def compact_transcript(raw: dict[str, Any]) -> dict[str, Any]:
+    """Convert whisper.cpp JSON without linguistic cleanup or reordering."""
+    entries = raw.get("transcription")
+    if not isinstance(entries, list):
+        raise TranscriptionError("Whisper JSON output must contain a 'transcription' list.")
+
+    segments: list[dict[str, Any]] = []
+    for index, entry in enumerate(entries):
+        if not isinstance(entry, dict):
+            raise TranscriptionError(f"transcription[{index}] must be an object.")
+        offsets = entry.get("offsets")
+        if not isinstance(offsets, dict):
+            raise TranscriptionError(f"transcription[{index}].offsets must be an object.")
+        start_ms = offsets.get("from")
+        end_ms = offsets.get("to")
+        if not isinstance(start_ms, (int, float)) or isinstance(start_ms, bool):
+            raise TranscriptionError(f"transcription[{index}].offsets.from must be a number.")
+        if not isinstance(end_ms, (int, float)) or isinstance(end_ms, bool):
+            raise TranscriptionError(f"transcription[{index}].offsets.to must be a number.")
+        if end_ms < start_ms:
+            raise TranscriptionError(
+                f"transcription[{index}].offsets.to must be greater than or equal to offsets.from."
+            )
+        text_value = entry.get("text", "")
+        if not isinstance(text_value, str):
+            raise TranscriptionError(f"transcription[{index}].text must be a string.")
+        text = text_value.strip()
+        if text:
+            segments.append(
+                {
+                    "id": len(segments),
+                    "start": float(start_ms) / 1000.0,
+                    "end": float(end_ms) / 1000.0,
+                    "text": text,
+                }
+            )
+    return {"text": " ".join(item["text"] for item in segments), "segments": segments}
+
+
+def _timestamp(seconds: float) -> str:
+    milliseconds = int(round(seconds * 1000))
+    hours, remainder = divmod(milliseconds, 3_600_000)
+    minutes, remainder = divmod(remainder, 60_000)
+    secs, millis = divmod(remainder, 1000)
+    return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}"
+
+
+def transcript_text(transcript: dict[str, Any]) -> str:
+    lines = [
+        f"[{_timestamp(item['start'])} - {_timestamp(item['end'])}] {item['text']}"
+        for item in transcript["segments"]
+    ]
+    return "\n".join(lines) + ("\n" if lines else "")
+
+
+def _write_json(path: Path, value: dict[str, Any]) -> None:
+    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+def transcribe_audio(
+    audio_path: Path,
+    model_path: Path,
+    output_dir: Path,
+    language: str = "auto",
+    *,
+    executable: str = "whisper-cli",
+    threads: str | int = DEFAULT_THREADS,
+    thread_detector: Callable[[], int] = physical_core_count,
+    runner: Callable[..., subprocess.CompletedProcess[str]] = subprocess.run,
+    monotonic: Callable[[], float] = time.monotonic,
+    now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
+) -> TranscriptionResult:
+    """Run one whisper.cpp call and write raw, compact, text, and metadata outputs."""
+    audio_path = Path(audio_path)
+    model_path = Path(model_path)
+    output_dir = Path(output_dir)
+    if not audio_path.is_file():
+        raise TranscriptionError(f"Audio file does not exist: {audio_path}")
+    if not model_path.is_file():
+        raise TranscriptionError(f"Whisper model does not exist: {model_path}")
+    if not isinstance(language, str) or not language.strip():
+        raise TranscriptionError("Language must be a non-empty string.")
+    if not executable.strip():
+        raise TranscriptionError("Whisper executable must be a non-empty string.")
+    if output_dir.exists() and not output_dir.is_dir():
+        raise TranscriptionError(f"Output directory path is not a directory: {output_dir}")
+    thread_count = resolve_threads(threads, thread_detector)
+
+    output_dir.mkdir(parents=True, exist_ok=True)
+    raw_output = output_dir / RAW_FILENAME
+    started_at = now().astimezone(timezone.utc)
+    started = monotonic()
+
+    with tempfile.TemporaryDirectory(prefix=".whisper-", dir=output_dir) as temp_name:
+        temporary_prefix = Path(temp_name) / "whisper_raw"
+        command: Sequence[str] = (
+            executable,
+            "-m", str(model_path),
+            "-f", str(audio_path),
+            "-l", language.strip(),
+            "-t", str(thread_count),
+            "-fa",
+            "-oj",
+            "-of", str(temporary_prefix),
+        )
+        try:
+            completed = runner(command, check=False, capture_output=True, text=True)
+        except OSError as exc:
+            raise TranscriptionError(f"Could not start {BACKEND}: {exc}") from exc
+        runtime_seconds = monotonic() - started
+        temporary_raw = temporary_prefix.with_suffix(".json")
+        if temporary_raw.is_file():
+            shutil.copyfile(temporary_raw, raw_output)
+        if completed.returncode != 0:
+            detail = completed.stderr.strip() or completed.stdout.strip() or "no diagnostic output"
+            raise TranscriptionError(
+                f"{BACKEND} failed with exit code {completed.returncode}: {detail}"
+            )
+        if not raw_output.is_file():
+            raise TranscriptionError(f"{BACKEND} completed without producing JSON output.")
+
+    raw_data = _json_object(raw_output)
+    transcript = compact_transcript(raw_data)
+    transcript_json_path = output_dir / TRANSCRIPT_FILENAME
+    transcript_text_path = output_dir / TEXT_FILENAME
+    metadata_path = output_dir / METADATA_FILENAME
+    _write_json(transcript_json_path, transcript)
+    transcript_text_path.write_text(transcript_text(transcript), encoding="utf-8")
+    duration = max((item["end"] for item in transcript["segments"]), default=None)
+    metadata = {
+        "input_file": str(audio_path.resolve()),
+        "model": str(model_path.resolve()),
+        "backend": BACKEND,
+        "whisper_executable": executable,
+        "language": language.strip(),
+        "threads": thread_count,
+        "threads_option": str(threads),
+        "flash_attention": True,
+        "vulkan_support_detected": _vulkan_support(raw_data),
+        "duration_seconds": duration,
+        "runtime_seconds": runtime_seconds,
+        "timestamp": started_at.isoformat(),
+        "output_files": {
+            "whisper_raw": RAW_FILENAME,
+            "transcript_json": TRANSCRIPT_FILENAME,
+            "transcript_text": TEXT_FILENAME,
+            "runtime_metadata": METADATA_FILENAME,
+        },
+    }
+    _write_json(metadata_path, metadata)
+    return TranscriptionResult(
+        output_dir=output_dir,
+        raw_output=raw_output,
+        transcript_json=transcript_json_path,
+        transcript_text=transcript_text_path,
+        runtime_metadata=metadata_path,
+        runtime_seconds=runtime_seconds,
+    )

tests/test_audio_preparation.py

@@ -0,0 +1,230 @@
+import subprocess
+import tempfile
+import unittest
+import wave
+from pathlib import Path
+from unittest.mock import patch
+
+from src.meeting_lab.audio.preparation import (
+    DEFAULT_NORMALIZATION_FILTER,
+    DEFAULT_NORMALIZATION_METHOD,
+    AudioPreparationError,
+    prepare_audio,
+)
+
+
+def _write_wav(
+    path: Path, *, channels: int = 1, sample_rate: int = 16_000, sample_width: int = 2
+) -> None:
+    with wave.open(str(path), "wb") as recording:
+        recording.setnchannels(channels)
+        recording.setsampwidth(sample_width)
+        recording.setframerate(sample_rate)
+        recording.writeframes(b"\x00" * channels * sample_width * 32)
+
+
+def _successful_runner(commands: list[list[str]]):
+    def run(command, **kwargs):
+        commands.append(list(command))
+        _write_wav(Path(command[-1]))
+        return subprocess.CompletedProcess(command, 0, "", "")
+
+    return run
+
+
+class AudioPreparationTests(unittest.TestCase):
+    def test_supported_inputs_are_prepared_with_normalization_on_and_off(self) -> None:
+        for suffix in (".wav", ".flac", ".m4a"):
+            for normalization_enabled in (True, False):
+                with (
+                    self.subTest(
+                        suffix=suffix, normalization_enabled=normalization_enabled
+                    ),
+                    tempfile.TemporaryDirectory() as directory,
+                ):
+                    root = Path(directory)
+                    source = root / f"meeting{suffix}"
+                    if suffix == ".wav":
+                        _write_wav(source)
+                    else:
+                        source.write_bytes(b"original encoded audio")
+                    original = source.read_bytes()
+                    destination = root / "run" / "audio" / "prepared.wav"
+                    commands: list[list[str]] = []
+
+                    with patch(
+                        "src.meeting_lab.audio.preparation.shutil.which",
+                        return_value="/usr/bin/ffmpeg",
+                    ):
+                        result = prepare_audio(
+                            source,
+                            destination,
+                            normalization_enabled=normalization_enabled,
+                            runner=_successful_runner(commands),
+                        )
+
+                    self.assertEqual(source.read_bytes(), original)
+                    self.assertEqual(result.prepared_path, destination)
+                    with wave.open(str(destination), "rb") as recording:
+                        self.assertEqual(recording.getnchannels(), 1)
+                        self.assertEqual(recording.getframerate(), 16_000)
+                        self.assertEqual(recording.getsampwidth(), 2)
+                        self.assertEqual(recording.getcomptype(), "NONE")
+                    self.assertEqual(commands[0][commands[0].index("-ac") + 1], "1")
+                    self.assertEqual(commands[0][commands[0].index("-ar") + 1], "16000")
+                    self.assertEqual(
+                        commands[0][commands[0].index("-c:a") + 1], "pcm_s16le"
+                    )
+                    self.assertEqual("-af" in commands[0], normalization_enabled)
+                    if normalization_enabled:
+                        self.assertEqual(
+                            commands[0][commands[0].index("-af") + 1],
+                            DEFAULT_NORMALIZATION_FILTER,
+                        )
+                    self.assertEqual(
+                        result.normalization_enabled, normalization_enabled
+                    )
+
+    def test_normalization_defaults_to_on_and_explicit_on_matches(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "meeting.wav"
+            _write_wav(source)
+            commands: list[list[str]] = []
+            with patch(
+                "src.meeting_lab.audio.preparation.shutil.which",
+                return_value="/usr/bin/ffmpeg",
+            ):
+                default = prepare_audio(
+                    source, root / "default.wav", runner=_successful_runner(commands)
+                )
+                explicit = prepare_audio(
+                    source,
+                    root / "explicit.wav",
+                    normalization_enabled=True,
+                    runner=_successful_runner(commands),
+                )
+
+            self.assertTrue(default.normalization_enabled)
+            self.assertTrue(explicit.normalization_enabled)
+            self.assertEqual(
+                commands[0][commands[0].index("-af") + 1],
+                commands[1][commands[1].index("-af") + 1],
+            )
+
+    def test_noncanonical_wav_is_normalized(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "stereo-48k.wav"
+            _write_wav(source, channels=2, sample_rate=48_000)
+            destination = root / "prepared.wav"
+
+            with patch(
+                "src.meeting_lab.audio.preparation.shutil.which",
+                return_value="/usr/bin/ffmpeg",
+            ):
+                prepare_audio(source, destination, runner=_successful_runner([]))
+
+            with wave.open(str(destination), "rb") as recording:
+                self.assertEqual(
+                    (recording.getnchannels(), recording.getframerate()), (1, 16_000)
+                )
+
+    def test_ffmpeg_missing_has_actionable_error(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "meeting.flac"
+            source.write_bytes(b"audio")
+
+            with (
+                patch(
+                    "src.meeting_lab.audio.preparation.shutil.which", return_value=None
+                ),
+                self.assertRaisesRegex(AudioPreparationError, "not found on PATH"),
+            ):
+                prepare_audio(source, root / "prepared.wav")
+
+    def test_ffmpeg_failure_includes_diagnostic_and_preserves_source(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "meeting.m4a"
+            source.write_bytes(b"original")
+
+            def fail(command, **kwargs):
+                return subprocess.CompletedProcess(command, 1, "", "decoder exploded")
+
+            with (
+                patch(
+                    "src.meeting_lab.audio.preparation.shutil.which",
+                    return_value="/usr/bin/ffmpeg",
+                ),
+                self.assertRaisesRegex(AudioPreparationError, "decoder exploded"),
+            ):
+                prepare_audio(source, root / "prepared.wav", runner=fail)
+
+            self.assertEqual(source.read_bytes(), b"original")
+            self.assertFalse((root / "prepared.wav").exists())
+
+    def test_prepared_audio_metadata_is_traceable(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "unknown_meeting.flac"
+            source.write_bytes(b"source")
+            destination = root / "audio" / "prepared.wav"
+            with patch(
+                "src.meeting_lab.audio.preparation.shutil.which",
+                return_value="/usr/bin/ffmpeg",
+            ):
+                result = prepare_audio(
+                    source, destination, runner=_successful_runner([])
+                )
+
+            metadata = result.metadata()
+            self.assertEqual(metadata["original_source_name"], "unknown_meeting.flac")
+            self.assertEqual(metadata["original_format"], "flac")
+            self.assertEqual(
+                metadata["prepared_audio_path"], str(destination.resolve())
+            )
+            self.assertEqual(metadata["preparation_method"], "ffmpeg")
+            self.assertTrue(metadata["normalization_enabled"])
+            self.assertEqual(
+                metadata["normalization_method"], DEFAULT_NORMALIZATION_METHOD
+            )
+            self.assertEqual(
+                metadata["normalization_filter"], DEFAULT_NORMALIZATION_FILTER
+            )
+            self.assertEqual(
+                metadata["canonical_output"],
+                {
+                    "container": "wav",
+                    "codec": "pcm_s16le",
+                    "channels": 1,
+                    "sample_rate_hz": 16_000,
+                    "bits_per_sample": 16,
+                },
+            )
+
+    def test_disabled_normalization_metadata_has_no_method_or_filter(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "meeting.m4a"
+            source.write_bytes(b"source")
+            with patch(
+                "src.meeting_lab.audio.preparation.shutil.which",
+                return_value="/usr/bin/ffmpeg",
+            ):
+                result = prepare_audio(
+                    source,
+                    root / "prepared.wav",
+                    normalization_enabled=False,
+                    runner=_successful_runner([]),
+                )
+
+            metadata = result.metadata()
+            self.assertFalse(metadata["normalization_enabled"])
+            self.assertIsNone(metadata["normalization_method"])
+            self.assertIsNone(metadata["normalization_filter"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_diarization.py

@@ -0,0 +1,222 @@
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from src.meeting_lab.diarization.alignment import align_transcript, write_diarized_transcript
+from src.meeting_lab.diarization.backend import (
+    DiarizationError,
+    diarize_audio,
+    run_container_pyannote,
+    select_device,
+)
+
+
+class FakeCuda:
+    def __init__(self, available: bool, *, name: str = "Test GPU", failure=None):
+        self.available = available
+        self.name = name
+        self.failure = failure
+
+    def is_available(self):
+        return self.available
+
+    def get_device_name(self, index):
+        if self.failure:
+            raise self.failure
+        return self.name
+
+
+class FakeTorch:
+    def __init__(self, available: bool, *, failure=None):
+        self.cuda = FakeCuda(available, failure=failure)
+        self.probes = []
+
+    def device(self, name):
+        return name
+
+    def zeros(self, size, *, device):
+        self.probes.append(device)
+        if self.cuda.failure:
+            raise self.cuda.failure
+        return [0]
+
+
+class DeviceSelectionTests(unittest.TestCase):
+    def test_auto_selects_usable_gpu(self):
+        torch = FakeTorch(True)
+        self.assertEqual(select_device("auto", torch), ("cuda", "Test GPU"))
+        self.assertEqual(torch.probes, ["cuda"])
+
+    def test_auto_falls_back_to_cpu(self):
+        self.assertEqual(select_device("auto", FakeTorch(False)), ("cpu", None))
+        self.assertEqual(
+            select_device("auto", FakeTorch(True, failure=RuntimeError("probe"))),
+            ("cpu", None),
+        )
+
+    def test_explicit_cpu_does_not_probe_gpu(self):
+        torch = FakeTorch(True)
+        self.assertEqual(select_device("cpu", torch), ("cpu", None))
+        self.assertEqual(torch.probes, [])
+
+    def test_explicit_gpu_fails_when_unavailable(self):
+        with self.assertRaisesRegex(DiarizationError, "GPU is unavailable"):
+            select_device("gpu", FakeTorch(False))
+
+
+class AlignmentTests(unittest.TestCase):
+    def test_exclusive_overlap_assigns_anonymous_speakers(self):
+        transcript = {
+            "text": "Original unchanged text.",
+            "segments": [
+                {"id": 0, "start": 0.0, "end": 4.0, "text": "Hallo"},
+                {"id": 1, "start": 4.0, "end": 6.0, "text": "Antwort"},
+            ],
+        }
+        original = json.loads(json.dumps(transcript))
+        turns = [
+            {"start": 0.0, "end": 3.0, "speaker_id": "SPEAKER_00"},
+            {"start": 3.0, "end": 6.0, "speaker_id": "SPEAKER_01"},
+        ]
+
+        derived = align_transcript(transcript, turns)
+
+        self.assertEqual(transcript, original)
+        self.assertEqual(derived["segments"][0]["speaker_id"], "SPEAKER_00")
+        self.assertEqual(derived["segments"][0]["speaker_overlap_seconds"], 3.0)
+        self.assertEqual(derived["segments"][1]["speaker_id"], "SPEAKER_01")
+        self.assertIn("SPEAKER_00: Hallo", derived["text"])
+        self.assertTrue(derived["speaker_labels_anonymous"])
+        self.assertEqual(derived["alignment_source"], "exclusive_diarization")
+
+    def test_speaker_aware_transcript_is_a_separate_artifact(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "transcript.json"
+            turns = root / "exclusive_turns.json"
+            source_text = json.dumps(
+                {"text": "Original", "segments": [{"start": 0, "end": 1, "text": "Hi"}]}
+            )
+            source.write_text(source_text, encoding="utf-8")
+            turns.write_text(
+                json.dumps([{"start": 0, "end": 1, "speaker_id": "SPEAKER_07"}]),
+                encoding="utf-8",
+            )
+
+            json_path, text_path = write_diarized_transcript(source, turns, root / "derived")
+
+            self.assertEqual(source.read_text(encoding="utf-8"), source_text)
+            self.assertNotEqual(json_path, source)
+            self.assertIn("SPEAKER_07", json_path.read_text(encoding="utf-8"))
+            self.assertIn("SPEAKER_07", text_path.read_text(encoding="utf-8"))
+
+
+class ContainerAdapterTests(unittest.TestCase):
+    def test_container_configuration_and_metadata_do_not_persist_token(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio = root / "audio.wav"
+            output = root / "output"
+            audio.write_bytes(b"audio")
+            observed = {}
+
+            def fake_runner(command, **kwargs):
+                observed["command"] = command
+                output.mkdir(exist_ok=True)
+                metadata = {
+                    "backend": "pyannote.audio",
+                    "model": "pyannote/speaker-diarization-community-1",
+                    "requested_device_mode": "gpu",
+                    "actual_device": "cuda",
+                    "credentials_persisted": False,
+                }
+                (output / "metadata.json").write_text(json.dumps(metadata))
+                return SimpleNamespace(returncode=0, stdout="ok", stderr="")
+
+            with patch.dict("os.environ", {"HF_TOKEN": "secret-token"}):
+                result = run_container_pyannote(
+                    audio,
+                    output,
+                    "gpu",
+                    image="test/image",
+                    container_args=("--device=/dev/test",),
+                    runner=fake_runner,
+                    uid_getter=lambda: 2345,
+                    gid_getter=lambda: 3456,
+                )
+
+            command = observed["command"]
+            shell_command = command[-1]
+            persisted = "".join(
+                path.read_text(encoding="utf-8")
+                for path in output.iterdir()
+                if path.is_file()
+            )
+            self.assertNotIn("secret-token", persisted)
+            self.assertNotIn("secret-token", command)
+            self.assertIn("HF_TOKEN", command)
+            self.assertIn("chown -R 2345:3456 /output", shell_command)
+            self.assertIn("chmod -R u+rwX /output", shell_command)
+            self.assertNotIn("1000:1000", shell_command)
+            device_index = command.index("--device=/dev/test")
+            self.assertLess(device_index, command.index("test/image"))
+            self.assertFalse(result.metadata["credentials_persisted"])
+            self.assertEqual(result.metadata["runtime_adapter"], "container")
+            self.assertTrue(
+                all(os.access(path, os.W_OK) for path in (output, *output.rglob("*")))
+            )
+
+    def test_unwritable_container_artifact_is_rejected_before_metadata_update(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio = root / "audio.wav"
+            output = root / "output"
+            audio.write_bytes(b"audio")
+
+            def fake_runner(command, **kwargs):
+                output.mkdir(exist_ok=True)
+                metadata = output / "metadata.json"
+                metadata.write_text("{}", encoding="utf-8")
+                metadata.chmod(0o444)
+                return SimpleNamespace(returncode=0, stdout="", stderr="")
+
+            with patch(
+                "src.meeting_lab.diarization.backend.os.access",
+                side_effect=lambda path, mode: Path(path).name != "metadata.json",
+            ):
+                with self.assertRaisesRegex(DiarizationError, "not writable"):
+                    run_container_pyannote(
+                        audio,
+                        output,
+                        "cpu",
+                        image="test/image",
+                        runner=fake_runner,
+                    )
+
+    def test_orchestrator_dispatches_runtime(self):
+        with patch(
+            "src.meeting_lab.diarization.backend.run_container_pyannote"
+        ) as container:
+            diarize_audio(
+                Path("audio.wav"),
+                Path("out"),
+                "cpu",
+                runtime="container",
+                container_image="image",
+                container_args=("--arg",),
+            )
+        container.assert_called_once_with(
+            Path("audio.wav"),
+            Path("out"),
+            "cpu",
+            image="image",
+            container_args=("--arg",),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_direct_protocol.py

@@ -0,0 +1,321 @@
+import json
+import tempfile
+import unittest
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import requests
+
+from scripts import run_direct_protocol
+from src.meeting_lab.llm import ollama
+from src.meeting_lab.llm.ollama import OllamaError, OllamaGeneration
+from src.meeting_lab.protocol.generate_direct_protocol import (
+    DirectProtocolError,
+    generate_direct_protocol,
+    load_compact_transcript,
+)
+
+
+VALID_CONTEXT = """schema_version: "1"
+meeting:
+  meeting_id: "test-meeting"
+  title: "Test Meeting"
+  language: "de"
+participants: []
+mentioned_people: []
+organization:
+  departments: []
+known_entities: {}
+"""
+
+
+def write_transcript(path: Path, text: str = "Wir besprechen den Projektstatus.") -> None:
+    path.write_text(json.dumps({"text": text, "segments": []}), encoding="utf-8")
+
+
+def generation(text: str = "# Meeting Protocol\n\n## Status\nUnveraendert.") -> OllamaGeneration:
+    return OllamaGeneration(
+        raw_response={
+            "response": text,
+            "done": True,
+            "done_reason": "stop",
+            "prompt_eval_count": 123,
+            "eval_count": 17,
+            "prompt_eval_duration": 1000,
+            "eval_duration": 2000,
+            "total_duration": 4000,
+        },
+        text=text,
+        client_wall_time_seconds=0.25,
+    )
+
+
+class TranscriptLoadingTests(unittest.TestCase):
+    def test_valid_transcript_is_accepted(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "transcript.json"
+            write_transcript(path)
+            self.assertEqual(load_compact_transcript(path), "Wir besprechen den Projektstatus.")
+
+    def test_missing_top_level_text_is_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "transcript.json"
+            path.write_text('{"segments": []}', encoding="utf-8")
+            with self.assertRaisesRegex(DirectProtocolError, "top-level 'text'"):
+                load_compact_transcript(path)
+
+    def test_empty_transcript_is_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "transcript.json"
+            write_transcript(path, "  \n")
+            with self.assertRaisesRegex(DirectProtocolError, "non-empty string"):
+                load_compact_transcript(path)
+
+    def test_malformed_json_is_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "transcript.json"
+            path.write_text("{", encoding="utf-8")
+            with self.assertRaisesRegex(DirectProtocolError, "not valid JSON"):
+                load_compact_transcript(path)
+
+
+class GeneratorTests(unittest.TestCase):
+    def test_prompt_requires_contextual_discussion_density_without_transcript_replay(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            transcript = Path(directory) / "transcript.json"
+            write_transcript(transcript)
+            result = generate_direct_protocol(
+                transcript,
+                model_check=Mock(return_value={}),
+                generation_call=Mock(return_value=generation()),
+            )
+
+        self.assertIn("vollständiges, strukturiertes", result.exact_prompt)
+        self.assertIn("relevante Diskussionsverläufe", result.exact_prompt)
+        self.assertIn("unterschiedliche Positionen", result.exact_prompt)
+        self.assertIn("Entscheidungsgrundlagen", result.exact_prompt)
+        self.assertIn("nicht am Meeting teilgenommen haben", result.exact_prompt)
+        self.assertIn("keine reine Wiedergabe des Transkripts", result.exact_prompt)
+        self.assertIn("nicht unnötig durch Wiederholungen", result.exact_prompt)
+
+    def test_optional_context_absent_and_generation_called_once(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            transcript = Path(directory) / "transcript.json"
+            write_transcript(transcript)
+            check = Mock(return_value={"model": "qwen3.6:35B-A3B"})
+            call = Mock(return_value=generation())
+
+            result = generate_direct_protocol(
+                transcript,
+                model_check=check,
+                generation_call=call,
+            )
+
+        self.assertIn("Kein Meeting-Kontext", result.exact_prompt)
+        self.assertEqual(check.call_count, 1)
+        self.assertEqual(call.call_count, 1)
+        self.assertEqual(call.call_args.args[1], "qwen3.6:35B-A3B")
+        self.assertEqual(call.call_args.kwargs["num_ctx"], 32768)
+        self.assertEqual(result.runtime_metadata["request_count"], 1)
+        self.assertEqual(result.runtime_metadata["prompt_token_count"], 123)
+        self.assertFalse(result.runtime_metadata["think"])
+        self.assertEqual(result.runtime_metadata["temperature"], 0.0)
+
+    def test_valid_context_is_loaded_and_rendered(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = root / "transcript.json"
+            context = root / "context.yaml"
+            write_transcript(transcript)
+            context.write_text(VALID_CONTEXT, encoding="utf-8")
+            result = generate_direct_protocol(
+                transcript,
+                context,
+                model_check=Mock(return_value={}),
+                generation_call=Mock(return_value=generation()),
+            )
+
+        self.assertIn("MEETING CONTEXT V1", result.exact_prompt)
+        self.assertIn("Test Meeting", result.exact_prompt)
+
+    def test_invalid_context_is_rejected_before_network_calls(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = root / "transcript.json"
+            context = root / "context.yaml"
+            write_transcript(transcript)
+            context.write_text("schema_version: wrong", encoding="utf-8")
+            check = Mock()
+            call = Mock()
+            with self.assertRaisesRegex(ValueError, "schema_version"):
+                generate_direct_protocol(
+                    transcript,
+                    context,
+                    model_check=check,
+                    generation_call=call,
+                )
+
+        check.assert_not_called()
+        call.assert_not_called()
+
+
+class OllamaTests(unittest.TestCase):
+    def test_unavailable_endpoint_failure(self) -> None:
+        with patch.object(ollama.requests, "get", side_effect=requests.ConnectionError("down")):
+            with self.assertRaisesRegex(OllamaError, "not reachable"):
+                ollama.require_model("http://127.0.0.1:11434", "model")
+
+    def test_missing_model_failure(self) -> None:
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {"models": [{"name": "other:model"}]}
+        with patch.object(ollama.requests, "get", return_value=response):
+            with self.assertRaisesRegex(OllamaError, "not installed"):
+                ollama.require_model("http://127.0.0.1:11434", "model")
+
+    def test_request_settings_and_raw_response(self) -> None:
+        raw = {"response": "# Meeting Protocol", "done": True}
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = raw
+        with patch.object(ollama.requests, "post", return_value=response) as post:
+            result = ollama.generate_once(
+                "http://localhost:11434",
+                "qwen3.8:27b",
+                "prompt",
+                timeout=30,
+                num_ctx=32768,
+                num_predict=8192,
+            )
+
+        self.assertEqual(post.call_count, 1)
+        payload = post.call_args.kwargs["json"]
+        self.assertEqual(payload["model"], "qwen3.8:27b")
+        self.assertEqual(payload["options"]["temperature"], 0.0)
+        self.assertEqual(payload["options"]["num_ctx"], 32768)
+        self.assertFalse(payload["think"])
+        self.assertFalse(payload["stream"])
+        self.assertEqual(result.raw_response, raw)
+
+    def test_malformed_response_failure_without_retry(self) -> None:
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {"message": "missing response"}
+        with patch.object(ollama.requests, "post", return_value=response) as post:
+            with self.assertRaisesRegex(OllamaError, "no string 'response'"):
+                ollama.generate_once("url", "model", "prompt", timeout=1, num_ctx=1, num_predict=1)
+        self.assertEqual(post.call_count, 1)
+
+    def test_empty_response_failure(self) -> None:
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {"response": "  "}
+        with patch.object(ollama.requests, "post", return_value=response):
+            with self.assertRaisesRegex(OllamaError, "empty protocol"):
+                ollama.generate_once("url", "model", "prompt", timeout=1, num_ctx=1, num_predict=1)
+
+
+class DirectProtocolCliTests(unittest.TestCase):
+    def test_artifacts_are_preserved_and_protocol_is_untouched(self) -> None:
+        protocol_text = "# Meeting Protocol\n\nExact output.  \n"
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = root / "source.json"
+            context = root / "source.yaml"
+            write_transcript(transcript)
+            context.write_text(VALID_CONTEXT, encoding="utf-8")
+            args = run_direct_protocol.parse_args(
+                [str(transcript), "--context", str(context), "--output-root", str(root / "runs")]
+            )
+            with patch.object(
+                run_direct_protocol,
+                "generate_direct_protocol",
+                return_value=type("Result", (), {
+                    "protocol_text": protocol_text,
+                    "exact_prompt": "exact prompt\n",
+                    "transcript_input": "selected transcript\n",
+                    "raw_response": {"response": protocol_text},
+                    "runtime_metadata": {"request_count": 1},
+                })(),
+            ) as generator:
+                code, run_dir, protocol_path = run_direct_protocol.run(args)
+
+            self.assertEqual(code, 0)
+            self.assertEqual(generator.call_count, 1)
+            self.assertEqual(protocol_path.read_text(encoding="utf-8"), protocol_text)
+            self.assertEqual(
+                (run_dir / "protocol/exact_prompt.txt").read_text(encoding="utf-8"),
+                "exact prompt\n",
+            )
+            self.assertEqual(
+                (run_dir / "protocol/transcript_input.txt").read_text(encoding="utf-8"),
+                "selected transcript\n",
+            )
+            self.assertEqual(
+                json.loads((run_dir / "protocol/raw_response.json").read_text())["response"],
+                protocol_text,
+            )
+            self.assertEqual(
+                json.loads((run_dir / "protocol/runtime_metadata.json").read_text())["request_count"],
+                1,
+            )
+            self.assertTrue((run_dir / "transcript/transcript.json").is_file())
+            self.assertTrue((run_dir / "context/meeting_context.yaml").is_file())
+            self.assertTrue((run_dir / "input_manifest.json").is_file())
+            self.assertEqual(json.loads((run_dir / "run_metadata.json").read_text())["status"], "completed")
+
+    def test_unique_run_directories_do_not_overwrite(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            fixed = datetime(2026, 8, 20, 12, 0, 0)
+            first = run_direct_protocol.create_unique_run_dir(root, "meeting", lambda: fixed)
+            marker = first / "keep.txt"
+            marker.write_text("keep", encoding="utf-8")
+            second = run_direct_protocol.create_unique_run_dir(root, "meeting", lambda: fixed)
+            self.assertEqual(first.name, "meeting_20260820_120000")
+            self.assertEqual(second.name, "meeting_20260820_120000_01")
+            self.assertEqual(marker.read_text(encoding="utf-8"), "keep")
+
+    def test_failure_after_directory_creation_preserves_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = run_direct_protocol.parse_args(
+                [str(root / "missing.json"), "--output-root", str(root / "runs")]
+            )
+            code, run_dir, protocol_path = run_direct_protocol.run(args)
+
+            metadata = json.loads((run_dir / "run_metadata.json").read_text())
+        self.assertEqual(code, 2)
+        self.assertIsNone(protocol_path)
+        self.assertEqual(metadata["status"], "failed")
+        self.assertIn("does not exist", metadata["failure"])
+
+    def test_semantic_pipeline_functions_are_never_invoked(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = root / "source.json"
+            write_transcript(transcript)
+            args = run_direct_protocol.parse_args(
+                [str(transcript), "--output-root", str(root / "runs")]
+            )
+            fake_result = type("Result", (), {
+                "protocol_text": "# Meeting Protocol",
+                "exact_prompt": "prompt",
+                "raw_response": {"response": "# Meeting Protocol"},
+                "runtime_metadata": {},
+            })()
+            with (
+                patch("src.meeting_lab.extraction.extract_chunks.extract_input") as extraction,
+                patch("src.meeting_lab.consolidation.consolidate_facts.call_ollama") as consolidation,
+                patch.object(run_direct_protocol, "generate_direct_protocol", return_value=fake_result),
+            ):
+                code, _run_dir, _protocol_path = run_direct_protocol.run(args)
+
+        self.assertEqual(code, 0)
+        extraction.assert_not_called()
+        consolidation.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_meeting_context.py

@@ -11,8 +11,10 @@ from src.meeting_lab.extraction.extract_chunks import (
 )
 from src.meeting_lab.models.meeting_context import (
     MeetingContextValidationError,
+    create_meeting_context,
     load_meeting_context,
     render_meeting_context_for_prompt,
+    serialize_meeting_context_yaml,
     validate_meeting_context,
 )
 
@@ -70,6 +72,38 @@ class MeetingContextTests(unittest.TestCase):
         with self.assertRaisesRegex(MeetingContextValidationError, "invalid value"):
             validate_meeting_context(data)
 
+    def test_missing_participant_attendance_defaults_to_present(self) -> None:
+        data = copy.deepcopy(self.context.data)
+        del data["participants"][0]["attendance_status"]
+
+        context = create_meeting_context(data)
+
+        self.assertEqual(
+            context.data["participants"][0]["attendance_status"], "present"
+        )
+
+    def test_explicit_mentioned_only_is_preserved(self) -> None:
+        data = copy.deepcopy(self.context.data)
+        data["mentioned_people"][0]["attendance_status"] = "mentioned_only"
+
+        context = create_meeting_context(data)
+
+        self.assertEqual(
+            context.data["mentioned_people"][0]["attendance_status"],
+            "mentioned_only",
+        )
+        self.assertIn(
+            "Mentioned but absent people:", render_meeting_context_for_prompt(context)
+        )
+
+    def test_mentioned_only_person_cannot_be_a_diarized_speaker(self) -> None:
+        data = copy.deepcopy(self.context.data)
+        mentioned_id = data["mentioned_people"][0]["person_id"]
+        data["speaker_mappings"] = {"SPEAKER_00": mentioned_id}
+
+        with self.assertRaisesRegex(MeetingContextValidationError, "unknown participant"):
+            validate_meeting_context(data)
+
     def test_prompt_representation_is_deterministic(self) -> None:
         first = render_meeting_context_for_prompt(self.context)
         second = render_meeting_context_for_prompt(self.context)
@@ -192,6 +226,51 @@ class MeetingContextTests(unittest.TestCase):
         self.assertNotIn("responsible: Björn", prompt_context)
         self.assertNotIn("responsible: Jovana", prompt_context)
 
+    def test_existing_context_without_speaker_mappings_remains_valid(self) -> None:
+        self.assertEqual(self.context.speaker_mappings, {})
+        self.assertIsNone(self.context.participant_for_speaker("SPEAKER_00"))
+
+    def test_explicit_speaker_mapping_is_authoritative(self) -> None:
+        data = copy.deepcopy(self.context.data)
+        participant = data["participants"][0]
+        data["speaker_mappings"] = {"SPEAKER_03": participant["participant_id"]}
+
+        context = create_meeting_context(data)
+        rendered = render_meeting_context_for_prompt(context)
+
+        self.assertEqual(
+            context.participant_for_speaker("SPEAKER_03")["participant_id"],
+            participant["participant_id"],
+        )
+        self.assertIsNone(context.participant_for_speaker("SPEAKER_04"))
+        self.assertIn("Confirmed diarization speaker mappings (authoritative)", rendered)
+        self.assertIn("Unmapped SPEAKER_XX labels must remain anonymous", rendered)
+
+    def test_speaker_mapping_must_reference_existing_participant(self) -> None:
+        data = copy.deepcopy(self.context.data)
+        data["speaker_mappings"] = {"SPEAKER_00": "unknown-person"}
+        with self.assertRaisesRegex(MeetingContextValidationError, "unknown participant"):
+            validate_meeting_context(data)
+
+    def test_speaker_mapping_label_must_use_pyannote_shape(self) -> None:
+        data = copy.deepcopy(self.context.data)
+        data["speaker_mappings"] = {
+            "Martin": data["participants"][0]["participant_id"]
+        }
+        with self.assertRaisesRegex(MeetingContextValidationError, "speaker label"):
+            validate_meeting_context(data)
+
+    def test_generated_context_yaml_is_deterministic_and_round_trips(self) -> None:
+        first = serialize_meeting_context_yaml(self.context)
+        second = serialize_meeting_context_yaml(self.context)
+        self.assertEqual(first, second)
+
+        SCRATCH_DIR.mkdir(exist_ok=True)
+        path = SCRATCH_DIR / "generated_context.yaml"
+        path.write_text(first, encoding="utf-8")
+        loaded = load_meeting_context(path)
+        self.assertEqual(loaded.data, self.context.data)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_mvp_api.py

@@ -0,0 +1,310 @@
+import json
+import shutil
+import subprocess
+import tempfile
+import unittest
+from dataclasses import replace
+from pathlib import Path
+from unittest.mock import patch
+
+from scripts import run_mvp_meeting as cli
+from src.meeting_lab.audio import PreparedAudio
+from src.meeting_lab.models.meeting_context import load_meeting_context
+from src.meeting_lab.orchestration import mvp as mvp_api
+from src.meeting_lab.orchestration.mvp import MvpMeetingConfig, MvpRunResult
+from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
+from src.meeting_lab.transcription.whisper import TranscriptionError, TranscriptionResult
+
+
+def context_data():
+    return {
+        "schema_version": "1",
+        "meeting": {
+            "meeting_id": "programmatic-test",
+            "title": "Programmatic Test",
+            "language": "de",
+            "date": None,
+            "objective": "API prüfen",
+            "notes": "",
+        },
+        "participants": [
+            {
+                "participant_id": "person-1",
+                "display_name": "Test Person",
+                "aliases": [],
+                "role": "Projektleitung",
+                "department": None,
+                "attendance_status": "present",
+                "notes": None,
+            }
+        ],
+        "speaker_mappings": {"SPEAKER_00": "person-1"},
+        "mentioned_people": [],
+        "organization": {"name": "Example", "departments": []},
+        "known_entities": {},
+        "context_rules": {"do_not_infer_responsibilities": True},
+    }
+
+
+def fake_transcribe(audio, model, output, language, **kwargs):
+    output.mkdir(parents=True, exist_ok=True)
+    raw = output / "whisper_raw.json"
+    transcript = output / "transcript.json"
+    text = output / "transcript.txt"
+    metadata = output / "runtime_metadata.json"
+    raw.write_text('{"transcription": []}\n', encoding="utf-8")
+    transcript.write_text(
+        json.dumps(
+            {
+                "text": "Ein kurzer Besprechungstext.",
+                "segments": [
+                    {
+                        "id": 0,
+                        "start": 0.0,
+                        "end": 1.0,
+                        "text": "Ein kurzer Besprechungstext.",
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    text.write_text("Ein kurzer Besprechungstext.\n", encoding="utf-8")
+    metadata.write_text('{"runtime_seconds": 0.1}\n', encoding="utf-8")
+    return TranscriptionResult(output, raw, transcript, text, metadata, 0.1)
+
+
+def fake_prepare(source, destination, **kwargs):
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copyfile(source, destination)
+    return PreparedAudio(source, source.suffix.removeprefix("."), destination, "ffmpeg", "ffmpeg")
+
+
+def fake_protocol(transcript, context, **kwargs):
+    rendered_context = load_meeting_context(context)
+    assert rendered_context.meeting_id == "programmatic-test"
+    return DirectProtocolResult(
+        protocol_text="# Meeting Protocol\n",
+        exact_prompt="prompt",
+        model_metadata={"model": kwargs["model"]},
+        runtime_metadata={"request_count": 1},
+        raw_response={"response": "# Meeting Protocol\n"},
+    )
+
+
+class MvpApiTests(unittest.TestCase):
+    def config(self, root: Path):
+        audio = root / "meeting.wav"
+        model = root / "model.bin"
+        audio.write_bytes(b"audio")
+        model.write_bytes(b"model")
+        return MvpMeetingConfig(
+            audio_file=audio,
+            whisper_model=model,
+            output_root=root / "runs",
+            model="test:model",
+        )
+
+    def test_programmatic_context_is_persisted_without_source_yaml(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            events = []
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
+                patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare),
+                patch.object(
+                    mvp_api, "generate_direct_protocol", side_effect=fake_protocol
+                ) as protocol_generator,
+                patch.object(subprocess, "run") as subprocess_run,
+            ):
+                result = mvp_api.run_mvp_meeting(
+                    config, meeting_context=context_data(), progress_sink=events.append
+                )
+
+            self.assertEqual(result.exit_code, 0)
+            context_path = result.run_dir / "context/meeting_context.yaml"
+            self.assertTrue(context_path.is_file())
+            persisted = load_meeting_context(context_path)
+            self.assertEqual(persisted.meeting_id, "programmatic-test")
+            self.assertEqual(persisted.speaker_mappings, {"SPEAKER_00": "person-1"})
+            subprocess_run.assert_not_called()
+            self.assertEqual(
+                [(event.stage, event.status) for event in events],
+                [
+                    ("preparing", "started"),
+                    ("preparing", "completed"),
+                    ("transcription", "started"),
+                    ("transcription", "completed"),
+                    ("protocol_generation", "started"),
+                    ("protocol_generation", "completed"),
+                    ("completed", "completed"),
+                ],
+            )
+            self.assertTrue(all(event.progress is None for event in events))
+            self.assertEqual(protocol_generator.call_args.kwargs["num_ctx"], 32_768)
+            self.assertEqual(
+                protocol_generator.call_args.kwargs["safe_input_token_budget"],
+                29_000,
+            )
+
+    def test_protocol_only_regeneration_reuses_diarized_artifacts(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            run_dir = root / "existing-run"
+            diarization_dir = run_dir / "diarization"
+            diarization_dir.mkdir(parents=True)
+            source = diarization_dir / "transcript_diarized.json"
+            source.write_text(
+                json.dumps(
+                    {
+                        "text": "SPEAKER_00: Existing statement.\n",
+                        "segments": [
+                            {
+                                "start": 0.0,
+                                "end": 1.0,
+                                "speaker_id": "SPEAKER_00",
+                                "text": "Existing statement.",
+                            }
+                        ],
+                        "speaker_labels_anonymous": True,
+                    }
+                ),
+                encoding="utf-8",
+            )
+            source_before = source.read_bytes()
+            mapped_context = context_data()
+            mapped_context["speaker_mappings"] = {"SPEAKER_00": "person-1"}
+            with (
+                patch.object(mvp_api, "prepare_audio") as preparation,
+                patch.object(mvp_api, "transcribe_audio") as transcription,
+                patch.object(mvp_api, "diarize_audio") as diarization,
+                patch.object(
+                    mvp_api, "generate_direct_protocol", side_effect=fake_protocol
+                ) as protocol,
+            ):
+                result = mvp_api.regenerate_mvp_protocol(
+                    run_dir,
+                    meeting_context=mapped_context,
+                    model="qwen3.8:27b",
+                    protocol_num_ctx=32_768,
+                    protocol_safe_input_token_budget=29_000,
+                )
+
+            self.assertEqual(result.exit_code, 0)
+            self.assertEqual(result.protocol_path, run_dir / "protocol.md")
+            preparation.assert_not_called()
+            transcription.assert_not_called()
+            diarization.assert_not_called()
+            self.assertEqual(protocol.call_args.args[0], source)
+            self.assertEqual(protocol.call_args.kwargs["num_ctx"], 32_768)
+            self.assertEqual(source.read_bytes(), source_before)
+            persisted = load_meeting_context(run_dir / "context/meeting_context.yaml")
+            self.assertEqual(persisted.speaker_mappings, {"SPEAKER_00": "person-1"})
+
+    def test_failure_emits_terminal_failure_event(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            events = []
+            with patch.object(
+                mvp_api,
+                "transcribe_audio",
+                side_effect=TranscriptionError("stopped"),
+            ), patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare):
+                result = mvp_api.run_mvp_meeting(config, progress_sink=events.append)
+
+            self.assertEqual(result.exit_code, 2)
+            self.assertEqual(events[-1].stage, "failed")
+            self.assertEqual(events[-1].status, "failed")
+            self.assertIn("transcription", events[-1].message)
+            metadata = json.loads((result.run_dir / "run_metadata.json").read_text())
+            self.assertEqual(metadata["status"], "failed")
+
+    def test_cli_defaults_and_wrapper_delegate_without_subprocess(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            args = cli.parse_args(
+                [str(config.audio_file), "--whisper-model", str(config.whisper_model)]
+            )
+            expected = MvpRunResult(0, root / "run", root / "run/protocol.md")
+            with patch.object(cli, "run_mvp_meeting", return_value=expected) as api:
+                actual = cli.run(args, context_override=context_data())
+
+            self.assertEqual(actual, (0, expected.run_dir, expected.protocol_path))
+            delegated = api.call_args.args[0]
+            self.assertEqual(delegated.diarization, "off")
+            self.assertEqual(delegated.language, "de")
+            self.assertEqual(delegated.whisper_executable, "whisper-cli")
+            self.assertEqual(delegated.ffmpeg_executable, "ffmpeg")
+            self.assertTrue(delegated.audio_normalization)
+            self.assertEqual(delegated.protocol_num_ctx, 32_768)
+            self.assertEqual(delegated.protocol_safe_input_token_budget, 29_000)
+            self.assertEqual(api.call_args.kwargs["meeting_context"], context_data())
+
+    def test_cli_explicit_audio_normalization_values_are_propagated(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            config = self.config(root)
+            enabled = cli.config_from_args(
+                cli.parse_args(
+                    [
+                        str(config.audio_file),
+                        "--whisper-model",
+                        str(config.whisper_model),
+                        "--audio-normalization",
+                    ]
+                )
+            )
+            disabled = cli.config_from_args(
+                cli.parse_args(
+                    [
+                        str(config.audio_file),
+                        "--whisper-model",
+                        str(config.whisper_model),
+                        "--no-audio-normalization",
+                    ]
+                )
+            )
+
+            self.assertTrue(enabled.audio_normalization)
+            self.assertFalse(disabled.audio_normalization)
+
+    def test_transcription_receives_prepared_wav_for_encoded_inputs(self):
+        for suffix in (".flac", ".m4a"):
+            with self.subTest(suffix=suffix), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                config = self.config(root)
+                encoded = config.audio_file.with_suffix(suffix)
+                config.audio_file.rename(encoded)
+                config = replace(config, audio_file=encoded)
+                received = []
+
+                def capture_transcribe(audio, *args, received_paths=received, **kwargs):
+                    received_paths.append(audio)
+                    return fake_transcribe(audio, *args, **kwargs)
+
+                with (
+                    patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare),
+                    patch.object(mvp_api, "transcribe_audio", side_effect=capture_transcribe),
+                    patch.object(mvp_api, "generate_direct_protocol", side_effect=fake_protocol),
+                ):
+                    result = mvp_api.run_mvp_meeting(
+                        config, meeting_context=context_data()
+                    )
+
+                self.assertEqual(result.exit_code, 0)
+                self.assertEqual(received, [result.run_dir / "audio" / "prepared.wav"])
+                manifest = json.loads(
+                    (result.run_dir / "audio" / "input_manifest.json").read_text()
+                )
+                self.assertEqual(manifest["format"], suffix.removeprefix("."))
+                self.assertEqual(
+                    manifest["prepared_audio"]["prepared_audio_path"],
+                    str((result.run_dir / "audio" / "prepared.wav").resolve()),
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mvp_orchestrator.py

@@ -0,0 +1,444 @@
+import json
+import shutil
+import tempfile
+import unittest
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from scripts import run_mvp_meeting
+from src.meeting_lab.audio import PreparedAudio
+from src.meeting_lab.audio.preparation import (
+    DEFAULT_NORMALIZATION_FILTER,
+    DEFAULT_NORMALIZATION_METHOD,
+)
+from src.meeting_lab.orchestration import mvp as mvp_api
+from src.meeting_lab.protocol.generate_direct_protocol import DirectProtocolResult
+from src.meeting_lab.diarization.backend import DiarizationResult
+from src.meeting_lab.transcription.whisper import TranscriptionError, TranscriptionResult
+
+
+VALID_CONTEXT = """schema_version: "1"
+meeting:
+  meeting_id: "mvp-test"
+  title: "MVP Test"
+  language: "de"
+participants: []
+mentioned_people: []
+organization:
+  departments: []
+known_entities: {}
+"""
+
+
+def protocol_result(model: str = "chosen:model") -> DirectProtocolResult:
+    text = "# Protokoll\n\nUnverändert.  \n"
+    return DirectProtocolResult(
+        protocol_text=text,
+        exact_prompt="exact prompt\n",
+        model_metadata={"model": model},
+        runtime_metadata={"model": model, "request_count": 1, "client_wall_time_seconds": 0.5},
+        raw_response={"response": text, "done": True},
+        transcript_input="selected transcript\n",
+    )
+
+
+def fake_transcribe(
+    audio_path: Path,
+    model_path: Path,
+    output_dir: Path,
+    language: str,
+    *,
+    executable: str,
+    threads: str | int,
+) -> TranscriptionResult:
+    output_dir.mkdir(parents=True, exist_ok=True)
+    raw = output_dir / "whisper_raw.json"
+    transcript = output_dir / "transcript.json"
+    text = output_dir / "transcript.txt"
+    metadata = output_dir / "runtime_metadata.json"
+    raw.write_text('{"transcription": []}\n', encoding="utf-8")
+    transcript.write_text(
+        json.dumps(
+            {
+                "text": "Besprechungstext.",
+                "segments": [
+                    {"id": 0, "start": 0.0, "end": 1.0, "text": "Besprechungstext."}
+                ],
+            }
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    text.write_text("Besprechungstext.\n", encoding="utf-8")
+    metadata.write_text('{"runtime_seconds": 1.25}\n', encoding="utf-8")
+    return TranscriptionResult(output_dir, raw, transcript, text, metadata, 1.25)
+
+
+def fake_diarize(audio_path, output_dir, device_mode, **kwargs):
+    output_dir.mkdir(parents=True, exist_ok=True)
+    metadata = {
+        "backend": "pyannote.audio",
+        "model": "pyannote/speaker-diarization-community-1",
+        "requested_device_mode": device_mode,
+        "actual_device": "cuda",
+        "device_name": "Fake GPU",
+        "runtime_seconds": 2.5,
+        "speaker_count": 1,
+        "credentials_persisted": False,
+    }
+    paths = {
+        "metadata": output_dir / "metadata.json",
+        "ordinary": output_dir / "diarization.rttm",
+        "exclusive": output_dir / "exclusive_diarization.rttm",
+        "turns": output_dir / "turns.json",
+        "exclusive_turns": output_dir / "exclusive_turns.json",
+    }
+    paths["metadata"].write_text(json.dumps(metadata), encoding="utf-8")
+    paths["ordinary"].write_text("", encoding="utf-8")
+    paths["exclusive"].write_text("", encoding="utf-8")
+    paths["turns"].write_text("[]", encoding="utf-8")
+    paths["exclusive_turns"].write_text(
+        json.dumps([{"start": 0, "end": 10, "speaker_id": "SPEAKER_00"}]),
+        encoding="utf-8",
+    )
+    return DiarizationResult(
+        output_dir,
+        paths["metadata"],
+        paths["ordinary"],
+        paths["exclusive"],
+        paths["turns"],
+        paths["exclusive_turns"],
+        metadata,
+    )
+
+
+def fake_prepare(source, destination, **kwargs):
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copyfile(source, destination)
+    normalization_enabled = kwargs.get("normalization_enabled", True)
+    return PreparedAudio(
+        source,
+        source.suffix.removeprefix("."),
+        destination,
+        "ffmpeg",
+        kwargs.get("ffmpeg_executable", "ffmpeg"),
+        normalization_enabled,
+        DEFAULT_NORMALIZATION_METHOD if normalization_enabled else None,
+        DEFAULT_NORMALIZATION_FILTER if normalization_enabled else None,
+    )
+
+
+class MvpOrchestratorTests(unittest.TestCase):
+    def setUp(self) -> None:
+        patcher = patch.object(mvp_api, "prepare_audio", side_effect=fake_prepare)
+        self.prepare_audio = patcher.start()
+        self.addCleanup(patcher.stop)
+
+    def create_inputs(self, root: Path) -> tuple[Path, Path, Path]:
+        audio = root / "team meeting.wav"
+        whisper_model = root / "ggml-model.bin"
+        context = root / "source-context.yaml"
+        audio.write_bytes(b"audio")
+        whisper_model.write_bytes(b"model")
+        context.write_text(VALID_CONTEXT, encoding="utf-8")
+        return audio, whisper_model, context
+
+    def args(self, root: Path, extra: list[str] | None = None):
+        audio, whisper_model, context = self.create_inputs(root)
+        values = [
+            str(audio),
+            "--whisper-model", str(whisper_model),
+            "--context", str(context),
+            "--output-root", str(root / "runs"),
+            "--language", "de",
+            "--model", "chosen:model",
+            "--ollama-endpoint", "http://ollama.test:11434",
+        ]
+        if extra:
+            values.extend(extra)
+        return run_mvp_meeting.parse_args(values)
+
+    def test_successful_full_orchestration_and_artifact_layout(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root)
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    return_value=protocol_result(),
+                ) as protocol,
+            ):
+                code, run_dir, protocol_path = run_mvp_meeting.run(args)
+
+            self.assertEqual(code, 0)
+            self.assertIsNotNone(run_dir)
+            self.assertEqual(protocol_path, run_dir / "protocol.md")
+            expected = {
+                "run_metadata.json",
+                "audio/input_manifest.json",
+                "audio/prepared.wav",
+                "audio/preparation_metadata.json",
+                "transcript/whisper_raw.json",
+                "transcript/transcript.json",
+                "transcript/transcript.txt",
+                "transcript/runtime_metadata.json",
+                "context/meeting_context.yaml",
+                "protocol/exact_prompt.txt",
+                "protocol/transcript_input.txt",
+                "protocol/raw_response.json",
+                "protocol/runtime_metadata.json",
+                "protocol.md",
+            }
+            self.assertTrue(all((run_dir / item).is_file() for item in expected))
+            metadata = json.loads((run_dir / "run_metadata.json").read_text())
+            self.assertEqual(metadata["status"], "completed")
+            self.assertEqual(metadata["model"], "chosen:model")
+            self.assertIsNone(metadata["failure"])
+            self.assertEqual(whisper.call_count, 1)
+            self.assertEqual(
+                whisper.call_args.args[0], run_dir / "audio" / "prepared.wav"
+            )
+            self.assertEqual(protocol.call_count, 1)
+            self.assertTrue(
+                self.prepare_audio.call_args.kwargs["normalization_enabled"]
+            )
+            preparation = json.loads(
+                (run_dir / "audio" / "preparation_metadata.json").read_text()
+            )
+            self.assertTrue(preparation["normalization_enabled"])
+
+    def test_cli_can_disable_audio_normalization_without_bypassing_preparation(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root, ["--no-audio-normalization"])
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    return_value=protocol_result(),
+                ),
+            ):
+                code, run_dir, _ = run_mvp_meeting.run(args)
+
+            self.assertEqual(code, 0)
+            self.prepare_audio.assert_called_once()
+            self.assertFalse(
+                self.prepare_audio.call_args.kwargs["normalization_enabled"]
+            )
+            self.assertEqual(
+                whisper.call_args.args[0], run_dir / "audio" / "prepared.wav"
+            )
+            preparation = json.loads(
+                (run_dir / "audio" / "preparation_metadata.json").read_text()
+            )
+            self.assertFalse(preparation["normalization_enabled"])
+            self.assertIsNone(preparation["normalization_method"])
+            self.assertIsNone(preparation["normalization_filter"])
+
+    def test_context_model_endpoint_and_whisper_options_are_forwarded(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(
+                root,
+                [
+                    "--whisper-executable",
+                    "/tools/whisper-cli",
+                    "--ffmpeg-executable",
+                    "/tools/ffmpeg",
+                    "--threads",
+                    "4",
+                ],
+            )
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe) as whisper,
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    return_value=protocol_result(),
+                ) as protocol,
+            ):
+                code, run_dir, _ = run_mvp_meeting.run(args)
+
+            self.assertEqual(code, 0)
+            self.assertEqual(whisper.call_args.args[3], "de")
+            self.assertEqual(whisper.call_args.kwargs["executable"], "/tools/whisper-cli")
+            self.assertEqual(whisper.call_args.kwargs["threads"], "4")
+            self.assertEqual(
+                self.prepare_audio.call_args.kwargs["ffmpeg_executable"],
+                "/tools/ffmpeg",
+            )
+            self.assertEqual(protocol.call_args.args[1], run_dir / "context/meeting_context.yaml")
+            self.assertEqual(protocol.call_args.kwargs["model"], "chosen:model")
+            self.assertEqual(
+                protocol.call_args.kwargs["endpoint"], "http://ollama.test:11434"
+            )
+
+    def test_diarization_is_off_by_default_and_preserves_protocol_input(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root)
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
+                patch.object(mvp_api, "diarize_audio") as diarization,
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    return_value=protocol_result(),
+                ) as protocol,
+            ):
+                code, run_dir, _ = run_mvp_meeting.run(args)
+
+            self.assertEqual(code, 0)
+            diarization.assert_not_called()
+            self.assertEqual(protocol.call_args.args[0], run_dir / "transcript/transcript.json")
+            self.assertFalse((run_dir / "diarization").exists())
+
+    def test_diarization_cli_propagates_and_uses_derived_protocol_input(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(
+                root,
+                [
+                    "--diarization", "gpu",
+                    "--diarization-runtime", "container",
+                    "--diarization-container-image", "rocm/test",
+                    "--diarization-container-arg=--device=/dev/kfd",
+                ],
+            )
+            with (
+                patch.object(
+                    mvp_api, "transcribe_audio", side_effect=fake_transcribe
+                ),
+                patch.object(
+                    mvp_api, "diarize_audio", side_effect=fake_diarize
+                ) as diarization,
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    return_value=protocol_result(),
+                ) as protocol,
+            ):
+                code, run_dir, _ = run_mvp_meeting.run(args)
+
+            self.assertEqual(code, 0)
+            self.assertEqual(diarization.call_args.args[2], "gpu")
+            self.assertEqual(
+                diarization.call_args.args[0], run_dir / "audio" / "prepared.wav"
+            )
+            self.assertEqual(diarization.call_args.kwargs["runtime"], "container")
+            self.assertEqual(
+                diarization.call_args.kwargs["container_args"], ("--device=/dev/kfd",)
+            )
+            derived = run_dir / "diarization/transcript_diarized.json"
+            self.assertEqual(protocol.call_args.args[0], derived)
+            self.assertIn("SPEAKER_00", derived.read_text(encoding="utf-8"))
+            self.assertEqual(
+                json.loads((run_dir / "transcript/transcript.json").read_text())["text"],
+                "Besprechungstext.",
+            )
+            run_metadata = json.loads((run_dir / "run_metadata.json").read_text())
+            self.assertTrue(run_metadata["diarization"]["enabled"])
+            self.assertNotIn("HF_TOKEN", json.dumps(run_metadata))
+
+    def test_whisper_failure_is_recorded_and_protocol_is_not_called(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root)
+            with (
+                patch.object(
+                    mvp_api,
+                    "transcribe_audio",
+                    side_effect=TranscriptionError("whisper stopped"),
+                ),
+                patch.object(mvp_api, "generate_direct_protocol") as protocol,
+            ):
+                code, run_dir, protocol_path = run_mvp_meeting.run(args)
+
+            metadata = json.loads((run_dir / "run_metadata.json").read_text())
+            self.assertEqual(code, 2)
+            self.assertIsNone(protocol_path)
+            self.assertEqual(metadata["status"], "failed")
+            self.assertEqual(metadata["failure"]["stage"], "whisper")
+            self.assertIn("whisper stopped", metadata["failure"]["message"])
+            protocol.assert_not_called()
+            self.assertTrue((run_dir / "audio/input_manifest.json").is_file())
+
+    def test_protocol_failure_preserves_transcript_and_failure_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root)
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    side_effect=ValueError("generation stopped"),
+                ),
+            ):
+                code, run_dir, protocol_path = run_mvp_meeting.run(args)
+
+            metadata = json.loads((run_dir / "run_metadata.json").read_text())
+            self.assertEqual(code, 2)
+            self.assertIsNone(protocol_path)
+            self.assertEqual(metadata["failure"]["stage"], "protocol")
+            self.assertTrue((run_dir / "transcript/transcript.json").is_file())
+            self.assertFalse((run_dir / "protocol.md").exists())
+
+    def test_unique_run_directories_do_not_overwrite(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            fixed = datetime(2026, 8, 21, 9, 15, 30)
+            first = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed)
+            marker = first / "keep.txt"
+            marker.write_text("keep", encoding="utf-8")
+            second = run_mvp_meeting.create_unique_run_dir(root, "team meeting", lambda: fixed)
+            self.assertEqual(first.name, "team_meeting_20260821_091530")
+            self.assertEqual(second.name, "team_meeting_20260821_091530_01")
+            self.assertEqual(marker.read_text(), "keep")
+
+    def test_semantic_pipeline_functions_are_never_invoked(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root)
+            with (
+                patch.object(mvp_api, "transcribe_audio", side_effect=fake_transcribe),
+                patch.object(
+                    mvp_api,
+                    "generate_direct_protocol",
+                    return_value=protocol_result(),
+                ),
+                patch("src.meeting_lab.chunking.chunk_transcript.build_chunks") as chunking,
+                patch("src.meeting_lab.extraction.extract_chunks.extract_input") as extraction,
+                patch("src.meeting_lab.consolidation.consolidate_facts.call_ollama") as consolidation,
+            ):
+                code, _, _ = run_mvp_meeting.run(args)
+
+            self.assertEqual(code, 0)
+            chunking.assert_not_called()
+            extraction.assert_not_called()
+            consolidation.assert_not_called()
+
+    def test_main_returns_nonzero_for_whisper_failure(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            args = self.args(root)
+            argv = [
+                str(args.audio_file),
+                "--whisper-model", str(args.whisper_model),
+                "--context", str(args.context),
+                "--output-root", str(args.output_root),
+            ]
+            with patch.object(
+                mvp_api,
+                "transcribe_audio",
+                side_effect=TranscriptionError("failed"),
+            ):
+                self.assertEqual(run_mvp_meeting.main(argv), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_protocol_transcript_input.py

@@ -0,0 +1,315 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from src.meeting_lab.diarization.alignment import diarized_transcript_text
+from src.meeting_lab.llm import ollama
+from src.meeting_lab.llm.ollama import OllamaGeneration
+from src.meeting_lab.protocol.direct_protocol_prompt import (
+    COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
+    build_direct_protocol_prompt,
+)
+from src.meeting_lab.protocol.generate_direct_protocol import (
+    DirectProtocolError,
+    estimate_input_tokens,
+    generate_direct_protocol,
+)
+from src.meeting_lab.protocol.transcript_input import compact_diarized_transcript
+
+
+def segments() -> list[dict[str, object]]:
+    return [
+        {"id": 0, "start": 0.0, "end": 1.0, "text": "First.", "speaker_id": "SPEAKER_01"},
+        {"id": 1, "start": 1.0, "end": 2.0, "text": "Second.", "speaker_id": "SPEAKER_01"},
+        {"id": 2, "start": 2.0, "end": 3.0, "text": "Third.", "speaker_id": "SPEAKER_04"},
+        {"id": 3, "start": 3.0, "end": 4.0, "text": "Unassigned.", "speaker_id": None},
+        {"id": 4, "start": 4.0, "end": 5.0, "text": "Last.", "speaker_id": "SPEAKER_01"},
+    ]
+
+
+def diarized_document(repetitions: int = 1) -> dict[str, object]:
+    source = segments() * repetitions
+    return {
+        "text": diarized_transcript_text(source),
+        "segments": source,
+        "speaker_labels_anonymous": True,
+        "alignment_source": "exclusive_diarization",
+    }
+
+
+def completed_generation() -> OllamaGeneration:
+    text = "# Meeting Protocol\n\nComplete."
+    return OllamaGeneration(
+        raw_response={
+            "response": text,
+            "done": True,
+            "done_reason": "stop",
+            "prompt_eval_count": 100,
+            "eval_count": 10,
+        },
+        text=text,
+        client_wall_time_seconds=0.1,
+    )
+
+
+class CompactDiarizedTranscriptTests(unittest.TestCase):
+    def test_adjacent_segments_group_and_transitions_remain_separate(self) -> None:
+        compact = compact_diarized_transcript(segments())
+
+        self.assertEqual(
+            [block.speaker_id for block in compact.blocks],
+            ["SPEAKER_01", "SPEAKER_04", "SPEAKER_UNASSIGNED", "SPEAKER_01"],
+        )
+        self.assertEqual(compact.blocks[0].segment_texts, ("First.", "Second."))
+        self.assertEqual(compact.blocks[-1].segment_texts, ("Last.",))
+        self.assertEqual(compact.text.count("SPEAKER_01:"), 2)
+
+    def test_every_segment_text_and_order_are_preserved(self) -> None:
+        source = segments()
+        compact = compact_diarized_transcript(source)
+
+        self.assertEqual(compact.source_segment_count, len(source))
+        self.assertEqual(compact.represented_segment_count, len(source))
+        self.assertEqual(
+            compact.segment_texts,
+            tuple(str(segment["text"]) for segment in source),
+        )
+        self.assertEqual(compact.segment_texts[0], "First.")
+        self.assertEqual(compact.segment_texts[-1], "Last.")
+        self.assertIn("SPEAKER_UNASSIGNED: Unassigned.", compact.text)
+
+    def test_compact_form_is_materially_smaller_than_per_segment_format(self) -> None:
+        source = [
+            {
+                "start": index,
+                "end": index + 1,
+                "text": "Repeated transcript content.",
+                "speaker_id": "SPEAKER_01",
+            }
+            for index in range(100)
+        ]
+
+        compact = compact_diarized_transcript(source).text
+        verbose = diarized_transcript_text(source)
+
+        self.assertLess(len(compact), len(verbose) * 0.6)
+
+
+class ProtocolInputBudgetTests(unittest.TestCase):
+    def _write(self, root: Path, document: dict[str, object]) -> Path:
+        path = root / "transcript.json"
+        path.write_text(json.dumps(document), encoding="utf-8")
+        return path
+
+    def test_token_estimate_uses_utf8_bytes_for_non_ascii_safety(self) -> None:
+        self.assertEqual(estimate_input_tokens("ä" * 44), 20)
+
+    def test_compact_diarized_representation_selected_within_budget(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            document = diarized_document()
+            transcript = self._write(root, document)
+            compact = compact_diarized_transcript(document["segments"]).text
+            budget = estimate_input_tokens(
+                build_direct_protocol_prompt(
+                    compact,
+                    instruction=COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
+                )
+            )
+            call = Mock(return_value=completed_generation())
+
+            result = generate_direct_protocol(
+                transcript,
+                safe_input_token_budget=budget,
+                model_check=Mock(return_value={}),
+                generation_call=call,
+            )
+
+        self.assertEqual(
+            result.runtime_metadata["selected_transcript_representation"],
+            "diarized_compact",
+        )
+        self.assertFalse(result.runtime_metadata["fallback_used"])
+        self.assertTrue(result.runtime_metadata["diarization_enabled"])
+        self.assertEqual(result.runtime_metadata["safe_input_token_budget"], budget)
+        self.assertEqual(result.transcript_input, compact)
+        self.assertEqual(call.call_count, 1)
+
+    def test_mapped_speakers_and_statements_reach_final_ollama_payload(self) -> None:
+        diarized = {
+            "text": "",
+            "segments": [
+                {
+                    "start": 0.0,
+                    "end": 1.0,
+                    "speaker_id": "SPEAKER_00",
+                    "text": "We will run the trial on Wednesday.",
+                },
+                {
+                    "start": 1.0,
+                    "end": 2.0,
+                    "speaker_id": "SPEAKER_01",
+                    "text": "I will prepare the raw materials before then.",
+                },
+                {
+                    "start": 2.0,
+                    "end": 3.0,
+                    "speaker_id": "SPEAKER_00",
+                    "text": "Good. Anna owns the material preparation.",
+                },
+            ],
+            "speaker_labels_anonymous": True,
+            "alignment_source": "exclusive_diarization",
+        }
+        context = {
+            "schema_version": "1",
+            "meeting": {
+                "meeting_id": "speaker-test",
+                "title": "Speaker test",
+                "language": "en",
+            },
+            "participants": [
+                {"participant_id": "martin", "display_name": "Martin"},
+                {"participant_id": "anna", "display_name": "Anna"},
+            ],
+            "speaker_mappings": {"SPEAKER_00": "martin", "SPEAKER_01": "anna"},
+            "mentioned_people": [],
+            "organization": {"departments": []},
+            "known_entities": {},
+        }
+        response = Mock()
+        response.raise_for_status.return_value = None
+        response.json.return_value = {"response": "# Meeting Protocol\n", "done": True}
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = self._write(root, diarized)
+            context_path = root / "context.yaml"
+            context_path.write_text(json.dumps(context), encoding="utf-8")
+            with patch.object(ollama.requests, "post", return_value=response) as post:
+                result = generate_direct_protocol(
+                    transcript,
+                    context_path,
+                    model="qwen3.8:27b",
+                    model_check=Mock(return_value={}),
+                )
+
+        prompt = post.call_args.kwargs["json"]["prompt"]
+        self.assertEqual(result.exact_prompt, prompt)
+        self.assertIn("- SPEAKER_00: Martin (participant_id: martin)", prompt)
+        self.assertIn("- SPEAKER_01: Anna (participant_id: anna)", prompt)
+        self.assertIn("SPEAKER_00: We will run the trial on Wednesday.", prompt)
+        self.assertIn(
+            "SPEAKER_01: I will prepare the raw materials before then.", prompt
+        )
+        self.assertIn("SPEAKER_00: Good. Anna owns the material preparation.", prompt)
+        self.assertNotIn("Martin: We will run the trial on Wednesday.", prompt)
+        self.assertNotIn("Anna: I will prepare the raw materials before then.", prompt)
+        self.assertIn("autoritativen SPEAKER_XX-zu-Teilnehmer-Zuordnungen", prompt)
+        self.assertIn("Ich-Zusage", prompt)
+        self.assertIn("nur erwähnten Personen", prompt)
+        self.assertIn("keine persönliche Verantwortung", prompt)
+
+    def test_plain_fallback_selected_when_diarized_compact_exceeds_budget(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            alternating_segments = [
+                {
+                    "id": index,
+                    "start": float(index),
+                    "end": float(index + 1),
+                    "text": "Word.",
+                    "speaker_id": f"SPEAKER_{index % 2:02d}",
+                }
+                for index in range(200)
+            ]
+            document = {
+                "text": diarized_transcript_text(alternating_segments),
+                "segments": alternating_segments,
+                "speaker_labels_anonymous": True,
+                "alignment_source": "exclusive_diarization",
+            }
+            transcript = self._write(root, document)
+            compact = compact_diarized_transcript(document["segments"]).text
+            plain = " ".join(str(segment["text"]) for segment in document["segments"])
+            compact_estimate = estimate_input_tokens(
+                build_direct_protocol_prompt(
+                    compact,
+                    instruction=COMPACT_DIARIZED_PROTOCOL_INSTRUCTION,
+                )
+            )
+            plain_estimate = estimate_input_tokens(build_direct_protocol_prompt(plain))
+            self.assertLess(plain_estimate, compact_estimate)
+
+            result = generate_direct_protocol(
+                transcript,
+                safe_input_token_budget=plain_estimate,
+                model_check=Mock(return_value={}),
+                generation_call=Mock(return_value=completed_generation()),
+            )
+
+        self.assertEqual(
+            result.runtime_metadata["selected_transcript_representation"],
+            "plain_transcript_fallback",
+        )
+        self.assertTrue(result.runtime_metadata["fallback_used"])
+        self.assertEqual(result.transcript_input, plain)
+        self.assertNotIn("SPEAKER_00", result.exact_prompt)
+        self.assertNotIn("SPEAKER_01", result.exact_prompt)
+        self.assertFalse(result.runtime_metadata["speaker_attribution_available"])
+        self.assertEqual(
+            result.runtime_metadata["speaker_attribution_loss_reason"],
+            "plain_transcript_fallback",
+        )
+
+    def test_oversized_plain_transcript_fails_before_any_network_call(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = self._write(
+                root,
+                {"text": "large input " * 1000, "segments": []},
+            )
+            model_check = Mock()
+            generation_call = Mock()
+
+            with self.assertRaisesRegex(
+                DirectProtocolError,
+                "No LLM request was made; silent truncation is not allowed",
+            ):
+                generate_direct_protocol(
+                    transcript,
+                    safe_input_token_budget=1,
+                    model_check=model_check,
+                    generation_call=generation_call,
+                )
+
+        model_check.assert_not_called()
+        generation_call.assert_not_called()
+
+    def test_existing_plain_path_and_metadata_remain_direct(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            transcript = self._write(
+                root,
+                {"text": "Plain original transcript.", "segments": []},
+            )
+            result = generate_direct_protocol(
+                transcript,
+                model_check=Mock(return_value={}),
+                generation_call=Mock(return_value=completed_generation()),
+            )
+
+        self.assertEqual(result.transcript_input, "Plain original transcript.")
+        self.assertEqual(
+            result.runtime_metadata["selected_transcript_representation"],
+            "plain_transcript",
+        )
+        self.assertFalse(result.runtime_metadata["fallback_used"])
+        self.assertFalse(result.runtime_metadata["diarization_enabled"])
+        self.assertIsNone(result.runtime_metadata["speaker_attribution_available"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_transcription.py

@@ -0,0 +1,221 @@
+import json
+import subprocess
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from scripts import transcribe_meeting
+from src.meeting_lab.transcription.whisper import (
+    TranscriptionError,
+    compact_transcript,
+    transcribe_audio,
+)
+
+
+RAW_RESULT = {
+    "params": {"language": "de"},
+    "transcription": [
+        {
+            "timestamps": {"from": "00:00:00,000", "to": "00:00:01,200"},
+            "offsets": {"from": 0, "to": 1200},
+            "text": " Hallo ",
+        },
+        {
+            "timestamps": {"from": "00:00:01,200", "to": "00:00:02,500"},
+            "offsets": {"from": 1200, "to": 2500},
+            "text": "Welt",
+        },
+    ],
+}
+
+
+class FakeRunner:
+    def __init__(self, *, returncode: int = 0, write_raw: bool = True) -> None:
+        self.returncode = returncode
+        self.write_raw = write_raw
+        self.commands: list[list[str]] = []
+
+    def __call__(self, command, **kwargs):
+        command = list(command)
+        self.commands.append(command)
+        if self.write_raw:
+            prefix = Path(command[command.index("-of") + 1])
+            prefix.with_suffix(".json").write_text(
+                json.dumps(RAW_RESULT, ensure_ascii=False), encoding="utf-8"
+            )
+        return subprocess.CompletedProcess(
+            command, self.returncode, stdout="", stderr="backend failure"
+        )
+
+
+class TranscriptionTests(unittest.TestCase):
+    def create_inputs(self, root: Path) -> tuple[Path, Path]:
+        audio = root / "meeting.wav"
+        model = root / "ggml-model.bin"
+        audio.write_bytes(b"audio")
+        model.write_bytes(b"model")
+        return audio, model
+
+    def test_missing_audio_file_is_rejected_before_execution(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            model = root / "model.bin"
+            model.write_bytes(b"model")
+            runner = FakeRunner()
+            with self.assertRaisesRegex(TranscriptionError, "Audio file does not exist"):
+                transcribe_audio(root / "missing.wav", model, root / "out", runner=runner)
+            self.assertEqual(runner.commands, [])
+
+    def test_invalid_parameters_are_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            with self.assertRaisesRegex(TranscriptionError, "model does not exist"):
+                transcribe_audio(audio, root / "missing.bin", root / "out")
+            with self.assertRaisesRegex(TranscriptionError, "Language must"):
+                transcribe_audio(audio, model, root / "out", "  ")
+            with self.assertRaisesRegex(TranscriptionError, "executable must"):
+                transcribe_audio(audio, model, root / "out", executable="")
+            with self.assertRaisesRegex(TranscriptionError, "Threads must"):
+                transcribe_audio(audio, model, root / "out", threads="zero")
+
+    def test_auto_threads_use_detected_physical_core_count_and_flash_attention(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            runner = FakeRunner()
+            result = transcribe_audio(
+                audio,
+                model,
+                root / "output",
+                runner=runner,
+                thread_detector=lambda: 6,
+            )
+
+            command = runner.commands[0]
+            self.assertEqual(command[command.index("-t") + 1], "6")
+            self.assertIn("-fa", command)
+            metadata = json.loads(result.runtime_metadata.read_text())
+            self.assertEqual(metadata["threads"], 6)
+            self.assertEqual(metadata["threads_option"], "auto")
+            self.assertTrue(metadata["flash_attention"])
+            self.assertEqual(metadata["whisper_executable"], "whisper-cli")
+
+    def test_explicit_thread_count_is_forwarded_unchanged(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            runner = FakeRunner()
+            result = transcribe_audio(
+                audio, model, root / "output", threads=12, runner=runner
+            )
+
+            command = runner.commands[0]
+            self.assertEqual(command[command.index("-t") + 1], "12")
+            metadata = json.loads(result.runtime_metadata.read_text())
+            self.assertEqual(metadata["threads"], 12)
+            self.assertEqual(metadata["threads_option"], "12")
+
+    def test_cli_forwards_explicit_thread_count_to_wrapper(self) -> None:
+        fake_result = Mock(
+            transcript_json=Path("output/transcript.json"), runtime_seconds=1.0
+        )
+        with patch.object(
+            transcribe_meeting, "transcribe_audio", return_value=fake_result
+        ) as transcribe:
+            code = transcribe_meeting.main(
+                [
+                    "meeting.wav",
+                    "--model", "model.bin",
+                    "--output-dir", "output",
+                    "--threads", "6",
+                ]
+            )
+
+        self.assertEqual(code, 0)
+        self.assertEqual(transcribe.call_args.kwargs["threads"], "6")
+
+    def test_compact_transcript_has_established_schema(self) -> None:
+        self.assertEqual(
+            compact_transcript(RAW_RESULT),
+            {
+                "text": "Hallo Welt",
+                "segments": [
+                    {"id": 0, "start": 0.0, "end": 1.2, "text": "Hallo"},
+                    {"id": 1, "start": 1.2, "end": 2.5, "text": "Welt"},
+                ],
+            },
+        )
+
+    def test_success_preserves_raw_output_and_creates_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            runner = FakeRunner()
+            times = iter([10.0, 12.25])
+            result = transcribe_audio(
+                audio,
+                model,
+                root / "output",
+                "de",
+                runner=runner,
+                monotonic=lambda: next(times),
+                now=lambda: datetime(2026, 8, 21, 8, 30, tzinfo=timezone.utc),
+            )
+
+            self.assertEqual(json.loads(result.raw_output.read_text()), RAW_RESULT)
+            self.assertEqual(json.loads(result.transcript_json.read_text())["text"], "Hallo Welt")
+            self.assertEqual(
+                result.transcript_text.read_text(),
+                "[00:00:00.000 - 00:00:01.200] Hallo\n"
+                "[00:00:01.200 - 00:00:02.500] Welt\n",
+            )
+            metadata = json.loads(result.runtime_metadata.read_text())
+            self.assertEqual(metadata["backend"], "whisper.cpp")
+            self.assertEqual(metadata["language"], "de")
+            self.assertEqual(metadata["duration_seconds"], 2.5)
+            self.assertEqual(metadata["runtime_seconds"], 2.25)
+            self.assertEqual(metadata["timestamp"], "2026-08-21T08:30:00+00:00")
+            self.assertEqual(metadata["input_file"], str(audio.resolve()))
+            self.assertEqual(metadata["model"], str(model.resolve()))
+
+    def test_output_paths_are_deterministic(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            output = root / "chosen-output"
+            result = transcribe_audio(audio, model, output, runner=FakeRunner())
+            self.assertEqual(result.raw_output, output / "whisper_raw.json")
+            self.assertEqual(result.transcript_json, output / "transcript.json")
+            self.assertEqual(result.transcript_text, output / "transcript.txt")
+            self.assertEqual(result.runtime_metadata, output / "runtime_metadata.json")
+            self.assertEqual(
+                set(json.loads(result.runtime_metadata.read_text())["output_files"].values()),
+                {"whisper_raw.json", "transcript.json", "transcript.txt", "runtime_metadata.json"},
+            )
+
+    def test_backend_failure_raises_and_preserves_available_raw_output(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            output = root / "output"
+            with self.assertRaisesRegex(TranscriptionError, "exit code 7.*backend failure"):
+                transcribe_audio(audio, model, output, runner=FakeRunner(returncode=7))
+            self.assertEqual(json.loads((output / "whisper_raw.json").read_text()), RAW_RESULT)
+            self.assertFalse((output / "transcript.json").exists())
+            self.assertFalse((output / "runtime_metadata.json").exists())
+
+    def test_missing_backend_output_is_reported(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            audio, model = self.create_inputs(root)
+            with self.assertRaisesRegex(TranscriptionError, "without producing JSON"):
+                transcribe_audio(
+                    audio, model, root / "output", runner=FakeRunner(write_raw=False)
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()