Add post-diarization speaker review workflow

PROJECT_KNOWLEDGE.md

@@ -81,7 +81,8 @@ Imported audio
        (optionally normalize loudness; default on)
     -> transcribe with whisper.cpp and large-v3-turbo
     -> optionally diarize with pyannote.audio Community-1
-    -> generate a protocol directly from the full transcript
+    -> when diarized, wait for optional speaker review before explicit protocol generation
+    -> otherwise generate a protocol directly from the full transcript
     -> review and edit by a human
 ```
 
@@ -121,7 +122,10 @@ Speaker selectors filter participants using a snapshot of all current widget
 selections, falling back to saved mappings before first interaction. Filtering
 retains existing assignments; clearing a selection releases the participant.
 Completeness counts and warnings cover detected speakers only and do not gate
-protocol regeneration or change mapping persistence.
+protocol generation or change mapping persistence. A diarized run persists the
+`awaiting_speaker_review` checkpoint after diarization; an empty, partial or
+complete mapping can then explicitly generate a protocol from the existing
+artifacts. Failed generation preserves that checkpoint and saved mappings for retry.
 
 ## Progress Contract
 

README.md

@@ -35,7 +35,8 @@ Audio file
     -> FFmpeg preparation (mono, 16 kHz PCM WAV; normalization optional)
     -> whisper.cpp transcription (large-v3-turbo)
     -> optional pyannote.audio Community-1 diarization
-    -> direct full-transcript protocol generation
+    -> optional speaker review and explicit protocol generation (when diarized)
+       or direct full-transcript protocol generation (without diarization)
     -> human review and export
 ```
 
@@ -212,6 +213,22 @@ From the Meeting Assistant repository, start the UI with:
 PYTHONPATH=src:../meeting-lab .venv/bin/streamlit run src/mka/ui/streamlit_app.py
 ```
 
+### North launcher
+
+In the post-diarization Assistant worktree on North, use the checked-in
+launcher instead of creating another virtual environment:
+
+```bash
+./run_north.sh
+```
+
+It reuses North's validated Assistant virtual environment and external
+Whisper/Ollama/Docker-ROCm runtime while loading Meeting Assistant from this
+worktree and Meeting Lab from
+`/opt/git-projekts/meeting-lab`. `HF_TOKEN`, when present, is
+passed through to the disposable diarization container without being stored by
+the launcher.
+
 Streamlit opens `http://localhost:8501` by default. Installing the sibling
 Meeting Lab project supplies its runtime requirements such as PyYAML and
 Requests. Optional diarization dependencies are needed only when diarization

run_north.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+assistant_root="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)"
+assistant_src="$assistant_root/src"
+lab_root=/opt/git-projekts/meeting-lab
+north_venv=/opt/git-projekts/meeting-assistant/.venv
+whisper_executable=/home/martin/whisper.cpp/whisper.cpp/build/bin/whisper-cli
+whisper_model=/home/martin/whisper.cpp/whisper.cpp/models/ggml-large-v3-turbo.bin
+
+require_directory() {
+    if [[ ! -d "$1" ]]; then
+        printf 'Required directory is missing: %s\n' "$1" >&2
+        exit 1
+    fi
+}
+
+require_file() {
+    if [[ ! -f "$1" ]]; then
+        printf 'Required file is missing: %s\n' "$1" >&2
+        exit 1
+    fi
+}
+
+require_executable() {
+    if [[ ! -x "$1" ]]; then
+        printf 'Required executable is missing or not executable: %s\n' "$1" >&2
+        exit 1
+    fi
+}
+
+require_directory "$assistant_src"
+require_file "$assistant_src/mka/ui/streamlit_app.py"
+require_directory "$lab_root"
+require_file "$lab_root/src/meeting_lab/__init__.py"
+require_directory "$north_venv"
+require_executable "$north_venv/bin/streamlit"
+require_executable "$whisper_executable"
+require_file "$whisper_model"
+if ! command -v docker >/dev/null 2>&1; then
+    printf 'Required executable is not on PATH: docker\n' >&2
+    exit 1
+fi
+
+export MKA_WHISPER_EXECUTABLE="$whisper_executable"
+export MKA_WHISPER_MODEL="$whisper_model"
+export MKA_PROTOCOL_MODEL=qwen3.8:27b
+export MKA_OLLAMA_ENDPOINT=http://127.0.0.1:11434
+export MKA_DIARIZATION_MODE=auto
+export MKA_DIARIZATION_RUNTIME=container
+export MKA_DIARIZATION_CONTAINER_IMAGE=rocm/pytorch:rocm7.2.1_ubuntu24.04_py3.12_pytorch_release_2.9.1
+export MKA_DIARIZATION_CONTAINER_ARGS='["--device=/dev/kfd","--device=/dev/dri","--group-add","video"]'
+
+# Keep feature sources ahead of the reused venv's editable installations.
+export PYTHONPATH="$assistant_src:$lab_root"
+
+exec "$north_venv/bin/streamlit" run "$assistant_src/mka/ui/streamlit_app.py"

