922 lines
33 KiB
Python
922 lines
33 KiB
Python
import json
|
|
from dataclasses import dataclass, replace
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from mka.application.config import AppSettings
|
|
from mka.application.meeting_service import (
|
|
MeetingDetails,
|
|
MeetingProcessingService,
|
|
ParticipantInput,
|
|
ProcessingOptions,
|
|
)
|
|
from mka.application.progress_timing import ProcessingTimer
|
|
|
|
|
|
@dataclass
|
|
class FakeContext:
|
|
data: dict[str, Any]
|
|
|
|
@property
|
|
def meeting_id(self) -> str:
|
|
return self.data["meeting"]["meeting_id"]
|
|
|
|
|
|
class FakeMeetingLab:
|
|
def __init__(self, run_dir: Path) -> None:
|
|
self.run_dir = run_dir
|
|
self.context_data: dict[str, Any] | None = None
|
|
self.config_values: dict[str, Any] | None = None
|
|
self.fail = False
|
|
self.regeneration: dict[str, Any] | None = None
|
|
|
|
def create_context(self, data: dict[str, Any]) -> FakeContext:
|
|
self.context_data = data
|
|
if not data["meeting"]["title"]:
|
|
raise ValueError("meeting.title must be present and non-empty")
|
|
participant_ids = [item["participant_id"] for item in data["participants"]]
|
|
if len(participant_ids) != len(set(participant_ids)):
|
|
raise ValueError("Duplicate participant_id")
|
|
return FakeContext(data)
|
|
|
|
def create_config(self, values: dict[str, Any]) -> dict[str, Any]:
|
|
self.config_values = values
|
|
return values
|
|
|
|
def run(self, config: Any, meeting_context: Any, progress_sink: Any) -> Any:
|
|
self.run_dir.mkdir(parents=True, exist_ok=True)
|
|
progress_sink(
|
|
SimpleNamespace(
|
|
stage="preparing",
|
|
status="started",
|
|
elapsed_seconds=0.1,
|
|
progress=None,
|
|
message=None,
|
|
)
|
|
)
|
|
progress_sink(
|
|
SimpleNamespace(
|
|
stage="preparing",
|
|
status="completed",
|
|
elapsed_seconds=0.2,
|
|
progress=None,
|
|
message=None,
|
|
)
|
|
)
|
|
progress_sink(
|
|
SimpleNamespace(
|
|
stage="transcription",
|
|
status="started",
|
|
elapsed_seconds=0.2,
|
|
progress=0.25,
|
|
message="transcribing",
|
|
)
|
|
)
|
|
if self.fail:
|
|
(self.run_dir / "run_metadata.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"failure": {
|
|
"stage": "whisper",
|
|
"type": "TranscriptionError",
|
|
"message": "model failed",
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
progress_sink(
|
|
SimpleNamespace(
|
|
stage="failed",
|
|
status="failed",
|
|
elapsed_seconds=0.3,
|
|
progress=None,
|
|
message="transcription: model failed",
|
|
)
|
|
)
|
|
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)
|
|
|
|
def regenerate_protocol(
|
|
self,
|
|
run_dir: Path,
|
|
meeting_context: Any,
|
|
progress_sink: Any,
|
|
**options: Any,
|
|
) -> Any:
|
|
self.regeneration = {
|
|
"run_dir": run_dir,
|
|
"meeting_context": meeting_context,
|
|
"options": options,
|
|
}
|
|
progress_sink(
|
|
SimpleNamespace(
|
|
stage="protocol_generation",
|
|
status="started",
|
|
elapsed_seconds=0.0,
|
|
progress=None,
|
|
message=None,
|
|
)
|
|
)
|
|
protocol = Path(run_dir) / "protocol.md"
|
|
protocol.write_text("# Regenerated protocol\n", encoding="utf-8")
|
|
protocol_dir = Path(run_dir) / "protocol"
|
|
protocol_dir.mkdir(exist_ok=True)
|
|
(protocol_dir / "runtime_metadata.json").write_text(
|
|
json.dumps({"speaker_attribution_available": True}), encoding="utf-8"
|
|
)
|
|
return SimpleNamespace(exit_code=0, run_dir=run_dir, protocol_path=protocol)
|
|
|
|
|
|
def make_service(tmp_path: Path) -> tuple[MeetingProcessingService, FakeMeetingLab]:
|
|
model = tmp_path / "model.bin"
|
|
model.write_bytes(b"model")
|
|
gateway = FakeMeetingLab(tmp_path / "backend-run")
|
|
settings = AppSettings(
|
|
data_root=tmp_path / "meetings",
|
|
whisper_model=model,
|
|
glossary_database=tmp_path / "glossary.sqlite3",
|
|
whisper_executable="/opt/whisper-cli",
|
|
protocol_model="test:model",
|
|
diarization_mode="gpu",
|
|
)
|
|
return MeetingProcessingService(settings, gateway), gateway
|
|
|
|
|
|
def meeting() -> MeetingDetails:
|
|
return MeetingDetails(
|
|
title="Architecture Review",
|
|
language="de",
|
|
meeting_date=date(2026, 8, 23),
|
|
description="Review the MVP.",
|
|
)
|
|
|
|
|
|
def participants() -> list[ParticipantInput]:
|
|
return [
|
|
ParticipantInput(
|
|
participant_id="martin",
|
|
display_name="Martin",
|
|
role="Project lead",
|
|
organization="Engineering",
|
|
),
|
|
ParticipantInput(
|
|
participant_id="alex",
|
|
display_name="Alex",
|
|
organization="Engineering",
|
|
),
|
|
]
|
|
|
|
|
|
def write_speaker_review_artifacts(run_dir: Path) -> Path:
|
|
diarization_dir = run_dir / "diarization"
|
|
context_dir = run_dir / "context"
|
|
diarization_dir.mkdir(parents=True)
|
|
context_dir.mkdir()
|
|
transcript_path = diarization_dir / "transcript_diarized.json"
|
|
transcript_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"speaker_labels_anonymous": True,
|
|
"segments": [
|
|
{
|
|
"speaker_id": "SPEAKER_01",
|
|
"text": "I will prepare all raw materials before Wednesday.",
|
|
},
|
|
{"speaker_id": "SPEAKER_00", "text": "Yes."},
|
|
{
|
|
"speaker_id": "SPEAKER_00",
|
|
"text": "We will run the production trial on Wednesday.",
|
|
},
|
|
{
|
|
"speaker_id": "SPEAKER_00",
|
|
"text": "The trial requires the complete production team.",
|
|
},
|
|
{"speaker_id": None, "text": "Unassigned text."},
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
(context_dir / "meeting_context.yaml").write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": "1",
|
|
"meeting": {
|
|
"meeting_id": "speaker-review",
|
|
"title": "Speaker review",
|
|
"language": "en",
|
|
},
|
|
"participants": [
|
|
{"participant_id": "martin", "display_name": "Martin"},
|
|
{"participant_id": "anna", "display_name": "Anna"},
|
|
],
|
|
"speaker_mappings": {},
|
|
"mentioned_people": [],
|
|
"organization": {"departments": []},
|
|
"known_entities": {},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return transcript_path
|
|
|
|
|
|
def write_custom_speaker_review_artifacts(
|
|
run_dir: Path, segments: list[dict[str, str | None]]
|
|
) -> Path:
|
|
"""Write custom diarized segments using the standard review context fixture."""
|
|
transcript_path = write_speaker_review_artifacts(run_dir)
|
|
transcript_path.write_text(
|
|
json.dumps({"speaker_labels_anonymous": True, "segments": segments}),
|
|
encoding="utf-8",
|
|
)
|
|
return transcript_path
|
|
|
|
|
|
def test_build_context_uses_actual_v1_shape(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
|
|
context = service.build_context(meeting(), participants())
|
|
|
|
assert context.meeting_id == "architecture-review"
|
|
assert gateway.context_data is not None
|
|
assert gateway.context_data["meeting"]["date"] == "2026-08-23"
|
|
assert gateway.context_data["meeting"]["notes"] == "Review the MVP."
|
|
assert gateway.context_data["participants"][0] == {
|
|
"participant_id": "martin",
|
|
"display_name": "Martin",
|
|
"aliases": [],
|
|
"role": "Project lead",
|
|
"department": "engineering",
|
|
"attendance_status": "present",
|
|
"notes": None,
|
|
}
|
|
assert gateway.context_data["organization"]["departments"] == [
|
|
{"id": "engineering", "name": "Engineering", "aliases": []}
|
|
]
|
|
|
|
|
|
def test_build_context_preserves_explicit_speaker_mapping(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
|
|
service.build_context(meeting(), participants(), {"SPEAKER_00": "martin"})
|
|
|
|
assert gateway.context_data is not None
|
|
assert gateway.context_data["speaker_mappings"] == {"SPEAKER_00": "martin"}
|
|
|
|
|
|
def test_build_context_includes_only_active_authoritative_glossary_terms(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
service.glossary.create("Secugrid HS", "product", aliases=("Sikirgut", "Secugrid H S"))
|
|
inactive = service.glossary.create("Old Name", "other")
|
|
service.glossary.set_active(inactive.id, False)
|
|
|
|
service.build_context(meeting(), participants())
|
|
|
|
assert gateway.context_data is not None
|
|
assert gateway.context_data["known_entities"] == {
|
|
"Authoritative terminology": ["Secugrid HS (aliases: Secugrid H S, Sikirgut)"]
|
|
}
|
|
rules = gateway.context_data["context_rules"]
|
|
assert "do not invent matches" in rules["glossary_canonical_spelling"]
|
|
assert "inside compounds" in rules["glossary_core_terms"]
|
|
assert "Old Name" not in str(gateway.context_data)
|
|
|
|
|
|
def test_glossary_is_rendered_into_meeting_lab_protocol_context(tmp_path: Path) -> None:
|
|
meeting_context = pytest.importorskip("src.meeting_lab.models.meeting_context")
|
|
service, gateway = make_service(tmp_path)
|
|
service.glossary.create("PBAT", "acronym", aliases=("P B A T",))
|
|
context = service.build_context(meeting(), participants())
|
|
|
|
real_context = meeting_context.create_meeting_context(context.data)
|
|
prompt_context = meeting_context.render_meeting_context_for_prompt(real_context)
|
|
|
|
assert "Authoritative terminology: PBAT (aliases: P B A T)" in prompt_context
|
|
assert "Use canonical glossary spellings" in prompt_context
|
|
assert "use canonical core terms inside compounds" in prompt_context
|
|
|
|
|
|
def test_build_context_translates_mentioned_only_person(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
people = participants() + [
|
|
ParticipantInput(
|
|
participant_id="sam",
|
|
display_name="Sam",
|
|
attendance_status="mentioned_only",
|
|
)
|
|
]
|
|
|
|
service.build_context(meeting(), people)
|
|
|
|
assert gateway.context_data is not None
|
|
assert [item["participant_id"] for item in gateway.context_data["participants"]] == [
|
|
"martin",
|
|
"alex",
|
|
]
|
|
assert gateway.context_data["mentioned_people"] == [
|
|
{
|
|
"person_id": "sam",
|
|
"display_name": "Sam",
|
|
"aliases": [],
|
|
"role": None,
|
|
"department": None,
|
|
"attendance_status": "mentioned_only",
|
|
"notes": None,
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("language", ["de", "en"])
|
|
def test_process_translates_configuration_and_disables_diarization(
|
|
tmp_path: Path,
|
|
language: str,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
outcome = service.process(
|
|
audio,
|
|
replace(meeting(), language=language),
|
|
participants(),
|
|
ProcessingOptions(diarization_enabled=False),
|
|
)
|
|
|
|
assert outcome.succeeded
|
|
assert gateway.config_values is not None
|
|
assert gateway.config_values["language"] == language
|
|
assert service.build_context(replace(meeting(), language=language), participants()).data[
|
|
"meeting"
|
|
]["language"] == language
|
|
assert gateway.config_values["diarization"] == "off"
|
|
assert gateway.config_values["model"] == "test:model"
|
|
assert gateway.config_values["protocol_num_ctx"] == 32_768
|
|
assert gateway.config_values["protocol_safe_input_token_budget"] == 29_000
|
|
assert gateway.config_values["whisper_executable"] == "/opt/whisper-cli"
|
|
assert gateway.config_values["ffmpeg_executable"] == "ffmpeg"
|
|
assert gateway.config_values["audio_normalization"] is True
|
|
assert gateway.config_values["diarization_runtime"] == "native"
|
|
assert gateway.config_values["diarization_container_args"] == ()
|
|
assert gateway.config_values["output_root"] == (
|
|
tmp_path / "meetings" / "architecture-review" / "runs"
|
|
)
|
|
|
|
|
|
def test_process_propagates_disabled_audio_normalization(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
audio = tmp_path / "meeting.m4a"
|
|
audio.write_bytes(b"audio")
|
|
|
|
outcome = service.process(
|
|
audio,
|
|
meeting(),
|
|
participants(),
|
|
ProcessingOptions(audio_normalization=False),
|
|
)
|
|
|
|
assert outcome.succeeded
|
|
assert gateway.config_values is not None
|
|
assert gateway.config_values["audio_normalization"] is False
|
|
|
|
|
|
def test_processing_options_default_to_audio_normalization_on() -> None:
|
|
assert ProcessingOptions().audio_normalization is True
|
|
|
|
|
|
def test_process_propagates_enabled_diarization_and_progress(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
audio = tmp_path / "meeting.flac"
|
|
audio.write_bytes(b"audio")
|
|
events = []
|
|
|
|
outcome = service.process(
|
|
audio,
|
|
meeting(),
|
|
participants(),
|
|
ProcessingOptions(diarization_enabled=True),
|
|
progress_sink=events.append,
|
|
)
|
|
|
|
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"),
|
|
]
|
|
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(
|
|
service.settings,
|
|
diarization_runtime="container",
|
|
diarization_container_image="runtime/image:tag",
|
|
diarization_container_args=(
|
|
"--network=host",
|
|
"--label",
|
|
"meeting-test",
|
|
),
|
|
)
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
outcome = service.process(
|
|
audio,
|
|
meeting(),
|
|
participants(),
|
|
ProcessingOptions(diarization_enabled=True),
|
|
)
|
|
|
|
assert outcome.succeeded
|
|
assert gateway.config_values is not None
|
|
assert gateway.config_values["diarization_container_args"] == (
|
|
"--network=host",
|
|
"--label",
|
|
"meeting-test",
|
|
)
|
|
|
|
|
|
def test_result_and_user_edit_are_preserved_separately(tmp_path: Path) -> None:
|
|
service, _ = make_service(tmp_path)
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
outcome = service.process(audio, meeting(), participants(), ProcessingOptions())
|
|
edited_path = service.save_edited_protocol(outcome.run_dir, "# Reviewed\n")
|
|
|
|
assert outcome.original_protocol == "# Generated protocol\n"
|
|
assert outcome.protocol_path.read_text(encoding="utf-8") == "# Generated protocol\n"
|
|
assert edited_path.read_text(encoding="utf-8") == "# Reviewed\n"
|
|
|
|
|
|
def test_failure_reports_stage_and_preserves_run_dir(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
gateway.fail = True
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
outcome = service.process(audio, meeting(), participants(), ProcessingOptions())
|
|
|
|
assert not outcome.succeeded
|
|
assert outcome.failed_stage == "transcription"
|
|
assert outcome.error_message == "TranscriptionError: model failed"
|
|
assert outcome.run_dir == gateway.run_dir
|
|
assert (gateway.run_dir / "run_metadata.json").is_file()
|
|
|
|
|
|
def test_speaker_review_lists_detected_labels_participants_and_excerpts(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
source = write_speaker_review_artifacts(gateway.run_dir)
|
|
|
|
review = service.load_speaker_mapping_review(gateway.run_dir, excerpts_per_speaker=2)
|
|
|
|
assert review is not None
|
|
assert [speaker.speaker_label for speaker in review.speakers] == [
|
|
"SPEAKER_00",
|
|
"SPEAKER_01",
|
|
]
|
|
assert review.speakers[0].excerpts == (
|
|
"We will run the production trial on Wednesday.",
|
|
"The trial requires the complete production team.",
|
|
)
|
|
assert review.participants == (("martin", "Martin"), ("anna", "Anna"))
|
|
assert review.current_mappings == {}
|
|
assert "SPEAKER_00" in source.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_speaker_review_prefers_content_rich_excerpts_over_acknowledgements_and_questions(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
acknowledgement = "Okay, that sounds good."
|
|
question = "Could we continue now?"
|
|
rich_excerpts = [
|
|
"I will complete the Atlas deployment in Berlin on 14 September with the database team.",
|
|
"I recommend moving the Orion product launch because the supplier test is incomplete.",
|
|
"We decided that I will own the Kubernetes migration for the payments service.",
|
|
"My experience with the Frankfurt factory rollout suggests a two-week validation period.",
|
|
"I will present the security audit results to the Apollo project steering group.",
|
|
"We should reserve twelve hours for the API integration and production monitoring.",
|
|
]
|
|
write_custom_speaker_review_artifacts(
|
|
gateway.run_dir,
|
|
[{"speaker_id": "SPEAKER_00", "text": text} for text in [acknowledgement, question, *rich_excerpts]],
|
|
)
|
|
|
|
review = service.load_speaker_mapping_review(gateway.run_dir)
|
|
|
|
assert review is not None
|
|
excerpts = review.speakers[0].excerpts
|
|
assert len(excerpts) == 6
|
|
assert acknowledgement not in excerpts
|
|
assert question not in excerpts
|
|
assert set(excerpts) == set(rich_excerpts)
|
|
|
|
|
|
def test_speaker_review_avoids_near_duplicate_excerpts(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
primary = "I will lead the Atlas API migration for the Berlin platform team next Monday."
|
|
duplicate = "I will lead the Atlas API migration for the Berlin platform group next Monday."
|
|
alternatives = [
|
|
"I recommend documenting the supplier quality decision in the project register.",
|
|
"My team will complete the production calibration before the factory trial.",
|
|
"I will present the customer research findings at the Munich planning workshop.",
|
|
"Our database monitoring threshold needs approval from the operations group.",
|
|
"I will coordinate the Orion release checklist with the support organization.",
|
|
]
|
|
write_custom_speaker_review_artifacts(
|
|
gateway.run_dir,
|
|
[{"speaker_id": "SPEAKER_00", "text": text} for text in [primary, duplicate, *alternatives]],
|
|
)
|
|
|
|
review = service.load_speaker_mapping_review(gateway.run_dir)
|
|
|
|
assert review is not None
|
|
excerpts = review.speakers[0].excerpts
|
|
assert primary in excerpts
|
|
assert duplicate not in excerpts
|
|
assert set(excerpts) == {primary, *alternatives}
|
|
|
|
|
|
def test_speaker_review_spreads_selected_excerpts_across_meeting_positions(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
positions = {
|
|
0: "I will lead the Atlas deployment review with the engineering team marker alpha.",
|
|
1: "I will lead the Atlas deployment review with the engineering team marker bravo.",
|
|
6: "I will lead the Atlas deployment review with the engineering team marker charlie.",
|
|
7: "I will lead the Atlas deployment review with the engineering team marker delta.",
|
|
12: "I will lead the Atlas deployment review with the engineering team marker echo.",
|
|
13: "I will lead the Atlas deployment review with the engineering team marker foxtrot.",
|
|
18: "I will lead the Atlas deployment review with the engineering team marker golf.",
|
|
19: "I will lead the Atlas deployment review with the engineering team marker hotel.",
|
|
}
|
|
segments = [
|
|
{
|
|
"speaker_id": "SPEAKER_00" if index in positions else "SPEAKER_01",
|
|
"text": positions.get(index, "Acknowledged."),
|
|
}
|
|
for index in range(20)
|
|
]
|
|
write_custom_speaker_review_artifacts(gateway.run_dir, segments)
|
|
|
|
review = service.load_speaker_mapping_review(gateway.run_dir)
|
|
|
|
assert review is not None
|
|
assert review.speakers[0].excerpts == tuple(positions[index] for index in (0, 1, 6, 7, 12, 18))
|
|
|
|
|
|
def test_protocol_only_regeneration_persists_mappings_without_rewriting_transcript(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
service.glossary.create("ENLYZE", "organization", aliases=("Enlyse",))
|
|
source = write_speaker_review_artifacts(gateway.run_dir)
|
|
original_source = source.read_bytes()
|
|
events = []
|
|
|
|
outcome = service.regenerate_protocol(
|
|
gateway.run_dir,
|
|
{"SPEAKER_00": "martin"},
|
|
progress_sink=events.append,
|
|
)
|
|
|
|
assert outcome.succeeded
|
|
assert outcome.original_protocol == "# Regenerated protocol\n"
|
|
assert outcome.speaker_attribution_available is True
|
|
assert gateway.context_data is not None
|
|
assert gateway.context_data["speaker_mappings"] == {"SPEAKER_00": "martin"}
|
|
assert gateway.context_data["known_entities"] == {
|
|
"Authoritative terminology": ["ENLYZE (aliases: Enlyse)"]
|
|
}
|
|
assert gateway.regeneration is not None
|
|
assert gateway.regeneration["meeting_context"].data["speaker_mappings"] == {
|
|
"SPEAKER_00": "martin"
|
|
}
|
|
assert gateway.regeneration["options"]["protocol_num_ctx"] == 32_768
|
|
assert gateway.regeneration["options"]["glossary_aliases"] == {"Enlyse": "ENLYZE"}
|
|
assert events[0].stage == "protocol_generation"
|
|
assert source.read_bytes() == original_source
|
|
|
|
|
|
def test_protocol_regeneration_allows_unmapped_and_rejects_duplicate_participant(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
write_speaker_review_artifacts(gateway.run_dir)
|
|
|
|
outcome = service.regenerate_protocol(gateway.run_dir, {})
|
|
|
|
assert outcome.succeeded
|
|
assert gateway.context_data is not None
|
|
assert gateway.context_data["speaker_mappings"] == {}
|
|
|
|
with pytest.raises(ValueError, match="only one speaker"):
|
|
service.regenerate_protocol(
|
|
gateway.run_dir,
|
|
{"SPEAKER_00": "martin", "SPEAKER_01": "martin"},
|
|
)
|
|
|
|
|
|
@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"
|
|
protocol_dir.mkdir(parents=True)
|
|
(protocol_dir / "runtime_metadata.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"speaker_attribution_available": False,
|
|
"speaker_attribution_loss_reason": "plain_transcript_fallback",
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert MeetingProcessingService._speaker_attribution_available(run_dir) is False
|
|
|
|
|
|
def test_uploaded_source_is_preserved_in_meeting_directory(tmp_path: Path) -> None:
|
|
service, _ = make_service(tmp_path)
|
|
source = SimpleNamespace(getbuffer=lambda: b"source audio")
|
|
|
|
destination = service.preserve_upload("meeting-1", "../unsafe.wav", source)
|
|
|
|
assert destination.parent == tmp_path / "meetings" / "meeting-1" / "uploads"
|
|
assert destination.name.endswith("_unsafe.wav")
|
|
assert destination.read_bytes() == b"source audio"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("profile", "threads"),
|
|
[("fast", 16), ("efficient", 10), ("powersave", 4)],
|
|
)
|
|
def test_process_resolves_profile_for_protocol_generation(
|
|
tmp_path: Path, profile: str, threads: int
|
|
) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
service.process(
|
|
audio,
|
|
meeting(),
|
|
participants(),
|
|
ProcessingOptions(performance_profile=profile),
|
|
)
|
|
|
|
assert gateway.config_values is not None
|
|
assert gateway.config_values["protocol_num_thread"] == threads
|
|
|
|
|
|
def test_process_defaults_to_backend_thread_selection(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
service.process(audio, meeting(), participants(), ProcessingOptions())
|
|
|
|
assert gateway.config_values is not None
|
|
assert gateway.config_values["protocol_num_thread"] is None
|
|
|
|
|
|
def test_protocol_regeneration_propagates_explicit_profile(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
write_speaker_review_artifacts(gateway.run_dir)
|
|
|
|
service.regenerate_protocol(
|
|
gateway.run_dir,
|
|
{},
|
|
performance_profile="fast",
|
|
)
|
|
|
|
assert gateway.regeneration is not None
|
|
assert gateway.regeneration["options"]["protocol_num_thread"] == 16
|
|
|
|
|
|
def test_process_passes_only_active_explicit_glossary_aliases(tmp_path: Path) -> None:
|
|
service, gateway = make_service(tmp_path)
|
|
service.glossary.create("Luminy", "product", aliases=("Lumini",))
|
|
inactive = service.glossary.create("PBAT", "acronym", aliases=("PBRT",))
|
|
service.glossary.set_active(inactive.id, False)
|
|
audio = tmp_path / "meeting.wav"
|
|
audio.write_bytes(b"audio")
|
|
|
|
service.process(audio, meeting(), participants(), ProcessingOptions())
|
|
|
|
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
|