ОРРКЕСТРАЦИЯ - Оркестрация домена: ввести step-validator и починить исторический as_of_date в item-trace

2026-04-14 23:01:23 +03:00
parent c020ef08e1
commit 82a020e302
16 changed files with 1414 additions and 75 deletions
@@ -35,6 +35,31 @@ SCENARIO_PACK_SCHEMA_VERSION = "domain_scenario_pack_v1"
 ACTIVE_DOMAIN_CONTRACT_SCHEMA_VERSION = "active_domain_contract_v1"
 AUTONOMOUS_LOOP_SCHEMA_VERSION = "domain_autonomous_loop_v1"

+TOP_LEVEL_NOISE_PATTERNS = (
+    re.compile(r"^(?:status|статус(?: результата)?)\b", re.IGNORECASE),
+    re.compile(r"^(?:что учтено|сводка)\b", re.IGNORECASE),
+    re.compile(r"^блок\s+\d+\b", re.IGNORECASE),
+    re.compile(r"^(?:подтверждение|опорные документы|сервисно)\b", re.IGNORECASE),
+)
+
+DEFAULT_INVARIANT_SEVERITY: dict[str, str] = {
+    "wrong_intent": "P0",
+    "wrong_capability": "P0",
+    "wrong_followup_action": "P0",
+    "wrong_recipe": "P0",
+    "wrong_result_mode": "P0",
+    "wrong_as_of_date": "P0",
+    "wrong_period_from": "P0",
+    "wrong_period_to": "P0",
+    "missing_required_filter": "P0",
+    "forbidden_capability_selected": "P0",
+    "forbidden_recipe_selected": "P0",
+    "focus_object_missing": "P0",
+    "wrong_date_scope_state": "P0",
+    "direct_answer_missing": "P0",
+    "top_level_noise_present": "P0",
+}
+

 def dump_json(payload: Any) -> str:
    return json.dumps(payload, ensure_ascii=False, indent=2)
@@ -268,6 +293,115 @@ def normalize_string_list(raw_values: Any) -> list[str]:
    return values


+def normalize_validation_filters(raw_filters: Any) -> dict[str, str]:
+    if not isinstance(raw_filters, dict):
+        return {}
+    normalized: dict[str, str] = {}
+    for raw_key, raw_value in raw_filters.items():
+        key = str(raw_key or "").strip()
+        if not key:
+            continue
+        if key in {"as_of_date", "period_from", "period_to"}:
+            normalized_value = normalize_iso_date(raw_value)
+            if normalized_value:
+                normalized[key] = normalized_value
+            continue
+        text_value = str(raw_value or "").strip()
+        if text_value:
+            normalized[key] = text_value
+    return normalized
+
+
+def normalize_invariant_severity(raw_mapping: Any) -> dict[str, str]:
+    if not isinstance(raw_mapping, dict):
+        return {}
+    normalized: dict[str, str] = {}
+    for raw_key, raw_value in raw_mapping.items():
+        key = str(raw_key or "").strip()
+        value = str(raw_value or "").strip().upper()
+        if not key or value not in {"P0", "P1", "WARNING"}:
+            continue
+        normalized[key] = value
+    return normalized
+
+
+def normalize_identifier(value: Any) -> str:
+    return str(value or "").strip().lower()
+
+
+def identifiers_match(actual: Any, expected: Any) -> bool:
+    actual_id = normalize_identifier(actual)
+    expected_id = normalize_identifier(expected)
+    if not actual_id or not expected_id:
+        return False
+    return actual_id == expected_id or actual_id.endswith(expected_id) or expected_id.endswith(actual_id)
+
+
+def identifier_in_list(actual: Any, expected_values: list[str]) -> bool:
+    return any(identifiers_match(actual, expected) for expected in expected_values if expected)
+
+
+def first_non_empty_lines(text: str, limit: int = 3) -> list[str]:
+    output: list[str] = []
+    for raw_line in str(text or "").splitlines():
+        cleaned = raw_line.strip()
+        if not cleaned:
+            continue
+        output.append(cleaned)
+        if len(output) >= limit:
+            break
+    return output
+
+
+def build_node_contract_index(raw_contract: dict[str, Any]) -> dict[str, dict[str, Any]]:
+    scenario_tree = raw_contract.get("scenario_tree")
+    if not isinstance(scenario_tree, dict):
+        return {}
+    node_index: dict[str, dict[str, Any]] = {}
+    for section_key in ("root_nodes", "critical_nodes", "supporting_nodes"):
+        raw_nodes = scenario_tree.get(section_key)
+        if not isinstance(raw_nodes, list):
+            continue
+        for raw_node in raw_nodes:
+            if not isinstance(raw_node, dict):
+                continue
+            node_id = str(raw_node.get("node_id") or "").strip()
+            if not node_id:
+                continue
+            node_index[node_id] = {
+                "expected_intents": normalize_string_list(raw_node.get("expected_intents")),
+                "required_answer_shape": str(raw_node.get("expected_answer_shape") or "").strip() or None,
+                "required_carryover_invariants": normalize_string_list(raw_node.get("required_carryover_invariants")),
+                "ordering_rule": str(raw_node.get("ordering_rule") or "").strip() or None,
+            }
+    return node_index
+
+
+def enrich_step_with_node_contract(raw_step: Any, node_contract_index: dict[str, dict[str, Any]]) -> Any:
+    if not isinstance(raw_step, dict):
+        return raw_step
+    node_id = str(raw_step.get("node_id") or "").strip()
+    node_defaults = node_contract_index.get(node_id) or {}
+    if not node_defaults:
+        return raw_step
+    enriched = dict(raw_step)
+    if not enriched.get("expected_intents") and node_defaults.get("expected_intents"):
+        enriched["expected_intents"] = list(node_defaults["expected_intents"])
+    if not enriched.get("required_answer_shape") and node_defaults.get("required_answer_shape"):
+        enriched["required_answer_shape"] = node_defaults["required_answer_shape"]
+    merged_invariants = list(
+        dict.fromkeys(
+            normalize_string_list(node_defaults.get("required_carryover_invariants"))
+            + normalize_string_list(enriched.get("required_carryover_invariants"))
+        )
+    )
+    if merged_invariants:
+        enriched["required_carryover_invariants"] = merged_invariants
+    if not enriched.get("ordering_rule") and node_defaults.get("ordering_rule"):
+        enriched["ordering_rule"] = node_defaults["ordering_rule"]
+    return enriched
+
+
 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}

