diff --git a/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql b/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql new file mode 100644 index 000000000..a6f1a8da9 --- /dev/null +++ b/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql @@ -0,0 +1,14 @@ +-- awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql +-- Bound the autonomous runtime Telegram latest-receipt query without widening +-- its 700ms statement timeout or changing the receipt eligibility contract. +-- +-- CONCURRENTLY must run outside a transaction block. + +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_awooop_outbound_msg_telegram_receipt_latest + ON awooop_outbound_message (project_id, queued_at DESC) + INCLUDE (send_status) + WHERE channel_type = 'telegram' + AND ( + source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' + OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent' + ); diff --git a/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16_down.sql b/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16_down.sql new file mode 100644 index 000000000..943cf6d96 --- /dev/null +++ b/apps/api/migrations/awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16_down.sql @@ -0,0 +1,6 @@ +-- awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16_down.sql +-- Roll back only the Telegram receipt hot-path index owned by this migration. +-- +-- CONCURRENTLY must run outside a transaction block. + +DROP INDEX CONCURRENTLY IF EXISTS idx_awooop_outbound_msg_telegram_receipt_latest; diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index 8dc4e771b..e1a1d09f1 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -31,7 +31,7 @@ import json from typing import Any from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, status -from fastapi.responses import StreamingResponse +from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel, Field from src.core.config import settings @@ -1359,6 +1359,12 @@ async def post_agent99_telegram_lifecycle( finally: clear_project_context(context_tokens) if receipt.get("ok") is not True: + if receipt.get("delivery_status") == "pending_unknown": + return JSONResponse( + status_code=status.HTTP_202_ACCEPTED, + content=receipt, + headers={"Retry-After": "2"}, + ) raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=( diff --git a/apps/api/src/jobs/baseline_snapshot.py b/apps/api/src/jobs/baseline_snapshot.py index 9ef33df12..2ef85f2f5 100644 --- a/apps/api/src/jobs/baseline_snapshot.py +++ b/apps/api/src/jobs/baseline_snapshot.py @@ -32,14 +32,16 @@ import json from datetime import timedelta import structlog -from sqlalchemy import func, select, text +from sqlalchemy import func, select from src.core.redis_client import get_redis from src.db.base import get_db_context from src.db.models import ( + AlertOperationLog, AutoRepairExecution, IncidentRecord, KnowledgeEntryRecord, + MCPAuditLog, ) from src.utils.timezone import now_taipei @@ -117,18 +119,15 @@ async def _count_mcp_calls_24h(since_24h) -> int: """ MCP 呼叫次數/24h。 - Phase 0：無 MCP Calls Table → 從 audit_logs 嘗試計數。 - Phase 1 建立 PreDecisionInvestigator 後，此處改為查 mcp_tool_calls 表。 + Count the production MCP audit ledger. ``audit_logs`` is the legacy + operation-result schema and has no generic ``action`` column. """ try: async with get_db_context() as db: - # audit_logs 中 action='mcp_call' — Phase 0 預期 0 筆 result = await db.execute( - text( - "SELECT COUNT(*) FROM audit_logs " - "WHERE action = 'mcp_call' AND created_at >= :since" - ), - {"since": since_24h}, + select(func.count(MCPAuditLog.id)).where( + MCPAuditLog.created_at >= since_24h + ) ) return result.scalar_one_or_none() or 0 except Exception: @@ -277,12 +276,11 @@ async def _db_metrics(since_24h) -> dict: ) metrics["learning_writes_24h"] = r.scalar_one_or_none() or 0 - # audit_logs 24h 計數（Phase 0 預期 = 0） + # Immutable alert-operation audit events in the last 24 hours. r = await db.execute( - text( - "SELECT COUNT(*) FROM audit_logs WHERE created_at >= :since" - ), - {"since": since_24h}, + select(func.count(AlertOperationLog.id)).where( + AlertOperationLog.created_at >= since_24h + ) ) metrics["audit_logs_24h"] = r.scalar_one_or_none() or 0 diff --git a/apps/api/src/services/agent99_telegram_lifecycle.py b/apps/api/src/services/agent99_telegram_lifecycle.py index 8df6831e0..c2c0d4ae0 100644 --- a/apps/api/src/services/agent99_telegram_lifecycle.py +++ b/apps/api/src/services/agent99_telegram_lifecycle.py @@ -118,11 +118,26 @@ async def deliver_agent99_telegram_lifecycle( isinstance(response, dict) and response.get("_awooop_provider_send_performed") is True ) + initial_result = response.get("result") if isinstance(response, dict) else None + provider_message_id_hint = ( + str(initial_result.get("message_id") or "") + if isinstance(initial_result, dict) + else "" + ) + initial_delivery_context = ( + response.get("_awooop_delivery_context") + if isinstance(response, dict) + and isinstance(response.get("_awooop_delivery_context"), dict) + else {} + ) + provider_destination_verified = bool( + initial_delivery_context.get("destination_binding_verified") is True + ) reconciliation_attempts = 0 # A provider send can finish before the durable outbox finalizer becomes - # readable. Re-entering with the same delivery identity only re-reads the - # send-once reservation; it cannot send a second provider message. + # readable. Reconciliation only reads/finalizes the exact existing row; + # it never re-enters the provider sender. for delay_seconds in _DURABLE_RECONCILE_DELAYS_SECONDS: delivery_status = ( str(response.get("_awooop_delivery_status") or "") @@ -133,17 +148,22 @@ async def deliver_agent99_telegram_lifecycle( break await asyncio.sleep(delay_seconds) reconciliation_attempts += 1 - response = await gateway.send_agent99_lifecycle_receipt(**send_kwargs) - provider_send_performed = bool( - provider_send_performed - or ( - isinstance(response, dict) - and response.get("_awooop_provider_send_performed") is True - ) + response = await gateway.reconcile_agent99_lifecycle_receipt( + delivery_id=payload.delivery_id, + incident_id=payload.incident_id, + state_key=payload.state_key, + project_id=payload.project_id, + expected_provider_message_id=( + provider_message_id_hint or None + ), ) result = response.get("result") if isinstance(response, dict) else None - message_id = str(result.get("message_id") or "") if isinstance(result, dict) else "" + message_id = ( + str(result.get("message_id") or "") + if isinstance(result, dict) + else "" + ) or provider_message_id_hint delivery_context = ( response.get("_awooop_delivery_context") if isinstance(response, dict) @@ -161,6 +181,7 @@ async def deliver_agent99_telegram_lifecycle( ) destination_verified = bool( delivery_status == "sent_reused" + or provider_destination_verified or delivery_context.get("destination_binding_verified") is True ) ok = bool( 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 a06f9fc5b..5e1ef4106 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -8673,7 +8673,7 @@ _RUNTIME_TELEGRAM_COUNTS_DIRECT_SQL = """ _RUNTIME_TELEGRAM_LATEST_SQL = """ SELECT message_id::text AS message_id, - run_id::text AS outbound_run_id, + run_id::text AS run_id, COALESCE( source_envelope ->> 'automation_run_id', source_envelope #>> '{callback_reply,automation_run_id}', @@ -8707,7 +8707,7 @@ _RUNTIME_TELEGRAM_LATEST_SQL = """ _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL = """ SELECT message_id::text AS message_id, - run_id::text AS outbound_run_id, + run_id::text AS run_id, COALESCE( source_envelope ->> 'automation_run_id', source_envelope #>> '{callback_reply,automation_run_id}', diff --git a/apps/api/src/services/audit_sink.py b/apps/api/src/services/audit_sink.py index 69d5dda01..c4c4ce9b0 100644 --- a/apps/api/src/services/audit_sink.py +++ b/apps/api/src/services/audit_sink.py @@ -9,7 +9,7 @@ AwoooP Phase 4.4: Audit log 寫入前的 sanitization pipeline（ADR-116） - PII / secret pattern 硬攔（不可被 caller 繞過） - 攔截清單：GCP IP、PostgreSQL password、Telegram token、SSH key、Bearer token 等 - redaction 後原值不可還原（替換為 [REDACTED:]） -- 所有 audit 寫入透過此 sink（禁止其他 service 直接 INSERT audit_logs） +- 所有 generic audit 寫入透過此 sink（禁止直接假設 audit_logs 新版 schema） 使用： from src.services.audit_sink import write_audit @@ -34,6 +34,23 @@ import structlog logger = structlog.get_logger(__name__) +AWOOOP_AUDIT_SCHEMA_VERSION = "awooop_sanitized_audit_v1" +AWOOOP_AUDIT_REDACTION_VERSION = "audit_sink_v1" + + +class AuditDurabilityError(RuntimeError): + """Fail-closed error for a required immutable audit acknowledgement.""" + + def __init__(self, error_code: str) -> None: + self.error_code = error_code + super().__init__(error_code) + + +def audit_idempotency_key_sha256(idempotency_key: str) -> str: + """Return the non-secret stable identity stored in immutable context.""" + + return hashlib.sha256(idempotency_key.encode("utf-8")).hexdigest() + # ───────────────────────────────────────────────────────────────────────────── # Redaction patterns（ADR-116 P1-08） @@ -149,27 +166,33 @@ async def write_audit( details: dict[str, Any] | None = None, run_id: str | None = None, trace_id: str | None = None, -) -> None: + require_durable: bool = False, + idempotency_key: str | None = None, +) -> dict[str, Any] | None: """ 統一 audit log 寫入入口（Phase 4+ 所有 service 必須透過此方法）。 1. sanitize details（PII / secret redaction） 2. 附加 run_id / trace_id（可觀測性） - 3. INSERT audit_logs（非阻擋 background task） - """ - import asyncio + 3. INSERT immutable alert_operation_log USER_ACTION/context event - asyncio.create_task( - _write_audit_impl( - project_id=project_id, - action=action, - resource_type=resource_type, - resource_id=resource_id, - details=details, - run_id=run_id, - trace_id=trace_id, - ), - name="audit_sink_write", + Critical authorization callers set ``require_durable=True`` and provide an + idempotency key. That path is serialized with a transaction-scoped + PostgreSQL advisory lock and returns only after the DB context commits. + """ + if require_durable and not str(idempotency_key or "").strip(): + raise AuditDurabilityError("audit_idempotency_key_required") + + return await _write_audit_impl( + project_id=project_id, + action=action, + resource_type=resource_type, + resource_id=resource_id, + details=details, + run_id=run_id, + trace_id=trace_id, + require_durable=require_durable, + idempotency_key=idempotency_key, ) @@ -182,40 +205,168 @@ async def _write_audit_impl( details: dict[str, Any] | None, run_id: str | None, trace_id: str | None, -) -> None: + require_durable: bool = False, + idempotency_key: str | None = None, +) -> dict[str, Any] | None: try: from sqlalchemy import text as sa_text + from src.db.base import get_db_context clean_details: dict[str, Any] = sanitize(details or {}) - if run_id: - clean_details["_run_id"] = run_id - if trace_id: - clean_details["_trace_id"] = trace_id + clean_action = _redact_string(str(action or "unknown"))[:200] + clean_resource_type = _redact_string(str(resource_type or "unknown"))[:80] + clean_resource_id = _redact_string(str(resource_id or "unknown"))[:200] + clean_run_id = _redact_string(str(run_id or ""))[:200] + clean_trace_id = _redact_string(str(trace_id or ""))[:200] + clean_project_id = _redact_string(str(project_id or ""))[:64] + actor = _redact_string( + str(clean_details.get("approver_id") or f"awooop:{clean_project_id}") + )[:100] + idempotency_key_value = str(idempotency_key or "").strip() + idempotency_key_sha256 = ( + audit_idempotency_key_sha256(idempotency_key_value) + if idempotency_key_value + else "" + ) + context = { + "schema_version": AWOOOP_AUDIT_SCHEMA_VERSION, + "redaction_version": AWOOOP_AUDIT_REDACTION_VERSION, + "project_id": clean_project_id, + "action": clean_action, + "resource_type": clean_resource_type, + "resource_id": clean_resource_id, + "run_id": clean_run_id, + "trace_id": clean_trace_id, + "details": clean_details, + "durable_write_required": require_durable, + "idempotency_key_sha256": idempotency_key_sha256, + } + params = { + "actor": actor, + "action_detail": clean_action, + "context": json.dumps(context, ensure_ascii=False), + } async with get_db_context(project_id) as db: + if idempotency_key_value: + await db.execute( + sa_text(""" + SELECT pg_advisory_xact_lock( + hashtextextended(:idempotency_key, 0) + ) + """), + {"idempotency_key": idempotency_key_value}, + ) + existing_result = await db.execute( + sa_text(""" + SELECT COUNT(*) AS matching_count, + MAX(created_at) AS created_at + FROM alert_operation_log + WHERE event_type = 'USER_ACTION' + AND action_detail = :action_detail + AND context ->> 'schema_version' = :schema_version + AND context ->> 'project_id' = :project_id + AND context ->> 'resource_type' = :resource_type + AND context ->> 'resource_id' = :resource_id + AND context ->> 'run_id' = :run_id + AND context ->> 'idempotency_key_sha256' + = :idempotency_key_sha256 + """), + { + "action_detail": clean_action, + "schema_version": AWOOOP_AUDIT_SCHEMA_VERSION, + "project_id": clean_project_id, + "resource_type": clean_resource_type, + "resource_id": clean_resource_id, + "run_id": clean_run_id, + "idempotency_key_sha256": idempotency_key_sha256, + }, + ) + existing = existing_result.mappings().one() + existing_count = int(existing["matching_count"] or 0) + if existing_count > 1: + raise AuditDurabilityError( + "audit_idempotency_duplicate_detected" + ) + if existing_count == 1: + return { + "status": "duplicate_existing", + "durable_write_ack": True, + "inserted": False, + "created_at": existing["created_at"], + } + await db.execute( sa_text(""" - INSERT INTO audit_logs - (project_id, action, resource_type, resource_id, details) + INSERT INTO alert_operation_log + (event_type, actor, action_detail, success, context) VALUES - (:project_id, :action, :resource_type, :resource_id, CAST(:details AS JSONB)) + ( + 'USER_ACTION', + :actor, + :action_detail, + TRUE, + CAST(:context AS JSONB) + ) """), - { - "project_id": project_id, - "action": action, - "resource_type": resource_type, - "resource_id": resource_id, - "details": json.dumps(clean_details), - }, + params, ) + if idempotency_key_value: + readback_result = await db.execute( + sa_text(""" + SELECT COUNT(*) AS matching_count, + MAX(created_at) AS created_at + FROM alert_operation_log + WHERE event_type = 'USER_ACTION' + AND action_detail = :action_detail + AND context ->> 'schema_version' = :schema_version + AND context ->> 'project_id' = :project_id + AND context ->> 'resource_type' = :resource_type + AND context ->> 'resource_id' = :resource_id + AND context ->> 'run_id' = :run_id + AND context ->> 'idempotency_key_sha256' + = :idempotency_key_sha256 + """), + { + "action_detail": clean_action, + "schema_version": AWOOOP_AUDIT_SCHEMA_VERSION, + "project_id": clean_project_id, + "resource_type": clean_resource_type, + "resource_id": clean_resource_id, + "run_id": clean_run_id, + "idempotency_key_sha256": idempotency_key_sha256, + }, + ) + readback = readback_result.mappings().one() + if int(readback["matching_count"] or 0) != 1: + raise AuditDurabilityError( + "audit_durable_readback_not_exactly_once" + ) + return { + "status": "inserted_verified", + "durable_write_ack": True, + "inserted": True, + "created_at": readback["created_at"], + } + return { + "status": "inserted", + "durable_write_ack": True, + "inserted": True, + "created_at": None, + } + except AuditDurabilityError: + raise except Exception as exc: logger.warning( "audit_sink_write_failed", action=action, resource_id=resource_id, - error=str(exc), + error_type=type(exc).__name__, ) + if require_durable: + raise AuditDurabilityError("audit_durable_write_failed") from exc + return None # ───────────────────────────────────────────────────────────────────────────── diff --git a/apps/api/src/services/contract_service.py b/apps/api/src/services/contract_service.py index 5a56bdf74..4c36cd249 100644 --- a/apps/api/src/services/contract_service.py +++ b/apps/api/src/services/contract_service.py @@ -26,7 +26,7 @@ from __future__ import annotations import hashlib import hmac import json -from datetime import datetime, timezone +from datetime import UTC, datetime from typing import Any from uuid import UUID @@ -37,6 +37,7 @@ from src.core.config import settings from src.db.awooop_models import AwoooPContractRevision from src.models.awooop_contracts import validate_contract_body from src.repositories import contract_repository +from src.services.audit_sink import write_audit as write_sanitized_audit logger = structlog.get_logger(__name__) @@ -123,9 +124,9 @@ def _verify_publish_signature( ) return True - message = f"{revision_id}:{body_hash}:{publisher_id}".encode("utf-8") + message = f"{revision_id}:{body_hash}:{publisher_id}".encode() expected = hmac.new( - secret.encode("utf-8"), message, hashlib.sha256 + secret.encode(), message, hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected, signature) @@ -270,7 +271,7 @@ async def publish( ): raise ContractSignatureError() - published_at = datetime.now(timezone.utc) + published_at = datetime.now(UTC) revision = await contract_repository.mark_published( revision_id=revision_id, project_id=project_id, @@ -420,30 +421,19 @@ async def _write_audit( resource_id: str, details: dict[str, Any], ) -> None: - """寫入 audit_log（非阻擋，失敗只 warning）""" + """Project the contract event into the immutable generic audit ledger.""" try: - from sqlalchemy import text as sa_text - from src.db.base import get_db_context - async with get_db_context(project_id) as db: - await db.execute( - sa_text(""" - INSERT INTO audit_logs - (project_id, action, resource_type, resource_id, details) - VALUES - (:project_id, :action, :resource_type, :resource_id, CAST(:details AS JSONB)) - """), - { - "project_id": project_id, - "action": action, - "resource_type": resource_type, - "resource_id": resource_id, - "details": json.dumps(details), - }, - ) + await write_sanitized_audit( + project_id=project_id, + action=action, + resource_type=resource_type, + resource_id=resource_id, + details=details, + ) except Exception as exc: logger.warning( "contract_audit_write_failed", action=action, resource_id=resource_id, - error=str(exc), + error_type=type(exc).__name__, ) diff --git a/apps/api/src/services/paid_provider_canary_authorization.py b/apps/api/src/services/paid_provider_canary_authorization.py index da7ab465c..42530703b 100644 --- a/apps/api/src/services/paid_provider_canary_authorization.py +++ b/apps/api/src/services/paid_provider_canary_authorization.py @@ -27,6 +27,10 @@ from sqlalchemy import func, select, text from src.db.awooop_models import AwoooPRunState, AwoooPRunStepJournal from src.db.base import get_db_context +from src.services.audit_sink import ( + AWOOOP_AUDIT_SCHEMA_VERSION, + audit_idempotency_key_sha256, +) logger = structlog.get_logger(__name__) @@ -37,11 +41,25 @@ PAID_CANARY_TRIGGER_TYPE = "api" PAID_CANARY_TRIGGER_REF = "paid-provider-canary:AIA-SRE-013" PAID_CANARY_APPROVAL_STEP = "operator_console.approve" PAID_CANARY_APPROVAL_AUDIT_ACTION = "run.approval.approve" +PAID_CANARY_APPROVAL_AUDIT_IDEMPOTENCY_PREFIX = ( + "awooop:paid-provider-canary:approval:v1" +) PAID_CANARY_START_ACTION = "claude_gemini_sanitized_sre_canary" PAID_CANARY_APPROVAL_TTL_SECONDS = 900 PAID_CANARY_PERCENT = 5 _SAFE_TRACE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{0,191}$") + +def paid_provider_canary_approval_audit_idempotency_key( + run_id: UUID | str, +) -> str: + """Return the exact single-run lock/deduplication identity.""" + + return ( + f"{PAID_CANARY_APPROVAL_AUDIT_IDEMPOTENCY_PREFIX}:" + f"{PAID_CANARY_PROJECT_ID}:{run_id}" + ) + _PAID_CANARY_AUTHORIZATION_INPUT: dict[str, Any] = { "schema_version": "paid_provider_canary_authorization_input_v1", "project_id": PAID_CANARY_PROJECT_ID, @@ -175,29 +193,44 @@ class _SqlPaidProviderCanaryAuthorizationEvidenceReader: ) step_row = step_result.one() - # ``audit_logs`` is the durable sink used by decide_approval(). - # Query it directly because the legacy ORM model does not expose - # the AwoooP action/resource/details columns present in production. + approval_idempotency_key = ( + paid_provider_canary_approval_audit_idempotency_key(run_id) + ) audit_result = await db.execute( text( """ SELECT COUNT(*) AS approval_audit_count, MAX(created_at) AS approval_audit_created_at - FROM audit_logs - WHERE project_id = :project_id - AND action = :action - AND resource_type = 'run' - AND resource_id = :run_id - AND details ->> '_run_id' = :run_id - AND details ->> 'decision' = 'approve' - AND details ->> 'new_state' = 'running' - AND COALESCE(details ->> 'approver_id', '') <> '' + FROM alert_operation_log + WHERE event_type = 'USER_ACTION' + AND action_detail = :action + AND success IS TRUE + AND COALESCE(actor, '') <> '' + AND context ->> 'schema_version' = :schema_version + AND context ->> 'project_id' = :project_id + AND context ->> 'resource_type' = 'run' + AND context ->> 'resource_id' = :run_id + AND context ->> 'run_id' = :run_id + AND context ->> 'trace_id' = :trace_id + AND context ->> 'idempotency_key_sha256' + = :idempotency_key_sha256 + AND context #>> '{details,decision}' = 'approve' + AND context #>> '{details,new_state}' = 'running' + AND COALESCE( + context #>> '{details,approver_id}', + '' + ) <> '' """ ), { "project_id": project_id, "action": PAID_CANARY_APPROVAL_AUDIT_ACTION, "run_id": str(run_id), + "trace_id": str(run.trace_id or ""), + "schema_version": AWOOOP_AUDIT_SCHEMA_VERSION, + "idempotency_key_sha256": audit_idempotency_key_sha256( + approval_idempotency_key + ), }, ) audit_row = audit_result.mappings().one() diff --git a/apps/api/src/services/platform_operator_service.py b/apps/api/src/services/platform_operator_service.py index edd3b8957..7a9fa9038 100644 --- a/apps/api/src/services/platform_operator_service.py +++ b/apps/api/src/services/platform_operator_service.py @@ -45,7 +45,7 @@ from src.services.ai_agent_result_capture_owner_release_approval_gate import ( from src.services.ai_agent_result_capture_release_verifier_preflight_gate import ( load_latest_ai_agent_result_capture_release_verifier_preflight_gate, ) -from src.services.audit_sink import write_audit +from src.services.audit_sink import AuditDurabilityError, write_audit from src.services.awooop_ansible_audit_service import summarize_ansible_execution from src.services.awooop_approval_token import issue_approval_token, record_approval from src.services.awooop_truth_chain_service import ( @@ -81,6 +81,9 @@ from src.services.operator_summary_cache import ( get_cached_operator_summary_async, store_operator_summary_async, ) +from src.services.paid_provider_canary_authorization import ( + paid_provider_canary_approval_audit_idempotency_key, +) from src.services.paid_provider_canary_gate5_run import ( paid_provider_canary_approval_contract_errors, ) @@ -8873,6 +8876,7 @@ async def decide_approval( paid_canary_contract_errors = ( paid_provider_canary_approval_contract_errors(run) ) + is_paid_provider_canary = paid_canary_contract_errors is not None if paid_canary_contract_errors: raise HTTPException( status_code=status.HTTP_409_CONFLICT, @@ -8882,6 +8886,7 @@ async def decide_approval( ), ) is_projection_only_gate5 = run.trigger_type == _ADR100_GATE5_PROJECTION_TRIGGER + run_trace_id = str(run.trace_id or "") approval_token_jti: str | None = None new_state: str @@ -8987,22 +8992,63 @@ async def decide_approval( reason=reason, ) - try: - await write_audit( - project_id=project_id, - action=f"run.approval.{decision}", - resource_type="run", - resource_id=run_id, - details={ - "approver_id": approver_id, - "decision": decision, - "reason": reason, - "new_state": new_state, - }, - run_id=run_id, - ) - except Exception as exc: - logger.warning("approval_audit_write_failed", run_id=run_id, error=str(exc)) + audit_kwargs = { + "project_id": project_id, + "action": f"run.approval.{decision}", + "resource_type": "run", + "resource_id": run_id, + "details": { + "approver_id": approver_id, + "decision": decision, + "reason": reason, + "new_state": new_state, + }, + "run_id": run_id, + "trace_id": run_trace_id, + } + if is_paid_provider_canary and decision == "approve": + try: + audit_receipt = await write_audit( + **audit_kwargs, + require_durable=True, + idempotency_key=( + paid_provider_canary_approval_audit_idempotency_key(run_id) + ), + ) + except AuditDurabilityError as exc: + logger.error( + "critical_approval_audit_write_failed", + run_id=run_id, + error_code=exc.error_code, + ) + await _terminalize_paid_canary_audit_failure( + run_id=run_uuid, + project_id=project_id, + error_code=exc.error_code, + ) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="critical_approval_durable_audit_unavailable", + ) from exc + if not audit_receipt or audit_receipt.get("durable_write_ack") is not True: + await _terminalize_paid_canary_audit_failure( + run_id=run_uuid, + project_id=project_id, + error_code="audit_durable_ack_missing", + ) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="critical_approval_durable_audit_unavailable", + ) + else: + try: + await write_audit(**audit_kwargs) + except Exception as exc: + logger.warning( + "approval_audit_write_failed", + run_id=run_id, + error_type=type(exc).__name__, + ) return { "run_id": run_id, @@ -9012,6 +9058,33 @@ async def decide_approval( } +async def _terminalize_paid_canary_audit_failure( + *, + run_id: UUID, + project_id: str, + error_code: str, +) -> None: + """Release the active Gate 5 slot when its durable approval audit fails.""" + + try: + await transition( + run_id, + project_id, + "failed", + error_code="E-PAID-CANARY-AUDIT", + error_detail=( + "critical approval durable audit unavailable; " + f"safe_code={error_code}" + ), + ) + except Exception as exc: + logger.error( + "critical_approval_audit_terminalize_failed", + run_id=str(run_id), + error_type=type(exc).__name__, + ) + + async def _record_approval_projection_guard_step( *, run_id: UUID, diff --git a/apps/api/src/services/telegram_gateway.py b/apps/api/src/services/telegram_gateway.py index 92d399ffd..734f0e758 100644 --- a/apps/api/src/services/telegram_gateway.py +++ b/apps/api/src/services/telegram_gateway.py @@ -6169,6 +6169,152 @@ class TelegramGateway: await asyncio.sleep(0.05 * (attempt + 1)) return False + async def _read_controlled_apply_result_outbound( + self, + *, + identity: dict[str, str], + ) -> dict[str, object]: + """Read one exact durable send-once row without reserving or sending.""" + from sqlalchemy import text + + from src.db.base import get_db_context + + delivery_run_id = _controlled_apply_result_delivery_run_id(identity) + try: + async with get_db_context(identity["project_id"]) as db: + result = await db.execute( + text(""" + SELECT + message_id::text AS message_id, + send_status, + provider_message_id, + ( + source_envelope #>> + '{visual_delivery,requested}' + ) = 'true' AS visual_requested + FROM awooop_outbound_message + WHERE project_id = :project_id + AND run_id = CAST(:run_id AS uuid) + AND channel_type = 'telegram' + AND source_envelope #>> + '{callback_reply,action}' = :delivery_kind + AND source_envelope #>> + '{callback_reply,project_id}' = :project_id + AND source_envelope #>> + '{callback_reply,automation_run_id}' = + :automation_run_id + AND source_envelope #>> + '{callback_reply,incident_id}' = :incident_id + AND source_envelope #>> + '{callback_reply,apply_op_id}' = :apply_op_id + ORDER BY queued_at ASC + LIMIT 1 + """), + { + "project_id": identity["project_id"], + "run_id": str(delivery_run_id), + "delivery_kind": identity["delivery_kind"], + "automation_run_id": identity["automation_run_id"], + "incident_id": identity["incident_id"], + "apply_op_id": identity["apply_op_id"], + }, + ) + row = result.mappings().one_or_none() + except Exception as exc: + logger.warning( + "controlled_apply_result_durable_readback_failed", + automation_run_id=identity["automation_run_id"], + incident_id=identity["incident_id"], + apply_op_id=identity["apply_op_id"], + error_type=type(exc).__name__, + ) + return { + "status": "durable_readback_failed", + "run_id": str(delivery_run_id), + } + + if row is None: + return { + "status": "reservation_missing", + "run_id": str(delivery_run_id), + } + + readback = dict(row) + message_id = str(readback.get("message_id") or "") + provider_message_id = str( + readback.get("provider_message_id") or "" + ) + send_status = str(readback.get("send_status") or "") + if send_status == "sent" and provider_message_id: + return { + "status": "sent", + "run_id": str(delivery_run_id), + "message_id": message_id, + "provider_message_id": provider_message_id, + "visual_requested": readback.get("visual_requested") is True, + } + if send_status == "shadow": + return { + "status": "suppressed", + "run_id": str(delivery_run_id), + "message_id": message_id, + } + if send_status == "pending" and message_id: + return { + "status": "pending_unknown", + "run_id": str(delivery_run_id), + "message_id": message_id, + } + return { + "status": "durable_state_invalid", + "run_id": str(delivery_run_id), + "message_id": message_id, + } + + async def _reconcile_controlled_apply_result_outbound( + self, + *, + identity: dict[str, str], + expected_provider_message_id: str | None = None, + ) -> dict[str, object]: + """Finalize/read one existing row; never reserve or contact Telegram.""" + readback = await self._read_controlled_apply_result_outbound( + identity=identity + ) + provider_message_id = str(expected_provider_message_id or "").strip() + if readback.get("status") == "sent": + durable_provider_message_id = str( + readback.get("provider_message_id") or "" + ) + if ( + provider_message_id + and durable_provider_message_id != provider_message_id + ): + return { + **readback, + "status": "provider_message_id_mismatch", + } + return readback + if readback.get("status") != "pending_unknown" or not provider_message_id: + return readback + + await self._finalize_controlled_apply_result_outbound( + identity=identity, + reservation=readback, + provider_message_id=provider_message_id, + ) + terminal_readback = await self._read_controlled_apply_result_outbound( + identity=identity + ) + if terminal_readback.get("status") == "sent" and str( + terminal_readback.get("provider_message_id") or "" + ) != provider_message_id: + return { + **terminal_readback, + "status": "provider_message_id_mismatch", + } + return terminal_readback + async def _send_request( self, method: str, @@ -11782,6 +11928,52 @@ class TelegramGateway: payload["reply_to_message_id"] = reply_to_message_id return await self._send_request(method, payload) + async def reconcile_agent99_lifecycle_receipt( + self, + *, + delivery_id: str, + incident_id: str, + state_key: str, + project_id: str = "awoooi", + expected_provider_message_id: str | None = None, + ) -> dict[str, object]: + """Read/finalize one lifecycle outbox row without provider egress.""" + identity = { + "project_id": project_id or "awoooi", + "automation_run_id": delivery_id, + "incident_id": incident_id, + "apply_op_id": state_key, + "delivery_kind": "agent99_lifecycle", + } + readback = await self._reconcile_controlled_apply_result_outbound( + identity=identity, + expected_provider_message_id=expected_provider_message_id, + ) + delivery_status = str(readback.get("status") or "unknown") + provider_message_id = str( + readback.get("provider_message_id") or "" + ) + sent = delivery_status == "sent" and bool(provider_message_id) + return { + "ok": sent, + "result": ( + {"message_id": provider_message_id} if provider_message_id else {} + ), + "_awooop_outbound_mirror_acknowledged": sent, + "_awooop_delivery_status": ( + "sent_reused" if sent else delivery_status + ), + "_awooop_provider_send_performed": False, + "_awooop_visual_delivery_verified": bool( + sent and readback.get("visual_requested") is True + ), + "_awooop_delivery_context": { + "destination_binding_verified": sent, + "durable_exact_row_readback": True, + }, + "_awooop_durable_readback_performed": True, + } + async def send_controlled_apply_result_receipt( self, *, diff --git a/apps/api/tests/test_agent99_telegram_lifecycle_api.py b/apps/api/tests/test_agent99_telegram_lifecycle_api.py index 6800fca14..20f1fe3a4 100644 --- a/apps/api/tests/test_agent99_telegram_lifecycle_api.py +++ b/apps/api/tests/test_agent99_telegram_lifecycle_api.py @@ -5,6 +5,7 @@ import base64 import hashlib import os import struct +from unittest.mock import AsyncMock os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test") @@ -124,7 +125,7 @@ def test_lifecycle_endpoint_rejects_wrong_token(monkeypatch) -> None: assert response.json()["detail"] == "agent99_lifecycle_auth_failed" -def test_lifecycle_endpoint_requires_durable_gateway_ack(monkeypatch) -> None: +def test_lifecycle_endpoint_returns_structured_pending_receipt(monkeypatch) -> None: monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") async def failed_delivery(request): @@ -145,8 +146,36 @@ def test_lifecycle_endpoint_requires_durable_gateway_ack(monkeypatch) -> None: json=payload(), ) + assert response.status_code == 202 + assert response.headers["retry-after"] == "2" + assert response.json()["ok"] is False + assert response.json()["delivery_status"] == "pending_unknown" + assert response.json()["delivery_id"] == payload()["delivery_id"] + + +def test_lifecycle_endpoint_fails_closed_for_non_pending_delivery(monkeypatch) -> None: + monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected") + + async def failed_delivery(request): + return { + "ok": False, + "delivery_id": request.delivery_id, + "delivery_status": "durable_readback_failed", + } + + monkeypatch.setattr( + agents_api, + "deliver_agent99_telegram_lifecycle", + failed_delivery, + ) + response = app_client().post( + "/api/v1/agents/agent99/telegram-lifecycle", + headers={"X-Agent99-Lifecycle-Token": "expected"}, + json=payload(), + ) + assert response.status_code == 503 - assert "pending_unknown" in response.json()["detail"] + assert "durable_readback_failed" in response.json()["detail"] def test_lifecycle_endpoint_returns_same_delivery_receipt(monkeypatch) -> None: @@ -220,6 +249,47 @@ def test_lifecycle_outbox_identity_is_stable_and_separate() -> None: ) != gateway_service._controlled_apply_result_delivery_run_id(controlled_identity) +@pytest.mark.asyncio +async def test_lifecycle_durable_readback_binds_exact_identity_without_egress( + monkeypatch, +) -> None: + gateway = object.__new__(gateway_service.TelegramGateway) + reconcile = AsyncMock( + return_value={ + "status": "sent", + "provider_message_id": "8124", + "visual_requested": True, + } + ) + monkeypatch.setattr( + gateway, + "_reconcile_controlled_apply_result_outbound", + reconcile, + ) + + result = await gateway.reconcile_agent99_lifecycle_receipt( + delivery_id=payload()["delivery_id"], + incident_id=payload()["incident_id"], + state_key=payload()["state_key"], + expected_provider_message_id="8124", + ) + + assert result["ok"] is True + assert result["_awooop_delivery_status"] == "sent_reused" + assert result["_awooop_provider_send_performed"] is False + assert result["_awooop_durable_readback_performed"] is True + reconcile.assert_awaited_once_with( + identity={ + "project_id": "awoooi", + "automation_run_id": payload()["delivery_id"], + "incident_id": payload()["incident_id"], + "apply_op_id": payload()["state_key"], + "delivery_kind": "agent99_lifecycle", + }, + expected_provider_message_id="8124", + ) + + @pytest.mark.asyncio async def test_lifecycle_service_uses_durable_canonical_sender(monkeypatch) -> None: calls: list[dict] = [] @@ -261,34 +331,37 @@ async def test_lifecycle_service_uses_durable_canonical_sender(monkeypatch) -> N async def test_lifecycle_reconciles_pending_send_without_duplicate_provider_message( monkeypatch, ) -> None: - calls: list[dict] = [] + send_calls: list[dict] = [] + reconcile_calls: list[dict] = [] sleeps: list[float] = [] - responses = [ - { - "ok": True, - "result": {"message_id": 9126}, - "_awooop_outbound_mirror_acknowledged": False, - "_awooop_delivery_status": "pending_unknown", - "_awooop_provider_send_performed": True, - "_awooop_visual_delivery_verified": True, - "_awooop_delivery_context": { - "destination_binding_verified": True, - }, + initial_response = { + "ok": True, + "result": {"message_id": 9126}, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "pending_unknown", + "_awooop_provider_send_performed": True, + "_awooop_visual_delivery_verified": True, + "_awooop_delivery_context": { + "destination_binding_verified": True, }, - { - "ok": True, - "result": {"message_id": 9126}, - "_awooop_outbound_mirror_acknowledged": True, - "_awooop_delivery_status": "sent_reused", - "_awooop_provider_send_performed": False, - "_awooop_visual_delivery_verified": True, - }, - ] + } + reconciled_response = { + "ok": True, + "result": {"message_id": 9126}, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "sent_reused", + "_awooop_provider_send_performed": False, + "_awooop_visual_delivery_verified": True, + } class Gateway: async def send_agent99_lifecycle_receipt(self, **kwargs): - calls.append(kwargs) - return responses.pop(0) + send_calls.append(kwargs) + return initial_response + + async def reconcile_agent99_lifecycle_receipt(self, **kwargs): + reconcile_calls.append(kwargs) + return reconciled_response async def no_sleep(delay: float) -> None: sleeps.append(delay) @@ -313,32 +386,106 @@ async def test_lifecycle_reconciles_pending_send_without_duplicate_provider_mess assert receipt["durable_reconciliation_attempts"] == 1 assert receipt["visual_sent"] is True assert sleeps == [0.25] - assert len(calls) == 2 - assert calls[0] == calls[1] - assert calls[0]["delivery_id"] == payload()["delivery_id"] + assert len(send_calls) == 1 + assert len(reconcile_calls) == 1 + assert send_calls[0]["delivery_id"] == payload()["delivery_id"] + assert reconcile_calls[0] == { + "delivery_id": payload()["delivery_id"], + "incident_id": payload()["incident_id"], + "state_key": payload()["state_key"], + "project_id": "awoooi", + "expected_provider_message_id": "9126", + } + + +@pytest.mark.asyncio +async def test_lifecycle_follower_only_reads_inflight_durable_delivery( + monkeypatch, +) -> None: + send_calls: list[dict] = [] + reconcile_calls: list[dict] = [] + + class Gateway: + async def send_agent99_lifecycle_receipt(self, **kwargs): + send_calls.append(kwargs) + return { + "ok": False, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "pending_unknown", + "_awooop_provider_send_performed": False, + } + + async def reconcile_agent99_lifecycle_receipt(self, **kwargs): + reconcile_calls.append(kwargs) + return { + "ok": True, + "result": {"message_id": 9128}, + "_awooop_outbound_mirror_acknowledged": True, + "_awooop_delivery_status": "sent_reused", + "_awooop_provider_send_performed": False, + "_awooop_delivery_context": { + "destination_binding_verified": True, + "durable_exact_row_readback": True, + }, + } + + async def no_sleep(_delay: float) -> None: + return None + + monkeypatch.setattr( + lifecycle_service, + "get_telegram_gateway", + lambda: Gateway(), + ) + monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep) + + receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle( + Agent99TelegramLifecycleRequest.model_validate(payload()) + ) + + assert receipt["ok"] is True + assert receipt["provider_message_id"] == "9128" + assert receipt["provider_send_performed"] is False + assert len(send_calls) == 1 + assert len(reconcile_calls) == 1 + assert reconcile_calls[0]["expected_provider_message_id"] is None @pytest.mark.asyncio async def test_lifecycle_reconciliation_fails_closed_after_bounded_readbacks( monkeypatch, ) -> None: - calls: list[dict] = [] + send_calls: list[dict] = [] + reconcile_calls: list[dict] = [] sleeps: list[float] = [] class Gateway: async def send_agent99_lifecycle_receipt(self, **kwargs): - calls.append(kwargs) + send_calls.append(kwargs) return { "ok": True, "result": {"message_id": 9127}, "_awooop_outbound_mirror_acknowledged": False, "_awooop_delivery_status": "pending_unknown", - "_awooop_provider_send_performed": len(calls) == 1, + "_awooop_provider_send_performed": True, "_awooop_delivery_context": { "destination_binding_verified": True, }, } + async def reconcile_agent99_lifecycle_receipt(self, **kwargs): + reconcile_calls.append(kwargs) + return { + "ok": False, + "_awooop_outbound_mirror_acknowledged": False, + "_awooop_delivery_status": "pending_unknown", + "_awooop_provider_send_performed": False, + "_awooop_delivery_context": { + "destination_binding_verified": False, + "durable_exact_row_readback": True, + }, + } + async def no_sleep(delay: float) -> None: sleeps.append(delay) @@ -358,8 +505,12 @@ async def test_lifecycle_reconciliation_fails_closed_after_bounded_readbacks( assert receipt["provider_send_performed"] is True assert receipt["durable_reconciliation_attempts"] == 3 assert sleeps == [0.25, 0.5, 1.0] - assert len(calls) == 4 - assert all(call == calls[0] for call in calls) + assert len(send_calls) == 1 + assert len(reconcile_calls) == 3 + assert all( + call["expected_provider_message_id"] == "9127" + for call in reconcile_calls + ) @pytest.mark.asyncio 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 0d5cf28fc..97e729ddd 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -3449,6 +3449,82 @@ def test_runtime_telegram_receipt_queries_include_alert_notifications() -> None: assert "controlled_apply_result" in _RUNTIME_TELEGRAM_LATEST_SQL +def test_telegram_latest_dbapi_timeout_cannot_false_green_same_run_closure() -> None: + automation_run_id = "run-telegram-latest-timeout" + candidate_op_id = "11111111-1111-4111-8111-111111111111" + check_op_id = "22222222-2222-4222-8222-222222222222" + apply_op_id = "33333333-3333-4333-8333-333333333333" + incident_id = "INC-TELEGRAM-LATEST-TIMEOUT" + readback = build_runtime_receipt_readback_from_rows( + db_read_status="partial", + operation_latest_rows=[ + { + "op_id": candidate_op_id, + "operation_type": "ansible_candidate_matched", + "status": "dry_run", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + }, + { + "op_id": check_op_id, + "parent_op_id": candidate_op_id, + "operation_type": "ansible_check_mode_executed", + "status": "success", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + }, + { + "op_id": apply_op_id, + "parent_op_id": check_op_id, + "operation_type": "ansible_apply_executed", + "status": "success", + "automation_run_id": automation_run_id, + "source_candidate_op_id": candidate_op_id, + "check_mode_op_id": check_op_id, + "incident_id": incident_id, + "returncode": "0", + }, + ], + verifier_latest_rows=[ + { + "id": "verifier-telegram-timeout", + "apply_op_id": apply_op_id, + "automation_run_id": automation_run_id, + "verification_result": "success", + } + ], + km_latest_rows=[ + { + "id": "km-telegram-timeout", + "path_type": f"ansible_apply_receipt:{apply_op_id[:8]}", + "automation_run_id": automation_run_id, + "status": "linked", + } + ], + telegram_count_rows=[ + {"send_status": "sent", "total": 1, "recent": 1}, + ], + telegram_latest_rows=[], + partial_query_failures=[ + {"query_name": "telegram_latest", "error_type": "DBAPIError"}, + ], + error_type="partial_query_failures", + ) + + closure = readback["latest_flow_closure"] + assert closure["has_post_apply_verifier"] is True + assert closure["has_km_writeback"] is True + assert closure["has_telegram_receipt"] is False + assert closure["missing"] == ["telegram_receipt"] + assert closure["closed"] is False + assert readback["runtime_receipt_readback_recovery"]["status"] == ( + "degraded_runtime_receipt_partial_query_failures" + ) + strict = build_ai_agent_strict_runtime_completion_from_readback(readback) + assert strict["latest_flow_closed"] is False + assert strict["closed"] is False + + def test_runtime_operation_latest_queries_prioritize_latest_apply_chain() -> None: for sql in (_RUNTIME_OPERATION_LATEST_SQL, _RUNTIME_OPERATION_LATEST_DIRECT_SQL): assert "WITH latest_apply_chain AS" in sql diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index be6000ce4..7d72d19bd 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -815,6 +815,111 @@ async def test_controlled_result_pending_reservation_never_resends_provider( finalize.assert_awaited_once() +@pytest.mark.asyncio +async def test_durable_reconcile_finalizes_exact_pending_row_without_provider_send( + monkeypatch: pytest.MonkeyPatch, +) -> None: + pending_row = { + "message_id": "00000000-0000-0000-0000-000000000202", + "send_status": "pending", + "provider_message_id": None, + "visual_requested": True, + } + sent_row = { + **pending_row, + "send_status": "sent", + "provider_message_id": "456", + } + db = _SequenceDB( + _MappingResult(row=pending_row), + _MappingResult(), + _MappingResult( + row={"send_status": "sent", "provider_message_id": "456"} + ), + _MappingResult(row=sent_row), + ) + + @asynccontextmanager + async def fake_get_db_context(_project_id): + yield db + + monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) + gateway = TelegramGateway() + gateway._http_client = AsyncMock() + identity = { + "project_id": "awoooi", + "automation_run_id": "agent99-lifecycle-exact-readback", + "incident_id": "INC-20260716-AG99ACK", + "apply_op_id": "recovered|info", + "delivery_kind": "agent99_lifecycle", + } + + result = await gateway._reconcile_controlled_apply_result_outbound( + identity=identity, + expected_provider_message_id="456", + ) + + assert result == { + "status": "sent", + "run_id": str( + telegram_gateway_module._controlled_apply_result_delivery_run_id( + identity + ) + ), + "message_id": pending_row["message_id"], + "provider_message_id": "456", + "visual_requested": True, + } + gateway._http_client.post.assert_not_awaited() + assert len(db.statements) == 4 + assert "{callback_reply,automation_run_id}" in db.statements[0] + assert db.parameters[0]["automation_run_id"] == identity["automation_run_id"] + assert db.parameters[0]["incident_id"] == identity["incident_id"] + assert db.parameters[0]["apply_op_id"] == identity["apply_op_id"] + assert "UPDATE awooop_outbound_message" in db.statements[2] + + +@pytest.mark.asyncio +async def test_durable_reconcile_rejects_provider_message_identity_mismatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + db = _SequenceDB( + _MappingResult( + row={ + "message_id": "00000000-0000-0000-0000-000000000202", + "send_status": "sent", + "provider_message_id": "456", + "visual_requested": False, + } + ) + ) + + @asynccontextmanager + async def fake_get_db_context(_project_id): + yield db + + monkeypatch.setattr("src.db.base.get_db_context", fake_get_db_context) + gateway = TelegramGateway() + gateway._http_client = AsyncMock() + identity = { + "project_id": "awoooi", + "automation_run_id": "agent99-lifecycle-provider-mismatch", + "incident_id": "INC-20260716-AG99ACK", + "apply_op_id": "recovered|info", + "delivery_kind": "agent99_lifecycle", + } + + result = await gateway._reconcile_controlled_apply_result_outbound( + identity=identity, + expected_provider_message_id="999", + ) + + assert result["status"] == "provider_message_id_mismatch" + assert result["provider_message_id"] == "456" + assert len(db.statements) == 1 + gateway._http_client.post.assert_not_awaited() + + @pytest.mark.asyncio async def test_controlled_result_reuses_durable_sent_reservation( monkeypatch: pytest.MonkeyPatch, diff --git a/apps/api/tests/test_awooop_audit_ledger_contract.py b/apps/api/tests/test_awooop_audit_ledger_contract.py new file mode 100644 index 000000000..28d4e9837 --- /dev/null +++ b/apps/api/tests/test_awooop_audit_ledger_contract.py @@ -0,0 +1,302 @@ +from __future__ import annotations + +import asyncio +import json +import os +from contextlib import asynccontextmanager +from datetime import UTC, datetime, timedelta +from typing import Any +from uuid import UUID + +import pytest + +os.environ.setdefault( + "DATABASE_URL", + "postgresql+asyncpg://test:test@localhost/test", +) + +from src.db import base as db_base # noqa: E402 +from src.db.awooop_models import AwoooPRunState # noqa: E402 +from src.jobs import baseline_snapshot # noqa: E402 +from src.services import ( # noqa: E402 + audit_sink, +) +from src.services import ( # noqa: E402 + paid_provider_canary_authorization as authorization, +) + + +class _MappingResult: + def __init__(self, mapping: dict[str, Any]) -> None: + self.mapping = mapping + + def mappings(self) -> _MappingResult: + return self + + def one(self) -> dict[str, Any]: + return self.mapping + + +class _ConcurrentLedger: + def __init__(self) -> None: + self.lock = asyncio.Lock() + self.rows: list[dict[str, Any]] = [] + self.statements: list[str] = [] + + +class _ConcurrentDb: + def __init__(self, ledger: _ConcurrentLedger) -> None: + self.ledger = ledger + self.lock_held = False + + async def execute( + self, + statement: Any, + params: dict[str, Any] | None = None, + ) -> _MappingResult | None: + sql = " ".join(str(statement).split()) + self.ledger.statements.append(sql) + if "pg_advisory_xact_lock" in sql: + await self.ledger.lock.acquire() + self.lock_held = True + return None + if "SELECT COUNT(*) AS matching_count" in sql: + return _MappingResult( + { + "matching_count": len(self.ledger.rows), + "created_at": ( + datetime(2026, 7, 16, 8, 0, tzinfo=UTC) + if self.ledger.rows + else None + ), + } + ) + if "INSERT INTO alert_operation_log" in sql: + assert params is not None + self.ledger.rows.append(json.loads(str(params["context"]))) + return None + raise AssertionError(f"unexpected SQL: {sql}") + + +@pytest.mark.asyncio +async def test_required_audit_is_awaited_exact_once_without_race( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ledger = _ConcurrentLedger() + + @asynccontextmanager + async def fake_db_context(project_id: str): + assert project_id == "awoooi" + db = _ConcurrentDb(ledger) + try: + yield db + finally: + if db.lock_held: + ledger.lock.release() + + monkeypatch.setattr(db_base, "get_db_context", fake_db_context) + kwargs = { + "project_id": "awoooi", + "action": "run.approval.approve", + "resource_type": "run", + "resource_id": "267a3d26-3c29-470f-8fe3-388a2f408f39", + "run_id": "267a3d26-3c29-470f-8fe3-388a2f408f39", + "trace_id": "trace-paid-canary", + "details": { + "approver_id": "operator:test", + "decision": "approve", + "new_state": "running", + }, + "require_durable": True, + "idempotency_key": "paid-canary:exact-run", + } + + first, second = await asyncio.gather( + audit_sink.write_audit(**kwargs), + audit_sink.write_audit(**kwargs), + ) + + assert {first["status"], second["status"]} == { + "inserted_verified", + "duplicate_existing", + } + assert first["durable_write_ack"] is True + assert second["durable_write_ack"] is True + assert len(ledger.rows) == 1 + context = ledger.rows[0] + assert context["schema_version"] == "awooop_sanitized_audit_v1" + assert context["project_id"] == "awoooi" + assert context["run_id"] == kwargs["run_id"] + assert context["trace_id"] == "trace-paid-canary" + assert context["details"]["decision"] == "approve" + sql = " ".join(ledger.statements) + assert "pg_advisory_xact_lock" in sql + assert "INSERT INTO alert_operation_log" in sql + assert "INSERT INTO audit_logs" not in sql + + +@pytest.mark.asyncio +async def test_required_audit_rejects_preexisting_duplicate_rows( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ledger = _ConcurrentLedger() + ledger.rows.extend([{}, {}]) + + @asynccontextmanager + async def fake_db_context(project_id: str): + db = _ConcurrentDb(ledger) + try: + yield db + finally: + if db.lock_held: + ledger.lock.release() + + monkeypatch.setattr(db_base, "get_db_context", fake_db_context) + + with pytest.raises(audit_sink.AuditDurabilityError) as raised: + await audit_sink.write_audit( + project_id="awoooi", + action="run.approval.approve", + resource_type="run", + resource_id="267a3d26-3c29-470f-8fe3-388a2f408f39", + run_id="267a3d26-3c29-470f-8fe3-388a2f408f39", + trace_id="trace-paid-canary", + details={"decision": "approve"}, + require_durable=True, + idempotency_key="paid-canary:duplicate-run", + ) + + assert raised.value.error_code == "audit_idempotency_duplicate_detected" + assert len(ledger.rows) == 2 + + +class _SequenceResult: + def __init__( + self, + *, + scalar: Any = None, + row: Any = None, + mapping: dict[str, Any] | None = None, + ) -> None: + self.scalar = scalar + self.row = row + self.mapping = mapping + + def scalar_one_or_none(self) -> Any: + return self.scalar + + def one(self) -> Any: + return self.row if self.row is not None else self.mapping + + def mappings(self) -> _SequenceResult: + return self + + +class _SequenceDb: + def __init__(self, results: list[_SequenceResult]) -> None: + self.results = results + self.calls: list[tuple[Any, dict[str, Any] | None]] = [] + + async def execute( + self, + statement: Any, + params: dict[str, Any] | None = None, + ) -> _SequenceResult: + self.calls.append((statement, params)) + return self.results.pop(0) + + +@pytest.mark.asyncio +async def test_paid_canary_reader_uses_exact_same_run_user_action_schema( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run_id = UUID("267a3d26-3c29-470f-8fe3-388a2f408f39") + trace_id = "00-0123456789abcdef0123456789abcdef-0123456789abcdef-01" + now = datetime(2026, 7, 16, 8, 0, tzinfo=UTC) + run = AwoooPRunState( + run_id=run_id, + project_id="awoooi", + agent_id="awoooi-paid-provider-canary", + state="running", + trace_id=trace_id, + trigger_type="api", + trigger_ref="paid-provider-canary:AIA-SRE-013", + is_shadow=False, + input_sha256=authorization.paid_provider_canary_authorization_input_sha256(), + timeout_at=now + timedelta(minutes=5), + ) + db = _SequenceDb( + [ + _SequenceResult(scalar=run), + _SequenceResult(row=(1, "success", False, now - timedelta(seconds=30))), + _SequenceResult( + mapping={ + "approval_audit_count": 1, + "approval_audit_created_at": now - timedelta(seconds=20), + } + ), + _SequenceResult(mapping={"prior_canary_start_count": 0}), + ] + ) + + @asynccontextmanager + async def fake_db_context(project_id: str): + assert project_id == "awoooi" + yield db + + monkeypatch.setattr(authorization, "get_db_context", fake_db_context) + + evidence = await authorization._SqlPaidProviderCanaryAuthorizationEvidenceReader().read( + run_id=run_id, + project_id="awoooi", + work_item_id="AIA-SRE-013", + ) + + assert evidence is not None + assert evidence.approval_audit_count == 1 + audit_statement, audit_params = db.calls[2] + audit_sql = " ".join(str(audit_statement).split()) + assert "FROM alert_operation_log" in audit_sql + assert "event_type = 'USER_ACTION'" in audit_sql + assert "FROM audit_logs" not in audit_sql + assert "context ->> 'project_id' = :project_id" in audit_sql + assert "context ->> 'resource_id' = :run_id" in audit_sql + assert "context ->> 'run_id' = :run_id" in audit_sql + assert "context ->> 'trace_id' = :trace_id" in audit_sql + assert "context #>> '{details,decision}' = 'approve'" in audit_sql + assert audit_params is not None + assert audit_params["run_id"] == str(run_id) + assert audit_params["trace_id"] == trace_id + expected_key = authorization.paid_provider_canary_approval_audit_idempotency_key( + run_id + ) + assert audit_params["idempotency_key_sha256"] == ( + audit_sink.audit_idempotency_key_sha256(expected_key) + ) + + +@pytest.mark.asyncio +async def test_baseline_mcp_count_uses_mcp_audit_ledger( + monkeypatch: pytest.MonkeyPatch, +) -> None: + statements: list[Any] = [] + + class _Db: + async def execute(self, statement: Any) -> _SequenceResult: + statements.append(statement) + return _SequenceResult(scalar=7) + + @asynccontextmanager + async def fake_db_context(): + yield _Db() + + monkeypatch.setattr(baseline_snapshot, "get_db_context", fake_db_context) + + count = await baseline_snapshot._count_mcp_calls_24h( + datetime(2026, 7, 15, 8, 0, tzinfo=UTC) + ) + + assert count == 7 + sql = " ".join(str(statements[0]).split()) + assert "mcp_audit_log" in sql + assert "audit_logs.action" not in sql diff --git a/apps/api/tests/test_awooop_outbound_message_hot_path_indexes.py b/apps/api/tests/test_awooop_outbound_message_hot_path_indexes.py new file mode 100644 index 000000000..1e849b902 --- /dev/null +++ b/apps/api/tests/test_awooop_outbound_message_hot_path_indexes.py @@ -0,0 +1,73 @@ +from pathlib import Path + +from src.services.ai_agent_autonomous_runtime_control import ( + _RUNTIME_RECEIPT_STATEMENT_TIMEOUT_MS, + _RUNTIME_TELEGRAM_COUNTS_DIRECT_SQL, + _RUNTIME_TELEGRAM_COUNTS_SQL, + _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL, + _RUNTIME_TELEGRAM_LATEST_SQL, +) + +ROOT = Path(__file__).resolve().parents[1] +MIGRATION = ( + "awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16.sql" +) +ROLLBACK = ( + "awooop_outbound_message_telegram_receipt_hot_path_index_2026-07-16_down.sql" +) + + +def _read_migration(name: str) -> str: + return (ROOT / "migrations" / name).read_text() + + +def _normalize_sql(sql: str) -> str: + return " ".join(sql.split()) + + +def test_telegram_receipt_hot_path_index_is_concurrent_and_transactionless() -> None: + migration = _normalize_sql(_read_migration(MIGRATION)) + rollback = _normalize_sql(_read_migration(ROLLBACK)) + + assert "BEGIN;" not in migration + assert "COMMIT;" not in migration + assert migration.count("CREATE INDEX CONCURRENTLY IF NOT EXISTS") == 1 + assert ( + "idx_awooop_outbound_msg_telegram_receipt_latest " + "ON awooop_outbound_message (project_id, queued_at DESC) " + "INCLUDE (send_status)" + ) in migration + assert "BEGIN;" not in rollback + assert "COMMIT;" not in rollback + assert ( + "DROP INDEX CONCURRENTLY IF EXISTS " + "idx_awooop_outbound_msg_telegram_receipt_latest;" + ) in rollback + + +def test_telegram_receipt_index_predicate_matches_all_runtime_queries() -> None: + predicate = _normalize_sql(""" + channel_type = 'telegram' + AND ( + source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' + OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent' + ) + """) + sources = ( + _read_migration(MIGRATION), + _RUNTIME_TELEGRAM_COUNTS_SQL, + _RUNTIME_TELEGRAM_COUNTS_DIRECT_SQL, + _RUNTIME_TELEGRAM_LATEST_SQL, + _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL, + ) + + for source in sources: + assert predicate in _normalize_sql(source) + + +def test_telegram_latest_run_id_alias_and_timeout_contract_stay_aligned() -> None: + for sql in (_RUNTIME_TELEGRAM_LATEST_SQL, _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL): + assert "run_id::text AS run_id" in sql + assert "outbound_run_id" not in sql + + assert _RUNTIME_RECEIPT_STATEMENT_TIMEOUT_MS == 700 diff --git a/apps/api/tests/test_paid_provider_canary_gate5_run.py b/apps/api/tests/test_paid_provider_canary_gate5_run.py index 3cf271a66..71d4e9331 100644 --- a/apps/api/tests/test_paid_provider_canary_gate5_run.py +++ b/apps/api/tests/test_paid_provider_canary_gate5_run.py @@ -254,6 +254,100 @@ async def test_generic_approval_service_rejects_expired_paid_canary_before_trans assert "timeout_not_expired" in str(exc.value.detail) +@pytest.mark.asyncio +async def test_paid_canary_audit_failure_terminalizes_active_gate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + run = _run(now=datetime.now(UTC)) + + class _Result: + def scalar_one_or_none(self) -> AwoooPRunState: + return run + + class _Db: + async def execute(self, *_: object, **__: object) -> _Result: + return _Result() + + @asynccontextmanager + async def fake_db_context(project_id: str): + assert project_id == "awoooi" + yield _Db() + + transition_calls: list[dict[str, object]] = [] + audit_calls: list[dict[str, object]] = [] + + async def fake_record_approval(**_: object) -> int: + return 1 + + async def fake_transition( + run_id: UUID, + project_id: str, + to_state: str, + **kwargs: object, + ) -> None: + transition_calls.append( + { + "run_id": run_id, + "project_id": project_id, + "to_state": to_state, + **kwargs, + } + ) + + async def fake_record_step(**_: object) -> None: + return None + + async def failing_write_audit(**kwargs: object) -> None: + audit_calls.append(kwargs) + raise platform_operator_service.AuditDurabilityError( + "audit_durable_write_failed" + ) + + monkeypatch.setattr(platform_operator_service, "get_db_context", fake_db_context) + monkeypatch.setattr( + platform_operator_service, + "issue_approval_token", + lambda **_: "e30.eyJqdGkiOiJqdGktdGVzdCJ9.sig", + ) + monkeypatch.setattr( + platform_operator_service, + "record_approval", + fake_record_approval, + ) + monkeypatch.setattr(platform_operator_service, "transition", fake_transition) + monkeypatch.setattr( + platform_operator_service, + "_record_approval_decision_step", + fake_record_step, + ) + monkeypatch.setattr( + platform_operator_service, + "write_audit", + failing_write_audit, + ) + + with pytest.raises(HTTPException) as raised: + await platform_operator_service.decide_approval( + run_id=str(run.run_id), + project_id="awoooi", + decision="approve", + approver_id="operator:test", + reason="bounded canary", + ) + + assert raised.value.status_code == 503 + assert raised.value.detail == "critical_approval_durable_audit_unavailable" + assert [call["to_state"] for call in transition_calls] == ["running", "failed"] + assert transition_calls[1]["error_code"] == "E-PAID-CANARY-AUDIT" + assert len(audit_calls) == 1 + assert audit_calls[0]["require_durable"] is True + assert audit_calls[0]["run_id"] == str(run.run_id) + assert audit_calls[0]["trace_id"] == run.trace_id + assert "paid-provider-canary:approval:v1" in str( + audit_calls[0]["idempotency_key"] + ) + + @pytest.mark.asyncio async def test_operator_route_uses_authenticated_principal_and_fixed_body( monkeypatch: pytest.MonkeyPatch, diff --git a/apps/api/tests/test_postgres_jsonb_bind_contract.py b/apps/api/tests/test_postgres_jsonb_bind_contract.py index e8e5467e6..f9f1b4fb8 100644 --- a/apps/api/tests/test_postgres_jsonb_bind_contract.py +++ b/apps/api/tests/test_postgres_jsonb_bind_contract.py @@ -28,15 +28,19 @@ class _CaptureDb: self.calls.append((statement, params)) -def _assert_jsonb_bind_contract(db: _CaptureDb) -> None: +def _assert_immutable_audit_jsonb_bind_contract(db: _CaptureDb) -> None: assert len(db.calls) == 1 statement, params = db.calls[0] normalized_sql = " ".join(str(statement).split()) - assert ":details::jsonb" not in normalized_sql - assert "CAST(:details AS JSONB)" in normalized_sql - assert "details" in statement._bindparams - assert "detail" not in statement._bindparams - assert json.loads(params["details"])["result"] == "verified" + assert "INSERT INTO alert_operation_log" in normalized_sql + assert "INSERT INTO audit_logs" not in normalized_sql + assert ":context::jsonb" not in normalized_sql + assert "CAST(:context AS JSONB)" in normalized_sql + assert "context" in statement._bindparams + assert "details" not in statement._bindparams + context = json.loads(params["context"]) + assert context["schema_version"] == "awooop_sanitized_audit_v1" + assert context["details"]["result"] == "verified" @pytest.mark.asyncio @@ -62,7 +66,7 @@ async def test_audit_sink_uses_asyncpg_safe_jsonb_bind( trace_id="trace-1", ) - _assert_jsonb_bind_contract(db) + _assert_immutable_audit_jsonb_bind_contract(db) @pytest.mark.asyncio @@ -86,4 +90,4 @@ async def test_contract_service_uses_asyncpg_safe_jsonb_bind( details={"result": "verified"}, ) - _assert_jsonb_bind_contract(db) + _assert_immutable_audit_jsonb_bind_contract(db)