Planner Autonomy: проверять catalog-alignment в truth harness

scripts/domain_truth_harness.py

@@ -24,6 +24,9 @@ TECHNICAL_QUESTION_FIELDS = (
     "expected_capability",
     "expected_recipe",
     "expected_result_mode",
+    "expected_catalog_alignment_status",
+    "expected_catalog_chain_top_match",
+    "expected_catalog_selected_matches_top",
     "required_filters",
     "forbidden_capabilities",
     "forbidden_recipes",
@@ -89,6 +92,13 @@ def normalize_step_spec(index: int, raw_step: Any) -> dict[str, Any]:
     normalized_step["allowed_limited_reason_categories"] = normalize_pattern_list(
         step.get("allowed_limited_reason_categories")
     )
+    normalized_step["expected_catalog_alignment_status"] = (
+        str(step.get("expected_catalog_alignment_status") or "").strip() or None
+    )
+    normalized_step["expected_catalog_chain_top_match"] = (
+        str(step.get("expected_catalog_chain_top_match") or "").strip() or None
+    )
+    normalized_step["expected_catalog_selected_matches_top"] = step.get("expected_catalog_selected_matches_top")
     normalized_step["required_answer_patterns_any"] = normalize_pattern_list(step.get("required_answer_patterns_any"))
     normalized_step["required_answer_patterns_all"] = normalize_pattern_list(step.get("required_answer_patterns_all"))
     normalized_step["required_direct_answer_patterns_any"] = normalize_pattern_list(
@@ -312,6 +322,17 @@ def normalize_actual_filter_value(filter_key: str, raw_value: Any) -> str:
     return str(raw_value or "").strip()
 
 
+def normalize_optional_bool(value: Any) -> bool | None:
+    if isinstance(value, bool):
+        return value
+    raw = str(value or "").strip().lower()
+    if raw in {"true", "1", "yes", "y"}:
+        return True
+    if raw in {"false", "0", "no", "n"}:
+        return False
+    return None
+
+
 def evaluate_truth_step(
     *,
     step: dict[str, Any],
@@ -328,6 +349,7 @@ def evaluate_truth_step(
     selected_recipe = str(step_state.get("selected_recipe") or "").strip()
     capability_id = str(step_state.get("capability_id") or "").strip()
     catalog_alignment_status = str(step_state.get("mcp_discovery_catalog_chain_alignment_status") or "").strip()
+    catalog_chain_top_match = str(step_state.get("mcp_discovery_catalog_chain_top_match") or "").strip()
     limited_reason_category = str(step_state.get("limited_reason_category") or "").strip()
     extracted_filters = (
         step_state.get("extracted_filters") if isinstance(step_state.get("extracted_filters"), dict) else {}
@@ -351,6 +373,64 @@ def evaluate_truth_step(
             severity="warning",
         )
 
+    expected_catalog_alignment_status = str(
+        resolve_nested_placeholders(
+            step.get("expected_catalog_alignment_status"),
+            step_results,
+            bindings,
+            runtime_bindings,
+        )
+        or ""
+    ).strip()
+    if expected_catalog_alignment_status and catalog_alignment_status != expected_catalog_alignment_status:
+        append_finding(
+            findings,
+            step,
+            "wrong_catalog_alignment_status",
+            "Catalog-chain alignment status does not match the expected planner/catalog verdict for this step.",
+            actual=catalog_alignment_status or None,
+            expected=expected_catalog_alignment_status,
+        )
+
+    expected_catalog_chain_top_match = str(
+        resolve_nested_placeholders(
+            step.get("expected_catalog_chain_top_match"),
+            step_results,
+            bindings,
+            runtime_bindings,
+        )
+        or ""
+    ).strip()
+    if expected_catalog_chain_top_match and catalog_chain_top_match != expected_catalog_chain_top_match:
+        append_finding(
+            findings,
+            step,
+            "wrong_catalog_chain_top_match",
+            "Top reviewed catalog-chain match does not match the expected chain for this step.",
+            actual=catalog_chain_top_match or None,
+            expected=expected_catalog_chain_top_match,
+        )
+
+    expected_catalog_selected_matches_top = normalize_optional_bool(
+        resolve_nested_placeholders(
+            step.get("expected_catalog_selected_matches_top"),
+            step_results,
+            bindings,
+            runtime_bindings,
+        )
+    )
+    if expected_catalog_selected_matches_top is not None:
+        actual_catalog_selected_matches_top = step_state.get("mcp_discovery_catalog_chain_selected_matches_top") is True
+        if actual_catalog_selected_matches_top != expected_catalog_selected_matches_top:
+            append_finding(
+                findings,
+                step,
+                "wrong_catalog_selected_matches_top",
+                "Selected chain top-match flag does not match the expected planner/catalog verdict for this step.",
+                actual=actual_catalog_selected_matches_top,
+                expected=expected_catalog_selected_matches_top,
+            )
+
     if step_state.get("question_resolved") != step["question_template"]:
         append_finding(
             findings,

scripts/scenario_acceptance_policy.py

@@ -145,7 +145,12 @@ def _is_meta_context_code(code: str) -> bool:
 
 
 def _is_catalog_alignment_code(code: str) -> bool:
-    return code == "catalog_alignment_divergence"
+    return code in {
+        "catalog_alignment_divergence",
+        "wrong_catalog_alignment_status",
+        "wrong_catalog_chain_top_match",
+        "wrong_catalog_selected_matches_top",
+    }
 
 
 def _derive_step_invariant_failures(step: dict[str, Any], findings: list[dict[str, Any]]) -> dict[str, bool]:

scripts/test_domain_case_loop_step_state.py

@@ -81,6 +81,39 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
         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")
+
 
 if __name__ == "__main__":
     unittest.main()