From 847a6ed0d507b4a560f46efdb45985ec917fc2dd Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 11:46:08 +0800 Subject: [PATCH] fix(alerts): close controlled canary lifecycle --- apps/api/src/api/v1/webhooks.py | 84 +++++++- .../alert_operation_log_repository.py | 8 +- .../services/awooop_ansible_audit_service.py | 18 ++ .../awooop_ansible_check_mode_service.py | 202 ++++++++++++++++-- .../services/awooop_ansible_post_verifier.py | 14 ++ .../telegram_alert_ai_automation_matrix.py | 18 +- .../tests/test_awooop_truth_chain_service.py | 113 ++++++++++ ...telegram_alert_ai_automation_matrix_api.py | 28 ++- .../tests/test_webhooks_auto_repair_labels.py | 15 +- .../playbooks/awoooi-auto-repair-canary.yml | 73 +++++++ 10 files changed, 528 insertions(+), 45 deletions(-) create mode 100644 infra/ansible/playbooks/awoooi-auto-repair-canary.yml diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index e7605a80c..187021310 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -27,7 +27,7 @@ import hashlib import hmac import time import uuid -from typing import Literal +from typing import Any, Literal from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, status from pydantic import BaseModel, Field @@ -684,6 +684,8 @@ async def _push_to_telegram_background( auto_tuning_command: str = "", # 2026-04-02 ogt: 修復 ai_provider 未傳遞 → Telegram 顯示「AI 仲裁判定」而非具體模型名稱 ai_provider: str = "", + playbook_name: str = "", + automation_state: str = "", # 2026-04-08 ogt: 補傳 incident_id 以啟用詳情/重診/歷史按鈕 incident_id: str = "", # ADR-073 Fix: 傳入 notification_type 以正確路由 TYPE-4D Config Drift 卡片 @@ -791,6 +793,8 @@ async def _push_to_telegram_background( ai_tokens=ai_tokens, ai_cost=ai_cost, ai_provider=ai_provider, + playbook_name=playbook_name, + automation_state=automation_state, incident_id=incident_id, # ADR-075 斷點 B 修復: 傳入分類以啟用動態按鈕 alert_category=alert_category, @@ -2433,19 +2437,40 @@ async def _process_new_alert_background( ) _is_heartbeat = is_heartbeat_alertname(alertname) + _controlled_handoff: dict[str, Any] | None = None + if ( + repair_candidate_result.candidate_found + and _controlled_ai_policy_allows(fallback_create.risk_level) + and not _is_heartbeat + ): + _controlled_handoff = await _try_auto_repair_background( + incident_id=fallback_incident_id, + approval_id=str(approval.id), + alert_type=alert_type, + target_resource=target_resource, + namespace=namespace, + risk_level=fallback_create.risk_level.value, + ) + if _controlled_handoff.get("queued") is True: + telegram_root_cause = ( + "MCP evidence + PlayBook trust 已完成候選判定；" + "AI 已排入 single-writer check-mode / controlled apply / verifier。" + ) + primary_responsibility = "AI_CONTROLLED_EXECUTOR" + else: + queue_blockers = ",".join( + str(value) + for value in _controlled_handoff.get("active_blockers") or [] + ) or "controlled_queue_write_not_acknowledged" + telegram_root_cause = ( + "修復候選已命中，但受控 executor queue 尚未受理；" + f"AI blocker：{queue_blockers}。" + ) + primary_responsibility = "AI_CONTROLLED_QUEUE_REPAIR" if not _is_heartbeat: from src.repositories.alert_operation_log_repository import get_alert_operation_log_repository _op_log_fallback = get_alert_operation_log_repository() if repair_candidate_result.candidate_found: - if _controlled_ai_policy_allows(fallback_create.risk_level): - await _try_auto_repair_background( - incident_id=fallback_incident_id, - approval_id=str(approval.id), - alert_type=alert_type, - target_resource=target_resource, - namespace=namespace, - risk_level=fallback_create.risk_level.value, - ) await _op_log_fallback.append( "PLAYBOOK_DRAFT_CREATED", incident_id=fallback_incident_id, @@ -2456,10 +2481,41 @@ async def _process_new_alert_background( context={ "alertname": alertname, "auto_repair_flag": bool(can_auto_repair), + "controlled_ai_policy_allowed": ( + _controlled_ai_policy_allows( + fallback_create.risk_level + ) + ), + "owner_policy": "global_product_governance_v2", "playbook_id": fallback_create.matched_playbook_id, - "candidate_status": "ready_for_approval", + "candidate_status": ( + str(_controlled_handoff.get("status") or "") + if _controlled_handoff + else "controlled_queue_not_requested" + ), + "automation_run_id": ( + _controlled_handoff.get("automation_run_id") + if _controlled_handoff + else None + ), }, ) + if ( + _controlled_handoff + and _controlled_handoff.get("queued") is not True + ): + await _escalate_auto_repair_unavailable( + incident_id=fallback_incident_id, + approval_id=str(approval.id), + alert_type=alert_type, + target_resource=target_resource, + namespace=namespace, + failure_reason=telegram_root_cause, + attempted_actions=( + "llm_fallback -> mcp_evidence -> playbook_trust -> " + "single_writer_controlled_queue" + ), + ) elif repair_candidate_result.draft_ready_for_owner_review: await _op_log_fallback.append( "PLAYBOOK_DRAFT_CREATED", @@ -2515,6 +2571,12 @@ async def _process_new_alert_background( hit_count=1, primary_responsibility=primary_responsibility, confidence=candidate_confidence, + playbook_name=str(fallback_create.matched_playbook_id or ""), + automation_state=( + str(_controlled_handoff.get("status") or "") + if _controlled_handoff + else "controlled_queue_not_requested" + ), namespace=namespace, incident_id=fallback_incident_id, notification_type=notification_type, diff --git a/apps/api/src/repositories/alert_operation_log_repository.py b/apps/api/src/repositories/alert_operation_log_repository.py index 540be94fb..eaef8a753 100644 --- a/apps/api/src/repositories/alert_operation_log_repository.py +++ b/apps/api/src/repositories/alert_operation_log_repository.py @@ -74,6 +74,7 @@ class AlertOperationLogRepository: success: bool | None = None, error_message: str | None = None, context: dict[str, Any] | None = None, + project_id: str | None = None, ) -> AlertOperationLog | None: """ 寫入一筆操作事件 @@ -102,7 +103,12 @@ class AlertOperationLogRepository: return None try: - async with get_db_context() as db: + db_context = ( + get_db_context(project_id) + if project_id + else get_db_context() + ) + async with db_context as db: record = AlertOperationLog( incident_id=incident_id, approval_id=approval_id, diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index daaf1d8f3..1e4d8f0b5 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -36,6 +36,23 @@ ANSIBLE_OPERATION_TYPES = frozenset({ }) _CATALOG: tuple[dict[str, Any], ...] = ( + { + "catalog_id": "ansible:awoooi-auto-repair-canary", + "playbook_path": "infra/ansible/playbooks/awoooi-auto-repair-canary.yml", + "check_mode_playbook_path": "infra/ansible/playbooks/awoooi-auto-repair-canary.yml", + "inventory_hosts": ["host_121"], + "domains": ["awoooi", "auto_repair", "k3s", "controlled_canary"], + "keywords": [ + "awooopautorepaircanaryt16", + "awoooi-auto-repair-canary", + "auto repair canary", + "controlled canary", + ], + "supports_check_mode": True, + "auto_apply_enabled": True, + "approval_required": False, + "risk_level": "low", + }, { "catalog_id": "ansible:wazuh-manager-posture-readback", "playbook_path": "infra/ansible/playbooks/wazuh-manager-posture-readback.yml", @@ -632,6 +649,7 @@ def build_ansible_decision_audit_payload( for row in candidates[:5] ], "proposal_source": proposal_data.get("source", ""), + "approval_id": str(proposal_data.get("approval_id") or ""), "proposal_risk_level": proposal_risk_level, "proposal_action_preview": str( proposal_data.get("action") diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index ff318a906..41ba82b5b 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -18,13 +18,15 @@ from dataclasses import dataclass, replace from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any +from uuid import UUID import structlog from sqlalchemy import select, text from src.core.config import settings from src.db.base import get_db_context -from src.db.models import PlaybookRecord +from src.db.models import ApprovalRecord, PlaybookRecord +from src.models.approval import ApprovalStatus from src.services.ai_automation_runtime_contract import ( AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION, ) @@ -432,6 +434,7 @@ def build_ansible_check_mode_claim_input( return { "automation_run_id": str(source_candidate_op_id), "incident_id": incident_id, + "approval_id": str(candidate_input.get("approval_id") or ""), "executor": "ansible", "execution_backend": "ansible", "execution_mode": "check_mode", @@ -719,6 +722,107 @@ def _build_apply_result_payload(result: AnsibleRunResult) -> tuple[str, dict[str return status, output, dry_run_result, error +def _approval_id_for_claim(claim: AnsibleCheckModeClaim) -> str: + raw_value = str(claim.input_payload.get("approval_id") or "").strip() + if not raw_value: + return "" + try: + return str(UUID(raw_value)) + except ValueError: + return "" + + +async def _append_alert_lifecycle_receipt( + claim: AnsibleCheckModeClaim, + event_type: str, + *, + apply_op_id: str, + success: bool, + action_detail: str, + project_id: str, + error_message: str | None = None, + post_verifier_passed: bool | None = None, +) -> bool: + from src.repositories.alert_operation_log_repository import ( + get_alert_operation_log_repository, + ) + + record = await get_alert_operation_log_repository().append( + event_type, + incident_id=claim.incident_id, + approval_id=_approval_id_for_claim(claim) or None, + actor="awoooi-ansible-executor-broker", + action_detail=action_detail, + success=success, + error_message=error_message, + context={ + "automation_run_id": _automation_run_id_for_claim(claim), + "catalog_id": claim.catalog_id, + "check_mode_op_id": claim.op_id, + "apply_op_id": apply_op_id, + "single_writer_executor": "awoooi-ansible-executor-broker", + "post_verifier_passed": post_verifier_passed, + }, + project_id=project_id, + ) + return record is not None + + +async def _finalize_controlled_approval_projection( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + apply_op_id: str, + project_id: str, +) -> bool: + approval_id = _approval_id_for_claim(claim) + if not approval_id: + return False + success = bool( + result.returncode == 0 and result.post_verifier_passed is True + ) + try: + async with get_db_context(project_id) as db: + selected = await db.execute( + select(ApprovalRecord).where(ApprovalRecord.id == approval_id) + ) + approval = selected.scalar_one_or_none() + if approval is None: + return False + metadata = dict(approval.extra_metadata or {}) + metadata.update({ + "automation_run_id": _automation_run_id_for_claim(claim), + "single_writer_executor": "awoooi-ansible-executor-broker", + "execution_kind": "controlled_apply", + "repair_attempted": True, + "repair_executed": result.returncode == 0, + "post_verifier_passed": result.post_verifier_passed is True, + "apply_op_id": apply_op_id, + }) + approval.extra_metadata = metadata + approval.status = ( + ApprovalStatus.EXECUTION_SUCCESS + if success + else ApprovalStatus.EXECUTION_FAILED + ) + approval.resolved_at = datetime.now(UTC) + if not success: + approval.rejection_reason = _tail( + result.stderr or "independent_post_verifier_failed", + 2000, + ) + except Exception as exc: + logger.warning( + "ansible_controlled_apply_approval_projection_failed", + incident_id=claim.incident_id, + approval_id=approval_id, + apply_op_id=apply_op_id, + error_type=type(exc).__name__, + ) + return False + return True + + def _build_auto_repair_execution_receipt( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -3876,31 +3980,47 @@ async def run_controlled_apply_for_claim( ) apply_op_id = str(inserted.scalar_one()) + lifecycle_started = await _append_alert_lifecycle_receipt( + claim, + "EXECUTION_STARTED", + apply_op_id=apply_op_id, + success=True, + action_detail=f"controlled_apply_started:{claim.catalog_id}", + project_id=project_id, + ) cancelled = False - try: - spec = build_ansible_apply_command( - playbook_path=claim.apply_playbook_path, - inventory_hosts=claim.inventory_hosts, - ) - result = await _run_ansible_command( - spec, - timeout_seconds=bounded_timeout_seconds, - ) - except asyncio.CancelledError: - cancelled = True - result = AnsibleRunResult( - returncode=130, - stdout="", - stderr="ansible_controlled_apply_interrupted_by_worker_shutdown", - duration_ms=0, - ) - except Exception as exc: + if not lifecycle_started: result = AnsibleRunResult( returncode=1, stdout="", - stderr=f"ansible_controlled_apply_runtime_error: {exc}", + stderr="alert_lifecycle_start_receipt_not_persisted", duration_ms=0, ) + else: + try: + spec = build_ansible_apply_command( + playbook_path=claim.apply_playbook_path, + inventory_hosts=claim.inventory_hosts, + ) + result = await _run_ansible_command( + spec, + timeout_seconds=bounded_timeout_seconds, + ) + except asyncio.CancelledError: + cancelled = True + result = AnsibleRunResult( + returncode=130, + stdout="", + stderr="ansible_controlled_apply_interrupted_by_worker_shutdown", + duration_ms=0, + ) + except Exception as exc: + result = AnsibleRunResult( + returncode=1, + stdout="", + stderr=f"ansible_controlled_apply_runtime_error: {exc}", + duration_ms=0, + ) status, output, dry_run_result, error = _build_apply_result_payload(result) async with get_db_context(project_id) as db: @@ -3973,6 +4093,33 @@ async def run_controlled_apply_for_claim( *learning_receipts, ), ) + approval_projection_written = await _finalize_controlled_approval_projection( + claim, + verified_result, + apply_op_id=apply_op_id, + project_id=project_id, + ) + execution_success = bool( + verified_result.returncode == 0 + and verified_result.post_verifier_passed is True + ) + lifecycle_completed = await _append_alert_lifecycle_receipt( + claim, + "EXECUTION_COMPLETED", + apply_op_id=apply_op_id, + success=execution_success, + action_detail=f"controlled_apply_completed:{claim.catalog_id}", + project_id=project_id, + error_message=( + None + if execution_success + else _tail( + verified_result.stderr or "independent_post_verifier_failed", + 2000, + ) + ), + post_verifier_passed=verified_result.post_verifier_passed, + ) telegram_receipt_sent = await _send_controlled_apply_telegram_receipt( claim, result, @@ -3980,6 +4127,17 @@ async def run_controlled_apply_for_claim( writeback=writeback, project_id=project_id, ) + lifecycle_telegram_written = False + if telegram_receipt_sent: + lifecycle_telegram_written = await _append_alert_lifecycle_receipt( + claim, + "TELEGRAM_RESULT_SENT", + apply_op_id=apply_op_id, + success=execution_success, + action_detail="controlled_apply_result_receipt_sent", + project_id=project_id, + post_verifier_passed=verified_result.post_verifier_passed, + ) logger.info( "ansible_controlled_apply_completed", @@ -3999,7 +4157,11 @@ async def run_controlled_apply_for_claim( playbook_trust_written=writeback.get("trust_learning"), timeline_projection_written=timeline_receipt is not None, runtime_stage_receipts_written=runtime_stage_receipts_written, + approval_projection_written=approval_projection_written, + alert_lifecycle_started=lifecycle_started, + alert_lifecycle_completed=lifecycle_completed, telegram_receipt_sent=telegram_receipt_sent, + alert_lifecycle_telegram_written=lifecycle_telegram_written, ) return verified_result diff --git a/apps/api/src/services/awooop_ansible_post_verifier.py b/apps/api/src/services/awooop_ansible_post_verifier.py index 13bb5a944..ecfb56cec 100644 --- a/apps/api/src/services/awooop_ansible_post_verifier.py +++ b/apps/api/src/services/awooop_ansible_post_verifier.py @@ -44,6 +44,20 @@ ProbeRunner = Callable[..., Awaitable[ReadOnlyProbeResult]] _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = { + "ansible:awoooi-auto-repair-canary": ( + AssetPostcondition( + "host_121_awoooi_canary_rollout_ready", + "service", + "host_121", + "sudo -n /usr/local/bin/kubectl --kubeconfig=/etc/rancher/k3s/k3s.yaml -n awoooi-prod rollout status deployment/awoooi-auto-repair-canary --timeout=30s >/dev/null", + ), + AssetPostcondition( + "host_121_awoooi_canary_ready_replicas", + "metric", + "host_121", + "test \"$(sudo -n /usr/local/bin/kubectl --kubeconfig=/etc/rancher/k3s/k3s.yaml -n awoooi-prod get deployment awoooi-auto-repair-canary -o jsonpath='{.status.readyReplicas}:{.status.replicas}')\" = '1:1'", + ), + ), "ansible:wazuh-manager-posture-readback": ( AssetPostcondition( "host_112_wazuh_manager_unit_loaded", diff --git a/apps/api/src/services/telegram_alert_ai_automation_matrix.py b/apps/api/src/services/telegram_alert_ai_automation_matrix.py index 91d65e255..76a90e288 100644 --- a/apps/api/src/services/telegram_alert_ai_automation_matrix.py +++ b/apps/api/src/services/telegram_alert_ai_automation_matrix.py @@ -164,12 +164,12 @@ def load_latest_telegram_alert_ai_automation_matrix( else "missing_or_not_proven_for_direct_path" ), "controlled_queue": ( - "no_direct_migration_queue_needed" + "ready_not_applicable_no_direct_migration_queue_needed" if direct_gap_closed else "missing_or_not_proven_for_direct_path" ), "post_verifier": ( - "metadata_inventory_clear" + "ready_metadata_inventory_clear" if direct_gap_closed else "metadata_inventory_ready" ), @@ -191,7 +191,7 @@ def load_latest_telegram_alert_ai_automation_matrix( "surface_kind": "ai_digest_policy", "status": "no_send_policy_ready", "known_item_count": int(digest_rollups["rule_count"]), - "db_or_log_receipt": "draft_only_no_runtime_write", + "db_or_log_receipt": "ready_source_policy_receipt_no_runtime_write", "ai_route": "ready_action_required_and_failure_only_rules", "controlled_queue": "ready_policy_gate_no_direct_send", "post_verifier": "ready_policy_loader_validation", @@ -224,12 +224,12 @@ def load_latest_telegram_alert_ai_automation_matrix( else "migration_plan_ready" ), "controlled_queue": ( - "no_direct_migration_queue_needed" + "ready_not_applicable_no_direct_migration_queue_needed" if migration_candidate_count == 0 else "ready_for_controlled_apply_review" ), "post_verifier": ( - "owner_acceptance_ledger_clear" + "ready_owner_acceptance_ledger_clear" if migration_candidate_count == 0 else "owner_acceptance_ledger_ready" ), @@ -253,7 +253,7 @@ def load_latest_telegram_alert_ai_automation_matrix( "known_item_count": source_proof["ai_alert_card_delivery_readback_endpoint_count"], "db_or_log_receipt": "ready_awooop_outbound_message_query", "ai_route": "ready_event_type_lane_target_filters", - "controlled_queue": "read_only_query_no_send", + "controlled_queue": "ready_not_applicable_read_only_query_no_send", "post_verifier": "ready_source_contract_present", "learning_writeback": "ready_delivery_receipt_summary_and_learning_writeback_refs", "manual_default_gap_count": 0, @@ -846,7 +846,11 @@ def _build_summary( source_proof: dict[str, Any], next_priority_action: dict[str, Any], ) -> dict[str, Any]: - ready = lambda key, prefix: sum(1 for item in matrix if str(item[key]).startswith(prefix)) + def ready(key: str, prefix: str) -> int: + return sum( + 1 for item in matrix if str(item[key]).startswith(prefix) + ) + return { "telegram_alert_surface_count": len(matrix), "known_direct_send_gap_count": direct_gap_count, diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index eb3637e69..fc75a8cb2 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -21,6 +21,7 @@ from src.services.awooop_ansible_check_mode_service import ( _EXECUTION_CAPABILITY_FALLBACK_OPERATION_TYPE, AnsibleCheckModeClaim, AnsibleRunResult, + _append_alert_lifecycle_receipt, _append_runtime_stage_receipts_to_apply, _automation_operation_log_incident_id, _build_auto_repair_execution_receipt, @@ -1394,6 +1395,118 @@ def test_ansible_decision_audit_payload_marks_repair_candidate_queue_claimable() assert payload["input"]["execution_priority"] == 0 +def test_ansible_canary_candidate_preserves_approval_and_controlled_apply() -> None: + approval_id = "00000000-0000-0000-0000-000000000044" + incident = SimpleNamespace( + incident_id="INC-CONTROLLED-CANARY", + project_id="awoooi", + alert_category="infrastructure", + notification_type="TYPE-3", + severity=SimpleNamespace(value="P2"), + affected_services=["awoooi-auto-repair-canary"], + signals=[ + SimpleNamespace( + alert_name="AwoooPAutoRepairCanaryT16", + labels={ + "alertname": "AwoooPAutoRepairCanaryT16", + "deployment": "awoooi-auto-repair-canary", + "namespace": "awoooi-prod", + }, + annotations={}, + ) + ], + ) + + payload = build_ansible_decision_audit_payload( + incident=incident, + proposal_data={ + "source": "alert_webhook_controlled_router", + "risk_level": "medium", + "action": "awoooi-auto-repair-canary", + "approval_id": approval_id, + }, + decision_path="repair_candidate_controlled_queue", + not_used_reason="queue bounded no-traffic canary", + ) + + assert payload is not None + assert payload["input"]["approval_id"] == approval_id + candidate = payload["input"]["executor_candidates"][0] + assert candidate["catalog_id"] == "ansible:awoooi-auto-repair-canary" + assert candidate["inventory_hosts"] == ["host_121"] + assert candidate["approval_required"] is False + claim = build_ansible_check_mode_claim_input( + source_candidate_op_id="00000000-0000-0000-0000-000000000045", + candidate_input=payload["input"], + ) + assert claim["approval_id"] == approval_id + assert claim["controlled_apply_allowed"] is True + assert claim["apply_playbook_path"] == ( + "infra/ansible/playbooks/awoooi-auto-repair-canary.yml" + ) + + +@pytest.mark.asyncio +async def test_alert_lifecycle_receipt_uses_same_run_and_project_context( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.repositories import alert_operation_log_repository as repo_module + + captured: dict[str, object] = {} + + class FakeRepository: + async def append(self, event_type: str, **kwargs): + captured["event_type"] = event_type + captured.update(kwargs) + return SimpleNamespace(id="receipt-1") + + monkeypatch.setattr( + repo_module, + "get_alert_operation_log_repository", + lambda: FakeRepository(), + ) + claim = AnsibleCheckModeClaim( + op_id="00000000-0000-0000-0000-000000000046", + source_candidate_op_id="00000000-0000-0000-0000-000000000047", + incident_id="INC-CONTROLLED-CANARY", + catalog_id="ansible:awoooi-auto-repair-canary", + playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml", + apply_playbook_path="infra/ansible/playbooks/awoooi-auto-repair-canary.yml", + inventory_hosts=("host_121",), + risk_level="low", + input_payload={ + "automation_run_id": "00000000-0000-0000-0000-000000000047", + "approval_id": "00000000-0000-0000-0000-000000000048", + }, + ) + + written = await _append_alert_lifecycle_receipt( + claim, + "EXECUTION_STARTED", + apply_op_id="00000000-0000-0000-0000-000000000049", + success=True, + action_detail="controlled_apply_started", + project_id="awoooi", + ) + + assert written is True + assert captured["event_type"] == "EXECUTION_STARTED" + assert captured["project_id"] == "awoooi" + assert captured["approval_id"] == "00000000-0000-0000-0000-000000000048" + assert captured["context"]["automation_run_id"] == ( + "00000000-0000-0000-0000-000000000047" + ) + + +def test_controlled_apply_bridges_terminal_projection_and_telegram_receipt() -> None: + source = inspect.getsource(run_controlled_apply_for_claim) + + assert '"EXECUTION_STARTED"' in source + assert '"EXECUTION_COMPLETED"' in source + assert '"TELEGRAM_RESULT_SENT"' in source + assert "_finalize_controlled_approval_projection" in source + + def test_ansible_decision_audit_payload_exposes_check_mode_safety_flags( monkeypatch, ) -> None: diff --git a/apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py b/apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py index 8dba34709..d4d2a5994 100644 --- a/apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py +++ b/apps/api/tests/test_telegram_alert_ai_automation_matrix_api.py @@ -39,11 +39,11 @@ def test_telegram_alert_ai_automation_matrix_loader_returns_gap_matrix(): assert summary["ops_script_direct_send_gap_count"] == 0 assert summary["api_direct_send_gap_count"] == 0 assert summary["gateway_normalized_callsite_count"] >= 50 - assert summary["db_or_log_receipt_ready_surface_count"] >= 2 - assert summary["ai_route_ready_surface_count"] >= 3 - assert summary["controlled_queue_ready_surface_count"] >= 2 - assert summary["post_verifier_ready_surface_count"] >= 3 - assert summary["learning_writeback_ready_surface_count"] >= 3 + assert summary["db_or_log_receipt_ready_surface_count"] == 9 + assert summary["ai_route_ready_surface_count"] == 9 + assert summary["controlled_queue_ready_surface_count"] == 9 + assert summary["post_verifier_ready_surface_count"] == 9 + assert summary["learning_writeback_ready_surface_count"] == 9 assert summary["manual_default_gap_count"] == 0 assert summary["direct_gap_migration_candidate_count"] == 0 assert summary["owner_acceptance_candidate_count"] == 0 @@ -78,16 +78,34 @@ def test_telegram_alert_ai_automation_matrix_loader_returns_gap_matrix(): assert matrix["telegram_direct_bot_api_egress_gap"]["db_or_log_receipt"] == ( "ready_no_direct_routes_remaining" ) + assert matrix["telegram_direct_bot_api_egress_gap"]["controlled_queue"] == ( + "ready_not_applicable_no_direct_migration_queue_needed" + ) + assert matrix["telegram_direct_bot_api_egress_gap"]["post_verifier"] == ( + "ready_metadata_inventory_clear" + ) assert matrix["telegram_egress_controlled_migration_candidates"]["status"] == ( "controlled_migration_complete_no_candidates" ) assert matrix["telegram_action_required_digest_policy"]["known_item_count"] == 8 + assert matrix["telegram_action_required_digest_policy"]["db_or_log_receipt"] == ( + "ready_source_policy_receipt_no_runtime_write" + ) + assert matrix["telegram_egress_controlled_migration_candidates"][ + "controlled_queue" + ] == "ready_not_applicable_no_direct_migration_queue_needed" + assert matrix["telegram_egress_controlled_migration_candidates"][ + "post_verifier" + ] == "ready_owner_acceptance_ledger_clear" assert matrix["telegram_ai_alert_card_delivery_readback_endpoint"][ "db_or_log_receipt" ] == "ready_awooop_outbound_message_query" assert matrix["telegram_ai_alert_card_delivery_readback_endpoint"][ "learning_writeback" ] == "ready_delivery_receipt_summary_and_learning_writeback_refs" + assert matrix["telegram_ai_alert_card_delivery_readback_endpoint"][ + "controlled_queue" + ] == "ready_not_applicable_read_only_query_no_send" assert matrix["telegram_alert_learning_registry_readback"]["status"] == ( "registry_readback_present" ) diff --git a/apps/api/tests/test_webhooks_auto_repair_labels.py b/apps/api/tests/test_webhooks_auto_repair_labels.py index 1467ae7f1..91ce8df00 100644 --- a/apps/api/tests/test_webhooks_auto_repair_labels.py +++ b/apps/api/tests/test_webhooks_auto_repair_labels.py @@ -1,8 +1,12 @@ from __future__ import annotations +import inspect from types import SimpleNamespace -from src.api.v1.webhooks import _auto_repair_action_label +from src.api.v1.webhooks import ( + _auto_repair_action_label, + _process_new_alert_background, +) def test_auto_repair_action_label_includes_executed_steps() -> None: @@ -26,3 +30,12 @@ def test_auto_repair_action_label_uses_target_when_steps_missing() -> None: assert label == "auto_repair_playbook:PB-TEST api:awoooi-prod" + +def test_fallback_candidate_enters_single_writer_controlled_queue() -> None: + source = inspect.getsource(_process_new_alert_background) + + assert "repair_candidate_result.candidate_found" in source + assert "_controlled_ai_policy_allows(fallback_create.risk_level)" in source + assert "_controlled_handoff = await _try_auto_repair_background" in source + assert 'primary_responsibility = "AI_CONTROLLED_EXECUTOR"' in source + assert '"automation_run_id": (' in source diff --git a/infra/ansible/playbooks/awoooi-auto-repair-canary.yml b/infra/ansible/playbooks/awoooi-auto-repair-canary.yml new file mode 100644 index 000000000..672d64651 --- /dev/null +++ b/infra/ansible/playbooks/awoooi-auto-repair-canary.yml @@ -0,0 +1,73 @@ +--- +- name: AWOOOI bounded auto-repair canary rollout + hosts: host_121 + gather_facts: false + become: false + serial: 1 + any_errors_fatal: true + + vars: + awoooi_kubeconfig: /etc/rancher/k3s/k3s.yaml + awoooi_namespace: awoooi-prod + awoooi_canary_deployment: awoooi-auto-repair-canary + + tasks: + - name: Verify the no-traffic canary deployment is ready + ansible.builtin.command: + argv: + - /usr/bin/sudo + - -n + - /usr/local/bin/kubectl + - --kubeconfig={{ awoooi_kubeconfig }} + - -n + - "{{ awoooi_namespace }}" + - get + - deployment + - "{{ awoooi_canary_deployment }}" + - -o + - jsonpath={.metadata.name}:{.status.readyReplicas}/{.status.replicas} + register: canary_preflight + check_mode: false + changed_when: false + failed_when: >- + canary_preflight.rc != 0 or + canary_preflight.stdout != 'awoooi-auto-repair-canary:1/1' + no_log: true + + - name: Restart only the bounded no-traffic canary + ansible.builtin.command: + argv: + - /usr/bin/sudo + - -n + - /usr/local/bin/kubectl + - --kubeconfig={{ awoooi_kubeconfig }} + - -n + - "{{ awoooi_namespace }}" + - rollout + - restart + - deployment/{{ awoooi_canary_deployment }} + register: canary_restart + when: not ansible_check_mode + changed_when: canary_restart.rc == 0 + failed_when: canary_restart.rc != 0 + no_log: true + + - name: Verify the canary rollout completed + ansible.builtin.command: + argv: + - /usr/bin/sudo + - -n + - /usr/local/bin/kubectl + - --kubeconfig={{ awoooi_kubeconfig }} + - -n + - "{{ awoooi_namespace }}" + - rollout + - status + - deployment/{{ awoooi_canary_deployment }} + - --timeout=90s + register: canary_rollout + check_mode: false + when: not ansible_check_mode + changed_when: false + failed_when: canary_rollout.rc != 0 + no_log: true