from __future__ import annotations import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.domain_case_loop import ( build_scenario_acceptance_matrix, carry_forward_analysis_context, derive_pack_final_status, load_scenario_pack, merge_scenario_date_scope, ) def test_carry_forward_analysis_context_preserves_followup_anchor() -> None: scenario_state = { "semantic_memory": { "date_scope": {"as_of_date": "2020-03-31"}, } } analysis_context = {"as_of_date": "2026-04-13", "source": "current_analysis"} carried = carry_forward_analysis_context(scenario_state, analysis_context) assert carried["as_of_date"] == "2026-04-13" assert carried["source"] == "current_analysis" def test_carry_forward_analysis_context_fills_missing_anchor() -> None: scenario_state = { "semantic_memory": { "date_scope": {"as_of_date": "2020-03-31"}, } } carried = carry_forward_analysis_context(scenario_state, {}) assert carried["as_of_date"] == "2020-03-31" assert carried["source"] == "scenario_state_carryover" def test_merge_scenario_date_scope_preserves_historical_anchor_on_followup() -> None: previous_date_scope = {"as_of_date": "2020-03-31", "source": "exact_anchor"} current_date_scope = {"as_of_date": "2026-04-13", "source": "current_analysis"} merged = merge_scenario_date_scope( previous_date_scope, current_date_scope, depends_on=["step_01_anchor"], ) assert merged["as_of_date"] == "2020-03-31" assert merged["source"] == "current_analysis" def test_load_scenario_pack_accepts_active_domain_contract(tmp_path) -> None: manifest_path = tmp_path / "active_domain_contract.json" manifest_path.write_text( json.dumps( { "schema_version": "active_domain_contract_v1", "status": "active", "domain_id": "inventory_stock_supplier_provenance", "runtime_domain": "inventory_stock", "title": "Warehouse domain", "default_analysis_context": {"as_of_date": "2021-09-30"}, "observed_anchors": { "warehouse": "Основной склад", "organization": "ООО \\Альтернатива Плюс\\", "historical_as_of_date": "2019-03-31", "current_as_of_date_example": "2021-09-30", "focus_item_historical": "Столешница 600*3050*26 дуб ниагара", }, "question_pool": { "questions": [ {"question_id": "Q01", "node_id": "N01_stock_snapshot", "text": "Q1"}, {"question_id": "Q19", "node_id": "N03_selected_item_supplier", "text": "Q19"}, ] }, "scenario_tree": { "critical_edges": [ { "edge_id": "E01_snapshot_to_selected_item_supplier", "from_node": "N01_stock_snapshot", "to_node": "N03_selected_item_supplier", "primary_user_path": True, } ] }, "orchestration_pack": { "pack_id": "inventory_active_contract_smoke", "scenarios": [ { "scenario_id": "inventory_selected_item_provenance", "title": "Selected item provenance", "question_ids": ["Q01", "Q19"], "steps": [ { "step_id": "step_01_snapshot", "question_id": "Q01", "node_id": "N01_stock_snapshot", "question": "Какие товары сейчас лежат на складе", }, { "step_id": "step_02_supplier", "question_id": "Q19", "node_id": "N03_selected_item_supplier", "question": "По выбранному объекту \"Столешница 600*3050*26 дуб ниагара\": кто это поставил нам", }, ], } ], }, }, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) pack = load_scenario_pack(manifest_path) assert pack["schema_version"] == "domain_scenario_pack_v1" assert pack["source_schema_version"] == "active_domain_contract_v1" assert pack["domain"] == "inventory_stock" assert pack["bindings"]["observed_warehouse"] == "Основной склад" assert pack["bindings"]["focus_item_historical"] == "Столешница 600*3050*26 дуб ниагара" assert pack["scenarios"][0]["question_ids"] == ["Q01", "Q19"] assert pack["scenarios"][0]["steps"][1]["question_id"] == "Q19" def test_build_scenario_acceptance_matrix_marks_green_edge_when_covering_scenario_is_accepted() -> None: pack = { "pack_id": "inventory_active_contract_smoke", "domain": "inventory_stock", "source_contract": {"domain_id": "inventory_stock_supplier_provenance", "title": "Warehouse domain"}, "question_pool": { "questions": [ {"question_id": "Q01", "node_id": "N01_stock_snapshot"}, {"question_id": "Q19", "node_id": "N03_selected_item_supplier"}, ] }, "scenario_tree": { "critical_nodes": [ { "node_id": "N03_selected_item_supplier", "covers_question_ids": ["Q19"], "required_wording_families": ["canonical"], } ], "critical_edges": [ { "edge_id": "E01_snapshot_to_selected_item_supplier", "from_node": "N01_stock_snapshot", "to_node": "N03_selected_item_supplier", "primary_user_path": True, } ], "primary_user_paths": [ {"path_id": "P01_snapshot_to_supplier", "nodes": ["N01_stock_snapshot", "N03_selected_item_supplier"]} ], }, "scenarios": [ { "scenario_id": "inventory_selected_item_provenance", "question_ids": ["Q01", "Q19"], "steps": [ { "step_id": "step_01_snapshot", "question_id": "Q01", "node_id": "N01_stock_snapshot", "paraphrase_family": "canonical", }, { "step_id": "step_02_supplier", "question_id": "Q19", "node_id": "N03_selected_item_supplier", "paraphrase_family": "canonical", }, ], } ], } scenario_results = [ { "scenario_id": "inventory_selected_item_provenance", "final_status": "accepted", "session_id": "asst-demo", "artifact_dir": "artifacts/domain_runs/demo", } ] matrix = build_scenario_acceptance_matrix(pack, scenario_results) assert "E01_snapshot_to_selected_item_supplier" in matrix assert "| E01_snapshot_to_selected_item_supplier | green |" in matrix assert "| P01_snapshot_to_supplier | green |" in matrix def test_build_scenario_acceptance_matrix_marks_partial_when_wording_family_is_missing() -> None: pack = { "pack_id": "inventory_active_contract_smoke", "domain": "inventory_stock", "source_contract": {"domain_id": "inventory_stock_supplier_provenance", "title": "Warehouse domain"}, "question_pool": { "questions": [ {"question_id": "Q19", "node_id": "N03_selected_item_supplier"}, ] }, "scenario_tree": { "critical_nodes": [ { "node_id": "N03_selected_item_supplier", "covers_question_ids": ["Q19"], "required_wording_families": ["canonical", "ui_selected_object_colloquial"], } ] }, "scenarios": [ { "scenario_id": "inventory_selected_item_provenance", "question_ids": ["Q19"], "steps": [ { "step_id": "step_01_supplier", "question_id": "Q19", "node_id": "N03_selected_item_supplier", "paraphrase_family": "canonical", } ], } ], } scenario_results = [ { "scenario_id": "inventory_selected_item_provenance", "final_status": "accepted", "session_id": "asst-demo", "artifact_dir": "artifacts/domain_runs/demo", } ] matrix = build_scenario_acceptance_matrix(pack, scenario_results) assert "| N03_selected_item_supplier | partial |" in matrix def test_derive_pack_final_status_downgrades_accepted_when_matrix_contains_partial_coverage() -> None: pack = { "pack_id": "inventory_active_contract_smoke", "domain": "inventory_stock", "scenarios": [ { "scenario_id": "inventory_selected_item_provenance", "question_ids": ["Q19"], "steps": [ { "step_id": "step_01_supplier", "question_id": "Q19", "node_id": "N03_selected_item_supplier", "paraphrase_family": "canonical", } ], }, ], "scenario_tree": { "critical_nodes": [ { "node_id": "N03_selected_item_supplier", "covers_question_ids": ["Q19"], "required_wording_families": ["canonical", "ui_selected_object_colloquial"], } ] }, "question_pool": { "questions": [ {"question_id": "Q19", "node_id": "N03_selected_item_supplier"}, ] }, } scenario_results = [ { "scenario_id": "inventory_selected_item_provenance", "final_status": "accepted", "session_id": "asst-demo", "artifact_dir": "artifacts/domain_runs/demo", } ] assert derive_pack_final_status(pack, scenario_results) == "partial"