wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(agent): close verified alert lifecycle
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 3m24s
Details
CD Pipeline / build-and-deploy (push) Successful in 6m38s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m33s
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 13:18:16 +08:00
parent f4fd3ebf87
commit 47ab6613dd
4 changed files with 1048 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py
View File

@@ -27,8 +27,7 @@ from src.services.awooop_ansible_audit_service import (
record_ansible_decision_audit,
)
from src.services.awooop_ansible_check_mode_service import (
backfill_missing_auto_repair_execution_receipts_once,
run_failed_apply_check_mode_replay_once,
preflight_failed_apply_retry_queue_once,
)
from src.services.evidence_snapshot import EvidenceSnapshot
from src.services.pre_decision_investigator import get_pre_decision_investigator
@@ -83,7 +82,18 @@ async def _fetch_missing_candidate_incidents(
frequency_snapshot,
affected_services,
signals,
decision_chain
decision_chain,
(
SELECT approval.id
FROM approval_records approval
WHERE approval.incident_id = incidents.incident_id
AND lower(approval.status::text) IN (
'approved',
'execution_failed'
)
ORDER BY approval.updated_at DESC
LIMIT 1
) AS approval_id
FROM incidents
WHERE (project_id = :project_id OR project_id IS NULL)
AND created_at >= NOW() - (:window_hours * INTERVAL '1 hour')
@@ -97,6 +107,13 @@ async def _fetch_missing_candidate_incidents(
AND existing.input ->> 'executor' = 'ansible'
AND existing.input ->> 'decision_path' = 'repair_candidate_controlled_queue'
AND coalesce(existing.incident_id::text, existing.input ->> 'incident_id') = incidents.incident_id::text
AND NOT EXISTS (
SELECT 1
FROM automation_operation_log terminal
WHERE terminal.parent_op_id = existing.op_id
AND terminal.operation_type
= 'ansible_execution_skipped'
)
)
ORDER BY created_at DESC
LIMIT :scan_limit
@@ -116,6 +133,7 @@ def _build_backfill_proposal(incident: dict[str, Any]) -> dict[str, Any]:
"risk_level": str(incident.get("severity") or ""),
"action": "enqueue_allowlisted_ansible_check_mode",
"alertname": incident.get("alertname"),
"approval_id": str(incident.get("approval_id") or ""),
}
@@ -731,10 +749,10 @@ async def enqueue_missing_ansible_candidates_once(
recorder: Recorder = record_ansible_decision_audit,
receipt_backfiller: Callable[
..., Awaitable[dict[str, Any]]
] = backfill_missing_auto_repair_execution_receipts_once,
] | None = None,
retry_replayer: Callable[
..., Awaitable[dict[str, Any]]
] = run_failed_apply_check_mode_replay_once,
] = preflight_failed_apply_retry_queue_once,
evidence_collector: EvidenceCollector = (
collect_ansible_candidate_pre_decision_evidence
),
@@ -827,14 +845,17 @@ async def enqueue_missing_ansible_candidates_once(
logger.info("awooop_ansible_failed_apply_retry_priority_tick", **stats)
return stats
receipt_stats = await receipt_backfiller(
project_id=project_id,
window_hours=bounded_window_hours,
limit=bounded_limit,
)
stats["repair_receipts_backfilled"] = int(receipt_stats.get("written") or 0)
if receipt_stats.get("error") and not stats["error"]:
stats["error"] = receipt_stats["error"]
if receipt_backfiller is not None:
receipt_stats = await receipt_backfiller(
project_id=project_id,
window_hours=bounded_window_hours,
limit=bounded_limit,
)
stats["repair_receipts_backfilled"] = int(
receipt_stats.get("written") or 0
)
if receipt_stats.get("error") and not stats["error"]:
stats["error"] = receipt_stats["error"]
incidents = await _fetch_missing_candidate_incidents(
project_id=project_id,

713
apps/api/src/services/awooop_ansible_check_mode_service.py
View File

@@ -61,6 +61,21 @@ _EXECUTION_CAPABILITY_OPERATION_TYPES = frozenset(
"ansible_executor_capability_expired",
}
)
_VERIFIED_CLOSURE_REQUIRED_STAGE_IDS = frozenset(
{
"mcp_context",
"service_log_evidence",
"normalized_asset_identity",
"source_truth_diff",
"risk_policy_decision",
"executor_log_projection",
"retry_or_rollback",
"km_playbook_writeback",
"rag_writeback",
"playbook_trust",
"timeline_projection",
}
)
FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap"
REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519")
REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts")
@@ -747,6 +762,38 @@ async def _append_alert_lifecycle_receipt(
get_alert_operation_log_repository,
)
automation_run_id = _automation_run_id_for_claim(claim)
try:
async with get_db_context(project_id) as db:
existing = await db.execute(
text("""
SELECT EXISTS (
SELECT 1
FROM alert_operation_log
WHERE incident_id = :incident_id
AND event_type::text = :event_type
AND context ->> 'apply_op_id' = :apply_op_id
AND context ->> 'automation_run_id' = :automation_run_id
)
"""),
{
"incident_id": claim.incident_id,
"event_type": event_type,
"apply_op_id": apply_op_id,
"automation_run_id": automation_run_id,
},
)
if existing.scalar() is True:
return True
except Exception as exc:
logger.warning(
"ansible_alert_lifecycle_receipt_readback_failed",
incident_id=claim.incident_id,
apply_op_id=apply_op_id,
event_type=event_type,
error_type=type(exc).__name__,
)
record = await get_alert_operation_log_repository().append(
event_type,
incident_id=claim.incident_id,
@@ -756,7 +803,7 @@ async def _append_alert_lifecycle_receipt(
success=success,
error_message=error_message,
context={
"automation_run_id": _automation_run_id_for_claim(claim),
"automation_run_id": automation_run_id,
"catalog_id": claim.catalog_id,
"check_mode_op_id": claim.op_id,
"apply_op_id": apply_op_id,
@@ -1988,6 +2035,393 @@ async def _append_runtime_stage_receipts_to_apply(
return False
def _runtime_stage_ids(value: Any) -> set[str]:
payload = _json_loads(value)
receipts = payload.get("runtime_stage_receipts")
if not isinstance(receipts, list):
return set()
return {
str(receipt.get("stage_id") or "")
for receipt in receipts
if isinstance(receipt, Mapping) and receipt.get("durable_receipt") is True
}
async def _read_verified_apply_closure_prerequisites(
claim: AnsibleCheckModeClaim,
*,
apply_op_id: str,
project_id: str,
) -> dict[str, Any]:
"""Read back every durable prerequisite before resolving an incident."""
automation_run_id = _automation_run_id_for_claim(claim)
approval_id = _approval_id_for_claim(claim)
canonical_playbook_id = canonical_ansible_playbook_id(claim.catalog_id)
async with get_db_context(project_id) as db:
result = await db.execute(
text("""
SELECT
apply.status = 'success' AS apply_success,
apply.input AS apply_input,
coalesce(
apply.output ->> 'independent_post_verifier_passed',
'false'
) = 'true' AS apply_verifier_terminal,
check_mode.status = 'success' AS check_mode_success,
candidate.op_id IS NOT NULL AS candidate_present,
EXISTS (
SELECT 1
FROM auto_repair_executions receipt
WHERE receipt.incident_id = :incident_id
AND receipt.triggered_by = 'ansible_controlled_apply'
AND receipt.success IS TRUE
AND receipt.executed_steps::text LIKE
'%' || :apply_op_id || '%'
) AS auto_repair_receipt,
EXISTS (
SELECT 1
FROM incident_evidence evidence
WHERE evidence.incident_id = :incident_id
AND evidence.post_execution_state ->> 'apply_op_id'
= :apply_op_id
AND evidence.verification_result = 'success'
) AS post_verifier_receipt,
EXISTS (
SELECT 1
FROM knowledge_entries km
WHERE km.related_incident_id = :incident_id
AND km.path_type = :km_path_type
AND km.related_playbook_id = :canonical_playbook_id
AND km.updated_at IS NOT NULL
) AS km_writeback_receipt,
EXISTS (
SELECT 1
FROM automation_operation_log learning
WHERE learning.operation_type
= 'ansible_learning_writeback_recorded'
AND learning.parent_op_id = apply.op_id
AND learning.status = 'success'
AND learning.output ->> 'learning_recorded' = 'true'
AND learning.output ->> 'trust_updated' = 'true'
) AS playbook_trust_receipt,
EXISTS (
SELECT 1
FROM playbooks playbook
WHERE playbook.playbook_id = :canonical_playbook_id
AND playbook.success_count > 0
AND playbook.last_used_at IS NOT NULL
) AS playbook_readback,
EXISTS (
SELECT 1
FROM timeline_events timeline
WHERE timeline.incident_id = :incident_id
AND timeline.actor = 'ansible_controlled_apply_worker'
AND timeline.status = 'success'
AND timeline.description LIKE
'%apply_op_id=' || :apply_op_id || ';%'
) AS timeline_readback,
EXISTS (
SELECT 1
FROM awooop_outbound_message outbound
WHERE outbound.project_id = :project_id
AND outbound.channel_type = 'telegram'
AND outbound.send_status = 'sent'
AND outbound.provider_message_id IS NOT NULL
AND outbound.source_envelope
#>> '{callback_reply,action}'
= 'controlled_apply_result'
AND coalesce(
outbound.source_envelope
->> 'automation_run_id',
outbound.source_envelope
#>> '{callback_reply,automation_run_id}'
) = :automation_run_id
AND coalesce(
outbound.source_envelope
#>> '{callback_reply,incident_id}',
outbound.source_envelope
#>> '{source_refs,incident_ids,0}'
) = :incident_id
) AS telegram_receipt,
(
:approval_id = ''
OR EXISTS (
SELECT 1
FROM approval_records approval
WHERE approval.id = :approval_id
AND lower(approval.status::text)
= 'execution_success'
AND approval.extra_metadata ->> 'apply_op_id'
= :apply_op_id
AND approval.extra_metadata
->> 'post_verifier_passed' = 'true'
)
) AS approval_projection,
EXISTS (
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = :incident_id
AND lifecycle.event_type::text
= 'EXECUTION_COMPLETED'
AND lifecycle.success IS TRUE
AND lifecycle.context ->> 'apply_op_id'
= :apply_op_id
AND lifecycle.context ->> 'automation_run_id'
= :automation_run_id
) AS execution_lifecycle,
EXISTS (
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = :incident_id
AND lifecycle.event_type::text
= 'TELEGRAM_RESULT_SENT'
AND lifecycle.success IS TRUE
AND lifecycle.context ->> 'apply_op_id'
= :apply_op_id
AND lifecycle.context ->> 'automation_run_id'
= :automation_run_id
) AS telegram_lifecycle
FROM automation_operation_log apply
LEFT JOIN automation_operation_log check_mode
ON check_mode.op_id = apply.parent_op_id
AND check_mode.operation_type = 'ansible_check_mode_executed'
LEFT JOIN automation_operation_log candidate
ON candidate.op_id = check_mode.parent_op_id
AND candidate.operation_type = 'ansible_candidate_matched'
WHERE apply.op_id = CAST(:apply_op_id AS uuid)
AND apply.operation_type = 'ansible_apply_executed'
LIMIT 1
"""),
{
"apply_op_id": apply_op_id,
"incident_id": claim.incident_id,
"project_id": project_id,
"automation_run_id": automation_run_id,
"approval_id": approval_id,
"canonical_playbook_id": canonical_playbook_id,
"km_path_type": f"ansible_apply_receipt:{apply_op_id[:8]}",
},
)
row = result.mappings().one_or_none()
if row is None:
return {
"ready": False,
"missing": ["apply_operation_readback"],
"receipts": {},
"stage_ids": [],
}
stage_ids = _runtime_stage_ids(row.get("apply_input"))
receipts = {
"candidate": row.get("candidate_present") is True,
"check_mode": row.get("check_mode_success") is True,
"controlled_apply": row.get("apply_success") is True,
"post_apply_verifier": (
row.get("apply_verifier_terminal") is True
and row.get("post_verifier_receipt") is True
),
"auto_repair_execution_receipt": row.get("auto_repair_receipt") is True,
"km_playbook_writeback": row.get("km_writeback_receipt") is True,
"rag_writeback": "rag_writeback" in stage_ids,
"playbook_trust": (
row.get("playbook_trust_receipt") is True
and row.get("playbook_readback") is True
),
"timeline_projection": (
"timeline_projection" in stage_ids
and row.get("timeline_readback") is True
),
"telegram_receipt": row.get("telegram_receipt") is True,
"approval_projection": row.get("approval_projection") is True,
"execution_lifecycle": row.get("execution_lifecycle") is True,
"telegram_lifecycle": row.get("telegram_lifecycle") is True,
"required_runtime_stage_receipts": (
_VERIFIED_CLOSURE_REQUIRED_STAGE_IDS.issubset(stage_ids)
),
}
missing = sorted(name for name, ready in receipts.items() if not ready)
return {
"ready": not missing,
"missing": missing,
"receipts": receipts,
"stage_ids": sorted(stage_ids),
"required_stage_ids": sorted(_VERIFIED_CLOSURE_REQUIRED_STAGE_IDS),
"automation_run_id": automation_run_id,
"canonical_playbook_id": canonical_playbook_id,
}
async def _read_incident_closure_readback(
claim: AnsibleCheckModeClaim,
*,
apply_op_id: str,
project_id: str,
) -> dict[str, Any]:
automation_run_id = _automation_run_id_for_claim(claim)
async with get_db_context(project_id) as db:
result = await db.execute(
text("""
SELECT
incident.status::text AS incident_status,
incident.resolved_at,
incident.outcome,
EXISTS (
SELECT 1
FROM jsonb_array_elements(
coalesce(
apply.input -> 'runtime_stage_receipts',
'[]'::jsonb
)
) receipt(value)
WHERE receipt.value ->> 'stage_id' = 'incident_closure'
AND receipt.value ->> 'automation_run_id'
= :automation_run_id
AND receipt.value ->> 'durable_receipt' = 'true'
) AS closure_receipt,
EXISTS (
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = :incident_id
AND lifecycle.event_type::text = 'RESOLVED'
AND lifecycle.success IS TRUE
AND lifecycle.context ->> 'apply_op_id'
= :apply_op_id
AND lifecycle.context ->> 'automation_run_id'
= :automation_run_id
) AS resolved_lifecycle
FROM incidents incident
JOIN automation_operation_log apply
ON apply.op_id = CAST(:apply_op_id AS uuid)
WHERE incident.incident_id = :incident_id
AND incident.project_id = :project_id
LIMIT 1
"""),
{
"incident_id": claim.incident_id,
"project_id": project_id,
"apply_op_id": apply_op_id,
"automation_run_id": automation_run_id,
},
)
row = result.mappings().one_or_none()
if row is None:
return {"closed": False, "missing": ["incident_readback"]}
outcome = _json_loads(row.get("outcome"))
terminal = outcome.get("automation_terminal")
terminal_ready = bool(
isinstance(terminal, Mapping)
and terminal.get("automation_run_id") == automation_run_id
and terminal.get("apply_op_id") == apply_op_id
and terminal.get("incident_resolved") is True
)
checks = {
"incident_status": str(row.get("incident_status") or "").upper()
in {"RESOLVED", "CLOSED"},
"incident_resolved_at": row.get("resolved_at") is not None,
"incident_terminal_readback": terminal_ready,
"incident_closure_receipt": row.get("closure_receipt") is True,
"resolved_lifecycle": row.get("resolved_lifecycle") is True,
}
missing = sorted(name for name, ready in checks.items() if not ready)
return {
"closed": not missing,
"missing": missing,
"checks": checks,
"incident_status": str(row.get("incident_status") or ""),
"automation_run_id": automation_run_id,
"apply_op_id": apply_op_id,
}
async def _finalize_verified_apply_closure(
claim: AnsibleCheckModeClaim,
*,
apply_op_id: str,
project_id: str,
) -> dict[str, Any]:
prerequisites = await _read_verified_apply_closure_prerequisites(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
)
if prerequisites.get("ready") is not True:
return {
"status": "closure_receipts_pending",
"closed": False,
"missing": list(prerequisites.get("missing") or []),
}
terminal = await _record_incident_terminal_disposition(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
terminal_type="controlled_apply_verified_closed",
success_terminal=True,
telegram_receipt_acknowledged=True,
)
if terminal is None:
return {
"status": "incident_terminal_write_pending",
"closed": False,
"missing": ["incident_terminal_readback"],
}
closure_receipt = _runtime_stage_receipt(
claim,
stage_id="incident_closure",
evidence_ref=f"incidents:{claim.incident_id}:automation_terminal",
detail={
**terminal,
"closure_prerequisite_count": len(prerequisites["receipts"]),
"closure_prerequisites_verified": True,
},
)
receipt_written = await _append_runtime_stage_receipts_to_apply(
apply_op_id=apply_op_id,
receipts=(closure_receipt,),
project_id=project_id,
)
resolved_lifecycle = await _append_alert_lifecycle_receipt(
claim,
"RESOLVED",
apply_op_id=apply_op_id,
success=True,
action_detail="controlled_apply_verified_closed",
project_id=project_id,
post_verifier_passed=True,
)
if not receipt_written or not resolved_lifecycle:
return {
"status": "closure_projection_pending",
"closed": False,
"missing": [
name
for name, ready in {
"incident_closure_receipt": receipt_written,
"resolved_lifecycle": resolved_lifecycle,
}.items()
if not ready
],
}
readback = await _read_incident_closure_readback(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
)
return {
"status": (
"controlled_apply_closed"
if readback.get("closed") is True
else "closure_readback_pending"
),
**readback,
}
async def _record_incident_terminal_disposition(
claim: AnsibleCheckModeClaim,
*,
@@ -2772,6 +3206,97 @@ async def _send_controlled_apply_telegram_receipt(
return False
async def _reconcile_verified_apply_closure_projections(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
*,
apply_op_id: str,
writeback: dict[str, Any],
project_id: str,
) -> dict[str, Any]:
"""Repair terminal projections without ever executing the apply again."""
if not (
result.returncode == 0
and result.post_verifier_passed is True
and writeback.get("verification_passed") is True
):
return {
"status": "not_verified_success",
"closed": False,
"runtime_apply_executed": False,
}
approval_projection = await _finalize_controlled_approval_projection(
claim,
result,
apply_op_id=apply_op_id,
project_id=project_id,
)
execution_lifecycle = await _append_alert_lifecycle_receipt(
claim,
"EXECUTION_COMPLETED",
apply_op_id=apply_op_id,
success=True,
action_detail=f"controlled_apply_completed:{claim.catalog_id}",
project_id=project_id,
post_verifier_passed=True,
)
readback = await _read_verified_apply_closure_prerequisites(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
)
telegram_receipt = bool(
(readback.get("receipts") or {}).get("telegram_receipt")
)
if not telegram_receipt:
await _send_controlled_apply_telegram_receipt(
claim,
result,
apply_op_id=apply_op_id,
writeback=writeback,
project_id=project_id,
)
readback = await _read_verified_apply_closure_prerequisites(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
)
telegram_receipt = bool(
(readback.get("receipts") or {}).get("telegram_receipt")
)
telegram_lifecycle = False
if telegram_receipt:
telegram_lifecycle = await _append_alert_lifecycle_receipt(
claim,
"TELEGRAM_RESULT_SENT",
apply_op_id=apply_op_id,
success=True,
action_detail="controlled_apply_result_receipt_sent",
project_id=project_id,
post_verifier_passed=True,
)
closure = await _finalize_verified_apply_closure(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
)
return {
**closure,
"approval_projection_written": (
approval_projection or not _approval_id_for_claim(claim)
),
"execution_lifecycle_written": execution_lifecycle,
"telegram_receipt_acknowledged": telegram_receipt,
"telegram_lifecycle_written": telegram_lifecycle,
"runtime_apply_executed": False,
}
async def backfill_missing_auto_repair_execution_receipts_once(
*,
project_id: str = "awoooi",
@@ -2790,6 +3315,8 @@ async def backfill_missing_auto_repair_execution_receipts_once(
"playbook_trust_written": 0,
"timeline_projection_written": 0,
"runtime_stage_receipts_written": 0,
"incident_closure_written": 0,
"telegram_receipt_acknowledged": 0,
"skipped": 0,
"error": None,
}
@@ -2893,11 +3420,58 @@ async def backfill_missing_auto_repair_execution_receipts_once(
)::timestamptz <= NOW() - INTERVAL '5 minutes'
)
)
OR (
apply.status = 'success'
AND (
NOT EXISTS (
SELECT 1
FROM jsonb_array_elements(
coalesce(
apply.input
-> 'runtime_stage_receipts',
'[]'::jsonb
)
) AS receipt(value)
WHERE receipt.value ->> 'stage_id'
= 'incident_closure'
)
OR EXISTS (
SELECT 1
FROM incidents incident
WHERE incident.incident_id = coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
)
AND incident.project_id = :project_id
AND (
incident.resolved_at IS NULL
OR upper(incident.status::text)
NOT IN ('RESOLVED', 'CLOSED')
)
)
OR NOT EXISTS (
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
)
AND lifecycle.event_type::text
= 'RESOLVED'
AND lifecycle.context ->> 'apply_op_id'
= apply.op_id::text
)
)
)
)
ORDER BY apply.created_at DESC
LIMIT :limit
"""),
{"window_hours": max(1, window_hours), "limit": max(1, limit)},
{
"project_id": project_id,
"window_hours": max(1, window_hours),
"limit": max(1, limit),
},
)
rows = [dict(row) for row in result.mappings().all()]
stats["scanned"] = len(rows)
@@ -2964,6 +3538,17 @@ async def backfill_missing_auto_repair_execution_receipts_once(
),
):
stats["runtime_stage_receipts_written"] += 1
closure = await _reconcile_verified_apply_closure_projections(
claim,
verified_result,
apply_op_id=str(row.get("op_id") or ""),
writeback=writeback,
project_id=project_id,
)
if closure.get("telegram_receipt_acknowledged") is True:
stats["telegram_receipt_acknowledged"] += 1
if closure.get("closed") is True:
stats["incident_closure_written"] += 1
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning("ansible_auto_repair_execution_receipt_backfill_failed", **stats)
@@ -3029,6 +3614,46 @@ async def _load_open_failed_apply_retry_row(
return dict(row) if row is not None else None
async def preflight_failed_apply_retry_queue_once(
*,
project_id: str = "awoooi",
window_hours: int = 24,
**_kwargs: Any,
) -> dict[str, Any]:
"""Query retry backlog without using executor transport or writing state."""
try:
row = await _load_open_failed_apply_retry_row(
project_id=project_id,
window_hours=window_hours,
)
return {
"scanned": 1 if row is not None else 0,
"replayed": 0,
"check_mode_passed": 0,
"check_mode_failed": 0,
"runtime_stage_receipt_written": 0,
"blockers": [],
"error": None,
"query_only": True,
"runtime_apply_executed": False,
"execution_owner": "awoooi-ansible-executor-broker",
}
except Exception as exc:
return {
"scanned": 0,
"replayed": 0,
"check_mode_passed": 0,
"check_mode_failed": 0,
"runtime_stage_receipt_written": 0,
"blockers": [],
"error": type(exc).__name__,
"query_only": True,
"runtime_apply_executed": False,
"execution_owner": "awoooi-ansible-executor-broker",
}
async def run_failed_apply_check_mode_replay_once(
*,
project_id: str = "awoooi",
@@ -4138,6 +4763,28 @@ async def run_controlled_apply_for_claim(
project_id=project_id,
post_verifier_passed=verified_result.post_verifier_passed,
)
try:
closure = await _reconcile_verified_apply_closure_projections(
claim,
verified_result,
apply_op_id=apply_op_id,
writeback=writeback,
project_id=project_id,
)
except Exception as exc:
closure = {
"status": "closure_reconcile_retry_required",
"closed": False,
"missing": [f"closure_reconcile:{type(exc).__name__}"],
}
logger.warning(
"ansible_verified_apply_closure_reconcile_failed",
automation_run_id=_automation_run_id_for_claim(claim),
incident_id=claim.incident_id,
apply_op_id=apply_op_id,
error_type=type(exc).__name__,
runtime_apply_replay_required=False,
)
logger.info(
"ansible_controlled_apply_completed",
@@ -4162,6 +4809,9 @@ async def run_controlled_apply_for_claim(
alert_lifecycle_completed=lifecycle_completed,
telegram_receipt_sent=telegram_receipt_sent,
alert_lifecycle_telegram_written=lifecycle_telegram_written,
incident_closure_status=closure.get("status"),
incident_closure_readback=closure.get("closed") is True,
incident_closure_missing=closure.get("missing") or [],
)
return verified_result
@@ -4178,6 +4828,61 @@ async def run_pending_check_modes_once(
effective_timeout_seconds = (
timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS
)
receipt_stats = await backfill_missing_auto_repair_execution_receipts_once(
project_id=project_id,
window_hours=24,
limit=max(1, min(int(limit), 5)),
)
receipt_backfill_summary = {
"repair_receipt_backfill_scanned": int(
receipt_stats.get("scanned") or 0
),
"repair_receipt_backfill_written": int(
receipt_stats.get("written") or 0
),
"repair_receipt_closure_written": int(
receipt_stats.get("incident_closure_written") or 0
),
"repair_receipt_telegram_acknowledged": int(
receipt_stats.get("telegram_receipt_acknowledged") or 0
),
"repair_receipt_backfill_error": (
str(receipt_stats.get("error") or "")[:500] or None
),
}
receipt_backfill_priority_tick = bool(
receipt_backfill_summary["repair_receipt_backfill_scanned"]
or receipt_backfill_summary["repair_receipt_backfill_error"]
)
receipt_backfill_summary["repair_receipt_backfill_priority_tick"] = (
receipt_backfill_priority_tick
)
if receipt_backfill_priority_tick:
logger.info(
"ansible_execution_broker_receipt_backfill_priority_tick",
project_id=project_id,
**receipt_backfill_summary,
)
return {
"claimed": 0,
"reclaimed": 0,
"catalog_replayed": 0,
"completed": 0,
"failed": 0,
"apply_completed": 0,
"apply_failed": 0,
"apply_blocked": 0,
"capability_issued": 0,
"capability_revoked": 0,
"capability_expired": expired_capability_count,
"capability_revoke_failed": 0,
"catalog_replay_error": None,
"error": receipt_backfill_summary[
"repair_receipt_backfill_error"
],
"blockers": [],
**receipt_backfill_summary,
}
retry_stats = await run_failed_apply_check_mode_replay_once(
project_id=project_id,
window_hours=24,
@@ -4236,6 +4941,7 @@ async def run_pending_check_modes_once(
"catalog_replay_error": None,
"error": failed_apply_retry_error,
"blockers": failed_apply_retry_blockers,
**receipt_backfill_summary,
**failed_apply_retry_summary,
}
blockers = _runtime_blockers()
@@ -4247,6 +4953,7 @@ async def run_pending_check_modes_once(
"failed": 0,
"capability_expired": expired_capability_count,
"blockers": blockers,
**receipt_backfill_summary,
**failed_apply_retry_summary,
}
transport_blockers = await recent_ansible_transport_blockers(project_id=project_id)
@@ -4258,6 +4965,7 @@ async def run_pending_check_modes_once(
"failed": 0,
"capability_expired": expired_capability_count,
"blockers": transport_blockers,
**receipt_backfill_summary,
**failed_apply_retry_summary,
}
reclaimed_claims = await claim_stale_pending_check_modes(
@@ -4421,6 +5129,7 @@ async def run_pending_check_modes_once(
"capability_expired": expired_capability_count,
"capability_revoke_failed": capability_revoke_failed,
"catalog_replay_error": catalog_replay_error,
**receipt_backfill_summary,
**failed_apply_retry_summary,
"blockers": [],
}

292
apps/api/tests/test_ansible_verified_closure.py Normal file
View File

@@ -0,0 +1,292 @@
from __future__ import annotations
from unittest.mock import AsyncMock
import pytest
from src.jobs import awooop_ansible_candidate_backfill_job as candidate_job
from src.services import awooop_ansible_check_mode_service as service
def _claim() -> service.AnsibleCheckModeClaim:
return service.AnsibleCheckModeClaim(
op_id="00000000-0000-0000-0000-000000000101",
source_candidate_op_id="00000000-0000-0000-0000-000000000100",
incident_id="INC-20260711-D037E5",
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="medium",
input_payload={
"automation_run_id": "00000000-0000-0000-0000-000000000102",
"approval_id": "00000000-0000-0000-0000-000000000103",
},
)
def _verified_result() -> service.AnsibleRunResult:
return service.AnsibleRunResult(
returncode=0,
stdout="",
stderr="",
duration_ms=25,
post_verifier_passed=True,
)
def test_runtime_stage_ids_only_accepts_durable_receipts() -> None:
stage_ids = service._runtime_stage_ids(
{
"runtime_stage_receipts": [
{
"stage_id": "mcp_context",
"durable_receipt": True,
},
{
"stage_id": "telegram_receipt",
"durable_receipt": False,
},
]
}
)
assert stage_ids == {"mcp_context"}
@pytest.mark.asyncio
async def test_closure_refuses_to_resolve_when_any_receipt_is_missing(
monkeypatch: pytest.MonkeyPatch,
) -> None:
terminal_writer = AsyncMock()
monkeypatch.setattr(
service,
"_read_verified_apply_closure_prerequisites",
AsyncMock(
return_value={
"ready": False,
"missing": ["telegram_receipt"],
"receipts": {},
}
),
)
monkeypatch.setattr(
service,
"_record_incident_terminal_disposition",
terminal_writer,
)
result = await service._finalize_verified_apply_closure(
_claim(),
apply_op_id="00000000-0000-0000-0000-000000000104",
project_id="awoooi",
)
assert result == {
"status": "closure_receipts_pending",
"closed": False,
"missing": ["telegram_receipt"],
}
terminal_writer.assert_not_awaited()
@pytest.mark.asyncio
async def test_closure_resolves_only_after_durable_readback(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
service,
"_read_verified_apply_closure_prerequisites",
AsyncMock(return_value={"ready": True, "receipts": {"all": True}}),
)
monkeypatch.setattr(
service,
"_record_incident_terminal_disposition",
AsyncMock(
return_value={
"automation_run_id": (
"00000000-0000-0000-0000-000000000102"
),
"apply_op_id": "00000000-0000-0000-0000-000000000104",
"incident_resolved": True,
}
),
)
receipt_writer = AsyncMock(return_value=True)
lifecycle_writer = AsyncMock(return_value=True)
monkeypatch.setattr(
service,
"_append_runtime_stage_receipts_to_apply",
receipt_writer,
)
monkeypatch.setattr(
service,
"_append_alert_lifecycle_receipt",
lifecycle_writer,
)
monkeypatch.setattr(
service,
"_read_incident_closure_readback",
AsyncMock(
return_value={
"closed": True,
"missing": [],
"incident_status": "RESOLVED",
}
),
)
result = await service._finalize_verified_apply_closure(
_claim(),
apply_op_id="00000000-0000-0000-0000-000000000104",
project_id="awoooi",
)
assert result["status"] == "controlled_apply_closed"
assert result["closed"] is True
closure_receipt = receipt_writer.await_args.kwargs["receipts"][0]
assert closure_receipt["stage_id"] == "incident_closure"
assert closure_receipt["durable_receipt"] is True
assert lifecycle_writer.await_args.args[1] == "RESOLVED"
@pytest.mark.asyncio
async def test_projection_replay_sends_missing_receipt_without_reapplying(
monkeypatch: pytest.MonkeyPatch,
) -> None:
readback = AsyncMock(
side_effect=[
{"receipts": {"telegram_receipt": False}},
{"receipts": {"telegram_receipt": True}},
]
)
telegram_sender = AsyncMock(return_value={"ok": True})
monkeypatch.setattr(
service,
"_finalize_controlled_approval_projection",
AsyncMock(return_value=True),
)
monkeypatch.setattr(
service,
"_append_alert_lifecycle_receipt",
AsyncMock(return_value=True),
)
monkeypatch.setattr(
service,
"_read_verified_apply_closure_prerequisites",
readback,
)
monkeypatch.setattr(
service,
"_send_controlled_apply_telegram_receipt",
telegram_sender,
)
monkeypatch.setattr(
service,
"_finalize_verified_apply_closure",
AsyncMock(return_value={"status": "controlled_apply_closed", "closed": True}),
)
result = await service._reconcile_verified_apply_closure_projections(
_claim(),
_verified_result(),
apply_op_id="00000000-0000-0000-0000-000000000104",
writeback={
"verification_passed": True,
"verification_result": "success",
"verification": True,
"learning": True,
},
project_id="awoooi",
)
assert result["closed"] is True
assert result["runtime_apply_executed"] is False
telegram_sender.assert_awaited_once()
def test_backfill_preserves_approval_and_replaces_terminal_skipped_candidate() -> None:
proposal = candidate_job._build_backfill_proposal(
{
"alertname": "AwoooPAutoRepairCanaryT16",
"severity": "medium",
"approval_id": "00000000-0000-0000-0000-000000000103",
}
)
query_source = "\n".join(
str(value)
for value in candidate_job._fetch_missing_candidate_incidents.__code__.co_consts
)
assert proposal["approval_id"] == "00000000-0000-0000-0000-000000000103"
assert "approval_records" in query_source
assert "ansible_execution_skipped" in query_source
assert "terminal.parent_op_id = existing.op_id" in query_source
@pytest.mark.asyncio
async def test_broker_reconciles_receipts_before_claiming_fresh_work(
monkeypatch: pytest.MonkeyPatch,
) -> None:
retry_replayer = AsyncMock()
monkeypatch.setattr(
service,
"_expire_stale_ansible_execution_capabilities",
AsyncMock(return_value=0),
)
monkeypatch.setattr(
service,
"backfill_missing_auto_repair_execution_receipts_once",
AsyncMock(
return_value={
"scanned": 1,
"written": 1,
"incident_closure_written": 1,
"telegram_receipt_acknowledged": 1,
"error": None,
}
),
)
monkeypatch.setattr(
service,
"run_failed_apply_check_mode_replay_once",
retry_replayer,
)
result = await service.run_pending_check_modes_once(limit=1)
assert result["claimed"] == 0
assert result["repair_receipt_backfill_priority_tick"] is True
assert result["repair_receipt_closure_written"] == 1
retry_replayer.assert_not_awaited()
@pytest.mark.asyncio
async def test_signal_worker_retry_preflight_is_query_only(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
service,
"_load_open_failed_apply_retry_row",
AsyncMock(return_value={"op_id": "apply-op"}),
)
result = await service.preflight_failed_apply_retry_queue_once(
project_id="awoooi",
window_hours=24,
)
assert result["scanned"] == 1
assert result["replayed"] == 0
assert result["query_only"] is True
assert result["runtime_apply_executed"] is False
assert result["execution_owner"] == "awoooi-ansible-executor-broker"
def test_candidate_worker_defaults_to_query_only_retry_preflight() -> None:
defaults = candidate_job.enqueue_missing_ansible_candidates_once.__kwdefaults__
assert defaults is not None
assert defaults["retry_replayer"] is service.preflight_failed_apply_retry_queue_once

11
apps/api/tests/test_awooop_truth_chain_service.py
View File

@@ -5,6 +5,7 @@ import os
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
@@ -2071,6 +2072,11 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims
"_expire_stale_ansible_execution_capabilities",
no_expired_capabilities,
)
monkeypatch.setattr(
service,
"backfill_missing_auto_repair_execution_receipts_once",
AsyncMock(return_value={"scanned": 0, "written": 0, "error": None}),
)
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
monkeypatch.setattr(
service,
@@ -2133,6 +2139,11 @@ async def test_execution_broker_runs_failed_apply_retry_before_all_candidate_cla
"_expire_stale_ansible_execution_capabilities",
no_expired_capabilities,
)
monkeypatch.setattr(
service,
"backfill_missing_auto_repair_execution_receipts_once",
AsyncMock(return_value={"scanned": 0, "written": 0, "error": None}),
)
monkeypatch.setattr(service, "_runtime_blockers", lambda: [])
monkeypatch.setattr(
service,