feat(awooop): show ansible apply gate handoff
Some checks failed
Code Review / ai-code-review (push) Successful in 15s
CD Pipeline / tests (push) Successful in 1m36s
CD Pipeline / build-and-deploy (push) Successful in 4m56s
CD Pipeline / post-deploy-checks (push) Successful in 1m35s
Ansible / Reboot Recovery Contract / validate (push) Has been cancelled

This commit is contained in:
Your Name
2026-06-25 21:30:22 +08:00
parent 57de2b0229
commit ead737266a
7 changed files with 324 additions and 0 deletions

View File

@@ -3979,6 +3979,113 @@ def _status_chain_ansible_dry_run_only(
)
def _status_chain_ansible_apply_gate_handoff(
*,
ansible_dry_run_only: bool,
execution_section: dict[str, Any],
incident_ids: list[str],
source_id: str | None,
verification: str,
) -> dict[str, Any] | None:
"""Build the owner-review handoff after Ansible check-mode succeeds."""
if not ansible_dry_run_only:
return None
ansible = execution_section.get("ansible")
if not isinstance(ansible, dict):
return None
candidate_playbooks = (
ansible.get("candidate_playbooks")
if isinstance(ansible.get("candidate_playbooks"), list)
else []
)
first_candidate = (
candidate_playbooks[0]
if candidate_playbooks and isinstance(candidate_playbooks[0], dict)
else {}
)
source_ref = source_id or (incident_ids[0] if incident_ids else "unknown")
safe_source_ref = "".join(
ch if ch.isalnum() or ch in {"-", "_"} else "-"
for ch in str(source_ref)
).strip("-") or "unknown"
catalog_id = (
ansible.get("latest_catalog_id")
or first_candidate.get("catalog_id")
or "ansible-candidate"
)
check_mode_playbook = ansible.get("latest_playbook_path") or "--"
apply_playbook = (
first_candidate.get("playbook_path")
or str(check_mode_playbook).replace("-readonly.yml", ".yml")
or "--"
)
latest_status = str(ansible.get("latest_status") or "unknown").lower()
latest_returncode = str(ansible.get("latest_returncode") or "")
dry_run_passed = latest_status == "success" and latest_returncode in {"", "0"}
verifier_ready = str(verification).lower() in {"verified", "success", "healthy"}
return {
"schema_version": "awooop_automation_handoff_v1",
"kind": "ansible_check_mode_apply_gate",
"status": "owner_review_required",
"source_id": source_ref,
"work_item_id": f"ansible-apply-gate:awoooi:{safe_source_ref}",
"decision_effect": "none",
"runtime_execution_authorized": False,
"writes_runtime_state": False,
"owner_review_gate": "required_before_apply",
"next_action": "owner_review_apply_gate_or_create_verifier_plan",
"asset_ids": {
"dry_run": f"ansible-check-mode:{catalog_id}",
"apply_candidate": f"ansible-apply-candidate:{catalog_id}",
"verifier": f"verifier-plan:{safe_source_ref}",
},
"candidate": {
"catalog_id": catalog_id,
"check_mode_playbook_path": check_mode_playbook,
"apply_playbook_path": apply_playbook,
"risk_level": first_candidate.get("risk_level") or "medium",
"match_score": first_candidate.get("match_score"),
},
"gates": [
{
"key": "dry_run",
"status": "passed" if dry_run_passed else "warning",
"detail": (
f"check={_safe_int(ansible.get('check_mode_total'))}; "
f"rc={ansible.get('latest_returncode') if ansible.get('latest_returncode') is not None else '--'}"
),
},
{
"key": "apply_gate",
"status": "blocked",
"detail": (
f"apply={_safe_int(ansible.get('apply_total'))}; "
f"approval={ansible.get('approval_source') or '--'}"
),
},
{
"key": "verifier",
"status": "passed" if verifier_ready else "blocked",
"detail": f"verification={verification or 'missing'}",
},
],
"owner_review_checklist": [
"確認 check-mode 沒有寫入與破壞性變更",
"確認 apply playbook 與 target selector 精準匹配",
"確認 rollback owner、維護窗口與 blast radius",
"確認 verifier plan 可在 apply 後回寫 Runs / KM / PlayBook trust",
],
"forbidden_actions": [
"no_direct_systemctl_without_owner_review",
"no_ansible_apply_without_approval_receipt",
"no_runtime_gate_raise_from_dry_run_only",
],
}
def _latest_remediation_history_item(
history: dict[str, Any] | None,
) -> dict[str, Any]:
@@ -4820,6 +4927,13 @@ def _build_awooop_status_chain(
source_section = _status_chain_source_section(truth_chain)
if source_correlation is not None:
source_section["correlation"] = source_correlation
automation_handoff = _status_chain_ansible_apply_gate_handoff(
ansible_dry_run_only=ansible_dry_run_only,
execution_section=execution_section,
incident_ids=incident_ids,
source_id=source_id,
verification=str(verification),
)
blockers = [
str(item)
for item in [
@@ -4874,6 +4988,7 @@ def _build_awooop_status_chain(
"needs_human": needs_human,
"next_step": next_step,
"operator_outcome": outcome,
"automation_handoff": automation_handoff,
"blockers": blockers[:8],
"fetch_error": fetch_error,
"evidence": {