diff --git a/README.md b/README.md
index c58d903c08..76e6534551 100644
--- a/README.md
+++ b/README.md
@@ -123,12 +123,12 @@ Every AWS provider scan will enqueue an Attack Paths ingestion job automatically
 
 | Provider | Checks | Services | [Compliance Frameworks](https://docs.prowler.com/user-guide/compliance/tutorials/compliance) | [Categories](https://docs.prowler.com/user-guide/cli/tutorials/misc#categories) | Support | Interface |
 |---|---|---|---|---|---|---|
-| AWS | 615 | 86 | 47 | 19 | Official | UI, API, CLI |
-| Azure | 190 | 22 | 21 | 16 | Official | UI, API, CLI |
+| AWS | 621 | 86 | 47 | 19 | Official | UI, API, CLI |
+| Azure | 191 | 22 | 21 | 16 | Official | UI, API, CLI |
 | GCP | 109 | 20 | 19 | 12 | Official | UI, API, CLI |
-| Kubernetes | 90 | 7 | 8 | 11 | Official | UI, API, CLI |
+| Kubernetes | 92 | 7 | 8 | 11 | Official | UI, API, CLI |
 | GitHub | 24 | 3 | 2 | 5 | Official | UI, API, CLI |
-| M365 | 109 | 10 | 6 | 10 | Official | UI, API, CLI |
+| M365 | 111 | 10 | 6 | 10 | Official | UI, API, CLI |
 | OCI | 52 | 14 | 5 | 10 | Official | UI, API, CLI |
 | Alibaba Cloud | 63 | 9 | 6 | 9 | Official | UI, API, CLI |
 | Cloudflare | 29 | 3 | 2 | 5 | Official | UI, API, CLI |
diff --git a/api/changelog.d/compliance-overview-single-transaction.changed.md b/api/changelog.d/compliance-overview-single-transaction.changed.md
new file mode 100644
index 0000000000..0e7a6714b3
--- /dev/null
+++ b/api/changelog.d/compliance-overview-single-transaction.changed.md
@@ -0,0 +1 @@
+Compliance overview ingest now runs in a single transaction per scan with a configurable `COPY` batch size (`DJANGO_COMPLIANCE_COPY_BATCH_SIZE`, default 2000), reducing write pressure on the database
diff --git a/api/src/backend/tasks/jobs/scan.py b/api/src/backend/tasks/jobs/scan.py
index 39db32abe4..36dc806ff3 100644
--- a/api/src/backend/tasks/jobs/scan.py
+++ b/api/src/backend/tasks/jobs/scan.py
@@ -6,7 +6,7 @@ import re
 import time
 import uuid
 from collections import defaultdict
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from datetime import UTC, datetime
 from typing import Any
 
@@ -49,6 +49,7 @@ from celery.utils.log import get_task_logger
 from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
 from config.env import env
 from config.settings.celery import CELERY_DEADLOCK_ATTEMPTS
+from django.core.exceptions import ImproperlyConfigured
 from django.db import DatabaseError, IntegrityError, OperationalError, transaction
 from django.db.models import (
     Case,
@@ -99,6 +100,16 @@ COMPLIANCE_REQUIREMENT_COPY_COLUMNS = (
 FINDINGS_MICRO_BATCH_SIZE = env.int("DJANGO_FINDINGS_MICRO_BATCH_SIZE", default=3000)
 # Controls how many rows each ORM bulk_create/bulk_update call sends to Postgres.
 SCAN_DB_BATCH_SIZE = env.int("DJANGO_SCAN_DB_BATCH_SIZE", default=1000)
+# Rows per COPY statement when ingesting compliance requirement overviews. All
+# batches of a scan share one transaction/commit; the batch size only bounds the
+# client-side CSV buffer and how long each individual COPY statement runs on the
+# writer (memory footprint, lock time and slow-statement logging under load).
+COMPLIANCE_COPY_BATCH_SIZE = env.int("DJANGO_COMPLIANCE_COPY_BATCH_SIZE", default=2000)
+if COMPLIANCE_COPY_BATCH_SIZE < 1:
+    raise ImproperlyConfigured(
+        "DJANGO_COMPLIANCE_COPY_BATCH_SIZE must be a positive integer, got "
+        f"{COMPLIANCE_COPY_BATCH_SIZE}"
+    )
 # Throttle scan progress persistence: minimum progress delta (fraction 0-1)
 # between two persisted progress updates.
 PROGRESS_THROTTLE_DELTA = env.float("DJANGO_SCAN_PROGRESS_THROTTLE_DELTA", default=0.01)
@@ -356,30 +367,36 @@ def _bulk_update_resource_failed_findings_counts(
                 raise
 
 
-def _copy_compliance_requirement_rows(
-    tenant_id: str, rows: list[dict[str, Any]]
-) -> None:
-    """Stream compliance requirement rows into Postgres using COPY.
+class ComplianceRowScopeError(ValueError):
+    """A compliance requirement row does not belong to the scan being ingested."""
 
-    We leverage the admin connection (when available) to bypass the COPY + RLS
-    restriction, writing only the fields required by
-    ``ComplianceRequirementOverview``.
 
-    Args:
-        tenant_id: Target tenant UUID.
-        rows: List of row dictionaries prepared by
-            :func:`create_compliance_requirements`.
+def _compliance_requirement_rows_to_csv(
+    rows: list[dict[str, Any]], tenant_id: str, scan_id: str
+) -> io.StringIO:
+    """Serialize compliance requirement rows into a CSV buffer for COPY.
+
+    COPY runs on the admin connection, which bypasses RLS, so every row is
+    checked against the expected tenant/scan before it is written: a mismatched
+    row would otherwise be inserted verbatim into another tenant's data.
     """
-
     csv_buffer = io.StringIO()
     writer = csv.writer(csv_buffer)
 
     datetime_now = datetime.now(tz=UTC)
     for row in rows:
+        row_tenant_id = str(row.get("tenant_id"))
+        row_scan_id = str(row.get("scan_id"))
+        if row_tenant_id != tenant_id or row_scan_id != scan_id:
+            raise ComplianceRowScopeError(
+                "Compliance requirement row does not belong to the scan being "
+                f"ingested (expected tenant {tenant_id} / scan {scan_id}, got "
+                f"tenant {row_tenant_id} / scan {row_scan_id})"
+            )
         writer.writerow(
             [
                 str(row.get("id")),
-                str(row.get("tenant_id")),
+                row_tenant_id,
                 (row.get("inserted_at") or datetime_now).isoformat(),
                 row.get("compliance_id") or "",
                 row.get("framework") or "",
@@ -393,65 +410,100 @@ def _copy_compliance_requirement_rows(
                 row.get("total_checks", 0),
                 row.get("passed_findings", 0),
                 row.get("total_findings", 0),
-                str(row.get("scan_id")),
+                row_scan_id,
             ]
         )
 
     csv_buffer.seek(0)
+    return csv_buffer
+
+
+def _copy_compliance_requirement_rows(
+    tenant_id: str, scan_id: str, rows: Iterable[dict[str, Any]], batch_size: int
+) -> int:
+    """Replace a scan's compliance requirement rows using batched COPY.
+
+    We leverage the admin connection (when available) to bypass the COPY + RLS
+    restriction. The scan's DELETE and every COPY batch run on one connection
+    inside a single transaction with a single commit, so the writer takes one
+    fsync per scan instead of one per batch, and a failed ingest rolls back
+    without committing a partial delete/insert (which a retry would otherwise
+    delete again, feeding dead rows to autovacuum).
+
+    Args:
+        tenant_id: Target tenant UUID.
+        scan_id: Scan whose previous rows are replaced.
+        rows: Iterable of row dictionaries, consumed lazily batch by batch.
+        batch_size: Number of rows per COPY statement.
+
+    Returns:
+        int: total number of rows staged and committed.
+
+    Raises:
+        ComplianceRowScopeError: A row belongs to another tenant or scan.
+    """
+    # Normalized once so the per-row scope check compares like with like even if
+    # the caller passes UUID instances instead of strings.
+    tenant_id = str(tenant_id)
+    scan_id = str(scan_id)
+    total_rows = 0
+    batch_num = 0
     copy_sql = (
         "COPY compliance_requirements_overviews ("
         + ", ".join(COMPLIANCE_REQUIREMENT_COPY_COLUMNS)
         + ") FROM STDIN WITH (FORMAT CSV, DELIMITER ',', QUOTE '\"', ESCAPE '\"', NULL '\\N')"
     )
 
-    try:
-        with psycopg_connection(MainRouter.admin_db) as connection:
-            connection.autocommit = False
-            try:
-                with connection.cursor() as cursor:
-                    cursor.execute(SET_CONFIG_QUERY, [POSTGRES_TENANT_VAR, tenant_id])
-                    cursor.copy_expert(copy_sql, csv_buffer)
-                connection.commit()
-            except Exception:
-                connection.rollback()
-                raise
-    finally:
-        csv_buffer.close()
+    with psycopg_connection(MainRouter.admin_db) as connection:
+        connection.autocommit = False
+        try:
+            with connection.cursor() as cursor:
+                cursor.execute(SET_CONFIG_QUERY, [POSTGRES_TENANT_VAR, tenant_id])
+                # Idempotent re-run: clearing this scan's rows inside the same
+                # transaction keeps delete + reinsert atomic.
+                cursor.execute(
+                    "DELETE FROM compliance_requirements_overviews "
+                    "WHERE tenant_id = %s AND scan_id = %s",
+                    [tenant_id, scan_id],
+                )
+                for batch, _is_last in batched(rows, batch_size):
+                    if not batch:
+                        continue
+                    batch_num += 1
+                    csv_buffer = _compliance_requirement_rows_to_csv(
+                        batch, tenant_id, scan_id
+                    )
+                    try:
+                        cursor.copy_expert(copy_sql, csv_buffer)
+                    finally:
+                        csv_buffer.close()
+                    total_rows += len(batch)
+                    logger.info(
+                        f"Compliance COPY batch {batch_num}: staged {len(batch)} rows "
+                        f"({total_rows} total)"
+                    )
+            connection.commit()
+        except Exception:
+            connection.rollback()
+            raise
+
+    return total_rows
 
 
-def _persist_compliance_requirement_rows(
-    tenant_id: str, rows: Iterable[dict[str, Any]], batch_size: int = 10000
+def _bulk_create_compliance_requirement_rows(
+    tenant_id: str, scan_id: str, rows: Iterable[dict[str, Any]], batch_size: int
 ) -> int:
-    """Persist compliance requirement rows using batched COPY with ORM fallback.
+    """Replace a scan's compliance requirement rows via the ORM.
 
-    ``rows`` is consumed lazily in batches, so peak memory stays at ~``batch_size``
-    rows instead of the full set. A batch that fails COPY falls back to an ORM
-    ``bulk_create`` of just that batch.
-
-    Args:
-        tenant_id: Target tenant UUID.
-        rows: Iterable of row dictionaries reflecting the compliance overview
-            state for a scan.
-        batch_size: Number of rows per COPY batch (default: 10000).
-
-    Returns:
-        int: total number of rows persisted.
+    Fallback for when COPY is unavailable; the delete and every ``bulk_create``
+    share one RLS transaction so the replacement stays atomic.
     """
     total_rows = 0
-    batch_num = 0
-
-    for batch, _is_last in batched(rows, batch_size):
-        if not batch:
-            continue
-        batch_num += 1
-        try:
-            _copy_compliance_requirement_rows(tenant_id, batch)
-        except Exception as error:
-            logger.exception(
-                f"COPY bulk insert for compliance requirements batch {batch_num} "
-                "failed; falling back to ORM bulk_create for this batch",
-                exc_info=error,
-            )
+    with rls_transaction(tenant_id):
+        ComplianceRequirementOverview.objects.filter(scan_id=scan_id).delete()
+        for batch, _is_last in batched(rows, batch_size):
+            if not batch:
+                continue
             fallback_objects = [
                 ComplianceRequirementOverview(
                     id=row["id"],
@@ -473,20 +525,58 @@ def _persist_compliance_requirement_rows(
                 )
                 for row in batch
             ]
-            with rls_transaction(tenant_id):
-                ComplianceRequirementOverview.objects.bulk_create(
-                    fallback_objects, batch_size=500
-                )
-
-        total_rows += len(batch)
-        logger.info(
-            f"Compliance COPY batch {batch_num}: inserted {len(batch)} rows "
-            f"({total_rows} total)"
-        )
-
+            ComplianceRequirementOverview.objects.bulk_create(
+                fallback_objects, batch_size=500
+            )
+            total_rows += len(batch)
     return total_rows
 
 
+def _persist_compliance_requirement_rows(
+    tenant_id: str,
+    scan_id: str,
+    rows_factory: Callable[[], Iterable[dict[str, Any]]],
+    batch_size: int | None = None,
+) -> int:
+    """Persist a scan's compliance requirement rows, replacing any previous ones.
+
+    ``rows_factory`` must return a fresh row iterator on every call: the COPY
+    path consumes it lazily in batches (peak memory ~``batch_size`` rows), and
+    if COPY fails the whole ingest falls back to a single ORM transaction that
+    re-iterates the rows.
+
+    Args:
+        tenant_id: Target tenant UUID.
+        scan_id: Scan whose compliance overview rows are being replaced.
+        rows_factory: Callable returning an iterable of row dictionaries.
+        batch_size: Rows per COPY/bulk_create batch (default:
+            ``COMPLIANCE_COPY_BATCH_SIZE``).
+
+    Returns:
+        int: total number of rows persisted.
+    """
+    if batch_size is None:
+        batch_size = COMPLIANCE_COPY_BATCH_SIZE
+
+    try:
+        return _copy_compliance_requirement_rows(
+            tenant_id, scan_id, rows_factory(), batch_size
+        )
+    except ComplianceRowScopeError:
+        # Cross-tenant/scan rows are a bug in the caller, not a COPY failure:
+        # retrying through the ORM would persist the very rows we rejected.
+        raise
+    except Exception as error:
+        logger.exception(
+            "COPY bulk insert for compliance requirements failed; "
+            "falling back to ORM bulk_create",
+            exc_info=error,
+        )
+        return _bulk_create_compliance_requirement_rows(
+            tenant_id, scan_id, rows_factory(), batch_size
+        )
+
+
 def _create_compliance_summaries(
     tenant_id: str, scan_id: str, requirement_statuses: dict
 ) -> None:
@@ -885,15 +975,19 @@ def _process_finding_micro_batch(
                         # Denormalized resource arrays populated directly on insert
                         # (was previously a separate bulk_update; saves a CASE WHEN
                         # over thousands of rows per micro-batch).
-                        resource_regions=[resource_instance.region]
-                        if resource_instance.region
-                        else [],
-                        resource_services=[resource_instance.service]
-                        if resource_instance.service
-                        else [],
-                        resource_types=[resource_instance.type]
-                        if resource_instance.type
-                        else [],
+                        resource_regions=(
+                            [resource_instance.region]
+                            if resource_instance.region
+                            else []
+                        ),
+                        resource_services=(
+                            [resource_instance.service]
+                            if resource_instance.service
+                            else []
+                        ),
+                        resource_types=(
+                            [resource_instance.type] if resource_instance.type else []
+                        ),
                     )
                     findings_to_create.append(finding_instance)
                     resource_denormalized_data.append(
@@ -1708,8 +1802,10 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                 )
 
             # Yield rows lazily (consumed batch-by-batch by COPY) so peak memory
-            # stays bounded; tally requirement_statuses in the same pass.
+            # stays bounded; tally requirement_statuses in the same pass. The
+            # ORM fallback re-iterates from scratch, so the tally resets first.
             def _iter_compliance_requirement_rows():
+                requirement_statuses.clear()
                 for region in regions:
                     region_stats = region_requirement_stats.get(region, {})
                     region_findings = findings_count_by_compliance.get(region, {})
@@ -1773,12 +1869,10 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                                 "total_findings": total_findings,
                             }
 
-            # Idempotent re-run: clear this scan's rows before re-inserting.
-            with rls_transaction(tenant_id):
-                ComplianceRequirementOverview.objects.filter(scan_id=scan_id).delete()
-
+            # The delete of the scan's previous rows happens inside the same
+            # transaction as the inserts (see _copy_compliance_requirement_rows).
             requirements_created = _persist_compliance_requirement_rows(
-                tenant_id, _iter_compliance_requirement_rows()
+                tenant_id_str, scan_id_str, _iter_compliance_requirement_rows
             )
 
         # Create pre-aggregated summaries for fast compliance overview lookups
diff --git a/api/src/backend/tasks/tests/test_scan.py b/api/src/backend/tasks/tests/test_scan.py
index 2a66985276..cad5e3d343 100644
--- a/api/src/backend/tasks/tests/test_scan.py
+++ b/api/src/backend/tasks/tests/test_scan.py
@@ -26,6 +26,7 @@ from prowler.lib.check.models import Severity
 from prowler.lib.outputs.finding import Status
 from tasks.jobs.scan import (
     _ATTACK_SURFACE_MAPPING_CACHE,
+    ComplianceRowScopeError,
     _aggregate_findings_by_region,
     _bulk_update_resource_failed_findings_counts,
     _copy_compliance_requirement_rows,
@@ -2314,9 +2315,9 @@ class TestCreateComplianceRequirements:
             create_compliance_requirements(tenant_id, scan_id)
 
             mock_persist.assert_called_once()
-            persisted_rows = mock_persist.call_args[0][1]
+            rows_factory = mock_persist.call_args[0][2]
             requirement_row = next(
-                row for row in persisted_rows if row["requirement_id"] == "1.1"
+                row for row in rows_factory() if row["requirement_id"] == "1.1"
             )
             assert requirement_row["requirement_status"] == "FAIL"
 
@@ -2454,18 +2455,26 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(
+                str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+            )
 
         mock_psycopg_connection.assert_called_once_with("admin")
         connection.cursor.assert_called_once()
-        cursor.execute.assert_called_once()
+        # One execute for set_config plus one for the scan's DELETE.
+        assert cursor.execute.call_count == 2
+        delete_sql, delete_params = cursor.execute.call_args_list[1][0]
+        assert "DELETE FROM compliance_requirements_overviews" in delete_sql
+        assert delete_params == [str(row["tenant_id"]), str(row["scan_id"])]
         cursor.copy_expert.assert_called_once()
+        connection.commit.assert_called_once()
 
         csv_rows = list(csv.reader(StringIO(captured["data"])))
         assert csv_rows[0][0] == str(row["id"])
         assert csv_rows[0][5] == ""
         assert csv_rows[0][-1] == str(row["scan_id"])
 
+    @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.filter")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -2473,7 +2482,7 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("copy failed"),
     )
     def test_persist_compliance_requirement_rows_fallback(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create
+        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_filter
     ):
         inserted_at = datetime.now(UTC)
         row = {
@@ -2494,16 +2503,22 @@ class TestComplianceRequirementCopy:
         }
 
         tenant_id = row["tenant_id"]
+        scan_id = str(row["scan_id"])
 
         ctx = MagicMock()
         ctx.__enter__.return_value = None
         ctx.__exit__.return_value = False
         mock_rls_transaction.return_value = ctx
 
-        _persist_compliance_requirement_rows(tenant_id, [row])
+        _persist_compliance_requirement_rows(tenant_id, scan_id, lambda: [row])
 
-        mock_copy.assert_called_once_with(tenant_id, [row])
+        mock_copy.assert_called_once()
+        assert mock_copy.call_args[0][0] == tenant_id
+        assert mock_copy.call_args[0][1] == scan_id
         mock_rls_transaction.assert_called_once_with(tenant_id)
+        # The fallback replaces the scan's rows: delete + insert atomically.
+        mock_filter.assert_called_once_with(scan_id=scan_id)
+        mock_filter.return_value.delete.assert_called_once()
         mock_bulk_create.assert_called_once()
 
         args, kwargs = mock_bulk_create.call_args
@@ -2515,13 +2530,18 @@ class TestComplianceRequirementCopy:
 
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
-    @patch("tasks.jobs.scan._copy_compliance_requirement_rows")
+    @patch("tasks.jobs.scan._copy_compliance_requirement_rows", return_value=0)
     def test_persist_compliance_requirement_rows_no_rows(
         self, mock_copy, mock_rls_transaction, mock_bulk_create
     ):
-        _persist_compliance_requirement_rows(str(uuid.uuid4()), [])
+        # Even with no rows the COPY path runs: it must clear the scan's
+        # previous rows so a re-run with fewer findings drops stale data.
+        total = _persist_compliance_requirement_rows(
+            str(uuid.uuid4()), str(uuid.uuid4()), lambda: []
+        )
 
-        mock_copy.assert_not_called()
+        assert total == 0
+        mock_copy.assert_called_once()
         mock_rls_transaction.assert_not_called()
         mock_bulk_create.assert_not_called()
 
@@ -2610,11 +2630,12 @@ class TestComplianceRequirementCopy:
         ]
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(tenant_id, rows)
+            _copy_compliance_requirement_rows(tenant_id, str(scan_id), rows, 2000)
 
         mock_psycopg_connection.assert_called_once_with("admin")
         connection.cursor.assert_called_once()
-        cursor.execute.assert_called_once()
+        # set_config + DELETE of the scan's previous rows.
+        assert cursor.execute.call_count == 2
         cursor.copy_expert.assert_called_once()
 
         csv_rows = list(csv.reader(StringIO(captured["data"])))
@@ -2644,6 +2665,60 @@ class TestComplianceRequirementCopy:
         assert csv_rows[2][5] == "2.0"
         assert csv_rows[2][9] == "MANUAL"
 
+    @patch("tasks.jobs.scan.psycopg_connection")
+    def test_copy_compliance_requirement_rows_batches_share_one_transaction(
+        self, mock_psycopg_connection, settings
+    ):
+        """Every COPY batch runs on the same connection with a single commit."""
+        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
+
+        connection = MagicMock()
+        cursor = MagicMock()
+        cursor_context = MagicMock()
+        cursor_context.__enter__.return_value = cursor
+        cursor_context.__exit__.return_value = False
+        connection.cursor.return_value = cursor_context
+        connection.__enter__.return_value = connection
+        connection.__exit__.return_value = False
+
+        context_manager = MagicMock()
+        context_manager.__enter__.return_value = connection
+        context_manager.__exit__.return_value = False
+        mock_psycopg_connection.return_value = context_manager
+
+        tenant_id = str(uuid.uuid4())
+        scan_id = str(uuid.uuid4())
+        inserted_at = datetime.now(UTC)
+        rows = [
+            {
+                "id": uuid.uuid4(),
+                "tenant_id": tenant_id,
+                "inserted_at": inserted_at,
+                "compliance_id": "cisa_aws",
+                "framework": "CISA",
+                "version": "1.0",
+                "description": f"Requirement {index}",
+                "region": "us-east-1",
+                "requirement_id": f"req-{index}",
+                "requirement_status": "PASS",
+                "passed_checks": 1,
+                "failed_checks": 0,
+                "total_checks": 1,
+                "scan_id": scan_id,
+            }
+            for index in range(3)
+        ]
+
+        with patch.object(MainRouter, "admin_db", "admin"):
+            total = _copy_compliance_requirement_rows(tenant_id, scan_id, rows, 1)
+
+        assert total == 3
+        # One connection, three COPY statements, one commit for the whole scan.
+        mock_psycopg_connection.assert_called_once_with("admin")
+        assert cursor.copy_expert.call_count == 3
+        connection.commit.assert_called_once()
+        connection.rollback.assert_not_called()
+
     @patch("tasks.jobs.scan.psycopg_connection")
     def test_copy_compliance_requirement_rows_null_values(
         self, mock_psycopg_connection, settings
@@ -2691,7 +2766,9 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(
+                str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+            )
 
         csv_rows = list(csv.reader(StringIO(captured["data"])))
         assert len(csv_rows) == 1
@@ -2747,7 +2824,9 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(
+                str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+            )
 
         # Verify CSV was generated (csv module handles escaping automatically)
         csv_rows = list(csv.reader(StringIO(captured["data"])))
@@ -2808,7 +2887,9 @@ class TestComplianceRequirementCopy:
 
         before_call = datetime.now(UTC)
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(
+                str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+            )
         after_call = datetime.now(UTC)
 
         csv_rows = list(csv.reader(StringIO(captured["data"])))
@@ -2861,12 +2942,84 @@ class TestComplianceRequirementCopy:
 
         with patch.object(MainRouter, "admin_db", "admin"):
             with pytest.raises(Exception, match="COPY command failed"):
-                _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+                _copy_compliance_requirement_rows(
+                    str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+                )
 
         # Verify rollback was called
         connection.rollback.assert_called_once()
         connection.commit.assert_not_called()
 
+    @pytest.mark.parametrize("mismatched_field", ["tenant_id", "scan_id"])
+    @patch("tasks.jobs.scan.psycopg_connection")
+    def test_copy_compliance_requirement_rows_rejects_out_of_scope_rows(
+        self, mock_psycopg_connection, mismatched_field, settings
+    ):
+        """COPY bypasses RLS, so rows from another tenant/scan must be rejected."""
+        settings.DATABASES.setdefault("admin", settings.DATABASES["default"])
+
+        connection = MagicMock()
+        cursor = MagicMock()
+        cursor_context = MagicMock()
+        cursor_context.__enter__.return_value = cursor
+        cursor_context.__exit__.return_value = False
+        connection.cursor.return_value = cursor_context
+        connection.__enter__.return_value = connection
+        connection.__exit__.return_value = False
+
+        context_manager = MagicMock()
+        context_manager.__enter__.return_value = connection
+        context_manager.__exit__.return_value = False
+        mock_psycopg_connection.return_value = context_manager
+
+        tenant_id = str(uuid.uuid4())
+        scan_id = str(uuid.uuid4())
+        row = {
+            "id": uuid.uuid4(),
+            "tenant_id": tenant_id,
+            "compliance_id": "test",
+            "framework": "Test",
+            "version": "1.0",
+            "description": "desc",
+            "region": "us-east-1",
+            "requirement_id": "req-1",
+            "requirement_status": "PASS",
+            "passed_checks": 1,
+            "failed_checks": 0,
+            "total_checks": 1,
+            "scan_id": scan_id,
+        }
+        row[mismatched_field] = str(uuid.uuid4())
+
+        with patch.object(MainRouter, "admin_db", "admin"):
+            with pytest.raises(ComplianceRowScopeError):
+                _copy_compliance_requirement_rows(tenant_id, scan_id, [row], 2000)
+
+        cursor.copy_expert.assert_not_called()
+        connection.rollback.assert_called_once()
+        connection.commit.assert_not_called()
+
+    @patch("tasks.jobs.scan.ComplianceRequirementOverview")
+    @patch("tasks.jobs.scan.rls_transaction")
+    @patch(
+        "tasks.jobs.scan._copy_compliance_requirement_rows",
+        side_effect=ComplianceRowScopeError("out of scope"),
+    )
+    def test_persist_compliance_requirement_rows_does_not_fall_back_on_scope_error(
+        self, mock_copy, mock_rls_transaction, mock_model
+    ):
+        """A scope violation is a caller bug: the ORM fallback must not persist it."""
+        tenant_id = str(uuid.uuid4())
+        scan_id = str(uuid.uuid4())
+
+        with pytest.raises(ComplianceRowScopeError):
+            _persist_compliance_requirement_rows(tenant_id, scan_id, lambda: [])
+
+        mock_copy.assert_called_once()
+        mock_rls_transaction.assert_not_called()
+        mock_model.objects.filter.assert_not_called()
+        mock_model.objects.bulk_create.assert_not_called()
+
     @patch("tasks.jobs.scan.psycopg_connection")
     def test_copy_compliance_requirement_rows_transaction_rollback_on_set_config_error(
         self, mock_psycopg_connection, settings
@@ -2909,7 +3062,9 @@ class TestComplianceRequirementCopy:
 
         with patch.object(MainRouter, "admin_db", "admin"):
             with pytest.raises(Exception, match="SET prowler.tenant_id failed"):
-                _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+                _copy_compliance_requirement_rows(
+                    str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+                )
 
         # Verify rollback was called
         connection.rollback.assert_called_once()
@@ -2955,7 +3110,9 @@ class TestComplianceRequirementCopy:
         }
 
         with patch.object(MainRouter, "admin_db", "admin"):
-            _copy_compliance_requirement_rows(str(row["tenant_id"]), [row])
+            _copy_compliance_requirement_rows(
+                str(row["tenant_id"]), str(row["scan_id"]), [row], 2000
+            )
 
         # Verify commit was called and rollback was not
         connection.commit.assert_called_once()
@@ -2966,9 +3123,10 @@ class TestComplianceRequirementCopy:
     @patch("tasks.jobs.scan._copy_compliance_requirement_rows")
     def test_persist_compliance_requirement_rows_success(self, mock_copy):
         """Test successful COPY path without fallback to ORM."""
-        mock_copy.return_value = None  # Success, no exception
+        mock_copy.return_value = 1  # Success, no exception
 
         tenant_id = str(uuid.uuid4())
+        scan_id = str(uuid.uuid4())
         rows = [
             {
                 "id": uuid.uuid4(),
@@ -2984,16 +3142,21 @@ class TestComplianceRequirementCopy:
                 "passed_checks": 1,
                 "failed_checks": 0,
                 "total_checks": 1,
-                "scan_id": uuid.uuid4(),
+                "scan_id": scan_id,
             }
         ]
 
-        _persist_compliance_requirement_rows(tenant_id, rows)
+        total = _persist_compliance_requirement_rows(tenant_id, scan_id, lambda: rows)
 
-        # Verify COPY was called
-        mock_copy.assert_called_once_with(tenant_id, rows)
+        assert total == 1
+        mock_copy.assert_called_once()
+        copy_args = mock_copy.call_args[0]
+        assert copy_args[0] == tenant_id
+        assert copy_args[1] == scan_id
+        assert list(copy_args[2]) == rows
 
     @patch("tasks.jobs.scan.logger")
+    @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.filter")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -3001,7 +3164,12 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("COPY failed"),
     )
     def test_persist_compliance_requirement_rows_fallback_logging(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_logger
+        self,
+        mock_copy,
+        mock_rls_transaction,
+        mock_bulk_create,
+        mock_filter,
+        mock_logger,
     ):
         """Test logger.exception is called when COPY fails and fallback occurs."""
         tenant_id = str(uuid.uuid4())
@@ -3027,7 +3195,9 @@ class TestComplianceRequirementCopy:
         ctx.__exit__.return_value = False
         mock_rls_transaction.return_value = ctx
 
-        _persist_compliance_requirement_rows(tenant_id, [row])
+        _persist_compliance_requirement_rows(
+            tenant_id, str(row["scan_id"]), lambda: [row]
+        )
 
         # Verify logger.exception was called
         mock_logger.exception.assert_called_once()
@@ -3036,6 +3206,7 @@ class TestComplianceRequirementCopy:
         assert "falling back to ORM" in args[0]
         assert kwargs.get("exc_info") is not None
 
+    @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.filter")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -3043,7 +3214,7 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("copy failed"),
     )
     def test_persist_compliance_requirement_rows_fallback_multiple_rows(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create
+        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_filter
     ):
         """Test ORM fallback with multiple rows."""
         tenant_id = str(uuid.uuid4())
@@ -3090,10 +3261,14 @@ class TestComplianceRequirementCopy:
         ctx.__exit__.return_value = False
         mock_rls_transaction.return_value = ctx
 
-        _persist_compliance_requirement_rows(tenant_id, rows)
+        total = _persist_compliance_requirement_rows(
+            tenant_id, str(scan_id), lambda: rows
+        )
 
-        mock_copy.assert_called_once_with(tenant_id, rows)
+        assert total == 2
+        mock_copy.assert_called_once()
         mock_rls_transaction.assert_called_once_with(tenant_id)
+        mock_filter.assert_called_once_with(scan_id=str(scan_id))
         mock_bulk_create.assert_called_once()
 
         args, kwargs = mock_bulk_create.call_args
@@ -3117,6 +3292,7 @@ class TestComplianceRequirementCopy:
         assert objects[1].passed_checks == 2
         assert objects[1].failed_checks == 3
 
+    @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.filter")
     @patch("tasks.jobs.scan.ComplianceRequirementOverview.objects.bulk_create")
     @patch("tasks.jobs.scan.rls_transaction")
     @patch(
@@ -3124,7 +3300,7 @@ class TestComplianceRequirementCopy:
         side_effect=Exception("copy failed"),
     )
     def test_persist_compliance_requirement_rows_fallback_all_fields(
-        self, mock_copy, mock_rls_transaction, mock_bulk_create
+        self, mock_copy, mock_rls_transaction, mock_bulk_create, mock_filter
     ):
         """Test ORM fallback correctly maps all fields from row dict to model."""
         tenant_id = str(uuid.uuid4())
@@ -3154,7 +3330,7 @@ class TestComplianceRequirementCopy:
         ctx.__exit__.return_value = False
         mock_rls_transaction.return_value = ctx
 
-        _persist_compliance_requirement_rows(tenant_id, [row])
+        _persist_compliance_requirement_rows(tenant_id, str(scan_id), lambda: [row])
 
         args, kwargs = mock_bulk_create.call_args
         objects = args[0]
diff --git a/docs/user-guide/cli/tutorials/parallel-execution.mdx b/docs/user-guide/cli/tutorials/parallel-execution.mdx
index 93b8ef381b..d659116b50 100644
--- a/docs/user-guide/cli/tutorials/parallel-execution.mdx
+++ b/docs/user-guide/cli/tutorials/parallel-execution.mdx
@@ -184,7 +184,7 @@ $combinedCsv | Export-Csv -Path "CombinedCSV.csv" -NoTypeInformation
 
 ## TODO: Additional Improvements
 
-Some services need to instantiate another service to perform a check. For instance, `cloudwatch` will instantiate Prowler's `iam` service to perform the `cloudwatch_cross_account_sharing_disabled` check. When the `iam` service is instantiated, it will perform the `__init__` function, and pull all the information required for that service. This provides an opportunity for an improvement in the above script to group related services together so that the `iam` services (or any other cross-service references) isn't repeatedily instantiated by grouping dependant services together. A complete mapping between these services still needs to be further investigated, but these are the cross-references that have been noted:
+Some services need to instantiate another service to perform a check. For instance, `cloudwatch` will instantiate Prowler's `iam` service to perform the `cloudwatch_cross_account_sharing_disabled` check. When the `iam` service is instantiated, it will perform the `__init__` function, and pull all the information required for that service. This provides an opportunity for an improvement in the above script to group related services together so that the `iam` services (or any other cross-service references) aren't repeatedly instantiated by grouping dependent services together. A complete mapping between these services still needs to be further investigated, but these are the cross-references that have been noted:
 
 * inspector2 needs lambda and ec2
 * cloudwatch needs iam
diff --git a/docs/user-guide/cli/tutorials/reporting.mdx b/docs/user-guide/cli/tutorials/reporting.mdx
index 19a14c9ae2..47e46dc2aa 100644
--- a/docs/user-guide/cli/tutorials/reporting.mdx
+++ b/docs/user-guide/cli/tutorials/reporting.mdx
@@ -114,7 +114,7 @@ The CSV format follows a standardized structure across all providers. The follow
 
 #### CSV Headers Mapping
 
-The following table shows the mapping between the CSV headers and the the providers fields:
+The following table shows the mapping between the CSV headers and the providers fields:
 
 | Open Source Consolidated| AWS| GCP| AZURE| KUBERNETES
 |----------|----------|----------|----------|----------
diff --git a/docs/user-guide/providers/oci/authentication.mdx b/docs/user-guide/providers/oci/authentication.mdx
index 9627e88cd4..cc0e75cded 100644
--- a/docs/user-guide/providers/oci/authentication.mdx
+++ b/docs/user-guide/providers/oci/authentication.mdx
@@ -434,7 +434,7 @@ prowler oci --oci-config-file /path/to/config
 
 **Cause**: Insufficient IAM permissions
 
-**Solution**: Add required policies (see [Required Permissions](./getting-started-oci.md#required-permissions))
+**Solution**: Add required policies (see [Required Permissions](/user-guide/providers/oci/getting-started-oci#required-permissions))
 
 ### Configuration Validation
 
diff --git a/docs/user-guide/providers/oci/getting-started-oci.mdx b/docs/user-guide/providers/oci/getting-started-oci.mdx
index 2f610cd09c..6a8c8de3e9 100644
--- a/docs/user-guide/providers/oci/getting-started-oci.mdx
+++ b/docs/user-guide/providers/oci/getting-started-oci.mdx
@@ -58,7 +58,7 @@ Before you begin, ensure you have:
 
 ### Authentication
 
-Prowler supports multiple authentication methods for OCI. For detailed authentication setup, see the [OCI Authentication Guide](./authentication).
+Prowler supports multiple authentication methods for OCI. For detailed authentication setup, see the [OCI Authentication Guide](/user-guide/providers/oci/authentication).
 
 **Note:** OCI Session Authentication and Config File Authentication both use the same `~/.oci/config` file. The difference is how the config file is generated - automatically via browser (session auth) or manually with API keys.
 
@@ -107,7 +107,7 @@ The easiest and most secure method is using OCI session authentication, which au
 
 #### Alternative: Manual API Key Setup
 
-If you prefer to manually generate API keys instead of using browser-based session authentication, see the detailed instructions in the [Authentication Guide](./authentication#config-file-authentication-manual-api-key-setup).
+If you prefer to manually generate API keys instead of using browser-based session authentication, see the detailed instructions in the [Authentication Guide](/user-guide/providers/oci/authentication#config-file-authentication-manual-api-key-setup).
 
 **Note:** Both methods use the same `~/.oci/config` file - the difference is that manual setup uses static API keys while session authentication uses temporary session tokens.
 
diff --git a/prowler/changelog.d/alibabacloud-security-group-policy-case.fixed.md b/prowler/changelog.d/alibabacloud-security-group-policy-case.fixed.md
new file mode 100644
index 0000000000..cef4ff076c
--- /dev/null
+++ b/prowler/changelog.d/alibabacloud-security-group-policy-case.fixed.md
@@ -0,0 +1 @@
+Alibaba Cloud SSH and RDP security group checks no longer produce false negatives when allowed rules use capitalized `Policy="Accept"` values
diff --git a/prowler/changelog.d/bucket-validation-syntaxwarning.fixed.md b/prowler/changelog.d/bucket-validation-syntaxwarning.fixed.md
new file mode 100644
index 0000000000..ceba09541c
--- /dev/null
+++ b/prowler/changelog.d/bucket-validation-syntaxwarning.fixed.md
@@ -0,0 +1 @@
+Fix invalid escape sequence `SyntaxWarning` raised on startup by the S3 bucket name validation regex
diff --git a/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet.py b/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet.py
index 72e579bbc6..45cfa81bda 100644
--- a/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet.py
+++ b/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet.py
@@ -26,7 +26,7 @@ class ecs_securitygroup_restrict_rdp_internet(Check):
 
             for ingress_rule in security_group.ingress_rules:
                 # Check if rule allows traffic (policy == "accept")
-                if ingress_rule.get("policy", "accept") != "accept":
+                if str(ingress_rule.get("policy", "accept")).lower() != "accept":
                     continue
 
                 # Check protocol (tcp for RDP)
diff --git a/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet.py b/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet.py
index 9dfdd182e1..0315a69207 100644
--- a/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet.py
+++ b/prowler/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet.py
@@ -26,7 +26,7 @@ class ecs_securitygroup_restrict_ssh_internet(Check):
 
             for ingress_rule in security_group.ingress_rules:
                 # Check if rule allows traffic (policy == "accept")
-                if ingress_rule.get("policy", "accept") != "accept":
+                if str(ingress_rule.get("policy", "accept")).lower() != "accept":
                     continue
 
                 # Check protocol (tcp for SSH)
diff --git a/prowler/providers/aws/lib/arguments/arguments.py b/prowler/providers/aws/lib/arguments/arguments.py
index 50f4665b2d..2d1632422b 100644
--- a/prowler/providers/aws/lib/arguments/arguments.py
+++ b/prowler/providers/aws/lib/arguments/arguments.py
@@ -235,7 +235,7 @@ def validate_arguments(arguments: Namespace) -> tuple[bool, str]:
 def validate_bucket(bucket_name: str) -> str:
     """validate_bucket validates that the input bucket_name is valid"""
     if search(
-        "^(?!^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$)(?!.*\.{2})(?!.*\.-)(?!.*-\.)(?!^xn--)(?!^sthree-)(?!^amzn-s3-demo-)(?!.*--table-s3$)[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$",
+        r"^(?!^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$)(?!.*\.{2})(?!.*\.-)(?!.*-\.)(?!^xn--)(?!^sthree-)(?!^amzn-s3-demo-)(?!.*--table-s3$)[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]$",
         bucket_name,
     ):
         return bucket_name
diff --git a/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet_test.py b/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet_test.py
index 17709a94a7..adbd82d11a 100644
--- a/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet_test.py
+++ b/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_rdp_internet/ecs_securitygroup_restrict_rdp_internet_test.py
@@ -37,7 +37,7 @@ class TestEcsSecurityGroupRestrictRdpInternet:
                         "ip_protocol": "tcp",
                         "source_cidr_ip": "0.0.0.0/0",
                         "port_range": "3389/3389",
-                        "policy": "accept",
+                        "policy": "Accept",
                     }
                 ],
             )
@@ -80,7 +80,7 @@ class TestEcsSecurityGroupRestrictRdpInternet:
                         "ip_protocol": "tcp",
                         "source_cidr_ip": "10.0.0.0/24",
                         "port_range": "3389/3389",
-                        "policy": "accept",
+                        "policy": "Accept",
                     }
                 ],
             )
diff --git a/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet_test.py b/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet_test.py
index 3278ce9a80..718baedce0 100644
--- a/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet_test.py
+++ b/tests/providers/alibabacloud/services/ecs/ecs_securitygroup_restrict_ssh_internet/ecs_securitygroup_restrict_ssh_internet_test.py
@@ -37,7 +37,7 @@ class TestEcsSecurityGroupRestrictSSHInternet:
                         "ip_protocol": "tcp",
                         "source_cidr_ip": "0.0.0.0/0",
                         "port_range": "22/22",
-                        "policy": "accept",
+                        "policy": "Accept",
                     }
                 ],
             )
