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

fix(agent): reconcile missing retry terminal projections
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 3m17s
Details
CD Pipeline / build-and-deploy (push) Successful in 7m55s
Details
CD Pipeline / post-deploy-checks (push) Successful in 2m13s
Details
AWOOOI Harbor 110 Local Repair / workflow-shape (push) Successful in 1s
Details
AWOOOI Harbor 110 Local Repair / harbor-110-local-repair (push) Successful in 18s
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 15:01:22 +08:00
parent 18c7b7f651
commit 86bba878e4
2 changed files with 490 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -3317,10 +3317,44 @@ async def backfill_missing_auto_repair_execution_receipts_once(
"runtime_stage_receipts_written": 0,
"incident_closure_written": 0,
"telegram_receipt_acknowledged": 0,
"retry_terminal_projection_scanned": 0,
"retry_terminal_projection_written": 0,
"retry_terminal_projection_incident_receipt_written": 0,
"retry_terminal_projection_lifecycle_written": 0,
"retry_terminal_projection_verified": 0,
"retry_terminal_projection_runtime_apply_executed": False,
"skipped": 0,
"error": None,
}
try:
projection = await backfill_missing_retry_terminal_projections_once(
project_id=project_id,
window_hours=window_hours,
limit=max(1, min(limit, 5)),
)
stats["retry_terminal_projection_scanned"] = int(
projection.get("scanned") or 0
)
stats["retry_terminal_projection_written"] = int(
projection.get("written") or 0
)
stats["retry_terminal_projection_incident_receipt_written"] = int(
projection.get("incident_receipt_written") or 0
)
stats["retry_terminal_projection_lifecycle_written"] = int(
projection.get("lifecycle_written") or 0
)
stats["retry_terminal_projection_verified"] = int(
projection.get("verified") or 0
)
projection_error = str(projection.get("error") or "")[:500]
if stats["retry_terminal_projection_scanned"] or projection_error:
stats["scanned"] = stats["retry_terminal_projection_scanned"]
stats["incident_closure_written"] = stats[
"retry_terminal_projection_written"
]
stats["error"] = projection_error or None
return stats
async with get_db_context(project_id) as db:
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
result = await db.execute(
@@ -3654,6 +3688,327 @@ async def preflight_failed_apply_retry_queue_once(
}
async def _load_missing_retry_terminal_projection_rows(
*,
project_id: str,
window_hours: int,
limit: int,
) -> list[dict[str, Any]]:
"""Load verified no-write retries whose incident projection is incomplete."""
async with get_db_context(project_id) as db:
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
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,
replay.op_id::text AS replay_op_id,
retry_receipt.receipt #>> '{detail,terminal_type}'
AS terminal_type,
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 coalesce(
outbound.source_envelope
->> 'automation_run_id',
outbound.source_envelope #>>
'{callback_reply,automation_run_id}',
outbound.source_envelope #>>
'{source_refs,automation_run_ids,0}'
) = apply.input ->> 'automation_run_id'
AND coalesce(
outbound.source_envelope #>>
'{callback_reply,incident_id}',
outbound.source_envelope #>>
'{source_refs,incident_ids,0}'
) = coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
)
) AS telegram_receipt_acknowledged
FROM automation_operation_log apply
JOIN automation_operation_log replay
ON replay.parent_op_id = apply.op_id
AND replay.operation_type = 'ansible_execution_skipped'
AND replay.input ->> 'execution_mode'
= 'controlled_retry_check_mode_replay'
AND replay.status IN ('success', 'failed')
AND replay.output ->> 'runtime_apply_executed' = 'false'
JOIN incidents incident
ON incident.incident_id = coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
)
AND incident.project_id = :project_id
JOIN LATERAL (
SELECT receipt.value AS receipt
FROM jsonb_array_elements(
coalesce(
apply.input -> 'runtime_stage_receipts',
'[]'::jsonb
)
) receipt(value)
WHERE receipt.value ->> 'stage_id' = 'retry_or_rollback'
AND receipt.value #>>
'{detail,retry_check_mode_op_id}' = replay.op_id::text
AND receipt.value #>>
'{detail,failed_apply_op_id}' = apply.op_id::text
AND receipt.value #>>
'{detail,verified_no_write_terminal}' = 'true'
AND receipt.value #>>
'{detail,runtime_apply_executed}' = 'false'
LIMIT 1
) retry_receipt ON TRUE
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.incident_id = incident.incident_id
AND verifier.post_execution_state ->> 'apply_op_id'
= apply.op_id::text
AND verifier.verification_result IN ('failed', 'timeout')
)
AND (
cast(incident.outcome AS jsonb) #>>
'{automation_terminal,automation_run_id}'
IS DISTINCT FROM apply.input ->> 'automation_run_id'
OR cast(incident.outcome AS jsonb) #>>
'{automation_terminal,apply_op_id}'
IS DISTINCT FROM apply.op_id::text
OR cast(incident.outcome AS jsonb) #>>
'{automation_terminal,retry_op_id}'
IS DISTINCT FROM replay.op_id::text
OR cast(incident.outcome AS jsonb) #>>
'{automation_terminal,terminal_type}'
IS DISTINCT FROM retry_receipt.receipt #>>
'{detail,terminal_type}'
OR cast(incident.outcome AS jsonb) #>>
'{automation_terminal,no_write_terminal}'
IS DISTINCT FROM 'true'
OR NOT EXISTS (
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = incident.incident_id
AND lifecycle.event_type::text
= 'EXECUTION_COMPLETED'
AND lifecycle.success IS FALSE
AND lifecycle.context ->> 'apply_op_id'
= apply.op_id::text
AND lifecycle.context ->> 'automation_run_id'
= apply.input ->> 'automation_run_id'
)
)
ORDER BY replay.created_at DESC
LIMIT :limit
"""),
{
"project_id": project_id,
"window_hours": max(1, window_hours),
"limit": max(1, min(limit, 5)),
},
)
return [dict(row) for row in result.mappings().all()]
async def _verify_retry_terminal_projection_readback(
claim: AnsibleCheckModeClaim,
*,
apply_op_id: str,
replay_op_id: str,
terminal_type: str,
project_id: str,
) -> bool:
automation_run_id = _automation_run_id_for_claim(claim)
async with get_db_context(project_id) as db:
await db.execute(text("SET LOCAL statement_timeout = '5000ms'"))
result = await db.execute(
text("""
SELECT
upper(incident.status::text) = 'MITIGATING'
AND cast(incident.outcome AS jsonb) #>>
'{automation_terminal,schema_version}'
= 'ansible_incident_terminal_disposition_v1'
AND cast(incident.outcome AS jsonb) #>>
'{automation_terminal,automation_run_id}'
= :automation_run_id
AND cast(incident.outcome AS jsonb) #>>
'{automation_terminal,apply_op_id}' = :apply_op_id
AND cast(incident.outcome AS jsonb) #>>
'{automation_terminal,retry_op_id}' = :replay_op_id
AND cast(incident.outcome AS jsonb) #>>
'{automation_terminal,terminal_type}' = :terminal_type
AND cast(incident.outcome AS jsonb) #>>
'{automation_terminal,no_write_terminal}' = 'true'
AND 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 #>> '{detail,apply_op_id}'
= :apply_op_id
AND receipt.value #>> '{detail,retry_op_id}'
= :replay_op_id
AND receipt.value #>>
'{detail,repository_readback_verified}' = 'true'
AND receipt.value ->> 'durable_receipt' = 'true'
)
AND EXISTS (
SELECT 1
FROM alert_operation_log lifecycle
WHERE lifecycle.incident_id = incident.incident_id
AND lifecycle.event_type::text = 'EXECUTION_COMPLETED'
AND lifecycle.success IS FALSE
AND lifecycle.context ->> 'apply_op_id' = :apply_op_id
AND lifecycle.context ->> 'automation_run_id'
= :automation_run_id
) AS projection_verified
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
"""),
{
"automation_run_id": automation_run_id,
"apply_op_id": apply_op_id,
"replay_op_id": replay_op_id,
"terminal_type": terminal_type,
"incident_id": claim.incident_id,
"project_id": project_id,
},
)
return result.scalar() is True
async def backfill_missing_retry_terminal_projections_once(
*,
project_id: str = "awoooi",
window_hours: int = 24,
limit: int = 1,
) -> dict[str, Any]:
"""Repair missing incident terminals from durable no-write retry receipts."""
stats: dict[str, Any] = {
"scanned": 0,
"written": 0,
"incident_receipt_written": 0,
"lifecycle_written": 0,
"verified": 0,
"runtime_apply_executed": False,
"error": None,
}
try:
rows = await _load_missing_retry_terminal_projection_rows(
project_id=project_id,
window_hours=window_hours,
limit=limit,
)
stats["scanned"] = len(rows)
for row in rows:
reconstructed = _claim_from_apply_operation_row(row)
if reconstructed is None:
continue
claim, _failed_apply_result = reconstructed
apply_op_id = str(row.get("op_id") or "")
replay_op_id = str(row.get("replay_op_id") or "")
terminal_type = str(row.get("terminal_type") or "")
if not (
apply_op_id
and replay_op_id
and terminal_type.startswith("no_write_replay_")
and row.get("telegram_receipt_acknowledged") is True
):
continue
terminal = await _record_incident_terminal_disposition(
claim,
apply_op_id=apply_op_id,
project_id=project_id,
terminal_type=terminal_type,
success_terminal=False,
retry_op_id=replay_op_id,
telegram_receipt_acknowledged=True,
)
if terminal is None:
continue
incident_receipt_written = (
await _append_runtime_stage_receipts_to_apply(
apply_op_id=apply_op_id,
receipts=(
_runtime_stage_receipt(
claim,
stage_id="incident_closure",
evidence_ref=(
f"incidents:{claim.incident_id}:automation_terminal"
),
detail=terminal,
derived_from_durable_chain=True,
),
),
project_id=project_id,
)
)
lifecycle_written = await _append_alert_lifecycle_receipt(
claim,
"EXECUTION_COMPLETED",
apply_op_id=apply_op_id,
success=False,
action_detail="controlled_retry_no_write_terminal_backfilled",
project_id=project_id,
error_message="failed_apply_retry_terminal",
post_verifier_passed=False,
)
stats["incident_receipt_written"] += int(
incident_receipt_written
)
stats["lifecycle_written"] += int(lifecycle_written)
if not incident_receipt_written or not lifecycle_written:
continue
verified = await _verify_retry_terminal_projection_readback(
claim,
apply_op_id=apply_op_id,
replay_op_id=replay_op_id,
terminal_type=terminal_type,
project_id=project_id,
)
stats["verified"] += int(verified)
stats["written"] += int(verified)
except Exception as exc:
stats["error"] = f"{type(exc).__name__}: {exc}"[:500]
logger.warning(
"ansible_retry_terminal_projection_backfill_failed",
project_id=project_id,
**stats,
)
return stats
async def run_failed_apply_check_mode_replay_once(
*,
project_id: str = "awoooi",
@@ -4857,6 +5212,16 @@ async def run_pending_check_modes_once(
"repair_receipt_telegram_acknowledged": int(
receipt_stats.get("telegram_receipt_acknowledged") or 0
),
"retry_terminal_projection_scanned": int(
receipt_stats.get("retry_terminal_projection_scanned") or 0
),
"retry_terminal_projection_written": int(
receipt_stats.get("retry_terminal_projection_written") or 0
),
"retry_terminal_projection_verified": int(
receipt_stats.get("retry_terminal_projection_verified") or 0
),
"retry_terminal_projection_runtime_apply_executed": False,
"repair_receipt_backfill_error": (
str(receipt_stats.get("error") or "")[:500] or None
),

125
apps/api/tests/test_ansible_verified_closure.py
View File

@@ -236,6 +236,21 @@ async def test_broker_reconciles_receipts_before_claiming_fresh_work(
"_expire_stale_ansible_execution_capabilities",
AsyncMock(return_value=0),
)
monkeypatch.setattr(
service,
"backfill_missing_retry_terminal_projections_once",
AsyncMock(
return_value={
"scanned": 0,
"written": 0,
"incident_receipt_written": 0,
"lifecycle_written": 0,
"verified": 0,
"runtime_apply_executed": False,
"error": None,
}
),
)
monkeypatch.setattr(
service,
"backfill_missing_auto_repair_execution_receipts_once",
@@ -263,6 +278,116 @@ async def test_broker_reconciles_receipts_before_claiming_fresh_work(
retry_replayer.assert_not_awaited()
@pytest.mark.asyncio
async def test_retry_terminal_projection_backfill_writes_and_verifies_no_apply(
monkeypatch: pytest.MonkeyPatch,
) -> None:
row = {
"op_id": "00000000-0000-0000-0000-000000000104",
"replay_op_id": "00000000-0000-0000-0000-000000000105",
"terminal_type": "no_write_replay_passed_waiting_repair_candidate",
"telegram_receipt_acknowledged": True,
}
monkeypatch.setattr(
service,
"_load_missing_retry_terminal_projection_rows",
AsyncMock(return_value=[row]),
)
monkeypatch.setattr(
service,
"_claim_from_apply_operation_row",
lambda _row: (
_claim(),
service.AnsibleRunResult(
returncode=1,
stdout="",
stderr="",
duration_ms=0,
),
),
)
terminal_writer = AsyncMock(
return_value={
"automation_run_id": "00000000-0000-0000-0000-000000000102",
"apply_op_id": row["op_id"],
"retry_op_id": row["replay_op_id"],
"terminal_type": row["terminal_type"],
"repository_readback_verified": True,
}
)
receipt_writer = AsyncMock(return_value=True)
lifecycle_writer = AsyncMock(return_value=True)
verifier = AsyncMock(return_value=True)
monkeypatch.setattr(
service, "_record_incident_terminal_disposition", terminal_writer
)
monkeypatch.setattr(
service, "_append_runtime_stage_receipts_to_apply", receipt_writer
)
monkeypatch.setattr(
service, "_append_alert_lifecycle_receipt", lifecycle_writer
)
monkeypatch.setattr(
service, "_verify_retry_terminal_projection_readback", verifier
)
result = await service.backfill_missing_retry_terminal_projections_once()
assert result == {
"scanned": 1,
"written": 1,
"incident_receipt_written": 1,
"lifecycle_written": 1,
"verified": 1,
"runtime_apply_executed": False,
"error": None,
}
terminal_writer.assert_awaited_once()
assert terminal_writer.await_args.kwargs["success_terminal"] is False
receipt = receipt_writer.await_args.kwargs["receipts"][0]
assert receipt["stage_id"] == "incident_closure"
assert receipt["derived_from_durable_chain"] is True
assert lifecycle_writer.await_args.args[1] == "EXECUTION_COMPLETED"
assert lifecycle_writer.await_args.kwargs["success"] is False
@pytest.mark.asyncio
async def test_broker_prioritizes_retry_terminal_projection_backfill(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
service,
"_expire_stale_ansible_execution_capabilities",
AsyncMock(return_value=0),
)
receipt_backfill = AsyncMock(
return_value={
"scanned": 1,
"written": 0,
"incident_closure_written": 1,
"telegram_receipt_acknowledged": 0,
"retry_terminal_projection_scanned": 1,
"retry_terminal_projection_written": 1,
"retry_terminal_projection_verified": 1,
"retry_terminal_projection_runtime_apply_executed": False,
"error": None,
}
)
monkeypatch.setattr(
service,
"backfill_missing_auto_repair_execution_receipts_once",
receipt_backfill,
)
result = await service.run_pending_check_modes_once(limit=1)
assert result["claimed"] == 0
assert result["retry_terminal_projection_verified"] == 1
assert result["retry_terminal_projection_runtime_apply_executed"] is False
assert result["repair_receipt_backfill_priority_tick"] is True
receipt_backfill.assert_awaited_once()
@pytest.mark.asyncio
async def test_signal_worker_retry_preflight_is_query_only(
monkeypatch: pytest.MonkeyPatch,