fix(api): normalize missing automation blockers
All checks were successful
Code Review / ai-code-review (push) Successful in 13s
CD Pipeline / tests (push) Successful in 1m35s
E2E Health Check / e2e-health (push) Successful in 30s
CD Pipeline / build-and-deploy (push) Successful in 4m32s
CD Pipeline / post-deploy-checks (push) Successful in 1m36s

This commit is contained in:
Your Name
2026-06-25 23:56:06 +08:00
parent 20a3961083
commit 4c85db183e
4 changed files with 75 additions and 1 deletions

View File

@@ -27,6 +27,44 @@ def _first_text(values: list[Any]) -> str | None:
return None
def _verification_blocker_code(value: Any) -> str:
status = str(value or "missing").strip().lower()
if status in {"", "none", "null", "missing", "--"}:
return "verification_missing"
if status in {"success", "passed", "healthy", "verified", "repaired", "ok"}:
return "verification_recorded"
return f"verification_{status}"
def normalize_operator_blockers(
blockers: list[Any],
facts: dict[str, Any] | None = None,
) -> list[str]:
"""Translate gate names into operator-facing missing/degraded blockers.
Automation quality gates are named after the desired evidence
(for example ``verification_recorded``). When that gate is the blocker,
the operator must see what is missing, not the name of the target state.
"""
facts = facts or {}
normalized: list[str] = []
for raw in blockers:
if not raw:
continue
item = str(raw)
if item == "auto_repair_recorded" and _safe_int(
facts.get("auto_repair_execution_records")
) <= 0:
item = "auto_repair_missing"
elif item == "verification_recorded":
item = _verification_blocker_code(facts.get("verification_result"))
elif item == "learning_recorded" and _safe_int(facts.get("knowledge_entries")) <= 0:
item = "learning_missing"
if item not in normalized:
normalized.append(item)
return normalized
def _build_notification(
*,
mode: str,
@@ -222,6 +260,7 @@ def build_operator_outcome(
]
if item
]
blockers = normalize_operator_blockers(blockers, facts)
verification = str(facts.get("verification_result") or "missing")
has_repair_execution = _safe_int(facts.get("effective_execution_records")) > 0 or _safe_int(
facts.get("auto_repair_execution_records")

View File

@@ -61,7 +61,7 @@ from src.services.ollama_failover_manager import (
get_ollama_failover_manager,
)
from src.services.ollama_health_monitor import HealthReport, HealthStatus
from src.services.operator_outcome import build_operator_outcome
from src.services.operator_outcome import build_operator_outcome, normalize_operator_blockers
from src.services.operator_summary_cache import (
get_cached_operator_summary_async,
store_operator_summary_async,
@@ -5038,6 +5038,7 @@ def _build_awooop_status_chain(
]
if item
]
blockers = normalize_operator_blockers(blockers, facts)
if fetch_error:
blockers.append("truth_chain_fetch_failed")
outcome = {}

View File

@@ -1803,6 +1803,8 @@ def test_awooop_status_chain_does_not_treat_ansible_check_mode_as_repair() -> No
chain["operator_outcome"]["execution_result"]["completion_status"]
== "dry_run_completed_no_apply"
)
assert "verification_missing" in chain["blockers"]
assert "verification_recorded" not in chain["blockers"]
assert chain["automation_handoff"]["kind"] == "ansible_check_mode_apply_gate"
assert chain["automation_handoff"]["status"] == "owner_review_required"
assert chain["automation_handoff"]["runtime_execution_authorized"] is False

View File

@@ -8,6 +8,38 @@ from src.services.approval_execution import ApprovalExecutionService
from src.services.operator_outcome import build_operator_outcome
def test_operator_outcome_normalizes_missing_evidence_blockers() -> None:
outcome = build_operator_outcome(
truth_status={
"current_stage": "execution_succeeded",
"stage_status": "success",
"needs_human": True,
"blockers": [],
},
automation_quality={
"verdict": "execution_unverified",
"facts": {
"effective_execution_records": 1,
"auto_repair_execution_records": 0,
"verification_result": None,
"knowledge_entries": 0,
},
"blockers": [
"auto_repair_recorded",
"verification_recorded",
"learning_recorded",
],
},
)
assert "auto_repair_missing" in outcome["blockers"]
assert "verification_missing" in outcome["blockers"]
assert "learning_missing" in outcome["blockers"]
assert "auto_repair_recorded" not in outcome["blockers"]
assert "verification_recorded" not in outcome["blockers"]
assert "learning_recorded" not in outcome["blockers"]
def test_operator_outcome_marks_diagnostic_only_as_manual_action_required() -> None:
outcome = build_operator_outcome(
truth_status={