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