@@ -777,8 +911,18 @@ def build_failed_step_state(
        "depends_on": step["depends_on"],
        "question_template": step["question_template"],
        "question_resolved": question_resolved,
+        "expected_intents": step.get("expected_intents") or [],
        "expected_capability": step.get("expected_capability"),
+        "expected_recipe": step.get("expected_recipe"),
        "expected_result_mode": step.get("expected_result_mode"),
+        "required_filters": step.get("required_filters") or {},
+        "forbidden_capabilities": step.get("forbidden_capabilities") or [],
+        "forbidden_recipes": step.get("forbidden_recipes") or [],
+        "required_state_objects": step.get("required_state_objects") or [],
+        "required_answer_shape": step.get("required_answer_shape"),
+        "forbidden_answer_patterns": step.get("forbidden_answer_patterns") or [],
+        "required_carryover_invariants": step.get("required_carryover_invariants") or [],
+        "invariant_severity": step.get("invariant_severity") or {},
        "reply_type": "backend_error" if status == "blocked" else "unresolved_followup",
        "assistant_message_id": None,
        "trace_id": None,
@@ -790,6 +934,11 @@ def build_failed_step_state(
        "route_expectation_status": None,
        "result_mode": None,
        "response_type": None,
+        "assistant_text": "",
+        "top_non_empty_lines": [],
+        "actual_direct_answer": None,
+        "extracted_filters": {},
+        "focus_object": None,
        "fallback_type": failure_type,
        "mcp_call_status": None,
        "balance_confirmed": None,
@@ -798,6 +947,11 @@ def build_failed_step_state(
        "date_scope": None,
        "organization_scope": None,
        "entries": [],
+        "execution_status": status,
+        "acceptance_status": status,
+        "violated_invariants": [],
+        "warnings": [],
+        "hard_fail": status == "blocked",
        "status": status,
        "failure_type": failure_type,
        "error_message": error_message,
@@ -816,13 +970,22 @@ def normalize_step_definition(index: int, raw_step: Any) -> dict[str, Any]:
            "question_template": question_template,
            "depends_on": [],
            "analysis_context": {},
+            "expected_intents": [],
            "expected_capability": None,
+            "expected_recipe": None,
            "expected_result_mode": None,
            "question_id": None,
            "node_id": None,
            "node_role": None,
            "paraphrase_family": None,
+            "required_filters": {},
+            "forbidden_capabilities": [],
+            "forbidden_recipes": [],
+            "required_state_objects": [],
+            "required_answer_shape": None,
+            "forbidden_answer_patterns": [],
            "required_carryover_invariants": [],
+            "invariant_severity": {},
            "ordering_rule": None,
        }
    if not isinstance(raw_step, dict):
@@ -845,13 +1008,24 @@ def normalize_step_definition(index: int, raw_step: Any) -> dict[str, Any]:
        "question_template": question_template,
        "depends_on": depends_on,
        "analysis_context": normalize_analysis_context(raw_step.get("analysis_context")),
+        "expected_intents": normalize_string_list(raw_step.get("expected_intents") or raw_step.get("expected_intent")),
        "expected_capability": str(raw_step.get("expected_capability") or "").strip() or None,
+        "expected_recipe": str(raw_step.get("expected_recipe") or raw_step.get("expected_selected_recipe") or "").strip() or None,
        "expected_result_mode": str(raw_step.get("expected_result_mode") or "").strip() or None,
        "question_id": str(raw_step.get("question_id") or "").strip() or None,
        "node_id": str(raw_step.get("node_id") or "").strip() or None,
        "node_role": str(raw_step.get("node_role") or raw_step.get("role") or "").strip() or None,
        "paraphrase_family": str(raw_step.get("paraphrase_family") or raw_step.get("wording_family") or "").strip() or None,
+        "required_filters": normalize_validation_filters(raw_step.get("required_filters")),
+        "forbidden_capabilities": normalize_string_list(raw_step.get("forbidden_capabilities")),
+        "forbidden_recipes": normalize_string_list(raw_step.get("forbidden_recipes")),
+        "required_state_objects": normalize_string_list(raw_step.get("required_state_objects")),
+        "required_answer_shape": (
+            str(raw_step.get("required_answer_shape") or raw_step.get("expected_answer_shape") or "").strip() or None
+        ),
+        "forbidden_answer_patterns": normalize_string_list(raw_step.get("forbidden_answer_patterns")),
        "required_carryover_invariants": normalize_string_list(raw_step.get("required_carryover_invariants")),
+        "invariant_severity": normalize_invariant_severity(raw_step.get("invariant_severity")),
        "ordering_rule": str(raw_step.get("ordering_rule") or "").strip() or None,
    }

@@ -936,6 +1110,7 @@ def convert_active_domain_contract_to_pack(raw_contract: dict[str, Any]) -> dict
        raise RuntimeError("Active domain contract must define `runtime_domain`")

    domain_id = str(raw_contract.get("domain_id") or runtime_domain).strip() or runtime_domain
+    node_contract_index = build_node_contract_index(raw_contract)
    bindings = build_active_contract_bindings(raw_contract)
    bindings.update(normalize_bindings(orchestration_pack.get("bindings")))

@@ -947,6 +1122,18 @@ def convert_active_domain_contract_to_pack(raw_contract: dict[str, Any]) -> dict
    pack_id = str(orchestration_pack.get("pack_id") or "").strip() or slugify_case_id(domain_id, None)
    title = str(orchestration_pack.get("title") or raw_contract.get("title") or domain_id).strip() or domain_id
    description = str(orchestration_pack.get("description") or raw_contract.get("domain_goal") or "").strip() or None
+    enriched_scenarios: list[Any] = []
+    for raw_scenario in raw_scenarios:
+        if not isinstance(raw_scenario, dict):
+            enriched_scenarios.append(raw_scenario)
+            continue
+        enriched_scenario = dict(raw_scenario)
+        raw_steps = enriched_scenario.get("steps")
+        if isinstance(raw_steps, list):
+            enriched_scenario["steps"] = [
+                enrich_step_with_node_contract(raw_step, node_contract_index) for raw_step in raw_steps
+            ]
+        enriched_scenarios.append(enriched_scenario)

    return {
        "schema_version": SCENARIO_PACK_SCHEMA_VERSION,
@@ -958,7 +1145,7 @@ def convert_active_domain_contract_to_pack(raw_contract: dict[str, Any]) -> dict
        "description": description,
        "analysis_context": analysis_context,
        "bindings": bindings,
-        "scenarios": raw_scenarios,
+        "scenarios": enriched_scenarios,
        "scenario_tree": raw_contract.get("scenario_tree") if isinstance(raw_contract.get("scenario_tree"), dict) else {},
        "acceptance_contract": (
            raw_contract.get("acceptance_contract") if isinstance(raw_contract.get("acceptance_contract"), dict) else {}
@@ -1135,7 +1322,7 @@ def extract_structured_entries(answer_text: str) -> list[dict[str, Any]]:
    return entries


-def derive_step_status(reply_type: str | None, debug_payload: dict[str, Any]) -> str:
+def derive_step_execution_status(reply_type: str | None, debug_payload: dict[str, Any]) -> str:
    if reply_type == "backend_error":
        return "blocked"
    capability_route_mode = str(debug_payload.get("capability_route_mode") or "").strip()
@@ -1156,6 +1343,153 @@ def derive_step_status(reply_type: str | None, debug_payload: dict[str, Any]) ->
    return "needs_exact_capability"


+def derive_step_status(reply_type: str | None, debug_payload: dict[str, Any]) -> str:
+    return derive_step_execution_status(reply_type, debug_payload)
+
+
+def should_require_direct_answer(step_state: dict[str, Any]) -> bool:
+    required_answer_shape = str(step_state.get("required_answer_shape") or "").strip()
+    if required_answer_shape:
+        return True
+    return str(step_state.get("node_role") or "").strip() in {"root", "critical_child"}
+
+
+def is_top_level_noise_line(line: str) -> bool:
+    cleaned = str(line or "").strip()
+    if not cleaned:
+        return False
+    return any(pattern.search(cleaned) for pattern in TOP_LEVEL_NOISE_PATTERNS)
+
+
+def derive_invariant_severity(step_state: dict[str, Any], violation_code: str) -> str:
+    overrides = step_state.get("invariant_severity")
+    if isinstance(overrides, dict):
+        override = str(overrides.get(violation_code) or "").strip().upper()
+        if override in {"P0", "P1", "WARNING"}:
+            return override
+    return DEFAULT_INVARIANT_SEVERITY.get(violation_code, "P1")
+
+
+def acceptance_status_from_execution(execution_status: str, hard_fail: bool) -> str:
+    if execution_status == "blocked":
+        return "blocked"
+    if execution_status == "needs_exact_capability":
+        return "needs_exact_capability"
+    if hard_fail:
+        return "rejected"
+    if execution_status == "exact":
+        return "validated"
+    return "rejected"
+
+
+def validate_step_contract(step_state: dict[str, Any]) -> dict[str, Any]:
+    state = dict(step_state)
+    execution_status = str(state.get("execution_status") or state.get("status") or "").strip() or "needs_exact_capability"
+    actual_direct_answer = str(state.get("actual_direct_answer") or "").strip()
+    top_non_empty_lines = state.get("top_non_empty_lines") if isinstance(state.get("top_non_empty_lines"), list) else []
+    extracted_filters = state.get("extracted_filters") if isinstance(state.get("extracted_filters"), dict) else {}
+    date_scope = state.get("date_scope") if isinstance(state.get("date_scope"), dict) else {}
+    violated_invariants: list[str] = []
+    warnings: list[str] = []
+
+    expected_intents = normalize_string_list(state.get("expected_intents"))
+    if expected_intents and not identifier_in_list(state.get("detected_intent"), expected_intents):
+        violated_invariants.append("wrong_intent")
+
+    expected_capability = state.get("expected_capability")
+    if expected_capability and not identifiers_match(state.get("capability_id"), expected_capability):
+        required_state_objects = set(normalize_string_list(state.get("required_state_objects")))
+        required_state_objects.update(normalize_string_list(state.get("required_carryover_invariants")))
+        violation_code = "wrong_followup_action" if "focus_object" in required_state_objects else "wrong_capability"
+        violated_invariants.append(violation_code)
+
+    expected_recipe = state.get("expected_recipe")
+    if expected_recipe and not identifiers_match(state.get("selected_recipe"), expected_recipe):
+        violated_invariants.append("wrong_recipe")
+
+    expected_result_mode = str(state.get("expected_result_mode") or "").strip()
+    actual_result_mode = str(state.get("result_mode") or "").strip()
+    if expected_result_mode and actual_result_mode and normalize_identifier(actual_result_mode) != normalize_identifier(expected_result_mode):
+        violated_invariants.append("wrong_result_mode")
+
+    for forbidden_capability in normalize_string_list(state.get("forbidden_capabilities")):
+        if identifiers_match(state.get("capability_id"), forbidden_capability):
+            violated_invariants.append("forbidden_capability_selected")
+            break
+
+    for forbidden_recipe in normalize_string_list(state.get("forbidden_recipes")):
+        if identifiers_match(state.get("selected_recipe"), forbidden_recipe):
+            violated_invariants.append("forbidden_recipe_selected")
+            break
+
+    required_filters = normalize_validation_filters(state.get("required_filters"))
+    required_as_of_date_from_context = normalize_iso_date(
+        (state.get("analysis_context") or {}).get("as_of_date") if isinstance(state.get("analysis_context"), dict) else None
+    )
+    if required_as_of_date_from_context and "as_of_date" not in required_filters:
+        required_filters["as_of_date"] = required_as_of_date_from_context
+
+    for filter_key, expected_value in required_filters.items():
+        actual_value = ""
+        if filter_key in {"as_of_date", "period_from", "period_to"}:
+            actual_value = normalize_iso_date(extracted_filters.get(filter_key))
+        else:
+            actual_value = str(extracted_filters.get(filter_key) or "").strip()
+        if not actual_value:
+            violated_invariants.append("missing_required_filter")
+            continue
+        if actual_value != expected_value:
+            if filter_key == "as_of_date":
+                violated_invariants.append("wrong_as_of_date")
+            elif filter_key == "period_from":
+                violated_invariants.append("wrong_period_from")
+            elif filter_key == "period_to":
+                violated_invariants.append("wrong_period_to")
+            else:
+                violated_invariants.append("missing_required_filter")
+
+    required_state_objects = set(normalize_string_list(state.get("required_state_objects")))
+    required_state_objects.update(
+        item
+        for item in normalize_string_list(state.get("required_carryover_invariants"))
+        if item in {"focus_object"}
+    )
+    focus_object = state.get("focus_object") if isinstance(state.get("focus_object"), dict) else {}
+    if "focus_object" in required_state_objects:
+        has_focus_object = bool(str(focus_object.get("object_id") or "").strip() or str(focus_object.get("label") or "").strip())
+        if not has_focus_object:
+            violated_invariants.append("focus_object_missing")
+
+    if "date_scope" in normalize_string_list(state.get("required_carryover_invariants")) and required_filters.get("as_of_date"):
+        current_date_scope = normalize_iso_date(date_scope.get("as_of_date"))
+        if current_date_scope and current_date_scope != required_filters["as_of_date"]:
+            violated_invariants.append("wrong_date_scope_state")
+
+    if should_require_direct_answer(state):
+        if not actual_direct_answer or is_top_level_noise_line(actual_direct_answer):
+            violated_invariants.append("direct_answer_missing")
+
+    first_top_line = str(top_non_empty_lines[0] if top_non_empty_lines else "").strip()
+    if first_top_line and is_top_level_noise_line(first_top_line):
+        violated_invariants.append("top_level_noise_present")
+
+    forbidden_answer_patterns = normalize_string_list(state.get("forbidden_answer_patterns"))
+    if forbidden_answer_patterns and top_non_empty_lines:
+        joined_top_block = "\n".join(str(line) for line in top_non_empty_lines)
+        for pattern in forbidden_answer_patterns:
+            if pattern and re.search(pattern, joined_top_block, flags=re.IGNORECASE):
+                warnings.append(f"forbidden_answer_pattern:{pattern}")
+
+    unique_violations = list(dict.fromkeys(violated_invariants))
+    hard_fail = any(derive_invariant_severity(state, code) == "P0" for code in unique_violations)
+    state["violated_invariants"] = unique_violations
+    state["warnings"] = list(dict.fromkeys(warnings))
+    state["hard_fail"] = hard_fail
+    state["acceptance_status"] = acceptance_status_from_execution(execution_status, hard_fail)
+    state["status"] = state["acceptance_status"]
+    return state
+
+
 def build_scenario_step_state(
    *,
    scenario_id: str,
@@ -1167,6 +1501,9 @@ def build_scenario_step_state(
    entries: list[dict[str, Any]],
 ) -> dict[str, Any]:
    debug_payload = turn_artifact.get("technical_debug_payload")
+    if not isinstance(debug_payload, dict):
+        assistant_debug = turn_artifact.get("assistant_message", {}).get("debug") if isinstance(turn_artifact.get("assistant_message"), dict) else None
+        debug_payload = assistant_debug if isinstance(assistant_debug, dict) else {}
    debug = debug_payload if isinstance(debug_payload, dict) else {}
    session_summary = turn_artifact.get("session_summary")
    summary = session_summary if isinstance(session_summary, dict) else {}
@@ -1177,6 +1514,9 @@ def build_scenario_step_state(
    assistant_message = turn_artifact.get("assistant_message")
    assistant_item = assistant_message if isinstance(assistant_message, dict) else {}
    reply_type = assistant_item.get("reply_type")
+    assistant_text = str(assistant_item.get("text") or "")
+    top_non_empty = first_non_empty_lines(assistant_text, limit=3)
+    analysis_context = step.get("analysis_context") if isinstance(step.get("analysis_context"), dict) else {}

    step_state = {
        "schema_version": SCENARIO_STEP_STATE_SCHEMA_VERSION,
@@ -1188,8 +1528,19 @@ def build_scenario_step_state(
        "depends_on": step["depends_on"],
        "question_template": step["question_template"],
        "question_resolved": question_resolved,
+        "analysis_context": analysis_context,
+        "expected_intents": step.get("expected_intents") or [],
        "expected_capability": step.get("expected_capability"),
+        "expected_recipe": step.get("expected_recipe"),
        "expected_result_mode": step.get("expected_result_mode"),
+        "required_filters": step.get("required_filters") or {},
+        "forbidden_capabilities": step.get("forbidden_capabilities") or [],
+        "forbidden_recipes": step.get("forbidden_recipes") or [],
+        "required_state_objects": step.get("required_state_objects") or [],
+        "required_answer_shape": step.get("required_answer_shape"),
+        "forbidden_answer_patterns": step.get("forbidden_answer_patterns") or [],
+        "required_carryover_invariants": step.get("required_carryover_invariants") or [],
+        "invariant_severity": step.get("invariant_severity") or {},
        "reply_type": reply_type,
        "assistant_message_id": assistant_item.get("message_id"),
        "trace_id": assistant_item.get("trace_id"),
@@ -1201,6 +1552,11 @@ def build_scenario_step_state(
        "route_expectation_status": debug.get("route_expectation_status"),
        "result_mode": debug.get("result_mode"),
        "response_type": debug.get("response_type"),
+        "assistant_text": assistant_text,
+        "top_non_empty_lines": top_non_empty,
+        "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,
        "fallback_type": debug.get("fallback_type"),
        "mcp_call_status": debug.get("mcp_call_status"),
        "balance_confirmed": debug.get("balance_confirmed"),
@@ -1210,8 +1566,10 @@ def build_scenario_step_state(
        "organization_scope": context.get("organization_scope"),
        "entries": entries,
    }
-    step_state["status"] = derive_step_status(reply_type if isinstance(reply_type, str) else None, debug)
-    return step_state
+    step_state["execution_status"] = derive_step_execution_status(reply_type if isinstance(reply_type, str) else None, debug)
+    step_state["acceptance_status"] = step_state["execution_status"]
+    step_state["status"] = step_state["acceptance_status"]
+    return validate_step_contract(step_state)


 def save_scenario_step_bundle(
@@ -1233,8 +1591,8 @@ def save_scenario_step_bundle(
    write_text(step_dir / "resolved_question.txt", f"{step_state['question_resolved']}\n")


-def derive_scenario_status(step_outputs: dict[str, dict[str, Any]]) -> str:
-    statuses = [str(item.get("status") or "") for item in step_outputs.values()]
+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:
        return "blocked"
    if any(status == "blocked" for status in statuses):
@@ -1243,10 +1601,23 @@ def derive_scenario_status(step_outputs: dict[str, dict[str, Any]]) -> str:
        return "needs_exact_capability"
    if any(status == "partial" for status in statuses):
        return "partial"
-    return "accepted"
+    return "exact"


-def build_scenario_summary(manifest: dict[str, Any], scenario_state: dict[str, Any], final_status: str) -> str:
+def derive_scenario_status(step_outputs: dict[str, dict[str, Any]]) -> str:
+    statuses = [str(item.get("acceptance_status") or item.get("status") or "") for item in step_outputs.values()]
+    if not statuses:
+        return "blocked"
+    if any(status == "blocked" for status in statuses):
+        return "blocked"
+    if any(status == "needs_exact_capability" for status in statuses):
+        return "needs_exact_capability"
+    if any(status in {"partial", "rejected"} for status in statuses):
+        return "partial"
+    return "accepted" if all(status == "validated" for status in statuses) else "partial"
+
+
+def build_scenario_summary(manifest: dict[str, Any], scenario_state: dict[str, Any], final_status: str, execution_status: str) -> str:
    lines = [
        "# Scenario summary",
        "",
@@ -1254,6 +1625,7 @@ def build_scenario_summary(manifest: dict[str, Any], scenario_state: dict[str, A
        f"- domain: `{manifest['domain']}`",
        f"- title: {manifest['title']}",
        f"- session_id: `{scenario_state.get('session_id') or 'n/a'}`",
+        f"- execution_status: `{execution_status}`",
        f"- final_status: `{final_status}`",
        "",
        "## Steps",
@@ -1263,20 +1635,27 @@ def build_scenario_summary(manifest: dict[str, Any], scenario_state: dict[str, A
        lines.extend(
            [
                f"{index}. `{step['step_id']}` - {step['question_template']}",
-                f"status: `{step_output.get('status') or 'n/a'}`",
+                f"execution_status: `{step_output.get('execution_status') or 'n/a'}`",
+                f"acceptance_status: `{step_output.get('acceptance_status') or step_output.get('status') or 'n/a'}`",
                f"question_resolved: {step_output.get('question_resolved') or 'n/a'}",
                f"intent: `{step_output.get('detected_intent') or 'n/a'}`",
                f"recipe: `{step_output.get('selected_recipe') or 'n/a'}`",
                f"capability: `{step_output.get('capability_id') or 'n/a'}`",
                f"result_mode: `{step_output.get('result_mode') or 'n/a'}`",
                f"result_set: `{step_output.get('active_result_set_id') or 'n/a'}`",
+                f"violated_invariants: {', '.join(step_output.get('violated_invariants') or []) or 'none'}",
                "",
            ]
        )
    return "\n".join(lines).strip() + "\n"


-def build_scenario_final_status(manifest: dict[str, Any], scenario_state: dict[str, Any], final_status: str) -> str:
+def build_scenario_final_status(
+    manifest: dict[str, Any],
+    scenario_state: dict[str, Any],
+    final_status: str,
+    execution_status: str,
+) -> str:
    reason = {
        "accepted": "all scenario steps executed in one assistant session with no unresolved route or capability gaps",
        "partial": "scenario captured successfully, but at least one step still needs exact capability enablement or route hardening",
@@ -1288,6 +1667,7 @@ def build_scenario_final_status(manifest: dict[str, Any], scenario_state: dict[s
        # Final status

        - status: `{final_status}`
+        - execution_status: `{execution_status}`
        - scenario_id: `{manifest['scenario_id']}`
        - session_id: `{scenario_state.get('session_id') or 'n/a'}`
        - reason: {reason}
@@ -1492,14 +1872,15 @@ def execute_scenario_manifest(
            f"{step['step_id']} -> {result['step_state']['status']}"
        )

+    execution_status = derive_scenario_execution_status(scenario_state["step_outputs"])
    final_status = derive_scenario_status(scenario_state["step_outputs"])
    write_text(scenario_dir / "scenario_output.md", last_export_markdown or "")
-    write_text(scenario_dir / "scenario_summary.md", build_scenario_summary(manifest, scenario_state, final_status))
-    write_text(scenario_dir / "final_status.md", build_scenario_final_status(manifest, scenario_state, final_status))
+    write_text(scenario_dir / "scenario_summary.md", build_scenario_summary(manifest, scenario_state, final_status, execution_status))
+    write_text(scenario_dir / "final_status.md", build_scenario_final_status(manifest, scenario_state, final_status, execution_status))
    if scenario_state.get("session_id"):
        write_text(scenario_dir / "session_id.txt", f"{scenario_state['session_id']}\n")
    print(f"[domain-case-loop] saved scenario artifacts to {scenario_dir}")
-    print(f"[domain-case-loop] final_status={final_status}")
+    print(f"[domain-case-loop] execution_status={execution_status} final_status={final_status}")
    return scenario_state, final_status


@@ -1645,13 +2026,19 @@ def handle_run_scenario(args: argparse.Namespace) -> int:
    return 0


-def build_pack_summary(pack: dict[str, Any], scenario_results: list[dict[str, Any]], final_status: str) -> str:
+def build_pack_summary(
+    pack: dict[str, Any],
+    scenario_results: list[dict[str, Any]],
+    final_status: str,
+    execution_status: str,
+) -> str:
    lines = [
        "# Pack summary",
        "",
        f"- pack_id: `{pack['pack_id']}`",
        f"- domain: `{pack['domain']}`",
        f"- title: {pack['title']}",
+        f"- execution_status: `{execution_status}`",
        f"- final_status: `{final_status}`",
        "",
        "## Scenarios",
@@ -1660,7 +2047,8 @@ def build_pack_summary(pack: dict[str, Any], scenario_results: list[dict[str, An
        lines.extend(
            [
                f"{index}. `{item['scenario_id']}` - {item['title']}",
-                f"status: `{item['final_status']}`",
+                f"execution_status: `{item.get('execution_status') or 'n/a'}`",
+                f"acceptance_status: `{item['final_status']}`",
                f"session_id: `{item.get('session_id') or 'n/a'}`",
                f"artifact_dir: `{item['artifact_dir']}`",
                "",
@@ -1669,7 +2057,12 @@ def build_pack_summary(pack: dict[str, Any], scenario_results: list[dict[str, An
    return "\n".join(lines).strip() + "\n"


-def build_pack_final_status(pack: dict[str, Any], scenario_results: list[dict[str, Any]], final_status: str) -> str:
+def build_pack_final_status(
+    pack: dict[str, Any],
+    scenario_results: list[dict[str, Any]],
+    final_status: str,
+    execution_status: str,
+) -> str:
    expected_scenarios = len(pack.get("scenarios") or [])
    executed_scenarios = len(scenario_results)
    has_missing_scenarios = executed_scenarios < expected_scenarios
@@ -1689,6 +2082,7 @@ def build_pack_final_status(pack: dict[str, Any], scenario_results: list[dict[st
        # Final status

        - status: `{final_status}`
+        - execution_status: `{execution_status}`
        - pack_id: `{pack['pack_id']}`
        - domain: `{pack['domain']}`
        - reason: {reason}
@@ -1709,6 +2103,19 @@ def derive_coverage_status(statuses: list[str]) -> str:
    return "partial"


+def derive_pack_execution_status(scenario_results: list[dict[str, Any]]) -> str:
+    aggregate_statuses = [str(item.get("execution_status") or "") for item in scenario_results if isinstance(item, dict)]
+    if not aggregate_statuses:
+        return "blocked"
+    if any(status == "blocked" for status in aggregate_statuses):
+        return "blocked"
+    if any(status == "needs_exact_capability" for status in aggregate_statuses):
+        return "needs_exact_capability"
+    if any(status == "partial" for status in aggregate_statuses):
+        return "partial"
+    return "exact"
+
+
 def derive_pack_final_status(pack: dict[str, Any], scenario_results: list[dict[str, Any]]) -> str:
    aggregate_statuses = [item["final_status"] for item in scenario_results]
    if not aggregate_statuses:
@@ -2099,6 +2506,8 @@ def compact_step_output_for_review(step_output: Any) -> dict[str, Any]:
                entry_titles_sample.append(title)
    return {
        "status": step_output.get("status"),
+        "execution_status": step_output.get("execution_status"),
+        "acceptance_status": step_output.get("acceptance_status"),
        "question_resolved": step_output.get("question_resolved"),
        "detected_intent": step_output.get("detected_intent"),
        "selected_recipe": step_output.get("selected_recipe"),
@@ -2106,6 +2515,8 @@ def compact_step_output_for_review(step_output: Any) -> dict[str, Any]:
        "result_mode": step_output.get("result_mode"),
        "answer_shape": step_output.get("answer_shape"),
        "actual_direct_answer": step_output.get("actual_direct_answer"),
+        "violated_invariants": step_output.get("violated_invariants"),
+        "warnings": step_output.get("warnings"),
        "fallback_type": step_output.get("fallback_type"),
        "mcp_call_status": step_output.get("mcp_call_status"),
        "failure_type": step_output.get("failure_type"),
@@ -2140,6 +2551,7 @@ def build_pack_review_bundle(pack_dir: Path) -> str:
            "pack_id": pack_state.get("pack_id"),
            "domain": pack_state.get("domain"),
            "title": pack_state.get("title"),
+            "execution_status": pack_state.get("execution_status"),
            "final_status": pack_state.get("final_status"),
            "scenario_results": pack_state.get("scenario_results"),
        },
@@ -2387,12 +2799,14 @@ def handle_run_pack(args: argparse.Namespace) -> int:
            {
                "scenario_id": scenario_manifest["scenario_id"],
                "title": scenario_manifest["title"],
+                "execution_status": derive_scenario_execution_status(scenario_state.get("step_outputs") or {}),
                "final_status": scenario_final_status,
                "session_id": scenario_state.get("session_id"),
                "artifact_dir": str(scenario_dir),
            }
        )

+    execution_status = derive_pack_execution_status(scenario_results)
    final_status = derive_pack_final_status(pack, scenario_results)

    pack_state = {
@@ -2403,15 +2817,16 @@ def handle_run_pack(args: argparse.Namespace) -> int:
        "analysis_context": pack.get("analysis_context") or {},
        "bindings": pack.get("bindings") or {},
        "scenario_results": scenario_results,
+        "execution_status": execution_status,
        "final_status": final_status,
        "updated_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
    }
    write_text(pack_dir / "scenario_acceptance_matrix.md", build_scenario_acceptance_matrix(pack, scenario_results))
    write_json(pack_dir / "pack_state.json", pack_state)
-    write_text(pack_dir / "pack_summary.md", build_pack_summary(pack, scenario_results, final_status))
-    write_text(pack_dir / "final_status.md", build_pack_final_status(pack, scenario_results, final_status))
+    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))
    print(f"[domain-case-loop] saved pack artifacts to {pack_dir}")
-    print(f"[domain-case-loop] final_status={final_status}")
+    print(f"[domain-case-loop] execution_status={execution_status} final_status={final_status}")
    return 0