src/mka/application/meeting_service.py

@@ -92,6 +92,8 @@ class ProcessingOutcome:
     failed_stage: str | None = None
     error_message: str | None = None
     speaker_attribution_available: bool | None = None
+    awaiting_speaker_review: bool = False
+    detected_speaker_count: int | None = None
 
 
 @dataclass(frozen=True)
@@ -249,6 +251,7 @@ class MeetingProcessingService:
                 "diarization_runtime": self.settings.diarization_runtime,
                 "diarization_container_image": (self.settings.diarization_container_image),
                 "diarization_container_args": self.settings.diarization_container_args,
+                "stop_after_diarization": options.diarization_enabled,
             }
         )
         current_stage: str | None = None
@@ -299,6 +302,15 @@ class MeetingProcessingService:
                 failed_stage=failed_stage or current_stage or "preparing",
                 error_message=failure_message,
             )
+        if result.protocol_path is None:
+            return ProcessingOutcome(
+                succeeded=True,
+                run_dir=result.run_dir,
+                original_protocol=None,
+                protocol_path=None,
+                awaiting_speaker_review=True,
+                detected_speaker_count=self._detected_speaker_count(result.run_dir),
+            )
         protocol_path = Path(result.protocol_path)
         return ProcessingOutcome(
             succeeded=True,
@@ -401,6 +413,13 @@ class MeetingProcessingService:
         context_data["speaker_mappings"] = dict(sorted(speaker_mappings.items()))
         self._apply_glossary(context_data)
         context = self.meeting_lab.create_context(context_data)
+        # A checkpoint mapping is user input, not a derived protocol artifact.  Save it
+        # before inference so a failed generation remains retryable with the same review.
+        if context_path.is_symlink():
+            context_path.unlink()
+        context_path.write_text(
+            yaml.safe_dump(context.data, allow_unicode=True, sort_keys=False), encoding="utf-8"
+        )
 
         def relay(event: Any) -> None:
             self._update_timer(event)
@@ -432,6 +451,7 @@ class MeetingProcessingService:
             self.timer.finish_run()
             raise
         self.timer.finish_run()
+        self._persist_timing(Path(result.run_dir))
         protocol_path = Path(result.protocol_path)
         return ProcessingOutcome(
             succeeded=True,
@@ -473,6 +493,18 @@ class MeetingProcessingService:
         value = metadata.get("speaker_attribution_available")
         return value if isinstance(value, bool) else None
 
+    @staticmethod
+    def _detected_speaker_count(run_dir: Path | None) -> int | None:
+        if run_dir is None:
+            return None
+        metadata_path = Path(run_dir) / "run_metadata.json"
+        try:
+            metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return None
+        value = metadata.get("diarization", {}).get("speaker_count")
+        return value if isinstance(value, int) else None
+
     @staticmethod
     def _read_failure(run_dir: Path | None) -> dict[str, str]:
         if run_dir is None:

src/mka/ui/streamlit_app.py

@@ -554,7 +554,15 @@ def _render_result() -> None:
             st.caption("Intermediate artifacts and run metadata were preserved here.")
         return
 
-    st.success("Processing completed. Review the generated protocol before use.")
+    awaiting_review = getattr(outcome, "awaiting_speaker_review", False)
+    if awaiting_review:
+        count = getattr(outcome, "detected_speaker_count", None)
+        detected = f"{count} speakers detected" if count is not None else "speakers detected"
+        st.info(
+            f"Diarization complete — {detected}. Assign speakers if desired, then generate the protocol."
+        )
+    else:
+        st.success("Processing completed. Review the generated protocol before use.")
     st.caption(f"Run artifacts: {outcome.run_dir}")
     if outcome.speaker_attribution_available is False:
         st.warning(
@@ -563,17 +571,18 @@ def _render_result() -> None:
         )
     if message := st.session_state.pop("speaker_mapping_message", None):
         st.success(message)
-    with st.expander("Original generated protocol", expanded=False):
-        st.markdown(outcome.original_protocol or "")
-    edited = st.text_area(
-        "Editable protocol",
-        key="edited_protocol",
-        height=500,
-        help="The original protocol.md remains unchanged.",
-    )
-    if st.button("Save edited protocol", type="primary"):
-        path = MeetingProcessingService.save_edited_protocol(outcome.run_dir, edited)
-        st.success(f"Saved edited protocol to {path}")
+    if not awaiting_review:
+        with st.expander("Original generated protocol", expanded=False):
+            st.markdown(outcome.original_protocol or "")
+        edited = st.text_area(
+            "Editable protocol",
+            key="edited_protocol",
+            height=500,
+            help="The original protocol.md remains unchanged.",
+        )
+        if st.button("Save edited protocol", type="primary"):
+            path = MeetingProcessingService.save_edited_protocol(outcome.run_dir, edited)
+            st.success(f"Saved edited protocol to {path}")
 
     try:
         service = MeetingProcessingService(AppSettings.from_environment(), MeetingLabGateway())
@@ -590,7 +599,7 @@ def _render_result() -> None:
     if duplicate_assignments:
         st.error("Each participant can be assigned to only one detected speaker.")
     if st.button(
-        "Regenerate protocol with confirmed speakers",
+        "Generate protocol" if awaiting_review else "Regenerate protocol with confirmed speakers",
         disabled=duplicate_assignments,
         type="primary",
     ):
@@ -610,7 +619,7 @@ def _render_result() -> None:
                     performance_profile=selected_profile,
                 ),
                 states,
-                "Starting protocol regeneration",
+                "Starting protocol generation" if awaiting_review else "Starting protocol regeneration",
             )
         except (OSError, RuntimeError, ValueError) as exc:
             _remember_regeneration_timing(
@@ -618,13 +627,13 @@ def _render_result() -> None:
                 "Protocol regeneration failed",
                 service.timer.snapshot(),
             )