@@ -81,7 +81,7 @@ class TestEcsSecurityGroupRestrictSSHInternet:
                         "ip_protocol": "tcp",
                         "source_cidr_ip": "10.0.0.0/24",
                         "port_range": "22/22",
-                        "policy": "accept",
+                        "policy": "Accept",
                     }
                 ],
             )
diff --git a/ui/changelog.d/ui-sentry-actionability.fixed.md b/ui/changelog.d/ui-sentry-actionability.fixed.md
new file mode 100644
index 0000000000..7a4b8ce2ca
--- /dev/null
+++ b/ui/changelog.d/ui-sentry-actionability.fixed.md
@@ -0,0 +1 @@
+UI Sentry alerts now suppress non-actionable warnings and expected API/control-flow noise while preserving actionable runtime failures
diff --git a/ui/instrumentation-client.ts b/ui/instrumentation-client.ts
index 910af90010..34dcde2f64 100644
--- a/ui/instrumentation-client.ts
+++ b/ui/instrumentation-client.ts
@@ -21,6 +21,10 @@ import {
   startProgress,
 } from "@/components/shadcn/navigation-progress/use-navigation-progress";
 import { getRuntimeConfigClient } from "@/lib/get-runtime-config.client";
+import {
+  applySentryEventPolicy,
+  SENTRY_EVENT_SOURCE,
+} from "@/sentry/event-policy";
 
 export const NAVIGATION_TYPE = {
   PUSH: "push",
@@ -96,25 +100,9 @@ if (typeof window !== "undefined" && sentryDsn) {
     ],
 
     beforeSend(event, hint) {
-      // Filter out noise: ResizeObserver errors (common browser quirk, not real bugs)
-      if (event.message?.includes("ResizeObserver")) {
-        return null; // Don't send to Sentry
-      }
-
-      // Filter out non-actionable errors
       if (event.exception) {
         const error = hint.originalException;
 
-        // Don't send cancelled requests
-        if (
-          error &&
-          typeof error === "object" &&
-          "name" in error &&
-          error.name === "AbortError"
-        ) {
-          return null;
-        }
-
         // Add additional context for API errors
         if (
           error &&
@@ -130,7 +118,9 @@ if (typeof window !== "undefined" && sentryDsn) {
         }
       }
 
-      return event; // Send to Sentry
+      return applySentryEventPolicy(event, hint, {
+        source: SENTRY_EVENT_SOURCE.CLIENT,
+      });
     },
   });
 
