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

fix(agent): replay stdin boundary failures without apply
All checks were successful
CD Pipeline / workflow-shape (push) Successful in 1s
Details
CD Pipeline / cancel-stale-cd (push) Has been skipped
Details
CD Pipeline / tests (push) Successful in 2m34s
Details
CD Pipeline / build-and-deploy (push) Successful in 7m28s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m18s
Details

Browse Source
This commit is contained in:
ogt
2026-07-14 10:43:00 +08:00
parent 708f976275
commit fc3b08afe3
2 changed files with 641 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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,

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

@@ -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