224 lines
8.6 KiB
Python
224 lines
8.6 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
import agent_runtime_manifest as runtime_manifest
|
||
import save_agent_semantic_run as saver
|
||
|
||
|
||
def mojibake(value: str) -> str:
|
||
return value.encode("utf-8").decode("cp1251")
|
||
|
||
|
||
def write_json(path: Path, payload: object) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
class SaveAgentSemanticRunTests(unittest.TestCase):
|
||
def write_clean_truth_run(self, run_dir: Path, *, include_runtime: bool) -> None:
|
||
write_json(
|
||
run_dir / "pack_state.json",
|
||
{
|
||
"final_status": "accepted",
|
||
"review_overall_status": "pass",
|
||
"acceptance_gate_passed": True,
|
||
"no_unresolved_p0": True,
|
||
"unresolved_p0_count": 0,
|
||
"steps_total": 1,
|
||
"steps_passed": 1,
|
||
"steps_failed": 0,
|
||
},
|
||
)
|
||
write_json(run_dir / "truth_review.json", {"summary": {"overall_status": "pass"}})
|
||
write_json(
|
||
run_dir / "business_review.json",
|
||
{
|
||
"overall_business_status": "pass",
|
||
"steps_with_business_failures": 0,
|
||
"steps_with_business_warnings": 0,
|
||
},
|
||
)
|
||
if include_runtime:
|
||
write_json(
|
||
run_dir / runtime_manifest.EFFECTIVE_RUNTIME_FILE_NAME,
|
||
{
|
||
"schema_version": runtime_manifest.EFFECTIVE_RUNTIME_SCHEMA_VERSION,
|
||
"runner": "domain_truth_harness.run-live",
|
||
"git_sha": "test-sha",
|
||
"llm_provider": "local",
|
||
"llm_model": "test-model",
|
||
"temperature": 0.0,
|
||
"max_output_tokens": 2048,
|
||
"prompt_version": "normalizer_v2_0_2",
|
||
"prompt_source": "file",
|
||
"prompt_hash": "abc123",
|
||
"prompt_registry_status": "pass",
|
||
},
|
||
)
|
||
|
||
def test_validate_truth_harness_run_refuses_missing_effective_runtime(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
run_dir = Path(tmp)
|
||
self.write_clean_truth_run(run_dir, include_runtime=False)
|
||
|
||
with self.assertRaisesRegex(RuntimeError, "reproducibility manifest"):
|
||
saver.validate_truth_harness_run_dir(run_dir)
|
||
|
||
def test_validate_truth_harness_run_includes_effective_runtime_summary(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
run_dir = Path(tmp)
|
||
self.write_clean_truth_run(run_dir, include_runtime=True)
|
||
|
||
metadata = saver.validate_truth_harness_run_dir(run_dir)
|
||
|
||
self.assertEqual(metadata["validation_status"], "accepted_live_replay")
|
||
self.assertEqual(metadata["effective_runtime"]["runner"], "domain_truth_harness.run-live")
|
||
self.assertEqual(metadata["effective_runtime"]["llm_model"], "test-model")
|
||
|
||
def write_domain_case_loop_pack(self, run_dir: Path, *, target_count: int = 0, p1_count: int = 0) -> None:
|
||
write_json(
|
||
run_dir / "pack_state.json",
|
||
{
|
||
"pack_id": "agent_pack",
|
||
"execution_status": "exact",
|
||
"acceptance_status": "accepted",
|
||
"final_status": "accepted",
|
||
"scenario_results": [{"scenario_id": "s01", "final_status": "accepted"}],
|
||
},
|
||
)
|
||
write_json(
|
||
run_dir / "repair_targets.json",
|
||
{
|
||
"target_count": target_count,
|
||
"acceptance_status": "accepted",
|
||
"final_status": "accepted",
|
||
"severity_counts": {"P0": 0, "P1": p1_count, "P2": 0},
|
||
},
|
||
)
|
||
write_json(
|
||
run_dir / runtime_manifest.EFFECTIVE_RUNTIME_FILE_NAME,
|
||
{
|
||
"schema_version": runtime_manifest.EFFECTIVE_RUNTIME_SCHEMA_VERSION,
|
||
"runner": "domain_case_loop.run-pack",
|
||
"git_sha": "test-sha",
|
||
"llm_provider": "local",
|
||
"llm_model": "test-model",
|
||
"temperature": 0.0,
|
||
"max_output_tokens": 2048,
|
||
"prompt_version": "normalizer_v2_0_2",
|
||
"prompt_source": "file",
|
||
"prompt_hash": "abc123",
|
||
"prompt_registry_status": "pass",
|
||
},
|
||
)
|
||
|
||
def test_validate_domain_case_loop_pack_accepts_clean_pack(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
run_dir = Path(tmp)
|
||
self.write_domain_case_loop_pack(run_dir)
|
||
|
||
metadata = saver.validate_accepted_run_dir(run_dir)
|
||
|
||
self.assertEqual(metadata["validation_status"], "accepted_domain_case_loop_pack")
|
||
self.assertEqual(metadata["effective_runtime"]["runner"], "domain_case_loop.run-pack")
|
||
self.assertEqual(metadata["repair_target_count"], 0)
|
||
|
||
def test_validate_domain_case_loop_pack_refuses_open_repair_targets(self) -> None:
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
run_dir = Path(tmp)
|
||
self.write_domain_case_loop_pack(run_dir, target_count=1, p1_count=1)
|
||
|
||
with self.assertRaisesRegex(RuntimeError, "not clean"):
|
||
saver.validate_accepted_run_dir(run_dir)
|
||
|
||
def test_extract_questions_resolves_scenario_pack_bindings(self) -> None:
|
||
spec = {
|
||
"schema_version": "domain_scenario_pack_v1",
|
||
"bindings": {
|
||
"main_organization": mojibake("ООО Альтернатива Плюс"),
|
||
"control_year": "2020",
|
||
"svk_counterparty": mojibake("Группа СВК"),
|
||
},
|
||
"scenarios": [
|
||
{
|
||
"scenario_id": "biz",
|
||
"steps": [
|
||
{
|
||
"question": mojibake("Дай обзор {{bindings.main_organization}} за {{bindings.control_year}} год."),
|
||
"semantic_tags": ["business_overview", "money"],
|
||
},
|
||
{
|
||
"question": mojibake("Отдельно по {{bindings.svk_counterparty}} покажи документы."),
|
||
"semantic_tags": ["counterparty", "documents"],
|
||
},
|
||
],
|
||
}
|
||
],
|
||
}
|
||
|
||
questions = saver.extract_questions_from_spec(spec)
|
||
|
||
self.assertEqual(
|
||
questions,
|
||
[
|
||
"Дай обзор ООО Альтернатива Плюс за 2020 год.",
|
||
"Отдельно по Группа СВК покажи документы.",
|
||
],
|
||
)
|
||
self.assertFalse(any("{{bindings." in question for question in questions))
|
||
self.assertEqual(
|
||
saver.extract_semantic_tags(spec),
|
||
["business_overview", "counterparty", "documents", "money"],
|
||
)
|
||
|
||
def test_extract_questions_repairs_mojibake_before_saved_session(self) -> None:
|
||
spec = {
|
||
"schema_version": "domain_truth_harness_spec_v1",
|
||
"bindings": {
|
||
"organization": mojibake('ООО "Альтернатива Плюс"'),
|
||
},
|
||
"steps": [
|
||
{
|
||
"question": mojibake("Какие остатки на складе у {{bindings.organization}}?"),
|
||
"semantic_tags": ["inventory"],
|
||
}
|
||
],
|
||
}
|
||
|
||
questions = saver.extract_questions_from_spec(spec)
|
||
snapshot = saver.build_snapshot_payload(
|
||
generation_id="gen-test",
|
||
title="AGENT | test",
|
||
questions=questions,
|
||
metadata={},
|
||
)
|
||
|
||
self.assertEqual(
|
||
questions,
|
||
['Какие остатки на складе у ООО "Альтернатива Плюс"?'],
|
||
)
|
||
self.assertEqual(snapshot["questions"], questions)
|
||
self.assertEqual(snapshot["session"]["items"][0]["text"], questions[0])
|
||
self.assertNotIn("Рљ", questions[0])
|
||
|
||
def test_extract_questions_refuses_unresolved_bindings(self) -> None:
|
||
spec = {
|
||
"questions": ["Что с НДС за {{bindings.control_year}} год?"],
|
||
"bindings": {},
|
||
}
|
||
|
||
with self.assertRaisesRegex(RuntimeError, "unresolved bindings"):
|
||
saver.extract_questions_from_spec(spec)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|