Planner Autonomy: добавить invariant catalog-alignment

This commit is contained in:
2026-05-01 16:21:10 +03:00
parent 91529d897d
commit e58a9664e0
4 changed files with 69 additions and 3 deletions
+9
View File
@@ -144,6 +144,10 @@ def _is_meta_context_code(code: str) -> bool:
)
def _is_catalog_alignment_code(code: str) -> bool:
return code == "catalog_alignment_divergence"
def _derive_step_invariant_failures(step: dict[str, Any], findings: list[dict[str, Any]]) -> dict[str, bool]:
codes = [str(item.get("code") or "").strip() for item in findings]
selected_object_step = _has_selected_object_signal(step)
@@ -155,6 +159,7 @@ def _derive_step_invariant_failures(step: dict[str, Any], findings: list[dict[st
"truth_gate": any(_is_truth_gate_code(code) for code in codes),
"human_answer_quality": any(_is_human_answer_quality_code(code) for code in codes),
"meta_context_integrity": meta_context_step and any(_is_meta_context_code(code) for code in codes),
"catalog_alignment": any(_is_catalog_alignment_code(code) for code in codes),
}
@@ -171,6 +176,7 @@ def build_scenario_acceptance_matrix(
"truth_gate": 0,
"human_answer_quality": 0,
"meta_context_integrity": 0,
"catalog_alignment": 0,
}
for index, step in enumerate(spec.get("steps") or [], start=1):
@@ -217,6 +223,7 @@ def build_scenario_acceptance_matrix(
"truth_gate_ok": invariant_failure_counts["truth_gate"] == 0,
"human_answer_quality_ok": invariant_failure_counts["human_answer_quality"] == 0,
"meta_context_integrity_ok": invariant_failure_counts["meta_context_integrity"] == 0,
"catalog_alignment_ok": invariant_failure_counts["catalog_alignment"] == 0,
}
critical_rows = [row for row in rows if row["criticality"] == "critical"]
critical_path_green = bool(critical_rows) and all(row["review_status"] == "pass" for row in critical_rows)
@@ -323,6 +330,7 @@ def build_scenario_acceptance_matrix_markdown(acceptance_matrix: dict[str, Any])
f"- truth_gate_ok: `{invariants.get('truth_gate_ok')}`",
f"- human_answer_quality_ok: `{invariants.get('human_answer_quality_ok')}`",
f"- meta_context_integrity_ok: `{invariants.get('meta_context_integrity_ok')}`",
f"- catalog_alignment_ok: `{invariants.get('catalog_alignment_ok')}`",
"",
"## Steps",
]
@@ -359,4 +367,5 @@ def build_truth_harness_final_status_markdown(pack_state: dict[str, Any]) -> str
f"- truth_gate_ok: `{invariants.get('truth_gate_ok')}`\n"
f"- human_answer_quality_ok: `{invariants.get('human_answer_quality_ok')}`\n"
f"- meta_context_integrity_ok: `{invariants.get('meta_context_integrity_ok')}`\n"
f"- catalog_alignment_ok: `{invariants.get('catalog_alignment_ok')}`\n"
)
@@ -163,6 +163,55 @@ class ScenarioAcceptancePolicyTests(unittest.TestCase):
self.assertTrue(row["meta_context_step"])
self.assertIn("meta_context_integrity", row["invariant_failures"])
def test_flags_catalog_alignment_invariant_when_planner_diverges_from_catalog_top(self) -> None:
spec = {
"scenario_id": "demo_planner_alignment",
"domain": "planner_autonomy",
"title": "Planner alignment",
"steps": [
{
"step_id": "step_01",
"title": "Catalog alignment",
"question_template": "проверь цепочку MCP",
"criticality": "critical",
"semantic_tags": ["planner_alignment"],
}
],
}
scenario_state = {
"session_id": "asst-align",
"step_outputs": {
"step_01": {
"review_status": "warning",
"reply_type": "factual",
"detected_intent": "counterparty_turnover",
"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,
"review_findings": [
{"code": "catalog_alignment_divergence", "severity": "warning"},
],
}
},
}
review_summary = {
"review_source": "live_strict_replay",
"overall_status": "warning",
"steps_total": 1,
"steps_passed": 0,
"steps_with_warning": 1,
"steps_failed": 0,
}
acceptance_matrix = sap.build_scenario_acceptance_matrix(spec, scenario_state, review_summary)
pack_state = sap.derive_truth_harness_pack_state(spec, scenario_state, review_summary, acceptance_matrix)
self.assertEqual(pack_state["final_status"], "partial")
self.assertFalse(pack_state["invariants"]["catalog_alignment_ok"])
self.assertEqual(pack_state["unresolved_p1_count"], 1)
self.assertIn("catalog_alignment", acceptance_matrix["rows"][0]["invariant_failures"])
if __name__ == "__main__":
unittest.main()