diff --git a/ui/lib/get-runtime-config.client.test.ts b/ui/lib/get-runtime-config.client.test.ts
index d46b633d38..d788a01132 100644
--- a/ui/lib/get-runtime-config.client.test.ts
+++ b/ui/lib/get-runtime-config.client.test.ts
@@ -105,6 +105,8 @@ describe("getRuntimeConfigClient", () => {
         "reoDevClientId",
         "sentryDsn",
         "sentryEnvironment",
+        "stripePublishableKey",
+        "stripePublishableKeyV2",
       ].sort(),
     );
     expect(config.apiBaseUrl).toBe("https://api.example.com/api/v1");
diff --git a/ui/lib/get-runtime-config.client.ts b/ui/lib/get-runtime-config.client.ts
index 9aa4ea15c8..e1bcd2bf1a 100644
--- a/ui/lib/get-runtime-config.client.ts
+++ b/ui/lib/get-runtime-config.client.ts
@@ -21,6 +21,8 @@ const pickConfig = (
   posthogHost: parsed.posthogHost ?? null,
   reoDevClientId: parsed.reoDevClientId ?? null,
   cloudBillingEnabled: parsed.cloudBillingEnabled ?? false,
+  stripePublishableKey: parsed.stripePublishableKey ?? null,
+  stripePublishableKeyV2: parsed.stripePublishableKeyV2 ?? null,
 });
 
 // Reads the <head> island once (memoized); all-null during SSR or if it's
