diff --git a/apps/api/src/api/v1/platform/events.py b/apps/api/src/api/v1/platform/events.py index 4424a8131..d9023062e 100644 --- a/apps/api/src/api/v1/platform/events.py +++ b/apps/api/src/api/v1/platform/events.py @@ -18,7 +18,11 @@ from src.core.awooop_operator_auth import ( AwoooPOperatorPrincipal, verify_awooop_operator, ) -from src.core.context import clear_project_context, get_current_project_context, set_project_context +from src.core.context import ( + clear_project_context, + get_current_project_context, + set_project_context, +) from src.services.channel_event_dossier_service import ( RecurrenceWorkItemHandoffKind, RecurrenceWorkItemMode, @@ -50,6 +54,7 @@ def _empty_recurrence_summary() -> dict[str, Any]: "unlinked_event_total": 0, "auto_repair_linked_total": 0, "verified_repair_group_total": 0, + "agent99_callback_verified_group_total": 0, "open_work_item_group_total": 0, "manual_gate_group_total": 0, "controlled_apply_gate_group_total": 0, @@ -239,6 +244,7 @@ class ChannelEventRecurrenceSummary(BaseModel): unlinked_event_total: int auto_repair_linked_total: int = 0 verified_repair_group_total: int = 0 + agent99_callback_verified_group_total: int = 0 open_work_item_group_total: int = 0 manual_gate_group_total: int = 0 controlled_apply_gate_group_total: int = 0 diff --git a/apps/api/src/services/channel_event_dossier_service.py b/apps/api/src/services/channel_event_dossier_service.py index a8d7d0ee0..bb620b5d8 100644 --- a/apps/api/src/services/channel_event_dossier_service.py +++ b/apps/api/src/services/channel_event_dossier_service.py @@ -33,6 +33,9 @@ _SOURCE_CORRELATION_APPLY_SCHEMA_VERSION = "awooop_source_correlation_apply_v1" _SOURCE_CORRELATION_APPLY_STAGE = "source_correlation_linked" _SOURCE_CORRELATION_REVIEW_ACTOR = "awooop_source_correlation_review_service" _SOURCE_CORRELATION_APPLY_ACTOR = "awooop_source_correlation_apply_service" +_AGENT99_COMPLETION_CALLBACK_ACTOR = "agent99_completion_callback" +_AGENT99_COMPLETION_CALLBACK_SCHEMA = "agent99_completion_callback_v1" +_AGENT99_CALLBACK_REPAIR_EVIDENCE = "agent99_completion_callback" _INCIDENT_ID_RE = re.compile(r"\bINC-\d{8}-[A-Z0-9]{4,}\b") RecurrenceWorkItemMode = Literal["auto", "ticket", "reverify", "approval_review", "observe"] RecurrenceWorkItemHandoffKind = Literal["ticket_proposal", "manual_review"] @@ -412,7 +415,13 @@ def build_dossier_recurrence( 1 for item in items if _as_dict(item.get("repair_summary")).get("status") - == "auto_repair_verified" + in {"auto_repair_verified", "agent99_callback_verified"} + ), + "agent99_callback_verified_group_total": sum( + 1 + for item in items + if _as_dict(item.get("repair_summary")).get("status") + == "agent99_callback_verified" ), "open_work_item_group_total": sum( 1 @@ -464,6 +473,7 @@ def _repair_status( *, incident_id: str | None, latest_run_state: str | None, + latest_stage: str | None, repair_summary: dict[str, Any] | None, ) -> str: if not incident_id: @@ -473,6 +483,18 @@ def _repair_status( verification = str( repair_summary.get("latest_verification_result") or "" ).lower() + if ( + repair_summary.get("repair_evidence_kind") + == _AGENT99_CALLBACK_REPAIR_EVIDENCE + ): + if ( + latest_success is True + and verification == "success" + and str(latest_stage or "").lower() == "resolved" + and latest_run_state == "completed" + ): + return "agent99_callback_verified" + return "agent99_callback_verification_stale" if latest_success is True and verification == "success": return "auto_repair_verified" if latest_success is True: @@ -494,6 +516,7 @@ def _work_item_status(repair_status: str) -> str: return "none" if repair_status in { "auto_repair_verified", + "agent99_callback_verified", "source_correlation_accepted", "source_correlation_rejected", }: @@ -502,6 +525,10 @@ def _work_item_status(repair_status: str) -> str: def _work_item_kind(repair_status: str, auto_repair_id: Any) -> str: + if repair_status == "agent99_callback_verified": + return "verified_repair" + if repair_status == "agent99_callback_verification_stale": + return "verification" if repair_status in { "source_correlation_review", "source_correlation_accepted", @@ -526,6 +553,8 @@ def _work_item_next_step(repair_status: str) -> str: "source_correlation_rejected": "monitor_for_new_provider_evidence", "source_correlation_applied": "verify_source_link_in_status_chain", "auto_repair_succeeded_unverified": "run_post_verification", + "agent99_callback_verified": "monitor_recurrence", + "agent99_callback_verification_stale": "rerun_agent99_verifier", "auto_repair_failed": "triage_failed_repair", "auto_repair_recorded": "review_repair_record", "controlled_apply_gate": "evaluate_controlled_apply", @@ -543,6 +572,8 @@ def _work_item_reason(repair_status: str) -> str: "source_correlation_rejected": "provider_native_evidence_rejected", "source_correlation_applied": "provider_native_evidence_link_applied", "auto_repair_succeeded_unverified": "auto_repair_missing_verification", + "agent99_callback_verified": "agent99_callback_durable_verified", + "agent99_callback_verification_stale": "latest_event_not_verified_resolved", "auto_repair_failed": "auto_repair_failed", "auto_repair_recorded": "auto_repair_record_needs_review", "controlled_apply_gate": "controlled_apply_required", @@ -573,6 +604,7 @@ def _attach_work_item_summary( status_value = _repair_status( incident_id=latest_incident_id, latest_run_state=group.get("latest_run_state"), + latest_stage=group.get("latest_stage"), repair_summary=repair_summary, ) if _needs_source_correlation_review(group, latest_incident_id): @@ -667,7 +699,10 @@ def _attach_work_item_summary( "kind": _work_item_kind(status_value, auto_repair_id), "next_step": work_item_next_step, "reason": work_item_reason, - "needs_human": work_status == "open", + "needs_human": bool( + work_status == "open" + and status_value != "agent99_callback_verification_stale" + ), } @@ -2050,6 +2085,133 @@ async def _fetch_auto_repair_summaries_by_incident( return summaries +async def _fetch_agent99_callback_summaries_by_incident( + db: Any, + incident_ids: list[str], + *, + project_id: str, +) -> dict[str, dict[str, Any]]: + """Fetch independently verified Agent99 callback receipts by incident.""" + visible_incident_ids = incident_ids[:_MAX_REPAIR_INCIDENTS] + if not visible_incident_ids: + return {} + + placeholders: list[str] = [] + params: dict[str, Any] = { + "actor": _AGENT99_COMPLETION_CALLBACK_ACTOR, + "schema_version": _AGENT99_COMPLETION_CALLBACK_SCHEMA, + "project_id": project_id, + } + for index, incident_id in enumerate(visible_incident_ids): + key = f"agent99_incident_id_{index}" + placeholders.append(f":{key}") + params[key] = incident_id + + result = await db.execute( + text( + f""" + WITH callback_events AS ( + SELECT + operation.incident_id, + operation.context ->> 'callback_id' AS callback_id, + max(operation.context ->> 'run_id') AS callback_run_id, + bool_or( + operation.event_type = 'EXECUTION_COMPLETED' + AND operation.success IS TRUE + ) AS execution_verified, + bool_or( + operation.event_type = 'RESOLVED' + AND operation.success IS TRUE + ) AS resolution_verified, + bool_or( + coalesce(operation.context ->> 'verifier_passed', 'false') + = 'true' + ) AS verifier_passed, + bool_or( + coalesce( + operation.context ->> 'source_event_resolved', + 'false' + ) = 'true' + ) AS source_event_resolved, + count(*) AS operation_receipt_count, + max(operation.created_at) AS latest_callback_at + FROM alert_operation_log operation + WHERE operation.actor = :actor + AND operation.context ->> 'schema_version' = :schema_version + AND coalesce( + operation.context ->> 'project_id', + :project_id + ) = :project_id + AND operation.context ->> 'callback_id' IS NOT NULL + AND operation.incident_id IN ({", ".join(placeholders)}) + GROUP BY + operation.incident_id, + operation.context ->> 'callback_id' + ), + verified_callbacks AS ( + SELECT * + FROM callback_events + WHERE execution_verified IS TRUE + AND resolution_verified IS TRUE + AND verifier_passed IS TRUE + AND source_event_resolved IS TRUE + ), + ranked AS ( + SELECT + verified_callbacks.*, + count(*) OVER ( + PARTITION BY verified_callbacks.incident_id + ) AS verified_callback_total, + row_number() OVER ( + PARTITION BY verified_callbacks.incident_id + ORDER BY verified_callbacks.latest_callback_at DESC + ) AS rn + FROM verified_callbacks + ) + SELECT + incident_id, + callback_id, + callback_run_id, + operation_receipt_count, + latest_callback_at, + verified_callback_total + FROM ranked + WHERE rn = 1 + """ + ), + params, + ) + + summaries: dict[str, dict[str, Any]] = {} + for row in result.mappings().all(): + item = dict(row) + incident_id = str(item.get("incident_id") or "") + if not incident_id: + continue + summaries[incident_id] = { + "schema_version": "awooop_recurrence_repair_summary_v1", + "incident_id": incident_id, + "repair_evidence_kind": _AGENT99_CALLBACK_REPAIR_EVIDENCE, + "latest_auto_repair_id": None, + "latest_agent99_callback_id": item.get("callback_id"), + "latest_agent99_run_id": item.get("callback_run_id"), + "latest_success": True, + "latest_verification_result": "success", + "latest_triggered_by": _AGENT99_COMPLETION_CALLBACK_ACTOR, + "latest_callback_at": item.get("latest_callback_at"), + "operation_receipt_count": int( + item.get("operation_receipt_count") or 0 + ), + "verified_callback_total": int( + item.get("verified_callback_total") or 0 + ), + "auto_repair_total": 0, + "success_total": int(item.get("verified_callback_total") or 0), + "failed_total": 0, + } + return summaries + + async def _fetch_source_review_decisions_by_work_item( db: Any, *, @@ -2348,10 +2510,20 @@ async def fetch_channel_event_dossier_recurrence( params, ) rows = [dict(row) for row in result.mappings().all()] + incident_ids = _collect_incident_ids_from_rows(rows) repair_summaries = await _fetch_auto_repair_summaries_by_incident( db, - _collect_incident_ids_from_rows(rows), + incident_ids, ) + agent99_callback_summaries = ( + await _fetch_agent99_callback_summaries_by_incident( + db, + incident_ids, + project_id=effective_project_id, + ) + ) + for incident_id, callback_summary in agent99_callback_summaries.items(): + repair_summaries.setdefault(incident_id, callback_summary) source_review_decisions = await _fetch_source_review_decisions_by_work_item( db, project_id=effective_project_id, diff --git a/apps/api/tests/test_channel_event_dossier_service.py b/apps/api/tests/test_channel_event_dossier_service.py index 353ae619f..2e0582b99 100644 --- a/apps/api/tests/test_channel_event_dossier_service.py +++ b/apps/api/tests/test_channel_event_dossier_service.py @@ -782,6 +782,174 @@ def test_build_dossier_recurrence_opens_work_item_for_completed_run_without_repa } +def test_agent99_durable_callback_closes_recurrence_automation_gap() -> None: + incident_id = "INC-20260712-72CBC5" + recurrence = build_dossier_recurrence( + [ + { + "event_id": "11111111-1111-4111-8111-111111111111", + "project_id": "awoooi", + "channel_type": "internal", + "provider_event_id": "agent99:resolved:callback-1", + "content_hash": "a" * 64, + "content_preview": f"Agent99 Recover resolved {incident_id}", + "content_redacted": f"Agent99 Recover resolved {incident_id}", + "redaction_version": "audit_sink_v1", + "source_envelope": { + "provider": "agent99", + "stage": "resolved", + "source_refs": {"incident_ids": [incident_id]}, + "log_correlation": { + "target_resource": "cold-start-gate", + "fingerprint": "agent99-cold-start", + }, + }, + "is_duplicate": False, + "provider_ts": None, + "received_at": "2026-07-14T09:30:28", + "run_id": UUID("22222222-2222-4222-8222-222222222222"), + "run_state": "completed", + "run_agent_id": "agent99", + } + ], + project_id="awoooi", + limit=20, + repair_summaries_by_incident={ + incident_id: { + "schema_version": "awooop_recurrence_repair_summary_v1", + "incident_id": incident_id, + "repair_evidence_kind": "agent99_completion_callback", + "latest_auto_repair_id": None, + "latest_agent99_callback_id": "agent99:callback-1:resolved", + "latest_success": True, + "latest_verification_result": "success", + "operation_receipt_count": 2, + "verified_callback_total": 1, + "auto_repair_total": 0, + "success_total": 1, + "failed_total": 0, + } + }, + ) + + item = recurrence["items"][0] + assert recurrence["summary"]["verified_repair_group_total"] == 1 + assert recurrence["summary"]["agent99_callback_verified_group_total"] == 1 + assert recurrence["summary"]["automation_gap_group_total"] == 0 + assert recurrence["summary"]["open_work_item_group_total"] == 0 + assert item["repair_summary"]["status"] == "agent99_callback_verified" + assert item["work_item"] == { + "schema_version": "awooop_recurrence_work_item_link_v1", + "work_item_id": f"incident:{incident_id}", + "incident_id": incident_id, + "matched_incident_id": None, + "auto_repair_id": None, + "status": "closed", + "kind": "verified_repair", + "next_step": "monitor_recurrence", + "reason": "agent99_callback_durable_verified", + "needs_human": False, + } + + +def test_agent99_callback_does_not_close_reopened_recurrence() -> None: + incident_id = "INC-20260712-72CBC5" + recurrence = build_dossier_recurrence( + [ + { + "event_id": "11111111-1111-4111-8111-111111111111", + "project_id": "awoooi", + "channel_type": "internal", + "provider_event_id": "agent99:firing:callback-2", + "content_hash": "b" * 64, + "content_preview": f"Agent99 firing {incident_id}", + "content_redacted": f"Agent99 firing {incident_id}", + "redaction_version": "audit_sink_v1", + "source_envelope": { + "provider": "agent99", + "stage": "firing", + "source_refs": {"incident_ids": [incident_id]}, + "log_correlation": {"fingerprint": "agent99-cold-start"}, + }, + "is_duplicate": False, + "provider_ts": None, + "received_at": "2026-07-14T09:35:00", + "run_id": UUID("22222222-2222-4222-8222-222222222222"), + "run_state": "completed", + "run_agent_id": "agent99", + } + ], + project_id="awoooi", + limit=20, + repair_summaries_by_incident={ + incident_id: { + "repair_evidence_kind": "agent99_completion_callback", + "latest_success": True, + "latest_verification_result": "success", + } + }, + ) + + item = recurrence["items"][0] + assert item["repair_summary"]["status"] == ( + "agent99_callback_verification_stale" + ) + assert item["work_item"]["status"] == "open" + assert item["work_item"]["needs_human"] is False + assert item["work_item"]["next_step"] == "rerun_agent99_verifier" + assert recurrence["summary"]["verified_repair_group_total"] == 0 + + +@pytest.mark.asyncio +async def test_agent99_callback_summary_requires_execution_and_resolution() -> None: + captured: list[dict[str, object]] = [] + + class FakeMappings: + def all(self) -> list[dict[str, object]]: + return [ + { + "incident_id": "INC-20260712-72CBC5", + "callback_id": "agent99:callback-1:resolved", + "callback_run_id": "sre-alert-1", + "operation_receipt_count": 2, + "latest_callback_at": "2026-07-14T09:30:29", + "verified_callback_total": 1, + } + ] + + class FakeResult: + def mappings(self) -> FakeMappings: + return FakeMappings() + + class FakeDb: + async def execute(self, statement, params): # noqa: ANN001 + captured.append({"sql": str(statement), "params": params}) + return FakeResult() + + result = await channel_event_dossier_service._fetch_agent99_callback_summaries_by_incident( + FakeDb(), + ["INC-20260712-72CBC5"], + project_id="awoooi", + ) + + summary = result["INC-20260712-72CBC5"] + assert summary["repair_evidence_kind"] == "agent99_completion_callback" + assert summary["latest_verification_result"] == "success" + assert summary["operation_receipt_count"] == 2 + assert summary["verified_callback_total"] == 1 + sql = str(captured[0]["sql"]) + assert "EXECUTION_COMPLETED" in sql + assert "RESOLVED" in sql + assert "verifier_passed" in sql + assert "source_event_resolved" in sql + assert captured[0]["params"] == { + "actor": "agent99_completion_callback", + "schema_version": "agent99_completion_callback_v1", + "project_id": "awoooi", + "agent99_incident_id_0": "INC-20260712-72CBC5", + } + + def test_build_recurrence_work_item_preview_selects_ticket_mode() -> None: recurrence = build_dossier_recurrence( [