-            st.error(f"Protocol regeneration failed: {exc}")
+            st.error(f"Protocol generation failed: {exc}")
         else:
             _remember_regeneration_timing(states, message, timing)
             st.session_state.outcome = regenerated
             _queue_edited_protocol(regenerated.original_protocol or "")
             st.session_state.speaker_mapping_message = (
-                "Speaker mappings saved and protocol regenerated."
+                "Speaker mappings saved and protocol generated."
             )
             st.rerun()
 

tests/test_alpha_pair.py

@@ -104,9 +104,12 @@ def test_paired_anonymous_generation_and_mapped_regeneration(
         ProcessingOptions(diarization_enabled=True, performance_profile=profile),
     )
     assert outcome.succeeded
+    assert outcome.awaiting_speaker_review
     assert prepare.call_args.kwargs["normalization_enabled"] is True
     assert transcribe.call_args.args[3] == language
     root = outcome.run_dir
+    assert calls == []
+    service.regenerate_protocol(root, {}, performance_profile=profile)
     first = root / "protocol/generations/001"
     before = {p.name: p.read_bytes() for p in first.iterdir()}
     original = (root / "diarization/transcript_diarized.json").read_bytes()

tests/test_meeting_service.py

@@ -6,6 +6,7 @@ from types import SimpleNamespace
 from typing import Any
 
 import pytest
+import yaml
 
 from mka.application.config import AppSettings
 from mka.application.meeting_service import (
@@ -99,6 +100,24 @@ class FakeMeetingLab:
                 )
             )
             return SimpleNamespace(exit_code=2, run_dir=self.run_dir, protocol_path=None)
+        if config["stop_after_diarization"]:
+            write_speaker_review_artifacts(self.run_dir)
+            (self.run_dir / "run_metadata.json").write_text(
+                json.dumps(
+                    {
+                        "status": "awaiting_speaker_review",
+                        "diarization": {"speaker_count": 2},
+                    }
+                ),
+                encoding="utf-8",
+            )
+            progress_sink(
+                SimpleNamespace(
+                    stage="diarization", status="completed", elapsed_seconds=0.3,
+                    progress=None, message=None,
+                )
+            )
+            return SimpleNamespace(exit_code=0, run_dir=self.run_dir, protocol_path=None)
         protocol = self.run_dir / "protocol.md"
         protocol.write_text("# Generated protocol\n", encoding="utf-8")
         return SimpleNamespace(exit_code=0, run_dir=self.run_dir, protocol_path=protocol)
@@ -387,7 +406,7 @@ def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) ->
     audio.write_bytes(b"audio")
     events = []
 