diff --git a/ui/lib/runtime-config.shared.ts b/ui/lib/runtime-config.shared.ts
index e4fbbeceed..6e820296d9 100644
--- a/ui/lib/runtime-config.shared.ts
+++ b/ui/lib/runtime-config.shared.ts
@@ -10,6 +10,8 @@ export interface RuntimePublicConfig {
   posthogHost: string | null; // reserved
   reoDevClientId: string | null; // reserved
   cloudBillingEnabled: boolean;
+  stripePublishableKey: string | null; // reserved
+  stripePublishableKeyV2: string | null; // reserved
 }
 
 export const RUNTIME_CONFIG_SCRIPT_ID = "__PROWLER_RUNTIME_CONFIG__";
@@ -25,4 +27,6 @@ export const EMPTY_RUNTIME_PUBLIC_CONFIG: RuntimePublicConfig = {
   posthogHost: null,
   reoDevClientId: null,
   cloudBillingEnabled: false,
+  stripePublishableKey: null,
+  stripePublishableKeyV2: null,
 };
diff --git a/ui/lib/runtime-config.ts b/ui/lib/runtime-config.ts
index eda71bc114..f1ef52900b 100644
--- a/ui/lib/runtime-config.ts
+++ b/ui/lib/runtime-config.ts
@@ -49,5 +49,13 @@ export async function getRuntimePublicConfig(): Promise<RuntimePublicConfig> {
     // server-side for V1/V2 routing). Default (unset) is off.
     cloudBillingEnabled:
       (readEnv("CLOUD_BILLING_ENABLED") ?? "false") !== "false",
+    stripePublishableKey: readEnv(
+      "UI_CLOUD_STRIPE_PUBLISHABLE_KEY",
+      "NEXT_PUBLIC_STRIPE_PUBLISHABLE_KEY",
+    ),
+    stripePublishableKeyV2: readEnv(
+      "UI_CLOUD_STRIPE_PUBLISHABLE_KEY_V2",
+      "NEXT_PUBLIC_STRIPE_PUBLISHABLE_KEY_V2",
+    ),
   };
 }
