diff --git a/scripts/domain_case_loop.py b/scripts/domain_case_loop.py index 5f328c8..6b38cfe 100644 --- a/scripts/domain_case_loop.py +++ b/scripts/domain_case_loop.py @@ -983,6 +983,84 @@ def drop_none_values(payload: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in payload.items() if value is not None} +def first_entry_field(entries: Any, *field_names: str) -> Any: + if not isinstance(entries, list): + return None + for entry in entries: + if not isinstance(entry, dict): + continue + fields = entry.get("fields") if isinstance(entry.get("fields"), dict) else {} + raw_fields = entry.get("raw_fields") if isinstance(entry.get("raw_fields"), dict) else {} + for field_name in field_names: + for candidate in (entry.get(field_name), fields.get(field_name), raw_fields.get(field_name)): + value = str(candidate or "").strip() + if value: + return value + return None + + +def build_reusable_inventory_bundle(step_state: dict[str, Any]) -> dict[str, Any] | None: + intent = str(step_state.get("detected_intent") or "").strip() + if intent not in {"inventory_purchase_provenance_for_item", "inventory_purchase_documents_for_item"}: + return None + entries = step_state.get("entries") if isinstance(step_state.get("entries"), list) else [] + focus_object = step_state.get("focus_object") if isinstance(step_state.get("focus_object"), dict) else {} + if not entries and not focus_object: + return None + return drop_none_values( + { + "source_step_id": step_state.get("step_id"), + "source_intent": intent, + "source_route": step_state.get("last_confirmed_route") or step_state.get("selected_recipe"), + "focus_object": focus_object or None, + "date_scope": step_state.get("date_scope") if isinstance(step_state.get("date_scope"), dict) else None, + "supplier_if_known": first_entry_field(entries, "контрагент", "counterparty", "supplier"), + "first_purchase_date": first_entry_field(entries, "date", "дата"), + "source_document_if_known": first_entry_field(entries, "title", "item", "документ", "document"), + "entries": entries or None, + } + ) + + +def build_scenario_semantic_memory_after_step( + previous_semantic_memory: Any, + step: dict[str, Any], + step_state: dict[str, Any], +) -> dict[str, Any]: + previous_memory = previous_semantic_memory if isinstance(previous_semantic_memory, dict) else {} + previous_date_scope = previous_memory.get("date_scope") + focus_object = step_state.get("focus_object") if isinstance(step_state.get("focus_object"), dict) else None + entries = step_state.get("entries") if isinstance(step_state.get("entries"), list) else None + memory = drop_none_values( + { + "latest_step_id": step.get("step_id"), + "latest_step_status": step_state.get("status"), + "active_result_set_id": step_state.get("active_result_set_id"), + "last_confirmed_route": step_state.get("last_confirmed_route"), + "date_scope": merge_scenario_date_scope( + previous_date_scope, + step_state.get("date_scope"), + depends_on=step.get("depends_on") or [], + ), + "organization_scope": step_state.get("organization_scope"), + "focus_object": focus_object, + "selected_object_ref": focus_object, + "comparison_scope": step_state.get("comparison_scope"), + "entries": entries, + } + ) + bundle = build_reusable_inventory_bundle(step_state) + if bundle: + memory["provenance_bundle"] = bundle + if str(step_state.get("detected_intent") or "").strip() == "inventory_purchase_documents_for_item": + memory["purchase_documents_bundle"] = bundle + if bundle.get("supplier_if_known"): + memory["supplier_if_known"] = bundle.get("supplier_if_known") + if bundle.get("first_purchase_date"): + memory["first_purchase_date"] = bundle.get("first_purchase_date") + return memory + + def http_json(url: str, *, method: str = "GET", payload: dict[str, Any] | None = None, timeout: int = 30) -> dict[str, Any]: data = None headers = {"Accept": "application/json"} @@ -3581,22 +3659,11 @@ def execute_scenario_manifest( scenario_state["session_id"] = result["session_id"] scenario_state["step_outputs"][step["step_id"]] = result["step_state"] - previous_semantic_memory = scenario_state.get("semantic_memory") or {} - previous_date_scope = previous_semantic_memory.get("date_scope") if isinstance(previous_semantic_memory, dict) else None - current_date_scope = result["step_state"].get("date_scope") - scenario_state["semantic_memory"] = { - "latest_step_id": step["step_id"], - "latest_step_status": result["step_state"].get("status"), - "active_result_set_id": result["step_state"].get("active_result_set_id"), - "last_confirmed_route": result["step_state"].get("last_confirmed_route"), - "date_scope": merge_scenario_date_scope( - previous_date_scope, - current_date_scope, - depends_on=step.get("depends_on") or [], - ), - "organization_scope": result["step_state"].get("organization_scope"), - "entries": result["step_state"].get("entries"), - } + scenario_state["semantic_memory"] = build_scenario_semantic_memory_after_step( + scenario_state.get("semantic_memory"), + step, + result["step_state"], + ) scenario_state["updated_at"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat() save_scenario_step_bundle( step_dir=step_dir, @@ -5099,6 +5166,46 @@ def analyst_target_contradicts_validated_step(target: dict[str, Any], step_snaps return False +def analyst_target_contradicts_pack_acceptance_status( + repair_targets: dict[str, Any], + target: dict[str, Any], +) -> bool: + pack_acceptance_status = str(repair_targets.get("acceptance_status") or "").strip() + pack_final_status = str(repair_targets.get("final_status") or "").strip() + if pack_acceptance_status != "accepted" or pack_final_status != "partial": + return False + target_text = " ".join( + str(value or "") + for value in ( + target.get("scenario_id"), + target.get("step_id"), + target.get("problem_type"), + target.get("fix_goal"), + target.get("minimal_patch_direction"), + " ".join(normalize_string_list(target.get("violated_invariants"))), + ) + ).casefold() + mentions_pack_status = ( + "pack_state" in target_text + and "acceptance_status" in target_text + and "final_status" in target_text + ) + if not mentions_pack_status: + return False + if "pack_run" not in target_text and "pack-level" not in target_text: + return False + return any( + marker in target_text + for marker in ( + "запрещает итоговый", + "нет `acceptance_status`", + "acceptance_status is absent", + "without `acceptance_status`", + "legacy-gate", + ) + ) + + def merge_analyst_priority_repair_targets( repair_targets: dict[str, Any], analyst_verdict: dict[str, Any], @@ -5148,6 +5255,19 @@ def merge_analyst_priority_repair_targets( } ) continue + if analyst_target_contradicts_pack_acceptance_status(repair_targets, analyst_target): + suppressed_analyst_targets.append( + { + "target_id": target_id, + "reason": "analyst_target_contradicts_pack_acceptance_status", + "pack_status": { + "acceptance_status": repair_targets.get("acceptance_status"), + "final_status": repair_targets.get("final_status"), + "execution_status": repair_targets.get("execution_status"), + }, + } + ) + continue existing = merged_by_id.get(target_id) if existing: existing_severity = normalize_repair_target_severity(existing.get("severity")) @@ -5514,6 +5634,7 @@ def build_pack_review_bundle(pack_dir: Path) -> str: "domain": pack_state.get("domain"), "title": pack_state.get("title"), "execution_status": pack_state.get("execution_status"), + "acceptance_status": pack_state.get("acceptance_status"), "final_status": pack_state.get("final_status"), "scenario_results": pack_state.get("scenario_results"), }, @@ -5613,6 +5734,7 @@ def build_analyst_loop_prompt( - `accepted` is allowed only if quality_score >= {target_score}, unresolved_p0_count = 0, and regression_detected = false; - `accepted` is forbidden if the evidence bundle shows `pack_state.acceptance_status != accepted` (or, for old artifacts without `acceptance_status`, `pack_state.final_status != accepted`) or the deterministic repair targets still contain any `P0` or `P1` items; - `pack_state.final_status=partial` with `acceptance_status=accepted` means the run still had partial execution evidence; treat it as a proof-strength signal, not as a presentation bug by itself; + - do not create a priority target solely to change `pack_state.final_status=partial` when `pack_state.acceptance_status=accepted`; only create a pack-status target if `acceptance_status` is missing/not accepted or if the user-facing scenario semantics are actually wrong; - `accepted` also requires `direct_answer_ok = true`, `business_usefulness_ok = true`, `temporal_honesty_ok = true`, and `field_truth_ok = true`; - Treat validated bounded MCP discovery as the semantic route when `bounded_mcp_answer_validated = true`, `mcp_discovery_response_applied = true`, `mcp_discovery_response_candidate_status = ready_for_guarded_use`, and `mcp_discovery_effective_intents` matches the business question. The legacy address route may be only a seed lane; do not call that silent heuristic masking unless the selected discovery chain is wrong, not ready, not applied, or the user-facing answer/state is semantically wrong. - Use `mcp_discovery_route_candidate_status`, `mcp_discovery_route_candidate_missing_axes`, and `mcp_discovery_route_candidate_enablement_reason` to distinguish a valid user-scope clarification from a real missing reviewed route. Do not ask the coder to overfit the visible answer when the correct next action is route enablement or missing-axis clarification. diff --git a/scripts/test_domain_case_loop_lead_handoff.py b/scripts/test_domain_case_loop_lead_handoff.py index 0eb1eff..d4c3598 100644 --- a/scripts/test_domain_case_loop_lead_handoff.py +++ b/scripts/test_domain_case_loop_lead_handoff.py @@ -631,6 +631,81 @@ class DomainCaseLoopLeadHandoffTests(unittest.TestCase): self.assertEqual(merged["target_count"], 0) self.assertEqual(merged["severity_counts"], {"P0": 0, "P1": 0, "P2": 0}) + def test_pack_status_target_is_suppressed_when_acceptance_status_is_accepted(self) -> None: + repair_targets = { + "pack_id": "pack_run", + "domain": "inventory_margin_ranking", + "execution_status": "partial", + "acceptance_status": "accepted", + "final_status": "partial", + "target_count": 0, + "severity_counts": {"P0": 0, "P1": 0, "P2": 0}, + "priority_foci": [], + "targets": [], + "step_validation_index": {}, + } + analyst_verdict = { + "priority_targets": [ + { + "scenario_id": "pack_run", + "step_id": None, + "severity": "P1", + "problem_type": "presentation_gap", + "fix_goal": ( + "Довести pack-level статусную семантику до непротиворечивой: либо записывать " + "`pack_state.acceptance_status=accepted`, либо не оставлять legacy-gate в состоянии " + "`final_status=partial`, если pack реально проходит analyst gate; пока это запрещает итоговый accepted." + ), + } + ] + } + + merged = dcl.merge_analyst_priority_repair_targets(repair_targets, analyst_verdict) + + self.assertEqual(merged["suppressed_analyst_priority_target_count"], 1) + self.assertEqual( + merged["suppressed_analyst_priority_targets"][0]["reason"], + "analyst_target_contradicts_pack_acceptance_status", + ) + self.assertEqual(merged["target_count"], 0) + self.assertEqual(merged["severity_counts"], {"P0": 0, "P1": 0, "P2": 0}) + + def test_pack_review_bundle_exposes_acceptance_status(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + pack_dir = Path(tmp) / "pack" + pack_dir.mkdir() + (pack_dir / "pack_state.json").write_text( + json.dumps( + { + "pack_id": "pack_run", + "domain": "inventory_margin_ranking", + "title": "Demo", + "execution_status": "partial", + "acceptance_status": "accepted", + "final_status": "partial", + "scenario_results": [], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + (pack_dir / "repair_targets.json").write_text( + json.dumps( + { + "target_count": 0, + "severity_counts": {"P0": 0, "P1": 0, "P2": 0}, + "targets": [], + }, + ensure_ascii=False, + ), + encoding="utf-8", + ) + + bundle = json.loads(dcl.build_pack_review_bundle(pack_dir)) + + self.assertEqual(bundle["pack_state"]["acceptance_status"], "accepted") + self.assertEqual(bundle["pack_state"]["final_status"], "partial") + if __name__ == "__main__": unittest.main() diff --git a/scripts/test_domain_case_loop_step_state.py b/scripts/test_domain_case_loop_step_state.py index a05c1c2..77926f6 100644 --- a/scripts/test_domain_case_loop_step_state.py +++ b/scripts/test_domain_case_loop_step_state.py @@ -15,6 +15,52 @@ 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()