Retain live processing timings across UI reruns

src/mka/application/meeting_service.py

@@ -21,6 +21,7 @@ from mka.application.glossary import (
     render_glossary_terms,
 )
 from mka.application.performance import DEFAULT_PERFORMANCE_PROFILE, resolve_ollama_num_thread
+from mka.application.progress_timing import ProcessingTimer
 
 STAGES = ("preparing", "transcription", "diarization", "protocol_generation")
 
@@ -120,11 +121,13 @@ class MeetingProcessingService:
         settings: AppSettings,
         meeting_lab: MeetingLabPort,
         glossary: GlossaryRepository | None = None,
+        timer: ProcessingTimer | None = None,
     ) -> None:
         self.settings = settings
         self.meeting_lab = meeting_lab
         self.glossary = glossary or GlossaryRepository(settings.glossary_database)
         self.glossary.initialize()
+        self.timer = timer or ProcessingTimer()
 
     def build_context(
         self,
@@ -235,11 +238,11 @@ class MeetingProcessingService:
                 "model": self.settings.protocol_model,
                 "ollama_endpoint": self.settings.ollama_endpoint,
                 "protocol_num_ctx": self.settings.protocol_num_ctx,
-                "glossary_aliases": glossary_alias_mapping(self.glossary.list(active_only=True)),
-                "protocol_num_thread": resolve_ollama_num_thread(options.performance_profile),
                 "protocol_safe_input_token_budget": (
                     self.settings.protocol_safe_input_token_budget
                 ),
+                "protocol_num_thread": resolve_ollama_num_thread(options.performance_profile),
+                "glossary_aliases": glossary_alias_mapping(self.glossary.list(active_only=True)),
                 "diarization": (
                     self.settings.diarization_mode if options.diarization_enabled else "off"
                 ),
@@ -249,11 +252,13 @@ class MeetingProcessingService:
             }
         )
         current_stage: str | None = None
+        self.timer.start_run()
 
         def relay(event: Any) -> None:
             nonlocal current_stage
             if event.stage in STAGES and event.status == "started":
                 current_stage = event.stage
