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()