Усилить память и статусный gate агентного loop

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 {}
+        scenario_state["semantic_memory"] = build_scenario_semantic_memory_after_step(
-        previous_date_scope = previous_semantic_memory.get("date_scope") if isinstance(previous_semantic_memory, dict) else None
+            scenario_state.get("semantic_memory"),
-        current_date_scope = result["step_state"].get("date_scope")
+            step,
-        scenario_state["semantic_memory"] = {
+            result["step_state"],
-            "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["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.

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()

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()