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 636c13ae7..2d02b3edc 100644 --- a/apps/api/src/services/ai_agent_autonomous_runtime_control.py +++ b/apps/api/src/services/ai_agent_autonomous_runtime_control.py @@ -3720,9 +3720,7 @@ def _autonomous_execution_loop_ledger( for row in telegram_rows if _durable_controlled_apply_telegram_receipt( row, - require_provider_receipt=( - catalog_id == _WAZUH_POSTURE_CATALOG_ID - ), + catalog_id=catalog_id, ) and ( not incident_id @@ -3943,29 +3941,122 @@ def _autonomous_execution_loop_ledger( def _durable_controlled_apply_telegram_receipt( row: Mapping[str, Any], *, - require_provider_receipt: bool = False, + catalog_id: str = "", + retry_terminal_receipt: bool = False, ) -> bool: """Accept provider sends or policy-complete no-send suppression receipts.""" if str(row.get("action") or "") != "controlled_apply_result": return False + wazuh_posture = catalog_id == _WAZUH_POSTURE_CATALOG_ID send_status = str(row.get("send_status") or "") if send_status == "sent": - return ( - bool(row.get("provider_message_id")) - if require_provider_receipt - else True + if not wazuh_posture: + return True + if not row.get("provider_message_id"): + return False + policy_version = str(row.get("notification_policy_version") or "") + provider_delivery = str( + row.get("notification_provider_delivery") or "" ) - if require_provider_receipt: + lifecycle_state = str(row.get("notification_lifecycle_state") or "") + if retry_terminal_receipt: + execution_kind = str( + row.get("notification_execution_kind") or "" + ) + return bool( + str(row.get("notification_disposition") or "") == "sent" + and lifecycle_state == "failed" + and ( + ( + execution_kind == "no_write_posture_readback" + and ( + ( + policy_version + == "telegram_posture_lifecycle_transition_v1" + and provider_delivery == "state_transition" + ) + or ( + policy_version + == "telegram_posture_state_digest_v1" + and provider_delivery == "digest" + ) + ) + ) + or ( + execution_kind == "no_write_replay" + and policy_version == "telegram_failure_digest_v2" + and provider_delivery == "digest" + ) + ) + ) + return bool( + policy_version == "telegram_posture_lifecycle_transition_v1" + and str(row.get("notification_disposition") or "") == "sent" + and provider_delivery == "state_transition" + and lifecycle_state == "recovered" + ) + if send_status != "shadow" or row.get("provider_message_id"): + return False + policy_version = str(row.get("notification_policy_version") or "") + provider_delivery = str(row.get("notification_provider_delivery") or "") + suppression_reason = str(row.get("notification_suppression_reason") or "") + generic_shadow = bool( + policy_version + and str(row.get("notification_disposition") or "") == "suppressed" + and provider_delivery in {"shadow_only", "state_transition", "digest"} + and row.get("notification_provider_send_suppressed") is True + ) + if not wazuh_posture: + return generic_shadow + lifecycle_state = str(row.get("notification_lifecycle_state") or "") + if retry_terminal_receipt: + if not generic_shadow or lifecycle_state != "failed": + return False + execution_kind = str(row.get("notification_execution_kind") or "") + return bool( + ( + execution_kind == "no_write_posture_readback" + and policy_version + == "telegram_posture_lifecycle_transition_v1" + and provider_delivery == "state_transition" + and suppression_reason == "duplicate_lifecycle_state" + ) + or ( + execution_kind == "no_write_posture_readback" + and policy_version == "telegram_posture_state_digest_v1" + and provider_delivery == "shadow_only" + and suppression_reason == "historical_projection_backfill" + ) + or ( + execution_kind == "no_write_replay" + and policy_version == "telegram_failure_digest_v2" + and ( + ( + provider_delivery == "digest" + and suppression_reason + == "duplicate_failure_fingerprint_window" + ) + or ( + provider_delivery == "shadow_only" + and suppression_reason == "historical_projection_backfill" + ) + ) + ) + ) + if not generic_shadow or lifecycle_state != "recovered": return False 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 + ( + policy_version == "telegram_posture_lifecycle_transition_v1" + and provider_delivery == "state_transition" + and suppression_reason == "duplicate_lifecycle_state" + ) + or ( + policy_version == "telegram_posture_state_digest_v1" + and provider_delivery == "shadow_only" + and suppression_reason == "historical_projection_backfill" + ) ) @@ -4038,9 +4129,7 @@ def _latest_flow_closure( for row in telegram_rows if _durable_controlled_apply_telegram_receipt( row, - require_provider_receipt=( - catalog_id == _WAZUH_POSTURE_CATALOG_ID - ), + catalog_id=catalog_id, ) and ( not incident_id @@ -5041,6 +5130,15 @@ def _build_autonomous_retry_rollback_terminal_readback( "notification_provider_delivery": row.get( "telegram_notification_provider_delivery" ), + "notification_execution_kind": row.get( + "telegram_notification_execution_kind" + ), + "notification_lifecycle_state": row.get( + "telegram_notification_lifecycle_state" + ), + "notification_suppression_reason": row.get( + "telegram_notification_suppression_reason" + ), "notification_provider_send_suppressed": row.get( "telegram_provider_send_suppressed" ), @@ -5109,7 +5207,11 @@ def _build_autonomous_retry_rollback_terminal_readback( ), "telegram_receipt_same_run": bool( row.get("telegram_message_id") - and _durable_controlled_apply_telegram_receipt(telegram_receipt) + and _durable_controlled_apply_telegram_receipt( + telegram_receipt, + catalog_id=str(row.get("catalog_id") or ""), + retry_terminal_receipt=True, + ) and row.get("telegram_automation_run_id") == automation_run_id ), "public_safe_receipt": bool( @@ -8146,6 +8248,12 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """ AS telegram_notification_disposition, telegram.notification_provider_delivery AS telegram_notification_provider_delivery, + telegram.notification_execution_kind + AS telegram_notification_execution_kind, + telegram.notification_lifecycle_state + AS telegram_notification_lifecycle_state, + telegram.notification_suppression_reason + AS telegram_notification_suppression_reason, telegram.provider_send_suppressed AS telegram_provider_send_suppressed, replay.created_at AS retry_created_at @@ -8263,6 +8371,17 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """ outbound.source_envelope #>> '{notification_policy,provider_delivery}' AS notification_provider_delivery, + coalesce( + outbound.source_envelope #>> + '{callback_reply,execution_kind}', + outbound.source_envelope ->> 'execution_kind' + ) AS notification_execution_kind, + outbound.source_envelope #>> + '{notification_policy,lifecycle_state}' + AS notification_lifecycle_state, + outbound.source_envelope #>> + '{notification_policy,suppression_reason}' + AS notification_suppression_reason, outbound.source_envelope #>> '{notification_policy,provider_send_performed}' = 'false' AS provider_send_suppressed @@ -8270,7 +8389,10 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """ WHERE outbound.project_id = :project_id AND outbound.channel_type = 'telegram' AND ( - outbound.send_status = 'sent' + ( + outbound.send_status = 'sent' + AND outbound.provider_message_id IS NOT NULL + ) OR ( outbound.send_status = 'shadow' AND outbound.provider_message_id IS NULL @@ -8302,6 +8424,17 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """ replay.input ->> 'automation_run_id', apply.input ->> 'automation_run_id' ) + AND coalesce( + outbound.source_envelope #>> + '{callback_reply,incident_id}', + outbound.source_envelope #>> + '{source_refs,incident_ids,0}' + ) = coalesce( + apply.incident_id::text, + apply.input ->> 'incident_id' + ) + AND outbound.source_envelope #>> + '{callback_reply,apply_op_id}' = apply.op_id::text ORDER BY outbound.sent_at DESC NULLS LAST, outbound.queued_at DESC LIMIT 1 ) telegram ON TRUE @@ -8789,6 +8922,10 @@ _RUNTIME_TELEGRAM_LATEST_SQL = """ AS notification_disposition, source_envelope #>> '{notification_policy,provider_delivery}' AS notification_provider_delivery, + source_envelope #>> '{notification_policy,lifecycle_state}' + AS notification_lifecycle_state, + source_envelope #>> '{notification_policy,suppression_reason}' + AS notification_suppression_reason, source_envelope #>> '{notification_policy,provider_send_performed}' = 'false' AS notification_provider_send_suppressed, queued_at, @@ -8844,6 +8981,10 @@ _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL = """ AS notification_disposition, source_envelope #>> '{notification_policy,provider_delivery}' AS notification_provider_delivery, + source_envelope #>> '{notification_policy,lifecycle_state}' + AS notification_lifecycle_state, + source_envelope #>> '{notification_policy,suppression_reason}' + AS notification_suppression_reason, source_envelope #>> '{notification_policy,provider_send_performed}' = 'false' AS notification_provider_send_suppressed, queued_at, 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 b239c0fca..4b87d9f56 100644 --- a/apps/api/tests/test_ai_agent_autonomous_runtime_control.py +++ b/apps/api/tests/test_ai_agent_autonomous_runtime_control.py @@ -3580,13 +3580,24 @@ def test_runtime_telegram_receipt_queries_include_alert_notifications() -> None: 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 + for sql in ( + _RUNTIME_TELEGRAM_LATEST_SQL, + runtime_control_module._RUNTIME_TELEGRAM_LATEST_DIRECT_SQL, + runtime_control_module._RUNTIME_RETRY_TERMINAL_LATEST_SQL, + ): + assert "notification_lifecycle_state" in sql + assert "notification_suppression_reason" in sql + retry_sql = runtime_control_module._RUNTIME_RETRY_TERMINAL_LATEST_SQL + assert "notification_execution_kind" in retry_sql + assert "{callback_reply,apply_op_id}" in retry_sql + assert "{callback_reply,incident_id}" in retry_sql @pytest.mark.parametrize( ("catalog_id", "shadow_completes"), [ ("ansible:generic-posture-readback", True), - ("ansible:wazuh-manager-posture-readback", False), + ("ansible:wazuh-manager-posture-readback", True), ], ) def test_same_run_suppressed_shadow_respects_provider_receipt_policy( @@ -3658,6 +3669,8 @@ def test_same_run_suppressed_shadow_respects_provider_receipt_policy( ), "notification_disposition": "suppressed", "notification_provider_delivery": "state_transition", + "notification_lifecycle_state": "recovered", + "notification_suppression_reason": "duplicate_lifecycle_state", "notification_provider_send_suppressed": True, } ], @@ -3677,6 +3690,235 @@ def test_same_run_suppressed_shadow_respects_provider_receipt_policy( assert telegram_stage["run_id_matches_expected"] is shadow_completes +@pytest.mark.parametrize( + ( + "send_status", + "policy_version", + "provider_delivery", + "lifecycle_state", + "suppression_reason", + "expected", + ), + [ + ( + "shadow", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "recovered", + "duplicate_lifecycle_state", + True, + ), + ( + "shadow", + "telegram_posture_state_digest_v1", + "shadow_only", + "recovered", + "historical_projection_backfill", + True, + ), + ( + "shadow", + "telegram_posture_state_digest_v1", + "digest", + "failed", + "duplicate_failure_fingerprint_window", + False, + ), + ( + "shadow", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "recovered", + "cooldown_active", + False, + ), + ( + "sent", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "recovered", + "", + True, + ), + ( + "sent", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "failed", + "", + False, + ), + ], +) +def test_wazuh_apply_receipt_requires_exact_recovered_lifecycle_provenance( + send_status: str, + policy_version: str, + provider_delivery: str, + lifecycle_state: str, + suppression_reason: str, + expected: bool, +) -> None: + assert runtime_control_module._durable_controlled_apply_telegram_receipt( + { + "action": "controlled_apply_result", + "send_status": send_status, + "provider_message_id": "42" if send_status == "sent" else None, + "notification_policy_version": policy_version, + "notification_disposition": ( + "sent" if send_status == "sent" else "suppressed" + ), + "notification_provider_delivery": provider_delivery, + "notification_lifecycle_state": lifecycle_state, + "notification_suppression_reason": suppression_reason, + "notification_provider_send_suppressed": send_status == "shadow", + }, + catalog_id="ansible:wazuh-manager-posture-readback", + ) is expected + + +def test_wazuh_duplicate_lifecycle_shadow_closes_strict_same_run_eighteen_stage_loop() -> ( + None +): + automation_run_id = "wazuh-candidate-op" + incident_id = "INC-WAZUH-SHADOW-STRICT-CLOSURE" + apply_op_id = "wazuh-apply-op" + catalog_id = "ansible:wazuh-manager-posture-readback" + non_execution_stages = set(AI_AUTOMATION_REQUIRED_LOOP_STAGES) - set( + AI_AUTOMATION_EXECUTION_CORRELATED_STAGES + ) + + def stage_receipt(stage_id: str) -> dict[str, object]: + receipt: dict[str, object] = { + "schema_version": "ai_automation_stage_receipt_v1", + "stage_id": stage_id, + "automation_run_id": automation_run_id, + "evidence_ref": f"receipt:{stage_id}", + "durable_receipt": True, + } + if stage_id == "rag_writeback": + receipt["detail"] = { + "durable_write_acknowledged": True, + "source_id": f"ansible-learning:{apply_op_id}", + "rag_chunk_count": 1, + "chunk_index_verified": True, + } + elif stage_id == "playbook_trust": + receipt["detail"] = { + "schema_version": "ansible_playbook_trust_writeback_v2", + "identity_schema_version": "ansible_playbook_identity_v1", + "identity_fingerprint": "a" * 64, + "canonical_playbook_id": "PB-WAZUH-MANAGER-POSTURE", + "playbook_row_version": 1, + "playbook_row_updated_at": "2026-07-22T12:00:00+00:00", + "playbook_row_fingerprint": "b" * 64, + "trust_score": 1.0, + "trust_observation_count": 1, + "learning_recorded": True, + "trust_updated": True, + "repository_write_acknowledged": True, + "repository_readback_verified": True, + "operation_receipt_readback_verified": True, + "durable_write_acknowledged": True, + "writer_source_sha": "6b87cf6d7", + "raw_log_payload_stored": False, + "secret_value_stored": False, + } + return receipt + + readback = build_runtime_receipt_readback_from_rows( + operation_latest_rows=[ + { + "op_id": automation_run_id, + "operation_type": "ansible_candidate_matched", + "status": "dry_run", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + "catalog_id": catalog_id, + }, + { + "op_id": "wazuh-check-op", + "parent_op_id": automation_run_id, + "operation_type": "ansible_check_mode_executed", + "status": "success", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + "catalog_id": catalog_id, + "returncode": "0", + }, + { + "op_id": apply_op_id, + "parent_op_id": "wazuh-check-op", + "source_candidate_op_id": automation_run_id, + "check_mode_op_id": "wazuh-check-op", + "operation_type": "ansible_apply_executed", + "status": "success", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + "catalog_id": catalog_id, + "returncode": "0", + "runtime_stage_receipts": [ + stage_receipt(stage_id) for stage_id in non_execution_stages + ], + }, + ], + auto_repair_latest_rows=[ + { + "id": "wazuh-auto-repair", + "incident_id": incident_id, + "catalog_id": catalog_id, + "result_status": "success", + "automation_run_id": automation_run_id, + "executed_steps_text": f'["apply:{apply_op_id}"]', + } + ], + verifier_latest_rows=[ + { + "id": "wazuh-verifier", + "incident_id": incident_id, + "apply_op_id": apply_op_id, + "automation_run_id": automation_run_id, + "verification_result": "success", + } + ], + km_latest_rows=[ + { + "id": "wazuh-km", + "related_incident_id": incident_id, + "path_type": f"ansible_apply_receipt:{apply_op_id[:8]}", + "automation_run_id": automation_run_id, + "status": "linked", + } + ], + telegram_latest_rows=[ + { + "message_id": "wazuh-telegram-shadow", + "automation_run_id": automation_run_id, + "incident_id": incident_id, + "action": "controlled_apply_result", + "send_status": "shadow", + "provider_message_id": None, + "notification_policy_version": ( + "telegram_posture_lifecycle_transition_v1" + ), + "notification_disposition": "suppressed", + "notification_provider_delivery": "state_transition", + "notification_lifecycle_state": "recovered", + "notification_suppression_reason": "duplicate_lifecycle_state", + "notification_provider_send_suppressed": True, + } + ], + ) + + strict = build_ai_agent_strict_runtime_completion_from_readback(readback) + + assert strict["required_stage_count"] == 18 + assert strict["present_stage_count"] == 18 + assert strict["missing_stage_ids"] == [] + assert strict["full_trace_same_run_correlation_proven"] is True + assert strict["completion_percent"] == 100 + assert strict["closed"] is True + + def test_unclassified_shadow_cannot_satisfy_telegram_receipt() -> None: assert runtime_control_module._durable_controlled_apply_telegram_receipt( { @@ -4173,6 +4415,113 @@ def test_retry_terminal_accepts_policy_complete_suppressed_receipt() -> None: assert terminal["controls"]["telegram_receipt_same_run"] is True +@pytest.mark.parametrize( + ( + "execution_kind", + "send_status", + "policy_version", + "provider_delivery", + "lifecycle_state", + "suppression_reason", + "expected_closed", + ), + [ + ( + "no_write_posture_readback", + "sent", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "failed", + "", + True, + ), + ( + "no_write_posture_readback", + "shadow", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "failed", + "duplicate_lifecycle_state", + True, + ), + ( + "no_write_replay", + "sent", + "telegram_failure_digest_v2", + "digest", + "failed", + "", + True, + ), + ( + "no_write_replay", + "shadow", + "telegram_failure_digest_v2", + "digest", + "failed", + "duplicate_failure_fingerprint_window", + True, + ), + ( + "no_write_replay", + "shadow", + "telegram_failure_digest_v2", + "shadow_only", + "failed", + "historical_projection_backfill", + True, + ), + ( + "no_write_posture_readback", + "shadow", + "telegram_posture_lifecycle_transition_v1", + "state_transition", + "recovered", + "duplicate_lifecycle_state", + False, + ), + ], +) +def test_wazuh_retry_terminal_requires_exact_failed_lifecycle_receipt( + execution_kind: str, + send_status: str, + policy_version: str, + provider_delivery: str, + lifecycle_state: str, + suppression_reason: str, + expected_closed: bool, +) -> None: + row = _closed_retry_terminal_row() + row.update( + { + "catalog_id": "ansible:wazuh-manager-posture-readback", + "telegram_send_status": send_status, + "telegram_provider_message_id": ( + "wazuh-provider-message" if send_status == "sent" else None + ), + "telegram_notification_execution_kind": execution_kind, + "telegram_notification_policy_version": policy_version, + "telegram_notification_disposition": ( + "sent" if send_status == "sent" else "suppressed" + ), + "telegram_notification_provider_delivery": provider_delivery, + "telegram_notification_lifecycle_state": lifecycle_state, + "telegram_notification_suppression_reason": suppression_reason, + "telegram_provider_send_suppressed": send_status == "shadow", + } + ) + + terminal = build_runtime_receipt_readback_from_rows( + retry_terminal_latest_rows=[row] + )["autonomous_retry_rollback_terminal"] + + assert terminal["closed"] is expected_closed + assert terminal["controls"]["telegram_receipt_same_run"] is expected_closed + assert ( + "telegram_receipt_same_run_not_verified" in terminal["active_blockers"] + ) is not expected_closed + + 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"