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

feat(agent): execute failed apply check-mode replay

Browse Source
This commit is contained in:
ogt
2026-07-10 19:51:29 +08:00
parent 5487b128e3
commit bb66db6e57
4 changed files with 284 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,9 @@
"""AwoooP Ansible candidate backfill worker.
This worker closes the gap between "AI found an allowlisted PlayBook candidate"
and "the check-mode worker has a durable AOL row to claim". It does not execute
host writes by itself; it only writes ``ansible_candidate_matched`` rows for
recent unresolved incidents that already match the static Ansible catalog.
and "the check-mode worker has a durable AOL row to claim". It also runs a
bounded ``--check --diff`` replay for a failed controlled apply. It never retries
the runtime apply by itself.
"""
from __future__ import annotations
@@ -23,6 +23,7 @@ from src.services.awooop_ansible_audit_service import (
)
from src.services.awooop_ansible_check_mode_service import (
backfill_missing_auto_repair_execution_receipts_once,
run_failed_apply_check_mode_replay_once,
)
logger = structlog.get_logger(__name__)
@@ -101,7 +102,12 @@ async def enqueue_missing_ansible_candidates_once(
limit: int | None = None,
window_hours: int | None = None,
recorder: Recorder = record_ansible_decision_audit,
receipt_backfiller: Callable[..., Awaitable[dict[str, Any]]] = backfill_missing_auto_repair_execution_receipts_once,
receipt_backfiller: Callable[
..., Awaitable[dict[str, Any]]
] = backfill_missing_auto_repair_execution_receipts_once,
retry_replayer: Callable[
..., Awaitable[dict[str, Any]]
] = run_failed_apply_check_mode_replay_once,
) -> dict[str, Any]:
"""Backfill missing Ansible candidate rows for recent unresolved incidents."""
@@ -112,6 +118,10 @@ async def enqueue_missing_ansible_candidates_once(
"queued": 0,
"already_existing_or_write_skipped": 0,
"no_catalog_candidate": 0,
"repair_receipts_backfilled": 0,
"failed_apply_retry_replayed": 0,
"failed_apply_retry_check_mode_passed": 0,
"failed_apply_retry_check_mode_failed": 0,
"error": None,
}
@@ -128,6 +138,9 @@ async def enqueue_missing_ansible_candidates_once(
"already_existing_or_write_skipped": 0,
"no_catalog_candidate": 0,
"repair_receipts_backfilled": 0,
"failed_apply_retry_replayed": 0,
"failed_apply_retry_check_mode_passed": 0,
"failed_apply_retry_check_mode_failed": 0,
"error": None,
}
@@ -168,6 +181,21 @@ async def enqueue_missing_ansible_candidates_once(
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"]
retry_stats = await retry_replayer(
project_id=project_id,
window_hours=bounded_window_hours,
)
stats["failed_apply_retry_replayed"] = int(
retry_stats.get("replayed") or 0
)
stats["failed_apply_retry_check_mode_passed"] = int(
retry_stats.get("check_mode_passed") or 0
)
stats["failed_apply_retry_check_mode_failed"] = int(
retry_stats.get("check_mode_failed") or 0
)
if retry_stats.get("error") and not stats["error"]:
stats["error"] = retry_stats["error"]
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning("awooop_ansible_candidate_backfill_once_failed", **stats)

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

@@ -1175,6 +1175,231 @@ async def backfill_missing_auto_repair_execution_receipts_once(
return stats
async def run_failed_apply_check_mode_replay_once(
*,
project_id: str = "awoooi",
window_hours: int = 24,
timeout_seconds: int | None = None,
) -> dict[str, Any]:
"""Replay one failed apply in no-write check mode and persist its terminal receipt."""
stats: dict[str, Any] = {
"scanned": 0,
"replayed": 0,
"check_mode_passed": 0,
"check_mode_failed": 0,
"runtime_apply_executed": 0,
"blockers": [],
"error": None,
}
blockers = _runtime_blockers()
if blockers:
stats["blockers"] = blockers
return stats
transport_blockers = await recent_ansible_transport_blockers(project_id=project_id)
if transport_blockers:
stats["blockers"] = transport_blockers
return stats
try:
async with get_db_context(project_id) as db:
result = await db.execute(
text("""
SELECT
apply.op_id::text AS op_id,
apply.parent_op_id::text AS parent_op_id,
coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
) AS incident_id,
apply.input,
apply.output,
apply.dry_run_result,
apply.error,
apply.duration_ms,
apply.status,
apply.created_at
FROM automation_operation_log apply
WHERE apply.operation_type = 'ansible_apply_executed'
AND apply.status = 'failed'
AND apply.created_at >= NOW() - (
:window_hours * INTERVAL '1 hour'
)
AND EXISTS (
SELECT 1
FROM incident_evidence verifier
WHERE verifier.post_execution_state ->> 'apply_op_id'
= apply.op_id::text
AND verifier.verification_result = 'failed'
)
AND NOT EXISTS (
SELECT 1
FROM automation_operation_log replay
WHERE replay.operation_type = 'ansible_execution_skipped'
AND replay.parent_op_id = apply.op_id
AND replay.input ->> 'execution_mode'
= 'controlled_retry_check_mode_replay'
)
ORDER BY apply.created_at DESC
LIMIT 1
FOR UPDATE SKIP LOCKED
"""),
{"window_hours": max(1, window_hours)},
)
row = result.mappings().one_or_none()
if row is None:
return stats
stats["scanned"] = 1
reconstructed = _claim_from_apply_operation_row(dict(row))
if reconstructed is None:
stats["error"] = "failed_apply_receipt_could_not_be_reconstructed"
return stats
claim, _failed_apply_result = reconstructed
apply_op_id = str(row.get("op_id") or "")
automation_run_id = str(
claim.input_payload.get("automation_run_id")
or claim.source_candidate_op_id
)
input_payload = {
**claim.input_payload,
"automation_run_id": automation_run_id,
"execution_mode": "controlled_retry_check_mode_replay",
"retry_of_apply_op_id": apply_op_id,
"check_mode": True,
"diff": True,
"apply_enabled": False,
"approval_required_before_apply": True,
"controlled_apply_allowed": False,
"controlled_apply_blocker": (
"retry_requires_repair_and_new_apply_gate"
),
}
claimed = await db.execute(
text("""
INSERT INTO automation_operation_log (
operation_type, actor, status, incident_id,
input, output, dry_run_result,
parent_op_id, tags
) VALUES (
'ansible_execution_skipped',
'ansible_controlled_retry_worker',
'pending',
:incident_db_id,
CAST(:input AS jsonb),
'{}'::jsonb,
CAST(:dry_run_result AS jsonb),
CAST(:parent_op_id AS uuid),
:tags
)
RETURNING op_id
"""),
{
"incident_db_id": _automation_operation_log_incident_id(
claim.incident_id
),
"input": json.dumps(input_payload, ensure_ascii=False),
"dry_run_result": json.dumps(
{
"execution_mode": (
"controlled_retry_check_mode_replay"
),
"retry_of_apply_op_id": apply_op_id,
"check_mode_executed": False,
"runtime_apply_executed": False,
"claim_state": "claimed",
},
ensure_ascii=False,
),
"parent_op_id": apply_op_id,
"tags": [
"ansible",
"controlled_retry",
"check_mode_replay",
"no_runtime_apply",
f"automation_run_id:{automation_run_id}",
],
},
)
replay_op_id = str(claimed.scalar_one())
effective_timeout = (
timeout_seconds or settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS
)
try:
spec = build_ansible_check_mode_command(
playbook_path=claim.playbook_path,
inventory_hosts=claim.inventory_hosts,
)
replay_result = await _run_ansible_command(
spec,
timeout_seconds=effective_timeout,
)
except Exception as exc:
replay_result = AnsibleRunResult(
returncode=1,
stdout="",
stderr=f"ansible_retry_check_mode_runtime_error: {exc}",
duration_ms=0,
)
status, output, dry_run_result, error = _build_result_payload(
replay_result,
controlled_apply_allowed=False,
controlled_apply_blocker="retry_requires_repair_and_new_apply_gate",
)
output.update({
"execution_mode": "controlled_retry_check_mode_replay",
"retry_of_apply_op_id": apply_op_id,
"runtime_apply_executed": False,
"terminal_disposition": (
"no_write_replay_passed_waiting_repair_candidate"
if replay_result.returncode == 0
else "no_write_replay_failed_waiting_playbook_or_transport_repair"
),
})
dry_run_result.update({
"execution_mode": "controlled_retry_check_mode_replay",
"retry_of_apply_op_id": apply_op_id,
"runtime_apply_executed": False,
})
async with get_db_context(project_id) as db:
await db.execute(
text("""
UPDATE automation_operation_log
SET status = :status,
output = CAST(:output AS jsonb),
dry_run_result = CAST(:dry_run_result AS jsonb),
error = :error,
duration_ms = :duration_ms,
stderr_feed_back = :stderr
WHERE op_id = CAST(:op_id AS uuid)
"""),
{
"status": status,
"output": json.dumps(output, ensure_ascii=False),
"dry_run_result": json.dumps(dry_run_result, ensure_ascii=False),
"error": _tail(error or "", 2000) or None,
"duration_ms": replay_result.duration_ms,
"stderr": _tail(replay_result.stderr, _STDERR_LIMIT),
"op_id": replay_op_id,
},
)
stats["replayed"] = 1
stats[
"check_mode_passed" if replay_result.returncode == 0 else "check_mode_failed"
] = 1
logger.info(
"ansible_failed_apply_check_mode_replay_completed",
automation_run_id=automation_run_id,
apply_op_id=apply_op_id,
replay_op_id=str(replay_op_id),
returncode=replay_result.returncode,
runtime_apply_executed=False,
)
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning("ansible_failed_apply_check_mode_replay_failed", **stats)
return stats
async def claim_pending_check_modes(
*,
project_id: str = "awoooi",

13
apps/api/tests/test_awooop_ansible_candidate_backfill_job.py
View File

@@ -66,6 +66,15 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo
assert kwargs["project_id"] == "awoooi"
return {"written": 2, "error": None}
async def fake_retry_replayer(**kwargs):
assert kwargs["project_id"] == "awoooi"
return {
"replayed": 1,
"check_mode_passed": 1,
"check_mode_failed": 0,
"error": None,
}
monkeypatch.setattr(job, "get_db_context", fake_db_context)
monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True)
@@ -75,11 +84,15 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo
window_hours=24,
recorder=fake_recorder,
receipt_backfiller=fake_receipt_backfiller,
retry_replayer=fake_retry_replayer,
)
assert result["queued"] == 1
assert result["no_catalog_candidate"] == 0
assert result["repair_receipts_backfilled"] == 2
assert result["failed_apply_retry_replayed"] == 1
assert result["failed_apply_retry_check_mode_passed"] == 1
assert result["failed_apply_retry_check_mode_failed"] == 0
assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue"
assert recorded[0]["incident"]["incident_id"] == "INC-20260627-NODE110"

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

@@ -35,6 +35,7 @@ from src.services.awooop_ansible_check_mode_service import (
finalize_check_mode_claim,
recent_ansible_transport_blockers,
run_controlled_apply_for_claim,
run_failed_apply_check_mode_replay_once,
run_pending_check_modes_once,
)
from src.services.awooop_truth_chain_service import (
@@ -1778,6 +1779,19 @@ def test_ansible_live_controlled_apply_sends_telegram_receipt_but_backfill_does_
assert inspect.iscoroutinefunction(_send_controlled_apply_telegram_receipt)
def test_failed_apply_retry_replay_is_no_write_and_idempotent() -> None:
source = inspect.getsource(run_failed_apply_check_mode_replay_once)
assert "FOR UPDATE SKIP LOCKED" in source
assert "controlled_retry_check_mode_replay" in source
assert "build_ansible_check_mode_command" in source
assert "controlled_apply_allowed=False" in source
assert '"runtime_apply_executed": False' in source
assert "retry_requires_repair_and_new_apply_gate" in source
assert "NOT EXISTS" in source
assert "run_controlled_apply_for_claim" not in source
def test_ansible_check_and_apply_rows_persist_canonical_automation_run_id() -> None:
finalize_source = inspect.getsource(finalize_check_mode_claim)
apply_source = inspect.getsource(run_controlled_apply_for_claim)