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

fix(agent): harden apply closure and telegram ownership
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 2m36s
Details
CD Pipeline / build-and-deploy (push) Successful in 7m2s
Details
CD Pipeline / post-deploy-checks (push) Successful in 1m53s
Details

Browse Source
This commit is contained in:
ogt
2026-07-11 09:56:54 +08:00
parent 03d758b6bd
commit 0b8a271270
7 changed files with 706 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,10 +20,11 @@ from pathlib import Path
from typing import Any
import structlog
from sqlalchemy import text
from sqlalchemy import select, text
from src.core.config import settings
from src.db.base import get_db_context
from src.db.models import PlaybookRecord
from src.services.ai_automation_runtime_contract import (
AI_AUTOMATION_STAGE_RECEIPT_SCHEMA_VERSION,
)
@@ -36,6 +37,7 @@ from src.services.awooop_ansible_learning_writeback import (
from src.services.awooop_ansible_post_verifier import (
run_ansible_asset_post_verifier,
)
from src.services.playbook_evolver import TRUST_ARCHIVE_THRESHOLD
logger = structlog.get_logger(__name__)
@@ -43,6 +45,9 @@ _SAFE_HOST_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
_PLAYBOOK_PREFIX = Path("infra/ansible/playbooks")
_STDOUT_LIMIT = 20_000
_STDERR_LIMIT = 12_000
_CONTROLLED_RETRY_MAX_ATTEMPTS = 1
_CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS = 15
_CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS = 30
_EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300
_EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800
_EXECUTION_CAPABILITY_SAFETY_MARGIN_SECONDS = 5
@@ -215,6 +220,181 @@ def _safe_candidate(input_payload: dict[str, Any]) -> dict[str, Any]:
raise ValueError("no_safe_check_mode_candidate")
def _apply_ansible_playbook_trust_gate(
candidate_input: dict[str, Any],
trust_rows: list[Mapping[str, Any]],
) -> tuple[dict[str, Any], dict[str, Any]]:
"""Filter failed PlayBooks before they can claim another runtime apply."""
candidates = candidate_input.get("executor_candidates")
if not isinstance(candidates, list):
candidates = []
trust_by_id = {
str(row.get("playbook_id") or ""): row
for row in trust_rows
if str(row.get("playbook_id") or "")
}
eligible: list[dict[str, Any]] = []
decisions: list[dict[str, Any]] = []
for candidate in candidates:
if not isinstance(candidate, dict):
continue
catalog_id = str(candidate.get("catalog_id") or "")
canonical_id = canonical_ansible_playbook_id(catalog_id)
row = trust_by_id.get(canonical_id)
status = str((row or {}).get("status") or "").strip().lower()
trust_score = float((row or {}).get("trust_score") or 0.0)
success_count = int((row or {}).get("success_count") or 0)
failure_count = int((row or {}).get("failure_count") or 0)
retired = status in {"deprecated", "archived"}
learned_failure = bool(
row
and failure_count > success_count
and trust_score < TRUST_ARCHIVE_THRESHOLD
)
circuit_open = retired or learned_failure
reason = (
"playbook_status_retired"
if retired
else "playbook_trust_below_archive_threshold"
if learned_failure
else "playbook_trust_allows_controlled_canary"
if row
else "playbook_unobserved_canary_allowed"
)
decisions.append(
{
"catalog_id": catalog_id,
"canonical_playbook_id": canonical_id,
"trust_record_present": row is not None,
"status": status or None,
"trust_score": round(trust_score, 6) if row else None,
"success_count": success_count,
"failure_count": failure_count,
"review_required": bool((row or {}).get("review_required")),
"circuit_open": circuit_open,
"reason": reason,
}
)
if not circuit_open:
eligible.append(candidate)
blocked_count = sum(1 for item in decisions if item["circuit_open"])
status = (
"circuit_open"
if decisions and not eligible
else "degraded_candidates_filtered"
if blocked_count
else "closed"
)
gate = {
"schema_version": "ansible_playbook_trust_claim_gate_v1",
"policy_source": "playbook_evolver.TRUST_ARCHIVE_THRESHOLD",
"trust_archive_threshold": TRUST_ARCHIVE_THRESHOLD,
"status": status,
"candidate_count": len(decisions),
"eligible_candidate_count": len(eligible),
"blocked_candidate_count": blocked_count,
"owner_review_used_as_terminal": False,
"next_controlled_action": (
"generate_or_verify_playbook_repair_candidate"
if status == "circuit_open"
else "continue_with_highest_ranked_eligible_candidate"
),
"decisions": decisions,
}
return (
{
**candidate_input,
"executor_candidates": eligible,
"playbook_trust_gate": gate,
},
gate,
)
async def _guard_candidate_input_by_playbook_trust(
db: Any,
candidate_input: dict[str, Any],
) -> tuple[dict[str, Any], dict[str, Any]]:
candidates = candidate_input.get("executor_candidates")
canonical_ids = sorted(
{
canonical_ansible_playbook_id(str(candidate.get("catalog_id") or ""))
for candidate in candidates or []
if isinstance(candidate, dict) and candidate.get("catalog_id")
}
)
trust_rows: list[Mapping[str, Any]] = []
if canonical_ids:
result = await db.execute(
select(
PlaybookRecord.playbook_id,
PlaybookRecord.status,
PlaybookRecord.trust_score,
PlaybookRecord.success_count,
PlaybookRecord.failure_count,
PlaybookRecord.review_required,
).where(PlaybookRecord.playbook_id.in_(canonical_ids))
)
trust_rows = list(result.mappings().all())
return _apply_ansible_playbook_trust_gate(candidate_input, trust_rows)
async def _revalidate_claim_playbook_trust(
db: Any,
claim: AnsibleCheckModeClaim,
) -> bool:
candidate_input = {
**claim.input_payload,
"executor_candidates": [
{
"catalog_id": claim.catalog_id,
"playbook_path": claim.apply_playbook_path,
"check_mode_playbook_path": claim.playbook_path,
"inventory_hosts": list(claim.inventory_hosts),
"risk_level": claim.risk_level,
}
],
}
_, gate = await _guard_candidate_input_by_playbook_trust(
db,
candidate_input,
)
claim.input_payload["playbook_trust_gate"] = gate
if gate.get("status") != "circuit_open":
return True
claim.input_payload["controlled_apply_allowed"] = False
claim.input_payload["controlled_apply_blocker"] = (
"playbook_trust_circuit_open"
)
await db.execute(
text("""
UPDATE automation_operation_log
SET input = coalesce(input, '{}'::jsonb) || jsonb_build_object(
'controlled_apply_allowed', false,
'controlled_apply_blocker', 'playbook_trust_circuit_open',
'playbook_trust_gate', CAST(:trust_gate AS jsonb)
),
dry_run_result = coalesce(
dry_run_result,
'{}'::jsonb
) || jsonb_build_object(
'controlled_apply_allowed', false,
'controlled_apply_blocker', 'playbook_trust_circuit_open',
'playbook_trust_gate', CAST(:trust_gate AS jsonb)
)
WHERE op_id = CAST(:op_id AS uuid)
AND operation_type = 'ansible_check_mode_executed'
"""),
{
"trust_gate": json.dumps(gate, ensure_ascii=False),
"op_id": claim.op_id,
},
)
return False
def _allowed_controlled_apply_risks() -> set[str]:
raw = str(settings.AWOOOP_ANSIBLE_CONTROLLED_APPLY_ALLOWED_RISK_LEVELS or "")
return {item.strip().lower() for item in raw.split(",") if item.strip()}
@@ -276,6 +456,7 @@ def build_ansible_check_mode_claim_input(
"target_selector": candidate_input.get("target_selector") or {},
"source_truth_diff": candidate_input.get("source_truth_diff") or {},
"risk_policy_decision": candidate_input.get("risk_policy_decision") or {},
"playbook_trust_gate": candidate_input.get("playbook_trust_gate") or {},
"source_candidate_op_id": source_candidate_op_id,
"catalog_id": safe["catalog_id"],
"playbook_path": safe["playbook_path"],
@@ -1641,34 +1822,14 @@ async def _record_runtime_stage_receipts(
return False
async def _record_retry_runtime_stage_receipt(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
async def _append_runtime_stage_receipts_to_apply(
*,
apply_op_id: str,
replay_op_id: str,
receipts: tuple[dict[str, Any], ...],
project_id: str,
) -> bool:
"""Append the verified no-write retry terminal to the original apply run."""
receipt = _runtime_stage_receipt(
claim,
stage_id="retry_or_rollback",
evidence_ref=f"automation_operation_log:{replay_op_id}:dry_run_result",
detail={
"failed_apply_op_id": apply_op_id,
"retry_check_mode_op_id": replay_op_id,
"terminal_type": (
"no_write_replay_passed_waiting_repair_candidate"
if result.returncode == 0
else "no_write_replay_failed_waiting_playbook_or_transport_repair"
),
"check_mode_replay_performed": True,
"check_mode_replay_returncode": result.returncode,
"runtime_apply_executed": False,
"rollback_performed": False,
},
)
if not receipts:
return False
try:
async with get_db_context(project_id) as db:
updated = await db.execute(
@@ -1679,40 +1840,327 @@ async def _record_retry_runtime_stage_receipt(
'{runtime_stage_receipts}',
coalesce(
(
SELECT jsonb_agg(existing_receipt)
FROM jsonb_array_elements(
coalesce(
apply.input -> 'runtime_stage_receipts',
'[]'::jsonb
SELECT jsonb_agg(
deduplicated.receipt
ORDER BY deduplicated.stage_id
)
FROM (
SELECT DISTINCT ON (
merged.receipt ->> 'stage_id'
)
) AS existing_receipt
WHERE existing_receipt ->> 'stage_id'
<> 'retry_or_rollback'
merged.receipt,
merged.receipt ->> 'stage_id' AS stage_id
FROM jsonb_array_elements(
coalesce(
apply.input -> 'runtime_stage_receipts',
'[]'::jsonb
) || CAST(:receipts AS jsonb)
) WITH ORDINALITY AS merged(receipt, position)
WHERE merged.receipt ->> 'stage_id' IS NOT NULL
ORDER BY
merged.receipt ->> 'stage_id',
merged.position DESC
) deduplicated
),
'[]'::jsonb
) || jsonb_build_array(CAST(:receipt AS jsonb)),
),
true
)
WHERE apply.op_id = CAST(:apply_op_id AS uuid)
AND apply.operation_type = 'ansible_apply_executed'
"""),
{
"receipt": json.dumps(receipt, ensure_ascii=False),
"receipts": json.dumps(receipts, ensure_ascii=False),
"apply_op_id": apply_op_id,
},
)
return bool(updated.rowcount)
except Exception as exc:
logger.warning(
"ansible_retry_runtime_stage_receipt_write_failed",
automation_run_id=_automation_run_id_for_claim(claim),
"ansible_runtime_terminal_receipt_append_failed",
apply_op_id=apply_op_id,
replay_op_id=replay_op_id,
error=str(exc),
error_type=type(exc).__name__,
)
return False
async def _record_incident_terminal_disposition(
claim: AnsibleCheckModeClaim,
*,
apply_op_id: str,
project_id: str,
terminal_type: str,
success_terminal: bool,
retry_op_id: str | None = None,
telegram_receipt_acknowledged: bool,
) -> dict[str, Any] | None:
automation_run_id = _automation_run_id_for_claim(claim)
incident_status = "RESOLVED" if success_terminal else "MITIGATING"
safe_next_action = (
"keep_verified_repair_and_monitor_recurrence"
if success_terminal
else "queue_ai_playbook_or_transport_repair_candidate"
)
terminal_payload = {
"schema_version": "ansible_incident_terminal_disposition_v1",
"automation_run_id": automation_run_id,
"apply_op_id": apply_op_id,
"retry_op_id": retry_op_id,
"terminal_type": terminal_type,
"incident_status": incident_status,
"incident_resolved": success_terminal,
"no_write_terminal": not success_terminal,
"safe_next_action": safe_next_action,
"telegram_receipt_acknowledged": telegram_receipt_acknowledged,
"raw_log_payload_stored": False,
"secret_value_stored": False,
}
try:
async with get_db_context(project_id) as db:
updated = await db.execute(
text("""
UPDATE incidents
SET status = CAST(:incident_status AS incidentstatus),
outcome = CAST(
coalesce(CAST(outcome AS jsonb), '{}'::jsonb)
|| jsonb_build_object(
'automation_terminal',
CAST(:terminal_payload AS jsonb)
)
AS json
),
resolved_at = CASE
WHEN :success_terminal THEN NOW()
ELSE resolved_at
END,
updated_at = NOW()
WHERE incident_id = :incident_id
AND project_id = :project_id
AND (
:success_terminal
OR upper(status::text) NOT IN ('RESOLVED', 'CLOSED')
)
RETURNING
status::text AS incident_status,
updated_at,
resolved_at,
outcome
"""),
{
"incident_status": incident_status,
"terminal_payload": json.dumps(
terminal_payload,
ensure_ascii=False,
),
"success_terminal": success_terminal,
"incident_id": claim.incident_id,
"project_id": project_id,
},
)
row = updated.mappings().one_or_none()
if row is None:
return None
outcome = _json_loads(row.get("outcome"))
readback = outcome.get("automation_terminal")
if not isinstance(readback, Mapping):
return None
if (
readback.get("automation_run_id") != automation_run_id
or readback.get("apply_op_id") != apply_op_id
or readback.get("terminal_type") != terminal_type
):
return None
return {
**terminal_payload,
"incident_id": claim.incident_id,
"incident_status": str(row.get("incident_status") or ""),
"incident_row_version": (
row["updated_at"].isoformat()
if row.get("updated_at") is not None
else None
),
"repository_write_acknowledged": True,
"repository_readback_verified": True,
"durable_write_acknowledged": True,
"writer_source_sha": os.getenv(
"AWOOOI_BUILD_COMMIT_SHA",
"",
).strip().lower()
or None,
}
except Exception as exc:
logger.warning(
"ansible_incident_terminal_disposition_write_failed",
automation_run_id=automation_run_id,
incident_id=claim.incident_id,
apply_op_id=apply_op_id,
error_type=type(exc).__name__,
)
return None
async def _record_retry_runtime_stage_receipt(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
*,
apply_op_id: str,
replay_op_id: str,
project_id: str,
) -> bool:
"""Append the verified no-write retry terminal to the original apply run."""
automation_run_id = _automation_run_id_for_claim(claim)
terminal_type = (
"no_write_replay_passed_waiting_repair_candidate"
if result.returncode == 0
else "no_write_replay_failed_waiting_playbook_or_transport_repair"
)
retry_idempotency_key = f"ansible-retry:{apply_op_id}:check-mode"
replay_readback_verified = False
telegram_receipt_acknowledged = False
try:
async with get_db_context(project_id) as db:
replay_readback = await db.execute(
text("""
SELECT status, input, output, dry_run_result
FROM automation_operation_log
WHERE op_id = CAST(:replay_op_id AS uuid)
AND operation_type = 'ansible_execution_skipped'
AND parent_op_id = CAST(:apply_op_id AS uuid)
"""),
{
"replay_op_id": replay_op_id,
"apply_op_id": apply_op_id,
},
)
replay_row = replay_readback.mappings().one_or_none()
replay_input = _json_loads(
replay_row.get("input") if replay_row else None
)
replay_output = _json_loads(
replay_row.get("output") if replay_row else None
)
replay_readback_verified = bool(
replay_row
and replay_row.get("status")
== ("success" if result.returncode == 0 else "failed")
and replay_input.get("automation_run_id")
== automation_run_id
and replay_input.get("retry_of_apply_op_id") == apply_op_id
and replay_output.get("runtime_apply_executed") is False
and replay_output.get("terminal_disposition") == terminal_type
)
telegram_readback = await db.execute(
text("""
SELECT EXISTS (
SELECT 1
FROM awooop_outbound_message
WHERE project_id = :project_id
AND channel_type = 'telegram'
AND send_status = 'sent'
AND source_envelope #>> '{callback_reply,action}'
= 'controlled_apply_result'
AND COALESCE(
source_envelope ->> 'automation_run_id',
source_envelope #>>
'{callback_reply,automation_run_id}',
source_envelope #>>
'{source_refs,automation_run_ids,0}'
) = :automation_run_id
)
"""),
{
"project_id": project_id,
"automation_run_id": automation_run_id,
},
)
telegram_receipt_acknowledged = telegram_readback.scalar() is True
except Exception as exc:
logger.warning(
"ansible_retry_terminal_readback_failed",
automation_run_id=automation_run_id,
apply_op_id=apply_op_id,
replay_op_id=replay_op_id,
error_type=type(exc).__name__,
)
if not replay_readback_verified:
logger.warning(
"ansible_retry_terminal_receipt_not_written_unverified",
automation_run_id=automation_run_id,
apply_op_id=apply_op_id,
replay_op_id=replay_op_id,
)
return False
retry_receipt = _runtime_stage_receipt(
claim,
stage_id="retry_or_rollback",
evidence_ref=f"automation_operation_log:{replay_op_id}:dry_run_result",
detail={
"schema_version": "ansible_controlled_retry_terminal_v2",
"failed_apply_op_id": apply_op_id,
"retry_check_mode_op_id": replay_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": replay_readback_verified,
"retry_operation_readback_verified": replay_readback_verified,
"retry_attempt": 1,
"max_retry_attempts": _CONTROLLED_RETRY_MAX_ATTEMPTS,
"retry_idempotency_key": retry_idempotency_key,
"retry_error_backoff_min_seconds": (
_CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS
),
"retry_error_backoff_max_seconds": (
_CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS
),
"bounded_retry": True,
"repair_candidate_required": True,
"safe_next_action": (
"queue_ai_playbook_or_transport_repair_candidate"
),
"repository_readback_verified": replay_readback_verified,
"durable_write_acknowledged": replay_readback_verified,
"writer_source_sha": os.getenv(
"AWOOOI_BUILD_COMMIT_SHA",
"",
).strip().lower()
or None,
"raw_log_payload_stored": False,
"secret_value_stored": False,
},
)
incident_detail = 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=telegram_receipt_acknowledged,
)
receipts = [retry_receipt]
if incident_detail is not None:
receipts.append(
_runtime_stage_receipt(
claim,
stage_id="incident_closure",
evidence_ref=(
f"incidents:{claim.incident_id}:automation_terminal"
),
detail=incident_detail,
)
)
return await _append_runtime_stage_receipts_to_apply(
apply_op_id=apply_op_id,
receipts=tuple(receipts),
project_id=project_id,
)
async def _record_learning_writeback_receipt(
claim: AnsibleCheckModeClaim,
result: AnsibleRunResult,
@@ -2750,6 +3198,18 @@ async def claim_pending_check_modes(
for row in rows:
source_op_id = str(row["op_id"])
candidate_input = _json_loads(row["input"])
candidate_input, trust_gate = await _guard_candidate_input_by_playbook_trust(
db,
candidate_input,
)
if trust_gate.get("status") == "circuit_open":
await _insert_skipped_candidate(
db,
source_candidate_op_id=source_op_id,
candidate_input=candidate_input,
reason="playbook_trust_circuit_open",
)
continue
try:
claim_input = build_ansible_check_mode_claim_input(
source_candidate_op_id=source_op_id,
@@ -3099,6 +3559,9 @@ async def _insert_skipped_candidate(
candidate_input: dict[str, Any],
reason: str,
) -> None:
trust_gate = candidate_input.get("playbook_trust_gate")
if not isinstance(trust_gate, dict):
trust_gate = {}
input_payload = {
"automation_run_id": str(source_candidate_op_id),
"incident_id": _incident_id_from_payload(candidate_input),
@@ -3110,6 +3573,7 @@ async def _insert_skipped_candidate(
"apply_enabled": False,
"source_candidate_op_id": source_candidate_op_id,
"not_used_reason": reason,
"playbook_trust_gate": trust_gate,
}
await db.execute(
text("""
@@ -3138,12 +3602,19 @@ async def _insert_skipped_candidate(
"output": json.dumps({
"not_used_reason": reason,
"decision_effect": "skipped_before_runtime",
"next_controlled_action": (
"generate_or_verify_playbook_repair_candidate"
if reason == "playbook_trust_circuit_open"
else "repair_candidate_input_before_retry"
),
"playbook_trust_gate": trust_gate,
}, ensure_ascii=False),
"dry_run_result": json.dumps({
"check_mode_executed": False,
"apply_executed": False,
"skipped": True,
"reason": reason,
"playbook_trust_gate": trust_gate,
}, ensure_ascii=False),
"parent_op_id": source_candidate_op_id,
"tags": ["ansible", "check_mode", "skipped", "apply_locked"],
@@ -3318,6 +3789,16 @@ async def run_controlled_apply_for_claim(
idempotency_key=apply_idempotency_key,
)
return None
if not await _revalidate_claim_playbook_trust(db, claim):
logger.warning(
"ansible_controlled_apply_blocked_by_playbook_trust",
op_id=claim.op_id,
source_candidate_op_id=claim.source_candidate_op_id,
incident_id=claim.incident_id,
catalog_id=claim.catalog_id,
blocker="playbook_trust_circuit_open",
)
return None
inserted = await db.execute(
text("""
INSERT INTO automation_operation_log (