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 5bd172940..3a5fd003f 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -3717,8 +3717,7 @@ def _autonomous_execution_loop_ledger( ( row for row in telegram_rows - if str(row.get("send_status") or "") == "sent" - and str(row.get("action") or "") == "controlled_apply_result" + if _durable_controlled_apply_telegram_receipt(row) and ( not incident_id or str(row.get("incident_id") or "") == incident_id @@ -3935,6 +3934,27 @@ def _autonomous_execution_loop_ledger( } +def _durable_controlled_apply_telegram_receipt( + row: Mapping[str, Any], +) -> bool: + """Accept provider sends or policy-complete no-send suppression receipts.""" + + if str(row.get("action") or "") != "controlled_apply_result": + return False + send_status = str(row.get("send_status") or "") + if send_status == "sent": + return True + return bool( + send_status == "shadow" + and not row.get("provider_message_id") + and str(row.get("notification_policy_version") or "") + and str(row.get("notification_disposition") or "") == "suppressed" + and str(row.get("notification_provider_delivery") or "") + in {"shadow_only", "state_transition", "digest"} + and row.get("notification_provider_send_suppressed") is True + ) + + def _latest_flow_closure( *, operation_latest_rows: Iterable[Mapping[str, Any] | Any], @@ -4001,8 +4021,7 @@ def _latest_flow_closure( ( row for row in telegram_rows - if str(row.get("send_status") or "") == "sent" - and str(row.get("action") or "") == "controlled_apply_result" + if _durable_controlled_apply_telegram_receipt(row) and ( not incident_id or str(row.get("incident_id") or "") == incident_id @@ -4989,6 +5008,23 @@ def _build_autonomous_retry_rollback_terminal_readback( _bool_value(row.get("rollback_performed")) and _bool_value(row.get("rollback_verified")) ) + telegram_receipt = { + "action": "controlled_apply_result", + "send_status": row.get("telegram_send_status"), + "provider_message_id": row.get("telegram_provider_message_id"), + "notification_policy_version": row.get( + "telegram_notification_policy_version" + ), + "notification_disposition": row.get( + "telegram_notification_disposition" + ), + "notification_provider_delivery": row.get( + "telegram_notification_provider_delivery" + ), + "notification_provider_send_suppressed": row.get( + "telegram_provider_send_suppressed" + ), + } controls = { "retry_terminal_receipt_contract": bool( @@ -5053,7 +5089,7 @@ def _build_autonomous_retry_rollback_terminal_readback( ), "telegram_receipt_same_run": bool( row.get("telegram_message_id") - and row.get("telegram_send_status") == "sent" + and _durable_controlled_apply_telegram_receipt(telegram_receipt) and row.get("telegram_automation_run_id") == automation_run_id ), "public_safe_receipt": bool( @@ -8082,7 +8118,16 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """ km.automation_run_id AS km_automation_run_id, telegram.message_id::text AS telegram_message_id, telegram.send_status AS telegram_send_status, + telegram.provider_message_id AS telegram_provider_message_id, telegram.automation_run_id AS telegram_automation_run_id, + telegram.notification_policy_version + AS telegram_notification_policy_version, + telegram.notification_disposition + AS telegram_notification_disposition, + telegram.notification_provider_delivery + AS telegram_notification_provider_delivery, + telegram.provider_send_suppressed + AS telegram_provider_send_suppressed, replay.created_at AS retry_created_at FROM latest_retry replay JOIN automation_operation_log apply @@ -8181,17 +8226,50 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """ SELECT outbound.message_id, outbound.send_status, + outbound.provider_message_id, coalesce( outbound.source_envelope ->> 'automation_run_id', outbound.source_envelope #>> '{callback_reply,automation_run_id}', outbound.source_envelope #>> '{source_refs,automation_run_ids,0}' - ) AS automation_run_id + ) AS automation_run_id, + outbound.source_envelope #>> + '{notification_policy,policy_version}' + AS notification_policy_version, + outbound.source_envelope #>> + '{notification_policy,disposition}' + AS notification_disposition, + outbound.source_envelope #>> + '{notification_policy,provider_delivery}' + AS notification_provider_delivery, + outbound.source_envelope #>> + '{notification_policy,provider_send_performed}' + = 'false' AS provider_send_suppressed FROM awooop_outbound_message outbound WHERE outbound.project_id = :project_id AND outbound.channel_type = 'telegram' - AND outbound.send_status = 'sent' + AND ( + outbound.send_status = 'sent' + OR ( + outbound.send_status = 'shadow' + AND outbound.provider_message_id IS NULL + AND nullif( + outbound.source_envelope #>> + '{notification_policy,policy_version}', + '' + ) IS NOT NULL + AND outbound.source_envelope #>> + '{notification_policy,disposition}' + = 'suppressed' + AND outbound.source_envelope #>> + '{notification_policy,provider_delivery}' + IN ('shadow_only', 'state_transition', 'digest') + AND outbound.source_envelope #>> + '{notification_policy,provider_send_performed}' + = 'false' + ) + ) AND outbound.source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' AND coalesce( @@ -8665,6 +8743,7 @@ _RUNTIME_TELEGRAM_COUNTS_DIRECT_SQL = """ _RUNTIME_TELEGRAM_LATEST_SQL = """ + WITH ranked_telegram_receipts AS ( SELECT message_id::text AS message_id, run_id::text AS run_id, @@ -8684,8 +8763,25 @@ _RUNTIME_TELEGRAM_LATEST_SQL = """ source_envelope #>> '{callback_reply,action}', source_envelope #>> '{alert_notification,status}' ) AS action, + source_envelope #>> '{notification_policy,policy_version}' + AS notification_policy_version, + source_envelope #>> '{notification_policy,disposition}' + AS notification_disposition, + source_envelope #>> '{notification_policy,provider_delivery}' + AS notification_provider_delivery, + source_envelope #>> '{notification_policy,provider_send_performed}' + = 'false' AS notification_provider_send_suppressed, queued_at, - sent_at + sent_at, + row_number() OVER ( + PARTITION BY CASE + WHEN source_envelope #>> '{callback_reply,action}' + = 'controlled_apply_result' + THEN 'controlled_apply_result' + ELSE 'alert_notification_sent' + END + ORDER BY queued_at DESC + ) AS receipt_kind_rank FROM awooop_outbound_message WHERE project_id = :project_id AND channel_type = 'telegram' @@ -8693,12 +8789,16 @@ _RUNTIME_TELEGRAM_LATEST_SQL = """ source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent' ) + ) + SELECT * + FROM ranked_telegram_receipts + WHERE receipt_kind_rank <= :limit ORDER BY queued_at DESC - LIMIT :limit """ _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL = """ + WITH ranked_telegram_receipts AS ( SELECT message_id::text AS message_id, run_id::text AS run_id, @@ -8718,8 +8818,25 @@ _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL = """ source_envelope #>> '{callback_reply,action}', source_envelope #>> '{alert_notification,status}' ) AS action, + source_envelope #>> '{notification_policy,policy_version}' + AS notification_policy_version, + source_envelope #>> '{notification_policy,disposition}' + AS notification_disposition, + source_envelope #>> '{notification_policy,provider_delivery}' + AS notification_provider_delivery, + source_envelope #>> '{notification_policy,provider_send_performed}' + = 'false' AS notification_provider_send_suppressed, queued_at, - sent_at + sent_at, + row_number() OVER ( + PARTITION BY CASE + WHEN source_envelope #>> '{callback_reply,action}' + = 'controlled_apply_result' + THEN 'controlled_apply_result' + ELSE 'alert_notification_sent' + END + ORDER BY queued_at DESC + ) AS receipt_kind_rank FROM awooop_outbound_message WHERE project_id = $1 AND channel_type = 'telegram' @@ -8727,8 +8844,11 @@ _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL = """ source_envelope #>> '{callback_reply,action}' = 'controlled_apply_result' OR source_envelope #>> '{alert_notification,status}' = 'alert_notification_sent' ) + ) + SELECT * + FROM ranked_telegram_receipts + WHERE receipt_kind_rank <= $2 ORDER BY queued_at DESC - LIMIT $2 """ diff --git a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py index 19d99dc86..8e3ea3433 100644 --- a/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/src/services/iwooos_wazuh_controlled_executor_runtime.py @@ -562,6 +562,12 @@ _LIVE_RUNTIME_SQL = """ outbound.source_envelope #>> '{notification_policy,provider_delivery}' AS telegram_provider_delivery, + outbound.source_envelope #>> + '{notification_policy,policy_version}' + AS telegram_notification_policy_version, + outbound.source_envelope #>> + '{notification_policy,provider_send_performed}' + = 'false' AS telegram_provider_send_suppressed, outbound.sent_at AS telegram_sent_at FROM awooop_outbound_message outbound JOIN latest_candidate candidate @@ -1232,7 +1238,10 @@ def build_iwooos_wazuh_controlled_executor_runtime_readback( and data.get("telegram_execution_kind") == "no_write_posture_readback" and data.get("telegram_notification_disposition") == "suppressed" - and data.get("telegram_provider_delivery") == "shadow_only" + and str(data.get("telegram_notification_policy_version") or "") + and data.get("telegram_provider_delivery") + in {"shadow_only", "state_transition", "digest"} + and data.get("telegram_provider_send_suppressed") is True ) telegram_ready = telegram_provider_sent or telegram_notification_suppressed incident_ledger_closure_verified = bool( 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 ad3d2d255..f5b14215c 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -3578,6 +3578,99 @@ def test_runtime_telegram_receipt_queries_include_alert_notifications() -> None: assert "alert_notification_sent" in _RUNTIME_TELEGRAM_LATEST_SQL assert "controlled_apply_result" in _RUNTIME_TELEGRAM_COUNTS_SQL assert "controlled_apply_result" in _RUNTIME_TELEGRAM_LATEST_SQL + assert "PARTITION BY CASE" in _RUNTIME_TELEGRAM_LATEST_SQL + assert "receipt_kind_rank <= :limit" in _RUNTIME_TELEGRAM_LATEST_SQL + + +def test_same_run_suppressed_controlled_apply_shadow_is_durable_receipt() -> None: + candidate_op_id = "11111111-1111-4111-8111-111111111112" + automation_run_id = candidate_op_id + check_op_id = "22222222-2222-4222-8222-222222222223" + apply_op_id = "33333333-3333-4333-8333-333333333334" + incident_id = "INC-SUPPRESSED-TELEGRAM" + readback = build_runtime_receipt_readback_from_rows( + 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, + "source_candidate_op_id": candidate_op_id, + "check_mode_op_id": check_op_id, + "operation_type": "ansible_apply_executed", + "status": "success", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + }, + ], + verifier_latest_rows=[ + { + "id": "verifier-suppressed-telegram", + "apply_op_id": apply_op_id, + "automation_run_id": automation_run_id, + "verification_result": "success", + } + ], + km_latest_rows=[ + { + "id": "km-suppressed-telegram", + "path_type": f"ansible_apply_receipt:{apply_op_id[:8]}", + "automation_run_id": automation_run_id, + "status": "linked", + } + ], + telegram_latest_rows=[ + { + "message_id": "telegram-shadow-same-run", + "send_status": "shadow", + "provider_message_id": None, + "automation_run_id": automation_run_id, + "incident_id": incident_id, + "action": "controlled_apply_result", + "notification_policy_version": ( + "telegram_posture_lifecycle_transition_v1" + ), + "notification_disposition": "suppressed", + "notification_provider_delivery": "state_transition", + "notification_provider_send_suppressed": True, + } + ], + ) + + closure = readback["latest_flow_closure"] + assert closure["has_telegram_receipt"] is True + assert closure["missing"] == [] + ledger = readback["autonomous_execution_loop_ledger"] + telegram_stage = next( + stage for stage in ledger["stages"] if stage["stage_id"] == "telegram_receipt" + ) + assert telegram_stage["present"] is True + assert telegram_stage["status"] == "shadow" + assert telegram_stage["run_id_matches_expected"] is True + + +def test_unclassified_shadow_cannot_satisfy_telegram_receipt() -> None: + assert runtime_control_module._durable_controlled_apply_telegram_receipt( + { + "message_id": "telegram-shadow-unclassified", + "send_status": "shadow", + "provider_message_id": None, + "action": "controlled_apply_result", + } + ) is False def test_telegram_latest_dbapi_timeout_cannot_false_green_same_run_closure() -> None: @@ -4042,6 +4135,29 @@ def test_retry_rollback_terminal_closes_from_same_run_public_safe_receipts(): ] == 1 +def test_retry_terminal_accepts_policy_complete_suppressed_receipt() -> None: + row = _closed_retry_terminal_row() + row.update( + { + "telegram_send_status": "shadow", + "telegram_provider_message_id": None, + "telegram_notification_policy_version": ( + "telegram_failure_digest_v2" + ), + "telegram_notification_disposition": "suppressed", + "telegram_notification_provider_delivery": "digest", + "telegram_provider_send_suppressed": True, + } + ) + + terminal = build_runtime_receipt_readback_from_rows( + retry_terminal_latest_rows=[row] + )["autonomous_retry_rollback_terminal"] + + assert terminal["closed"] is True + assert terminal["controls"]["telegram_receipt_same_run"] is True + + def test_retry_rollback_terminal_fails_closed_on_run_mismatch_or_missing_proof(): row = _closed_retry_terminal_row() row["telegram_automation_run_id"] = "different-run" diff --git a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py index c27133f42..05d08fe1c 100644 --- a/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py +++ b/apps/api/tests/test_iwooos_wazuh_controlled_executor_runtime.py @@ -289,6 +289,10 @@ def test_wazuh_posture_shadow_receipt_closes_without_provider_send() -> None: "telegram_execution_kind": "no_write_posture_readback", "telegram_notification_disposition": "suppressed", "telegram_provider_delivery": "shadow_only", + "telegram_notification_policy_version": ( + "telegram_posture_state_digest_v1" + ), + "telegram_provider_send_suppressed": True, } ) @@ -310,6 +314,32 @@ def test_wazuh_posture_shadow_receipt_closes_without_provider_send() -> None: assert payload["boundaries"]["healthy_posture_notification_suppressed"] is True +def test_wazuh_posture_duplicate_lifecycle_shadow_is_terminal_receipt() -> None: + row = _complete_row() + row.update( + { + "telegram_send_status": "shadow", + "telegram_provider_message_id": None, + "telegram_execution_kind": "no_write_posture_readback", + "telegram_notification_disposition": "suppressed", + "telegram_provider_delivery": "state_transition", + "telegram_notification_policy_version": ( + "telegram_posture_lifecycle_transition_v1" + ), + "telegram_provider_send_suppressed": True, + } + ) + + payload = build_iwooos_wazuh_controlled_executor_runtime_readback( + row, + executor_enabled=True, + ) + + assert payload["summary"]["runtime_closed_count"] == 1 + assert payload["summary"]["telegram_receipt_count"] == 1 + assert payload["summary"]["telegram_provider_send_count"] == 0 + + def test_wazuh_runtime_readback_prefers_ingress_convergence_semantics() -> None: row = _complete_row() row["catalog_id"] = "ansible:wazuh-alertmanager-integration"