Усилить семантический контур агента и большой прогон

2026-05-25 13:00:50 +03:00
parent 7cc65e808e
commit 8ce007724a
87 changed files with 4941 additions and 14691 deletions
@@ -1972,6 +1972,11 @@ def derive_step_execution_status(reply_type: str | None, debug_payload: dict[str
        and mcp_discovery_candidate_status == "ready_for_guarded_use"
        and reply_type in {"factual", "factual_with_explanation", "partial_coverage"}
    ):
+        if (
+            mcp_discovery_candidate.get("hot_runtime_wired") is True
+            and debug_payload.get("mcp_discovery_execution_handoff_can_use_guarded_response") is True
+        ):
+            return "exact"
        return "partial"
    capability_route_mode = str(debug_payload.get("capability_route_mode") or "").strip()
    fallback_type = str(debug_payload.get("fallback_type") or "").strip()
@@ -2360,6 +2365,8 @@ def is_validated_memory_checkpoint_answer(
    business_review: dict[str, Any],
    violations: list[str],
 ) -> bool:
+    if is_validated_comparison_boundary_memory_checkpoint(state, business_review, violations):
+        return True
    tags = set(normalize_string_list(state.get("semantic_tags")))
    if "memory" not in tags:
        return False
@@ -2375,6 +2382,46 @@ def is_validated_memory_checkpoint_answer(
    )


+def is_validated_comparison_boundary_memory_checkpoint(
+    state: dict[str, Any],
+    business_review: dict[str, Any],
+    violations: list[str],
+) -> bool:
+    if violations:
+        return False
+    tags = set(normalize_string_list(state.get("semantic_tags")))
+    if not {"business_overview", "selected_object", "organization_clarification"} <= tags:
+        return False
+    comparison_scope = state.get("comparison_scope") if isinstance(state.get("comparison_scope"), dict) else {}
+    organization = comparison_scope.get("organization") if isinstance(comparison_scope.get("organization"), dict) else {}
+    counterparty = comparison_scope.get("counterparty") if isinstance(comparison_scope.get("counterparty"), dict) else {}
+    proof_bundles = comparison_scope.get("proof_bundles") if isinstance(comparison_scope.get("proof_bundles"), dict) else {}
+    value_flow_bundle = (
+        proof_bundles.get("counterparty_value_flow_bundle")
+        if isinstance(proof_bundles.get("counterparty_value_flow_bundle"), dict)
+        else None
+    )
+    document_bundle = (
+        proof_bundles.get("counterparty_document_bundle")
+        if isinstance(proof_bundles.get("counterparty_document_bundle"), dict)
+        else None
+    )
+    organization_label = str(organization.get("label") or "").strip()
+    counterparty_label = str(counterparty.get("label") or "").strip()
+    focus_object = state.get("focus_object") if isinstance(state.get("focus_object"), dict) else {}
+    focus_label = str(focus_object.get("label") or "").strip()
+    if not organization_label or not counterparty_label or not (value_flow_bundle or document_bundle):
+        return False
+    if focus_label and focus_label.casefold() != counterparty_label.casefold():
+        return False
+    return (
+        business_review.get("direct_answer_first_ok") is True
+        and business_review.get("answer_layering_ok") is True
+        and business_review.get("technical_garbage_present") is False
+        and business_review.get("business_usefulness_ok") is True
+    )
+
+
 def is_validated_confirmed_runtime_answer(
    state: dict[str, Any],
    execution_status: str,
@@ -2386,7 +2433,17 @@ def is_validated_confirmed_runtime_answer(
    if violations:
        return False
    if state.get("mcp_discovery_response_applied") is True:
-        return False
+        return (
+            str(state.get("mcp_discovery_response_candidate_status") or "").strip() == "ready_for_guarded_use"
+            and state.get("mcp_discovery_response_candidate_hot_runtime_wired") is True
+            and state.get("mcp_discovery_execution_handoff_can_use_guarded_response") is True
+            and str(state.get("reply_type") or "").strip()
+            in {"factual", "factual_with_explanation", "partial_coverage"}
+            and business_review.get("business_usefulness_ok") is True
+            and business_review.get("direct_answer_first_ok") is True
+            and business_review.get("answer_layering_ok") is True
+            and business_review.get("technical_garbage_present") is False
+        )
    if str(state.get("reply_type") or "").strip() not in {"factual", "factual_with_explanation", "empty_but_valid"}:
        return False
    if str(state.get("fallback_type") or "").strip() not in {"", "none"}:
@@ -2489,6 +2546,33 @@ def is_validated_clarification_answer(
    )


+def is_validated_missing_axis_clarification_answer(
+    state: dict[str, Any],
+    execution_status: str,
+    business_review: dict[str, Any],
+    violations: list[str],
+) -> bool:
+    if execution_status != "partial":
+        return False
+    if violations:
+        return False
+    if str(state.get("mcp_discovery_response_candidate_status") or "").strip() != "clarification_candidate":
+        return False
+    if str(state.get("mcp_discovery_route_candidate_status") or "").strip() != "needs_user_scope":
+        return False
+    if str(state.get("mcp_discovery_response_reply_type") or "").strip() != "clarification_required":
+        return False
+    if not normalize_string_list(state.get("mcp_discovery_route_candidate_missing_axes")):
+        return False
+    return (
+        business_review.get("business_usefulness_ok") is True
+        and business_review.get("direct_answer_first_ok") is True
+        and business_review.get("answer_layering_ok") is True
+        and business_review.get("technical_garbage_present") is False
+        and business_review.get("next_action_present") is True
+    )
+
+
 def _business_review_is_clean(step_state: dict[str, Any]) -> bool:
    business_review = step_state.get("business_first_review")
    if not isinstance(business_review, dict):
@@ -2736,6 +2820,12 @@ def validate_step_contract(step_state: dict[str, Any]) -> dict[str, Any]:
        business_review,
        unique_violations,
    )
+    missing_axis_clarification_validated = is_validated_missing_axis_clarification_answer(
+        state,
+        execution_status,
+        business_review,
+        unique_violations,
+    )
    state["violated_invariants"] = unique_violations
    state["warnings"] = list(dict.fromkeys(warnings))
    state["hard_fail"] = hard_fail
@@ -2744,6 +2834,7 @@ def validate_step_contract(step_state: dict[str, Any]) -> dict[str, Any]:
    state["runtime_factual_answer_validated"] = runtime_factual_validated
    state["guarded_insufficiency_validated"] = guarded_insufficiency_validated
    state["clarification_answer_validated"] = clarification_validated
+    state["missing_axis_clarification_validated"] = missing_axis_clarification_validated
    state["acceptance_status"] = acceptance_status_from_execution(
        execution_status,
        hard_fail,
@@ -2753,6 +2844,7 @@ def validate_step_contract(step_state: dict[str, Any]) -> dict[str, Any]:
            or runtime_factual_validated
            or guarded_insufficiency_validated
            or clarification_validated
+            or missing_axis_clarification_validated
        ),
    )
    state["status"] = state["acceptance_status"]
@@ -2913,6 +3005,7 @@ def build_scenario_step_state(
        "actual_direct_answer": top_non_empty[0] if top_non_empty else None,
        "extracted_filters": debug.get("extracted_filters") if isinstance(debug.get("extracted_filters"), dict) else {},
        "focus_object": context.get("active_focus_object") if isinstance(context.get("active_focus_object"), dict) else None,
+        "comparison_scope": context.get("comparison_scope") if isinstance(context.get("comparison_scope"), dict) else None,
        "fallback_type": debug.get("fallback_type"),
        "mcp_call_status": debug.get("mcp_call_status"),
        "balance_confirmed": debug.get("balance_confirmed"),
@@ -2947,6 +3040,20 @@ def save_scenario_step_bundle(
    write_text(step_dir / "resolved_question.txt", f"{step_state['question_resolved']}\n")


+def is_effectively_complete_partial_step(step_output: dict[str, Any]) -> bool:
+    execution_status = str(step_output.get("execution_status") or step_output.get("status") or "").strip()
+    if execution_status == "exact":
+        return True
+    if execution_status != "partial":
+        return False
+    if str(step_output.get("acceptance_status") or "").strip() != "validated":
+        return False
+    return (
+        step_output.get("missing_axis_clarification_validated") is True
+        or step_output.get("clarification_answer_validated") is True
+    )
+
+
 def derive_scenario_execution_status(step_outputs: dict[str, dict[str, Any]]) -> str:
    statuses = [str(item.get("execution_status") or item.get("status") or "") for item in step_outputs.values()]
    if not statuses:
@@ -2955,7 +3062,7 @@ def derive_scenario_execution_status(step_outputs: dict[str, dict[str, Any]]) ->
        return "blocked"
    if any(status == "needs_exact_capability" for status in statuses):
        return "needs_exact_capability"
-    if any(status == "partial" for status in statuses):
+    if any(not is_effectively_complete_partial_step(item) for item in step_outputs.values()):
        return "partial"
    return "exact"

@@ -3419,6 +3526,7 @@ def build_pack_summary(
    scenario_results: list[dict[str, Any]],
    final_status: str,
    execution_status: str,
+    acceptance_status: str,
 ) -> str:
    lines = [
        "# Pack summary",
@@ -3427,6 +3535,7 @@ def build_pack_summary(
        f"- domain: `{pack['domain']}`",
        f"- title: {pack['title']}",
        f"- execution_status: `{execution_status}`",
+        f"- acceptance_status: `{acceptance_status}`",
        f"- final_status: `{final_status}`",
        "",
        "## Scenarios",
@@ -3450,6 +3559,7 @@ def build_pack_final_status(
    scenario_results: list[dict[str, Any]],
    final_status: str,
    execution_status: str,
+    acceptance_status: str,
 ) -> str:
    expected_scenarios = len(pack.get("scenarios") or [])
    executed_scenarios = len(scenario_results)
@@ -3471,6 +3581,7 @@ def build_pack_final_status(

        - status: `{final_status}`
        - execution_status: `{execution_status}`
+        - acceptance_status: `{acceptance_status}`
        - pack_id: `{pack['pack_id']}`
        - domain: `{pack['domain']}`
        - reason: {reason}
@@ -3522,6 +3633,18 @@ def derive_pack_final_status(pack: dict[str, Any], scenario_results: list[dict[s
    return "accepted" if len(scenario_results) == len(pack.get("scenarios") or []) else "partial"


+def derive_pack_effective_final_status(acceptance_status: str, execution_status: str) -> str:
+    normalized_acceptance = str(acceptance_status or "").strip() or "partial"
+    normalized_execution = str(execution_status or "").strip() or "partial"
+    if normalized_acceptance in {"blocked", "needs_exact_capability"}:
+        return normalized_acceptance
+    if normalized_execution in {"blocked", "needs_exact_capability"}:
+        return normalized_execution
+    if normalized_acceptance == "accepted" and normalized_execution == "exact":
+        return "accepted"
+    return "partial"
+
+
 def build_scenario_acceptance_matrix(pack: dict[str, Any], scenario_results: list[dict[str, Any]]) -> str:
    scenario_status_map = {
        str(item.get("scenario_id") or ""): str(item.get("final_status") or "unknown")
@@ -3906,6 +4029,7 @@ def compact_step_output_for_review(step_output: Any) -> dict[str, Any]:
        "required_carryover_invariants": step_output.get("required_carryover_invariants"),
        "extracted_filters": step_output.get("extracted_filters"),
        "focus_object": step_output.get("focus_object"),
+        "comparison_scope": step_output.get("comparison_scope"),
        "date_scope": step_output.get("date_scope"),
        "result_mode": step_output.get("result_mode"),
        "truth_mode": step_output.get("truth_mode"),
@@ -4436,6 +4560,8 @@ def build_deterministic_repair_targets(
        "schema_version": "domain_pack_repair_targets_v1",
        "pack_id": pack_state.get("pack_id"),
        "domain": pack_state.get("domain"),
+        "execution_status": pack_state.get("execution_status"),
+        "acceptance_status": pack_state.get("acceptance_status"),
        "final_status": pack_state.get("final_status"),
        "target_count": len(targets),
        "severity_counts": severity_counts,
@@ -4588,6 +4714,18 @@ def analyst_target_contradicts_validated_step(target: dict[str, Any], step_snaps
        ):
            return True

+    if step_snapshot.get("memory_checkpoint_validated") is True and evidence_problem:
+        memory_checkpoint_markers = (
+            "machine-readable",
+            "comparison boundary",
+            "carryover",
+            "comparison bundle",
+            "reusable comparison",
+            "selected counterparty",
+        )
+        if any(marker in target_text for marker in memory_checkpoint_markers):
+            return True
+
    followup_problem = problem_type in {
        "followup_action_resolution_gap",
        "object_memory_gap",
@@ -4980,8 +5118,9 @@ def evaluate_deterministic_loop_gate(
    repair_targets: dict[str, Any],
 ) -> tuple[bool, str]:
    pack_final_status = str(pack_state.get("final_status") or "").strip() or "partial"
-    if pack_final_status != "accepted":
-        return False, f"pack_final_status={pack_final_status}"
+    pack_acceptance_status = str(pack_state.get("acceptance_status") or pack_final_status).strip() or "partial"
+    if pack_acceptance_status != "accepted":
+        return False, f"pack_acceptance_status={pack_acceptance_status};pack_final_status={pack_final_status}"
    severity_counts = repair_targets.get("severity_counts") if isinstance(repair_targets, dict) else {}
    if isinstance(severity_counts, dict):
        p0_count = int(severity_counts.get("P0") or 0)
@@ -5120,7 +5259,8 @@ def build_analyst_loop_prompt(

        Rules:
        - `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.final_status != accepted` or the deterministic repair targets still contain any `P0` or `P1` items;
+        - `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;
        - `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.
@@ -5998,7 +6138,8 @@ def handle_run_pack(args: argparse.Namespace) -> int:
        )

    execution_status = derive_pack_execution_status(scenario_results)
-    final_status = derive_pack_final_status(pack, scenario_results)
+    acceptance_status = derive_pack_final_status(pack, scenario_results)
+    final_status = derive_pack_effective_final_status(acceptance_status, execution_status)

    pack_state = {
        "schema_version": SCENARIO_PACK_SCHEMA_VERSION,
@@ -6009,6 +6150,7 @@ def handle_run_pack(args: argparse.Namespace) -> int:
        "bindings": pack.get("bindings") or {},
        "scenario_results": scenario_results,
        "execution_status": execution_status,
+        "acceptance_status": acceptance_status,
        "final_status": final_status,
        "updated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
    }
@@ -6018,8 +6160,8 @@ def handle_run_pack(args: argparse.Namespace) -> int:
    write_json(pack_dir / "pack_state.json", pack_state)
    write_json(pack_dir / "repair_targets.json", repair_targets)
    write_text(pack_dir / "repair_targets.md", build_repair_targets_summary(repair_targets))
-    write_text(pack_dir / "pack_summary.md", build_pack_summary(pack, scenario_results, final_status, execution_status))
-    write_text(pack_dir / "final_status.md", build_pack_final_status(pack, scenario_results, final_status, execution_status))
+    write_text(pack_dir / "pack_summary.md", build_pack_summary(pack, scenario_results, final_status, execution_status, acceptance_status))
+    write_text(pack_dir / "final_status.md", build_pack_final_status(pack, scenario_results, final_status, execution_status, acceptance_status))
    print(f"[domain-case-loop] saved pack artifacts to {pack_dir}")
    print(f"[domain-case-loop] execution_status={execution_status} final_status={final_status}")
    return 0
@@ -135,6 +135,45 @@ COMPANY_PROFIT_ANSWER_RE = re.compile(
    "(?:\u0447\u0438\u0441\u0442\u0430\u044f\\s+\u043f\u0440\u0438\u0431\u044b\u043b\u044c|\u043f\u0440\u0438\u0431\u044b\u043b\u044c\u044e|90/91/99|\u0444\u0438\u043d\u0440\u0435\u0437\u0443\u043b\u044c\u0442\u0430\u0442|\u0443\u0431\u044b\u0442\u043e\u043a)",
    re.IGNORECASE,
 )
+BACKEND_ERROR_ANSWER_RE = re.compile(
+    "(?:backend_error|internal\\s+error|"
+    "\u0432\u043d\u0443\u0442\u0440\u0435\u043d\u043d\\w*\\s+\u043e\u0448\u0438\u0431\u043a|"
+    "\u043d\u0435\\s+\u0443\u0434\u0430\u043b\u043e\u0441\u044c\\s+\u0437\u0430\u0432\u0435\u0440\u0448\u0438\u0442\u044c\\s+\u0440\u0430\u0437\u0431\u043e\u0440)",
+    re.IGNORECASE,
+)
+TOP_YEAR_QUESTION_RE = re.compile(
+    "(?:\u0441\u0430\u043c\\w*\\s+\u0434\u043e\u0445\u043e\u0434\u043d\\w*\\s+\u0433\u043e\u0434|"
+    "\u0434\u043e\u0445\u043e\u0434\u043d\\w*\\s+\u0433\u043e\u0434|"
+    "\u043b\u0443\u0447\u0448\\w*\\s+\u0433\u043e\u0434|top[-\\s]*year|best\\s+year)",
+    re.IGNORECASE,
+)
+DIRECT_TOP_YEAR_ANSWER_RE = re.compile(
+    "(?:(?:\u0441\u0430\u043c\\w*\\s+\u0434\u043e\u0445\u043e\u0434\u043d\\w*\\s+\u0433\u043e\u0434|"
+    "\u043b\u0443\u0447\u0448\\w*\\s+\u0433\u043e\u0434|\u043b\u0438\u0434\u0438\u0440\\w*|"
+    "\u0442\u043e\u043f[-\\s]*1|top[-\\s]*year|best\\s+year)[^\\n]{0,160}(?:19|20)\\d{2}|"
+    "(?:19|20)\\d{2}[^\\n]{0,120}(?:\u043b\u0438\u0434\u0438\u0440\\w*|"
+    "\u0441\u0430\u043c\\w*\\s+\u0434\u043e\u0445\u043e\u0434\u043d\\w*|"
+    "\u043b\u0443\u0447\u0448\\w*\\s+\u0433\u043e\u0434))",
+    re.IGNORECASE,
+)
+GENERIC_BUSINESS_OVERVIEW_ANSWER_RE = re.compile(
+    "(?:\u0432\u044b\u0433\u043b\u044f\u0434\u0438\u0442\\s+\u043a\u0430\u043a\\s+\u0431\u0438\u0437\u043d\u0435\u0441|"
+    "\u043a\u0440\u0443\u043f\u043d\\w*\\s+\u043a\u043e\u043d\u0442\u0440\u0430\u043a\u0442\\w*\\s+\u0434\u0435\u043d\u0435\u0436\u043d\\w*\\s+\u043f\u043e\u0442\u043e\u043a|"
+    "\u0431\u0438\u0437\u043d\u0435\u0441-\u043e\u0431\u0437\u043e\u0440|\u043e\u0431\u0449\\w*\\s+\u0431\u0438\u0437\u043d\u0435\u0441\\w*\\s+\u043e\u0431\u0437\u043e\u0440)",
+    re.IGNORECASE,
+)
+VAT_PURCHASE_DATE_QUESTION_RE = re.compile(
+    "(?:\u043d\u0434\u0441[^\\n]{0,80}\u0434\u0430\u0442[ау]\\s+\u043f\u043e\u043a\u0443\u043f\u043a|"
+    "\u0434\u0430\u0442[ау]\\s+\u043f\u043e\u043a\u0443\u043f\u043a[^\\n]{0,80}\u043d\u0434\u0441)",
+    re.IGNORECASE,
+)
+VAT_PURCHASE_DATE_BASIS_RE = re.compile(
+    "(?:\u0434\u0430\u0442[ауы]\\s+\u043f\u043e\u043a\u0443\u043f\u043a|\u043f\u043e\u043a\u0443\u043f\u043a[аи]\\s+\u043e\u0442|"
+    "\u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d\u0438[ея]\\s+\u043e\u0442|\u043e\u043f\u043e\u0440\u043d\\w*\\s+\u0434\u0430\u0442|"
+    "\u043f\u0435\u0440\u0432\\w*\\s+(?:\u043d\u0430\u0439\u0434\u0435\u043d\u043d\\w*\\s+)?(?:\u043f\u043e\u043a\u0443\u043f\u043a|\u043f\u043e\u0441\u0442\u0443\u043f\u043b\u0435\u043d)|"
+    "\u0434\u0430\u0442\\w*\\s+\u0437\u0430\u043a\u0443\u043f)",
+    re.IGNORECASE,
+)


 def now_iso() -> str:
@@ -360,6 +399,9 @@ def build_step_for_pair(pair: dict[str, Any]) -> dict[str, Any]:
            "business_answer_too_verbose": "P1",
            "bank_counterparty_misclassified_as_business_partner": "P1",
            "counterparty_value_flow_misrouted_to_company_profit": "P0",
+            "backend_error_response": "P0",
+            "year_ranking_direct_answer_missing": "P0",
+            "purchase_date_vat_anchor_ambiguous": "P1",
        },
    }

@@ -374,6 +416,20 @@ def marker_hits(text: str, markers: tuple[str, ...]) -> list[str]:
    return [marker for marker in markers if marker and marker.casefold() in lowered]


+def add_review_issue(
+    *,
+    issue_codes: list[str],
+    root_layers: list[str],
+    issue_code: str,
+    layers: tuple[str, ...],
+) -> None:
+    if issue_code not in issue_codes:
+        issue_codes.append(issue_code)
+    for layer in layers:
+        if layer and layer not in root_layers:
+            root_layers.append(layer)
+
+
 def detect_counterparty_value_flow_profit_mismatch(question: str, assistant_text: str) -> dict[str, Any] | None:
    question_text = str(question or "")
    answer_text = str(assistant_text or "")
@@ -412,6 +468,51 @@ def augment_gui_business_review(step_state: dict[str, Any]) -> dict[str, Any]:
    assistant_text = str(step_state.get("assistant_text") or "")
    issue_codes = [str(item) for item in review.get("issue_codes", []) if str(item).strip()]
    root_layers = [str(item) for item in review.get("suggested_root_cause_layers", []) if str(item).strip()]
+    semantic_details = (
+        dict(review.get("semantic_mismatch_details"))
+        if isinstance(review.get("semantic_mismatch_details"), dict)
+        else {}
+    )
+    first_answer_line = next((line.strip() for line in assistant_text.splitlines() if line.strip()), "")
+
+    if str(step_state.get("reply_type") or "").strip() == "backend_error" or BACKEND_ERROR_ANSWER_RE.search(assistant_text):
+        add_review_issue(
+            issue_codes=issue_codes,
+            root_layers=root_layers,
+            issue_code="backend_error_response",
+            layers=("runtime_error", "business_utility_gap"),
+        )
+        semantic_details["backend_error_response"] = {
+            "reply_type": step_state.get("reply_type"),
+            "first_answer_line": first_answer_line,
+        }
+
+    if TOP_YEAR_QUESTION_RE.search(question):
+        has_direct_top_year = bool(DIRECT_TOP_YEAR_ANSWER_RE.search(assistant_text))
+        generic_overview_first = bool(GENERIC_BUSINESS_OVERVIEW_ANSWER_RE.search(first_answer_line))
+        if not has_direct_top_year or generic_overview_first:
+            add_review_issue(
+                issue_codes=issue_codes,
+                root_layers=root_layers,
+                issue_code="year_ranking_direct_answer_missing",
+                layers=("answer_shape_mismatch", "business_utility_gap"),
+            )
+            semantic_details["year_ranking_direct_answer_missing"] = {
+                "has_direct_top_year": has_direct_top_year,
+                "generic_overview_first": generic_overview_first,
+                "first_answer_line": first_answer_line,
+            }
+
+    if VAT_PURCHASE_DATE_QUESTION_RE.search(question) and not VAT_PURCHASE_DATE_BASIS_RE.search(assistant_text):
+        add_review_issue(
+            issue_codes=issue_codes,
+            root_layers=root_layers,
+            issue_code="purchase_date_vat_anchor_ambiguous",
+            layers=("domain_anchor_gap", "field_mapping_gap"),
+        )
+        semantic_details["purchase_date_vat_anchor_ambiguous"] = {
+            "first_answer_line": first_answer_line,
+        }

    technical_hits = [str(item) for item in review.get("technical_garbage_hits", []) if str(item).strip()]
    for hit in marker_hits(assistant_text, GUI_TECHNICAL_LEAK_MARKERS):
@@ -434,17 +535,18 @@ def augment_gui_business_review(step_state: dict[str, Any]) -> dict[str, Any]:

    mismatch_details = detect_counterparty_value_flow_profit_mismatch(question, assistant_text)
    if mismatch_details:
-        issue_code = "counterparty_value_flow_misrouted_to_company_profit"
-        if issue_code not in issue_codes:
-            issue_codes.append(issue_code)
-        if "followup_action_resolution_gap" not in root_layers:
-            root_layers.append("followup_action_resolution_gap")
-        if "answer_shape_mismatch" not in root_layers:
-            root_layers.append("answer_shape_mismatch")
-        review["semantic_mismatch_details"] = mismatch_details
+        add_review_issue(
+            issue_codes=issue_codes,
+            root_layers=root_layers,
+            issue_code="counterparty_value_flow_misrouted_to_company_profit",
+            layers=("followup_action_resolution_gap", "answer_shape_mismatch"),
+        )
+        semantic_details["counterparty_value_flow_misrouted_to_company_profit"] = mismatch_details

    review["technical_garbage_present"] = bool(technical_hits)
    review["technical_garbage_hits"] = technical_hits
+    if semantic_details:
+        review["semantic_mismatch_details"] = semantic_details
    review["issue_codes"] = issue_codes
    review["suggested_root_cause_layers"] = list(dict.fromkeys(root_layers))
    review["business_usefulness_ok"] = not issue_codes
@@ -104,6 +104,117 @@ class AssistantStage1RunReviewTests(unittest.TestCase):
        self.assertEqual(review["question_quality_review"]["turns_total"], 2)
        self.assertIn("contextual_followup", review["question_quality_review"]["tag_counts"])

+    def test_review_flags_backend_error_as_business_blocker(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sessions_dir = root / "sessions"
+            reports_dir = root / "reports"
+            run_id = "assistant-stage1-backend-error"
+            session_file = sessions_dir / f"{run_id}-SAVED-001.json"
+            report_file = reports_dir / f"{run_id}.md"
+            write_json(
+                session_file,
+                session_payload(
+                    [
+                        {"role": "user", "text": "приветик - че как там дела"},
+                        {
+                            "role": "assistant",
+                            "text": "Сейчас не удалось завершить разбор из-за внутренней ошибки контуров LLM.",
+                            "reply_type": "backend_error",
+                            "message_id": "a-backend-error",
+                            "trace_id": "trace-backend-error",
+                            "debug": {},
+                        },
+                    ]
+                ),
+            )
+            report_file.parent.mkdir(parents=True, exist_ok=True)
+            report_file.write_text(f"# Assistant Stage 1 Eval Run\n\n- run_id: {run_id}\n", encoding="utf-8")
+
+            review = reviewer.build_run_review(
+                run_id=run_id,
+                session_files=[session_file],
+                report_path=report_file,
+            )
+
+        self.assertEqual(review["summary"]["overall_business_status"], "fail")
+        self.assertIn("backend_error_response", review["summary"]["issue_counts"])
+        target_by_issue = {item["issue_code"]: item for item in review["repair_targets"]}
+        self.assertEqual(target_by_issue["backend_error_response"]["severity"], "P0")
+
+    def test_review_flags_top_year_generic_overview_without_direct_year(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sessions_dir = root / "sessions"
+            reports_dir = root / "reports"
+            run_id = "assistant-stage1-top-year-generic"
+            session_file = sessions_dir / f"{run_id}-SAVED-001.json"
+            report_file = reports_dir / f"{run_id}.md"
+            write_json(
+                session_file,
+                session_payload(
+                    [
+                        {"role": "user", "text": "какой у нас самый доходный год"},
+                        {
+                            "role": "assistant",
+                            "text": "Коротко: по доступным данным компания выглядит как бизнес с крупными контрактными денежными потоками. Входящие деньги за доступное время: 10 000 руб.",
+                            "reply_type": "partial_coverage",
+                            "message_id": "a-top-year-generic",
+                            "trace_id": "trace-top-year-generic",
+                            "debug": {"capability_id": "address_customer_revenue_and_payments"},
+                        },
+                    ]
+                ),
+            )
+            report_file.parent.mkdir(parents=True, exist_ok=True)
+            report_file.write_text(f"# Assistant Stage 1 Eval Run\n\n- run_id: {run_id}\n", encoding="utf-8")
+
+            review = reviewer.build_run_review(
+                run_id=run_id,
+                session_files=[session_file],
+                report_path=report_file,
+            )
+
+        self.assertEqual(review["summary"]["overall_business_status"], "fail")
+        self.assertIn("year_ranking_direct_answer_missing", review["summary"]["issue_counts"])
+        target_by_issue = {item["issue_code"]: item for item in review["repair_targets"]}
+        self.assertEqual(target_by_issue["year_ranking_direct_answer_missing"]["severity"], "P0")
+
+    def test_review_accepts_direct_top_year_answer(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            sessions_dir = root / "sessions"
+            reports_dir = root / "reports"
+            run_id = "assistant-stage1-top-year-clean"
+            session_file = sessions_dir / f"{run_id}-SAVED-001.json"
+            report_file = reports_dir / f"{run_id}.md"
+            write_json(
+                session_file,
+                session_payload(
+                    [
+                        {"role": "user", "text": "какой у нас самый доходный год"},
+                        {
+                            "role": "assistant",
+                            "text": "Самый доходный год по подтвержденным поступлениям: 2021 (320 000 руб. по 1 операции). Это денежный поток, а не чистая прибыль.",
+                            "reply_type": "factual",
+                            "message_id": "a-top-year-clean",
+                            "trace_id": "trace-top-year-clean",
+                            "debug": {"capability_id": "address_customer_revenue_and_payments"},
+                        },
+                    ]
+                ),
+            )
+            report_file.parent.mkdir(parents=True, exist_ok=True)
+            report_file.write_text(f"# Assistant Stage 1 Eval Run\n\n- run_id: {run_id}\n", encoding="utf-8")
+
+            review = reviewer.build_run_review(
+                run_id=run_id,
+                session_files=[session_file],
+                report_path=report_file,
+            )
+
+        self.assertNotIn("year_ranking_direct_answer_missing", review["summary"]["issue_counts"])
+
    def test_save_run_review_materializes_machine_and_markdown_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            output_dir = Path(tmp) / "review"