import unittest import json import tempfile from pathlib import Path from unittest.mock import Mock from src.meeting_lab.consolidation.consolidate_facts import ( ConsolidationValidationError, DEFAULT_MIN_NUM_PREDICT, DEFAULT_NUM_PREDICT, build_consolidated_output, build_ollama_payload, call_ollama_with_repetition_retry, detect_repetitive_group_loop, estimate_response_tokens, fact_items, parse_model_json, repair_model_group_coverage, resolve_num_predict, validate_consolidated_output, validate_model_groups, ) def canonicalized_fixture(): fact_one = { "item_id": "fact_0001", "category": "fact", "text": "The lead maintains the list.", "evidence": "lead maintains the list", "source_file": "chunk_01_extraction.json", "source_index": 0, "original_value": "The lead maintains the list. | evidence", "source_references": [ { "source_file": "chunk_01_extraction.json", "source_index": 0, "evidence": "lead maintains the list", "original_value": "The lead maintains the list. | evidence", } ], "duplicate_count": 1, } fact_two = { "item_id": "fact_0002", "category": "fact", "text": "The head maintains the project list.", "evidence": "head maintains the project list", "source_file": "chunk_02_extraction.json", "source_index": 0, "original_value": "The head maintains the project list. | evidence", "source_references": [ { "source_file": "chunk_02_extraction.json", "source_index": 0, "evidence": "head maintains the project list", "original_value": "The head maintains the project list. | evidence", } ], "duplicate_count": 1, } decision = { "item_id": "decision_0001", "category": "decision", "text": "Ship it.", "evidence": "Agreed.", "source_file": "chunk_01_extraction.json", "source_index": 0, "original_value": "Ship it. | Agreed.", "source_references": [], "duplicate_count": 1, } return { "schema_version": "1", "source_files": ["chunk_01_extraction.json", "chunk_02_extraction.json"], "items": [fact_one, fact_two, decision], } class ConsolidateFactsTests(unittest.TestCase): @staticmethod def group(text="A", ids=None, reason="Singleton."): return { "canonical_text": text, "source_item_ids": ids or ["fact_0001"], "merge_reason": reason, } def test_payload_construction_disables_streaming_and_thinking_by_default(self): payload = build_ollama_payload( model="qwen3.5:9B", prompt="prompt", num_ctx=32768, num_predict=DEFAULT_NUM_PREDICT, think=False, ) self.assertEqual(payload["model"], "qwen3.5:9B") self.assertEqual(payload["prompt"], "prompt") self.assertIs(payload["stream"], False) self.assertIs(payload["think"], False) self.assertEqual(payload["format"], "json") self.assertEqual(payload["options"]["temperature"], 0.0) self.assertEqual(payload["options"]["num_ctx"], 32768) self.assertEqual(payload["options"]["num_predict"], DEFAULT_NUM_PREDICT) def test_payload_construction_can_enable_thinking_explicitly(self): payload = build_ollama_payload( model="qwen3.5:9B", prompt="prompt", num_ctx=16384, num_predict=1024, think=True, ) self.assertIs(payload["think"], True) self.assertEqual(payload["options"]["num_ctx"], 16384) self.assertEqual(payload["options"]["num_predict"], 1024) def test_explicit_num_predict_is_preserved(self): facts = [canonicalized_fixture()["items"][0]] self.assertEqual( resolve_num_predict( requested_num_predict=1234, facts=facts, prompt_token_estimate=100, num_ctx=32768, ), 1234, ) def test_adaptive_num_predict_scales_with_fact_payload(self): fixture = canonicalized_fixture() base_facts = fact_items(fixture) larger_facts = [] for index in range(80): item = dict(base_facts[index % len(base_facts)]) item["item_id"] = f"fact_{index + 1:04d}" item["text"] = item["text"] + " " + ("detail " * 20) item["evidence"] = item["evidence"] + " " + ("evidence " * 20) larger_facts.append(item) resolved = resolve_num_predict( requested_num_predict=None, facts=larger_facts, prompt_token_estimate=9000, num_ctx=32768, ) self.assertGreater(resolved, DEFAULT_MIN_NUM_PREDICT) self.assertLessEqual(resolved, 32768 - 9000 - 1024) def test_progeo_context_benchmark_needs_more_than_fixed_default_when_available(self): path = Path( "samples/benchmarks/progeo_meeting_context_v1_20260804_110913/" "canonicalizer/canonicalized_extractions.json" ) if not path.exists(): self.skipTest("Progeo context benchmark artifact is not available.") canonicalized = json.loads(path.read_text(encoding="utf-8-sig")) facts = fact_items(canonicalized) self.assertGreater(len(facts), 60) self.assertGreater(estimate_response_tokens(facts), DEFAULT_NUM_PREDICT) def test_grouping_validation_accepts_complete_singletons(self): groups = validate_model_groups( { "groups": [ { "canonical_text": "The lead maintains the list.", "source_item_ids": ["fact_0001"], "merge_reason": "Singleton.", }, { "canonical_text": "The head maintains the project list.", "source_item_ids": ["fact_0002"], "merge_reason": "Singleton.", }, ] }, {"fact_0001", "fact_0002"}, ) self.assertEqual(len(groups), 2) def test_valid_normal_consolidation_has_no_repetition_loop(self): text = json.dumps( {"groups": [self.group("A"), self.group("B", ["fact_0002"])]} ) detection = detect_repetitive_group_loop(text) self.assertFalse(detection["detected"]) self.assertEqual(detection["completed_group_count"], 2) def test_ordinary_duplicate_group_is_not_a_repetition_loop(self): duplicate = self.group() text = json.dumps({"groups": [duplicate, duplicate]}) detection = detect_repetitive_group_loop(text) self.assertFalse(detection["detected"]) self.assertEqual(detection["longest_consecutive_run"], 2) def test_repetitive_identical_group_sequence_is_detected_structurally(self): repeated = self.group("Same fact", ["fact_0001", "fact_0002"], "Equivalent.") text = json.dumps({"groups": [repeated, repeated, repeated]}) detection = detect_repetitive_group_loop(text) self.assertTrue(detection["detected"]) self.assertEqual(detection["longest_consecutive_run"], 3) self.assertEqual( detection["signature"], { "canonical_text": "Same fact", "source_item_ids": ["fact_0001", "fact_0002"], "merge_reason": "Equivalent.", }, ) def test_truncated_response_after_repetitive_loop_is_detected(self): repeated = json.dumps(self.group(), separators=(",", ":")) text = '{"groups":[' + ",".join([repeated] * 4) + ',{"canonical_text":' detection = detect_repetitive_group_loop(text) self.assertTrue(detection["detected"]) self.assertEqual(detection["completed_group_count"], 4) def test_detected_invalid_loop_gets_one_controlled_retry(self): repeated = json.dumps(self.group(), separators=(",", ":")) truncated = '{"groups":[' + ",".join([repeated] * 3) + ',{"canonical_text":' valid = json.dumps({"groups": [self.group()]}) call_fn = Mock( side_effect=[ (truncated, {"eval_count": 100, "done_reason": "length"}, 1.0), (valid, {"eval_count": 20, "done_reason": "stop"}, 0.5), ] ) with tempfile.TemporaryDirectory() as directory: raw_text, _data, runtime, attempts = call_ollama_with_repetition_retry( endpoint="http://example.invalid", model="test-model", prompt="base prompt", timeout=1, num_ctx=1000, num_predict=100, think=False, progress_interval=1, output_dir=Path(directory), call_fn=call_fn, ) self.assertTrue((Path(directory) / "raw_model_response.txt").exists()) self.assertTrue((Path(directory) / "raw_model_response_retry.txt").exists()) metadata = json.loads( (Path(directory) / "repetition_retry_metadata.json").read_text() ) self.assertEqual(raw_text, valid) self.assertEqual(runtime, 1.5) self.assertEqual(len(attempts), 2) self.assertEqual(call_fn.call_count, 2) self.assertEqual(metadata["attempts"][0]["repetition"]["detected"], True) first_prompt = call_fn.call_args_list[0].kwargs["prompt"] retry_prompt = call_fn.call_args_list[1].kwargs["prompt"] self.assertEqual(first_prompt, "base prompt") self.assertIn("Emit each semantic group exactly once", retry_prompt) def test_unrelated_malformed_json_is_not_retried(self): call_fn = Mock(return_value=("{invalid", {"done_reason": "length"}, 0.1)) with tempfile.TemporaryDirectory() as directory: with self.assertRaises(ConsolidationValidationError): call_ollama_with_repetition_retry( endpoint="http://example.invalid", model="test-model", prompt="prompt", timeout=1, num_ctx=1000, num_predict=100, think=False, progress_interval=1, output_dir=Path(directory), call_fn=call_fn, ) self.assertEqual(call_fn.call_count, 1) def test_retry_failure_preserves_both_attempts_and_raises(self): repeated = json.dumps(self.group(), separators=(",", ":")) truncated = '{"groups":[' + ",".join([repeated] * 3) + ',{"canonical_text":' call_fn = Mock( side_effect=[ (truncated, {"done_reason": "length"}, 1.0), (truncated, {"done_reason": "length"}, 1.0), ] ) with tempfile.TemporaryDirectory() as directory: root = Path(directory) with self.assertRaises(ConsolidationValidationError): call_ollama_with_repetition_retry( endpoint="http://example.invalid", model="test-model", prompt="prompt", timeout=1, num_ctx=1000, num_predict=100, think=False, progress_interval=1, output_dir=root, call_fn=call_fn, ) self.assertTrue((root / "raw_model_response.txt").exists()) self.assertTrue((root / "raw_ollama_response.json").exists()) self.assertTrue((root / "raw_model_response_retry.txt").exists()) self.assertTrue((root / "raw_ollama_response_retry.json").exists()) metadata = json.loads((root / "repetition_retry_metadata.json").read_text()) self.assertEqual(call_fn.call_count, 2) self.assertEqual(len(metadata["attempts"]), 2) self.assertIn("parse_error", metadata["attempts"][1]) def test_no_missing_source_ids(self): with self.assertRaisesRegex(ConsolidationValidationError, "Missing"): validate_model_groups( { "groups": [ { "canonical_text": "Only one.", "source_item_ids": ["fact_0001"], "merge_reason": "Singleton.", } ] }, {"fact_0001", "fact_0002"}, ) def test_no_duplicate_source_ids(self): with self.assertRaisesRegex(ConsolidationValidationError, "multiple"): validate_model_groups( { "groups": [ { "canonical_text": "One.", "source_item_ids": ["fact_0001"], "merge_reason": "Singleton.", }, { "canonical_text": "Again.", "source_item_ids": ["fact_0001"], "merge_reason": "Singleton.", }, ] }, {"fact_0001"}, ) def test_source_coverage_repair_restores_missing_singletons(self): fixture = canonicalized_fixture() facts = fact_items(fixture) repaired, changes = repair_model_group_coverage( { "groups": [ { "canonical_text": "The lead maintains the list.", "source_item_ids": ["fact_0001"], "merge_reason": "Singleton.", } ] }, facts, ) groups = validate_model_groups(repaired, {"fact_0001", "fact_0002"}) self.assertEqual(len(groups), 2) self.assertEqual(groups[1]["source_item_ids"], ["fact_0002"]) self.assertEqual( changes[0]["operation"], "restore_missing_source_id_as_singleton", ) def test_source_coverage_repair_removes_duplicate_occurrences(self): fixture = canonicalized_fixture() facts = fact_items(fixture) repaired, changes = repair_model_group_coverage( { "groups": [ { "canonical_text": "Merged.", "source_item_ids": ["fact_0001", "fact_0002"], "merge_reason": "Same.", }, { "canonical_text": "Duplicate.", "source_item_ids": ["fact_0002"], "merge_reason": "Duplicate.", }, ] }, facts, ) groups = validate_model_groups(repaired, {"fact_0001", "fact_0002"}) self.assertEqual(len(groups), 1) self.assertEqual(groups[0]["source_item_ids"], ["fact_0001", "fact_0002"]) self.assertEqual( [change["operation"] for change in changes], ["remove_duplicate_source_id", "remove_empty_group"], ) def test_merged_group_validation(self): groups = validate_model_groups( { "groups": [ { "canonical_text": "The lead maintains the project list.", "source_item_ids": ["fact_0001", "fact_0002"], "merge_reason": "Same proposition.", } ] }, {"fact_0001", "fact_0002"}, ) output = build_consolidated_output(canonicalized_fixture(), groups) validate_consolidated_output(canonicalized_fixture(), output) self.assertEqual(output["items"][0]["source_item_ids"], ["fact_0001", "fact_0002"]) def test_preservation_of_non_fact_categories(self): fixture = canonicalized_fixture() groups = validate_model_groups( { "groups": [ { "canonical_text": "The lead maintains the project list.", "source_item_ids": ["fact_0001", "fact_0002"], "merge_reason": "Same proposition.", } ] }, {"fact_0001", "fact_0002"}, ) output = build_consolidated_output(fixture, groups) self.assertEqual(output["items"][1:], fixture["items"][2:]) def test_invalid_model_json(self): with self.assertRaisesRegex(ConsolidationValidationError, "Invalid model JSON"): parse_model_json("{invalid") def test_unknown_source_item_ids(self): with self.assertRaisesRegex(ConsolidationValidationError, "unknown"): validate_model_groups( { "groups": [ { "canonical_text": "Unknown.", "source_item_ids": ["fact_9999"], "merge_reason": "Bad ID.", } ] }, {"fact_0001"}, ) if __name__ == "__main__": unittest.main()