+            self._update_timer(event)
             app_event = AppProgressEvent(
                 stage=event.stage,
                 status=event.status,
@@ -264,7 +269,13 @@ class MeetingProcessingService:
             if progress_sink is not None:
                 progress_sink(app_event)
 
-        result = self.meeting_lab.run(config, context, relay)
+        try:
+            result = self.meeting_lab.run(config, context, relay)
+        except Exception:
+            self.timer.finish_run()
+            raise
+        self.timer.finish_run()
+        self._persist_timing(result.run_dir)
         if result.exit_code != 0:
             failure = self._read_failure(result.run_dir)
             backend_stage = failure.get("stage")
@@ -392,6 +403,7 @@ class MeetingProcessingService:
         context = self.meeting_lab.create_context(context_data)
 
         def relay(event: Any) -> None:
+            self._update_timer(event)
             if progress_sink is not None:
                 progress_sink(
                     AppProgressEvent(
@@ -403,17 +415,23 @@ class MeetingProcessingService:
                     )
                 )
 
-        result = self.meeting_lab.regenerate_protocol(
-            Path(run_dir),
-            context,
-            relay,
-            model=self.settings.protocol_model,
-            ollama_endpoint=self.settings.ollama_endpoint,
-            protocol_num_ctx=self.settings.protocol_num_ctx,
-            glossary_aliases=glossary_alias_mapping(self.glossary.list(active_only=True)),
-            protocol_num_thread=resolve_ollama_num_thread(performance_profile),
-            protocol_safe_input_token_budget=(self.settings.protocol_safe_input_token_budget),
-        )
+        self.timer.start_run()
+        try:
+            result = self.meeting_lab.regenerate_protocol(
+                Path(run_dir),
+                context,
+                relay,
+                model=self.settings.protocol_model,
+                ollama_endpoint=self.settings.ollama_endpoint,
+                protocol_num_ctx=self.settings.protocol_num_ctx,
+                protocol_safe_input_token_budget=(self.settings.protocol_safe_input_token_budget),
+                protocol_num_thread=resolve_ollama_num_thread(performance_profile),
+                glossary_aliases=glossary_alias_mapping(self.glossary.list(active_only=True)),
+            )
+        except Exception:
+            self.timer.finish_run()
+            raise
+        self.timer.finish_run()
         protocol_path = Path(result.protocol_path)
         return ProcessingOutcome(
             succeeded=True,
@@ -474,3 +492,38 @@ class MeetingProcessingService:
         destination = Path(run_dir) / "protocol_edited.md"
         destination.write_text(text, encoding="utf-8")
         return destination
+
+    def _update_timer(self, event: Any) -> None:
+        """Apply a backend progress event to the shared per-run timer."""
+        if event.stage in STAGES and event.status == "started":
+            self.timer.start_stage(event.stage)
+        elif event.stage in STAGES and event.status in {"completed", "skipped"}:
+            self.timer.finish_stage(event.stage)
+        if event.stage in {"completed", "failed"}:
+            self.timer.finish_run()
+
+    def _persist_timing(self, run_dir: Path | None) -> None:
+        """Merge final monotonic durations into Meeting Lab run metadata."""
+        if run_dir is None:
+            return
+        metadata_path = Path(run_dir) / "run_metadata.json"
+        metadata: dict[str, Any] = {}
+        if metadata_path.is_file():
+            try:
+                existing = json.loads(metadata_path.read_text(encoding="utf-8"))
+                if isinstance(existing, dict):
+                    metadata = existing
+            except (OSError, json.JSONDecodeError):
+                return
+        snapshot = self.timer.snapshot()
+        metadata["timing"] = {
+            "stages_seconds": snapshot.stage_durations,
+            "total_seconds": snapshot.total_duration,
+        }
+        metadata_path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            metadata_path.write_text(
+                json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+            )
+        except OSError:
+            return

src/mka/application/progress_timing.py

@@ -0,0 +1,80 @@
+"""Thread-safe monotonic timing state for meeting-processing progress."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from threading import Lock
+
+
+@dataclass(frozen=True)
+class TimingSnapshot:
+    """Presentation-neutral runtime values measured in seconds."""
+
+    stage_durations: dict[str, float]
+    active_stage: str | None
+    total_duration: float
+    running: bool
+
+
+class ProcessingTimer:
+    """Track pipeline stages independently from pipeline business logic."""
+
+    def __init__(self, clock: Callable[[], float] = time.perf_counter) -> None:
+        self._clock = clock
+        self._lock = Lock()
+        self._run_started: float | None = None
+        self._run_finished: float | None = None
+        self._stage_started: dict[str, float] = {}
+        self._stage_finished: dict[str, float] = {}
+        self._active_stage: str | None = None
+
+    def start_run(self) -> None:
+        with self._lock:
+            self._run_started = self._clock()
+            self._run_finished = None
+            self._stage_started.clear()
+            self._stage_finished.clear()
+            self._active_stage = None
+
+    def start_stage(self, stage: str) -> None:
+        with self._lock:
+            now = self._clock()
+            if self._active_stage is not None and self._active_stage != stage:
+                self._stage_finished.setdefault(self._active_stage, now)
+            self._stage_started.setdefault(stage, now)
+            self._active_stage = stage
+
+    def finish_stage(self, stage: str) -> None:
+        with self._lock:
+            if stage in self._stage_started:
+                self._stage_finished.setdefault(stage, self._clock())
+            if self._active_stage == stage:
+                self._active_stage = None
+
+    def finish_run(self) -> None:
+        with self._lock:
+            if self._run_finished is not None:
+                return
+            now = self._clock()
+            if self._active_stage is not None:
+                self._stage_finished.setdefault(self._active_stage, now)
+                self._active_stage = None
+            if self._run_started is not None:
+                self._run_finished = now
+
+    def snapshot(self) -> TimingSnapshot:
+        with self._lock:
+            now = self._run_finished if self._run_finished is not None else self._clock()
+            durations = {
+                stage: max(0.0, self._stage_finished.get(stage, now) - started)
+                for stage, started in self._stage_started.items()
+            }
+            total = max(0.0, now - self._run_started) if self._run_started is not None else 0.0
+            return TimingSnapshot(
+                stage_durations=durations,
+                active_stage=self._active_stage,
+                total_duration=total,
+                running=self._run_started is not None and self._run_finished is None,
+            )

src/mka/ui/streamlit_app.py

@@ -2,7 +2,11 @@
 
 from __future__ import annotations
 
+import time
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
 from datetime import date
+from queue import Empty, Queue
 from typing import Any
 from uuid import uuid4
 
@@ -35,6 +39,7 @@ from mka.application.people_yaml import (
     import_people_yaml,
 )
 from mka.application.performance import DEFAULT_PERFORMANCE_PROFILE, PERFORMANCE_PROFILES
+from mka.application.progress_timing import TimingSnapshot
 from mka.application.run_inputs import (
     RunInputJsonError,
     RunInputState,
@@ -338,40 +343,132 @@ def _render_participants() -> list[ParticipantInput]:
     return people
 
 
-def _progress_callback(
+def _format_duration(seconds: float) -> str:
+    """Format numeric seconds as MM:SS or H:MM:SS."""
+    whole_seconds = max(0, int(seconds))
+    hours, remainder = divmod(whole_seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+    return f"{hours}:{minutes:02d}:{seconds:02d}" if hours else f"{minutes:02d}:{seconds:02d}"
+
+
+def _render_progress_status(
     status_box: Any,
     stage_table: Any,
-    progress_slot: Any,
     states: dict[str, str],
-) -> Any:
-    progress_bar = None
+    timing: TimingSnapshot,
+    message: str,
+) -> None:
+    suffix = " …" if timing.running else ""
+    status_box.info(f"{message} — total runtime {_format_duration(timing.total_duration)}{suffix}")
+    stage_table.table(
+        [
+            {
+                "Stage": STAGE_LABELS[stage],
+                "Status": states[stage],
+                "Runtime": (
+                    _format_duration(timing.stage_durations[stage])
+                    + (" …" if timing.active_stage == stage else "")
+                    if stage in timing.stage_durations
+                    else ""
+                ),
+            }
+            for stage in STAGES
+        ]
+        + [
+            {
+                "Stage": "Total runtime",
+                "Status": "",
+                "Runtime": _format_duration(timing.total_duration) + suffix,
+            }
+        ]
+    )
 
-    def update(event: AppProgressEvent) -> None:
-        nonlocal progress_bar
-        if event.stage in states:
-            states[event.stage] = "running" if event.status == "started" else event.status
-        if event.stage == "failed":
-            running = next(
+
+def _apply_progress_event(event: AppProgressEvent, states: dict[str, str]) -> None:
+    if event.stage in states:
+        states[event.stage] = "running" if event.status == "started" else event.status
+    if event.stage == "failed":
+        running = next((stage for stage, status in states.items() if status == "running"), None)
+        if running:
+            states[running] = "failed"
+
+
+def _run_with_live_progress(
+    service: MeetingProcessingService,
+    work: Callable[[Callable[[AppProgressEvent], None]], Any],
+    states: dict[str, str],
+    initial_message: str,
+) -> tuple[Any, dict[str, str], str, TimingSnapshot]:
+    """Run pipeline work while rendering the shared timer and progress events."""
+    status_box = st.empty()
+    stage_table = st.empty()
+    progress_slot = st.empty()
+    event_queue: Queue[AppProgressEvent] = Queue()
+    latest_message = initial_message
+    progress_bar = None
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        future = executor.submit(work, event_queue.put)
+        while not future.done():
+            try:
+                while True:
+                    event = event_queue.get_nowait()
+                    _apply_progress_event(event, states)
+                    latest_message = event.message or STAGE_LABELS.get(event.stage, event.stage)
+                    if event.progress is not None:
+                        if progress_bar is None:
+                            progress_bar = progress_slot.progress(0.0)
+                        progress_bar.progress(
+                            min(max(event.progress, 0.0), 1.0),
+                            text=(
+                                f"{STAGE_LABELS.get(event.stage, event.stage)}: "
+                                f"{event.progress:.0%}"
+                            ),
+                        )
+            except Empty:
+                pass
+            _render_progress_status(
+                status_box,
+                stage_table,
+                states,
+                service.timer.snapshot(),
+                latest_message,
+            )
+            time.sleep(0.2)
+        try:
+            result = future.result()
+        except Exception:
+            while not event_queue.empty():
+                event = event_queue.get_nowait()
+                _apply_progress_event(event, states)
+                latest_message = event.message or STAGE_LABELS.get(event.stage, event.stage)
+            service.timer.finish_run()
+            running_stage = next(
                 (stage for stage, status in states.items() if status == "running"),
                 None,
             )
-            if running:
-                states[running] = "failed"
-        elapsed = f"{event.elapsed_seconds:.1f} s"
-        message = event.message or STAGE_LABELS.get(event.stage, event.stage)
-        status_box.info(f"{message} — elapsed {elapsed}")
-        stage_table.table(
-            [{"Stage": STAGE_LABELS[stage], "Status": states[stage]} for stage in STAGES]
-        )
-        if event.progress is not None:
-            if progress_bar is None:
-                progress_bar = progress_slot.progress(0.0)
-            progress_bar.progress(
-                min(max(event.progress, 0.0), 1.0),
-                text=f"{STAGE_LABELS.get(event.stage, event.stage)}: {event.progress:.0%}",
-            )
+            if running_stage is not None:
+                states[running_stage] = "failed"
+            snapshot = service.timer.snapshot()
+            _render_progress_status(status_box, stage_table, states, snapshot, latest_message)
+            st.session_state["processing_progress"] = (states.copy(), latest_message, snapshot)
+            raise
+    while not event_queue.empty():
+        event = event_queue.get_nowait()
+        _apply_progress_event(event, states)
+        latest_message = event.message or STAGE_LABELS.get(event.stage, event.stage)
+    running_stage = next((stage for stage, status in states.items() if status == "running"), None)
+    if running_stage is not None:
+        states[running_stage] = "completed" if getattr(result, "succeeded", True) else "failed"
+    snapshot = service.timer.snapshot()
+    _render_progress_status(status_box, stage_table, states, snapshot, latest_message)
+    st.session_state["processing_progress"] = (states.copy(), latest_message, snapshot)
+    return result, states, latest_message, snapshot
 
-    return update
+
+def _remember_regeneration_timing(
+    states: dict[str, str], message: str, timing: TimingSnapshot
+) -> None:
+    st.session_state["regeneration_progress"] = (states.copy(), message, timing)
 
 
 def _speaker_options(
@@ -447,6 +544,9 @@ def _render_result() -> None:
         return
     st.divider()
     st.header("Protocol result")
+    if retained_progress := st.session_state.get("processing_progress"):
+        states, message, timing = retained_progress
+        _render_progress_status(st.empty(), st.empty(), states, timing, message)
     if not outcome.succeeded:
         st.error(f"Processing failed during {outcome.failed_stage}: {outcome.error_message}")
         if outcome.run_dir:
@@ -494,18 +594,33 @@ def _render_result() -> None:
         disabled=duplicate_assignments,
         type="primary",
     ):
+        st.session_state.pop("regeneration_progress", None)
+        run_dir = outcome.run_dir
+        mapping_items = tuple(selections.items())
+        selected_profile = st.session_state.get("performance_profile", DEFAULT_PERFORMANCE_PROFILE)
+        states = {stage: "skipped" for stage in STAGES}
+        states["protocol_generation"] = "pending"
         try:
-            with st.spinner("Regenerating protocol without rerunning audio processing..."):
-                regenerated = service.regenerate_protocol(
-                    outcome.run_dir,
-                    selections,
-                    performance_profile=st.session_state.get(
-                        "performance_profile", DEFAULT_PERFORMANCE_PROFILE
-                    ),
-                )
+            regenerated, states, message, timing = _run_with_live_progress(
+                service,
+                lambda progress_sink: service.regenerate_protocol(
+                    run_dir,
+                    dict(mapping_items),
+                    progress_sink=progress_sink,
+                    performance_profile=selected_profile,
+                ),
+                states,
+                "Starting protocol regeneration",
+            )
         except (OSError, RuntimeError, ValueError) as exc:
+            _remember_regeneration_timing(
+                states,
+                "Protocol regeneration failed",
+                service.timer.snapshot(),
+            )
             st.error(f"Protocol regeneration 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 = (
@@ -618,31 +733,34 @@ def main() -> None:
                 )
                 meeting_id = stable_id(title)
                 audio_path = service.preserve_upload(meeting_id, audio.name, audio)
+                st.session_state.pop("regeneration_progress", None)
+                st.session_state.pop("processing_progress", None)
                 st.header("Processing status")
-                status_box = st.empty()
-                stage_table = st.empty()
-                progress_slot = st.empty()
                 states = {stage: "pending" for stage in STAGES}
                 if not diarization_enabled:
                     states["diarization"] = "skipped"
-                callback = _progress_callback(status_box, stage_table, progress_slot, states)
-                outcome = service.process(
-                    audio_path,
-                    meeting,
-                    participants,
-                    ProcessingOptions(
-                        diarization_enabled=diarization_enabled,
-                        audio_normalization=audio_normalization,
-                        performance_profile=performance_profile,
+                outcome, _, _, _ = _run_with_live_progress(
+                    service,
+                    lambda progress_sink: service.process(
+                        audio_path,
+                        meeting,
+                        participants,
+                        ProcessingOptions(
+                            diarization_enabled=diarization_enabled,
+                            audio_normalization=audio_normalization,
+                            performance_profile=performance_profile,
+                        ),
+                        progress_sink=progress_sink,
                     ),
-                    progress_sink=callback,
+                    states,
+                    "Starting processing",
                 )
                 st.session_state.outcome = outcome
                 st.session_state.edited_protocol = outcome.original_protocol or ""
                 if outcome.succeeded:
-                    status_box.success("Processing completed.")
+                    st.success("Processing completed.")
                 else:
-                    status_box.error(
+                    st.error(
                         f"Processing failed during {outcome.failed_stage}: "
                         f"{outcome.error_message} Artifacts were preserved."
                     )