607 lines
22 KiB
Python
607 lines
22 KiB
Python
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_source_coverage_repair_keeps_valid_ids_unchanged(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
model_output = {
|
|
"groups": [self.group("Merged", ["fact_0001", "fact_0002"], "Same.")]
|
|
}
|
|
|
|
repaired, changes = repair_model_group_coverage(model_output, facts)
|
|
|
|
self.assertEqual(repaired, model_output)
|
|
self.assertEqual(changes, [])
|
|
validate_model_groups(repaired, {"fact_0001", "fact_0002"})
|
|
|
|
def test_source_coverage_repair_removes_unknown_id_but_keeps_valid_id(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
repaired, changes = repair_model_group_coverage(
|
|
{
|
|
"groups": [
|
|
self.group("Supported", ["fact_0001", "fact_9999"], "Partial.")
|
|
]
|
|
},
|
|
facts,
|
|
)
|
|
|
|
self.assertEqual(repaired["groups"][0]["source_item_ids"], ["fact_0001"])
|
|
self.assertEqual(changes[0]["operation"], "remove_unknown_source_id")
|
|
self.assertEqual(changes[0]["id"], "fact_9999")
|
|
|
|
def test_source_coverage_repair_removes_multiple_unknown_ids(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
repaired, changes = repair_model_group_coverage(
|
|
{
|
|
"groups": [
|
|
self.group(
|
|
"Supported",
|
|
["unknown_a", "fact_0001", "unknown_b", "fact_0002"],
|
|
"Partial.",
|
|
)
|
|
]
|
|
},
|
|
facts,
|
|
)
|
|
|
|
self.assertEqual(
|
|
repaired["groups"][0]["source_item_ids"],
|
|
["fact_0001", "fact_0002"],
|
|
)
|
|
self.assertEqual(
|
|
[change["id"] for change in changes],
|
|
["unknown_a", "unknown_b"],
|
|
)
|
|
|
|
def test_source_coverage_repair_drops_group_with_only_unknown_ids(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
repaired, changes = repair_model_group_coverage(
|
|
{"groups": [self.group("Unsupported", ["unknown_a"], "Unsupported.")]},
|
|
facts,
|
|
)
|
|
|
|
self.assertNotIn("Unsupported", [group["canonical_text"] for group in repaired["groups"]])
|
|
self.assertEqual(
|
|
[change["operation"] for change in changes[:2]],
|
|
["remove_unknown_source_id", "remove_empty_group"],
|
|
)
|
|
validate_model_groups(repaired, {"fact_0001", "fact_0002"})
|
|
|
|
def test_unknown_id_removal_that_empties_group_is_recorded(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
_repaired, changes = repair_model_group_coverage(
|
|
{
|
|
"groups": [
|
|
self.group("Unsupported", ["unknown_a", "unknown_b"], "Unsupported.")
|
|
]
|
|
},
|
|
facts,
|
|
)
|
|
|
|
self.assertEqual(
|
|
[change["operation"] for change in changes[:3]],
|
|
[
|
|
"remove_unknown_source_id",
|
|
"remove_unknown_source_id",
|
|
"remove_empty_group",
|
|
],
|
|
)
|
|
|
|
def test_unknown_id_removal_interacts_with_duplicate_repair(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
repaired, changes = repair_model_group_coverage(
|
|
{
|
|
"groups": [
|
|
self.group("First", ["fact_0001", "unknown_a"], "First."),
|
|
self.group("Second", ["fact_0001", "fact_0002"], "Second."),
|
|
]
|
|
},
|
|
facts,
|
|
)
|
|
|
|
self.assertEqual(repaired["groups"][0]["source_item_ids"], ["fact_0001"])
|
|
self.assertEqual(repaired["groups"][1]["source_item_ids"], ["fact_0002"])
|
|
self.assertEqual(
|
|
[change["operation"] for change in changes],
|
|
["remove_unknown_source_id", "remove_duplicate_source_id"],
|
|
)
|
|
validate_model_groups(repaired, {"fact_0001", "fact_0002"})
|
|
|
|
def test_unknown_removal_and_missing_id_singleton_restoration(self):
|
|
facts = fact_items(canonicalized_fixture())
|
|
repaired, changes = repair_model_group_coverage(
|
|
{
|
|
"groups": [
|
|
self.group("Supported", ["fact_0001", "unknown_a"], "Partial.")
|
|
]
|
|
},
|
|
facts,
|
|
)
|
|
|
|
self.assertEqual(repaired["groups"][-1]["source_item_ids"], ["fact_0002"])
|
|
self.assertEqual(
|
|
[change["operation"] for change in changes],
|
|
["remove_unknown_source_id", "restore_missing_source_id_as_singleton"],
|
|
)
|
|
validate_model_groups(repaired, {"fact_0001", "fact_0002"})
|
|
|
|
def test_final_validator_still_rejects_unknown_source_id(self):
|
|
with self.assertRaisesRegex(ConsolidationValidationError, "unknown"):
|
|
validate_model_groups(
|
|
{"groups": [self.group("Unsupported", ["fact_9999"], "Bad ID.")]},
|
|
{"fact_0001"},
|
|
)
|
|
|
|
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()
|