diff --git a/.gitea/workflows/cd.yaml b/.gitea/workflows/cd.yaml index 18dfe0221..bc40e5dbe 100644 --- a/.gitea/workflows/cd.yaml +++ b/.gitea/workflows/cd.yaml @@ -2705,10 +2705,170 @@ jobs: echo "❌ execution broker cannot reach any allowlisted SSH endpoint"; exit 1; } + read_workload_db_identity() { + workload="$1" + container="$2" + $KUBECTL exec -i "deployment/$workload" -n awoooi-prod \ + -c "$container" -- python - <<'PY' + import asyncio + import hashlib + import json + + from sqlalchemy import text + + from src.db.base import get_db_context + + + async def main() -> None: + try: + async with get_db_context("awoooi") as db: + result = await db.execute(text(""" + SELECT + current_user AS role_name, + rolconnlimit AS connection_limit + FROM pg_roles + WHERE rolname = current_user + """)) + row = result.mappings().one() + except Exception as exc: + print(json.dumps({ + "role_fingerprint": "", + "connection_limit": None, + "error_type": type(exc).__name__, + })) + return + print(json.dumps({ + "role_fingerprint": hashlib.sha256( + str(row["role_name"]).encode("utf-8") + ).hexdigest(), + "connection_limit": int(row["connection_limit"]), + "error_type": None, + })) + + + asyncio.run(main()) + PY + } + json_field() { + field="$1" + python3 -c \ + 'import json,sys; value=json.load(sys.stdin).get(sys.argv[1]); print(value if value is not None else "")' \ + "$field" + } + normalize_bool() { + case "${1:-}" in + true|TRUE|True|1) printf 'true' ;; + *) printf 'false' ;; + esac + } + workload_secret_projection() { + workload="$1" + container="$2" + explicit_ref=$($KUBECTL get "deployment/$workload" -n awoooi-prod \ + -o jsonpath="{.spec.template.spec.containers[?(@.name=='$container')].env[?(@.name=='DATABASE_URL')].valueFrom.secretKeyRef.name}" \ + || true) + env_from_refs=$($KUBECTL get "deployment/$workload" -n awoooi-prod \ + -o jsonpath="{range .spec.template.spec.containers[?(@.name=='$container')].envFrom[*]}{.secretRef.name}{'\\n'}{end}" \ + | sed '/^$/d' | paste -sd, - || true) + if [ -n "$explicit_ref" ]; then + printf '%s|%s\n' "$explicit_ref" "$env_from_refs" + else + printf '%s|%s\n' "$env_from_refs" "$env_from_refs" + fi + } + + API_DB_IDENTITY=$(read_workload_db_identity awoooi-api api \ + | tail -n 1 || true) + WORKER_DB_IDENTITY=$(read_workload_db_identity awoooi-worker worker \ + | tail -n 1 || true) + BROKER_DB_IDENTITY=$(read_workload_db_identity \ + awoooi-ansible-executor-broker broker | tail -n 1 || true) + API_DB_ROLE_FINGERPRINT=$(printf '%s' "$API_DB_IDENTITY" \ + | json_field role_fingerprint 2>/dev/null || true) + WORKER_DB_ROLE_FINGERPRINT=$(printf '%s' "$WORKER_DB_IDENTITY" \ + | json_field role_fingerprint 2>/dev/null || true) + BROKER_DB_ROLE_FINGERPRINT=$(printf '%s' "$BROKER_DB_IDENTITY" \ + | json_field role_fingerprint 2>/dev/null || true) + API_DB_CONNECTION_LIMIT=$(printf '%s' "$API_DB_IDENTITY" \ + | json_field connection_limit 2>/dev/null || true) + WORKER_DB_CONNECTION_LIMIT=$(printf '%s' "$WORKER_DB_IDENTITY" \ + | json_field connection_limit 2>/dev/null || true) + BROKER_DB_CONNECTION_LIMIT=$(printf '%s' "$BROKER_DB_IDENTITY" \ + | json_field connection_limit 2>/dev/null || true) + API_DB_CONNECTION_LIMIT=${API_DB_CONNECTION_LIMIT:-0} + WORKER_DB_CONNECTION_LIMIT=${WORKER_DB_CONNECTION_LIMIT:-0} + BROKER_DB_CONNECTION_LIMIT=${BROKER_DB_CONNECTION_LIMIT:-0} + + API_DB_PROJECTION=$(workload_secret_projection awoooi-api api) + WORKER_DB_PROJECTION=$(workload_secret_projection awoooi-worker worker) + BROKER_DB_PROJECTION=$(workload_secret_projection \ + awoooi-ansible-executor-broker broker) + API_DATABASE_SECRET_REF=${API_DB_PROJECTION%%|*} + WORKER_DATABASE_SECRET_REF=${WORKER_DB_PROJECTION%%|*} + BROKER_DATABASE_SECRET_REF=${BROKER_DB_PROJECTION%%|*} + API_SECRET_ENVFROM_REFS=${API_DB_PROJECTION#*|} + WORKER_SECRET_ENVFROM_REFS=${WORKER_DB_PROJECTION#*|} + BROKER_SECRET_ENVFROM_REFS=${BROKER_DB_PROJECTION#*|} + API_MONOLITHIC_SECRET_ENVFROM_ABSENT=false + WORKER_MONOLITHIC_SECRET_ENVFROM_ABSENT=false + BROKER_MONOLITHIC_SECRET_ENVFROM_ABSENT=false + if [ -z "$API_SECRET_ENVFROM_REFS" ]; then + API_MONOLITHIC_SECRET_ENVFROM_ABSENT=true + fi + if [ -z "$WORKER_SECRET_ENVFROM_REFS" ]; then + WORKER_MONOLITHIC_SECRET_ENVFROM_ABSENT=true + fi + if [ -z "$BROKER_SECRET_ENVFROM_REFS" ]; then + BROKER_MONOLITHIC_SECRET_ENVFROM_ABSENT=true + fi + + API_DB_NULL_POOL=$(normalize_bool "$($KUBECTL get deployment/awoooi-api \ + -n awoooi-prod -o jsonpath='{.spec.template.spec.containers[?(@.name=="api")].env[?(@.name=="DATABASE_NULL_POOL")].value}')") + WORKER_DB_NULL_POOL=$(normalize_bool "$($KUBECTL get deployment/awoooi-worker \ + -n awoooi-prod -o jsonpath='{.spec.template.spec.containers[?(@.name=="worker")].env[?(@.name=="DATABASE_NULL_POOL")].value}')") + BROKER_DB_NULL_POOL=$(normalize_bool "$($KUBECTL get deployment/awoooi-ansible-executor-broker \ + -n awoooi-prod -o jsonpath='{.spec.template.spec.containers[?(@.name=="broker")].env[?(@.name=="DATABASE_NULL_POOL")].value}')") + + DISTINCT_DATABASE_SECRET_REFS=false + if [ -n "$API_DATABASE_SECRET_REF" ] \ + && [ -n "$WORKER_DATABASE_SECRET_REF" ] \ + && [ -n "$BROKER_DATABASE_SECRET_REF" ] \ + && [ "$API_DATABASE_SECRET_REF" != "$WORKER_DATABASE_SECRET_REF" ] \ + && [ "$API_DATABASE_SECRET_REF" != "$BROKER_DATABASE_SECRET_REF" ] \ + && [ "$WORKER_DATABASE_SECRET_REF" != "$BROKER_DATABASE_SECRET_REF" ]; then + DISTINCT_DATABASE_SECRET_REFS=true + fi + DISTINCT_DB_ROLE_FINGERPRINTS=false + if [ -n "$API_DB_ROLE_FINGERPRINT" ] \ + && [ -n "$WORKER_DB_ROLE_FINGERPRINT" ] \ + && [ -n "$BROKER_DB_ROLE_FINGERPRINT" ] \ + && [ "$API_DB_ROLE_FINGERPRINT" != "$WORKER_DB_ROLE_FINGERPRINT" ] \ + && [ "$API_DB_ROLE_FINGERPRINT" != "$BROKER_DB_ROLE_FINGERPRINT" ] \ + && [ "$WORKER_DB_ROLE_FINGERPRINT" != "$BROKER_DB_ROLE_FINGERPRINT" ]; then + DISTINCT_DB_ROLE_FINGERPRINTS=true + fi + DB_IDENTITY_ISOLATED=false + if [ "$API_MONOLITHIC_SECRET_ENVFROM_ABSENT" = "true" ] \ + && [ "$WORKER_MONOLITHIC_SECRET_ENVFROM_ABSENT" = "true" ] \ + && [ "$DISTINCT_DATABASE_SECRET_REFS" = "true" ] \ + && [ "$DISTINCT_DB_ROLE_FINGERPRINTS" = "true" ]; then + DB_IDENTITY_ISOLATED=true + fi + CONNECTION_BUDGET_VERIFIED=false + if [ "$DB_IDENTITY_ISOLATED" = "true" ] \ + && [ "$API_DB_NULL_POOL" = "true" ] \ + && [ "$WORKER_DB_NULL_POOL" = "true" ] \ + && [ "$BROKER_DB_NULL_POOL" = "true" ] \ + && [ "$API_DB_CONNECTION_LIMIT" -ge 1 ] \ + && [ "$WORKER_DB_CONNECTION_LIMIT" -ge 1 ] \ + && [ "$BROKER_DB_CONNECTION_LIMIT" -ge 1 ]; then + CONNECTION_BUDGET_VERIFIED=true + fi + BOUNDARY_VERIFIED_AT=$(date -u +"%Y-%m-%dT%H:%M:%SZ") $KUBECTL create configmap awoooi-executor-boundary-verification \ -n awoooi-prod \ - --from-literal=schema_version=awoooi_executor_boundary_verification_v2 \ + --from-literal=schema_version=awoooi_executor_boundary_verification_v3 \ --from-literal=verified_source_sha="$SOURCE_REVISION" \ --from-literal=verified_at="$BOUNDARY_VERIFIED_AT" \ --from-literal=workflow_run_id="$WORKFLOW_RUN_ID" \ @@ -2727,6 +2887,29 @@ jobs: --from-literal=worker_ssh_egress_denied=true \ --from-literal=broker_ssh_egress_allowlisted=true \ --from-literal=legacy_namespace_egress_allow_absent=true \ + --from-literal=db_identity_verifier=runtime_role_fingerprint_and_secret_projection_v1 \ + --from-literal=api_database_secret_ref="$API_DATABASE_SECRET_REF" \ + --from-literal=worker_database_secret_ref="$WORKER_DATABASE_SECRET_REF" \ + --from-literal=broker_database_secret_ref="$BROKER_DATABASE_SECRET_REF" \ + --from-literal=api_db_role_fingerprint="$API_DB_ROLE_FINGERPRINT" \ + --from-literal=worker_db_role_fingerprint="$WORKER_DB_ROLE_FINGERPRINT" \ + --from-literal=broker_db_role_fingerprint="$BROKER_DB_ROLE_FINGERPRINT" \ + --from-literal=api_db_connection_limit="$API_DB_CONNECTION_LIMIT" \ + --from-literal=worker_db_connection_limit="$WORKER_DB_CONNECTION_LIMIT" \ + --from-literal=broker_db_connection_limit="$BROKER_DB_CONNECTION_LIMIT" \ + --from-literal=api_required_connection_budget=1 \ + --from-literal=worker_required_connection_budget=1 \ + --from-literal=broker_required_connection_budget=1 \ + --from-literal=api_database_null_pool="$API_DB_NULL_POOL" \ + --from-literal=worker_database_null_pool="$WORKER_DB_NULL_POOL" \ + --from-literal=broker_database_null_pool="$BROKER_DB_NULL_POOL" \ + --from-literal=api_monolithic_secret_env_from_absent="$API_MONOLITHIC_SECRET_ENVFROM_ABSENT" \ + --from-literal=worker_monolithic_secret_env_from_absent="$WORKER_MONOLITHIC_SECRET_ENVFROM_ABSENT" \ + --from-literal=broker_monolithic_secret_env_from_absent="$BROKER_MONOLITHIC_SECRET_ENVFROM_ABSENT" \ + --from-literal=distinct_database_secret_refs="$DISTINCT_DATABASE_SECRET_REFS" \ + --from-literal=distinct_db_role_fingerprints="$DISTINCT_DB_ROLE_FINGERPRINTS" \ + --from-literal=db_identity_isolated="$DB_IDENTITY_ISOLATED" \ + --from-literal=connection_budget_verified="$CONNECTION_BUDGET_VERIFIED" \ --dry-run=client -o yaml \ | $KUBECTL apply -f - $KUBECTL annotate configmap \ @@ -2741,6 +2924,12 @@ jobs: echo "❌ executor boundary receipt source mismatch"; exit 1; } echo "✅ AIA-P0-011 API/worker/broker production boundary verified" + if [ "$DB_IDENTITY_ISOLATED" != "true" ] \ + || [ "$CONNECTION_BUDGET_VERIFIED" != "true" ]; then + echo "⚠️ AIA-P0-011 workload DB identity/budget remains in progress" + else + echo "✅ AIA-P0-011 workload DB identity/budget verified" + fi # Health Check HEALTH_PASS=0 diff --git a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py index 47de8ccf6..2d9fc6871 100644 --- a/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py +++ b/apps/api/src/jobs/awooop_ansible_candidate_backfill_job.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import json +import random from collections.abc import Awaitable, Callable from types import SimpleNamespace from typing import Any @@ -42,6 +43,12 @@ _BACKFILL_REASON = ( "truth-chain found allowlisted Ansible catalog candidates but no durable " "candidate row existed; enqueue for check-mode worker" ) +_ERROR_BACKOFF_MIN_SECONDS = 15.0 +_ERROR_BACKOFF_MAX_SECONDS = 30.0 + + +def _error_backoff_seconds() -> float: + return random.uniform(_ERROR_BACKOFF_MIN_SECONDS, _ERROR_BACKOFF_MAX_SECONDS) async def _fetch_missing_candidate_incidents( @@ -242,10 +249,13 @@ async def enqueue_missing_ansible_candidates_once( "pre_decision_evidence_ready": 0, "pre_decision_evidence_blocked": 0, "repair_receipts_backfilled": 0, + "failed_apply_retry_scanned": 0, "failed_apply_retry_replayed": 0, "failed_apply_retry_check_mode_passed": 0, "failed_apply_retry_check_mode_failed": 0, "failed_apply_retry_stage_receipt_written": 0, + "failed_apply_retry_blockers": [], + "failed_apply_retry_priority_tick": False, "error": None, } @@ -264,14 +274,61 @@ async def enqueue_missing_ansible_candidates_once( "pre_decision_evidence_ready": 0, "pre_decision_evidence_blocked": 0, "repair_receipts_backfilled": 0, + "failed_apply_retry_scanned": 0, "failed_apply_retry_replayed": 0, "failed_apply_retry_check_mode_passed": 0, "failed_apply_retry_check_mode_failed": 0, "failed_apply_retry_stage_receipt_written": 0, + "failed_apply_retry_blockers": [], + "failed_apply_retry_priority_tick": False, "error": None, } try: + retry_stats = await retry_replayer( + project_id=project_id, + window_hours=bounded_window_hours, + ) + stats["failed_apply_retry_scanned"] = int(retry_stats.get("scanned") or 0) + stats["failed_apply_retry_replayed"] = int( + retry_stats.get("replayed") or 0 + ) + stats["failed_apply_retry_check_mode_passed"] = int( + retry_stats.get("check_mode_passed") or 0 + ) + stats["failed_apply_retry_check_mode_failed"] = int( + retry_stats.get("check_mode_failed") or 0 + ) + stats["failed_apply_retry_stage_receipt_written"] = int( + retry_stats.get("runtime_stage_receipt_written") or 0 + ) + retry_blockers = retry_stats.get("blockers") + if isinstance(retry_blockers, list): + stats["failed_apply_retry_blockers"] = [ + str(item)[:200] for item in retry_blockers + ] + retry_error = retry_stats.get("error") + if retry_error: + stats["error"] = str(retry_error)[:500] + stats["failed_apply_retry_priority_tick"] = bool( + stats["failed_apply_retry_scanned"] + or stats["failed_apply_retry_replayed"] + or stats["failed_apply_retry_blockers"] + or retry_error + ) + if stats["failed_apply_retry_priority_tick"]: + logger.info("awooop_ansible_failed_apply_retry_priority_tick", **stats) + return stats + + receipt_stats = await receipt_backfiller( + project_id=project_id, + window_hours=bounded_window_hours, + limit=bounded_limit, + ) + stats["repair_receipts_backfilled"] = int(receipt_stats.get("written") or 0) + if receipt_stats.get("error") and not stats["error"]: + stats["error"] = receipt_stats["error"] + incidents = await _fetch_missing_candidate_incidents( project_id=project_id, window_hours=bounded_window_hours, @@ -324,32 +381,6 @@ async def enqueue_missing_ansible_candidates_once( stats["queued"] += 1 else: stats["already_existing_or_write_skipped"] += 1 - receipt_stats = await receipt_backfiller( - project_id=project_id, - window_hours=bounded_window_hours, - limit=bounded_limit, - ) - stats["repair_receipts_backfilled"] = int(receipt_stats.get("written") or 0) - if receipt_stats.get("error") and not stats["error"]: - stats["error"] = receipt_stats["error"] - retry_stats = await retry_replayer( - project_id=project_id, - window_hours=bounded_window_hours, - ) - stats["failed_apply_retry_replayed"] = int( - retry_stats.get("replayed") or 0 - ) - stats["failed_apply_retry_check_mode_passed"] = int( - retry_stats.get("check_mode_passed") or 0 - ) - stats["failed_apply_retry_check_mode_failed"] = int( - retry_stats.get("check_mode_failed") or 0 - ) - stats["failed_apply_retry_stage_receipt_written"] = int( - retry_stats.get("runtime_stage_receipt_written") or 0 - ) - if retry_stats.get("error") and not stats["error"]: - stats["error"] = retry_stats["error"] except Exception as exc: stats["error"] = f"{type(exc).__name__}: {exc}"[:500] logger.warning("awooop_ansible_candidate_backfill_once_failed", **stats) @@ -372,14 +403,24 @@ async def run_awooop_ansible_candidate_backfill_loop() -> None: await asyncio.sleep(settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_STARTUP_SLEEP_SECONDS) while True: + sleep_seconds = float( + settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS + ) try: result = await enqueue_missing_ansible_candidates_once( limit=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_BATCH_LIMIT, window_hours=settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WINDOW_HOURS, ) - if result.get("queued") or result.get("error"): + if result.get("error") or result.get("failed_apply_retry_blockers"): + sleep_seconds = _error_backoff_seconds() + if ( + result.get("queued") + or result.get("error") + or result.get("failed_apply_retry_priority_tick") + ): logger.info("awooop_ansible_candidate_backfill_worker_tick", **result) except Exception as exc: logger.warning("awooop_ansible_candidate_backfill_worker_failed", error=str(exc)) + sleep_seconds = _error_backoff_seconds() - await asyncio.sleep(settings.AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_INTERVAL_SECONDS) + await asyncio.sleep(sleep_seconds) diff --git a/apps/api/src/jobs/awooop_ansible_check_mode_job.py b/apps/api/src/jobs/awooop_ansible_check_mode_job.py index ddbf2477b..b4236fb84 100644 --- a/apps/api/src/jobs/awooop_ansible_check_mode_job.py +++ b/apps/api/src/jobs/awooop_ansible_check_mode_job.py @@ -9,6 +9,7 @@ when the dry-run passes. Critical / break-glass catalog rows still stay blocked. from __future__ import annotations import asyncio +import random import structlog @@ -16,6 +17,12 @@ from src.core.config import settings from src.services.awooop_ansible_check_mode_service import run_pending_check_modes_once logger = structlog.get_logger(__name__) +_ERROR_BACKOFF_MIN_SECONDS = 15.0 +_ERROR_BACKOFF_MAX_SECONDS = 30.0 + + +def _error_backoff_seconds() -> float: + return random.uniform(_ERROR_BACKOFF_MIN_SECONDS, _ERROR_BACKOFF_MAX_SECONDS) async def run_awooop_ansible_check_mode_loop() -> None: @@ -32,14 +39,18 @@ async def run_awooop_ansible_check_mode_loop() -> None: await asyncio.sleep(settings.AWOOOP_ANSIBLE_CHECK_MODE_STARTUP_SLEEP_SECONDS) while True: + sleep_seconds = float(settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS) try: result = await run_pending_check_modes_once( limit=settings.AWOOOP_ANSIBLE_CHECK_MODE_BATCH_LIMIT, timeout_seconds=settings.AWOOOP_ANSIBLE_CHECK_MODE_TIMEOUT_SECONDS, ) + if result.get("error") or result.get("catalog_replay_error"): + sleep_seconds = _error_backoff_seconds() if result.get("claimed") or result.get("blockers"): logger.info("awooop_ansible_check_mode_worker_tick", **result) except Exception as exc: logger.warning("awooop_ansible_check_mode_worker_failed", error=str(exc)) + sleep_seconds = _error_backoff_seconds() - await asyncio.sleep(settings.AWOOOP_ANSIBLE_CHECK_MODE_INTERVAL_SECONDS) + await asyncio.sleep(sleep_seconds) diff --git a/apps/api/src/services/ai_agent_autonomous_runtime_control.py b/apps/api/src/services/ai_agent_autonomous_runtime_control.py index 010fdfe78..f2e523a1d 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -236,22 +236,71 @@ def _runtime_receipt_readback_cache_get_stale( _runtime_receipt_readback_cache.pop(key, None) return None cached_readback = copy.deepcopy(readback) + identity_bearing_partial = ( + _runtime_receipt_readback_has_identity_bearing_partial(cached_readback) + ) cached_readback["cache_fallback_active"] = True - cached_readback["record_quality"] = "cached_verified_runtime_receipt_readback" + cached_readback["record_quality"] = ( + "cached_identity_bearing_partial_runtime_receipt_readback" + if identity_bearing_partial + else "cached_verified_runtime_receipt_readback" + ) cached_readback["runtime_receipt_cache_fallback"] = { "active": True, "reason": fallback_reason, "live_db_read_status": live_db_read_status, "live_error_type": live_error_type, + "cached_db_read_status": str(cached_readback.get("db_read_status") or ""), + "identity_bearing_partial": identity_bearing_partial, "cache_age_seconds": round(cache_age_seconds, 3), "max_stale_seconds": _RUNTIME_RECEIPT_READBACK_STALE_FALLBACK_TTL_SECONDS, - "source": "in_process_last_verified_db_receipt_readback", + "source": ( + "in_process_last_known_identity_runtime_receipt_readback" + if identity_bearing_partial + else "in_process_last_verified_db_receipt_readback" + ), "durable_receipts_reused": True, "writes_on_read": False, } return cached_readback +def _runtime_receipt_readback_has_identity_bearing_partial( + readback: Mapping[str, Any], +) -> bool: + """Return whether a partial read preserves durable same-run identity evidence.""" + + if str(readback.get("db_read_status") or "") != "partial": + return False + loop_ledger = readback.get("autonomous_execution_loop_ledger") + if not isinstance(loop_ledger, Mapping): + return False + automation_run_id = str(loop_ledger.get("automation_run_id") or "") + if not automation_run_id: + return False + + stage_receipts = loop_ledger.get("same_run_stage_receipts") + if isinstance(stage_receipts, list) and any( + isinstance(receipt, Mapping) + and str(receipt.get("automation_run_id") or "") == automation_run_id + and receipt.get("durable_receipt") is True + for receipt in stage_receipts + ): + return True + + stages = loop_ledger.get("stages") + return bool( + isinstance(stages, list) + and any( + isinstance(stage, Mapping) + and stage.get("present") is True + and stage.get("run_id_matches_expected") is True + and str(stage.get("run_id") or "") == automation_run_id + for stage in stages + ) + ) + + def _runtime_receipt_readback_is_cacheable( readback: Mapping[str, Any], ) -> bool: @@ -261,8 +310,15 @@ def _runtime_receipt_readback_is_cacheable( partial_query_failures = readback.get("partial_query_failures") return bool( db_read_status == "partial" - and isinstance(partial_query_failures, list) - and _has_only_auxiliary_runtime_receipt_failures(partial_query_failures) + and ( + ( + isinstance(partial_query_failures, list) + and _has_only_auxiliary_runtime_receipt_failures( + partial_query_failures + ) + ) + or _runtime_receipt_readback_has_identity_bearing_partial(readback) + ) ) diff --git a/apps/api/src/services/awoooi_priority_work_order_readback.py b/apps/api/src/services/awoooi_priority_work_order_readback.py index 950312e8a..4894d880d 100644 --- a/apps/api/src/services/awoooi_priority_work_order_readback.py +++ b/apps/api/src/services/awoooi_priority_work_order_readback.py @@ -7311,6 +7311,12 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: trust_boundary_runtime_closed = bool( summary.get("ai_agent_executor_trust_boundary_runtime_closed") is True ) + db_identity_isolated = bool( + summary.get("ai_agent_workload_db_identity_isolated") is True + ) + connection_budget_verified = bool( + summary.get("ai_agent_workload_db_connection_budget_verified") is True + ) trust_boundary_controls = { "api_read_only_identity_verified": boundary_verified, "worker_broker_separation_verified": boundary_verified, @@ -7318,6 +7324,8 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: capability_lifecycle_closed and capability_same_run ), "production_rbac_mount_verifier_verified": boundary_verified, + "workload_db_identity_isolated": db_identity_isolated, + "workload_db_connection_budget_verified": connection_budget_verified, } trust_boundary_present_count = sum(trust_boundary_controls.values()) trust_boundary_work_item["runtime_progress"] = { @@ -7328,7 +7336,19 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: "required_control_count": len(trust_boundary_controls), "controls": trust_boundary_controls, "missing": list( - summary.get("ai_agent_execution_capability_missing") or [] + dict.fromkeys( + [ + *list( + summary.get("ai_agent_execution_capability_missing") or [] + ), + *list( + summary.get( + "ai_agent_workload_db_identity_active_blockers" + ) + or [] + ), + ] + ) ), "runtime_closed": trust_boundary_runtime_closed, } @@ -7359,6 +7379,14 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: ), "same_run_correlation": capability_same_run, "production_boundary_verified": boundary_verified, + "db_identity_isolated": db_identity_isolated, + "connection_budget_verified": connection_budget_verified, + "db_identity_receipt_ref": str( + summary.get("ai_agent_workload_db_identity_receipt_ref") or "" + ), + "db_identity_verifier": str( + summary.get("ai_agent_workload_db_identity_verifier") or "" + ), } trust_boundary_work_item["runtime_evidence_refs"] = [ str( @@ -7369,6 +7397,10 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None: + str(summary.get("ai_agent_autonomous_runtime_automation_run_id") or ""), "capability:" + str(summary.get("ai_agent_execution_capability_op_id") or ""), + "workload-db-identity:" + + str( + summary.get("ai_agent_workload_db_identity_receipt_ref") or "" + ), ] elif boundary_verified or capability_lifecycle_closed: trust_boundary_work_item["status"] = "in_progress" @@ -7729,6 +7761,9 @@ def apply_ai_automation_live_closure_readbacks( executor_trust_boundary = _dict( runtime_control.get("executor_trust_boundary") ) + database_identity_boundary = _dict( + executor_trust_boundary.get("database_identity_boundary") + ) strict_runtime_closed = bool( strict_runtime.get("schema_version") == "ai_agent_strict_runtime_completion_v2" @@ -7793,8 +7828,18 @@ def apply_ai_automation_live_closure_readbacks( executor_trust_boundary.get("production_boundary_verified") is True and executor_trust_boundary.get("source_sha_matches_deployment") is True ) + db_identity_isolated = bool( + database_identity_boundary.get("db_identity_isolated") is True + ) + connection_budget_verified = bool( + database_identity_boundary.get("connection_budget_verified") is True + ) trust_boundary_closed = bool( - strict_runtime_closed and capability_same_run and boundary_verified + strict_runtime_closed + and capability_same_run + and boundary_verified + and db_identity_isolated + and connection_budget_verified ) summary["ai_agent_executor_trust_boundary_verified"] = boundary_verified summary["ai_agent_execution_capability_lifecycle_closed"] = ( @@ -7806,6 +7851,26 @@ def apply_ai_automation_live_closure_readbacks( summary["ai_agent_executor_trust_boundary_runtime_closed"] = ( trust_boundary_closed ) + summary["ai_agent_workload_db_identity_isolated"] = db_identity_isolated + summary["ai_agent_workload_db_connection_budget_verified"] = ( + connection_budget_verified + ) + summary["ai_agent_workload_db_identity_receipt_ref"] = str( + database_identity_boundary.get("receipt_ref") or "" + ) + summary["ai_agent_workload_db_identity_verifier"] = str( + database_identity_boundary.get("verifier") or "" + ) + database_identity_blockers = list( + database_identity_boundary.get("active_blockers") or [] + ) + if not db_identity_isolated: + database_identity_blockers.append("db_identity_isolated_not_verified") + if not connection_budget_verified: + database_identity_blockers.append("connection_budget_verified_not_verified") + summary["ai_agent_workload_db_identity_active_blockers"] = list( + dict.fromkeys(database_identity_blockers) + ) summary["ai_agent_executor_trust_boundary_receipt_ref"] = str( executor_trust_boundary.get("receipt_ref") or "" ) 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 3017604d6..c4fb4525f 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -3194,17 +3194,25 @@ async def run_pending_check_modes_once( stale_after_seconds=max(300, effective_timeout_seconds + 120), ) remaining_limit = max(0, limit - len(reclaimed_claims)) - catalog_replay_claims = ( - await claim_catalog_drift_failed_check_modes( - project_id=project_id, - limit=remaining_limit, - candidate_max_age_hours=( - settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS - ), - ) - if remaining_limit - else [] - ) + catalog_replay_claims: list[AnsibleCheckModeClaim] = [] + catalog_replay_error: str | None = None + if remaining_limit: + try: + catalog_replay_claims = await claim_catalog_drift_failed_check_modes( + project_id=project_id, + limit=remaining_limit, + candidate_max_age_hours=( + settings.AWOOOP_ANSIBLE_CHECK_MODE_CANDIDATE_MAX_AGE_HOURS + ), + ) + except Exception as exc: + catalog_replay_error = type(exc).__name__ + logger.warning( + "ansible_catalog_drift_replay_claim_failed", + project_id=project_id, + error_type=catalog_replay_error, + fresh_candidate_claim_continues=True, + ) remaining_limit = max(0, remaining_limit - len(catalog_replay_claims)) fresh_claims = ( await claim_pending_check_modes( @@ -3331,5 +3339,6 @@ async def run_pending_check_modes_once( "capability_revoked": capability_revoked, "capability_expired": expired_capability_count, "capability_revoke_failed": capability_revoke_failed, + "catalog_replay_error": catalog_replay_error, "blockers": [], } diff --git a/apps/api/src/services/executor_trust_boundary_readback.py b/apps/api/src/services/executor_trust_boundary_readback.py index d9d0fa604..5104099a9 100644 --- a/apps/api/src/services/executor_trust_boundary_readback.py +++ b/apps/api/src/services/executor_trust_boundary_readback.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import copy import os +import re import time from collections.abc import Mapping from typing import Any @@ -13,8 +14,8 @@ import structlog logger = structlog.get_logger(__name__) -SCHEMA_VERSION = "awoooi_executor_trust_boundary_readback_v2" -RECEIPT_SCHEMA_VERSION = "awoooi_executor_boundary_verification_v2" +SCHEMA_VERSION = "awoooi_executor_trust_boundary_readback_v3" +RECEIPT_SCHEMA_VERSION = "awoooi_executor_boundary_verification_v3" DEFAULT_NAMESPACE = "awoooi-prod" DEFAULT_CONFIGMAP_NAME = "awoooi-executor-boundary-verification" _CACHE_TTL_SECONDS = 20.0 @@ -32,9 +33,181 @@ _REQUIRED_TRUE_FIELDS = ( "broker_ssh_egress_allowlisted", "legacy_namespace_egress_allow_absent", ) +_DB_WORKLOAD_IDS = ("api", "worker", "broker") +_ROLE_FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{64}$") _cache: tuple[float, dict[str, Any]] | None = None +def _truthy(value: str | None) -> bool: + return str(value or "").lower() == "true" + + +def _int_or_none(value: str | None) -> int | None: + try: + return int(str(value)) + except (TypeError, ValueError): + return None + + +def _role_fingerprint(value: str | None) -> str | None: + normalized = str(value or "").strip().lower() + return normalized if _ROLE_FINGERPRINT_PATTERN.fullmatch(normalized) else None + + +def _build_database_identity_boundary( + receipt: Mapping[str, str], + *, + receipt_ref: str, +) -> dict[str, Any]: + workloads: dict[str, dict[str, Any]] = {} + for workload_id in _DB_WORKLOAD_IDS: + workloads[workload_id] = { + "database_secret_ref": ( + receipt.get(f"{workload_id}_database_secret_ref") or None + ), + "role_fingerprint": _role_fingerprint( + receipt.get(f"{workload_id}_db_role_fingerprint") + ), + "connection_limit": _int_or_none( + receipt.get(f"{workload_id}_db_connection_limit") + ), + "required_connection_budget": _int_or_none( + receipt.get(f"{workload_id}_required_connection_budget") + ), + "database_null_pool": _truthy( + receipt.get(f"{workload_id}_database_null_pool") + ), + "monolithic_secret_env_from_absent": _truthy( + receipt.get( + f"{workload_id}_monolithic_secret_env_from_absent" + ) + ), + } + + secret_refs = [ + str(workloads[workload_id]["database_secret_ref"] or "") + for workload_id in _DB_WORKLOAD_IDS + ] + role_fingerprints = [ + str(workloads[workload_id]["role_fingerprint"] or "") + for workload_id in _DB_WORKLOAD_IDS + ] + distinct_secret_refs = bool( + all(secret_refs) + and len(set(secret_refs)) == len(_DB_WORKLOAD_IDS) + and _truthy(receipt.get("distinct_database_secret_refs")) + ) + distinct_role_fingerprints = bool( + all(role_fingerprints) + and len(set(role_fingerprints)) == len(_DB_WORKLOAD_IDS) + and _truthy(receipt.get("distinct_db_role_fingerprints")) + ) + monolithic_env_from_absent = bool( + workloads["api"]["monolithic_secret_env_from_absent"] + and workloads["worker"]["monolithic_secret_env_from_absent"] + ) + null_pool_verified = all( + workloads[workload_id]["database_null_pool"] is True + for workload_id in _DB_WORKLOAD_IDS + ) + workload_budget_verified = all( + isinstance(workloads[workload_id]["connection_limit"], int) + and isinstance(workloads[workload_id]["required_connection_budget"], int) + and workloads[workload_id]["connection_limit"] >= workloads[workload_id][ + "required_connection_budget" + ] + and workloads[workload_id]["required_connection_budget"] > 0 + for workload_id in _DB_WORKLOAD_IDS + ) + db_identity_isolated = bool( + distinct_secret_refs + and distinct_role_fingerprints + and monolithic_env_from_absent + and _truthy(receipt.get("db_identity_isolated")) + ) + connection_budget_verified = bool( + db_identity_isolated + and null_pool_verified + and workload_budget_verified + and _truthy(receipt.get("connection_budget_verified")) + ) + + blockers: list[str] = [] + if not workloads["api"]["monolithic_secret_env_from_absent"]: + blockers.append("api_monolithic_secret_env_from_present") + if not workloads["worker"]["monolithic_secret_env_from_absent"]: + blockers.append("worker_monolithic_secret_env_from_present") + if not distinct_secret_refs: + blockers.append("database_secret_refs_not_distinct") + if not distinct_role_fingerprints: + blockers.append("database_role_fingerprints_not_distinct") + for workload_id in _DB_WORKLOAD_IDS: + workload = workloads[workload_id] + if workload["role_fingerprint"] is None: + blockers.append(f"{workload_id}_database_role_fingerprint_unavailable") + if workload["database_null_pool"] is not True: + blockers.append(f"{workload_id}_database_null_pool_not_verified") + connection_limit = workload["connection_limit"] + required_budget = workload["required_connection_budget"] + if ( + not isinstance(connection_limit, int) + or not isinstance(required_budget, int) + or required_budget <= 0 + or connection_limit < required_budget + ): + blockers.append(f"{workload_id}_connection_budget_not_verified") + if not db_identity_isolated: + blockers.append("db_identity_isolated_not_verified") + if not connection_budget_verified: + blockers.append("connection_budget_verified_not_verified") + + observed_limits = [ + workloads[workload_id]["connection_limit"] + for workload_id in _DB_WORKLOAD_IDS + ] + required_budgets = [ + workloads[workload_id]["required_connection_budget"] + for workload_id in _DB_WORKLOAD_IDS + ] + return { + "schema_version": "awoooi_workload_db_identity_budget_readback_v1", + "status": "verified_ready" if not blockers else "in_progress", + "receipt_ref": receipt_ref, + "verifier": receipt.get("db_identity_verifier") or None, + "db_identity_isolated": db_identity_isolated, + "connection_budget_verified": connection_budget_verified, + "workloads": workloads, + "controls": { + "api_monolithic_secret_env_from_absent": workloads["api"][ + "monolithic_secret_env_from_absent" + ], + "worker_monolithic_secret_env_from_absent": workloads["worker"][ + "monolithic_secret_env_from_absent" + ], + "distinct_database_secret_refs": distinct_secret_refs, + "distinct_db_role_fingerprints": distinct_role_fingerprints, + "database_null_pool_verified": null_pool_verified, + "workload_connection_budgets_verified": workload_budget_verified, + }, + "connection_budget": { + "observed_total": ( + sum(observed_limits) + if all(isinstance(value, int) for value in observed_limits) + else None + ), + "required_total": ( + sum(required_budgets) + if all(isinstance(value, int) for value in required_budgets) + else None + ), + }, + "active_blockers": blockers, + "writes_on_read": False, + "secret_values_read": False, + "role_names_exposed": False, + } + + def build_executor_trust_boundary_readback( data: Mapping[str, Any] | None, *, @@ -44,6 +217,7 @@ def build_executor_trust_boundary_readback( error_type: str | None = None, ) -> dict[str, Any]: receipt = {str(key): str(value) for key, value in (data or {}).items()} + receipt_ref = f"configmap:{namespace}/{configmap_name}" verified_source_sha = receipt.get("verified_source_sha", "").strip().lower() normalized_deployed_sha = deployed_source_sha.strip().lower() blockers: list[str] = [] @@ -60,12 +234,30 @@ def build_executor_trust_boundary_readback( blockers.append(f"{field}_not_verified") ready = not blockers + database_identity_boundary = _build_database_identity_boundary( + receipt, + receipt_ref=receipt_ref, + ) + full_workload_boundary_verified = bool( + ready + and database_identity_boundary["db_identity_isolated"] is True + and database_identity_boundary["connection_budget_verified"] is True + ) return { "schema_version": SCHEMA_VERSION, - "status": "verified_ready" if ready else "degraded", + "status": ( + "verified_ready" + if full_workload_boundary_verified + else ( + "network_boundary_verified_database_identity_in_progress" + if ready + else "degraded" + ) + ), "production_boundary_verified": ready, + "full_workload_boundary_verified": full_workload_boundary_verified, "namespace": namespace, - "receipt_ref": f"configmap:{namespace}/{configmap_name}", + "receipt_ref": receipt_ref, "receipt_schema_version": receipt.get("schema_version") or None, "verified_source_sha": verified_source_sha or None, "deployed_source_sha": normalized_deployed_sha or None, @@ -85,7 +277,12 @@ def build_executor_trust_boundary_readback( field: receipt.get(field, "").lower() == "true" for field in _REQUIRED_TRUE_FIELDS }, + "database_identity_boundary": database_identity_boundary, "active_blockers": blockers, + "full_boundary_active_blockers": [ + *blockers, + *database_identity_boundary["active_blockers"], + ], "writes_on_read": False, "secret_values_read": False, } diff --git a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py index f43df369a..298a61398 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -852,6 +852,89 @@ async def test_live_runtime_receipt_readback_uses_last_verified_cache_when_db_de assert fallback["writes_on_read"] is False +@pytest.mark.asyncio +async def test_live_runtime_receipt_preserves_identity_partial_when_db_unavailable( + monkeypatch, +): + runtime_control_module._clear_runtime_receipt_readback_cache() + calls = 0 + + async def _fake_uncached_loader( + *, + project_id: str, + lookback_hours: int, + limit: int, + ) -> dict: + nonlocal calls + calls += 1 + if calls == 1: + readback = build_runtime_receipt_readback_from_rows( + project_id=project_id, + lookback_hours=lookback_hours, + db_read_status="partial", + partial_query_failures=[ + {"query_name": "km_latest", "error_type": "QueryCanceled"}, + ], + error_type="partial_query_failures", + ) + readback["autonomous_execution_loop_ledger"] = { + "automation_run_id": "run-partial-94", + "same_run_stage_receipts": [ + { + "stage_id": "sensor_source_receipt", + "automation_run_id": "run-partial-94", + "durable_receipt": True, + } + ], + "stages": [], + } + return readback + return build_runtime_receipt_readback_from_rows( + project_id=project_id, + lookback_hours=lookback_hours, + db_read_status="unavailable", + error_type="RuntimeReceiptDbContextTimeout", + ) + + monkeypatch.setattr( + runtime_control_module, + "_load_ai_agent_autonomous_runtime_receipt_readback_uncached", + _fake_uncached_loader, + ) + + try: + first = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback() + cache_key = ("awoooi", 24, 20) + stored_at, cached_readback = ( + runtime_control_module._runtime_receipt_readback_cache[cache_key] + ) + runtime_control_module._runtime_receipt_readback_cache[cache_key] = ( + stored_at - 21.0, + cached_readback, + ) + second = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback() + finally: + runtime_control_module._clear_runtime_receipt_readback_cache() + + assert calls == 2 + assert first["db_read_status"] == "partial" + assert second["db_read_status"] == "partial" + assert second["autonomous_execution_loop_ledger"]["automation_run_id"] == ( + "run-partial-94" + ) + assert second["cache_fallback_active"] is True + assert second["record_quality"] == ( + "cached_identity_bearing_partial_runtime_receipt_readback" + ) + fallback = second["runtime_receipt_cache_fallback"] + assert fallback["live_db_read_status"] == "unavailable" + assert fallback["cached_db_read_status"] == "partial" + assert fallback["identity_bearing_partial"] is True + assert fallback["source"] == ( + "in_process_last_known_identity_runtime_receipt_readback" + ) + + @pytest.mark.asyncio async def test_live_runtime_receipt_lock_timeout_uses_last_verified_cache( monkeypatch, diff --git a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py index 01d11cd13..0f42dd4b7 100644 --- a/apps/api/tests/test_awoooi_priority_work_order_readback_api.py +++ b/apps/api/tests/test_awoooi_priority_work_order_readback_api.py @@ -126,14 +126,26 @@ def _autonomous_runtime_control_trust_boundary_closed() -> dict: } } payload["executor_trust_boundary"] = { - "schema_version": "awoooi_executor_trust_boundary_readback_v1", + "schema_version": "awoooi_executor_trust_boundary_readback_v3", "status": "verified_ready", "production_boundary_verified": True, + "full_workload_boundary_verified": True, "source_sha_matches_deployment": True, "verified_source_sha": "source-p0-011", "receipt_ref": ( "configmap:awoooi-prod/awoooi-executor-boundary-verification" ), + "database_identity_boundary": { + "schema_version": "awoooi_workload_db_identity_budget_readback_v1", + "status": "verified_ready", + "receipt_ref": ( + "configmap:awoooi-prod/awoooi-executor-boundary-verification" + ), + "verifier": "runtime_role_fingerprint_and_secret_projection_v1", + "db_identity_isolated": True, + "connection_budget_verified": True, + "active_blockers": [], + }, } return payload @@ -1970,11 +1982,48 @@ def test_ai_automation_trust_boundary_runtime_receipts_advance_order() -> None: assert items["AIA-P0-011"]["completion_evidence"][ "capability_op_id" ] == "capability-p0-011" + assert items["AIA-P0-011"]["completion_evidence"][ + "db_identity_isolated" + ] is True + assert items["AIA-P0-011"]["completion_evidence"][ + "connection_budget_verified" + ] is True + assert items["AIA-P0-011"]["runtime_progress"]["required_control_count"] == 6 assert program["summary"]["next_work_item_id"] == "AIA-P0-002" assert program["summary"]["done_count"] == 2 assert program["summary"]["completion_percent"] == 10 +def test_ai_automation_trust_boundary_requires_db_identity_and_budget_receipt() -> None: + payload = load_latest_awoooi_priority_work_order_readback() + runtime_control = _autonomous_runtime_control_trust_boundary_closed() + runtime_control["executor_trust_boundary"].pop("database_identity_boundary") + runtime_control["executor_trust_boundary"][ + "full_workload_boundary_verified" + ] = False + + apply_ai_automation_live_closure_readbacks( + payload, + autonomous_runtime_control=runtime_control, + ) + + program = payload["ai_automation_program_ledger"] + items = {item["id"]: item for item in program["work_items"]} + trust_boundary = items["AIA-P0-011"] + assert items["AIA-P0-001"]["status"] == "done" + assert trust_boundary["status"] == "in_progress" + assert trust_boundary["runtime_progress"]["runtime_closed"] is False + assert trust_boundary["runtime_progress"]["controls"][ + "workload_db_identity_isolated" + ] is False + assert trust_boundary["runtime_progress"]["controls"][ + "workload_db_connection_budget_verified" + ] is False + assert program["summary"]["next_work_item_id"] == "AIA-P0-011" + assert program["summary"]["done_count"] == 1 + assert program["summary"]["completion_percent"] == 5 + + def test_awoooi_priority_work_order_readback_endpoint_returns_snapshot( monkeypatch: pytest.MonkeyPatch, ): diff --git a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py index 813ea65ac..1d7ebf1a3 100644 --- a/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py +++ b/apps/api/tests/test_awooop_ansible_candidate_backfill_job.py @@ -81,16 +81,20 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo return True async def fake_receipt_backfiller(**kwargs): + call_order.append("receipt") assert kwargs["project_id"] == "awoooi" return {"written": 2, "error": None} async def fake_retry_replayer(**kwargs): + call_order.append("retry") assert kwargs["project_id"] == "awoooi" return { - "replayed": 1, - "check_mode_passed": 1, + "scanned": 0, + "replayed": 0, + "check_mode_passed": 0, "check_mode_failed": 0, - "runtime_stage_receipt_written": 1, + "runtime_stage_receipt_written": 0, + "blockers": [], "error": None, } @@ -111,18 +115,94 @@ async def test_backfill_enqueues_catalog_matched_incident(monkeypatch: pytest.Mo assert result["queued"] == 1 assert result["no_catalog_candidate"] == 0 assert result["repair_receipts_backfilled"] == 2 - assert result["failed_apply_retry_replayed"] == 1 - assert result["failed_apply_retry_check_mode_passed"] == 1 + assert result["failed_apply_retry_replayed"] == 0 + assert result["failed_apply_retry_check_mode_passed"] == 0 assert result["failed_apply_retry_check_mode_failed"] == 0 - assert result["failed_apply_retry_stage_receipt_written"] == 1 + assert result["failed_apply_retry_stage_receipt_written"] == 0 + assert result["failed_apply_retry_blockers"] == [] + assert result["failed_apply_retry_priority_tick"] is False assert result["pre_decision_evidence_ready"] == 1 assert result["pre_decision_evidence_blocked"] == 0 - assert call_order == ["evidence", "verify", "record"] + assert call_order == ["retry", "receipt", "evidence", "verify", "record"] assert recorded[0]["decision_path"] == "repair_candidate_controlled_queue" assert recorded[0]["incident"]["incident_id"] == "INC-20260627-NODE110" assert recorded[0]["automation_run_id"] +@pytest.mark.asyncio +async def test_backfill_prioritizes_open_failed_apply_retry_before_fresh_candidates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + async def fake_retry_replayer(**kwargs): + assert kwargs["project_id"] == "awoooi" + return { + "scanned": 1, + "replayed": 1, + "check_mode_passed": 1, + "check_mode_failed": 0, + "runtime_stage_receipt_written": 1, + "blockers": [], + "error": None, + } + + async def fresh_scan_must_not_run(**_kwargs): + raise AssertionError("fresh candidates must wait for the open retry tick") + + monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True) + monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan_must_not_run) + receipt_backfiller = AsyncMock(return_value={"written": 0, "error": None}) + recorder = AsyncMock(return_value=True) + + result = await job.enqueue_missing_ansible_candidates_once( + retry_replayer=fake_retry_replayer, + receipt_backfiller=receipt_backfiller, + recorder=recorder, + ) + + assert result["failed_apply_retry_scanned"] == 1 + assert result["failed_apply_retry_replayed"] == 1 + assert result["failed_apply_retry_check_mode_passed"] == 1 + assert result["failed_apply_retry_stage_receipt_written"] == 1 + assert result["failed_apply_retry_priority_tick"] is True + assert result["queued"] == 0 + receipt_backfiller.assert_not_awaited() + recorder.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_backfill_surfaces_retry_blockers_without_queueing_fresh_work( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + async def fake_retry_replayer(**_kwargs): + return { + "scanned": 0, + "replayed": 0, + "blockers": ["database_connection_budget_exhausted"], + "error": None, + } + + async def fresh_scan_must_not_run(**_kwargs): + raise AssertionError("fresh candidates must not grow behind retry blockers") + + monkeypatch.setattr(job.settings, "ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER", True) + monkeypatch.setattr(job, "_fetch_missing_candidate_incidents", fresh_scan_must_not_run) + + result = await job.enqueue_missing_ansible_candidates_once( + retry_replayer=fake_retry_replayer, + receipt_backfiller=AsyncMock(return_value={"written": 0, "error": None}), + ) + + assert result["failed_apply_retry_blockers"] == [ + "database_connection_budget_exhausted" + ] + assert result["failed_apply_retry_priority_tick"] is True + assert result["queued"] == 0 + + @pytest.mark.asyncio async def test_backfill_retries_automatically_when_predecision_evidence_is_incomplete( monkeypatch: pytest.MonkeyPatch, @@ -224,3 +304,12 @@ def test_backfill_query_excludes_existing_ansible_candidate_rows() -> None: assert "NOT EXISTS" in joined assert "ansible_candidate_matched" in joined assert "resolved_at IS NULL" in joined + + +def test_backfill_worker_error_backoff_is_short_and_bounded() -> None: + from src.jobs import awooop_ansible_candidate_backfill_job as job + + delay = job._error_backoff_seconds() + + assert job._ERROR_BACKOFF_MIN_SECONDS <= delay <= job._ERROR_BACKOFF_MAX_SECONDS + assert job._ERROR_BACKOFF_MAX_SECONDS < 60 diff --git a/apps/api/tests/test_awooop_truth_chain_service.py b/apps/api/tests/test_awooop_truth_chain_service.py index 0e9e5b588..ac98891c0 100644 --- a/apps/api/tests/test_awooop_truth_chain_service.py +++ b/apps/api/tests/test_awooop_truth_chain_service.py @@ -1854,6 +1854,67 @@ def test_failed_check_mode_catalog_drift_replays_once() -> None: assert "claim_catalog_drift_failed_check_modes" in run_source +@pytest.mark.asyncio +async def test_catalog_drift_query_failure_does_not_block_fresh_candidate_claims( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from src.services import awooop_ansible_check_mode_service as service + + fresh_claim_attempted = False + + async def no_expired_capabilities(**_kwargs): + return 0 + + async def no_transport_blockers(**_kwargs): + return [] + + async def no_stale_claims(**_kwargs): + return [] + + async def failed_catalog_replay(**_kwargs): + raise RuntimeError("catalog replay query canceled") + + async def fresh_claims_continue(**_kwargs): + nonlocal fresh_claim_attempted + fresh_claim_attempted = True + return [] + + monkeypatch.setattr( + service, + "_expire_stale_ansible_execution_capabilities", + no_expired_capabilities, + ) + 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_stale_claims) + monkeypatch.setattr( + service, + "claim_catalog_drift_failed_check_modes", + failed_catalog_replay, + ) + monkeypatch.setattr(service, "claim_pending_check_modes", fresh_claims_continue) + + result = await service.run_pending_check_modes_once(limit=1) + + assert fresh_claim_attempted is True + assert result["claimed"] == 0 + assert result["catalog_replayed"] == 0 + assert result["catalog_replay_error"] == "RuntimeError" + + +def test_check_mode_worker_error_backoff_is_short_and_bounded() -> None: + from src.jobs import awooop_ansible_check_mode_job as job + + delay = job._error_backoff_seconds() + + assert job._ERROR_BACKOFF_MIN_SECONDS <= delay <= job._ERROR_BACKOFF_MAX_SECONDS + assert job._ERROR_BACKOFF_MAX_SECONDS < 60 + + def test_ansible_apply_receipt_backfill_queries_existing_apply_rows() -> None: source = inspect.getsource(backfill_missing_auto_repair_execution_receipts_once) diff --git a/apps/api/tests/test_executor_trust_boundary_readback.py b/apps/api/tests/test_executor_trust_boundary_readback.py index 09a56e4a4..3a5fd8bc1 100644 --- a/apps/api/tests/test_executor_trust_boundary_readback.py +++ b/apps/api/tests/test_executor_trust_boundary_readback.py @@ -25,6 +25,29 @@ def _verified_receipt(source_sha: str = "abc123") -> dict[str, str]: "worker_ssh_egress_denied": "true", "broker_ssh_egress_allowlisted": "true", "legacy_namespace_egress_allow_absent": "true", + "db_identity_verifier": "runtime_role_fingerprint_and_secret_projection_v1", + "api_database_secret_ref": "awoooi-api-db", + "worker_database_secret_ref": "awoooi-worker-db", + "broker_database_secret_ref": "awoooi-broker-db", + "api_db_role_fingerprint": "a" * 64, + "worker_db_role_fingerprint": "b" * 64, + "broker_db_role_fingerprint": "c" * 64, + "api_db_connection_limit": "2", + "worker_db_connection_limit": "2", + "broker_db_connection_limit": "2", + "api_required_connection_budget": "1", + "worker_required_connection_budget": "1", + "broker_required_connection_budget": "1", + "api_database_null_pool": "true", + "worker_database_null_pool": "true", + "broker_database_null_pool": "true", + "api_monolithic_secret_env_from_absent": "true", + "worker_monolithic_secret_env_from_absent": "true", + "broker_monolithic_secret_env_from_absent": "true", + "distinct_database_secret_refs": "true", + "distinct_db_role_fingerprints": "true", + "db_identity_isolated": "true", + "connection_budget_verified": "true", } @@ -36,9 +59,19 @@ def test_executor_trust_boundary_requires_live_controls_and_matching_source() -> assert readback["status"] == "verified_ready" assert readback["production_boundary_verified"] is True + assert readback["full_workload_boundary_verified"] is True assert readback["source_sha_matches_deployment"] is True assert readback["active_blockers"] == [] assert readback["secret_values_read"] is False + database_boundary = readback["database_identity_boundary"] + assert database_boundary["db_identity_isolated"] is True + assert database_boundary["connection_budget_verified"] is True + assert database_boundary["connection_budget"] == { + "observed_total": 6, + "required_total": 3, + } + assert database_boundary["secret_values_read"] is False + assert database_boundary["role_names_exposed"] is False def test_executor_trust_boundary_rejects_stale_or_incomplete_receipt() -> None: @@ -57,3 +90,39 @@ def test_executor_trust_boundary_rejects_stale_or_incomplete_receipt() -> None: ) assert incomplete["production_boundary_verified"] is False assert "worker_ssh_egress_denied_not_verified" in incomplete["active_blockers"] + + +def test_executor_trust_boundary_keeps_shared_database_identity_in_progress() -> None: + shared = _verified_receipt() + for workload_id in ("api", "worker", "broker"): + shared[f"{workload_id}_database_secret_ref"] = "awoooi-secrets" + shared[f"{workload_id}_db_role_fingerprint"] = "d" * 64 + shared["api_monolithic_secret_env_from_absent"] = "false" + shared["worker_monolithic_secret_env_from_absent"] = "false" + shared["distinct_database_secret_refs"] = "false" + shared["distinct_db_role_fingerprints"] = "false" + shared["db_identity_isolated"] = "false" + shared["connection_budget_verified"] = "false" + + readback = build_executor_trust_boundary_readback( + shared, + deployed_source_sha="abc123", + ) + + assert readback["production_boundary_verified"] is True + assert readback["full_workload_boundary_verified"] is False + assert readback["status"] == ( + "network_boundary_verified_database_identity_in_progress" + ) + database_boundary = readback["database_identity_boundary"] + assert database_boundary["db_identity_isolated"] is False + assert database_boundary["connection_budget_verified"] is False + assert "api_monolithic_secret_env_from_present" in database_boundary[ + "active_blockers" + ] + assert "database_secret_refs_not_distinct" in database_boundary[ + "active_blockers" + ] + assert "database_role_fingerprints_not_distinct" in database_boundary[ + "active_blockers" + ] diff --git a/apps/api/tests/test_signal_worker_ansible_executor_binding.py b/apps/api/tests/test_signal_worker_ansible_executor_binding.py index 685b98533..1474d753e 100644 --- a/apps/api/tests/test_signal_worker_ansible_executor_binding.py +++ b/apps/api/tests/test_signal_worker_ansible_executor_binding.py @@ -86,6 +86,7 @@ def test_api_manifest_has_read_only_identity_and_no_ssh_executor_mounts() -> Non assert pod_spec["serviceAccountName"] == "awoooi-api" assert env["SSH_MCP_ENABLED"] == "false" + assert env["DATABASE_NULL_POOL"] == "true" assert env["ENABLE_AWOOOP_ANSIBLE_CHECK_MODE_WORKER"] == "false" assert env["ENABLE_AWOOOP_ANSIBLE_CANDIDATE_BACKFILL_WORKER"] == "false" assert {"repair-ssh-key", "repair-known-hosts", "ssh-mcp-key"}.isdisjoint( @@ -154,9 +155,17 @@ def test_cd_prunes_and_verifies_api_executor_boundary_in_production() -> None: assert "repair-ssh-key|repair-known-hosts|ssh-mcp-key" in workflow assert "legacy cluster-wide executor binding still exists" in workflow assert "awoooi-executor-boundary-verification" in workflow - assert "awoooi_executor_boundary_verification_v2" in workflow + assert "awoooi_executor_boundary_verification_v3" in workflow assert "executor_boundary_public_readback=verified_ready" in workflow assert "verified_source_sha" in workflow + assert "runtime_role_fingerprint_and_secret_projection_v1" in workflow + assert "db_identity_isolated" in workflow + assert "connection_budget_verified" in workflow + assert "api_monolithic_secret_env_from_absent" in workflow + assert "distinct_database_secret_refs" in workflow + assert "distinct_db_role_fingerprints" in workflow + assert "current_user AS role_name" in workflow + assert "hashlib.sha256" in workflow assert "argocd.argoproj.io/sync-options=Prune=false" in workflow assert "argocd.argoproj.io/compare-options=IgnoreExtraneous" in workflow assert "github.com/kubernetes-sigs/kustomize" not in workflow diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 92891c4c2..c9dedd722 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -92,6 +92,10 @@ spec: value: "1" - name: DATABASE_MAX_OVERFLOW value: "0" + - name: DATABASE_NULL_POOL + # 2026-07-11 Codex: release each API connection after use while + # workload-specific DB identities are being split and verified. + value: "true" - name: IWOOOS_WAZUH_READONLY_ENABLED # 2026-06-30 Codex: controlled GitOps enablement after owner # metadata, manager registry acceptance, dry-run, rollback, and