From 673b0d57a41652575ae9cd2e2897b9d99b981fc5 Mon Sep 17 00:00:00 2001 From: ogt Date: Sat, 11 Jul 2026 16:18:32 +0800 Subject: [PATCH] fix(agent): bound stale retry terminalization --- .../awooop_ansible_check_mode_service.py | 45 ++++++++++--------- .../tests/test_ansible_verified_closure.py | 37 ++++++++++++++- 2 files changed, 58 insertions(+), 24 deletions(-) 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 88ed2eed7..d1a220b81 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -4799,17 +4799,28 @@ async def run_failed_apply_check_mode_replay_once( claim.input_payload.get("automation_run_id") or claim.source_candidate_op_id ) - - blockers = _runtime_blockers() - if blockers: - stats["blockers"] = blockers - return stats - transport_blockers = await recent_ansible_transport_blockers( - project_id=project_id + stale_replay_op_id = str(row.get("stale_replay_op_id") or "") + stale_recovery_count = int( + row.get("stale_replay_recovery_count") or 0 ) - if transport_blockers: - stats["blockers"] = transport_blockers - return stats + stale_retry_terminalized = bool( + stale_replay_op_id + and stale_recovery_count >= _CONTROLLED_RETRY_MAX_ATTEMPTS + ) + stats["stale_retry_reclaimed"] = int(bool(stale_replay_op_id)) + stats["stale_retry_terminalized"] = int(stale_retry_terminalized) + + if not stale_retry_terminalized: + 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 retry_claim_id = str(uuid4()) retry_claimed_at = datetime.now(UTC).isoformat() @@ -4834,9 +4845,6 @@ async def run_failed_apply_check_mode_replay_once( ), } async with get_db_context(project_id) as db: - stale_replay_op_id = str( - row.get("stale_replay_op_id") or "" - ) await db.execute( text(""" SELECT pg_advisory_xact_lock( @@ -4969,15 +4977,6 @@ async def run_failed_apply_check_mode_replay_once( stats["blockers"] = ["failed_apply_retry_claim_raced"] return stats replay_op_id = str(replay_op_id_value) - stale_recovery_count = int( - row.get("stale_replay_recovery_count") or 0 - ) - stale_retry_terminalized = bool( - stale_replay_op_id - and stale_recovery_count >= _CONTROLLED_RETRY_MAX_ATTEMPTS - ) - stats["stale_retry_reclaimed"] = int(bool(stale_replay_op_id)) - stats["stale_retry_terminalized"] = int(stale_retry_terminalized) if stale_retry_terminalized: replay_result = AnsibleRunResult( returncode=1, @@ -5014,6 +5013,7 @@ async def run_failed_apply_check_mode_replay_once( "approval_required_before_apply": False, "owner_review_required": False, "runtime_apply_executed": False, + "check_mode_executed": not stale_retry_terminalized, "check_mode_replay_performed": not stale_retry_terminalized, "stale_pending_retry_terminalized": stale_retry_terminalized, "bounded_execution_scope": "retry_check_mode_only", @@ -5033,6 +5033,7 @@ async def run_failed_apply_check_mode_replay_once( "approval_required_before_apply": False, "owner_review_required": False, "runtime_apply_executed": False, + "check_mode_executed": not stale_retry_terminalized, "check_mode_replay_performed": not stale_retry_terminalized, "stale_pending_retry_terminalized": stale_retry_terminalized, "bounded_execution_scope": "retry_check_mode_only", diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index 70f227c94..d476cad2c 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -1,6 +1,7 @@ from __future__ import annotations import inspect +import json from contextlib import asynccontextmanager from unittest.mock import AsyncMock @@ -759,11 +760,25 @@ async def test_stale_pending_retry_is_reclaimed_without_runtime_apply( ), ), ) - monkeypatch.setattr(service, "_runtime_blockers", lambda: []) + runtime_blockers = ( + ["ansible_binary_missing"] if expected_terminalized else [] + ) + monkeypatch.setattr( + service, + "_runtime_blockers", + lambda: runtime_blockers, + ) + transport_probe = AsyncMock( + return_value=( + ["ansible_transport_unavailable"] + if expected_terminalized + else [] + ) + ) monkeypatch.setattr( service, "recent_ansible_transport_blockers", - AsyncMock(return_value=[]), + transport_probe, ) monkeypatch.setattr( service, @@ -803,10 +818,28 @@ async def test_stale_pending_retry_is_reclaimed_without_runtime_apply( assert "UPDATE automation_operation_log replay" in db.statements[1] assert "reclaimed_after_stale_pending" in db.statements[1] assert "input ->> 'retry_claim_id'" in db.statements[2] + persisted_output = json.loads(db.parameters[2]["output"]) + persisted_dry_run = json.loads(db.parameters[2]["dry_run_result"]) + assert persisted_output["check_mode_executed"] is ( + not bool(expected_terminalized) + ) + assert persisted_output["check_mode_replay_performed"] is ( + not bool(expected_terminalized) + ) + assert persisted_output["runtime_apply_executed"] is False + assert persisted_dry_run["check_mode_executed"] is ( + not bool(expected_terminalized) + ) + assert persisted_dry_run["check_mode_replay_performed"] is ( + not bool(expected_terminalized) + ) + assert persisted_dry_run["runtime_apply_executed"] is False if expected_terminalized: replay_runner.assert_not_awaited() + transport_probe.assert_not_awaited() else: replay_runner.assert_awaited_once() + transport_probe.assert_awaited_once() receipt_writer.assert_awaited_once() assert receipt_writer.await_args.kwargs[ "check_mode_replay_performed"