"""Durable, idempotent Agent99 completion callback writeback.""" from __future__ import annotations from hashlib import sha256 from typing import Any from sqlalchemy import text from src.db.base import get_db_context from src.models.agent99_completion import Agent99CompletionCallbackRequest from src.repositories.alert_operation_log_repository import ( get_alert_operation_log_repository, ) from src.services.channel_hub import ( build_external_alert_provider_event_id, record_external_alert_event, ) _OPERATION_INCIDENT_ID_MAX_LENGTH = 30 def _operation_incident_id(alert_id: str | None) -> str | None: """Keep external alert identity while fitting the immutable event schema.""" if not alert_id: return None if len(alert_id) <= _OPERATION_INCIDENT_ID_MAX_LENGTH: return alert_id digest = sha256(alert_id.encode("utf-8")).hexdigest().upper()[:21] return f"INC-AG99-{digest}" async def _load_callback_readback( *, project_id: str, provider_event_id: str, callback_id: str, ) -> dict[str, Any] | None: async with get_db_context(project_id) as db: result = await db.execute( text( """ SELECT event_row.event_id::text AS event_id, event_row.run_id::text AS run_id, ( SELECT count(*) FROM alert_operation_log operation_row WHERE operation_row.context ->> 'callback_id' = :callback_id ) AS operation_count, ( SELECT count(*) FROM alert_operation_log operation_row WHERE operation_row.context ->> 'callback_id' = :callback_id AND operation_row.event_type = 'EXECUTION_COMPLETED' ) AS execution_count, ( SELECT count(*) FROM alert_operation_log operation_row WHERE operation_row.context ->> 'callback_id' = :callback_id AND operation_row.event_type = 'RESOLVED' ) AS resolved_count FROM awooop_conversation_event event_row WHERE event_row.project_id = :project_id AND event_row.channel_type = 'internal' AND event_row.provider_event_id = :provider_event_id LIMIT 1 """ ), { "project_id": project_id, "provider_event_id": provider_event_id, "callback_id": callback_id, }, ) row = result.mappings().first() return dict(row) if row else None def _operation_context(payload: Agent99CompletionCallbackRequest) -> dict[str, Any]: return { "schema_version": payload.schema_version, "callback_id": payload.callback_id, "project_id": payload.project_id, "run_id": payload.run_id, "trace_id": payload.trace_id, "work_item_id": payload.work_item_id, "alert_id": payload.alert_id, "operation_incident_id": _operation_incident_id(payload.alert_id), "correlation_key": payload.correlation_key, "source": payload.source, "mode": payload.mode, "outcome_state": payload.outcome_state, "controlled_apply": payload.controlled_apply, "transport_ok": payload.transport_ok, "verifier_name": payload.verifier_name, "verifier_passed": payload.verifier_passed, "source_event_resolved": payload.source_event_resolved, "source_event_resolution_policy": payload.source_event_resolution_policy, "duration_seconds": payload.duration_seconds, "evidence_ref": payload.evidence_ref, "alert_kind": payload.alert_kind, "alert_service": payload.alert_service, "alert_host": payload.alert_host, "telegram": payload.telegram.model_dump(mode="json"), "problem": payload.problem.model_dump(mode="json"), "occurred_at": payload.occurred_at.isoformat(), "stores_secret": False, "raw_log_stored": False, } def _owner_plane(alert_kind: str | None) -> str: value = (alert_kind or "").lower() if any(marker in value for marker in ("security", "wazuh", "siem", "cve")): return "iwooos" return "awoooi" async def record_agent99_completion_callback( payload: Agent99CompletionCallbackRequest, ) -> dict[str, Any]: """Write one callback to the AWOOOI/AwoooP truth chain and read it back.""" stage = payload.outcome_state provider_event_id = build_external_alert_provider_event_id( "agent99", payload.callback_id, stage, ) before = await _load_callback_readback( project_id=payload.project_id, provider_event_id=provider_event_id, callback_id=payload.callback_id, ) duplicate = before is not None terminal_resolved = bool( payload.outcome_state == "resolved" and payload.verifier_passed and payload.source_event_resolved ) severity = ( "info" if payload.outcome_state == "resolved" else "critical" if payload.outcome_state == "failed" else "warning" ) context = _operation_context(payload) operation_incident_id = _operation_incident_id(payload.alert_id) event_id = await record_external_alert_event( project_id=payload.project_id, provider="agent99", event_id=payload.callback_id, stage=stage, title=f"Agent99 {payload.mode} {payload.outcome_state}", severity=severity, namespace="agent99", target_resource=payload.alert_service or payload.mode, fingerprint=payload.correlation_key or payload.trace_id, incident_id=payload.alert_id, labels={ "agent": "agent99", "mode": payload.mode, "outcome_state": payload.outcome_state, "verifier_passed": str(payload.verifier_passed).lower(), "source_event_resolved": str(payload.source_event_resolved).lower(), "owner_plane": _owner_plane(payload.alert_kind), }, annotations={ "run_id": payload.run_id, "trace_id": payload.trace_id, "work_item_id": payload.work_item_id, "evidence_ref": payload.evidence_ref, }, payload=context, is_duplicate=duplicate, ) repository = get_alert_operation_log_repository() if int((before or {}).get("execution_count") or 0) == 0: await repository.append( "EXECUTION_COMPLETED", incident_id=operation_incident_id, actor="agent99_completion_callback", action_detail=f"{payload.mode}:{payload.outcome_state}", success=payload.verifier_passed, context=context, ) if terminal_resolved and int((before or {}).get("resolved_count") or 0) == 0: await repository.append( "RESOLVED", incident_id=operation_incident_id, actor="agent99_completion_callback", action_detail=f"{payload.mode}:verified_resolved", success=True, context=context, ) readback = await _load_callback_readback( project_id=payload.project_id, provider_event_id=provider_event_id, callback_id=payload.callback_id, ) operation_count = int((readback or {}).get("operation_count") or 0) execution_count = int((readback or {}).get("execution_count") or 0) resolved_count = int((readback or {}).get("resolved_count") or 0) durable_readback = bool( readback and str(readback.get("event_id") or "") == str(event_id) and readback.get("run_id") and execution_count > 0 and (not terminal_resolved or resolved_count > 0) ) return { "schema_version": "agent99_completion_callback_receipt_v1", "ok": durable_readback, "status": "duplicate_verified" if duplicate else "recorded_verified", "accepted": True, "duplicate": duplicate, "callback_id": payload.callback_id, "run_id": payload.run_id, "trace_id": payload.trace_id, "work_item_id": payload.work_item_id, "outcome_state": payload.outcome_state, "conversation_event_id": str(event_id), "awooop_run_id": (readback or {}).get("run_id"), "operation_receipt_count": operation_count, "execution_receipt_count": execution_count, "resolved_receipt_count": resolved_count, "durable_readback": durable_readback, "stores_secret": False, "raw_log_stored": False, }