diff --git a/CHANGELOG.md b/CHANGELOG.md index 34dc0c6..70a000e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,9 @@ ### Added +- Retained live stage and total timings for processing and protocol regeneration. +- Immutable diagnostic generation history with atomic latest-result publication. + - Auto/Fast/Efficient/Powersave protocol profiles with backend-selected threads by default, also applied during mapped-speaker regeneration. diff --git a/PROJECT_KNOWLEDGE.md b/PROJECT_KNOWLEDGE.md index ef7fa96..43f35cc 100644 --- a/PROJECT_KNOWLEDGE.md +++ b/PROJECT_KNOWLEDGE.md @@ -169,3 +169,14 @@ Active glossary aliases are forwarded to Meeting Lab as provenance metadata. Canonical terminology remains Meeting Context guidance; the exact protocol transcript input and the raw Whisper/diarization artifacts are not rewritten. `glossary_replacements` remains empty. + +Processing and protocol regeneration show measured monotonic stage/total timings, +including frozen failure durations. The latest timing display survives ordinary +Streamlit reruns. Initial-run durations are saved in `run_metadata.json`; +regeneration display timings remain session-local. Worker callbacks capture UI +configuration before dispatch and never read Streamlit session state. + +Meeting Lab retains immutable generation records including prompt, transcript +input, model response/metadata, glossary configuration, mappings and Meeting +Context. Latest protocol/diagnostic/context paths resolve through an atomic +`protocol/current` link. Copy whole runs with relative symlinks preserved. diff --git a/README.md b/README.md index 2eed545..4a803e9 100644 --- a/README.md +++ b/README.md @@ -234,3 +234,14 @@ Active glossary aliases are forwarded to Meeting Lab as provenance metadata. Canonical terminology remains Meeting Context guidance; the exact protocol transcript input and the raw Whisper/diarization artifacts are not rewritten. `glossary_replacements` remains empty. + +Processing and protocol regeneration show measured monotonic stage/total timings, +including frozen failure durations. The latest timing display survives ordinary +Streamlit reruns. Initial-run durations are saved in `run_metadata.json`; +regeneration display timings remain session-local. Worker callbacks capture UI +configuration before dispatch and never read Streamlit session state. + +Meeting Lab retains immutable generation records including prompt, transcript +input, model response/metadata, glossary configuration, mappings and Meeting +Context. Latest protocol/diagnostic/context paths resolve through an atomic +`protocol/current` link. Copy whole runs with relative symlinks preserved. diff --git a/src/mka/application/meeting_service.py b/src/mka/application/meeting_service.py index baae273..28c95a3 100644 --- a/src/mka/application/meeting_service.py +++ b/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 diff --git a/src/mka/application/progress_timing.py b/src/mka/application/progress_timing.py new file mode 100644 index 0000000..35eb13b --- /dev/null +++ b/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, + ) diff --git a/src/mka/ui/streamlit_app.py b/src/mka/ui/streamlit_app.py index e713e4e..96bdf54 100644 --- a/src/mka/ui/streamlit_app.py +++ b/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." ) diff --git a/tests/test_live_progress_ui.py b/tests/test_live_progress_ui.py new file mode 100644 index 0000000..253d7a7 --- /dev/null +++ b/tests/test_live_progress_ui.py @@ -0,0 +1,88 @@ +"""Exercise the real executor and Streamlit reruns without model calls.""" + +from pathlib import Path +from threading import current_thread +from types import SimpleNamespace + +import pytest +import streamlit +from streamlit.testing.v1 import AppTest + +from mka.application.meeting_service import AppProgressEvent, SpeakerMappingReview, SpeakerReview +from mka.application.progress_timing import ProcessingTimer +from mka.ui import streamlit_app as ui + + +class GuardedStreamlit: + """Fail if a worker touches the UI session proxy, including callback reads.""" + + def __getattr__(self, name): + if name == "session_state" and current_thread().name.startswith("ThreadPoolExecutor"): + raise AssertionError("Worker accessed Streamlit session state") + return getattr(streamlit, name) + + +@pytest.mark.parametrize("fails", [False, True]) +@pytest.mark.parametrize("mapped", [False, True]) +def test_regeneration_captures_ui_values_before_real_worker_and_retains_timing( + monkeypatch, fails, mapped +): + outcome = SimpleNamespace( + succeeded=True, + run_dir=Path("/tmp/alpha-mapping-test"), + speaker_attribution_available=True, + original_protocol="Existing protocol", + ) + review = SpeakerMappingReview((SpeakerReview("SPEAKER_00", ()),), (("a", "A"),), {}) + calls = [] + + class Service: + def __init__(self, *args): + self.timer = ProcessingTimer() + + def load_speaker_mapping_review(self, run_dir): + return review + + def regenerate_protocol(self, run_dir, selections, *, progress_sink, performance_profile): + assert current_thread().name.startswith("ThreadPoolExecutor") + calls.append((run_dir, selections, performance_profile)) + self.timer.start_run() + self.timer.start_stage("protocol_generation") + progress_sink(AppProgressEvent("protocol_generation", "started", 0.0)) + if fails: + raise RuntimeError("model unavailable") + self.timer.finish_stage("protocol_generation") + self.timer.finish_run() + progress_sink(AppProgressEvent("protocol_generation", "completed", 0.0)) + return outcome + + monkeypatch.setattr(ui, "st", GuardedStreamlit()) + monkeypatch.setattr(ui, "MeetingProcessingService", Service) + monkeypatch.setattr(ui.AppSettings, "from_environment", lambda: None) + monkeypatch.setattr(ui, "MeetingLabGateway", lambda: None) + + def result_app(outcome): + import streamlit as st + + from mka.ui.streamlit_app import _render_result + + st.session_state.outcome = outcome + st.session_state.performance_profile = "fast" + _render_result() + + app = AppTest.from_function(result_app, args=(outcome,)).run() + if mapped: + app.selectbox[0].select("a").run() + button = next(button for button in app.button if "Regenerate" in button.label) + assert not button.disabled + button.click().run() + assert not app.exception + assert calls == [(outcome.run_dir, {"SPEAKER_00": "a"} if mapped else {}, "fast")] + states, message, timing = app.session_state["processing_progress"] + assert states["protocol_generation"] == ("failed" if fails else "completed") + assert not timing.running + assert timing.total_duration >= timing.stage_durations["protocol_generation"] >= 0 + app.run() + assert not app.exception + assert app.session_state["processing_progress"][2] == timing + assert any("total runtime" in item.value for item in app.info) diff --git a/tests/test_meeting_service.py b/tests/test_meeting_service.py index 55ddbdd..ce5054f 100644 --- a/tests/test_meeting_service.py +++ b/tests/test_meeting_service.py @@ -14,6 +14,7 @@ from mka.application.meeting_service import ( ParticipantInput, ProcessingOptions, ) +from mka.application.progress_timing import ProcessingTimer @dataclass @@ -623,3 +624,111 @@ def test_process_passes_only_active_explicit_glossary_aliases(tmp_path: Path) -> assert gateway.config_values is not None assert gateway.config_values["glossary_aliases"] == {"Lumini": "Luminy"} + + +@pytest.mark.parametrize("fails", [False, True]) +def test_process_persists_completed_and_failed_timings(tmp_path: Path, fails: bool) -> None: + service, gateway = make_service(tmp_path) + gateway.fail = fails + values = iter([0.0, 1.0, 9.0, 10.0, 20.0]) + service.timer = ProcessingTimer(lambda: next(values)) + audio = tmp_path / "meeting.wav" + audio.write_bytes(b"audio") + + outcome = service.process(audio, meeting(), participants(), ProcessingOptions()) + + metadata = json.loads((gateway.run_dir / "run_metadata.json").read_text(encoding="utf-8")) + assert metadata["timing"] == { + "stages_seconds": {"preparing": 8.0, "transcription": 10.0}, + "total_seconds": 20.0, + } + assert outcome.succeeded is not fails + if fails: + assert metadata["failure"]["type"] == "TranscriptionError" + + +def test_regeneration_starts_fresh_timer_and_retains_final_duration( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + service, gateway = make_service(tmp_path) + write_speaker_review_artifacts(gateway.run_dir) + clock = SimpleNamespace(value=10.0) + service.timer = ProcessingTimer(lambda: clock.value) + audio = tmp_path / "meeting.wav" + audio.write_bytes(b"audio") + service.process(audio, meeting(), participants(), ProcessingOptions()) + + clock.value = 100.0 + observed = [] + + def regenerate(run_dir: Path, context: Any, progress_sink: Any, **options: Any) -> Any: + progress_sink( + SimpleNamespace( + stage="protocol_generation", + status="started", + elapsed_seconds=0.0, + progress=None, + message=None, + ) + ) + clock.value = 105.0 + observed.append(service.timer.snapshot()) + progress_sink( + SimpleNamespace( + stage="protocol_generation", + status="completed", + elapsed_seconds=5.0, + progress=None, + message=None, + ) + ) + clock.value = 109.0 + protocol = run_dir / "protocol.md" + protocol.write_text("# Regenerated protocol\n", encoding="utf-8") + return SimpleNamespace(run_dir=run_dir, protocol_path=protocol) + + monkeypatch.setattr(gateway, "regenerate_protocol", regenerate) + + service.regenerate_protocol(gateway.run_dir, {"SPEAKER_00": "martin"}) + final = service.timer.snapshot() + + assert observed[0].active_stage == "protocol_generation" + assert observed[0].stage_durations == {"protocol_generation": 5.0} + assert observed[0].total_duration == 5.0 + assert observed[0].running is True + assert final.stage_durations == {"protocol_generation": 5.0} + assert final.total_duration == 9.0 + assert final.running is False + + +def test_failed_regeneration_freezes_active_and_total_durations( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + service, gateway = make_service(tmp_path) + write_speaker_review_artifacts(gateway.run_dir) + clock = SimpleNamespace(value=20.0) + service.timer = ProcessingTimer(lambda: clock.value) + + def fail_regeneration(run_dir: Path, context: Any, progress_sink: Any, **options: Any) -> Any: + progress_sink( + SimpleNamespace( + stage="protocol_generation", + status="started", + elapsed_seconds=0.0, + progress=None, + message=None, + ) + ) + clock.value = 27.0 + raise RuntimeError("generation failed") + + monkeypatch.setattr(gateway, "regenerate_protocol", fail_regeneration) + + with pytest.raises(RuntimeError, match="generation failed"): + service.regenerate_protocol(gateway.run_dir, {}) + clock.value = 99.0 + final = service.timer.snapshot() + + assert final.stage_durations == {"protocol_generation": 7.0} + assert final.total_duration == 7.0 + assert final.running is False diff --git a/tests/test_progress_timing.py b/tests/test_progress_timing.py new file mode 100644 index 0000000..2380982 --- /dev/null +++ b/tests/test_progress_timing.py @@ -0,0 +1,45 @@ +from mka.application.progress_timing import ProcessingTimer + + +class FakeClock: + def __init__(self, value: float = 0.0) -> None: + self.value = value + + def __call__(self) -> float: + return self.value + + +def test_completed_duration_is_retained_while_active_stage_increases() -> None: + clock = FakeClock(10.0) + timer = ProcessingTimer(clock) + timer.start_run() + timer.start_stage("preparing") + clock.value = 18.0 + timer.finish_stage("preparing") + timer.start_stage("transcription") + + clock.value = 20.0 + first = timer.snapshot() + clock.value = 25.5 + second = timer.snapshot() + + assert first.stage_durations == {"preparing": 8.0, "transcription": 2.0} + assert second.stage_durations == {"preparing": 8.0, "transcription": 7.5} + assert second.total_duration == 15.5 + + +def test_failed_active_stage_and_total_duration_are_frozen() -> None: + clock = FakeClock(2.0) + timer = ProcessingTimer(clock) + timer.start_run() + clock.value = 5.0 + timer.start_stage("transcription") + clock.value = 14.0 + timer.finish_run() + clock.value = 99.0 + + snapshot = timer.snapshot() + + assert snapshot.stage_durations["transcription"] == 9.0 + assert snapshot.total_duration == 12.0 + assert snapshot.running is False diff --git a/tests/test_streamlit_app.py b/tests/test_streamlit_app.py index fee8def..fb1f56b 100644 --- a/tests/test_streamlit_app.py +++ b/tests/test_streamlit_app.py @@ -1,6 +1,7 @@ from datetime import date from mka.application.meeting_service import ParticipantInput +from mka.application.progress_timing import TimingSnapshot from mka.application.run_inputs import RunInputState from mka.ui import streamlit_app @@ -66,3 +67,29 @@ def test_imported_inputs_are_applied_via_pending_state_before_widgets( assert "source_media_1" not in state assert "pending_run_inputs" not in state assert state["source_media_0"] is existing_upload + + +def test_duration_formatting_covers_seconds_minutes_and_hours() -> None: + assert streamlit_app._format_duration(8.9) == "00:08" + assert streamlit_app._format_duration(7 * 60 + 18) == "07:18" + assert streamlit_app._format_duration(3600 + 3 * 60 + 42) == "1:03:42" + + +def test_completed_regeneration_timing_survives_frontend_rerun(monkeypatch) -> None: + state = {} + monkeypatch.setattr(streamlit_app.st, "session_state", state) + timing = TimingSnapshot( + stage_durations={"protocol_generation": 4.5}, + active_stage=None, + total_duration=4.5, + running=False, + ) + states = {stage: "skipped" for stage in streamlit_app.STAGES} + states["protocol_generation"] = "completed" + + streamlit_app._remember_regeneration_timing(states, "Protocol regenerated", timing) + + saved_states, saved_message, saved_timing = state["regeneration_progress"] + assert saved_states["protocol_generation"] == "completed" + assert saved_message == "Protocol regenerated" + assert saved_timing == timing