From e2b928ac19a11474a4eca1458ef1a673cd3946c2 Mon Sep 17 00:00:00 2001 From: dctouch Date: Mon, 15 Jun 2026 20:01:22 +0300 Subject: [PATCH] =?UTF-8?q?=D0=A3=D0=BA=D1=80=D0=B5=D0=BF=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20=D0=B2=D0=B0=D0=BB=D0=B8=D0=B4=D0=B0=D1=86=D0=B8=D1=8E?= =?UTF-8?q?=20=D0=B0=D0=B3=D0=B5=D0=BD=D1=82=D0=BD=D0=BE=D0=B3=D0=BE=20?= =?UTF-8?q?=D0=BB=D1=83=D0=BF=D0=B0=20=D0=BF=D0=BE=20=D1=81=D1=86=D0=B5?= =?UTF-8?q?=D0=BD=D0=B0=D1=80=D0=BD=D1=8B=D0=BC=20=D1=8F=D0=BA=D0=BE=D1=80?= =?UTF-8?q?=D1=8F=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/domain_case_loop.py | 86 +++++++++++++++- scripts/test_domain_case_loop_step_state.py | 103 ++++++++++++++++++++ 2 files changed, 188 insertions(+), 1 deletion(-) diff --git a/scripts/domain_case_loop.py b/scripts/domain_case_loop.py index f4519f2..a424ff2 100644 --- a/scripts/domain_case_loop.py +++ b/scripts/domain_case_loop.py @@ -1695,6 +1695,47 @@ def resolve_question_template(question_template: str, scenario_state: dict[str, return pattern.sub(replace, question_template).strip() +def resolve_scenario_contract_placeholders(value: Any, scenario_state: dict[str, Any]) -> Any: + if isinstance(value, str): + runtime_resolved = str(resolve_runtime_placeholders(value)) + if "{{" not in runtime_resolved: + return runtime_resolved + pattern = re.compile(r"{{\s*([^{}]+?)\s*}}") + + def replace(match: re.Match[str]) -> str: + value = lookup_placeholder_value(match.group(1), scenario_state) + if isinstance(value, (dict, list)): + return dump_json(value) + return str(value) + + return pattern.sub(replace, runtime_resolved) + if isinstance(value, list): + return [resolve_scenario_contract_placeholders(item, scenario_state) for item in value] + if isinstance(value, dict): + return { + key: resolve_scenario_contract_placeholders(item, scenario_state) + for key, item in value.items() + } + return value + + +def resolve_step_contract_placeholders(step: dict[str, Any], scenario_state: dict[str, Any]) -> dict[str, Any]: + resolved_step = dict(step) + resolved_step["required_filters"] = normalize_validation_filters( + resolve_scenario_contract_placeholders(step.get("required_filters") or {}, scenario_state) + ) + for pattern_field in ( + "forbidden_answer_patterns", + "required_answer_patterns_any", + "required_answer_patterns_all", + ): + if pattern_field in step: + resolved_step[pattern_field] = normalize_pattern_list( + resolve_scenario_contract_placeholders(step.get(pattern_field) or [], scenario_state) + ) + return resolved_step + + def build_failed_step_state( *, scenario_id: str, @@ -2313,6 +2354,9 @@ def step_intent_matches_expected(state: dict[str, Any], expected_intents: list[s observed_intents = [ str(state.get("detected_intent") or "").strip(), *normalize_string_list(state.get("mcp_discovery_effective_intents")), + str(state.get("mcp_discovery_selected_chain_id") or "").strip(), + str(state.get("mcp_discovery_route_candidate_fact_family") or "").strip(), + str(state.get("mcp_discovery_route_candidate_action_family") or "").strip(), ] return any(identifier_in_list(intent, expected_intents) for intent in observed_intents if intent) @@ -2551,6 +2595,8 @@ def is_validated_memory_checkpoint_answer( ) -> bool: if is_validated_comparison_boundary_memory_checkpoint(state, business_review, violations): return True + if is_validated_bare_organization_scope_checkpoint(state, business_review, violations): + return True tags = set(normalize_string_list(state.get("semantic_tags"))) if "memory" not in tags: return False @@ -2566,6 +2612,39 @@ def is_validated_memory_checkpoint_answer( ) +def is_validated_bare_organization_scope_checkpoint( + state: dict[str, Any], + business_review: dict[str, Any], + violations: list[str], +) -> bool: + if violations: + return False + tags = set(normalize_string_list(state.get("semantic_tags"))) + if not tags.intersection({"bare_org_scope", "company_scope", "company_selected"}): + return False + if str(state.get("living_chat_response_source") or "").strip() != "deterministic_data_scope_selection_contract": + return False + if str(state.get("living_chat_data_scope_probe_status") or "").strip() != "resolved": + return False + selected_organization = first_non_empty_string( + state.get("living_chat_selected_organization"), + state.get("assistant_active_organization"), + ) + if not selected_organization: + return False + assistant_text = str(state.get("assistant_text") or "") + normalized_answer = normalize_entity_filter_value_for_comparison(assistant_text) + normalized_organization = normalize_entity_filter_value_for_comparison(selected_organization) + if not normalized_organization or normalized_organization not in normalized_answer: + return False + return ( + business_review.get("business_usefulness_ok") is True + and business_review.get("direct_answer_first_ok") is True + and business_review.get("answer_layering_ok") is True + and business_review.get("technical_garbage_present") is False + ) + + def is_validated_comparison_boundary_memory_checkpoint( state: dict[str, Any], business_review: dict[str, Any], @@ -3235,6 +3314,10 @@ def build_scenario_step_state( "detected_intent": debug.get("detected_intent"), "selected_recipe": debug.get("selected_recipe"), "capability_id": debug.get("capability_id"), + "living_chat_response_source": debug.get("living_chat_response_source"), + "living_chat_data_scope_probe_status": debug.get("living_chat_data_scope_probe_status"), + "living_chat_selected_organization": debug.get("living_chat_selected_organization"), + "assistant_active_organization": debug.get("assistant_active_organization"), "capability_contract_id": first_non_empty_string( debug.get("capability_contract_id"), runtime_contract.get("capability_contract_id"), @@ -3600,6 +3683,7 @@ def execute_scenario_manifest( step_dir = steps_dir / step["step_id"] try: resolved_question = resolve_question_template(step["question_template"], scenario_state) + resolved_step = resolve_step_contract_placeholders(step, scenario_state) step_analysis_context = merge_analysis_context(manifest.get("analysis_context"), step.get("analysis_context")) step_analysis_context = carry_forward_analysis_context( scenario_state, @@ -3611,7 +3695,7 @@ def execute_scenario_manifest( args=args, domain=manifest["domain"], scenario_id=manifest["scenario_id"], - step=step, + step=resolved_step, step_index=step_index, session_id=scenario_state.get("session_id"), question_resolved=resolved_question, diff --git a/scripts/test_domain_case_loop_step_state.py b/scripts/test_domain_case_loop_step_state.py index 9b4cac2..cd6e06a 100644 --- a/scripts/test_domain_case_loop_step_state.py +++ b/scripts/test_domain_case_loop_step_state.py @@ -190,6 +190,109 @@ class DomainCaseLoopStepStateTests(unittest.TestCase): self.assertIn("required_answer_patterns_any_missing", step_state["violated_invariants"]) self.assertEqual(step_state["acceptance_status"], "rejected") + def test_binding_placeholder_required_filter_is_resolved_before_step_validation(self) -> None: + step = dcl.normalize_step_definition( + 1, + { + "step_id": "counterparty_binding_filter", + "title": "Counterparty binding filter", + "question": "Отдельно по {{bindings.counterparty}}: какой это поток?", + "required_filters": {"counterparty": "{{bindings.counterparty}}"}, + "required_answer_patterns_all": ["(?i){{bindings.counterparty}}|банк"], + }, + ) + scenario_state = { + "bindings": {"counterparty": "СБЕРБАНК"}, + "step_outputs": {}, + "semantic_memory": {}, + } + + resolved_step = dcl.resolve_step_contract_placeholders(step, scenario_state) + + self.assertEqual(resolved_step["required_filters"]["counterparty"], "СБЕРБАНК") + self.assertEqual(resolved_step["required_answer_patterns_all"], ["(?i)СБЕРБАНК|банк"]) + + step_state = dcl.build_scenario_step_state( + scenario_id="binding_filter_demo", + domain="agentic_loop", + step=resolved_step, + step_index=1, + question_resolved="Отдельно по СБЕРБАНК: какой это поток?", + analysis_context={}, + turn_artifact={ + "assistant_message": { + "reply_type": "factual", + "text": "СБЕРБАНК подтвержден как банковский/финансовый поток.", + "message_id": "msg-1", + "trace_id": "trace-1", + }, + "technical_debug_payload": { + "detected_mode": "address_query", + "detected_intent": "bank_operations_by_counterparty", + "selected_recipe": "address_bank_operations_by_counterparty_v1", + "capability_id": "bank_operations_drilldown", + "truth_gate_contract_status": "full_confirmed", + "mcp_call_status": "matched_non_empty", + "extracted_filters": {"counterparty": "СБЕРБАНК"}, + }, + "session_summary": {}, + }, + entries=[], + ) + + self.assertNotIn("missing_required_filter", step_state["violated_invariants"]) + + def test_expected_intent_accepts_discovery_chain_id(self) -> None: + self.assertTrue( + dcl.step_intent_matches_expected( + { + "detected_intent": "supplier_payouts_profile", + "mcp_discovery_selected_chain_id": "value_flow", + "mcp_discovery_route_candidate_fact_family": "value_flow", + "mcp_discovery_route_candidate_action_family": "payout", + }, + ["value_flow"], + ) + ) + + def test_bare_organization_scope_selection_is_validated_memory_checkpoint(self) -> None: + step_state = dcl.build_scenario_step_state( + scenario_id="org_scope_demo", + domain="agentic_loop", + step={ + "step_id": "step_01", + "title": "Choose organization", + "depends_on": [], + "question_template": "ООО Альтернатива Плюс", + "semantic_tags": ["bare_org_scope", "company_scope"], + "required_answer_patterns_all": ["(?i)фиксир|рабоч.*организац|контур"], + }, + step_index=1, + question_resolved="ООО Альтернатива Плюс", + 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_chat_response_source": "deterministic_data_scope_selection_contract", + "living_chat_data_scope_probe_status": "resolved", + "living_chat_selected_organization": 'ООО "Альтернатива Плюс"', + "assistant_active_organization": 'ООО "Альтернатива Плюс"', + }, + "session_summary": {}, + }, + entries=[], + ) + + self.assertEqual(step_state["status"], "validated") + self.assertTrue(step_state["memory_checkpoint_validated"]) + 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"},