diff --git a/ui/lib/server-actions-helper.test.ts b/ui/lib/server-actions-helper.test.ts
index 4644de68e6..5fa939687e 100644
--- a/ui/lib/server-actions-helper.test.ts
+++ b/ui/lib/server-actions-helper.test.ts
@@ -1,14 +1,14 @@
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import * as Sentry from "@sentry/nextjs";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 
-const {
-  captureExceptionMock,
-  captureMessageMock,
-  revalidatePathMock,
-  unstableRethrowMock,
-} = vi.hoisted(() => ({
-  captureExceptionMock: vi.fn(),
-  captureMessageMock: vi.fn(),
-  revalidatePathMock: vi.fn(),
+import {
+  isErrorAlreadyReported,
+  markErrorAsReported,
+} from "@/sentry/event-policy";
+
+import { handleApiError, handleApiResponse } from "./server-actions-helper";
+
+const { unstableRethrowMock } = vi.hoisted(() => ({
   unstableRethrowMock: vi.fn((error: unknown) => {
     if (error instanceof Error && error.message === "NEXT_REDIRECT") {
       throw error;
@@ -17,19 +17,19 @@ const {
 }));
 
 vi.mock("@sentry/nextjs", () => ({
-  captureException: captureExceptionMock,
-  captureMessage: captureMessageMock,
+  captureException: vi.fn(),
+  captureMessage: vi.fn(),
 }));
 
 vi.mock("next/cache", () => ({
-  revalidatePath: revalidatePathMock,
+  revalidatePath: vi.fn(),
 }));
 
 vi.mock("next/navigation", () => ({
   unstable_rethrow: unstableRethrowMock,
 }));
 
-vi.mock("./helper", () => ({
+vi.mock("@/lib/helper", () => ({
   GENERIC_SERVER_ERROR_MESSAGE:
     "Server is temporarily unavailable. Please try again in a few minutes.",
   getErrorMessage: (error: unknown) =>
@@ -39,49 +39,239 @@ vi.mock("./helper", () => ({
     /<html\b|<\/?body\b|<\/?h1\b/i.test(message) ? fallback : message.trim(),
 }));
 
-import { handleApiError, handleApiResponse } from "./server-actions-helper";
-
-describe("server action error handling", () => {
+describe("server-actions-helper", () => {
   beforeEach(() => {
     vi.clearAllMocks();
-    vi.spyOn(console, "error").mockImplementation(() => {});
   });
 
-  it("rethrows authentication redirect control flow before converting errors", () => {
-    // Given
-    const redirectError = new Error("NEXT_REDIRECT");
-
-    // When / Then
-    expect(() => handleApiError(redirectError)).toThrow(redirectError);
+  afterEach(() => {
+    vi.restoreAllMocks();
   });
 
-  it("returns ordinary errors as action data", () => {
-    // Given
-    const ordinaryError = new Error("API unavailable");
+  describe("handleApiResponse", () => {
+    it("should throw a generic server error instead of raw HTML for 5xx responses", async () => {
+      // Given
+      const response = new Response(
+        "<html><head><title>502 Bad Gateway</title></head><body><center><h1>502 Bad Gateway</h1></center></body></html>",
+        {
+          status: 502,
+          statusText: "Bad Gateway",
+          headers: { "content-type": "text/html" },
+        },
+      );
 
-    // When
-    const result = handleApiError(ordinaryError);
+      // When / Then
+      await expect(handleApiResponse(response)).rejects.toThrow(
+        "Server is temporarily unavailable. Please try again in a few minutes.",
+      );
+    });
 
-    // Then
-    expect(result).toEqual({ error: "API unavailable" });
-  });
+    it("should not capture controlled 4xx API validation responses", async () => {
+      // Given
+      const response = new Response(
+        JSON.stringify({ errors: [{ detail: "Provider already exists" }] }),
+        { status: 409, statusText: "Conflict" },
+      );
 
-  it("throws a generic server error instead of raw HTML for 5xx responses", async () => {
-    // Given
-    const response = new Response(
-      "<html><head><title>502 Bad Gateway</title></head><body><center><h1>502 Bad Gateway</h1></center></body></html>",
-      {
-        status: 502,
-        statusText: "Bad Gateway",
-        headers: { "content-type": "text/html" },
+      // When
+      const result = await handleApiResponse(response);
+
+      // Then
+      expect(Sentry.captureException).not.toHaveBeenCalled();
+      expect(result).toMatchObject({
+        error: "Provider already exists",
+        status: 409,
+      });
+    });
+
+    it.each([418, 429])(
+      "should capture unexpected %s API client failures before returning them",
+      async (status) => {
+        // Given
+        const response = new Response(
+          JSON.stringify({ message: "Unexpected API contract failure" }),
+          { status, statusText: "Unexpected Client Failure" },
+        );
+
+        // When
+        const result = await handleApiResponse(response);
+
+        // Then
+        expect(Sentry.captureException).toHaveBeenCalledTimes(1);
+        expect(Sentry.captureException).toHaveBeenCalledWith(
+          expect.any(Error),
+          expect.objectContaining({
+            tags: expect.objectContaining({
+              api_error: true,
+              error_source: "handleApiResponse",
+              error_type: "client_error",
+              status_code: status.toString(),
+            }),
+          }),
+        );
+        expect(result).toMatchObject({
+          error: "Unexpected API contract failure",
+          status,
+        });
       },
     );
 
-    // When / Then
-    const result = handleApiResponse(response);
-    await expect(result).rejects.toThrow(
-      "Server is temporarily unavailable. Please try again in a few minutes.",
-    );
+    it("should not mark server errors before Sentry processes the event", async () => {
+      // Given
+      const response = new Response(
+        JSON.stringify({ message: "backend down" }),
+        {
+          status: 500,
+          statusText: "Internal Server Error",
+        },
+      );
+
+      // When
+      await expect(handleApiResponse(response)).rejects.toThrow("backend down");
+
+      // Then
+      expect(Sentry.captureException).toHaveBeenCalledTimes(1);
+      const capturedError = vi.mocked(Sentry.captureException).mock
+        .calls[0]?.[0];
+      expect(isErrorAlreadyReported(capturedError)).toBe(false);
+    });
+
+    it("should fingerprint server errors by a normalized pathname", async () => {
+      // Given
+      const response = new Response(
+        JSON.stringify({ message: "backend down" }),
+        {
+          status: 500,
+          statusText: "Internal Server Error",
+        },
+      );
+      Object.defineProperty(response, "url", {
+        value:
+          "https://api.prowler.test/api/v1/providers/550e8400-e29b-41d4-a716-446655440000?tenant=123",
+      });
+
+      // When
+      await expect(handleApiResponse(response)).rejects.toThrow("backend down");
+
+      // Then
+      expect(Sentry.captureException).toHaveBeenCalledWith(
+        expect.any(Error),
+        expect.objectContaining({
+          fingerprint: ["api-server-error", "500", "/api/v1/providers/:id"],
+        }),
+      );
+    });
+
+    it("should fingerprint unexpected client failures by a normalized pathname", async () => {
+      // Given
+      const response = new Response(
+        JSON.stringify({ message: "Unexpected API contract failure" }),
+        { status: 429, statusText: "Too Many Requests" },
+      );
+      Object.defineProperty(response, "url", {
+        value: "https://api.prowler.test/api/v1/scans/12345?page=2",
+      });
+
+      // When
+      await handleApiResponse(response);
+
+      // Then
+      expect(Sentry.captureException).toHaveBeenCalledWith(
+        expect.any(Error),
+        expect.objectContaining({
+          fingerprint: [
+            "api-client-contract-error",
+            "429",
+            "/api/v1/scans/:id",
+          ],
+        }),
+      );
+    });
+  });
+
+  describe("handleApiError", () => {
+    it("should rethrow authentication redirect control flow", () => {
+      // Given
+      const redirectError = new Error("NEXT_REDIRECT");
+
+      // When / Then
+      expect(() => handleApiError(redirectError)).toThrow(redirectError);
+    });
+
+    it("should not recapture errors that were already reported", () => {
+      // Given
+      const error = new Error("Already reported failure");
+      vi.spyOn(console, "error").mockImplementation(() => undefined);
+      markErrorAsReported(error);
+
+      // When
+      const result = handleApiError(error);
+
+      // Then
+      expect(Sentry.captureException).not.toHaveBeenCalled();
+      expect(result).toEqual({ error: "Already reported failure" });
+    });
+
+    it("should not recapture errors already marked by Sentry", () => {
+      // Given
+      const error = new Error("Already captured failure");
+      Object.defineProperty(error, "__sentry_captured__", { value: true });
+      vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+      // When
+      const result = handleApiError(error);
+
+      // Then
+      expect(Sentry.captureException).not.toHaveBeenCalled();
+      expect(result).toEqual({ error: "Already captured failure" });
+    });
+
+    it("should capture unmarked request failure errors", () => {
+      // Given
+      const error = new Error("Request failed unexpectedly");
+      vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+      // When
+      const result = handleApiError(error);
+
+      // Then
+      expect(Sentry.captureException).toHaveBeenCalledTimes(1);
+      expect(Sentry.captureException).toHaveBeenCalledWith(
+        error,
+        expect.objectContaining({
+          tags: expect.objectContaining({
+            error_source: "handleApiError",
+            error_type: "unexpected_error",
+          }),
+        }),
+      );
+      expect(isErrorAlreadyReported(error)).toBe(false);
+      expect(result).toEqual({ error: "Request failed unexpectedly" });
+    });
+
+    it("should capture unmarked runtime errors that include expected HTTP status numbers", () => {
+      // Given
+      const error = new Error("Runtime worker 401 crashed unexpectedly");
+      vi.spyOn(console, "error").mockImplementation(() => undefined);
+
+      // When
+      const result = handleApiError(error);
+
+      // Then
+      expect(Sentry.captureException).toHaveBeenCalledTimes(1);
+      expect(Sentry.captureException).toHaveBeenCalledWith(
+        error,
+        expect.objectContaining({
+          tags: expect.objectContaining({
+            error_source: "handleApiError",
+            error_type: "unexpected_error",
+          }),
+        }),
+      );
+      expect(result).toEqual({
+        error: "Runtime worker 401 crashed unexpectedly",
+      });
+    });
   });
 
   it("returns authentication failures as ordinary API error data", async () => {
diff --git a/ui/lib/server-actions-helper.ts b/ui/lib/server-actions-helper.ts
index 1c36d347d1..b219f41b2f 100644
--- a/ui/lib/server-actions-helper.ts
+++ b/ui/lib/server-actions-helper.ts
@@ -3,6 +3,10 @@ import { revalidatePath } from "next/cache";
 import { unstable_rethrow } from "next/navigation";
 
 import { SentryErrorSource, SentryErrorType } from "@/sentry";
+import {
+  isErrorAlreadyReported,
+  isErrorCapturedBySentry,
+} from "@/sentry/event-policy";
 
 import {
   GENERIC_SERVER_ERROR_MESSAGE,
@@ -11,6 +15,14 @@ import {
   sanitizeErrorMessage,
 } from "./helper";
 
+const EXPECTED_HTTP_CLIENT_STATUS_CODES = new Set([401, 403, 404]);
+const CONTROLLED_CLIENT_STATUS_CODES = new Set([400, 409, 422]);
+const KNOWN_NON_ACTIONABLE_CLIENT_ERROR_MESSAGES = [
+  "already exists",
+  "duplicate",
+] as const;
+const UNKNOWN_URL_PATH_FINGERPRINT = "unknown-url-path";
+
 /**
  * Helper function to handle API responses consistently
  * Includes Sentry error tracking for debugging
@@ -79,37 +91,17 @@ export const handleApiResponse = async (
         fingerprint: [
           "api-server-error",
           response.status.toString(),
-          response.url,
+          getSentryFingerprintUrlPath(response.url),
         ],
       });
 
       throw serverError;
     }
 
-    // Client errors (4xx) - Only capture unexpected ones
-    if (![401, 403, 404].includes(response.status)) {
-      const clientError = new Error(
-        errorDetail ||
-          `Request failed (${response.status}): ${response.statusText}`,
-      );
-
-      Sentry.captureException(clientError, {
-        tags: {
-          api_error: true,
-          status_code: response.status.toString(),
-          error_type: SentryErrorType.CLIENT_ERROR,
-          error_source: SentryErrorSource.HANDLE_API_RESPONSE,
-        },
-        level: "warning",
-        contexts: {
-          api_response: errorContext,
-        },
-        fingerprint: [
-          "api-client-error",
-          response.status.toString(),
-          response.url,
-        ],
-      });
+    if (
+      !shouldSuppressApiClientFailure(response.status, errorDetail, errorsArray)
+    ) {
+      captureUnexpectedApiClientFailure(response, errorDetail, errorContext);
     }
 
     return errorsArray
@@ -161,33 +153,26 @@ export const handleApiError = (error: unknown): { error: string } => {
 
   // Check if this error was already captured by handleApiResponse
   const isAlreadyCaptured =
-    error instanceof Error &&
-    (error.message.includes("Server error") ||
-      error.message.includes("Request failed"));
+    isErrorAlreadyReported(error) || isErrorCapturedBySentry(error);
 
-  // Only capture if not already captured by handleApiResponse
+  // Only capture if not already captured by handleApiResponse.
+  // HTTP status-based suppression belongs in the structured Sentry event policy,
+  // not in string-only Error message matching.
   if (!isAlreadyCaptured) {
     if (error instanceof Error) {
-      // Don't capture expected errors
-      if (
-        !error.message.includes("401") &&
-        !error.message.includes("403") &&
-        !error.message.includes("404")
-      ) {
-        Sentry.captureException(error, {
-          tags: {
-            error_source: SentryErrorSource.HANDLE_API_ERROR,
-            error_type: SentryErrorType.UNEXPECTED_ERROR,
+      Sentry.captureException(error, {
+        tags: {
+          error_source: SentryErrorSource.HANDLE_API_ERROR,
+          error_type: SentryErrorType.UNEXPECTED_ERROR,
+        },
+        level: "error",
+        contexts: {
+          error_details: {
+            message: error.message,
+            stack: error.stack,
           },
-          level: "error",
-          contexts: {
-            error_details: {
-              message: error.message,
-              stack: error.stack,
-            },
-          },
-        });
-      }
+        },
+      });
     } else {
       // Capture non-Error objects
       Sentry.captureMessage(
@@ -210,3 +195,90 @@ export const handleApiError = (error: unknown): { error: string } => {
     error: getErrorMessage(error),
   };
 };
+
+function shouldSuppressApiClientFailure(
+  status: number,
+  errorDetail: unknown,
+  errorsArray: unknown[] | undefined,
+) {
+  if (status < 400 || status >= 500) {
+    return true;
+  }
+
+  if (EXPECTED_HTTP_CLIENT_STATUS_CODES.has(status)) {
+    return true;
+  }
+
+  return (
+    CONTROLLED_CLIENT_STATUS_CODES.has(status) &&
+    (hasStructuredApiErrors(errorsArray) ||
+      hasKnownNonActionableClientErrorMessage(errorDetail))
+  );
+}
+
+function captureUnexpectedApiClientFailure(
+  response: Response,
+  errorDetail: unknown,
+  errorContext: Record<string, unknown>,
+) {
+  const clientError = new Error(
+    typeof errorDetail === "string"
+      ? errorDetail
+      : `Unexpected API client failure (${response.status})`,
+  );
+
+  Sentry.captureException(clientError, {
+    tags: {
+      api_error: true,
+      status_code: response.status.toString(),
+      error_type: SentryErrorType.CLIENT_ERROR,
+      error_source: SentryErrorSource.HANDLE_API_RESPONSE,
+    },
+    level: "error",
+    contexts: {
+      api_response: errorContext,
+    },
+    fingerprint: [
+      "api-client-contract-error",
+      response.status.toString(),
+      getSentryFingerprintUrlPath(response.url),
+    ],
+  });
+}
+
+function getSentryFingerprintUrlPath(url: string) {
+  if (url.trim() === "") {
+    return UNKNOWN_URL_PATH_FINGERPRINT;
+  }
+
+  try {
+    const pathname = new URL(url).pathname;
+
+    return (
+      pathname
+        .replace(
+          /\/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}(?=\/|$)/gi,
+          "/:id",
+        )
+        .replace(/\/\d+(?=\/|$)/g, "/:id") || UNKNOWN_URL_PATH_FINGERPRINT
+    );
+  } catch {
+    return UNKNOWN_URL_PATH_FINGERPRINT;
+  }
+}
+
+function hasStructuredApiErrors(errorsArray: unknown[] | undefined) {
+  return Array.isArray(errorsArray) && errorsArray.length > 0;
+}
+
+function hasKnownNonActionableClientErrorMessage(errorDetail: unknown) {
+  if (typeof errorDetail !== "string") {
+    return false;
+  }
+
+  const normalizedErrorDetail = errorDetail.toLowerCase();
+
+  return KNOWN_NON_ACTIONABLE_CLIENT_ERROR_MESSAGES.some((message) =>
+    normalizedErrorDetail.includes(message),
+  );
+}
diff --git a/ui/sentry/event-policy.test.ts b/ui/sentry/event-policy.test.ts
new file mode 100644
index 0000000000..3f3c13d7fa
--- /dev/null
+++ b/ui/sentry/event-policy.test.ts
@@ -0,0 +1,308 @@
+import { describe, expect, it } from "vitest";
+
+import type {
+  SentryEventHint,
+  SentryEventPolicyOptions,
+  SentryPolicyEvent,
+} from "./event-policy";
+import {
+  applySentryEventPolicy,
+  isErrorAlreadyReported,
+  isErrorCapturedBySentry,
+  markErrorAsReported,
+} from "./event-policy";
+
+describe("applySentryEventPolicy", () => {
+  describe("when events are actionable", () => {
+    it("should keep error events", () => {
+      // Given
+      const event = {
+        level: "error",
+        message: "Unexpected failure",
+      } satisfies SentryPolicyEvent & { level: string; message: string };
+      const options: SentryEventPolicyOptions = { source: "client" };
+
+      // When
+      const result = applySentryEventPolicy(event, undefined, options);
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          kind: "runtime",
+          source: "client",
+        },
+      });
+    });
+
+    it("should keep fatal events", () => {
+      // Given
+      const event = { level: "fatal", message: "Runtime crashed" };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+    });
+
+    it("should keep events without a level", () => {
+      // Given
+      const event = { message: "Default Sentry event" };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+    });
+
+    it("should keep fatal runtime messages that contain an expected HTTP status number", () => {
+      // Given
+      const event = {
+        level: "fatal",
+        message: "Runtime worker 401 crashed unexpectedly",
+      };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          kind: "runtime",
+        },
+      });
+    });
+
+    it("should keep runtime error messages that contain an expected HTTP status number", () => {
+      // Given
+      const event = {
+        level: "error",
+        message: "Background import 404 failed after a null dereference",
+      };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          kind: "runtime",
+        },
+      });
+    });
+
+    it("should keep fatal runtime messages that mention status without HTTP context", () => {
+      // Given
+      const event = {
+        level: "fatal",
+        message: "State transition status 403 caused worker crash",
+      };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          kind: "runtime",
+        },
+      });
+    });
+
+    it("should keep fatal runtime messages that mention status code without HTTP context", () => {
+      // Given
+      const event = {
+        level: "fatal",
+        message: "State transition status code 403 caused worker crash",
+      };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          kind: "runtime",
+        },
+      });
+    });
+
+    it("should keep fatal runtime messages that mention response without HTTP context", () => {
+      // Given
+      const event = {
+        level: "fatal",
+        message: "Worker response 404 triggered fatal cache corruption",
+      };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          kind: "runtime",
+        },
+      });
+    });
+
+    it("should preserve existing tags on actionable API events", () => {
+      // Given
+      const event = {
+        level: "error",
+        message: "API response failed with status 500",
+        tags: {
+          feature: "providers",
+          status_code: "500",
+        },
+      };
+
+      // When
+      const result = applySentryEventPolicy(event, undefined, {
+        source: "server_action",
+      });
+
+      // Then
+      expect(result).toBe(event);
+      expect(result).toMatchObject({
+        tags: {
+          actionability: "actionable",
+          feature: "providers",
+          kind: "api",
+          source: "server_action",
+          status_code: "500",
+        },
+      });
+    });
+  });
+
+  describe("when events are non-actionable", () => {
+    it("should drop warning events", () => {
+      // Given
+      const event = { level: "warning", message: "Provider already exists" };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop Next.js redirect control-flow events", () => {
+      // Given
+      const event = { level: "error", message: "NEXT_REDIRECT" };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop AbortError events", () => {
+      // Given
+      const event = { level: "error", message: "The operation was aborted" };
+      const hint: SentryEventHint = {
+        originalException: new DOMException("Aborted", "AbortError"),
+      };
+
+      // When
+      const result = applySentryEventPolicy(event, hint);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop expected HTTP 401 events", () => {
+      // Given
+      const event = { level: "error", tags: { status_code: "401" } };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop expected HTTP 403 events", () => {
+      // Given
+      const event = { level: "error", message: "Request failed with 403" };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop expected HTTP 404 events", () => {
+      // Given
+      const event = { level: "error" };
+      const hint = { originalException: { status: 404 } };
+
+      // When
+      const result = applySentryEventPolicy(event, hint);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop clear HTTP messages with expected status codes", () => {
+      // Given
+      const event = {
+        level: "error",
+        message: "HTTP response status code 404",
+      };
+
+      // When
+      const result = applySentryEventPolicy(event);
+
+      // Then
+      expect(result).toBeNull();
+    });
+
+    it("should drop already-reported errors", () => {
+      // Given
+      const error = new Error("Already reported failure");
+      const event = { level: "error", message: error.message };
+      markErrorAsReported(error);
+
+      // When
+      const result = applySentryEventPolicy(event, {
+        originalException: error,
+      });
+
+      // Then
+      expect(isErrorAlreadyReported(error)).toBe(true);
+      expect(result).toBeNull();
+    });
+
+    it("should keep the first event for an error marked by Sentry", () => {
+      // Given
+      const error = new Error("First capture");
+      Object.defineProperty(error, "__sentry_captured__", { value: true });
+      const event = { level: "error", message: error.message };
+
+      // When
+      const result = applySentryEventPolicy(event, {
+        originalException: error,
+      });
+
+      // Then
+      expect(isErrorCapturedBySentry(error)).toBe(true);
+      expect(result).toBe(event);
+    });
+  });
+});
diff --git a/ui/sentry/event-policy.ts b/ui/sentry/event-policy.ts
new file mode 100644
index 0000000000..c8312abfb6
--- /dev/null
+++ b/ui/sentry/event-policy.ts
@@ -0,0 +1,290 @@
+const SENTRY_EVENT_LEVEL = {
+  WARNING: "warning",
+} as const;
+
+export const SENTRY_EVENT_SOURCE = {
+  CLIENT: "client",
+  EDGE: "edge",
+  SERVER: "server",
+  SERVER_ACTION: "server_action",
+} as const;
+
+const SENTRY_EVENT_KIND = {
+  API: "api",
+  RUNTIME: "runtime",
+} as const;
+
+const SENTRY_ACTIONABILITY = {
+  ACTIONABLE: "actionable",
+} as const;
+
+const EXPECTED_CONTROL_FLOW_MESSAGES = [
+  "NEXT_REDIRECT",
+  "NEXT_NOT_FOUND",
+  "AbortError",
+  "ResizeObserver",
+] as const;
+
+const HTTP_CONTEXT_MESSAGES = [
+  /\bapi\b/i,
+  /\bfetch\b/i,
+  /\bhttp\b/i,
+  /\brequest failed\b/i,
+] as const;
+
+const EXPECTED_HTTP_STATUS_CODES = new Set([401, 403, 404]);
+const EXPECTED_HTTP_STATUS_PATTERN = /(^|\D)(401|403|404)(\D|$)/;
+const REPORTED_ERROR_MARKER = Symbol.for("prowler.sentry.reported_error");
+const SENTRY_CAPTURED_MARKER = "__sentry_captured__";
+const reportedErrors = new WeakSet<object>();
+
+type SentryEventSource =
+  (typeof SENTRY_EVENT_SOURCE)[keyof typeof SENTRY_EVENT_SOURCE];
+type SentryPolicyTagValue = string | number | boolean | null | undefined;
+
+export interface SentryEventPolicyOptions {
+  source?: SentryEventSource;
+}
+
+export interface SentryEventHint {
+  originalException?: unknown;
+}
+
+export interface SentryPolicyEvent {
+  tags?: Record<string, SentryPolicyTagValue>;
+}
+
+type ObjectRecord = Record<PropertyKey, unknown>;
+type ReportedErrorRecord = Record<symbol, unknown>;
+
+export function applySentryEventPolicy<TEvent extends object>(
+  event: TEvent,
+  hint?: SentryEventHint,
+  options: SentryEventPolicyOptions = {},
+) {
+  if (shouldDropSentryEvent(event, hint)) {
+    return null;
+  }
+
+  tagActionableEvent(event, options);
+
+  return event;
+}
+
+export function markErrorAsReported(error: unknown) {
+  if (!isObjectLike(error)) {
+    return;
+  }
+
+  reportedErrors.add(error);
+
+  try {
+    Object.defineProperty(error, REPORTED_ERROR_MARKER, {
+      configurable: false,
+      enumerable: false,
+      value: true,
+    });
+  } catch {
+    // WeakSet fallback still suppresses duplicates for non-extensible objects.
+  }
+}
+
+export function isErrorAlreadyReported(error: unknown) {
+  if (!isObjectLike(error)) {
+    return false;
+  }
+
+  return (
+    reportedErrors.has(error) ||
+    (error as ReportedErrorRecord)[REPORTED_ERROR_MARKER] === true
+  );
+}
+
+export function isErrorCapturedBySentry(error: unknown) {
+  return (
+    isObjectLike(error) && getProperty(error, SENTRY_CAPTURED_MARKER) === true
+  );
+}
+
+function shouldDropSentryEvent(event: object, hint?: SentryEventHint) {
+  if (getStringProperty(event, "level") === SENTRY_EVENT_LEVEL.WARNING) {
+    return true;
+  }
+
+  if (isErrorAlreadyReported(hint?.originalException)) {
+    return true;
+  }
+
+  const messages = getEventMessages(event, hint);
+
+  if (hasExpectedControlFlowMessage(messages)) {
+    return true;
+  }
+
+  return hasExpectedHttpStatus(event, hint?.originalException, messages);
+}
+
+function tagActionableEvent(event: object, options: SentryEventPolicyOptions) {
+  const mutableEvent = event as SentryPolicyEvent;
+
+  mutableEvent.tags = {
+    ...mutableEvent.tags,
+    actionability: SENTRY_ACTIONABILITY.ACTIONABLE,
+    kind: inferEventKind(event),
+    ...(options.source ? { source: options.source } : {}),
+  };
+}
+
+function inferEventKind(event: object) {
+  const tags = getRecordProperty(event, "tags");
+  const errorType = getStringProperty(tags, "error_type");
+  const apiError = getProperty(tags, "api_error");
+
+  if (
+    errorType === "api_error" ||
+    apiError === true ||
+    getStatusFromTags(tags)
+  ) {
+    return SENTRY_EVENT_KIND.API;
+  }
+
+  return SENTRY_EVENT_KIND.RUNTIME;
+}
+
+function hasExpectedControlFlowMessage(messages: string[]) {
+  return EXPECTED_CONTROL_FLOW_MESSAGES.some((expectedMessage) =>
+    messages.some((message) => message.includes(expectedMessage)),
+  );
+}
+
+function hasExpectedHttpStatus(
+  event: object,
+  originalException: unknown,
+  messages: string[],
+) {
+  const tags = getRecordProperty(event, "tags");
+  const statusFromTags = getStatusFromTags(tags);
+  const statusFromError = getStatusFromError(originalException);
+
+  if (
+    (statusFromTags && EXPECTED_HTTP_STATUS_CODES.has(statusFromTags)) ||
+    (statusFromError && EXPECTED_HTTP_STATUS_CODES.has(statusFromError))
+  ) {
+    return true;
+  }
+
+  return (
+    hasExpectedHttpStatusMessage(messages) &&
+    hasApiOrHttpContext(event, messages)
+  );
+}
+
+function hasExpectedHttpStatusMessage(messages: string[]) {
+  return messages.some((message) => EXPECTED_HTTP_STATUS_PATTERN.test(message));
+}
+
+function hasApiOrHttpContext(event: object, messages: string[]) {
+  const tags = getRecordProperty(event, "tags");
+
+  return hasApiTags(tags) || hasHttpContextMessage(messages);
+}
+
+function hasApiTags(tags: unknown) {
+  return (
+    getStringProperty(tags, "error_type") === "api_error" ||
+    getProperty(tags, "api_error") === true
+  );
+}
+
+function hasHttpContextMessage(messages: string[]) {
+  return messages.some((message) =>
+    HTTP_CONTEXT_MESSAGES.some((pattern) => pattern.test(message)),
+  );
+}
+
+function getStatusFromTags(tags: unknown) {
+  return (
+    normalizeStatusCode(getProperty(tags, "status_code")) ??
+    normalizeStatusCode(getProperty(tags, "status")) ??
+    normalizeStatusCode(getProperty(tags, "http.status_code"))
+  );
+}
+
+function getStatusFromError(error: unknown): number | undefined {
+  const response = getRecordProperty(error, "response");
+
+  return (
+    normalizeStatusCode(getProperty(error, "status")) ??
+    normalizeStatusCode(getProperty(error, "statusCode")) ??
+    normalizeStatusCode(getProperty(error, "status_code")) ??
+    normalizeStatusCode(getProperty(response, "status"))
+  );
+}
+
+function getEventMessages(event: object, hint?: SentryEventHint) {
+  return [
+    getStringProperty(event, "message"),
+    ...getExceptionMessages(event),
+    getStringProperty(hint?.originalException, "name"),
+    getStringProperty(hint?.originalException, "message"),
+    typeof hint?.originalException === "string"
+      ? hint.originalException
+      : undefined,
+  ].filter(isString);
+}
+
+function getExceptionMessages(event: object) {
+  const exception = getRecordProperty(event, "exception");
+  const values = getProperty(exception, "values");
+
+  if (!Array.isArray(values)) {
+    return [];
+  }
+
+  return values.flatMap((value) => [
+    getStringProperty(value, "type"),
+    getStringProperty(value, "value"),
+  ]);
+}
+
+function normalizeStatusCode(value: unknown) {
+  if (typeof value === "number" && Number.isInteger(value)) {
+    return value;
+  }
+
+  if (typeof value === "string" && /^\d{3}$/.test(value)) {
+    return Number(value);
+  }
+
+  return undefined;
+}
+
+function getStringProperty(value: unknown, property: string) {
+  const propertyValue = getProperty(value, property);
+
+  return typeof propertyValue === "string" ? propertyValue : undefined;
+}
+
+function getRecordProperty(value: unknown, property: string) {
+  const propertyValue = getProperty(value, property);
+
+  return isRecord(propertyValue) ? propertyValue : undefined;
+}
+
+function getProperty(value: unknown, property: string) {
+  return isRecord(value) ? value[property] : undefined;
+}
+
+function isRecord(value: unknown): value is ObjectRecord {
+  return typeof value === "object" && value !== null;
+}
+
+function isObjectLike(value: unknown): value is object {
+  return (
+    (typeof value === "object" && value !== null) || typeof value === "function"
+  );
+}
+
+function isString(value: unknown): value is string {
+  return typeof value === "string";
+}
diff --git a/ui/sentry/index.ts b/ui/sentry/index.ts
index dca76ca0e0..9666b7002e 100644
--- a/ui/sentry/index.ts
+++ b/ui/sentry/index.ts
@@ -1,2 +1,3 @@
 // Re-export all Sentry utilities
+export * from "./event-policy";
 export * from "./utils";
diff --git a/ui/sentry/sentry-event-filter.test.ts b/ui/sentry/sentry-event-filter.test.ts
new file mode 100644
index 0000000000..370af5884e
--- /dev/null
+++ b/ui/sentry/sentry-event-filter.test.ts
@@ -0,0 +1,53 @@
+import { describe, expect, it } from "vitest";
+
+import { filterWarningSentryEvent } from "./sentry-event-filter";
+
+describe("filterWarningSentryEvent", () => {
+  describe("when the event level is warning", () => {
+    it("should drop the event", () => {
+      // Given
+      const event = { level: "warning", message: "Provider already exists" };
+
+      // When
+      const result = filterWarningSentryEvent(event);
+
+      // Then
+      expect(result).toBeNull();
+    });
+  });
+
+  describe("when the event level is actionable", () => {
+    it("should keep error events", () => {
+      // Given
+      const event = { level: "error", message: "Unexpected failure" };
+
+      // When
+      const result = filterWarningSentryEvent(event);
+
+      // Then
+      expect(result).toBe(event);
+    });
+
+    it("should keep fatal events", () => {
+      // Given
+      const event = { level: "fatal", message: "Runtime crashed" };
+
+      // When
+      const result = filterWarningSentryEvent(event);
+
+      // Then
+      expect(result).toBe(event);
+    });
+
+    it("should keep events without a level", () => {
+      // Given
+      const event = { message: "Default Sentry event" };
+
+      // When
+      const result = filterWarningSentryEvent(event);
+
+      // Then
+      expect(result).toBe(event);
+    });
+  });
+});
diff --git a/ui/sentry/sentry-event-filter.ts b/ui/sentry/sentry-event-filter.ts
new file mode 100644
index 0000000000..6396d50e66
--- /dev/null
+++ b/ui/sentry/sentry-event-filter.ts
@@ -0,0 +1,7 @@
+import { applySentryEventPolicy } from "./event-policy";
+
+// Backward-compatible export name: the implementation now applies the full
+// actionability policy, including warning drops and expected API noise filters.
+export function filterWarningSentryEvent<TEvent extends object>(event: TEvent) {
+  return applySentryEventPolicy(event);
+}
diff --git a/ui/sentry/sentry.edge.config.ts b/ui/sentry/sentry.edge.config.ts
index f4f10d0566..f8198902b5 100644
--- a/ui/sentry/sentry.edge.config.ts
+++ b/ui/sentry/sentry.edge.config.ts
@@ -2,6 +2,8 @@ import * as Sentry from "@sentry/nextjs";
 
 import { readGatedEnv } from "@/lib/integrations";
 
+import { applySentryEventPolicy, SENTRY_EVENT_SOURCE } from "./event-policy";
+
 const sentryDsn = readGatedEnv(
   "UI_SENTRY_ENABLED",
   "UI_SENTRY_DSN",
@@ -47,15 +49,13 @@ if (sentryDsn) {
     // 🔌 Integrations - Edge runtime doesn't support all integrations
     integrations: [],
 
-    // 🎣 Filter expected errors - Don't send noise to Sentry
+    // 🎣 Filter expected framework control-flow - Don't send noise to Sentry.
+    // HTTP status-based suppression belongs in applySentryEventPolicy, where
+    // structured event context prevents broad numeric matches from hiding crashes.
     ignoreErrors: [
       // NextAuth redirect errors - Expected behavior in auth flow
       "NEXT_REDIRECT",
       "NEXT_NOT_FOUND",
-      // Expected HTTP errors - Expected when users lack permissions
-      "401", // Unauthorized - expected when token expires
-      "403", // Forbidden - expected when no permissions
-      "404", // Not Found - expected for missing resources
     ],
 
     beforeSend(event, hint) {
@@ -65,20 +65,9 @@ if (sentryDsn) {
         runtime: "edge",
       };
 
-      const error = hint.originalException;
-
-      // Don't send NextAuth expected errors
-      if (
-        error &&
-        typeof error === "object" &&
-        "message" in error &&
-        typeof error.message === "string" &&
-        error.message.includes("NEXT_REDIRECT")
-      ) {
-        return null;
-      }
-
-      return event;
+      return applySentryEventPolicy(event, hint, {
+        source: SENTRY_EVENT_SOURCE.EDGE,
+      });
     },
   });
 }
diff --git a/ui/sentry/sentry.server.config.ts b/ui/sentry/sentry.server.config.ts
index 3a3ac491da..d1c7fef906 100644
--- a/ui/sentry/sentry.server.config.ts
+++ b/ui/sentry/sentry.server.config.ts
@@ -2,6 +2,8 @@ import * as Sentry from "@sentry/nextjs";
 
 import { readGatedEnv } from "@/lib/integrations";
 
+import { applySentryEventPolicy, SENTRY_EVENT_SOURCE } from "./event-policy";
+
 const sentryDsn = readGatedEnv(
   "UI_SENTRY_ENABLED",
   "UI_SENTRY_DSN",
@@ -52,15 +54,13 @@ if (sentryDsn) {
       }),
     ],
 
-    // 🎣 Filter expected errors - Don't send noise to Sentry
+    // 🎣 Filter expected framework control-flow - Don't send noise to Sentry.
+    // HTTP status-based suppression belongs in applySentryEventPolicy, where
+    // structured event context prevents broad numeric matches from hiding crashes.
     ignoreErrors: [
       // NextAuth redirect errors - Expected behavior
       "NEXT_REDIRECT",
       "NEXT_NOT_FOUND",
-      // Expected HTTP errors - Expected when users lack permissions
-      "401", // Unauthorized
-      "403", // Forbidden
-      "404", // Not Found
     ],
 
     beforeSend(event, hint) {
@@ -87,15 +87,12 @@ if (sentryDsn) {
               error_type: "api_error",
             };
           }
-
-          // Don't send NextAuth expected errors
-          if (error.message.includes("NEXT_REDIRECT")) {
-            return null;
-          }
         }
       }
 
-      return event;
+      return applySentryEventPolicy(event, hint, {
+        source: SENTRY_EVENT_SOURCE.SERVER,
+      });
     },
   });
 }
diff --git a/ui/tests/runtime-config/runtime-config-page.ts b/ui/tests/runtime-config/runtime-config-page.ts
index dcdf3e546c..430033eba4 100644
--- a/ui/tests/runtime-config/runtime-config-page.ts
+++ b/ui/tests/runtime-config/runtime-config-page.ts
@@ -20,6 +20,8 @@ export const RUNTIME_CONFIG_KEYS = [
   "posthogHost",
   "reoDevClientId",
   "cloudBillingEnabled",
+  "stripePublishableKey",
+  "stripePublishableKeyV2",
 ] as const satisfies ReadonlyArray<keyof RuntimePublicConfig>;
 
 /**
diff --git a/ui/types/env.d.ts b/ui/types/env.d.ts
index 515d7a6c03..c4ec4d85a3 100644
--- a/ui/types/env.d.ts
+++ b/ui/types/env.d.ts
@@ -30,6 +30,15 @@ declare global {
 
       CLOUD_BILLING_ENABLED?: "legacy" | "metronome" | "false";
 
+      // Cloud-only Stripe publishable keys (public; shipped to the browser).
+      // V1 = legacy billing, V2 = metronome.
+      /** @deprecated use UI_CLOUD_STRIPE_PUBLISHABLE_KEY */
+      NEXT_PUBLIC_STRIPE_PUBLISHABLE_KEY?: string;
+      UI_CLOUD_STRIPE_PUBLISHABLE_KEY?: string;
+      /** @deprecated use UI_CLOUD_STRIPE_PUBLISHABLE_KEY_V2 */
+      NEXT_PUBLIC_STRIPE_PUBLISHABLE_KEY_V2?: string;
+      UI_CLOUD_STRIPE_PUBLISHABLE_KEY_V2?: string;
+
       // Build-time public config
       NEXT_PUBLIC_IS_CLOUD_ENV?: "true" | "false";
       NEXT_PUBLIC_PROWLER_RELEASE_VERSION?: string;