Planner Autonomy: предупреждать о divergence catalog-alignment

This commit is contained in:
2026-05-01 15:42:00 +03:00
parent a63742f0d6
commit 91529d897d
4 changed files with 64 additions and 4 deletions
+21 -1
View File
@@ -285,11 +285,12 @@ def append_finding(
*,
actual: Any = None,
expected: Any = None,
severity: str | None = None,
) -> None:
findings.append(
{
"code": code,
"severity": step.get("criticality") or DEFAULT_CRITICALITY,
"severity": severity or step.get("criticality") or DEFAULT_CRITICALITY,
"message": message,
"actual": actual,
"expected": expected,
@@ -326,11 +327,30 @@ def evaluate_truth_step(
detected_intent = str(step_state.get("detected_intent") or "").strip()
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()
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 {}
)
if (
catalog_alignment_status in {"selected_lower_rank", "selected_outside_match_set"}
and not bool(step.get("allow_catalog_alignment_divergence"))
):
append_finding(
findings,
step,
"catalog_alignment_divergence",
"Planner selected chain diverges from the top reviewed catalog-chain match and needs semantic review.",
actual={
"alignment_status": catalog_alignment_status,
"top_match": step_state.get("mcp_discovery_catalog_chain_top_match"),
"selected_matches_top": step_state.get("mcp_discovery_catalog_chain_selected_matches_top"),
},
expected="selected_matches_top or explicit allow_catalog_alignment_divergence",
severity="warning",
)
if step_state.get("question_resolved") != step["question_template"]:
append_finding(
findings,
@@ -8,6 +8,7 @@ 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):
@@ -49,6 +50,37 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
self.assertEqual(step_state["mcp_discovery_catalog_chain_top_match"], "value_flow")
self.assertTrue(step_state["mcp_discovery_catalog_chain_selected_matches_top"])
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")
if __name__ == "__main__":
unittest.main()