from __future__ import annotations import os import shutil import sys import tempfile import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import domain_case_loop as dcl import domain_truth_harness as dth class DomainCaseLoopStepStateTests(unittest.TestCase): def test_scenario_semantic_memory_keeps_selected_inventory_bundle(self) -> None: step_state = { "step_id": "step_06_pronoun_selected_item_supplier", "status": "validated", "detected_intent": "inventory_purchase_provenance_for_item", "selected_recipe": "address_inventory_purchase_provenance_for_item_v1", "last_confirmed_route": "address_inventory_purchase_provenance_for_item_v1", "active_result_set_id": "rs-msg-demo", "organization_scope": 'ООО "Альтернатива Плюс"', "date_scope": { "as_of_date": "2017-12-31", "period_from": "2017-01-01", "period_to": "2017-12-31", }, "focus_object": { "object_type": "item", "object_id": "item:flag", "label": "Флаг геральдический. 100*150 см", }, "entries": [ { "title": "Поступление товаров и услуг 00000000016 от 01.06.2017", "fields": { "date": "01.06.2017", "контрагент": "ПрофТренд,ООО", }, } ], } memory = dcl.build_scenario_semantic_memory_after_step( previous_semantic_memory={}, step={"step_id": "step_06_pronoun_selected_item_supplier", "depends_on": ["step_05"]}, step_state=step_state, ) self.assertEqual(memory["focus_object"]["label"], "Флаг геральдический. 100*150 см") self.assertEqual(memory["selected_object_ref"]["object_id"], "item:flag") self.assertEqual(memory["supplier_if_known"], "ПрофТренд,ООО") self.assertEqual(memory["first_purchase_date"], "01.06.2017") self.assertEqual(memory["provenance_bundle"]["source_intent"], "inventory_purchase_provenance_for_item") self.assertEqual( memory["provenance_bundle"]["source_document_if_known"], "Поступление товаров и услуг 00000000016 от 01.06.2017", ) @unittest.skipIf(os.name != "nt", "extended-length path handling is Windows-specific") def test_write_json_supports_long_windows_artifact_paths(self) -> None: temp_dir = tempfile.mkdtemp() try: base = Path(temp_dir) long_path = ( base / ("artifact_root_" + "a" * 80) / ("scenario_" + "b" * 80) / ("step_" + "c" * 80) / "assistant_response.json" ) dcl.write_json(long_path, {"ok": True}) extended_path = dcl.resolve_windows_extended_path(long_path) self.assertTrue(extended_path.exists()) self.assertIn('"ok": true', extended_path.read_text(encoding="utf-8-sig")) finally: shutil.rmtree(dcl.resolve_windows_extended_path(Path(temp_dir)), ignore_errors=True) def test_preserves_mcp_catalog_alignment_debug_fields(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="planner_alignment_demo", domain="planner_autonomy", step={ "step_id": "step_01", "title": "Alignment visibility", "depends_on": [], "question_template": "show planner alignment", }, step_index=1, question_resolved="show planner alignment", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Confirmed answer", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "counterparty_turnover", "selected_recipe": "counterparty_turnover_by_period", "capability_id": "confirmed_counterparty_turnover", "mcp_discovery_catalog_chain_alignment_status": "selected_matches_top", "mcp_discovery_catalog_chain_top_match": "value_flow", "mcp_discovery_catalog_chain_selected_matches_top": True, "mcp_discovery_route_candidate_status": "needs_user_scope", "mcp_discovery_route_candidate_fact_family": "value_flow", "mcp_discovery_route_candidate_action_family": "turnover", "mcp_discovery_route_candidate_missing_axes": ["organization"], "mcp_discovery_route_candidate_provided_axes": ["period"], "mcp_discovery_route_candidate_executable_now": False, "mcp_discovery_route_candidate_enablement_reason": "Missing scope axes: organization", "mcp_discovery_route_candidate_next_action": "Ask the user for the missing scope axes before MCP execution.", }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["mcp_discovery_catalog_chain_alignment_status"], "selected_matches_top") self.assertEqual(step_state["mcp_discovery_catalog_chain_top_match"], "value_flow") self.assertTrue(step_state["mcp_discovery_catalog_chain_selected_matches_top"]) self.assertEqual(step_state["mcp_discovery_route_candidate_status"], "needs_user_scope") self.assertEqual(step_state["mcp_discovery_route_candidate_missing_axes"], ["organization"]) self.assertEqual(step_state["mcp_discovery_route_candidate_provided_axes"], ["period"]) self.assertFalse(step_state["mcp_discovery_route_candidate_executable_now"]) def test_scenario_execution_status_treats_validated_meta_and_guarded_partial_as_complete(self) -> None: step_outputs = { "smalltalk": { "execution_status": "needs_exact_capability", "acceptance_status": "validated", "clean_meta_chat_answer_validated": True, }, "exact_inventory": { "execution_status": "exact", "acceptance_status": "validated", }, "account_60_boundary": { "execution_status": "partial", "acceptance_status": "validated", "runtime_factual_answer_validated": True, }, } self.assertEqual(dcl.derive_scenario_execution_status(step_outputs), "partial") self.assertEqual(dcl.derive_scenario_status(step_outputs), "accepted") def test_today_scope_required_filter_and_direct_patterns_are_enforced(self) -> None: self.assertTrue(dcl.question_resets_temporal_scope("мы должны комуто денег на сегодня?")) step = dcl.normalize_step_definition( 1, { "step_id": "payables_today", "title": "Payables today", "question": "мы должны комуто денег на сегодня?", "expected_intents": ["payables_confirmed_as_of_date"], "required_filters": {"as_of_date": "{{runtime.today_iso}}"}, "required_direct_answer_patterns_any": ["{{runtime.today_dot_regex}}"], }, ) runtime_today = dcl.build_runtime_bindings()["today_iso"] runtime_today_pattern = dcl.build_runtime_bindings()["today_dot_regex"] self.assertEqual(step["required_filters"]["as_of_date"], runtime_today) self.assertEqual(step["required_answer_patterns_any"], [runtime_today_pattern]) step_state = dcl.validate_step_contract( { "execution_status": "exact", "reply_type": "factual", "detected_intent": "payables_confirmed_as_of_date", "required_filters": step["required_filters"], "required_answer_patterns_any": step["required_answer_patterns_any"], "extracted_filters": {"as_of_date": "2017-05-31"}, "assistant_text": "на 31.05.2017 мы должны 3.433.472,35 ₽.", "actual_direct_answer": "на 31.05.2017 мы должны 3.433.472,35 ₽.", "top_non_empty_lines": ["на 31.05.2017 мы должны 3.433.472,35 ₽."], } ) self.assertIn("wrong_as_of_date", step_state["violated_invariants"]) self.assertIn("required_answer_patterns_any_missing", step_state["violated_invariants"]) self.assertEqual(step_state["acceptance_status"], "rejected") def test_repair_targets_promote_route_candidate_enablement_gaps(self) -> None: repair_targets = dcl.build_deterministic_repair_targets( {"pack_id": "route_candidate_pack", "domain": "open_world", "final_status": "accepted"}, [ { "scenario_id": "route_candidate_demo", "title": "Route candidate demo", "artifact_dir": "artifacts/domain_runs/route_candidate_demo", "scenario_state": { "step_outputs": { "step_01": { "status": "accepted", "acceptance_status": "accepted", "execution_status": "validated", "question_resolved": "можно ли построить анализ качества поставщиков за 2020?", "reply_type": "partial_coverage", "mcp_discovery_selected_chain_id": "business_overview", "mcp_discovery_route_candidate_status": "needs_route_enablement", "mcp_discovery_route_candidate_fact_family": "supplier_quality", "mcp_discovery_route_candidate_action_family": "risk_or_quality_analysis", "mcp_discovery_route_candidate_missing_axes": [], "mcp_discovery_route_candidate_executable_now": False, "mcp_discovery_route_candidate_enablement_reason": "Reviewed supplier-quality route is not wired yet.", "mcp_discovery_route_candidate_next_action": "Enable a reviewed supplier-quality route before claiming this fact.", } } }, } ], ) self.assertEqual(repair_targets["target_count"], 1) self.assertEqual(repair_targets["severity_counts"]["P1"], 1) target = repair_targets["targets"][0] self.assertEqual(target["problem_type"], "route_candidate_enablement_gap") self.assertEqual(target["target_source"], "route_candidate_enablement") self.assertEqual(target["route_candidate"]["candidate_status"], "needs_route_enablement") self.assertEqual(repair_targets["route_candidate_status_counts"], {"needs_route_enablement": 1}) self.assertEqual(repair_targets["route_candidate_groups"][0]["selected_chain_id"], "business_overview") def test_lead_handoff_surfaces_route_candidate_groups(self) -> None: repair_targets = dcl.build_deterministic_repair_targets( {"pack_id": "route_candidate_pack", "domain": "open_world", "final_status": "accepted"}, [ { "scenario_id": "route_candidate_demo", "title": "Route candidate demo", "artifact_dir": "artifacts/domain_runs/route_candidate_demo", "scenario_state": { "step_outputs": { "step_01": { "status": "accepted", "acceptance_status": "accepted", "execution_status": "validated", "question_resolved": "можно ли построить анализ качества поставщиков за 2020?", "reply_type": "partial_coverage", "mcp_discovery_selected_chain_id": "business_overview", "mcp_discovery_route_candidate_status": "needs_route_enablement", "mcp_discovery_route_candidate_fact_family": "supplier_quality", "mcp_discovery_route_candidate_action_family": "risk_or_quality_analysis", "mcp_discovery_route_candidate_missing_axes": [], "mcp_discovery_route_candidate_executable_now": False, "mcp_discovery_route_candidate_enablement_reason": "Reviewed supplier-quality route is not wired yet.", "mcp_discovery_route_candidate_next_action": "Enable a reviewed supplier-quality route before claiming this fact.", } } }, } ], ) handoff = dcl.build_lead_coder_handoff( loop_state={"loop_id": "route_candidate_loop"}, iteration_id="iteration_00", pack_dir=Path("artifacts/domain_runs/route_candidate_pack"), analyst_verdict_path=Path("artifacts/domain_runs/route_candidate_pack/analyst_verdict.json"), repair_targets_path=Path("artifacts/domain_runs/route_candidate_pack/repair_targets.json"), business_audit_path=Path("artifacts/domain_runs/route_candidate_pack/business_audit.md"), analyst_verdict={"quality_score": 75}, repair_targets=repair_targets, target_score=88, loop_decision="needs_exact_capability", analyst_accepted_gate=False, accepted_gate=False, deterministic_gate_ok=False, deterministic_gate_reason="repair_targets_remaining=P0:0,P1:1", requires_user_decision=False, user_decision_type="none", user_decision_prompt=None, ) markdown = dcl.build_lead_coder_handoff_markdown(handoff) self.assertEqual(handoff["route_candidate_groups"][0]["candidate_status"], "needs_route_enablement") self.assertEqual(handoff["route_candidate_enablement_targets"][0]["problem_type"], "route_candidate_enablement_gap") self.assertIn("## Route Candidate Handoff Groups", markdown) self.assertIn("route_candidate_demo:step_01", markdown) def test_validated_ready_route_candidates_are_reviewed_without_repair_handoff(self) -> None: repair_targets = dcl.build_deterministic_repair_targets( {"pack_id": "route_candidate_pack", "domain": "inventory_stock", "final_status": "accepted"}, [ { "scenario_id": "inventory_sale_trace", "title": "Sale trace", "artifact_dir": "artifacts/domain_runs/inventory_sale_trace", "scenario_state": { "step_outputs": { "step_04": { "status": "validated", "acceptance_status": "validated", "execution_status": "exact", "question_resolved": "trace purchase to sale", "reply_type": "factual", "mcp_discovery_selected_chain_id": "metadata_inspection", "mcp_discovery_route_candidate_status": "ready_for_reviewed_execution", "mcp_discovery_route_candidate_fact_family": "schema_surface", "mcp_discovery_route_candidate_action_family": "inspect_surface", "mcp_discovery_route_candidate_missing_axes": ["metadata_scope"], "mcp_discovery_route_candidate_executable_now": True, "violated_invariants": [], "warnings": [], } } }, } ], ) self.assertEqual(repair_targets["target_count"], 0) self.assertEqual(repair_targets["route_candidate_handoff_count"], 0) self.assertEqual(repair_targets["route_candidate_status_counts"], {}) self.assertEqual(repair_targets["route_candidate_groups"], []) self.assertEqual(repair_targets["reviewed_route_candidate_count"], 1) self.assertEqual(repair_targets["reviewed_route_candidate_status_counts"], {"ready_for_reviewed_execution": 1}) self.assertEqual( repair_targets["reviewed_route_candidate_groups"][0]["group_id"], "ready_for_reviewed_execution|metadata_inspection|metadata_scope", ) def test_nonvalidated_ready_route_candidates_stay_in_repair_handoff(self) -> None: repair_targets = dcl.build_deterministic_repair_targets( {"pack_id": "route_candidate_pack", "domain": "inventory_stock", "final_status": "partial"}, [ { "scenario_id": "inventory_sale_trace", "title": "Sale trace", "artifact_dir": "artifacts/domain_runs/inventory_sale_trace", "scenario_state": { "step_outputs": { "step_04": { "status": "rejected", "acceptance_status": "rejected", "execution_status": "partial", "question_resolved": "trace purchase to sale", "reply_type": "partial_coverage", "mcp_discovery_selected_chain_id": "metadata_inspection", "mcp_discovery_route_candidate_status": "ready_for_reviewed_execution", "mcp_discovery_route_candidate_fact_family": "schema_surface", "mcp_discovery_route_candidate_action_family": "inspect_surface", "mcp_discovery_route_candidate_missing_axes": ["metadata_scope"], "mcp_discovery_route_candidate_executable_now": True, "violated_invariants": ["answer_shape_mismatch"], "warnings": [], } } }, } ], ) self.assertEqual(repair_targets["route_candidate_handoff_count"], 1) self.assertEqual(repair_targets["reviewed_route_candidate_count"], 0) self.assertEqual( repair_targets["route_candidate_groups"][0]["group_id"], "ready_for_reviewed_execution|metadata_inspection|metadata_scope", ) def test_analysis_context_date_is_not_implicit_business_filter(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="stage_pack_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "All-time summary", "depends_on": [], "question_template": "all-time money summary", }, step_index=1, question_resolved="all-time money summary", analysis_context={"as_of_date": "2026-05-09", "source": "stage_pack"}, turn_artifact={ "assistant_message": { "reply_type": "factual_with_explanation", "text": "Short: all-time confirmed money summary.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) self.assertNotIn("missing_required_filter", step_state["violated_invariants"]) self.assertNotIn("wrong_as_of_date", step_state["violated_invariants"]) def test_analysis_context_date_is_required_for_explicit_date_carryover(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="date_carryover_demo", domain="inventory", step={ "step_id": "step_01", "title": "Date carryover", "depends_on": [], "question_template": "stock on that date", "required_carryover_invariants": ["date_scope"], }, step_index=1, question_resolved="stock on that date", analysis_context={"as_of_date": "2021-03-31"}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Short: stock confirmed.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_on_hand_as_of_date", "selected_recipe": "address_inventory_on_hand_as_of_date_v1", "capability_id": "confirmed_inventory_on_hand_as_of_date", "capability_route_mode": "exact", "fallback_type": "none", "extracted_filters": {"as_of_date": "2020-03-31"}, }, "session_summary": {}, }, entries=[], ) self.assertIn("wrong_as_of_date", step_state["violated_invariants"]) def test_temporal_reset_question_skips_carried_date_scope(self) -> None: self.assertTrue(dcl.question_resets_temporal_scope("show money za all time")) self.assertTrue(dcl.question_resets_temporal_scope("сколько всего денег за все доступное время")) self.assertTrue(dcl.question_resets_temporal_scope("по чепурнову покажи все доки")) self.assertTrue(dcl.question_resets_temporal_scope("покажи документы по свк")) self.assertFalse(dcl.question_resets_temporal_scope("а документы по этому же договору за тот же период")) carried = dcl.carry_forward_analysis_context( { "semantic_memory": { "date_scope": { "as_of_date": "2020-12-31", "period_from": "2020-10-01", "period_to": "2020-12-31", }, "organization_scope": {"label": "ООО Альтернатива Плюс"}, } }, {}, prefer_carryover=True, carry_date_scope=False, ) self.assertNotIn("as_of_date", carried) self.assertEqual(carried["organization_scope"], {"label": "ООО Альтернатива Плюс"}) def test_temporal_reset_question_clears_stale_navigation_date_scope(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="docs_reset_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Docs root", "depends_on": [], "question_template": "по чепурнову покажи все доки", "expected_intents": ["list_documents_by_counterparty"], "expected_recipe": "address_documents_by_counterparty_v1", }, step_index=1, question_resolved="по чепурнову покажи все доки", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Контрагент: Чепурнов П.Д. Найдено документов: 1.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "list_documents_by_counterparty", "selected_recipe": "address_documents_by_counterparty_v1", "capability_id": "documents_drilldown", "capability_contract_id": "documents_drilldown", "truth_gate_contract_status": "full_confirmed", "extracted_filters": {"counterparty": "Чепурнов", "organization": "ООО Альтернатива Плюс"}, }, "session_summary": { "address_navigation_state": { "session_context": { "date_scope": { "as_of_date": "2017-05-31", "period_from": None, "period_to": None, } } } }, }, entries=[], ) self.assertIsNone(step_state["date_scope"]["as_of_date"]) self.assertEqual(step_state["date_scope"]["source"], "question_temporal_scope_reset") def test_open_items_exact_negative_answer_validates_without_rows(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="open_items_negative_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Open items account 60", "depends_on": [], "question_template": "хвосты покажи по счету 60 на август 2022", "expected_intents": ["open_items_by_counterparty_or_contract"], }, step_index=1, question_resolved="хвосты покажи по счету 60 на август 2022", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "По счету 60 за период 2022-08-01..2022-08-31 открытых остатков не найдено.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "open_items_by_counterparty_or_contract", "selected_recipe": "address_open_items_by_party_or_contract_v1", "capability_id": "address_open_items_by_counterparty_or_contract", "capability_contract_id": "address_open_items_by_counterparty_or_contract", "truth_gate_contract_status": "full_confirmed", "response_type": "FACTUAL_SUMMARY", "result_mode": "confirmed_balance", "truth_mode": "limited", "answer_shape": "limited_with_reason", "mcp_call_status": "no_raw_rows", "balance_confirmed": False, "extracted_filters": { "account": "60", "period_from": "2022-08-01", "period_to": "2022-08-31", "as_of_date": "2022-08-31", }, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["status"], "validated") self.assertTrue(step_state["runtime_factual_answer_validated"]) def test_required_organization_filter_accepts_quoted_legal_name(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="quoted_org_filter_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Inventory root", "depends_on": [], "question_template": "что там на складе по остаткам", "expected_intents": ["inventory_on_hand_as_of_date"], "required_filters": {"organization": "ООО Альтернатива Плюс"}, }, step_index=1, question_resolved="что там на складе по остаткам", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "На складе подтверждено 11 позиций.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_on_hand_as_of_date", "selected_recipe": "address_inventory_on_hand_as_of_date_v1", "capability_id": "confirmed_inventory_on_hand_as_of_date", "capability_contract_id": "confirmed_inventory_on_hand_as_of_date", "truth_gate_contract_status": "full_confirmed", "response_type": "FACTUAL_LIST", "truth_mode": "confirmed", "answer_shape": "confirmed_factual", "mcp_call_status": "matched_non_empty", "balance_confirmed": True, "extracted_filters": {"organization": 'ООО "Альтернатива Плюс"', "as_of_date": "2026-06-02"}, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["status"], "validated") self.assertNotIn("missing_required_filter", step_state["violated_invariants"]) def test_clean_meta_chat_answer_validates_without_exact_capability(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="meta_chat_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Capability meta", "depends_on": [], "question_template": "расскажи что можешь интересного", "semantic_tags": ["meta_capability", "human_answer_quality"], "required_answer_patterns_all": ["(?i)1с", "(?i)могу"], }, step_index=1, question_resolved="расскажи что можешь интересного", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual_with_explanation", "text": "Могу помочь с быстрым анализом данных 1С: долги, остатки, НДС и документы.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "chat", "detected_intent": None, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["status"], "validated") self.assertTrue(step_state["clean_meta_chat_answer_validated"]) def test_merge_scenario_date_scope_keeps_current_scope_over_stale_previous(self) -> None: merged = dcl.merge_scenario_date_scope( { "as_of_date": "2020-12-31", "period_from": "2020-10-01", "period_to": "2020-12-31", "source": "scenario_state_carryover", }, { "as_of_date": "2021-03-31", "period_from": "2021-03-01", "period_to": "2021-03-31", "source": "current_turn", }, depends_on=["previous_step"], ) self.assertEqual(merged["as_of_date"], "2021-03-31") self.assertEqual(merged["source"], "current_turn") def test_mcp_business_overview_all_time_scope_overrides_stale_session_date(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="business_overview_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "All-time money", "depends_on": ["previous_step"], "question_template": "all-time money summary", "expected_intents": ["business_overview"], }, step_index=1, question_resolved="all-time money summary", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Short: all-time confirmed money summary.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_supplier_stock_overlap_as_of_date", "selected_recipe": "address_inventory_supplier_stock_overlap_as_of_date_v1", "capability_id": "inventory_inventory_supplier_stock_overlap_as_of_date", "mcp_discovery_response_applied": True, "mcp_discovery_selected_chain_id": "business_overview", "mcp_discovery_catalog_chain_top_match": "business_overview", "mcp_discovery_response_candidate_v1": { "candidate_status": "ready_for_guarded_use", "reply_type": "partial_coverage", }, "assistant_mcp_discovery_entry_point_v1": { "bridge": { "pilot": { "derived_business_overview": { "period_scope": None, } } } }, }, "session_summary": { "address_navigation_state": { "session_context": { "active_result_set_id": "rs-stale", "date_scope": { "as_of_date": "2020-12-31", "period_from": "2020-10-01", "period_to": "2020-12-31", }, } } }, }, entries=[], ) self.assertEqual(step_state["date_scope"]["scope"], "all_time") self.assertIsNone(step_state["date_scope"]["as_of_date"]) self.assertEqual(step_state["active_result_set_id"], "mcp-discovery-msg-1") self.assertNotIn("wrong_date_scope_state", step_state["violated_invariants"]) def test_applied_ready_mcp_discovery_chain_satisfies_expected_intent(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="business_overview_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Business overview", "depends_on": [], "question_template": "business overview for 2020", "expected_intents": ["business_overview"], }, step_index=1, question_resolved="business overview for 2020", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Short: business overview from confirmed 1C rows.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_supplier_stock_overlap_as_of_date", "selected_recipe": "address_inventory_supplier_stock_overlap_as_of_date_v1", "capability_id": "inventory_inventory_supplier_stock_overlap_as_of_date", "mcp_discovery_response_applied": True, "mcp_discovery_selected_chain_id": "business_overview", "mcp_discovery_catalog_chain_top_match": "business_overview", "mcp_discovery_response_candidate_v1": { "candidate_status": "ready_for_guarded_use", "reply_type": "partial_coverage", }, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["mcp_discovery_effective_intents"], ["business_overview"]) self.assertNotIn("wrong_intent", step_state["violated_invariants"]) def test_ready_bounded_mcp_answer_can_validate_without_exact_route(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="business_overview_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Business overview", "depends_on": [], "question_template": "business overview for 2020", "expected_intents": ["business_overview"], "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved="business overview for 2020", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Short: confirmed bounded business overview from 1C rows.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_supplier_stock_overlap_as_of_date", "selected_recipe": "address_inventory_supplier_stock_overlap_as_of_date_v1", "capability_id": "inventory_inventory_supplier_stock_overlap_as_of_date", "mcp_discovery_response_applied": True, "mcp_discovery_selected_chain_id": "business_overview", "mcp_discovery_catalog_chain_top_match": "business_overview", "mcp_discovery_response_candidate_v1": { "candidate_status": "ready_for_guarded_use", "reply_type": "partial_coverage", }, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "partial") self.assertTrue(step_state["bounded_mcp_answer_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_required_answer_patterns_block_generic_bounded_mcp_summary(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="summary_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Summary", "depends_on": [], "question_template": "summarize company and SVK separately", "required_answer_shape": "direct_answer_first", "required_answer_patterns_all": ["SVK", "company"], }, step_index=1, question_resolved="summarize company and SVK separately", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Short: company money summary only.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "mcp_discovery_response_applied": True, "mcp_discovery_selected_chain_id": "business_overview", "mcp_discovery_catalog_chain_top_match": "business_overview", "mcp_discovery_response_candidate_v1": { "candidate_status": "ready_for_guarded_use", "reply_type": "partial_coverage", }, }, "session_summary": {}, }, entries=[], ) self.assertIn("required_answer_patterns_all_missing", step_state["violated_invariants"]) self.assertFalse(step_state["bounded_mcp_answer_validated"]) self.assertEqual(step_state["acceptance_status"], "rejected") def test_memory_checkpoint_can_validate_honest_no_scope_answer(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="memory_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Memory checkpoint", "depends_on": [], "question_template": "is any company or counterparty selected in the current dialog?", "semantic_tags": ["memory", "scope_guard"], "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved="is any company or counterparty selected in the current dialog?", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "В текущем диалоге не выбрана компания или контрагент; память не выдумываю.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "customer_revenue_and_payments", "fallback_type": "no_rows", }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "partial") self.assertTrue(step_state["memory_checkpoint_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_deterministic_chat_memory_checkpoint_validates_without_exact_capability(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="memory_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Memory checkpoint", "depends_on": [], "question_template": "current dialog memory checkpoint", "semantic_tags": ["memory", "scope_guard"], "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved="current dialog memory checkpoint", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual_with_explanation", "text": ( "Коротко: в текущем диалоге я не вижу выбранной компании, контрагента или позиции. " "Память про «Группа СВК» в этом диалоге не подтверждена." ), "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "chat", "fallback_type": "none", "living_router_reason": "memory_recap_followup_detected", "living_chat_response_source": "deterministic_memory_recap_contract", }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "partial") self.assertTrue(step_state["memory_checkpoint_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_confirmed_runtime_factual_answer_can_validate_without_exact_route_mode(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="runtime_factual_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Account 60 tails", "depends_on": [], "question_template": "show account 60 tails", "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved="show account 60 tails", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Коротко: по счету 60 найдено 8 строк хвостов; контрагентов с сигналом: 6.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "open_items_by_counterparty_or_contract", "selected_recipe": "address_open_items_by_party_or_contract_v1", "capability_id": "address_open_items_by_counterparty_or_contract", "capability_route_mode": "heuristic", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_LIST", "result_mode": "confirmed_balance", }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "partial") self.assertTrue(step_state["runtime_factual_answer_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_business_result_mode_accepts_clean_margin_confirmed_balance(self) -> None: question = ( "\u043a\u0430\u043a\u0430\u044f " "\u043d\u043e\u043c\u0435\u043d\u043a\u043b\u0430\u0442\u0443\u0440\u0430 " "\u0431\u044b\u043b\u0430 \u0441\u0430\u043c\u043e\u0439 " "\u043c\u0430\u0440\u0436\u0438\u043d\u0430\u043b\u044c\u043d\u043e\u0439 " "\u0432 2020" ) answer_text = ( "\u0421\u0430\u043c\u0430\u044f " "\u043c\u0430\u0440\u0436\u0438\u043d\u0430\u043b\u044c\u043d\u0430\u044f " "\u043f\u043e\u0437\u0438\u0446\u0438\u044f \u0437\u0430 " "\u043f\u0435\u0440\u0438\u043e\u0434 2020: " "\u0422\u043e\u0432\u0430\u0440 A \u2014 " "\u043c\u0430\u0440\u0436\u0430 42%, " "\u0432\u044b\u0440\u0443\u0447\u043a\u0430 100 000 " "\u0440\u0443\u0431., " "\u0441\u0435\u0431\u0435\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u043d\u0430\u044f " "\u0431\u0430\u0437\u0430 58 000 " "\u0440\u0443\u0431., " "\u0432\u0430\u043b\u043e\u0432\u0430\u044f " "\u0440\u0430\u0437\u043d\u0438\u0446\u0430 42 000 " "\u0440\u0443\u0431.\n" "\u0421\u043b\u0435\u0434\u0443\u044e\u0449\u0438\u0439 " "\u0448\u0430\u0433: \u043c\u043e\u0433\u0443 " "\u0440\u0430\u0441\u043a\u0440\u044b\u0442\u044c " "\u0441\u0442\u0440\u043e\u043a\u0438 " "\u0432\u044b\u0440\u0443\u0447\u043a\u0438 \u0438 " "\u0441\u0435\u0431\u0435\u0441\u0442\u043e\u0438\u043c\u043e\u0441\u0442\u043d\u043e\u0439 " "\u0431\u0430\u0437\u044b." ) step_state = dcl.build_scenario_step_state( scenario_id="margin_result_mode_demo", domain="margin_profitability", step={ "step_id": "step_01", "title": "Margin ranking", "depends_on": [], "question_template": question, "expected_intents": ["inventory_margin_ranking_for_nomenclature"], "expected_capability": "inventory_inventory_margin_ranking_for_nomenclature", "expected_recipe": "address_inventory_margin_ranking_for_nomenclature_v1", "expected_result_mode": "ranking_or_limited_accounting_answer", "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved=question, analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": answer_text, "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_margin_ranking_for_nomenclature", "selected_recipe": "address_inventory_margin_ranking_for_nomenclature_v1", "capability_id": "inventory_inventory_margin_ranking_for_nomenclature", "capability_route_mode": "exact", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_SUMMARY", "result_mode": "confirmed_balance", "truth_mode": "confirmed", "answer_shape": "confirmed_factual", "balance_confirmed": True, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "exact") self.assertNotIn("wrong_result_mode", step_state["violated_invariants"]) self.assertEqual(step_state["business_first_review"]["issue_codes"], []) self.assertEqual(step_state["acceptance_status"], "validated") def test_assistant_truth_policy_runtime_contract_validates_exact_factual_step(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="truth_contract_demo", domain="inventory_selected_object_reliability", step={ "step_id": "step_01", "title": "VAT by selected purchase date", "depends_on": [], "question_template": "calculate VAT by selected purchase date", "required_answer_shape": "direct_answer_first", "required_answer_patterns_all": ["VAT", "Q1 2015"], }, step_index=1, question_resolved="calculate VAT by selected purchase date", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "VAT for Q1 2015 is confirmed: 1,556,449.06. Basis: sales and purchase ledgers.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "vat_liability_confirmed_for_tax_period", "selected_recipe": "address_vat_liability_confirmed_tax_period_v1", "capability_id": "confirmed_vat_liability_for_tax_period", "capability_route_mode": "exact", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_SUMMARY", "result_mode": "confirmed_balance", "balance_confirmed": True, "mcp_discovery_route_candidate_status": "ready_for_reviewed_execution", "mcp_discovery_response_applied": False, "assistant_runtime_contract_v1": { "capability_contract_id": "confirmed_vat_liability_for_tax_period", "truth_gate_contract_status": "full_confirmed", "carryover_eligibility": "root_only", }, "assistant_truth_answer_policy_v1": { "truth_gate": { "coverage_status": "full", "evidence_grade": "strong", "truth_mode": "confirmed", "source_truth_gate_status": "full_confirmed", }, "answer_shape": { "answer_shape": "confirmed_factual", "capability_contract_id": "confirmed_vat_liability_for_tax_period", }, }, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["capability_contract_id"], "confirmed_vat_liability_for_tax_period") self.assertEqual(step_state["truth_gate_contract_status"], "full_confirmed") self.assertEqual(step_state["truth_mode"], "confirmed") self.assertEqual(step_state["coverage_status"], "full") self.assertEqual(step_state["answer_shape"], "confirmed_factual") self.assertTrue(step_state["runtime_factual_answer_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_analyst_shape_only_target_is_suppressed_for_truth_confirmed_runtime_step(self) -> None: repair_targets = { "targets": [], "step_validation_index": { "scenario_a:step_01": { "acceptance_status": "validated", "violated_invariants": [], "warnings": [], "runtime_factual_answer_validated": True, "truth_gate_contract_status": "full_confirmed", "assistant_text_excerpt": "Confirmed business answer for the selected item.", } }, } analyst_verdict = { "priority_targets": [ { "scenario_id": "scenario_a", "step_id": "step_01", "severity": "P1", "problem_type": "capability_gap", "root_cause_layers": ["runtime_capability_gap"], "fix_goal": "Remove shape-only acceptance and add runtime truth confirmation.", } ] } merged = dcl.merge_analyst_priority_repair_targets(repair_targets, analyst_verdict) self.assertEqual(merged["target_count"], 0) self.assertEqual(merged["suppressed_analyst_priority_target_count"], 1) self.assertEqual( merged["suppressed_analyst_priority_targets"][0]["reason"], "analyst_target_contradicts_validated_step", ) def test_analyst_temporal_cutoff_false_positive_is_suppressed_for_validated_step(self) -> None: repair_targets = { "targets": [], "step_validation_index": { "inventory_chain:s03": { "acceptance_status": "validated", "violated_invariants": [], "warnings": [], "runtime_factual_answer_validated": True, "extracted_filters": {"as_of_date": "2016-03-31"}, "assistant_text_excerpt": ( "По позиции до 31.03.2016 поставщик не подтвержден. " "Первая закупка: 05.02.2015. Последняя закупка: 22.07.2015. " "Опорный документ от 22.07.2015." ), } }, } analyst_verdict = { "priority_targets": [ { "scenario_id": "inventory_chain", "step_id": "s03", "severity": "P0", "problem_type": "other", "root_cause_layers": ["temporal_honesty_gap"], "fix_goal": "The answer declares a historical cutoff but cites proof rows with later dates.", } ] } merged = dcl.merge_analyst_priority_repair_targets(repair_targets, analyst_verdict) self.assertEqual(merged["target_count"], 0) self.assertEqual(merged["suppressed_analyst_priority_target_count"], 1) def test_analyst_temporal_cutoff_real_out_of_window_target_is_kept(self) -> None: repair_targets = { "targets": [], "step_validation_index": { "inventory_chain:s03": { "acceptance_status": "validated", "violated_invariants": [], "warnings": [], "runtime_factual_answer_validated": True, "extracted_filters": {"as_of_date": "2016-03-31"}, "assistant_text_excerpt": ( "По позиции до 31.03.2016 поставщик не подтвержден. " "Опорный документ от 25.05.2026." ), } }, } analyst_verdict = { "priority_targets": [ { "scenario_id": "inventory_chain", "step_id": "s03", "severity": "P0", "problem_type": "other", "root_cause_layers": ["temporal_honesty_gap"], "fix_goal": "The answer declares a historical cutoff but cites proof rows with later dates.", } ] } merged = dcl.merge_analyst_priority_repair_targets(repair_targets, analyst_verdict) self.assertEqual(merged["target_count"], 1) self.assertEqual(merged["suppressed_analyst_priority_target_count"], 0) def test_literal_result_mode_contract_still_rejects_mismatch(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="literal_result_mode_demo", domain="inventory", step={ "step_id": "step_01", "title": "Literal result mode", "depends_on": [], "question_template": "show stock", "expected_result_mode": "exact_inventory_balance", }, step_index=1, question_resolved="show stock", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Short: stock is confirmed.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_SUMMARY", "result_mode": "confirmed_balance", "truth_mode": "confirmed", "answer_shape": "confirmed_factual", "balance_confirmed": True, }, "session_summary": {}, }, entries=[], ) self.assertIn("wrong_result_mode", step_state["violated_invariants"]) def test_selected_object_purchase_document_mode_accepts_confirmed_factual_route(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="margin_selected_object_demo", domain="margin_profitability", step={ "step_id": "step_05_selected_item_documents", "title": "Selected item document", "depends_on": ["step_04_selected_item_purchase_date"], "question_template": "Which document confirms this selected item?", "expected_intents": ["inventory_purchase_documents_for_item"], "expected_capability": "inventory_inventory_purchase_documents_for_item", "expected_recipe": "address_inventory_purchase_documents_for_item_v1", "expected_result_mode": "selected_object_purchase_documents_or_honest_boundary", "required_answer_shape": "direct_answer_first", }, step_index=5, question_resolved="Which document confirms this selected item?", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Confirmation document for the selected item: Purchase document 0001.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "inventory_purchase_documents_for_item", "selected_recipe": "address_inventory_purchase_documents_for_item_v1", "capability_id": "inventory_inventory_purchase_documents_for_item", "capability_route_mode": "exact", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_SUMMARY", "result_mode": "confirmed_balance", "truth_mode": "confirmed", "answer_shape": "confirmed_factual", "balance_confirmed": True, "extracted_filters": {"item": "Selected item"}, }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "exact") self.assertNotIn("wrong_result_mode", step_state["violated_invariants"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_exact_confirmed_document_followup_sets_runtime_factual_validation(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="svk_pivot", domain="agentic_loop", step={ "step_id": "s02_svk_docs", "title": "Counterparty documents follow-up", "depends_on": ["s01_svk_money"], "question_template": "show documents by this chain", "semantic_tags": ["counterparty", "documents", "scope_guard"], "required_answer_shape": "direct_answer_first", }, step_index=2, question_resolved="show documents by this chain", analysis_context={"as_of_date": "2026-05-09"}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Контрагент: Группа СВК. Найдено документов: 19.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "list_documents_by_counterparty", "selected_recipe": "address_documents_by_counterparty_v1", "capability_id": "documents_drilldown", "capability_route_mode": "exact", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_LIST", "truth_mode": "confirmed", "answer_shape": "confirmed_factual", "coverage_status": "full", "evidence_grade": "strong", "extracted_filters": {"counterparty": "Группа СВК", "as_of_date": "2026-05-09"}, "focus_object": { "object_type": "counterparty", "object_id": "counterparty:группа свк", "label": "Группа СВК", }, }, "session_summary": {}, }, entries=[{"item": "2021-11-10T12:00:07Z"}], ) self.assertEqual(step_state["execution_status"], "exact") self.assertTrue(step_state["runtime_factual_answer_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_heuristic_open_items_guarded_insufficiency_validates_separately(self) -> None: answer_text = ( "\u041a\u043e\u0440\u043e\u0442\u043a\u043e: \u0442\u043e\u0447\u043d\u044b\u0439 " "\u043e\u0442\u043a\u0440\u044b\u0442\u044b\u0439 \u043e\u0441\u0442\u0430\u0442\u043e\u043a " "\u043f\u043e \u0441\u0447\u0435\u0442\u0443 60 \u043d\u0435 " "\u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d; \u043d\u0438\u0436\u0435 " "\u0442\u043e\u043b\u044c\u043a\u043e \u043f\u0440\u0435\u0434\u0432\u0430\u0440\u0438\u0442\u0435\u043b\u044c\u043d\u044b\u0435 " "\u0441\u0438\u0433\u043d\u0430\u043b\u044b \u043f\u043e \u0434\u0432\u0438\u0436\u0435\u043d\u0438\u044f\u043c: 8 " "\u0441\u0442\u0440\u043e\u043a.\n" "\u042d\u0442\u043e \u043d\u0435 \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043d\u043e\u0435 " "\u0441\u0430\u043b\u044c\u0434\u043e: \u0442\u0435\u043a\u0443\u0449\u0438\u0439 " "\u043a\u043e\u043d\u0442\u0443\u0440 \u0432\u0438\u0434\u0438\u0442 " "\u0434\u0432\u0438\u0436\u0435\u043d\u0438\u044f-\u043a\u0430\u043d\u0434\u0438\u0434\u0430\u0442\u044b, " "\u043d\u043e \u043d\u0435 \u0434\u043e\u043a\u0430\u0437\u044b\u0432\u0430\u0435\u0442 " "\u043e\u0441\u0442\u0430\u0442\u043e\u043a." ) step_state = dcl.build_scenario_step_state( scenario_id="runtime_factual_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Account 60 limited tails", "depends_on": [], "question_template": "show account 60 tails; say if exact data is unavailable", "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved="show account 60 tails; say if exact data is unavailable", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": answer_text, "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "open_items_by_counterparty_or_contract", "selected_recipe": "address_open_items_by_party_or_contract_v1", "capability_id": "address_open_items_by_counterparty_or_contract", "capability_route_mode": "heuristic", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_LIST", "result_mode": "heuristic_candidates", "balance_confirmed": False, "truth_mode": "limited", "answer_shape": "limited_with_reason", }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "partial") self.assertEqual(step_state["truth_mode"], "limited") self.assertEqual(step_state["answer_shape"], "limited_with_reason") self.assertFalse(step_state["runtime_factual_answer_validated"]) self.assertTrue(step_state["guarded_insufficiency_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_expected_clarification_partial_answer_validates(self) -> None: answer_text = ( "Для рейтинга прибыльности номенклатуры нужен период.\n\n" "Могу посчитать по номенклатуре: выручку без НДС, себестоимость реализации, " "валовую прибыль и маржинальность.\n\n" "Уточните период: месяц, квартал, год или весь доступный период." ) step_state = dcl.validate_step_contract( { "execution_status": "partial", "reply_type": "partial_coverage", "expected_result_mode": "clarification_required", "required_answer_shape": "direct_answer_first", "response_type": "LIMITED_WITH_REASON", "truth_mode": "clarification_required", "answer_shape": "clarification_required", "assistant_text": answer_text, "actual_direct_answer": "Для рейтинга прибыльности номенклатуры нужен период.", "top_non_empty_lines": [ "Для рейтинга прибыльности номенклатуры нужен период.", "Могу посчитать по номенклатуре: выручку без НДС, себестоимость реализации, валовую прибыль и маржинальность.", "Уточните период: месяц, квартал, год или весь доступный период.", ], } ) self.assertTrue(step_state["clarification_answer_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_inventory_margin_guarded_insufficiency_validates_without_exact_values(self) -> None: answer_text = ( "За период 01.05.2020 - 31.05.2020 рейтинг прибыльности номенклатуры построить нельзя.\n\n" "Что нашлось:\n" "- Есть реализация по 1 номенклатурной позиции.\n" "- Подтвержденной себестоимости реализации по этой позиции не найдено.\n" "- Поэтому валовую прибыль и маржинальность честно посчитать нельзя.\n" "Вывод: за период 01.05.2020 - 31.05.2020 нет достаточной базы для рейтинга " "«высокая / низкая прибыль» по номенклатуре.\n\n" "Что можно сделать дальше:\n" "- показать найденные реализации за этот период;\n" "- расширить период до квартала или года;\n" "- попробовать строгий расчет по проводкам 90.01 / 90.02.\n\n" "Граница ответа:\n" "- Прибыльность номенклатуры считаю только когда есть реализация и подтвержденная себестоимость реализации." ) step_state = dcl.validate_step_contract( { "execution_status": "partial", "reply_type": "partial_coverage", "expected_result_mode": "ranking_or_limited_accounting_answer", "required_answer_shape": "direct_answer_first", "detected_intent": "inventory_margin_ranking_for_nomenclature", "capability_id": "inventory_inventory_margin_ranking_for_nomenclature", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_SUMMARY", "truth_mode": "limited", "answer_shape": "limited_with_reason", "balance_confirmed": False, "assistant_text": answer_text, "actual_direct_answer": "За период 01.05.2020 - 31.05.2020 рейтинг прибыльности номенклатуры построить нельзя.", "top_non_empty_lines": [ "За период 01.05.2020 - 31.05.2020 рейтинг прибыльности номенклатуры построить нельзя.", "Что нашлось:", "- Есть реализация по 1 номенклатурной позиции.", "- Подтвержденной себестоимости реализации по этой позиции не найдено.", "- Поэтому валовую прибыль и маржинальность честно посчитать нельзя.", "Вывод: за период 01.05.2020 - 31.05.2020 нет достаточной базы для рейтинга «высокая / низкая прибыль» по номенклатуре.", "Что можно сделать дальше:", "- показать найденные реализации за этот период;", "- расширить период до квартала или года;", "- попробовать строгий расчет по проводкам 90.01 / 90.02.", "Граница ответа:", "- Прибыльность номенклатуры считаю только когда есть реализация и подтвержденная себестоимость реализации.", ], } ) self.assertTrue(step_state["guarded_insufficiency_validated"]) self.assertEqual(step_state["acceptance_status"], "validated") def test_heuristic_open_items_without_limitation_is_rejected(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="runtime_factual_demo", domain="agentic_loop", step={ "step_id": "step_01", "title": "Account 60 unguarded tails", "depends_on": [], "question_template": "show account 60 tails", "required_answer_shape": "direct_answer_first", }, step_index=1, question_resolved="show account 60 tails", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "Short: account 60 has 8 open-item rows and 6 counterparties.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": { "detected_mode": "address_query", "detected_intent": "open_items_by_counterparty_or_contract", "selected_recipe": "address_open_items_by_party_or_contract_v1", "capability_id": "address_open_items_by_counterparty_or_contract", "capability_route_mode": "heuristic", "fallback_type": "none", "mcp_call_status": "matched_non_empty", "response_type": "FACTUAL_LIST", "result_mode": "heuristic_candidates", "balance_confirmed": False, "truth_mode": "limited", "answer_shape": "limited_with_reason", }, "session_summary": {}, }, entries=[], ) self.assertEqual(step_state["execution_status"], "partial") self.assertFalse(step_state["runtime_factual_answer_validated"]) self.assertFalse(step_state["guarded_insufficiency_validated"]) self.assertEqual(step_state["acceptance_status"], "rejected") def test_truth_harness_warns_on_catalog_alignment_divergence(self) -> None: reviewed = dth.evaluate_truth_step( step={ "step_id": "step_01", "question_template": "show planner alignment", "criticality": "critical", "allowed_reply_types": [], }, step_state={ "question_resolved": "show planner alignment", "reply_type": "factual", "assistant_text": "Confirmed answer", "actual_direct_answer": "Confirmed answer", "detected_intent": "counterparty_turnover", "selected_recipe": "counterparty_turnover_by_period", "capability_id": "confirmed_counterparty_turnover", "mcp_discovery_catalog_chain_alignment_status": "selected_outside_match_set", "mcp_discovery_catalog_chain_top_match": "value_flow_comparison", "mcp_discovery_catalog_chain_selected_matches_top": False, "extracted_filters": {}, }, step_results={}, bindings={}, runtime_bindings={}, ) self.assertEqual(reviewed["review_status"], "warning") self.assertEqual(reviewed["warning_findings_count"], 1) self.assertEqual(reviewed["review_findings"][0]["code"], "catalog_alignment_divergence") self.assertEqual(reviewed["review_findings"][0]["severity"], "warning") def test_truth_harness_checks_expected_catalog_alignment_fields(self) -> None: reviewed = dth.evaluate_truth_step( step={ "step_id": "step_01", "question_template": "show planner alignment", "criticality": "critical", "allowed_reply_types": [], "expected_catalog_alignment_status": "selected_matches_top", "expected_catalog_chain_top_match": "value_flow_comparison", "expected_catalog_selected_matches_top": True, }, step_state={ "question_resolved": "show planner alignment", "reply_type": "factual", "assistant_text": "Confirmed answer", "actual_direct_answer": "Confirmed answer", "detected_intent": "counterparty_turnover", "selected_recipe": "counterparty_turnover_by_period", "capability_id": "confirmed_counterparty_turnover", "mcp_discovery_catalog_chain_alignment_status": "selected_matches_top", "mcp_discovery_catalog_chain_top_match": "value_flow", "mcp_discovery_catalog_chain_selected_matches_top": True, "extracted_filters": {}, }, step_results={}, bindings={}, runtime_bindings={}, ) self.assertEqual(reviewed["review_status"], "fail") self.assertEqual(reviewed["critical_findings_count"], 1) self.assertEqual(reviewed["review_findings"][0]["code"], "wrong_catalog_chain_top_match") def test_truth_harness_checks_expected_route_candidate_fields(self) -> None: reviewed = dth.evaluate_truth_step( step={ "step_id": "step_01", "question_template": "show route candidate", "criticality": "critical", "allowed_reply_types": [], "expected_route_candidate_status": "needs_user_scope", "expected_route_candidate_executable_now": False, "expected_route_candidate_missing_axes": ["organization", "period"], }, step_state={ "question_resolved": "show route candidate", "reply_type": "clarification_required", "assistant_text": "Please choose organization.", "actual_direct_answer": "Please choose organization.", "detected_intent": "counterparty_turnover", "selected_recipe": None, "capability_id": None, "mcp_discovery_route_candidate_status": "needs_user_scope", "mcp_discovery_route_candidate_missing_axes": ["organization"], "mcp_discovery_route_candidate_executable_now": False, "extracted_filters": {}, }, step_results={}, bindings={}, runtime_bindings={}, ) self.assertEqual(reviewed["review_status"], "fail") self.assertEqual(reviewed["critical_findings_count"], 1) self.assertEqual(reviewed["review_findings"][0]["code"], "missing_route_candidate_axes") def test_business_first_review_flags_dirty_direct_answer_surface(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="business_surface_demo", domain="business_overview", step={ "step_id": "step_01", "title": "Top year", "depends_on": [], "question_template": "какой у нас самый доходный год", }, step_index=1, question_resolved="какой у нас самый доходный год", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Коротко: Ограниченный бизнес-обзор по подтвержденным строкам 1С. " + ("лишний текст " * 220), "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) review = step_state["business_first_review"] self.assertFalse(review["direct_answer_first_ok"]) self.assertFalse(review["business_usefulness_ok"]) self.assertIn("business_direct_answer_missing", review["issue_codes"]) self.assertIn("answer_layering_noise", review["issue_codes"]) self.assertIn("business_answer_too_verbose", review["issue_codes"]) self.assertIn("business_direct_answer_missing", step_state["violated_invariants"]) def test_business_first_review_accepts_compact_direct_answer_surface(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="business_surface_demo", domain="business_overview", step={ "step_id": "step_01", "title": "Top year", "depends_on": [], "question_template": "какой у нас самый доходный год", }, step_index=1, question_resolved="какой у нас самый доходный год", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Коротко: самый доходный год в доступном денежном контуре 1С — 2015: 136 723 459,73 руб.\nМетод: считаю по подтвержденным входящим поступлениям.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) review = step_state["business_first_review"] self.assertTrue(review["direct_answer_first_ok"]) self.assertTrue(review["business_usefulness_ok"]) self.assertEqual(review["issue_codes"], []) def test_business_first_review_separates_direct_answer_from_later_technical_leak(self) -> None: question = "\u043a\u0430\u043a\u043e\u0439 \u0443 \u043d\u0430\u0441 \u0441\u0430\u043c\u044b\u0439 \u0434\u043e\u0445\u043e\u0434\u043d\u044b\u0439 \u0433\u043e\u0434" step_state = dcl.build_scenario_step_state( scenario_id="business_surface_demo", domain="business_overview", step={ "step_id": "step_01", "title": "Top year", "depends_on": [], "question_template": question, }, step_index=1, question_resolved=question, analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "2015 \u2014 \u0441\u0430\u043c\u044b\u0439 \u0434\u043e\u0445\u043e\u0434\u043d\u044b\u0439 \u0433\u043e\u0434 \u043f\u043e \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043d\u044b\u043c \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u043c \u0434\u0435\u043d\u044c\u0433\u0430\u043c.\nservice: capability_id=business_overview_route_template_v1", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) review = step_state["business_first_review"] self.assertTrue(review["direct_answer_first_ok"]) self.assertTrue(review["technical_garbage_present"]) self.assertIn("technical_garbage_in_answer", review["issue_codes"]) self.assertNotIn("business_direct_answer_missing", review["issue_codes"]) def test_business_first_review_rejects_found_rows_scaffold_as_direct_answer(self) -> None: question = "\u0441\u043a\u043e\u043b\u044c\u043a\u043e \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445 \u0434\u0435\u043d\u0435\u0433 \u0437\u0430 2020" step_state = dcl.build_scenario_step_state( scenario_id="semantic_gate_demo", domain="value_flow", step={ "step_id": "step_01", "title": "Direct money answer", "depends_on": [], "question_template": question, }, step_index=1, question_resolved=question, analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual", "text": "\u041a\u043e\u0440\u043e\u0442\u043a\u043e: \u041f\u043e \u0434\u0430\u043d\u043d\u044b\u043c 1\u0421 \u043d\u0430\u0439\u0434\u0435\u043d\u044b \u0441\u0442\u0440\u043e\u043a\u0438 \u0432\u0445\u043e\u0434\u044f\u0449\u0438\u0445 \u0434\u0435\u043d\u0435\u0436\u043d\u044b\u0445 \u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438\u0439; \u0441\u0443\u043c\u043c\u0443 \u043c\u043e\u0436\u043d\u043e \u043d\u0430\u0437\u044b\u0432\u0430\u0442\u044c \u0442\u043e\u043b\u044c\u043a\u043e \u0432 \u0440\u0430\u043c\u043a\u0430\u0445 \u043f\u0440\u043e\u0432\u0435\u0440\u043a\u0438.\n\u0427\u0442\u043e \u043f\u043e\u0434\u0442\u0432\u0435\u0440\u0436\u0434\u0435\u043d\u043e: 47 628 853,03 \u0440\u0443\u0431.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) review = step_state["business_first_review"] self.assertFalse(review["direct_answer_first_ok"]) self.assertFalse(review["answer_layering_ok"]) self.assertIn("business_direct_answer_missing", review["issue_codes"]) self.assertIn("answer_layering_noise", review["issue_codes"]) def test_business_first_review_allows_direct_answer_with_evidence_boundary(self) -> None: question = "\u043a\u0430\u043a\u043e\u0435 \u043d\u0435\u0442\u0442\u043e \u043f\u043e \u0434\u0435\u043d\u044c\u0433\u0430\u043c \u0441 \u0413\u0440\u0443\u043f\u043f\u0430 \u0421\u0412\u041a \u0437\u0430 2020" step_state = dcl.build_scenario_step_state( scenario_id="semantic_gate_demo", domain="value_flow", step={ "step_id": "step_01", "title": "Counterparty net flow", "depends_on": [], "question_template": question, }, step_index=1, question_resolved=question, analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "factual_with_explanation", "text": "\u041a\u043e\u0440\u043e\u0442\u043a\u043e: \u043f\u043e \u043a\u043e\u043d\u0442\u0440\u0430\u0433\u0435\u043d\u0442\u0443 \u0413\u0440\u0443\u043f\u043f\u0430 \u0421\u0412\u041a \u0437\u0430 2020 \u043f\u043e \u043d\u0430\u0439\u0434\u0435\u043d\u043d\u044b\u043c \u0441\u0442\u0440\u043e\u043a\u0430\u043c 1\u0421 \u043f\u043e\u043b\u0443\u0447\u0438\u043b\u0438 12 093 465 \u0440\u0443\u0431., \u0437\u0430\u043f\u043b\u0430\u0442\u0438\u043b\u0438 0 \u0440\u0443\u0431.; \u043d\u0435\u0442\u0442\u043e \u0432 \u043d\u0430\u0448\u0443 \u0441\u0442\u043e\u0440\u043e\u043d\u0443 12 093 465 \u0440\u0443\u0431.", "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) review = step_state["business_first_review"] self.assertTrue(review["direct_answer_first_ok"]) self.assertTrue(review["answer_layering_ok"]) self.assertEqual(review["issue_codes"], []) def test_semantic_audit_markdown_exposes_semantic_gate(self) -> None: markdown = dth.build_business_review_markdown( { "scenario_id": "semantic_gate_demo", "domain": "value_flow", "title": "Semantic gate demo", "session_id": "session-1", "overall_business_status": "fail", "steps_total": 1, "steps_with_business_failures": 1, "steps_with_business_warnings": 0, "issue_counts": {"answer_layering_noise": 1}, "steps": [], }, title="Semantic audit", ) self.assertIn("# Semantic audit", markdown) self.assertIn("semantic_gate_passed: `False`", markdown) def test_truth_harness_promotes_business_review_issues_to_findings(self) -> None: step_state = dcl.build_scenario_step_state( scenario_id="business_surface_demo", domain="business_overview", step={ "step_id": "step_01", "title": "Top year", "depends_on": [], "question_template": "какой у нас самый доходный год", }, step_index=1, question_resolved="какой у нас самый доходный год", analysis_context={}, turn_artifact={ "assistant_message": { "reply_type": "partial_coverage", "text": "Коротко: Ограниченный бизнес-обзор по подтвержденным строкам 1С. " + ("лишний текст " * 220), "message_id": "msg-1", "trace_id": "trace-1", }, "technical_debug_payload": {}, "session_summary": {}, }, entries=[], ) reviewed = dth.evaluate_truth_step( step={ "step_id": "step_01", "question_template": "какой у нас самый доходный год", "criticality": "critical", "allowed_reply_types": [], }, step_state=step_state, step_results={}, bindings={}, runtime_bindings={}, ) codes = [item["code"] for item in reviewed["review_findings"]] self.assertIn("business_review:business_direct_answer_missing", codes) self.assertIn("business_review:answer_layering_noise", codes) self.assertEqual(reviewed["review_status"], "fail") if __name__ == "__main__": unittest.main()