-    service.process(
+    outcome = service.process(
         audio,
         meeting(),
         participants(),
@@ -397,6 +416,9 @@ def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) ->
 
     assert gateway.config_values is not None
     assert gateway.config_values["diarization"] == "gpu"
+    assert gateway.config_values["stop_after_diarization"] is True
+    assert outcome.awaiting_speaker_review is True
+    assert outcome.detected_speaker_count == 2
     assert [(event.stage, event.status) for event in events[:2]] == [
         ("preparing", "started"),
         ("preparing", "completed"),
@@ -404,6 +426,33 @@ def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) ->
     assert events[2].progress == 0.25
 
 
+def test_diarization_checkpoint_defers_protocol_but_disabled_run_does_not(tmp_path: Path) -> None:
+    service, gateway = make_service(tmp_path)
+    audio = tmp_path / "meeting.wav"
+    audio.write_bytes(b"audio")
+
+    checkpoint = service.process(
+        audio, meeting(), participants(), ProcessingOptions(diarization_enabled=True)
+    )
+    assert checkpoint.succeeded
+    assert checkpoint.awaiting_speaker_review
+    assert checkpoint.original_protocol is None
+    assert checkpoint.protocol_path is None
+
+    normal_root = tmp_path / "without-diarization"
+    normal_root.mkdir()
+    normal_service, normal_gateway = make_service(normal_root)
+    normal_audio = normal_root / "meeting.wav"
+    normal_audio.write_bytes(b"audio")
+    normal = normal_service.process(
+        normal_audio, meeting(), participants(), ProcessingOptions(diarization_enabled=False)
+    )
+    assert normal.succeeded
+    assert not normal.awaiting_speaker_review
+    assert normal.protocol_path is not None
+    assert normal_gateway.config_values["stop_after_diarization"] is False
+
+
 def test_process_propagates_ordered_diarization_container_args(tmp_path: Path) -> None:
     service, gateway = make_service(tmp_path)
     service.settings = replace(
@@ -537,6 +586,49 @@ def test_protocol_regeneration_allows_unmapped_and_rejects_duplicate_participant
         )
 
 
+@pytest.mark.parametrize(
+    "mappings",
+    [
+        {"SPEAKER_00": "martin"},
+        {"SPEAKER_00": "martin", "SPEAKER_01": "anna"},
+    ],
+)
+def test_protocol_generation_uses_partial_and_complete_confirmed_mappings(
+    tmp_path: Path, mappings: dict[str, str]
+) -> None:
+    service, gateway = make_service(tmp_path)
+    write_speaker_review_artifacts(gateway.run_dir)
+
+    outcome = service.regenerate_protocol(gateway.run_dir, mappings)
+
+    assert outcome.succeeded
+    assert gateway.regeneration is not None
+    assert gateway.regeneration["meeting_context"].data["speaker_mappings"] == mappings
+
+
+def test_failed_protocol_generation_keeps_checkpoint_mapping_for_retry(tmp_path: Path) -> None:
+    service, gateway = make_service(tmp_path)
+    source = write_speaker_review_artifacts(gateway.run_dir)
+    original_source = source.read_bytes()
+    original_regeneration = gateway.regenerate_protocol
+
+    def fail_regeneration(*args, **kwargs):
+        raise RuntimeError("model unavailable")
+
+    gateway.regenerate_protocol = fail_regeneration
+    with pytest.raises(RuntimeError, match="model unavailable"):
+        service.regenerate_protocol(gateway.run_dir, {"SPEAKER_00": "martin"})
+
+    persisted = yaml.safe_load(
+        (gateway.run_dir / "context" / "meeting_context.yaml").read_text(encoding="utf-8")
+    )
+    assert persisted["speaker_mappings"] == {"SPEAKER_00": "martin"}
+    gateway.regenerate_protocol = original_regeneration
+    retry = service.regenerate_protocol(gateway.run_dir, {"SPEAKER_00": "martin"})
+    assert retry.succeeded
+    assert source.read_bytes() == original_source
+
+
 def test_fallback_attribution_loss_is_read_from_runtime_metadata(tmp_path: Path) -> None:
     run_dir = tmp_path / "run"
     protocol_dir = run_dir / "protocol"

tests/test_speaker_mapping_ui.py

@@ -80,7 +80,7 @@ def test_widget_reruns_filter_release_and_update_status():
     assert not app.exception
 
 
-def test_anonymous_regeneration_remains_allowed(monkeypatch):
+def test_checkpoint_allows_anonymous_protocol_generation(monkeypatch):
     from mka.application.meeting_service import SpeakerMappingReview, SpeakerReview
     from mka.ui import streamlit_app as ui
 
@@ -89,6 +89,8 @@ def test_anonymous_regeneration_remains_allowed(monkeypatch):
         run_dir=Path("/tmp/mapping-test"),
         speaker_attribution_available=True,
         original_protocol="Anonymous protocol",
+        awaiting_speaker_review=True,
+        detected_speaker_count=1,
     )
     review = SpeakerMappingReview((SpeakerReview("SPEAKER_00", ()),), (("a", "A"),), {})
     calls = []
@@ -124,9 +126,9 @@ def test_anonymous_regeneration_remains_allowed(monkeypatch):
         _render_result()
 
     app = AppTest.from_function(result_app, args=(outcome,)).run()
-    regenerate = app.button[1]
-    assert not regenerate.disabled
+    generate = next(button for button in app.button if button.label == "Generate protocol")
+    assert not generate.disabled
     assert app.warning
-    regenerate.click().run()
+    generate.click().run()
     assert calls == [{}]
     assert not app.exception