From fc3b08afe30703836ebed1c0d3c1ebf3ba511f04 Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 10:43:00 +0800 Subject: [PATCH] fix(agent): replay stdin boundary failures without apply --- .../awooop_ansible_check_mode_service.py | 470 +++++++++++++++++- .../tests/test_awooop_truth_chain_service.py | 172 +++++++ 2 files changed, 641 insertions(+), 1 deletion(-) 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 52ac2e8de..789890289 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -50,6 +50,7 @@ _STDERR_LIMIT = 12_000 _CONTROLLED_RETRY_MAX_ATTEMPTS = 1 _CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS = 15 _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS = 30 +_STDIN_BOUNDARY_REPLAY_WINDOW_HOURS = 7 * 24 _EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300 _EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800 _EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS = 5 @@ -3315,6 +3316,249 @@ async def _record_retry_runtime_stage_receipt( ) +def _build_stdin_boundary_replay_stage_receipt( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + failed_check_mode_op_id: str, + terminal_op_id: str, +) -> dict[str, Any]: + terminal_type = ( + "no_write_stdin_boundary_replay_passed_waiting_repair_candidate" + if result.returncode == 0 + else "no_write_stdin_boundary_replay_failed_waiting_transport_repair" + ) + return _runtime_stage_receipt( + claim, + stage_id="retry_or_rollback", + evidence_ref=( + f"automation_operation_log:{terminal_op_id}:dry_run_result" + ), + detail={ + "schema_version": ( + "ansible_failed_check_stdin_boundary_retry_terminal_v1" + ), + "failed_check_mode_op_id": failed_check_mode_op_id, + "retry_check_mode_op_id": claim.op_id, + "retry_terminal_op_id": terminal_op_id, + "terminal_type": terminal_type, + "check_mode_replay_performed": True, + "check_mode_replay_returncode": result.returncode, + "runtime_apply_executed": False, + "rollback_performed": False, + "verified_no_write_terminal": True, + "retry_operation_readback_verified": True, + "retry_attempt": 1, + "max_retry_attempts": 1, + "retry_idempotency_key": ( + f"ansible-check-retry:{failed_check_mode_op_id}:stdin-boundary" + ), + "bounded_retry": True, + "repair_candidate_required": True, + "safe_next_action": ( + "queue_new_controlled_apply_candidate_after_no_write_replay" + if result.returncode == 0 + else "queue_ai_transport_repair_candidate" + ), + "repository_readback_verified": True, + "durable_write_acknowledged": True, + "writer_source_sha": os.getenv( + "AWOOOI_BUILD_COMMIT_SHA", + "", + ).strip().lower() + or None, + "raw_log_payload_stored": False, + "secret_value_stored": False, + }, + ) + + +async def _record_stdin_boundary_replay_terminal( + claim: AnsibleCheckModeClaim, + result: AnsibleRunResult, + *, + project_id: str, +) -> bool: + """Persist one idempotent, public-safe terminal for a no-write replay.""" + + failed_check_mode_op_id = str( + claim.input_payload.get("replay_of_check_mode_op_id") or "" + ) + if not failed_check_mode_op_id: + return False + automation_run_id = _automation_run_id_for_claim(claim) + expected_status = "success" if result.returncode == 0 else "failed" + terminal_type = ( + "no_write_stdin_boundary_replay_passed_waiting_repair_candidate" + if result.returncode == 0 + else "no_write_stdin_boundary_replay_failed_waiting_transport_repair" + ) + try: + async with get_db_context(project_id) as db: + await db.execute( + text(""" + SELECT pg_advisory_xact_lock( + hashtextextended(:retry_terminal_lock_key, 0) + ) + """), + { + "retry_terminal_lock_key": ( + "ansible-check-retry-terminal:" + f"{failed_check_mode_op_id}" + ) + }, + ) + selected = await db.execute( + text(""" + SELECT op_id::text AS op_id + FROM automation_operation_log + WHERE operation_type = 'ansible_execution_skipped' + AND input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay_terminal' + AND input ->> 'retry_of_check_mode_op_id' + = :failed_check_mode_op_id + ORDER BY created_at DESC + LIMIT 1 + """), + {"failed_check_mode_op_id": failed_check_mode_op_id}, + ) + terminal_op_id = str(selected.scalar_one_or_none() or uuid4()) + receipt = _build_stdin_boundary_replay_stage_receipt( + claim, + result, + failed_check_mode_op_id=failed_check_mode_op_id, + terminal_op_id=terminal_op_id, + ) + terminal_input = { + "automation_run_id": automation_run_id, + "incident_id": claim.incident_id, + "catalog_id": claim.catalog_id, + "execution_mode": ( + "stdin_boundary_check_mode_replay_terminal" + ), + "retry_of_check_mode_op_id": failed_check_mode_op_id, + "retry_check_mode_op_id": claim.op_id, + "runtime_apply_executed": False, + "runtime_stage_receipts": [receipt], + } + terminal_output = { + "terminal_disposition": terminal_type, + "check_mode_replay_returncode": result.returncode, + "check_mode_replay_performed": True, + "runtime_apply_executed": False, + "verified_no_write_terminal": True, + "raw_log_payload_stored": False, + "secret_value_stored": False, + } + await db.execute( + text(""" + INSERT INTO automation_operation_log ( + op_id, operation_type, actor, status, incident_id, + input, output, dry_run_result, error, duration_ms, + parent_op_id, tags + ) + SELECT + CAST(:op_id AS uuid), + 'ansible_execution_skipped', + 'ansible_stdin_boundary_replay_worker', + :status, + :incident_db_id, + CAST(:input AS jsonb), + CAST(:output AS jsonb), + CAST(:dry_run_result AS jsonb), + :error, + :duration_ms, + CAST(:parent_op_id AS uuid), + :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log existing + WHERE existing.operation_type + = 'ansible_execution_skipped' + AND existing.input ->> 'execution_mode' + = 'stdin_boundary_check_mode_replay_terminal' + AND existing.input ->> 'retry_of_check_mode_op_id' + = :failed_check_mode_op_id + ) + """), + { + "op_id": terminal_op_id, + "status": expected_status, + "incident_db_id": _automation_operation_log_incident_id( + claim.incident_id + ), + "input": json.dumps(terminal_input, ensure_ascii=False), + "output": json.dumps(terminal_output, ensure_ascii=False), + "dry_run_result": json.dumps( + { + **terminal_output, + "check_mode": True, + "diff": True, + }, + ensure_ascii=False, + ), + "error": ( + None + if result.returncode == 0 + else ( + "stdin_boundary_replay_check_mode_failed_rc_" + f"{result.returncode}" + ) + ), + "duration_ms": result.duration_ms, + "parent_op_id": claim.op_id, + "tags": [ + "ansible", + "stdin_boundary_replay", + "verified_no_write_terminal", + f"automation_run_id:{automation_run_id}", + ], + "failed_check_mode_op_id": failed_check_mode_op_id, + }, + ) + readback = await db.execute( + text(""" + SELECT status, input, output + FROM automation_operation_log + WHERE op_id = CAST(:op_id AS uuid) + AND operation_type = 'ansible_execution_skipped' + LIMIT 1 + """), + {"op_id": terminal_op_id}, + ) + row = readback.mappings().one_or_none() + input_readback = _json_loads(row.get("input") if row else None) + output_readback = _json_loads(row.get("output") if row else None) + receipts = input_readback.get("runtime_stage_receipts") + return bool( + row + and row.get("status") == expected_status + and input_readback.get("automation_run_id") + == automation_run_id + and input_readback.get("retry_of_check_mode_op_id") + == failed_check_mode_op_id + and output_readback.get("runtime_apply_executed") is False + and output_readback.get("terminal_disposition") + == terminal_type + and isinstance(receipts, list) + and any( + isinstance(item, Mapping) + and item.get("stage_id") == "retry_or_rollback" + and item.get("durable_receipt") is True + for item in receipts + ) + ) + except Exception as exc: + logger.warning( + "ansible_stdin_boundary_replay_terminal_write_failed", + automation_run_id=automation_run_id, + failed_check_mode_op_id=failed_check_mode_op_id, + retry_check_mode_op_id=claim.op_id, + error_type=type(exc).__name__, + ) + return False + + async def _record_learning_writeback_receipt( claim: AnsibleCheckModeClaim, result: AnsibleRunResult, @@ -5598,6 +5842,169 @@ async def claim_stale_pending_check_modes( return claims +async def claim_stdin_boundary_failed_check_modes( + *, + project_id: str = "awoooi", + limit: int = 1, + window_hours: int = _STDIN_BOUNDARY_REPLAY_WINDOW_HOURS, +) -> list[AnsibleCheckModeClaim]: + """Replay historical nonblocking-stdin failures once, without apply.""" + + claims: list[AnsibleCheckModeClaim] = [] + async with get_db_context(project_id) as db: + result = await db.execute( + text(""" + SELECT + check_mode.op_id, + check_mode.parent_op_id, + coalesce( + check_mode.input ->> 'incident_id', + check_mode.incident_id::text + ) AS incident_id, + check_mode.input + FROM automation_operation_log check_mode + WHERE check_mode.operation_type = 'ansible_check_mode_executed' + AND check_mode.status = 'failed' + AND check_mode.created_at >= NOW() - ( + :window_hours * INTERVAL '1 hour' + ) + AND ( + coalesce(check_mode.error, '') + LIKE '%Ansible requires blocking IO%' + OR coalesce(check_mode.error, '') + LIKE '%Non-blocking file handles detected%' + OR coalesce(check_mode.stderr_feed_back, '') + LIKE '%Ansible requires blocking IO%' + OR coalesce(check_mode.stderr_feed_back, '') + LIKE '%Non-blocking file handles detected%' + OR coalesce(check_mode.output::text, '') + LIKE '%Ansible requires blocking IO%' + OR coalesce(check_mode.output::text, '') + LIKE '%Non-blocking file handles detected%' + OR coalesce(check_mode.dry_run_result::text, '') + LIKE '%Ansible requires blocking IO%' + OR coalesce(check_mode.dry_run_result::text, '') + LIKE '%Non-blocking file handles detected%' + ) + AND NOT EXISTS ( + SELECT 1 + FROM automation_operation_log replay + WHERE replay.operation_type + = 'ansible_check_mode_executed' + AND replay.input ->> 'replay_of_check_mode_op_id' + = check_mode.op_id::text + ) + ORDER BY check_mode.created_at ASC + LIMIT :scan_limit + FOR UPDATE SKIP LOCKED + """), + { + "window_hours": max(1, min(window_hours, 7 * 24)), + "scan_limit": min(100, max(10, limit * 10)), + }, + ) + for row_value in result.mappings().all(): + row = dict(row_value) + original = _claim_from_stale_check_mode_row(row) + if original is None: + continue + failed_check_mode_op_id = str(row.get("op_id") or "") + replay_input = { + **original.input_payload, + "execution_mode": "stdin_boundary_check_mode_replay", + "historical_stdin_boundary_replay": True, + "replay_of_check_mode_op_id": failed_check_mode_op_id, + "check_mode": True, + "diff": True, + "apply_enabled": False, + "controlled_apply_allowed": False, + "controlled_apply_blocker": ( + "historical_stdin_boundary_replay_no_write" + ), + "approval_required_before_apply": False, + "owner_review_required": False, + "bounded_execution_scope": "check_mode_replay_only", + "safe_next_action": ( + "queue_new_controlled_apply_candidate_after_no_write_replay" + ), + } + inserted = await db.execute( + text(""" + INSERT INTO automation_operation_log ( + operation_type, actor, status, incident_id, + input, output, dry_run_result, + parent_op_id, tags + ) + SELECT + 'ansible_check_mode_executed', + 'ansible_stdin_boundary_replay_worker', + 'pending', + :incident_db_id, + CAST(:input AS jsonb), + '{}'::jsonb, + CAST(:dry_run_result AS jsonb), + CAST(:parent_op_id AS uuid), + :tags + WHERE NOT EXISTS ( + SELECT 1 + FROM automation_operation_log replay + WHERE replay.operation_type + = 'ansible_check_mode_executed' + AND replay.input ->> 'replay_of_check_mode_op_id' + = :failed_check_mode_op_id + ) + RETURNING op_id + """), + { + "incident_db_id": _automation_operation_log_incident_id( + original.incident_id + ), + "input": json.dumps(replay_input, ensure_ascii=False), + "dry_run_result": json.dumps( + { + "execution_mode": ( + "stdin_boundary_check_mode_replay" + ), + "replay_of_check_mode_op_id": ( + failed_check_mode_op_id + ), + "check_mode_executed": False, + "runtime_apply_executed": False, + "claim_state": "claimed", + }, + ensure_ascii=False, + ), + "parent_op_id": original.source_candidate_op_id, + "tags": [ + "ansible", + "check_mode", + "stdin_boundary_replay", + "no_runtime_apply", + ], + "failed_check_mode_op_id": failed_check_mode_op_id, + }, + ) + op_id_value = inserted.scalar_one_or_none() + if op_id_value is None: + continue + claims.append( + AnsibleCheckModeClaim( + op_id=str(op_id_value), + source_candidate_op_id=original.source_candidate_op_id, + incident_id=original.incident_id, + catalog_id=original.catalog_id, + playbook_path=original.playbook_path, + apply_playbook_path=original.apply_playbook_path, + inventory_hosts=original.inventory_hosts, + risk_level=original.risk_level, + input_payload=replay_input, + ) + ) + if len(claims) >= max(1, limit): + break + return claims + + async def claim_catalog_drift_failed_check_modes( *, project_id: str = "awoooi", @@ -6438,6 +6845,28 @@ async def run_pending_check_modes_once( stale_after_seconds=max(300, effective_timeout_seconds + 120), ) remaining_limit = max(0, limit - len(reclaimed_claims)) + stdin_boundary_replay_claims: list[AnsibleCheckModeClaim] = [] + stdin_boundary_replay_error: str | None = None + if remaining_limit: + try: + stdin_boundary_replay_claims = ( + await claim_stdin_boundary_failed_check_modes( + project_id=project_id, + limit=remaining_limit, + ) + ) + except Exception as exc: + stdin_boundary_replay_error = type(exc).__name__ + logger.warning( + "ansible_stdin_boundary_replay_claim_failed", + project_id=project_id, + error_type=stdin_boundary_replay_error, + fresh_candidate_claim_continues=True, + ) + remaining_limit = max( + 0, + remaining_limit - len(stdin_boundary_replay_claims), + ) catalog_replay_claims: list[AnsibleCheckModeClaim] = [] catalog_replay_error: str | None = None if remaining_limit: @@ -6469,7 +6898,12 @@ async def run_pending_check_modes_once( if remaining_limit else [] ) - claims = [*reclaimed_claims, *catalog_replay_claims, *fresh_claims] + claims = [ + *reclaimed_claims, + *stdin_boundary_replay_claims, + *catalog_replay_claims, + *fresh_claims, + ] completed = 0 failed = 0 apply_completed = 0 @@ -6479,6 +6913,7 @@ async def run_pending_check_modes_once( capability_issued = 0 capability_revoked = 0 capability_revoke_failed = 0 + stdin_boundary_replay_terminal_written = 0 for claim in claims: capability_op_id = "" terminal_status = "broker_interrupted_before_execution" @@ -6499,6 +6934,34 @@ async def run_pending_check_modes_once( project_id=project_id, ) completed += 1 + if claim.input_payload.get( + "historical_stdin_boundary_replay" + ) is True: + terminal_written = ( + await _record_stdin_boundary_replay_terminal( + claim, + result, + project_id=project_id, + ) + ) + stdin_boundary_replay_terminal_written += int( + terminal_written + ) + apply_blocked += 1 + if result.returncode != 0: + failed += 1 + terminal_status = ( + "stdin_boundary_replay_failed_no_write_terminal" + if terminal_written + else "stdin_boundary_replay_failed_terminal_missing" + ) + else: + terminal_status = ( + "stdin_boundary_replay_passed_no_write_terminal" + if terminal_written + else "stdin_boundary_replay_passed_terminal_missing" + ) + continue if result.returncode != 0: failed += 1 terminal_status = "check_mode_failed" @@ -6581,6 +7044,11 @@ async def run_pending_check_modes_once( return { "claimed": len(claims), "reclaimed": len(reclaimed_claims), + "stdin_boundary_replayed": len(stdin_boundary_replay_claims), + "stdin_boundary_replay_terminal_written": ( + stdin_boundary_replay_terminal_written + ), + "stdin_boundary_replay_error": stdin_boundary_replay_error, "catalog_replayed": len(catalog_replay_claims), "completed": completed, "failed": failed, diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 5b7c1b149..72a4b36f5 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -41,6 +41,7 @@ from src.services.awooop_ansible_check_mode_service import ( _record_learning_writeback_receipt, _record_retry_runtime_stage_receipt, _record_runtime_stage_receipts, + _record_stdin_boundary_replay_terminal, _record_timeline_projection_receipt, _resolve_execution_capability_operation_type, _revoke_ansible_execution_capability, @@ -57,6 +58,7 @@ from src.services.awooop_ansible_check_mode_service import ( claim_catalog_drift_failed_check_modes, claim_pending_check_modes, claim_stale_pending_check_modes, + claim_stdin_boundary_failed_check_modes, detect_ansible_transport_blockers, finalize_check_mode_claim, recent_ansible_transport_blockers, @@ -2063,6 +2065,30 @@ def test_failed_check_mode_catalog_drift_replays_once() -> None: assert "claim_catalog_drift_failed_check_modes" in run_source +def test_failed_stdin_boundary_checks_replay_once_without_apply() -> None: + claim_source = inspect.getsource( + claim_stdin_boundary_failed_check_modes + ) + terminal_source = inspect.getsource( + _record_stdin_boundary_replay_terminal + ) + run_source = inspect.getsource(run_pending_check_modes_once) + + assert "Ansible requires blocking IO" in claim_source + assert "Non-blocking file handles detected" in claim_source + assert "FOR UPDATE SKIP LOCKED" in claim_source + assert "replay_of_check_mode_op_id" in claim_source + assert "historical_stdin_boundary_replay_no_write" in claim_source + assert '"controlled_apply_allowed": False' in claim_source + assert '"apply_enabled": False' in claim_source + assert "stdin_boundary_check_mode_replay_terminal" in terminal_source + assert "verified_no_write_terminal" in terminal_source + assert "runtime_stage_receipts" in terminal_source + assert "runtime_apply_executed" in terminal_source + assert "_record_stdin_boundary_replay_terminal" in run_source + assert "stdin_boundary_replay_passed_no_write_terminal" in run_source + + @pytest.mark.asyncio async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims( monkeypatch: pytest.MonkeyPatch, @@ -2116,6 +2142,11 @@ async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims no_transport_blockers, ) monkeypatch.setattr(service, "claim_stale_pending_check_modes", no_stale_claims) + monkeypatch.setattr( + service, + "claim_stdin_boundary_failed_check_modes", + no_stale_claims, + ) monkeypatch.setattr( service, "run_failed_apply_check_mode_replay_once", @@ -2192,6 +2223,11 @@ async def test_execution_broker_runs_retry_then_keeps_fresh_capacity_moving( "claim_stale_pending_check_modes", candidate_claim_continues, ) + monkeypatch.setattr( + service, + "claim_stdin_boundary_failed_check_modes", + candidate_claim_continues, + ) monkeypatch.setattr( service, "claim_catalog_drift_failed_check_modes", @@ -2214,6 +2250,142 @@ async def test_execution_broker_runs_retry_then_keeps_fresh_capacity_moving( assert result["failed_apply_retry_priority_tick"] is True +@pytest.mark.asyncio +async def test_execution_broker_stdin_replay_never_enters_apply( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services import awooop_ansible_check_mode_service as service + + replay_claim = AnsibleCheckModeClaim( + op_id="00000000-0000-0000-0000-000000000302", + source_candidate_op_id=( + "00000000-0000-0000-0000-000000000301" + ), + incident_id="INC-STDIN-REPLAY", + catalog_id="ansible:188-momo-backup-user", + playbook_path=( + "infra/ansible/playbooks/188-momo-backup-user.yml" + ), + apply_playbook_path=( + "infra/ansible/playbooks/188-momo-backup-user.yml" + ), + inventory_hosts=("host_188",), + risk_level="low", + input_payload={ + "automation_run_id": ( + "00000000-0000-0000-0000-000000000301" + ), + "historical_stdin_boundary_replay": True, + "replay_of_check_mode_op_id": ( + "00000000-0000-0000-0000-000000000300" + ), + "controlled_apply_allowed": False, + }, + ) + + async def no_expired_capabilities(**_kwargs): + return 0 + + async def no_transport_blockers(**_kwargs): + return [] + + async def no_claims(**_kwargs): + return [] + + async def stdin_replay_claims(**_kwargs): + return [replay_claim] + + async def no_failed_apply_retry(**_kwargs): + return { + "scanned": 0, + "replayed": 0, + "check_mode_passed": 0, + "check_mode_failed": 0, + "runtime_stage_receipt_written": 0, + "blockers": [], + "error": None, + } + + async def issue_capability(claim, **_kwargs): + return claim, "00000000-0000-0000-0000-000000000303" + + async def revoke_capability(*_args, **_kwargs): + return True + + async def successful_check(*_args, **_kwargs): + return AnsibleRunResult( + returncode=0, + stdout="", + stderr="", + duration_ms=12, + ) + + monkeypatch.setattr( + service, + "_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, + "run_failed_apply_check_mode_replay_once", + no_failed_apply_retry, + ) + monkeypatch.setattr(service, "_runtime_blockers", lambda: []) + monkeypatch.setattr( + service, + "recent_ansible_transport_blockers", + no_transport_blockers, + ) + monkeypatch.setattr( + service, + "claim_stale_pending_check_modes", + no_claims, + ) + monkeypatch.setattr( + service, + "claim_stdin_boundary_failed_check_modes", + stdin_replay_claims, + ) + monkeypatch.setattr( + service, + "_issue_ansible_execution_capability", + issue_capability, + ) + monkeypatch.setattr( + service, + "_revoke_ansible_execution_capability", + revoke_capability, + ) + monkeypatch.setattr( + service, + "run_claimed_check_mode", + successful_check, + ) + terminal_write = AsyncMock(return_value=True) + monkeypatch.setattr( + service, + "_record_stdin_boundary_replay_terminal", + terminal_write, + ) + apply = AsyncMock() + monkeypatch.setattr(service, "run_controlled_apply_for_claim", apply) + + result = await service.run_pending_check_modes_once(limit=1) + + assert result["claimed"] == 1 + assert result["stdin_boundary_replayed"] == 1 + assert result["stdin_boundary_replay_terminal_written"] == 1 + assert result["apply_completed"] == 0 + assert result["apply_blocked"] == 1 + apply.assert_not_awaited() + terminal_write.assert_awaited_once() + + def test_check_mode_worker_error_backoff_is_short_and_bounded() -> None: from src.jobs import awooop_ansible_check_mode_job as job