diff --git a/.github/workflows/uv-lock-automation.yaml b/.github/workflows/uv-lock-automation.yaml
index 8cfc2f9ca2..52c3364386 100644
--- a/.github/workflows/uv-lock-automation.yaml
+++ b/.github/workflows/uv-lock-automation.yaml
@@ -36,8 +36,6 @@ jobs:
           version: "0.6.14"
           python-version: 3.12.9
 
-      - run: uv pip install --python=3.12.9 pip
-
       - name: Generate UV lockfiles
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
diff --git a/backend/api_v2/notification.py b/backend/api_v2/notification.py
index 57e810c0e9..4cefe92d24 100644
--- a/backend/api_v2/notification.py
+++ b/backend/api_v2/notification.py
@@ -1,6 +1,6 @@
 import logging
 
-from notification_v2.helper import NotificationHelper
+from notification_v2.helper import dispatch_with_delivery_mode
 from notification_v2.models import Notification
 from pipeline_v2.dto import PipelineStatusPayload
 from workflow_manager.workflow_v2.enums import ExecutionStatus
@@ -60,6 +60,8 @@ def send(self) -> None:
             failed_files=failed_files,
         )
 
-        NotificationHelper.send_notification(
-            notifications=self.notifications, payload=payload_dto.to_dict()
+        dispatch_with_delivery_mode(
+            list(self.notifications),
+            payload_dto.to_dict(),
+            error_context=f"api={self.api.id}",
         )
diff --git a/backend/backend/settings/base.py b/backend/backend/settings/base.py
index 2d493a91c3..f20109203f 100644
--- a/backend/backend/settings/base.py
+++ b/backend/backend/settings/base.py
@@ -12,11 +12,12 @@
 import logging
 import os
 from pathlib import Path
-from urllib.parse import quote, urlparse
+from urllib.parse import quote
 
 import httpx
 from dotenv import find_dotenv, load_dotenv
 from utils.common_utils import CommonUtils
+from utils.cors_origin import normalize_web_app_origin
 
 missing_settings = []
 
@@ -68,10 +69,18 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str |
 )
 # Maximum number of files allowed per workflow page execution
 WORKFLOW_PAGE_MAX_FILES = int(os.environ.get("WORKFLOW_PAGE_MAX_FILES", 2))
-WEB_APP_ORIGIN_URL = os.environ.get("WEB_APP_ORIGIN_URL", "http://localhost:3000")
-parsed_url = urlparse(WEB_APP_ORIGIN_URL)
-WEB_APP_ORIGIN_URL_WITH_WILD_CARD = f"{parsed_url.scheme}://*.{parsed_url.netloc}"
+(
+    WEB_APP_ORIGIN_URL,
+    WEB_APP_ORIGIN_URL_WITH_WILD_CARD,
+    _CORS_SUBDOMAIN_REGEX,
+) = normalize_web_app_origin(
+    os.environ.get("WEB_APP_ORIGIN_URL", "http://localhost:3000")
+)
 CORS_ALLOWED_ORIGINS = [WEB_APP_ORIGIN_URL]
+# Wildcard subdomain regex consumed by django-cors-headers and (via the
+# RegexOrigin wrapper in utils/log_events.py) by the SocketIO engine.io
+# handshake — a single source of truth so HTTP and WS CORS cannot diverge.
+CORS_ALLOWED_ORIGIN_REGEXES = [_CORS_SUBDOMAIN_REGEX]
 
 DJANGO_APP_BACKEND_URL = os.environ.get("DJANGO_APP_BACKEND_URL", "http://localhost:8000")
 INTERNAL_SERVICE_API_KEY = os.environ.get("INTERNAL_SERVICE_API_KEY")
@@ -210,6 +219,15 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str |
 
 INDEXING_FLAG_TTL = int(get_required_setting("INDEXING_FLAG_TTL"))
 NOTIFICATION_TIMEOUT = int(get_required_setting("NOTIFICATION_TIMEOUT", "5"))
+# Window for clubbing BATCHED notifications — also the flush cadence (seconds).
+# Default 1800 (30 min). Per-notification buffer rows precompute flush_after at
+# enqueue time, so changing this only affects rows enqueued after the restart.
+NOTIFICATION_CLUB_INTERVAL = int(os.environ.get("NOTIFICATION_CLUB_INTERVAL", "1800"))
+# Retention for terminal NotificationBuffer rows (DISPATCHED / DEAD_LETTER).
+# PENDING rows are never GC'd regardless of age.
+NOTIFICATION_BUFFER_RETENTION_DAYS = int(
+    os.environ.get("NOTIFICATION_BUFFER_RETENTION_DAYS", "7")
+)
 ATOMIC_REQUESTS = CommonUtils.str_to_bool(
     os.environ.get("DJANGO_ATOMIC_REQUESTS", "False")
 )
@@ -689,4 +707,6 @@ def filter(self, record):
     )
     raise ValueError(ERROR_MESSAGE)
 
-ENABLE_HIGHLIGHT_API_DEPLOYMENT = os.environ.get("ENABLE_HIGHLIGHT_API_DEPLOYMENT", False)
+ENABLE_HIGHLIGHT_API_DEPLOYMENT = CommonUtils.str_to_bool(
+    os.environ.get("ENABLE_HIGHLIGHT_API_DEPLOYMENT", "False")
+)
diff --git a/backend/configuration/enums.py b/backend/configuration/enums.py
index f35033a9f7..a11a0b58e7 100644
--- a/backend/configuration/enums.py
+++ b/backend/configuration/enums.py
@@ -61,6 +61,12 @@ class ConfigKey(Enum):
         max_value=settings.MAX_PARALLEL_FILE_BATCHES_MAX_VALUE,
     )
 
+    NOTIFICATION_CLUB_INTERVAL = ConfigSpec(
+        default=settings.NOTIFICATION_CLUB_INTERVAL,
+        value_type=ConfigType.INT,
+        help_text="Window (seconds) for clubbing BATCHED notifications.",
+    )
+
     def cast_value(self, raw_value: Any):
         converters = {
             ConfigType.INT: int,
diff --git a/backend/notification_v2/clubbed_renderer.py b/backend/notification_v2/clubbed_renderer.py
new file mode 100644
index 0000000000..6fd0de0a84
--- /dev/null
+++ b/backend/notification_v2/clubbed_renderer.py
@@ -0,0 +1,138 @@
+"""Clubbed notification renderer.
+
+Builds one canonical JSON envelope from a group of buffered execution events
+and emits the platform-appropriate dispatch payload. Stays separate from the
+single-event SlackWebhook / APIWebhook providers so immediate-dispatch behavior
+stays untouched.
+
+Envelope shape (always the same — single-event groups use this too so consumers
+never need to branch on "is this batched?"):
+
+    {
+        "kind": "batch",
+        "summary": {
+            "pipeline": "<name>",
+            "interval_minutes": 30,
+            "total": N, "succeeded": S, "failed": F
+        },
+        "events": [{"execution_id": ..., "status": ..., "error": ...?}, ...]
+    }
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from notification_v2.enums import PlatformType
+
+logger = logging.getLogger(__name__)
+
+# Hard cap on events per dispatch — extras roll over to the next flush tick.
+# Bounds memory + payload size and prevents a runaway backlog from creating an
+# unbounded HTTP body.
+MAX_BATCH_SIZE = 500
+# How many events Slack renders inline before collapsing the rest under a
+# "… and K more" footer. Slack tolerates much larger payloads, but readability
+# tanks past ~25 lines.
+SLACK_MAX_DISPLAY_EVENTS = 25
+
+_SUCCESS_STATUSES = {"COMPLETED", "SUCCESS"}
+
+
+def _is_success(status: str | None) -> bool:
+    if not status:
+        return False
+    return status.upper() in _SUCCESS_STATUSES
+
+
+def _event_from_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    event: dict[str, Any] = {
+        "execution_id": payload.get("execution_id"),
+        "status": payload.get("status"),
+    }
+    error_message = payload.get("error_message")
+    if error_message:
+        event["error"] = error_message
+    return event
+
+
+def build_envelope(
+    payloads: list[dict[str, Any]], interval_seconds: int
+) -> dict[str, Any]:
+    """Build the canonical batch envelope.
+
+    Caps the events list at MAX_BATCH_SIZE; oldest-first ordering is the
+    caller's responsibility (the flush job sorts by created_at).
+    """
+    capped = payloads[:MAX_BATCH_SIZE]
+    succeeded = sum(1 for p in capped if _is_success(p.get("status")))
+    failed = len(capped) - succeeded
+    # Multiple pipelines can share an (org, url, auth_sig) group; we surface
+    # the first one's name as a representative. Mixed-pipeline batches are
+    # rare in practice and a v2 enhancement would aggregate distinct names.
+    pipeline_name = capped[0].get("pipeline_name") if capped else None
+    return {
+        "kind": "batch",
+        "summary": {
+            "pipeline": pipeline_name,
+            "interval_minutes": max(1, interval_seconds // 60),
+            "total": len(capped),
+            "succeeded": succeeded,
+            "failed": failed,
+        },
+        "events": [_event_from_payload(p) for p in capped],
+    }
+
+
+def _slack_event_line(event: dict[str, Any]) -> str:
+    parts = [f"— {event.get('execution_id') or 'unknown'}: {event.get('status')}"]
+    if event.get("error"):
+        parts.append(f"({event['error']})")
+    return " ".join(parts)
+
+
+def render_for_slack(envelope: dict[str, Any]) -> dict[str, Any]:
+    """Format the envelope as a Slack-compatible payload dict.
+
+    Returns the body shape Slack incoming webhooks expect (`text` field with
+    mrkdwn). Truncates inline events at SLACK_MAX_DISPLAY_EVENTS.
+    """
+    summary = envelope["summary"]
+    events: list[dict[str, Any]] = envelope["events"]
+    pipeline = summary.get("pipeline") or "pipeline"
+
+    header = f"*[Unstract] {summary['total']} executions for `{pipeline}`*"
+    counts = f"✅ {summary['succeeded']} succeeded  ❌ {summary['failed']} failed"
+
+    visible = events[:SLACK_MAX_DISPLAY_EVENTS]
+    lines = [_slack_event_line(e) for e in visible]
+    overflow = len(events) - len(visible)
+    if overflow > 0:
+        lines.append(f"… and {overflow} more executions")
+
+    body = "\n".join([header, counts, *lines])
+    return {"text": body}
+
+
+def render_clubbed_message(
+    payloads: list[dict[str, Any]], platform: str, interval_seconds: int
+) -> dict[str, Any]:
+    """Top-level entry point — returns the dispatch body for ``platform``.
+
+    Slack receives the rendered text payload; raw API webhooks receive the
+    canonical envelope unchanged so downstream consumers can parse it
+    programmatically.
+    """
+    envelope = build_envelope(payloads, interval_seconds)
+    if platform == PlatformType.SLACK.value:
+        return render_for_slack(envelope)
+    if platform == PlatformType.API.value:
+        return envelope
+    # Unknown platform — fall back to the raw envelope and warn so misrouted
+    # rows don't drop silently.
+    logger.warning(
+        "Unknown platform %s for clubbed dispatch; returning raw envelope",
+        platform,
+    )
+    return envelope
diff --git a/backend/notification_v2/enums.py b/backend/notification_v2/enums.py
index 991b08cac9..d6fed8b485 100644
--- a/backend/notification_v2/enums.py
+++ b/backend/notification_v2/enums.py
@@ -36,3 +36,37 @@ class PlatformType(Enum):
     @classmethod
     def choices(cls):
         return [(e.value, e.name.replace("_", " ").capitalize()) for e in cls]
+
+
+class DeliveryMode(Enum):
+    """Per-notification dispatch mode.
+
+    IMMEDIATE fires on every workflow completion (pre-existing behavior).
+    BATCHED buffers events into NotificationBuffer and flushes them as one
+    clubbed message per (org, webhook_url, auth_sig) every
+    NOTIFICATION_CLUB_INTERVAL seconds.
+    """
+
+    IMMEDIATE = "IMMEDIATE"
+    BATCHED = "BATCHED"
+
+    @classmethod
+    def choices(cls):
+        return [(e.value, e.name.replace("_", " ").capitalize()) for e in cls]
+
+
+class BufferStatus(Enum):
+    """Lifecycle states for a NotificationBuffer row.
+
+    PENDING       — waiting for the next flush tick.
+    DISPATCHED    — successfully sent as part of a clubbed message.
+    DEAD_LETTER   — Celery exhausted retries; terminal, never re-picked.
+    """
+
+    PENDING = "PENDING"
+    DISPATCHED = "DISPATCHED"
+    DEAD_LETTER = "DEAD_LETTER"
+
+    @classmethod
+    def choices(cls):
+        return [(e.value, e.name.replace("_", " ").capitalize()) for e in cls]
diff --git a/backend/notification_v2/helper.py b/backend/notification_v2/helper.py
index a454b9d82b..f217f2da18 100644
--- a/backend/notification_v2/helper.py
+++ b/backend/notification_v2/helper.py
@@ -1,35 +1,223 @@
+import hashlib
 import logging
+from collections.abc import Iterable
+from datetime import timedelta
 from typing import Any
 
-from notification_v2.enums import NotificationType, PlatformType
-from notification_v2.models import Notification
+from account_v2.models import Organization
+from django.utils import timezone
+
+from notification_v2.enums import (
+    AuthorizationType,
+    BufferStatus,
+    DeliveryMode,
+    NotificationType,
+    PlatformType,
+)
+from notification_v2.models import Notification, NotificationBuffer
 from notification_v2.provider.notification_provider import NotificationProvider
 from notification_v2.provider.registry import get_notification_provider
 
 logger = logging.getLogger(__name__)
 
+# Used as a stable salt-free input for SHA-256 grouping; collisions are
+# vanishingly improbable and the digest is never used as a security primitive.
+_AUTH_SIG_NONE = ""
+
+
+def compute_auth_sig(notification: Notification) -> str:
+    """SHA-256 hex of (auth_type + auth_key + auth_header) — never raw creds.
+
+    Identical auth configs produce the same sig (so grouping clubs them);
+    differing configs split into separate groups.
+    """
+    raw = "|".join(
+        [
+            notification.authorization_type or _AUTH_SIG_NONE,
+            notification.authorization_key or _AUTH_SIG_NONE,
+            notification.authorization_header or _AUTH_SIG_NONE,
+        ]
+    )
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
+def webhook_url_hash(url: str | None) -> str:
+    """Short, log-safe fingerprint of a webhook URL (first 8 chars of SHA-256)."""
+    if not url:
+        return "none"
+    return hashlib.sha256(url.encode("utf-8")).hexdigest()[:8]
+
+
+def get_org_club_interval_seconds(organization: Organization) -> int:
+    """Per-org override of NOTIFICATION_CLUB_INTERVAL, falling back to env default.
+
+    Reads from the generic configuration KV table; returns the env-derived
+    default when the org has no override. The value is read at enqueue time
+    and baked into the row's flush_after — see mfbt §EC-2 / §EC-8: changing
+    the override only affects rows enqueued after the change.
+    """
+    # Local import: configuration depends on Django settings at import time
+    # and notification_v2.helper is imported during app boot.
+    from configuration.enums import ConfigKey
+    from configuration.models import Configuration
+
+    return int(
+        Configuration.get_value_by_organization(
+            ConfigKey.NOTIFICATION_CLUB_INTERVAL, organization
+        )
+    )
+
+
+def build_webhook_headers(notification: Notification) -> dict[str, str]:
+    """Build HTTP headers for a webhook dispatch from the notification's auth.
+
+    Mirrors the logic in ``provider/webhook/webhook.py`` and the worker-side
+    ``get_webhook_headers`` so the clubbed dispatcher and the immediate path
+    produce identical headers for the same auth config.
+    """
+    headers = {"Content-Type": "application/json"}
+    auth_type_raw = (notification.authorization_type or "").upper()
+    auth_key = notification.authorization_key
+    auth_header = notification.authorization_header
+    if auth_type_raw == AuthorizationType.BEARER.value and auth_key:
+        headers["Authorization"] = f"Bearer {auth_key}"
+    elif auth_type_raw == AuthorizationType.API_KEY.value and auth_key:
+        headers["Authorization"] = auth_key
+    elif (
+        auth_type_raw == AuthorizationType.CUSTOM_HEADER.value
+        and auth_header
+        and auth_key
+    ):
+        headers[auth_header] = auth_key
+    return headers
+
+
+def _resolve_organization(notification: Notification) -> Organization | None:
+    """Walk pipeline/api FK to find the owning org. Notification has no direct FK."""
+    pipeline = notification.pipeline
+    if pipeline and pipeline.organization_id:
+        return pipeline.organization
+    api = notification.api
+    if api and api.organization_id:
+        return api.organization
+    return None
+
+
+def split_by_delivery_mode(
+    notifications: "Iterable[Notification]",
+) -> tuple[list[Notification], list[Notification]]:
+    """Partition into (IMMEDIATE, BATCHED). Unknown modes default to IMMEDIATE."""
+    immediate: list[Notification] = []
+    batched: list[Notification] = []
+    for n in notifications:
+        if n.delivery_mode == DeliveryMode.BATCHED.value:
+            batched.append(n)
+        else:
+            immediate.append(n)
+    return immediate, batched
+
+
+def dispatch_with_delivery_mode(
+    notifications: "Iterable[Notification]",
+    payload: dict[str, Any],
+    *,
+    error_context: str = "",
+) -> None:
+    """Single-call entry point that splits IMMEDIATE / BATCHED and dispatches.
+
+    IMMEDIATE rows fire synchronously via NotificationHelper. BATCHED rows
+    enqueue into NotificationBuffer; an enqueue failure is logged but does
+    not abort the loop — other notifications still get their chance.
+
+    ``error_context`` lets callers tag failures with their dispatch source
+    (pipeline id, api id) for easier triage.
+    """
+    immediate, batched = split_by_delivery_mode(notifications)
+    if immediate:
+        NotificationHelper.send_notification(notifications=immediate, payload=payload)
+    for notification in batched:
+        try:
+            enqueue(notification, payload)
+        except Exception:
+            logger.exception(
+                "Failed to enqueue BATCHED notification %s%s",
+                notification.id,
+                f" ({error_context})" if error_context else "",
+            )
+
+
+def enqueue(notification: Notification, payload: dict[str, Any]) -> NotificationBuffer:
+    """Buffer a single execution event for a BATCHED notification.
+
+    Computes auth_sig and flush_after at write time so existing PENDING rows
+    keep their original cadence even if NOTIFICATION_CLUB_INTERVAL or the
+    notification's auth changes mid-window. Returns the persisted row.
+
+    Raises ValueError if the notification has no resolvable organization
+    (defensive — the FK chain via pipeline/api always provides one in practice).
+    """
+    organization = _resolve_organization(notification)
+    if organization is None:
+        raise ValueError(
+            f"Notification {notification.id} has no resolvable organization "
+            "(neither pipeline nor api FK populated)"
+        )
+
+    interval_seconds = get_org_club_interval_seconds(organization)
+    flush_after = timezone.now() + timedelta(seconds=interval_seconds)
+    auth_sig = compute_auth_sig(notification)
+    platform = notification.platform or PlatformType.API.value
+
+    buffer_row = NotificationBuffer.objects.create(
+        notification=notification,
+        organization=organization,
+        webhook_url=notification.url,
+        payload=payload,
+        platform=platform,
+        auth_sig=auth_sig,
+        flush_after=flush_after,
+        status=BufferStatus.PENDING.value,
+    )
+
+    # Structured log: org + URL fingerprint only — never the raw URL or any
+    # part of the auth tuple. Downstream metrics consumers grep on metric=.
+    logger.info(
+        "metric=notification_buffer_enqueued_total platform=%s org_id=%s "
+        "webhook_url_hash=%s notification_id=%s buffer_id=%s flush_after=%s",
+        platform,
+        organization.organization_id,
+        webhook_url_hash(notification.url),
+        notification.id,
+        buffer_row.id,
+        flush_after.isoformat(),
+    )
+    return buffer_row
+
 
 class NotificationHelper:
     @classmethod
     def send_notification(cls, notifications: list[Notification], payload: Any) -> None:
-        """Send notification Sends notifications using the appropriate provider
-        based on the notification type and platform.
+        """Dispatch IMMEDIATE notifications via the registered provider.
 
-        This method iterates through a list of `Notification` objects, determines the
-        appropriate notification provider based on the notification's type and
-        platform, and sends the notification with the provided payload. If an error
-        occurs due to an invalid notification type or platform, it logs the error.
+        Iterates over notifications, resolves the provider for each
+        (notification_type, platform) pair, and fires the webhook task. BATCHED
+        notifications must be routed to ``enqueue()`` instead — callers branch
+        on ``notification.delivery_mode`` before reaching this method.
 
         Args:
-            notifications (list[Notification]): A list of `Notification` instances to
-                be processed and sent.
-            payload (Any): The data to be sent with the notification. This can be any
-                format expected by the provider
-
-        Returns:
-                None
+            notifications: Active Notification rows to dispatch synchronously.
+            payload: Provider-specific payload (typically a dict).
         """
         for notification in notifications:
+            if notification.delivery_mode == DeliveryMode.BATCHED.value:
+                # Callers should not reach here for BATCHED — log loudly so
+                # routing regressions are visible without breaking dispatch.
+                logger.warning(
+                    "BATCHED notification %s reached IMMEDIATE dispatch path; "
+                    "skipping. Caller must branch on delivery_mode.",
+                    notification.id,
+                )
+                continue
             notification_type = NotificationType(notification.notification_type)
             platform_type = PlatformType(notification.platform)
             try:
@@ -40,9 +228,13 @@ def send_notification(cls, notifications: list[Notification], payload: Any) -> N
                     notification=notification, payload=payload
                 )
                 notifier.send()
-                logger.info(f"Sending notification to {notification}")
+                logger.info("Sending notification to %s", notification)
             except ValueError as e:
                 logger.error(
-                    f"Error in notification type {notification_type} and platform "
-                    f"{platform_type} for notification {notification}: {e}"
+                    "Error in notification type %s and platform %s for "
+                    "notification %s: %s",
+                    notification_type,
+                    platform_type,
+                    notification,
+                    e,
                 )
diff --git a/backend/notification_v2/internal_api_views.py b/backend/notification_v2/internal_api_views.py
index 3e3f386b17..396bfb2081 100644
--- a/backend/notification_v2/internal_api_views.py
+++ b/backend/notification_v2/internal_api_views.py
@@ -9,13 +9,18 @@
 - These endpoints are not accessible from browsers and don't use session cookies
 """
 
+import json
 import logging
+from datetime import timedelta
 from typing import Any, cast
 
 from api_v2.models import APIDeployment
-from django.db.models import QuerySet
+from django.conf import settings
+from django.db import transaction
+from django.db.models import Min, QuerySet
 from django.http import HttpRequest, JsonResponse
 from django.shortcuts import get_object_or_404
+from django.utils import timezone
 from django.views.decorators.csrf import csrf_exempt
 from django.views.decorators.http import require_http_methods
 from pipeline_v2.models import Pipeline
@@ -23,7 +28,16 @@
 from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
 
-from notification_v2.models import Notification
+from backend.celery_service import app as celery_app
+from notification_v2.clubbed_renderer import render_clubbed_message
+from notification_v2.enums import BufferStatus, DeliveryMode
+from notification_v2.helper import (
+    build_webhook_headers,
+    enqueue,
+    get_org_club_interval_seconds,
+    webhook_url_hash,
+)
+from notification_v2.models import Notification, NotificationBuffer
 
 logger = logging.getLogger(__name__)
 
@@ -89,6 +103,9 @@ def _serialize_notification(n: Notification) -> dict[str, Any]:
         "max_retries": n.max_retries,
         "is_active": n.is_active,
         "notify_on_failures": n.notify_on_failures,
+        # Drives the worker-side IMMEDIATE-vs-BATCHED branch in
+        # workers/shared/patterns/notification/helper.py.
+        "delivery_mode": n.delivery_mode,
     }
 
 
@@ -271,3 +288,247 @@ def get_api_data(request: HttpRequest, api_id: str) -> JsonResponse:
         return JsonResponse(
             {"status": "error", "message": INTERNAL_SERVER_ERROR_MSG}, status=500
         )
+
+
+# Required fields on the enqueue endpoint body. Worker-side serialization
+# guarantees these — keep this list in sync with
+# workers/shared/patterns/notification/helper.py.
+_ENQUEUE_REQUIRED_FIELDS = (
+    "notification_id",
+    "execution_id",
+    "pipeline_id",
+    "pipeline_name",
+    "status",
+    "platform",
+)
+
+
+@csrf_exempt  # Safe: Internal API with Bearer token auth, service-to-service only
+@require_http_methods(["POST"])
+def enqueue_notification_buffer(request: HttpRequest) -> JsonResponse:
+    """Buffer one execution event from a callback worker.
+
+    Worker code is model-free: it forwards a notification_id + structured
+    payload here and lets the backend write the NotificationBuffer row.
+    Rejects rows whose source notification is not BATCHED so a worker
+    routing bug cannot silently divert IMMEDIATE traffic into the buffer.
+    """
+    try:
+        body = json.loads(request.body.decode("utf-8") or "{}")
+    except json.JSONDecodeError:
+        return JsonResponse(
+            {"status": "error", "message": "Invalid JSON body"}, status=400
+        )
+
+    missing = [f for f in _ENQUEUE_REQUIRED_FIELDS if not body.get(f)]
+    if missing:
+        return JsonResponse(
+            {
+                "status": "error",
+                "message": f"Missing required fields: {', '.join(missing)}",
+            },
+            status=400,
+        )
+
+    try:
+        notification = Notification.objects.get(id=body["notification_id"])
+    except Notification.DoesNotExist:
+        return JsonResponse(
+            {"status": "error", "message": "Notification not found"}, status=404
+        )
+
+    if notification.delivery_mode != DeliveryMode.BATCHED.value:
+        # Hard-fail rather than silently auto-correcting — surfaces worker
+        # routing regressions instead of letting them drain into the buffer.
+        return JsonResponse(
+            {
+                "status": "error",
+                "message": (
+                    "Notification delivery_mode is not BATCHED; refuse to enqueue"
+                ),
+            },
+            status=409,
+        )
+
+    payload = {
+        "execution_id": body["execution_id"],
+        "pipeline_id": body["pipeline_id"],
+        "pipeline_name": body["pipeline_name"],
+        "status": body["status"],
+        "error_message": body.get("error_message"),
+        "platform": body["platform"],
+    }
+    try:
+        buffer_row = enqueue(notification, payload)
+    except ValueError as e:
+        return JsonResponse({"status": "error", "message": str(e)}, status=400)
+
+    return JsonResponse(
+        {"status": "success", "buffer_row_id": str(buffer_row.id)}, status=201
+    )
+
+
+def _gc_terminal_rows() -> int:
+    """Delete DISPATCHED / DEAD_LETTER rows older than the retention window.
+
+    PENDING rows are intentionally untouched regardless of age — they
+    represent live work the flush job still owns.
+    """
+    cutoff = timezone.now() - timedelta(days=settings.NOTIFICATION_BUFFER_RETENTION_DAYS)
+    deleted_count, _ = NotificationBuffer.objects.filter(
+        status__in=[BufferStatus.DISPATCHED.value, BufferStatus.DEAD_LETTER.value],
+        created_at__lt=cutoff,
+    ).delete()
+    return int(deleted_count)
+
+
+def _dispatch_group(
+    org_id: Any,
+    webhook_url: str,
+    auth_sig: str,
+) -> tuple[int, int]:
+    """Dispatch a single (org, url, auth_sig) group; returns (rows, succeeded).
+
+    Caller already filtered groups to MIN(flush_after) <= now. Locks rows
+    with SKIP LOCKED so a sibling replica skips them rather than blocking.
+    Re-fetches the source Notification each time for live auth (record may
+    have been edited between enqueue and flush).
+    """
+    with transaction.atomic():
+        rows = list(
+            NotificationBuffer.objects.select_for_update(skip_locked=True)
+            .filter(
+                status=BufferStatus.PENDING.value,
+                organization_id=org_id,
+                webhook_url=webhook_url,
+                auth_sig=auth_sig,
+            )
+            .order_by("created_at")[:_PROCESS_BUFFER_CAP]
+        )
+        if not rows:
+            # Either another replica claimed the rows (SKIP LOCKED) or they
+            # transitioned out of PENDING between the GROUP BY scan and the
+            # row-level lock. Either way: nothing to do here.
+            return 0, 0
+
+        # Live auth — read from the FIRST row's notification. If multiple
+        # notifications collide on (url, auth_sig) we have, by definition,
+        # identical auth, so this is safe.
+        first_notification = rows[0].notification
+        platform = rows[0].platform
+        payloads = [r.payload for r in rows]
+        # Per-org interval read here is cosmetic — used only for the
+        # `interval_minutes` field in the rendered message body. The
+        # cadence-controlling read happened at enqueue time and is
+        # already baked into each row's flush_after (mfbt §EC-2).
+        interval_seconds = get_org_club_interval_seconds(rows[0].organization)
+        body = render_clubbed_message(payloads, platform, interval_seconds)
+        headers = build_webhook_headers(first_notification)
+
+        buffer_ids = [str(r.id) for r in rows]
+        try:
+            celery_app.send_task(
+                "send_webhook_notification",
+                args=[
+                    first_notification.url,
+                    body,
+                    headers,
+                    settings.NOTIFICATION_TIMEOUT,
+                ],
+                kwargs={
+                    "max_retries": first_notification.max_retries,
+                    "retry_delay": 10,
+                    "platform": platform,
+                },
+                queue="notifications",
+                link_error=celery_app.signature(
+                    "notification_v2.mark_buffer_dead_letter",
+                    kwargs={"buffer_row_ids": buffer_ids},
+                ),
+            )
+        except Exception:
+            # Broker hiccup — leave rows PENDING for the next tick rather
+            # than mark them DEAD_LETTER. `exception` keeps stack context.
+            logger.exception(
+                "Broker dispatch failed for group org=%s url_hash=%s",
+                org_id,
+                webhook_url_hash(webhook_url),
+            )
+            return 0, 0
+
+        now = timezone.now()
+        NotificationBuffer.objects.filter(id__in=buffer_ids).update(
+            status=BufferStatus.DISPATCHED.value,
+            dispatched_at=now,
+        )
+        logger.info(
+            "metric=notification_batch_dispatched_total platform=%s result=success "
+            "org_id=%s webhook_url_hash=%s rows=%d",
+            platform,
+            org_id,
+            webhook_url_hash(webhook_url),
+            len(rows),
+        )
+        return len(rows), len(rows)
+
+
+# Per-group cap; matches the renderer's MAX_BATCH_SIZE so the rendered
+# events list and the dispatched row set stay in lock-step. Anything beyond
+# this rolls into the next flush tick.
+_PROCESS_BUFFER_CAP = 500
+
+
+@csrf_exempt  # Safe: Internal API with Bearer token auth, service-to-service only
+@require_http_methods(["POST"])
+def process_notification_buffer(request: HttpRequest) -> JsonResponse:
+    """Flush PENDING groups that have hit their flush_after; then GC.
+
+    Algorithm:
+    1. GROUP BY (org, url, auth_sig), HAVING MIN(flush_after) <= NOW()
+    2. For each group, in its own transaction: lock-skip-locked rows,
+       render, dispatch a single Celery task, mark rows DISPATCHED.
+    3. Sweep terminal rows older than NOTIFICATION_BUFFER_RETENTION_DAYS.
+
+    Concurrency: SELECT FOR UPDATE SKIP LOCKED makes parallel calls safe —
+    each replica skips groups another worker is already dispatching.
+    """
+    now = timezone.now()
+    groups = list(
+        NotificationBuffer.objects.filter(status=BufferStatus.PENDING.value)
+        .values("organization_id", "webhook_url", "auth_sig")
+        .annotate(earliest_flush=Min("flush_after"))
+        .filter(earliest_flush__lte=now)
+    )
+
+    dispatched_groups = 0
+    dispatched_rows = 0
+    for group in groups:
+        try:
+            rows, _succeeded = _dispatch_group(
+                org_id=group["organization_id"],
+                webhook_url=group["webhook_url"],
+                auth_sig=group["auth_sig"],
+            )
+        except Exception:
+            logger.exception(
+                "Failed dispatching group org=%s url_hash=%s",
+                group["organization_id"],
+                webhook_url_hash(group["webhook_url"]),
+            )
+            continue
+        if rows > 0:
+            dispatched_groups += 1
+            dispatched_rows += rows
+
+    gc_deleted = _gc_terminal_rows()
+    return JsonResponse(
+        {
+            "status": "success",
+            "dispatched_groups": dispatched_groups,
+            "dispatched_rows": dispatched_rows,
+            # DEAD_LETTER transitions are async (Celery link_error) — this
+            # response only covers transitions visible to this request.
+            "dead_letter_rows": 0,
+            "gc_deleted_rows": gc_deleted,
+        }
+    )
diff --git a/backend/notification_v2/internal_serializers.py b/backend/notification_v2/internal_serializers.py
index 94669d64a4..db7a35ab32 100644
--- a/backend/notification_v2/internal_serializers.py
+++ b/backend/notification_v2/internal_serializers.py
@@ -23,6 +23,7 @@ class Meta:
             "platform",
             "max_retries",
             "is_active",
+            "delivery_mode",
             "created_at",
             "modified_at",
             "pipeline",
diff --git a/backend/notification_v2/internal_urls.py b/backend/notification_v2/internal_urls.py
index 0414761089..a0f87f0250 100644
--- a/backend/notification_v2/internal_urls.py
+++ b/backend/notification_v2/internal_urls.py
@@ -21,6 +21,17 @@
 router.register(r"", WebhookInternalViewSet, basename="webhook-internal")
 
 urlpatterns = [
+    # Buffered (clubbed) notification dispatch endpoints
+    path(
+        "buffer/enqueue/",
+        internal_api_views.enqueue_notification_buffer,
+        name="enqueue_notification_buffer",
+    ),
+    path(
+        "buffer/process/",
+        internal_api_views.process_notification_buffer,
+        name="process_notification_buffer",
+    ),
     # Notification data endpoints for workers
     path(
         "pipeline/<str:pipeline_id>/notifications/",
diff --git a/backend/notification_v2/migrations/0003_add_notification_buffer.py b/backend/notification_v2/migrations/0003_add_notification_buffer.py
new file mode 100644
index 0000000000..7f5224849f
--- /dev/null
+++ b/backend/notification_v2/migrations/0003_add_notification_buffer.py
@@ -0,0 +1,148 @@
+import uuid
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("account_v2", "0001_initial"),
+        ("notification_v2", "0002_notification_notify_on_failures"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="notification",
+            name="delivery_mode",
+            field=models.CharField(
+                choices=[("IMMEDIATE", "Immediate"), ("BATCHED", "Batched")],
+                default="IMMEDIATE",
+                max_length=16,
+                db_comment=(
+                    "IMMEDIATE fires on every completion (default, unchanged "
+                    "behavior). BATCHED buffers events and dispatches a single "
+                    "clubbed message per (org, webhook_url, auth_sig) every "
+                    "NOTIFICATION_CLUB_INTERVAL."
+                ),
+            ),
+        ),
+        migrations.CreateModel(
+            name="NotificationBuffer",
+            fields=[
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("modified_at", models.DateTimeField(auto_now=True)),
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                (
+                    "webhook_url",
+                    models.URLField(
+                        db_comment="Denormalized destination URL; grouping key.",
+                    ),
+                ),
+                (
+                    "payload",
+                    models.JSONField(
+                        db_comment=(
+                            "Pre-structured execution data (execution_id, status, "
+                            "error_message, pipeline_name, pipeline_type) — NOT a "
+                            "final rendered message. The renderer formats this at "
+                            "dispatch time."
+                        ),
+                    ),
+                ),
+                (
+                    "platform",
+                    models.CharField(
+                        choices=[("SLACK", "Slack"), ("API", "Api")],
+                        max_length=50,
+                        db_comment=(
+                            "SLACK / API — drives renderer selection at flush time."
+                        ),
+                    ),
+                ),
+                (
+                    "auth_sig",
+                    models.CharField(
+                        max_length=64,
+                        db_comment=(
+                            "SHA-256 hex of (auth_type + auth_key + auth_header), "
+                            "computed at enqueue time. Grouping key — never store "
+                            "raw credentials here."
+                        ),
+                    ),
+                ),
+                (
+                    "flush_after",
+                    models.DateTimeField(
+                        db_comment=(
+                            "created_at + NOTIFICATION_CLUB_INTERVAL, precomputed "
+                            "at enqueue. Read-at-enqueue contract: changing the "
+                            "env var only affects rows enqueued after the restart."
+                        ),
+                    ),
+                ),
+                ("dispatched_at", models.DateTimeField(blank=True, null=True)),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("PENDING", "Pending"),
+                            ("DISPATCHED", "Dispatched"),
+                            ("DEAD_LETTER", "Dead letter"),
+                        ],
+                        default="PENDING",
+                        max_length=16,
+                        db_comment=(
+                            "PENDING -> DISPATCHED on success, "
+                            "PENDING -> DEAD_LETTER on retry exhaustion."
+                        ),
+                    ),
+                ),
+                (
+                    "notification",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="buffer_rows",
+                        to="notification_v2.notification",
+                        db_comment=(
+                            "Source Notification. Cascade-delete is intentional: "
+                            "removing a Notification expresses intent to stop all "
+                            "future deliveries, including buffered ones."
+                        ),
+                    ),
+                ),
+                (
+                    "organization",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="notification_buffer_rows",
+                        to="account_v2.organization",
+                        db_comment=(
+                            "Tenant scope. Mandatory grouping key — prevents "
+                            "cross-tenant leakage at flush time."
+                        ),
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Notification Buffer",
+                "verbose_name_plural": "Notification Buffers",
+                "db_table": "notification_buffer",
+            },
+        ),
+        migrations.AddIndex(
+            model_name="notificationbuffer",
+            index=models.Index(
+                condition=models.Q(("status", "PENDING")),
+                fields=["organization", "webhook_url", "auth_sig", "flush_after"],
+                name="idx_notif_buffer_pending",
+            ),
+        ),
+    ]
diff --git a/backend/notification_v2/models.py b/backend/notification_v2/models.py
index e5238ec176..a8b077f339 100644
--- a/backend/notification_v2/models.py
+++ b/backend/notification_v2/models.py
@@ -1,13 +1,21 @@
 import uuid
 
+from account_v2.models import Organization
 from api_v2.models import APIDeployment
 from django.db import models
 from pipeline_v2.models import Pipeline
 from utils.models.base_model import BaseModel
 
-from .enums import AuthorizationType, NotificationType, PlatformType
+from .enums import (
+    AuthorizationType,
+    BufferStatus,
+    DeliveryMode,
+    NotificationType,
+    PlatformType,
+)
 
 NOTIFICATION_NAME_MAX_LENGTH = 255
+AUTH_SIG_LENGTH = 64  # SHA-256 hex digest
 
 
 class Notification(BaseModel):
@@ -55,6 +63,16 @@ class Notification(BaseModel):
             "(default), fire on every terminal completion."
         ),
     )
+    delivery_mode = models.CharField(
+        max_length=16,
+        choices=DeliveryMode.choices(),
+        default=DeliveryMode.IMMEDIATE.value,
+        db_comment=(
+            "IMMEDIATE fires on every completion (default, unchanged behavior). "
+            "BATCHED buffers events and dispatches a single clubbed message per "
+            "(org, webhook_url, auth_sig) every NOTIFICATION_CLUB_INTERVAL."
+        ),
+    )
     # Foreign keys to specific models
     pipeline = models.ForeignKey(
         Pipeline,
@@ -100,3 +118,92 @@ def __str__(self):
             f"Notification {self.id}: (Type: {self.notification_type}, "
             f"Platform: {self.platform}, Url: {self.url}))"
         )
+
+
+class NotificationBuffer(BaseModel):
+    """Per-execution event buffered for a BATCHED notification.
+
+    One row is written per workflow completion when the source Notification
+    has delivery_mode=BATCHED. The flush job groups rows by
+    (organization, webhook_url, auth_sig), renders one clubbed message per
+    group, and dispatches via the existing send_webhook_notification Celery
+    task. Group key includes auth_sig because two notifications may share the
+    same URL but use different credentials — they must dispatch separately.
+    """
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    notification = models.ForeignKey(
+        Notification,
+        on_delete=models.CASCADE,
+        related_name="buffer_rows",
+        db_comment=(
+            "Source Notification. Cascade-delete is intentional: removing a "
+            "Notification expresses intent to stop all future deliveries, "
+            "including buffered ones."
+        ),
+    )
+    organization = models.ForeignKey(
+        Organization,
+        on_delete=models.CASCADE,
+        related_name="notification_buffer_rows",
+        db_comment=(
+            "Tenant scope. Mandatory grouping key — prevents cross-tenant "
+            "leakage at flush time."
+        ),
+    )
+    webhook_url = models.URLField(
+        db_comment="Denormalized destination URL; grouping key.",
+    )
+    payload = models.JSONField(
+        db_comment=(
+            "Pre-structured execution data (execution_id, status, error_message, "
+            "pipeline_name, pipeline_type) — NOT a final rendered message. The "
+            "renderer formats this at dispatch time."
+        ),
+    )
+    platform = models.CharField(
+        max_length=50,
+        choices=PlatformType.choices(),
+        db_comment="SLACK / API — drives renderer selection at flush time.",
+    )
+    auth_sig = models.CharField(
+        max_length=AUTH_SIG_LENGTH,
+        db_comment=(
+            "SHA-256 hex of (auth_type + auth_key + auth_header), computed at "
+            "enqueue time. Grouping key — never store raw credentials here."
+        ),
+    )
+    flush_after = models.DateTimeField(
+        db_comment=(
+            "created_at + NOTIFICATION_CLUB_INTERVAL, precomputed at enqueue. "
+            "Read-at-enqueue contract: changing the env var only affects rows "
+            "enqueued after the restart."
+        ),
+    )
+    dispatched_at = models.DateTimeField(null=True, blank=True)
+    status = models.CharField(
+        max_length=16,
+        choices=BufferStatus.choices(),
+        default=BufferStatus.PENDING.value,
+        db_comment="PENDING -> DISPATCHED on success, PENDING -> DEAD_LETTER on retry exhaustion.",
+    )
+
+    class Meta:
+        verbose_name = "Notification Buffer"
+        verbose_name_plural = "Notification Buffers"
+        db_table = "notification_buffer"
+        indexes = [
+            # Partial covering index — supports Index Only Scans on the flush
+            # GROUP BY query and bounds index size to live PENDING backlog.
+            models.Index(
+                fields=["organization", "webhook_url", "auth_sig", "flush_after"],
+                name="idx_notif_buffer_pending",
+                condition=models.Q(status=BufferStatus.PENDING.value),
+            ),
+        ]
+
+    def __str__(self) -> str:
+        return (
+            f"NotificationBuffer {self.id}: status={self.status} "
+            f"flush_after={self.flush_after.isoformat() if self.flush_after else 'n/a'}"
+        )
diff --git a/backend/notification_v2/serializers.py b/backend/notification_v2/serializers.py
index 4cb4f3c4cb..b2602929fe 100644
--- a/backend/notification_v2/serializers.py
+++ b/backend/notification_v2/serializers.py
@@ -1,10 +1,18 @@
 from rest_framework import serializers
 from utils.input_sanitizer import validate_name_field
 
-from .enums import AuthorizationType, NotificationType, PlatformType
+from .enums import AuthorizationType, DeliveryMode, NotificationType, PlatformType
 from .models import Notification
 
 
+class NotificationSettingsSerializer(serializers.Serializer):
+    """Org-scoped notification batching settings (UNS-611 v2)."""
+
+    # No min/max here: mfbt is silent on bounds. Backend ConfigSpec accepts
+    # any int; constraining is a follow-up if/when product gives a number.
+    club_interval_seconds = serializers.IntegerField()
+
+
 class NotificationSerializer(serializers.ModelSerializer):
     notification_type = serializers.ChoiceField(choices=NotificationType.choices())
     authorization_type = serializers.ChoiceField(choices=AuthorizationType.choices())
@@ -13,6 +21,11 @@ class NotificationSerializer(serializers.ModelSerializer):
         max_value=4, min_value=0, default=0, required=False
     )
     notify_on_failures = serializers.BooleanField(default=False, required=False)
+    delivery_mode = serializers.ChoiceField(
+        choices=DeliveryMode.choices(),
+        default=DeliveryMode.IMMEDIATE.value,
+        required=False,
+    )
 
     class Meta:
         model = Notification
diff --git a/backend/notification_v2/tasks.py b/backend/notification_v2/tasks.py
new file mode 100644
index 0000000000..a143f9295e
--- /dev/null
+++ b/backend/notification_v2/tasks.py
@@ -0,0 +1,51 @@
+"""Celery tasks owned by notification_v2.
+
+Currently hosts ``mark_buffer_dead_letter`` — a thin task attached as a
+Celery ``link_error`` to the clubbed dispatch chain. When the underlying
+``send_webhook_notification`` task exhausts retries, this task converts
+the buffered rows from PENDING/DISPATCHED to terminal DEAD_LETTER so the
+flush job will not re-pick them.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterable
+from typing import Any
+
+from backend.celery_service import app as celery_app
+from notification_v2.enums import BufferStatus
+from notification_v2.models import NotificationBuffer
+
+logger = logging.getLogger(__name__)
+
+
+@celery_app.task(name="notification_v2.mark_buffer_dead_letter")
+def mark_buffer_dead_letter(
+    request: Any,
+    exc: Any = None,
+    traceback: Any = None,
+    *,
+    buffer_row_ids: Iterable[str] | None = None,
+) -> int:
+    """Mark a clubbed dispatch's rows as DEAD_LETTER on terminal failure.
+
+    Celery's ``link_error`` signature passes ``(request, exc, traceback)`` to
+    the callback; the actual buffer ids are bound at dispatch time via task
+    kwargs. Returns the row count for visibility in flower.
+    """
+    if not buffer_row_ids:
+        logger.warning(
+            "mark_buffer_dead_letter invoked without buffer_row_ids — nothing to do"
+        )
+        return 0
+    ids = list(buffer_row_ids)
+    updated: int = NotificationBuffer.objects.filter(id__in=ids).update(
+        status=BufferStatus.DEAD_LETTER.value
+    )
+    logger.warning(
+        "metric=notification_batch_dispatched_total result=dead_letter rows=%d " "exc=%r",
+        updated,
+        exc,
+    )
+    return updated
diff --git a/backend/notification_v2/tests/__init__.py b/backend/notification_v2/tests/__init__.py
deleted file mode 100644
index e69de29bb2..0000000000
diff --git a/backend/notification_v2/urls.py b/backend/notification_v2/urls.py
index 2e356b4003..e41bb00f3a 100644
--- a/backend/notification_v2/urls.py
+++ b/backend/notification_v2/urls.py
@@ -1,7 +1,7 @@
 from django.urls import path
 from rest_framework.urlpatterns import format_suffix_patterns
 
-from .views import NotificationViewSet
+from .views import NotificationSettingsView, NotificationViewSet
 
 notification_list = NotificationViewSet.as_view({"get": "list", "post": "create"})
 notification_detail = NotificationViewSet.as_view(
@@ -16,6 +16,13 @@
 urlpatterns = format_suffix_patterns(
     [
         path("", notification_list, name="notification-list"),
+        # Org-scoped notification batching settings (UNS-611 v2). Mounted
+        # before the <uuid:pk> route so "settings" is not interpreted as a UUID.
+        path(
+            "settings/",
+            NotificationSettingsView.as_view(),
+            name="notification-settings",
+        ),
         path("<uuid:pk>/", notification_detail, name="notification-detail"),
         path(
             "pipeline/<uuid:pipeline_uuid>/",
diff --git a/backend/notification_v2/views.py b/backend/notification_v2/views.py
index 1410256ab6..207d4acc6b 100644
--- a/backend/notification_v2/views.py
+++ b/backend/notification_v2/views.py
@@ -1,14 +1,26 @@
+import logging
+
 from api_v2.deployment_helper import DeploymentHelper
 from api_v2.exceptions import APINotFound
+from configuration.enums import ConfigKey
+from configuration.models import Configuration
 from pipeline_v2.exceptions import PipelineNotFound
 from pipeline_v2.models import Pipeline
 from pipeline_v2.pipeline_processor import PipelineProcessor
-from rest_framework import viewsets
+from platform_api.permissions import IsOrganizationAdmin
+from rest_framework import status, viewsets
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from utils.user_context import UserContext
 
 from notification_v2.constants import NotificationUrlConstant
 
 from .models import Notification
-from .serializers import NotificationSerializer
+from .serializers import NotificationSerializer, NotificationSettingsSerializer
+
+logger = logging.getLogger(__name__)
 
 
 class NotificationViewSet(viewsets.ModelViewSet):
@@ -39,3 +51,47 @@ def get_queryset(self):
             queryset = queryset.filter(api=api)
 
         return queryset
+
+
+class NotificationSettingsView(APIView):
+    """Org-scoped notification batching settings — currently just the club interval.
+
+    GET returns the org's effective interval (override or env-derived default).
+    PATCH writes/updates the override via the generic configuration KV table.
+
+    Read-at-enqueue contract (mfbt §EC-2 / §EC-8): updates take effect for
+    notifications enqueued after the change. Existing PENDING buffer rows
+    keep their original flush_after.
+    """
+
+    permission_classes = [IsAuthenticated, IsOrganizationAdmin]
+
+    def get(self, request: Request) -> Response:
+        organization = UserContext.get_organization()
+        value = Configuration.get_value_by_organization(
+            ConfigKey.NOTIFICATION_CLUB_INTERVAL, organization
+        )
+        return Response({"club_interval_seconds": int(value)})
+
+    def patch(self, request: Request) -> Response:
+        serializer = NotificationSettingsSerializer(data=request.data, partial=True)
+        serializer.is_valid(raise_exception=True)
+        organization = UserContext.get_organization()
+        new_value = serializer.validated_data.get("club_interval_seconds")
+        if new_value is None:
+            return Response(
+                {"detail": "club_interval_seconds is required."},
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        # ConfigKey.cast_value enforces type + any future bounds; bubble its
+        # ValueError up as a 400 instead of letting it 500.
+        try:
+            ConfigKey.NOTIFICATION_CLUB_INTERVAL.cast_value(new_value)
+        except ValueError as exc:
+            return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST)
+        Configuration.objects.update_or_create(
+            organization=organization,
+            key=ConfigKey.NOTIFICATION_CLUB_INTERVAL.name,
+            defaults={"value": str(new_value), "enabled": True},
+        )
+        return Response({"club_interval_seconds": int(new_value)})
diff --git a/backend/pipeline_v2/notification.py b/backend/pipeline_v2/notification.py
index ec82145054..5a40a37506 100644
--- a/backend/pipeline_v2/notification.py
+++ b/backend/pipeline_v2/notification.py
@@ -1,6 +1,6 @@
 import logging
 
-from notification_v2.helper import NotificationHelper
+from notification_v2.helper import dispatch_with_delivery_mode
 from notification_v2.models import Notification
 from workflow_manager.workflow_v2.enums import ExecutionStatus
 from workflow_manager.workflow_v2.models.execution import WorkflowExecution
@@ -90,7 +90,8 @@ def send(self) -> None:
             successful_files=successful_files,
             failed_files=failed_files,
         )
-
-        NotificationHelper.send_notification(
-            notifications=self.notifications, payload=payload_dto.to_dict()
+        dispatch_with_delivery_mode(
+            list(self.notifications),
+            payload_dto.to_dict(),
+            error_context=f"pipeline={self.pipeline.id}",
         )
diff --git a/backend/utils/cors_origin.py b/backend/utils/cors_origin.py
new file mode 100644
index 0000000000..aa640c6bc6
--- /dev/null
+++ b/backend/utils/cors_origin.py
@@ -0,0 +1,81 @@
+"""CORS origin helpers used by Django settings and SocketIO log events.
+
+Kept free of Django imports so the matching/normalization logic can be unit
+tested without bootstrapping the full project.
+"""
+
+from __future__ import annotations
+
+import re
+from urllib.parse import urlparse
+
+
+class RegexOrigin:
+    """Origin pattern that compares to strings via regex match.
+
+    python-socketio enforces CORS with ``origin in allowed_origins`` during the
+    engine.io handshake — overriding ``__eq__`` lets a single list entry cover
+    a wildcard subdomain so bad origins are rejected before ``connect`` runs.
+
+    Instances are intentionally unhashable: a hashable object must satisfy
+    ``a == b ⇒ hash(a) == hash(b)``, but ``__eq__`` here is asymmetric across
+    types (matches many strings, hashes only one pattern). Any code that put
+    one in a ``set``/``frozenset`` would silently break the CORS gate, so
+    ``__hash__`` is disabled to fail loud at construction time instead.
+
+    ``fullmatch`` (not ``match``) is used so ``$`` doesn't permit a trailing
+    newline — defense in depth even though WSGI strips them upstream.
+    """
+
+    def __init__(self, pattern: str) -> None:
+        self._regex = re.compile(pattern)
+
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, str):
+            return self._regex.fullmatch(other) is not None
+        return NotImplemented
+
+    __hash__ = None  # see class docstring
+
+
+def normalize_web_app_origin(env_value: str) -> tuple[str, str, str]:
+    """Parse and canonicalize ``WEB_APP_ORIGIN_URL`` for CORS/CSRF allow-lists.
+
+    Returns ``(origin, wildcard_origin, subdomain_regex)``:
+
+    - ``origin``: ``scheme://host[:port]`` form. Hostname is lowercased and
+      explicit default ports (:80 for http, :443 for https) are dropped, so
+      it matches what browsers serialize per RFC 6454.
+    - ``wildcard_origin``: same with a literal ``*.`` subdomain prefix, for
+      Django's ``CSRF_TRUSTED_ORIGINS`` (which fnmatches ``*``).
+    - ``subdomain_regex``: anchored pattern matching any subdomain of the
+      configured netloc, for ``CORS_ALLOWED_ORIGIN_REGEXES`` and SocketIO
+      via ``RegexOrigin``.
+
+    Raises:
+        ValueError: if the env value is not an http(s) URL with a host.
+    """
+    parsed = urlparse(env_value)
+    # `parsed.port` is a property that raises ValueError on malformed/out-of-range
+    # ports (e.g. `:abc`, `:99999`). Catch it here so misconfig surfaces with the
+    # same actionable message as every other validation failure.
+    try:
+        port = parsed.port
+    except ValueError as exc:
+        raise ValueError(
+            f"WEB_APP_ORIGIN_URL must be of the form http(s)://host[:port], "
+            f"got: {parsed.geturl()!r}"
+        ) from exc
+    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
+        raise ValueError(
+            f"WEB_APP_ORIGIN_URL must be of the form http(s)://host[:port], "
+            f"got: {parsed.geturl()!r}"
+        )
+    default_port = {"http": 80, "https": 443}[parsed.scheme]
+    netloc = parsed.hostname
+    if port and port != default_port:
+        netloc = f"{netloc}:{port}"
+    origin = f"{parsed.scheme}://{netloc}"
+    wildcard = f"{parsed.scheme}://*.{netloc}"
+    subdomain_regex = rf"^{re.escape(parsed.scheme)}://[^/]+\.{re.escape(netloc)}$"
+    return origin, wildcard, subdomain_regex
diff --git a/backend/utils/log_events.py b/backend/utils/log_events.py
index f2ff725820..dcb66ed81c 100644
--- a/backend/utils/log_events.py
+++ b/backend/utils/log_events.py
@@ -11,9 +11,15 @@
 from unstract.core.data_models import LogDataDTO
 from unstract.core.log_utils import get_validated_log_data, store_execution_log
 from utils.constants import ExecutionLogConstants
+from utils.cors_origin import RegexOrigin
 
 logger = logging.getLogger(__name__)
 
+
+_cors_allowed_origins: list[Any] = list(settings.CORS_ALLOWED_ORIGINS)
+for _pattern in getattr(settings, "CORS_ALLOWED_ORIGIN_REGEXES", []):
+    _cors_allowed_origins.append(RegexOrigin(_pattern))
+
 _kombu_kwargs: dict[str, Any] = {"url": settings.SOCKET_IO_MANAGER_URL}
 if getattr(settings, "SOCKET_IO_TRANSPORT_OPTIONS", None):
     _kombu_kwargs["connection_options"] = {
@@ -23,7 +29,7 @@
 sio = socketio.Server(
     # Allowed values: {threading, eventlet, gevent, gevent_uwsgi}
     async_mode="threading",
-    cors_allowed_origins=settings.CORS_ALLOWED_ORIGINS,
+    cors_allowed_origins=_cors_allowed_origins,
     logger=False,
     engineio_logger=False,
     always_connect=True,
diff --git a/backend/utils/tests/test_cors_origin.py b/backend/utils/tests/test_cors_origin.py
new file mode 100644
index 0000000000..65628b4405
--- /dev/null
+++ b/backend/utils/tests/test_cors_origin.py
@@ -0,0 +1,197 @@
+"""Regression tests for utils.cors_origin — the CORS origin matcher and
+URL normalizer that gate browser ``Origin`` headers on every Django request
+and SocketIO handshake.
+
+UN-3439: production socket connections silently failed for wildcard subdomains
+because python-socketio does exact-string comparison. These tests pin the
+contract so the next refactor can't reopen the hole.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+
+from utils.cors_origin import RegexOrigin, normalize_web_app_origin
+
+
+class TestRegexOrigin:
+    def test_subdomain_matches(self):
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        assert ("https://app.example.com" == ro) is True
+        assert ("https://api.example.com" == ro) is True
+
+    def test_deep_subdomain_matches(self):
+        """Multi-level subdomains are accepted — DNS-owned by the same party."""
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        assert ("https://dev.env.example.com" == ro) is True
+
+    def test_apex_does_not_match(self):
+        """Apex is covered by the exact-match CORS_ALLOWED_ORIGINS entry, not
+        the wildcard regex."""
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        assert ("https://example.com" == ro) is False
+
+    def test_lookalike_rejected(self):
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        assert ("https://attacker-example.com" == ro) is False
+        assert ("https://example.com.attacker.com" == ro) is False
+        assert ("https://x.example.com.attacker.com" == ro) is False
+
+    def test_wrong_scheme_rejected(self):
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        # NOSONAR — the `http://` URL is intentional test data: we are
+        # asserting it is *rejected* by an https-scoped pattern.
+        assert ("http://app.example.com" == ro) is False  # NOSONAR
+
+    def test_trailing_newline_rejected(self):
+        """``fullmatch`` (not ``match``) is required so ``$`` doesn't permit
+        a trailing ``\\n`` per Python regex semantics."""
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        assert ("https://app.example.com\n" == ro) is False
+
+    def test_in_operator_routes_to_eq(self):
+        """``origin in allowed_origins`` is exactly how python-socketio gates
+        the engine.io handshake — verify Python's reflected ``__eq__`` kicks in."""
+        allowed = [RegexOrigin(r"^https://[^/]+\.example\.com$")]
+        assert "https://app.example.com" in allowed
+        assert "https://evil.com" not in allowed
+
+    def test_non_string_returns_not_implemented(self):
+        """``__eq__`` must return the ``NotImplemented`` sentinel (not
+        ``False``) for non-strings so Python's reflected-equality protocol
+        can fall back to identity. Tested via direct dunder calls because
+        ``ro == None`` short-circuits before reaching ``__eq__``."""
+        ro = RegexOrigin(r"^x$")
+        assert ro.__eq__(None) is NotImplemented
+        assert ro.__eq__(42) is NotImplemented
+        assert ro.__eq__([]) is NotImplemented
+
+    def test_unhashable(self):
+        """``__hash__ = None`` prevents the equality/hash contract from being
+        violated if anyone wraps the allow-list in a ``set``/``frozenset``."""
+        ro = RegexOrigin(r"^x$")
+        with pytest.raises(TypeError):
+            hash(ro)
+        with pytest.raises(TypeError):
+            # `len({ro})` builds the set (calling __hash__) and consumes it via
+            # an explicit function call — keeps ruff from collapsing the
+            # statement and Sonar from flagging it as side-effect-free.
+            len({ro})
+
+    def test_no_redos(self):
+        """Pattern must complete on hostile input — ``[^/]+`` has no nested
+        quantifiers so backtracking is bounded. Threshold is generous (500ms)
+        so noisy CI runners don't flake; ReDoS would blow up by orders of
+        magnitude past this."""
+        ro = RegexOrigin(r"^https://[^/]+\.example\.com$")
+        hostile = "https://" + "a" * 10000 + ".evil.com"
+        start = time.perf_counter()
+        _ = hostile in [ro]
+        assert time.perf_counter() - start < 0.5
+
+
+class TestNormalizeWebAppOrigin:
+    def test_basic_https(self):
+        origin, wildcard, _ = normalize_web_app_origin("https://example.com")
+        assert origin == "https://example.com"
+        assert wildcard == "https://*.example.com"
+
+    def test_strips_trailing_slash(self):
+        origin, _, _ = normalize_web_app_origin("https://example.com/")
+        assert origin == "https://example.com"
+
+    def test_strips_path_and_query(self):
+        origin, _, _ = normalize_web_app_origin("https://example.com/path?q=1")
+        assert origin == "https://example.com"
+
+    def test_lowercases_hostname(self):
+        """Browsers serialize ``Origin`` with a lowercase host (RFC 6454);
+        django-cors-headers does case-sensitive string compare, so the env
+        value must be lowercased to match."""
+        origin, _, _ = normalize_web_app_origin("https://APP.EXAMPLE.COM")
+        assert origin == "https://app.example.com"
+
+    def test_drops_default_https_port(self):
+        """Browsers omit the explicit default port from ``Origin`` per
+        RFC 6454 — keeping ``:443`` would silently break exact match."""
+        origin, _, _ = normalize_web_app_origin("https://example.com:443")
+        assert origin == "https://example.com"
+
+    def test_drops_default_http_port(self):
+        # NOSONAR — `http://` URLs are intentional test data for the port
+        # normalization logic, not a runtime use of the insecure protocol.
+        origin, _, _ = normalize_web_app_origin("http://example.com:80")  # NOSONAR
+        assert origin == "http://example.com"  # NOSONAR
+
+    def test_keeps_non_default_port(self):
+        origin, wildcard, _ = normalize_web_app_origin("https://example.com:8443")
+        assert origin == "https://example.com:8443"
+        assert wildcard == "https://*.example.com:8443"
+
+    def test_localhost_default(self):
+        origin, _, _ = normalize_web_app_origin("http://localhost:3000")
+        assert origin == "http://localhost:3000"
+
+    @pytest.mark.parametrize(
+        "bad",
+        [
+            "",
+            "not-a-url",
+            "example.com",          # missing scheme
+            "//example.com",        # protocol-relative
+            "https://",             # missing host
+            "ftp://example.com",  # non-browser scheme  # NOSONAR — test input asserting ftp is rejected
+            "ws://example.com",  # not a top-level browser scheme
+            "https://example.com:abc",  # malformed port — urlparse raises on .port access
+            "https://example.com:99999",  # out-of-range port
+        ],
+    )
+    def test_rejects_misconfigured(self, bad):
+        """Fail fast at startup so misconfigured envs can't silently produce
+        CORS rules that match nothing real."""
+        with pytest.raises(ValueError, match="WEB_APP_ORIGIN_URL"):
+            normalize_web_app_origin(bad)
+
+
+class TestSubdomainRegexEndToEnd:
+    """End-to-end: the regex returned by ``normalize_web_app_origin`` is what
+    actually gates production. Verify via ``RegexOrigin`` (same path as
+    SocketIO uses)."""
+
+    def test_apex_env_accepts_subdomains(self):
+        _, _, pattern = normalize_web_app_origin("https://us-central.unstract.com")
+        ro = RegexOrigin(pattern)
+        # The exact failing origins from UN-3439:
+        assert "https://dev.env.us-central.unstract.com" in [ro]
+        assert "https://test.env.us-central.unstract.com" in [ro]
+
+    def test_apex_rejected_by_regex(self):
+        """Apex itself is *not* matched by the wildcard regex — that's the
+        exact-match CORS_ALLOWED_ORIGINS entry's job."""
+        _, _, pattern = normalize_web_app_origin("https://example.com")
+        ro = RegexOrigin(pattern)
+        assert "https://example.com" not in [ro]
+
+    @pytest.mark.parametrize(
+        "spoof",
+        [
+            "https://attacker-example.com",
+            "https://x.example.com.attacker.com",
+            "https://example.com.attacker.com",
+            "http://app.example.com",  # wrong scheme  # NOSONAR — test input asserting http is rejected
+            "https://app.example.com\n",  # trailing newline
+        ],
+    )
+    def test_spoofed_origins_rejected(self, spoof):
+        _, _, pattern = normalize_web_app_origin("https://example.com")
+        ro = RegexOrigin(pattern)
+        assert spoof not in [ro], f"should reject {spoof!r}"
+
+    def test_uppercase_env_still_accepts_lowercase_origin(self):
+        """Browser sends lowercase even if the env was set with uppercase;
+        normalization must canonicalize before regex compilation."""
+        _, _, pattern = normalize_web_app_origin("https://APP.EXAMPLE.COM")
+        ro = RegexOrigin(pattern)
+        assert "https://sub.app.example.com" in [ro]
diff --git a/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx b/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
index 65460376d4..17066ed06b 100644
--- a/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
+++ b/frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
@@ -2,6 +2,12 @@ import { Button, Checkbox, Form, Input, Select, Space } from "antd";
 import PropTypes from "prop-types";
 import { useEffect, useState } from "react";
 import { getBackendErrorDetail } from "../../../helpers/GetStaticData";
+import { useAxiosPrivate } from "../../../hooks/useAxiosPrivate";
+import { useSessionStore } from "../../../store/session-store";
+
+// Used only when the org's batch interval can't be fetched (network or auth
+// failure). Backend's env-derived default is also 30 min, so this matches.
+const FALLBACK_BATCH_INTERVAL_MINUTES = 30;
 
 const DEFAULT_FORM_DETAILS = {
   name: "",
@@ -13,6 +19,7 @@ const DEFAULT_FORM_DETAILS = {
   is_active: false,
   max_retries: 0,
   notify_on_failures: false,
+  delivery_mode: "IMMEDIATE",
   pipeline: "",
   api: "",
   url: "",
@@ -68,6 +75,37 @@ function CreateNotification({
   const [formDetails, setFormDetails] = useState(DEFAULT_FORM_DETAILS);
   const [backendErrors, setBackendErrors] = useState(null);
   const [resetForm, setResetForm] = useState(false);
+  const [batchIntervalMinutes, setBatchIntervalMinutes] = useState(
+    FALLBACK_BATCH_INTERVAL_MINUTES,
+  );
+  const axiosPrivate = useAxiosPrivate();
+  const { sessionDetails } = useSessionStore();
+
+  useEffect(() => {
+    // Read live org-scoped interval (UNS-611 v2). Fall back silently to the
+    // hardcoded 30-min default — the dropdown still labels something useful.
+    if (!sessionDetails?.orgId) {
+      return;
+    }
+    axiosPrivate({
+      method: "GET",
+      url: `/api/v1/unstract/${sessionDetails.orgId}/notifications/settings/`,
+    })
+      .then((res) => {
+        const seconds = res?.data?.club_interval_seconds;
+        if (typeof seconds === "number" && seconds > 0) {
+          setBatchIntervalMinutes(Math.max(1, Math.round(seconds / 60)));
+        }
+      })
+      .catch(() => {
+        // Non-fatal — keep fallback.
+      });
+  }, [sessionDetails?.orgId]);
+
+  const deliveryModes = [
+    { value: "IMMEDIATE", label: "Immediate" },
+    { value: "BATCHED", label: "Batched" },
+  ];
 
   useEffect(() => {
     if (editDetails) {
@@ -84,7 +122,18 @@ function CreateNotification({
   }, [formDetails]);
 
   const handleInputChange = (changedValues, allValues) => {
-    setFormDetails({ ...formDetails, ...allValues });
+    let nextValues = { ...formDetails, ...allValues };
+    // Failure alerts must not be delayed by the batch window — auto-select
+    // IMMEDIATE the moment the box is checked. The user can still override
+    // to BATCHED afterward and that choice will stick.
+    if (
+      Object.hasOwn(changedValues, "notify_on_failures") &&
+      changedValues.notify_on_failures === true
+    ) {
+      nextValues = { ...nextValues, delivery_mode: "IMMEDIATE" };
+      form.setFieldsValue({ delivery_mode: "IMMEDIATE" });
+    }
+    setFormDetails(nextValues);
     const changedFieldName = Object.keys(changedValues)[0];
     form.setFields([
       {
@@ -228,6 +277,18 @@ function CreateNotification({
       >
         <Checkbox>Notify on failures only</Checkbox>
       </Form.Item>
+      <Form.Item
+        label="Delivery Mode"
+        name="delivery_mode"
+        tooltip="Immediate fires on every completion. Batched buffers events and dispatches a single clubbed message per webhook every batch interval."
+        extra={
+          formDetails.delivery_mode === "BATCHED"
+            ? `Notifications will be batched and sent every ${batchIntervalMinutes} minutes. Org admins can change this in Platform Settings.`
+            : null
+        }
+      >
+        <Select options={deliveryModes} />
+      </Form.Item>
       <Form.Item className="display-flex-right">
         <Space>
           <Button onClick={() => setIsForm(false)}>Cancel</Button>
diff --git a/frontend/src/components/settings/platform/PlatformSettings.jsx b/frontend/src/components/settings/platform/PlatformSettings.jsx
index f255da6a73..0b39cebecf 100644
--- a/frontend/src/components/settings/platform/PlatformSettings.jsx
+++ b/frontend/src/components/settings/platform/PlatformSettings.jsx
@@ -3,7 +3,17 @@ import {
   CopyOutlined,
   DeleteOutlined,
 } from "@ant-design/icons";
-import { Button, Col, Divider, Input, Radio, Row, Typography } from "antd";
+import {
+  Button,
+  Col,
+  Divider,
+  Input,
+  InputNumber,
+  Radio,
+  Row,
+  Space,
+  Typography,
+} from "antd";
 import { useEffect, useState } from "react";
 import { useNavigate } from "react-router-dom";
 
@@ -38,6 +48,9 @@ function PlatformSettings() {
   const [keys, setKeys] = useState(defaultKeys);
   const [isLoadingIndex, setLoadingIndex] = useState(null);
   const [isDeletingIndex, setDeletingIndex] = useState(null);
+  // UI shows minutes; wire format (and ConfigSpec.value) is seconds.
+  const [batchIntervalMinutes, setBatchIntervalMinutes] = useState(null);
+  const [isSavingInterval, setIsSavingInterval] = useState(false);
   const { sessionDetails } = useSessionStore();
   const { setAlertDetails } = useAlertStore();
   const axiosPrivate = useAxiosPrivate();
@@ -45,6 +58,58 @@ function PlatformSettings() {
   const handleException = useExceptionHandler();
   const { setPostHogCustomEvent } = usePostHogEvents();
 
+  useEffect(() => {
+    // Load org-scoped batch interval (UNS-611 v2). Falls back silently to
+    // null on failure so the rest of the page still renders.
+    axiosPrivate({
+      method: "GET",
+      url: `/api/v1/unstract/${sessionDetails?.orgId}/notifications/settings/`,
+    })
+      .then((res) => {
+        const seconds = res?.data?.club_interval_seconds;
+        if (typeof seconds === "number" && seconds > 0) {
+          setBatchIntervalMinutes(Math.round(seconds / 60));
+        }
+      })
+      .catch(() => {
+        // Non-fatal — admin just won't see a pre-filled value.
+      });
+  }, []);
+
+  const handleSaveInterval = () => {
+    if (!batchIntervalMinutes || batchIntervalMinutes < 1) {
+      setAlertDetails({
+        type: "error",
+        content: "Batch interval must be a positive number of minutes.",
+      });
+      return;
+    }
+    setIsSavingInterval(true);
+    axiosPrivate({
+      method: "PATCH",
+      url: `/api/v1/unstract/${sessionDetails?.orgId}/notifications/settings/`,
+      headers: {
+        "X-CSRFToken": sessionDetails?.csrfToken,
+        "Content-Type": "application/json",
+      },
+      data: { club_interval_seconds: batchIntervalMinutes * 60 },
+    })
+      .then(() => {
+        setAlertDetails({
+          type: "success",
+          content: "Notification batch interval updated.",
+        });
+      })
+      .catch((err) => {
+        setAlertDetails(
+          handleException(err, "Failed to update batch interval"),
+        );
+      })
+      .finally(() => {
+        setIsSavingInterval(false);
+      });
+  };
+
   useEffect(() => {
     const requestOptions = {
       method: "GET",
@@ -331,6 +396,34 @@ function PlatformSettings() {
                   );
                 })}
               </div>
+              <Divider />
+              <div className="plt-set-batch-interval">
+                <Typography.Title level={5}>
+                  Notification batching
+                </Typography.Title>
+                <Typography.Text type="secondary">
+                  Batched notifications enqueued after a change pick up the new
+                  value; in-flight rows keep their original cadence.
+                </Typography.Text>
+                <div style={{ marginTop: 12 }}>
+                  <Space>
+                    <Typography.Text>Batch interval (minutes)</Typography.Text>
+                    <InputNumber
+                      min={1}
+                      value={batchIntervalMinutes}
+                      onChange={(v) => setBatchIntervalMinutes(v)}
+                      placeholder="e.g. 30"
+                    />
+                    <Button
+                      type="primary"
+                      onClick={handleSaveInterval}
+                      loading={isSavingInterval}
+                    >
+                      Save
+                    </Button>
+                  </Space>
+                </div>
+              </div>
             </div>
           </IslandLayout>
         </div>
diff --git a/unstract/sdk1/src/unstract/sdk1/adapters/base1.py b/unstract/sdk1/src/unstract/sdk1/adapters/base1.py
index cb1c738cf4..d1aa475ec4 100644
--- a/unstract/sdk1/src/unstract/sdk1/adapters/base1.py
+++ b/unstract/sdk1/src/unstract/sdk1/adapters/base1.py
@@ -474,6 +474,71 @@ def validate_model(adapter_metadata: dict[str, "Any"]) -> str:
             return f"vertex_ai/{model}"
 
 
+# AWS Bedrock auth helpers: shared by LLM and Embedding param classes.
+# `auth_type` is a UI-only selector (Access Keys vs IAM Role / Instance
+# Profile) that drives form rendering. The backend translates the user's
+# choice into actual credential handling here so that both validate()
+# methods stay symmetric and a single bug fix applies to both paths.
+_BEDROCK_AWS_KEY_FIELDS: tuple[str, ...] = (
+    "aws_access_key_id",
+    "aws_secret_access_key",
+)
+_BEDROCK_VALID_AUTH_TYPES: frozenset[str | None] = frozenset(
+    {None, "access_keys", "iam_role"}
+)
+
+
+def _resolve_bedrock_aws_credentials(
+    adapter_metadata: dict[str, "Any"],
+    validated: dict[str, "Any"],
+) -> dict[str, "Any"]:
+    """Apply auth_type semantics to the validated LiteLLM kwargs.
+
+    Three cases:
+    - ``auth_type == "iam_role"``: drop access keys unconditionally so a
+      previously-saved adapter switched into IAM Role mode does not leak
+      stale long-lived credentials. boto3's default credential chain
+      (IRSA / instance profile / env vars / AWS Profile) takes over.
+    - ``auth_type == "access_keys"`` (explicit): require non-blank values.
+      A blank submission must surface as a clear error rather than
+      silently fall through to the default chain (which would hide the
+      mistake and authenticate with whatever ambient creds the host has).
+    - ``auth_type is None`` (legacy adapters created before this field
+      existed): lenient strip of empty/missing keys. Preserves backwards
+      compatibility for stored configurations.
+
+    Raises:
+        ValueError: on unknown ``auth_type`` (typo / non-UI client) or on
+            blank credentials when ``auth_type == "access_keys"``.
+    """
+    auth_type = adapter_metadata.get("auth_type")
+    if auth_type not in _BEDROCK_VALID_AUTH_TYPES:
+        raise ValueError(
+            f"Unknown auth_type {auth_type!r}; expected one of "
+            f"{sorted(t for t in _BEDROCK_VALID_AUTH_TYPES if t)!r} or absent."
+        )
+
+    if auth_type == "iam_role":
+        for key in _BEDROCK_AWS_KEY_FIELDS:
+            validated.pop(key, None)
+        return validated
+
+    if auth_type == "access_keys":
+        for key in _BEDROCK_AWS_KEY_FIELDS:
+            value = validated.get(key)
+            if not isinstance(value, str) or not value.strip():
+                raise ValueError(f"{key} is required when auth_type is 'access_keys'.")
+        return validated
+
+    # Legacy adapters with no auth_type: strip blanks silently to
+    # preserve the pre-PR behaviour where empty key fields fell through
+    # to boto3's default chain.
+    for key in _BEDROCK_AWS_KEY_FIELDS:
+        if not validated.get(key):
+            validated.pop(key, None)
+    return validated
+
+
 class AWSBedrockLLMParameters(BaseChatCompletionParameters):
     """See https://docs.litellm.ai/docs/providers/bedrock."""
 
@@ -525,11 +590,12 @@ def validate(adapter_metadata: dict[str, "Any"]) -> dict[str, "Any"]:
                 result_metadata["thinking"] = thinking_config
                 result_metadata["temperature"] = 1
 
-        # Create validation metadata excluding control fields
+        # Create validation metadata excluding control fields. `auth_type` is
+        # a UI-only selector that drives form rendering; LiteLLM never sees it.
         validation_metadata = {
             k: v
             for k, v in result_metadata.items()
-            if k not in ("enable_thinking", "budget_tokens", "thinking")
+            if k not in ("enable_thinking", "budget_tokens", "thinking", "auth_type")
         }
 
         validated = AWSBedrockLLMParameters(**validation_metadata).model_dump()
@@ -538,7 +604,11 @@ def validate(adapter_metadata: dict[str, "Any"]) -> dict[str, "Any"]:
         if enable_thinking and "thinking" in result_metadata:
             validated["thinking"] = result_metadata["thinking"]
 
-        return validated
+        # Apply Bedrock auth semantics: IAM Role mode drops keys, Access
+        # Keys mode requires non-blank values, legacy (no auth_type) is
+        # lenient. Reads auth_type from result_metadata since validation_
+        # metadata strips it before Pydantic.
+        return _resolve_bedrock_aws_credentials(result_metadata, validated)
 
     @staticmethod
     def validate_model(adapter_metadata: dict[str, "Any"]) -> str:
@@ -959,8 +1029,10 @@ def validate_model(adapter_metadata: dict[str, "Any"]) -> str:
 class AWSBedrockEmbeddingParameters(BaseEmbeddingParameters):
     """See https://docs.litellm.ai/docs/providers/bedrock."""
 
-    aws_access_key_id: str | None
-    aws_secret_access_key: str | None
+    # Region is still mandatory — credentials are the only fields that
+    # may be absent (IAM Role / Instance Profile mode).
+    aws_access_key_id: str | None = None
+    aws_secret_access_key: str | None = None
     aws_region_name: str | None
 
     @staticmethod
@@ -973,7 +1045,16 @@ def validate(adapter_metadata: dict[str, "Any"]) -> dict[str, "Any"]:
         ):
             adapter_metadata["aws_region_name"] = adapter_metadata["region_name"]
 
-        return AWSBedrockEmbeddingParameters(**adapter_metadata).model_dump()
+        # `auth_type` is a UI-only selector; strip before LiteLLM kwargs.
+        validation_metadata = {
+            k: v for k, v in adapter_metadata.items() if k != "auth_type"
+        }
+
+        validated = AWSBedrockEmbeddingParameters(**validation_metadata).model_dump()
+
+        # Apply Bedrock auth semantics: IAM Role drops keys, Access Keys
+        # requires non-blank values, legacy (no auth_type) is lenient.
+        return _resolve_bedrock_aws_credentials(adapter_metadata, validated)
 
     @staticmethod
     def validate_model(adapter_metadata: dict[str, "Any"]) -> str:
diff --git a/unstract/sdk1/src/unstract/sdk1/adapters/embedding1/static/bedrock.json b/unstract/sdk1/src/unstract/sdk1/adapters/embedding1/static/bedrock.json
index 81829a0f49..200d8f5bfb 100644
--- a/unstract/sdk1/src/unstract/sdk1/adapters/embedding1/static/bedrock.json
+++ b/unstract/sdk1/src/unstract/sdk1/adapters/embedding1/static/bedrock.json
@@ -2,9 +2,7 @@
     "title": "Bedrock Embeddings",
     "type": "object",
     "required": [
-      "aws_secret_access_key",
       "region_name",
-      "aws_access_key_id",
       "model",
       "adapter_name"
     ],
@@ -21,18 +19,6 @@
         "default": "amazon.titan-embed-text-v2:0",
         "description": "Model name. Refer to [Bedrock's documentation](https://docs.aws.amazon.com/bedrock/latest/userguide/models-supported.html) for the list of available models."
       },
-      "aws_access_key_id": {
-        "type": "string",
-        "title": "AWS Access Key ID",
-        "description": "Provide your AWS Access Key ID",
-        "format": "password"
-      },
-      "aws_secret_access_key": {
-        "type": "string",
-        "title": "AWS Secret Access Key",
-        "description": "Provide your AWS Secret Access Key",
-        "format": "password"
-      },
       "region_name": {
         "type": "string",
         "title": "AWS Region name",
@@ -53,6 +39,60 @@
         "title": "Timeout",
         "default": 900,
         "description": "Timeout in seconds"
+      },
+      "auth_type": {
+        "type": "string",
+        "title": "Authentication Type",
+        "enum": [
+          "access_keys",
+          "iam_role"
+        ],
+        "enumNames": [
+          "Access Keys",
+          "IAM Role / Instance Profile (on-prem AWS only)"
+        ],
+        "default": "access_keys",
+        "description": "Choose **Access Keys** for any deployment (provide your AWS Access Key ID and Secret). Choose **IAM Role / Instance Profile** only when Unstract is hosted on AWS infrastructure that already has ambient credentials — for example EKS pods with IRSA, ECS tasks with a task role, or EC2 instances with an instance profile. The on-prem option requires no further input; boto3 picks up the host's identity automatically."
+      }
+    },
+    "dependencies": {
+      "auth_type": {
+        "oneOf": [
+          {
+            "properties": {
+              "auth_type": {
+                "enum": [
+                  "access_keys"
+                ]
+              },
+              "aws_access_key_id": {
+                "type": "string",
+                "title": "AWS Access Key ID",
+                "description": "Provide your AWS Access Key ID.",
+                "format": "password"
+              },
+              "aws_secret_access_key": {
+                "type": "string",
+                "title": "AWS Secret Access Key",
+                "description": "Provide your AWS Secret Access Key.",
+                "format": "password"
+              }
+            },
+            "required": [
+              "aws_access_key_id",
+              "aws_secret_access_key"
+            ]
+          },
+          {
+            "properties": {
+              "auth_type": {
+                "enum": [
+                  "iam_role"
+                ]
+              }
+            }
+          }
+        ]
       }
     }
   }
diff --git a/unstract/sdk1/src/unstract/sdk1/adapters/llm1/static/bedrock.json b/unstract/sdk1/src/unstract/sdk1/adapters/llm1/static/bedrock.json
index 9adb8b8ceb..03f23d1376 100644
--- a/unstract/sdk1/src/unstract/sdk1/adapters/llm1/static/bedrock.json
+++ b/unstract/sdk1/src/unstract/sdk1/adapters/llm1/static/bedrock.json
@@ -19,18 +19,6 @@
       "default": "amazon.titan-text-express-v1",
       "description": "Model name. Refer to [Bedrock's documentation](https://docs.aws.amazon.com/bedrock/latest/userguide/models-supported.html) for the list of available models."
     },
-    "aws_access_key_id": {
-      "type": "string",
-      "title": "AWS Access Key ID",
-      "description": "Provide your AWS Access Key ID. Leave empty if using AWS Profile or IAM role.",
-      "format": "password"
-    },
-    "aws_secret_access_key": {
-      "type": "string",
-      "title": "AWS Secret Access Key",
-      "description": "Provide your AWS Secret Access Key. Leave empty if using AWS Profile or IAM role.",
-      "format": "password"
-    },
     "region_name": {
       "type": "string",
       "title": "AWS Region Name",
@@ -75,6 +63,60 @@
       "title": "Enable Extended Thinking",
       "default": false,
       "description": "Enhance reasoning for complex tasks with step-by-step transparency. Available only for Claude 3.7 Sonnet."
+    },
+    "auth_type": {
+      "type": "string",
+      "title": "Authentication Type",
+      "enum": [
+        "access_keys",
+        "iam_role"
+      ],
+      "enumNames": [
+        "Access Keys",
+        "IAM Role / Instance Profile (on-prem AWS only)"
+      ],
+      "default": "access_keys",
+      "description": "Choose **Access Keys** for any deployment (provide your AWS Access Key ID and Secret). Choose **IAM Role / Instance Profile** only when Unstract is hosted on AWS infrastructure that already has ambient credentials — for example EKS pods with IRSA, ECS tasks with a task role, or EC2 instances with an instance profile. The on-prem option requires no further input; boto3 picks up the host's identity automatically."
+    }
+  },
+  "dependencies": {
+    "auth_type": {
+      "oneOf": [
+        {
+          "properties": {
+            "auth_type": {
+              "enum": [
+                "access_keys"
+              ]
+            },
+            "aws_access_key_id": {
+              "type": "string",
+              "title": "AWS Access Key ID",
+              "description": "Provide your AWS Access Key ID.",
+              "format": "password"
+            },
+            "aws_secret_access_key": {
+              "type": "string",
+              "title": "AWS Secret Access Key",
+              "description": "Provide your AWS Secret Access Key.",
+              "format": "password"
+            }
+          },
+          "required": [
+            "aws_access_key_id",
+            "aws_secret_access_key"
+          ]
+        },
+        {
+          "properties": {
+            "auth_type": {
+              "enum": [
+                "iam_role"
+              ]
+            }
+          }
+        }
+      ]
     }
   },
   "allOf": [
diff --git a/unstract/sdk1/tests/test_bedrock_adapter.py b/unstract/sdk1/tests/test_bedrock_adapter.py
new file mode 100644
index 0000000000..4d4b7c1cd7
--- /dev/null
+++ b/unstract/sdk1/tests/test_bedrock_adapter.py
@@ -0,0 +1,246 @@
+"""Unit tests for the AWS Bedrock LLM and embedding adapters.
+
+Covers the auth_type selector behaviour added alongside the IAM Role /
+Instance Profile mode, plus backwards compatibility for legacy adapter
+configurations stored before auth_type existed.
+"""
+
+import pytest
+from unstract.sdk1.adapters.base1 import (
+    AWSBedrockEmbeddingParameters,
+    AWSBedrockLLMParameters,
+)
+
+# ── LLM: validate auth_type semantics ────────────────────────────────────────
+
+
+def test_llm_legacy_no_auth_type_keeps_keys() -> None:
+    """Legacy adapters without auth_type must keep working unchanged."""
+    out = AWSBedrockLLMParameters.validate(
+        {
+            "model": "anthropic.claude-3-haiku-20240307-v1:0",
+            "region_name": "us-east-1",
+            "aws_access_key_id": "AKIAFAKE",
+            "aws_secret_access_key": "secret",
+        }
+    )
+    assert out["aws_access_key_id"] == "AKIAFAKE"
+    assert out["aws_secret_access_key"] == "secret"
+    assert out["aws_region_name"] == "us-east-1"
+    assert "auth_type" not in out
+
+
+def test_llm_access_keys_mode_keeps_keys_and_strips_auth_type() -> None:
+    out = AWSBedrockLLMParameters.validate(
+        {
+            "auth_type": "access_keys",
+            "model": "anthropic.claude-3-haiku-20240307-v1:0",
+            "region_name": "us-east-1",
+            "aws_access_key_id": "AKIAFAKE",
+            "aws_secret_access_key": "secret",
+        }
+    )
+    assert out["aws_access_key_id"] == "AKIAFAKE"
+    assert out["aws_secret_access_key"] == "secret"
+    assert "auth_type" not in out
+
+
+def test_llm_iam_role_mode_drops_keys_even_when_present() -> None:
+    """IAM Role mode unconditionally drops access keys.
+
+    A saved adapter switched into IAM mode must not silently leak the
+    previously stored long-lived credentials.
+    """
+    out = AWSBedrockLLMParameters.validate(
+        {
+            "auth_type": "iam_role",
+            "model": "us.anthropic.claude-haiku-4-5-20251001-v1:0",
+            "region_name": "us-east-1",
+            "aws_access_key_id": "STALE_KEY",
+            "aws_secret_access_key": "STALE_SECRET",
+        }
+    )
+    assert "aws_access_key_id" not in out
+    assert "aws_secret_access_key" not in out
+    assert out["aws_region_name"] == "us-east-1"
+    assert "auth_type" not in out
+
+
+def test_llm_iam_role_mode_with_no_keys() -> None:
+    out = AWSBedrockLLMParameters.validate(
+        {
+            "auth_type": "iam_role",
+            "model": "us.anthropic.claude-haiku-4-5-20251001-v1:0",
+            "region_name": "us-east-1",
+        }
+    )
+    assert "aws_access_key_id" not in out
+    assert "aws_secret_access_key" not in out
+
+
+def test_llm_access_keys_mode_blank_keys_raises() -> None:
+    """Blank values must surface a clear error.
+
+    Falling through to boto3's default chain would hide the user's
+    misconfiguration and authenticate with whatever ambient creds the
+    host happens to have.
+    """
+    with pytest.raises(ValueError, match="aws_access_key_id is required"):
+        AWSBedrockLLMParameters.validate(
+            {
+                "auth_type": "access_keys",
+                "model": "anthropic.claude-3-haiku-20240307-v1:0",
+                "region_name": "us-east-1",
+                "aws_access_key_id": "",
+                "aws_secret_access_key": "",
+            }
+        )
+
+
+def test_llm_access_keys_mode_whitespace_keys_raises() -> None:
+    with pytest.raises(ValueError, match="aws_secret_access_key is required"):
+        AWSBedrockLLMParameters.validate(
+            {
+                "auth_type": "access_keys",
+                "model": "anthropic.claude-3-haiku-20240307-v1:0",
+                "region_name": "us-east-1",
+                "aws_access_key_id": "AKIAFAKE",
+                "aws_secret_access_key": "   ",
+            }
+        )
+
+
+def test_llm_unknown_auth_type_raises() -> None:
+    """A typo or non-UI client must not silently fall through."""
+    with pytest.raises(ValueError, match="Unknown auth_type"):
+        AWSBedrockLLMParameters.validate(
+            {
+                "auth_type": "access_key",  # typo: missing 's'
+                "model": "anthropic.claude-3-haiku-20240307-v1:0",
+                "region_name": "us-east-1",
+                "aws_access_key_id": "AKIAFAKE",
+                "aws_secret_access_key": "secret",
+            }
+        )
+
+
+def test_llm_other_params_preserved_through_strip() -> None:
+    """Non-credential params survive the auth-type handling.
+
+    model_id, aws_profile_name, region, and thinking config must pass
+    through both the strip and the resolver unchanged.
+    """
+    out = AWSBedrockLLMParameters.validate(
+        {
+            "auth_type": "iam_role",
+            "model": "anthropic.claude-3-7-sonnet-20250219-v1:0",
+            "region_name": "us-east-1",
+            "aws_profile_name": "dev-profile",
+            "model_id": (
+                "arn:aws:bedrock:us-east-1:1234:application-inference-profile/abc"
+            ),
+            "enable_thinking": True,
+            "budget_tokens": 4096,
+        }
+    )
+    assert out["aws_profile_name"] == "dev-profile"
+    assert out["aws_region_name"] == "us-east-1"
+    assert out["model_id"].endswith("application-inference-profile/abc")
+    assert out["thinking"] == {"type": "enabled", "budget_tokens": 4096}
+
+
+# ── Embedding: same auth_type matrix ─────────────────────────────────────────
+
+
+def test_embedding_legacy_no_auth_type_keeps_keys() -> None:
+    out = AWSBedrockEmbeddingParameters.validate(
+        {
+            "model": "amazon.titan-embed-text-v2:0",
+            "region_name": "us-east-1",
+            "aws_access_key_id": "AKIAFAKE",
+            "aws_secret_access_key": "secret",
+        }
+    )
+    assert out["aws_access_key_id"] == "AKIAFAKE"
+    assert out["aws_secret_access_key"] == "secret"
+    assert "auth_type" not in out
+
+
+def test_embedding_access_keys_mode_keeps_keys() -> None:
+    out = AWSBedrockEmbeddingParameters.validate(
+        {
+            "auth_type": "access_keys",
+            "model": "amazon.titan-embed-text-v2:0",
+            "region_name": "us-east-1",
+            "aws_access_key_id": "AKIAFAKE",
+            "aws_secret_access_key": "secret",
+        }
+    )
+    assert out["aws_access_key_id"] == "AKIAFAKE"
+    assert out["aws_secret_access_key"] == "secret"
+    assert "auth_type" not in out
+
+
+def test_embedding_iam_role_mode_drops_stale_keys() -> None:
+    """Embedding-side parity with the LLM stale-key fix."""
+    out = AWSBedrockEmbeddingParameters.validate(
+        {
+            "auth_type": "iam_role",
+            "model": "amazon.titan-embed-text-v2:0",
+            "region_name": "us-east-1",
+            "aws_access_key_id": "STALE_KEY",
+            "aws_secret_access_key": "STALE_SECRET",
+        }
+    )
+    assert "aws_access_key_id" not in out
+    assert "aws_secret_access_key" not in out
+    assert out["aws_region_name"] == "us-east-1"
+
+
+def test_embedding_iam_role_mode_with_no_keys() -> None:
+    out = AWSBedrockEmbeddingParameters.validate(
+        {
+            "auth_type": "iam_role",
+            "model": "amazon.titan-embed-text-v2:0",
+            "region_name": "us-east-1",
+        }
+    )
+    assert "aws_access_key_id" not in out
+    assert "aws_secret_access_key" not in out
+
+
+def test_embedding_access_keys_mode_blank_keys_raises() -> None:
+    with pytest.raises(ValueError, match="aws_access_key_id is required"):
+        AWSBedrockEmbeddingParameters.validate(
+            {
+                "auth_type": "access_keys",
+                "model": "amazon.titan-embed-text-v2:0",
+                "region_name": "us-east-1",
+                "aws_access_key_id": "",
+                "aws_secret_access_key": "",
+            }
+        )
+
+
+def test_embedding_unknown_auth_type_raises() -> None:
+    with pytest.raises(ValueError, match="Unknown auth_type"):
+        AWSBedrockEmbeddingParameters.validate(
+            {
+                "auth_type": "iamrole",  # typo
+                "model": "amazon.titan-embed-text-v2:0",
+                "region_name": "us-east-1",
+            }
+        )
+
+
+def test_embedding_region_required_when_absent() -> None:
+    """aws_region_name is still mandatory even though credentials are not."""
+    from pydantic import ValidationError
+
+    with pytest.raises(ValidationError):
+        AWSBedrockEmbeddingParameters.validate(
+            {
+                "auth_type": "iam_role",
+                "model": "amazon.titan-embed-text-v2:0",
+            }
+        )
diff --git a/workers/log_consumer/process_notification_buffer.py b/workers/log_consumer/process_notification_buffer.py
new file mode 100755
index 0000000000..7f93a8dfe6
--- /dev/null
+++ b/workers/log_consumer/process_notification_buffer.py
@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Trigger backend processing of the notification buffer.
+
+Mirrors process_log_history.py: a thin wrapper around an internal API call
+that the log_consumer scheduler.sh fires on every tick. The backend owns the
+actual GROUP BY / SKIP LOCKED / dispatch logic. Idempotent — safe to run
+under multiple replicas (the backend's row-level lock prevents duplicate
+dispatch).
+
+Usage:
+    python process_notification_buffer.py
+"""
+
+import logging
+import os
+import sys
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Endpoint must match the URL registered in
+# backend/notification_v2/internal_urls.py + backend/backend/internal_base_urls.py
+PROCESS_BUFFER_ENDPOINT = "v1/webhook/buffer/process/"
+
+
+def process_notification_buffer() -> bool:
+    """Hit the backend's process-buffer endpoint; return True on success.
+
+    Returns False on auth/network failure so the calling scheduler can log
+    the failure and proceed to the next tick. Never raises — the scheduler
+    is supposed to keep ticking.
+    """
+    internal_api_base_url = os.getenv("INTERNAL_API_BASE_URL")
+    internal_api_key = os.getenv("INTERNAL_SERVICE_API_KEY")
+
+    if not internal_api_base_url:
+        logger.error("INTERNAL_API_BASE_URL environment variable not set")
+        return False
+    if not internal_api_key:
+        logger.error("INTERNAL_SERVICE_API_KEY environment variable not set")
+        return False
+
+    url = f"{internal_api_base_url.rstrip('/')}/{PROCESS_BUFFER_ENDPOINT}"
+    # Longer timeout than process_log_history (60s vs 30s): a flush tick can
+    # involve multiple Celery dispatches, GC, and per-group rendering.
+    transport = httpx.HTTPTransport(retries=3)
+    try:
+        with httpx.Client(transport=transport) as client:
+            response = client.post(
+                url,
+                headers={"Authorization": f"Bearer {internal_api_key}"},
+                timeout=60.0,
+            )
+    except httpx.HTTPError as e:
+        logger.error("HTTP error calling process-buffer: %s", e)
+        return False
+    except Exception:
+        logger.exception("Unexpected error calling process-buffer")
+        return False
+
+    if response.status_code != 200:
+        logger.error(
+            "Backend returned status %s on process-buffer: %s",
+            response.status_code,
+            response.text[:500],
+        )
+        return False
+
+    result = response.json()
+    if result.get("dispatched_groups", 0) > 0 or result.get("gc_deleted_rows", 0) > 0:
+        logger.info(
+            "Notification buffer flush: groups=%s rows=%s gc=%s",
+            result.get("dispatched_groups", 0),
+            result.get("dispatched_rows", 0),
+            result.get("gc_deleted_rows", 0),
+        )
+    return True
+
+
+if __name__ == "__main__":
+    success = process_notification_buffer()
+    sys.exit(0 if success else 1)
diff --git a/workers/log_consumer/scheduler.sh b/workers/log_consumer/scheduler.sh
index a5572b21eb..0bf7dca96d 100755
--- a/workers/log_consumer/scheduler.sh
+++ b/workers/log_consumer/scheduler.sh
@@ -1,19 +1,26 @@
 #!/usr/bin/env bash
-set -euo pipefail
+set -uo pipefail
+# Note: pipefail without -e — one task's failure must not abort the loop.
 
 INTERVAL="${LOG_HISTORY_CONSUMER_INTERVAL:-5}"
-TASK_NAME="process_log_history"
 
-# Task trigger command - can be overridden via environment variable
-DEFAULT_TRIGGER_CMD="/app/.venv/bin/python /app/log_consumer/process_log_history.py"
-TRIGGER_CMD="${TASK_TRIGGER_COMMAND:-$DEFAULT_TRIGGER_CMD}"
+# Task 1: log history consumer (existing).
+DEFAULT_LOG_HISTORY_CMD="/app/.venv/bin/python /app/log_consumer/process_log_history.py"
+LOG_HISTORY_CMD="${TASK_TRIGGER_COMMAND:-$DEFAULT_LOG_HISTORY_CMD}"
+
+# Task 2: notification buffer flush (UNS-611 clubbed dispatch).
+# The endpoint short-circuits on an empty PENDING set, so polling on the same
+# 5s tick is cheap. Real dispatch cadence is gated by NOTIFICATION_CLUB_INTERVAL
+# on the backend (rows precompute flush_after at enqueue time).
+DEFAULT_BUFFER_FLUSH_CMD="/app/.venv/bin/python /app/log_consumer/process_notification_buffer.py"
+BUFFER_FLUSH_CMD="${NOTIFICATION_BUFFER_TASK_COMMAND:-$DEFAULT_BUFFER_FLUSH_CMD}"
 
 echo "=========================================="
-echo "Log History Scheduler Starting"
+echo "Log Consumer Scheduler Starting"
 echo "=========================================="
-echo "Task: ${TASK_NAME}"
 echo "Interval: ${INTERVAL} seconds"
-echo "Trigger Command: ${TRIGGER_CMD}"
+echo "Task 1 (log history): ${LOG_HISTORY_CMD}"
+echo "Task 2 (notification buffer flush): ${BUFFER_FLUSH_CMD}"
 echo "=========================================="
 
 cleanup() {
@@ -27,20 +34,27 @@ cleanup() {
 
 trap cleanup SIGTERM SIGINT
 
+run_task() {
+    # $1 = display name, $2 = command. Returns the command's exit code but
+    # never propagates failure — the caller logs it and moves on.
+    local task_name="$1"
+    local cmd="$2"
+    echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] Triggering ${task_name}..."
+    if eval "${cmd}" 2>&1; then
+        echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] ✓ ${task_name} OK"
+    else
+        local exit_code=$?
+        echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] ✗ ${task_name} failed with exit code ${exit_code}"
+    fi
+}
+
 run_count=0
 
 while true; do
     run_count=$((run_count + 1))
 
-    echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] Triggering ${TASK_NAME}..."
-
-    if eval "${TRIGGER_CMD}" 2>&1; then
-        echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] ✓ Task completed successfully"
-    else
-        exit_code=$?
-        echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] ✗ Task failed with exit code ${exit_code}"
-        echo "[$(date '+%Y-%m-%d %H:%M:%S')] [Run #${run_count}] Will retry after ${INTERVAL} seconds"
-    fi
+    run_task "process_log_history" "${LOG_HISTORY_CMD}"
+    run_task "process_notification_buffer" "${BUFFER_FLUSH_CMD}"
 
     echo "[$(date '+%Y-%m-%d %H:%M:%S')] Sleeping for ${INTERVAL} seconds..."
     echo ""
diff --git a/workers/shared/patterns/notification/helper.py b/workers/shared/patterns/notification/helper.py
index 19f78fbb99..f46385bf0b 100644
--- a/workers/shared/patterns/notification/helper.py
+++ b/workers/shared/patterns/notification/helper.py
@@ -19,6 +19,90 @@
 
 logger = logging.getLogger(__name__)
 
+# Mirrors notification_v2.enums.DeliveryMode.BATCHED. Worker stays string-only
+# so it does not import Django enums.
+DELIVERY_MODE_BATCHED = "BATCHED"
+ENQUEUE_BUFFER_ENDPOINT = "v1/webhook/buffer/enqueue/"
+
+
+def _enqueue_to_buffer(
+    api_client: Any,
+    notification: dict[str, Any],
+    payload: NotificationPayload,
+) -> bool:
+    """POST a single execution event to the backend's buffer endpoint.
+
+    Worker writes nothing to the DB itself — the backend owns NotificationBuffer
+    rows. Returns True on success so callers can fall back to immediate dispatch
+    if the buffer endpoint is unavailable. The fallback decision is opinionated:
+    we keep behavior conservative and DON'T fall back so a misconfigured or
+    outage-mode backend can't silently turn BATCHED into IMMEDIATE.
+    """
+    try:
+        api_client._make_request(
+            method="POST",
+            endpoint=ENQUEUE_BUFFER_ENDPOINT,
+            data={
+                "notification_id": notification["id"],
+                "execution_id": payload.execution_id,
+                "pipeline_id": payload.pipeline_id,
+                "pipeline_name": payload.pipeline_name,
+                "status": payload.status.value
+                if hasattr(payload.status, "value")
+                else payload.status,
+                "error_message": payload.error_message,
+                "platform": notification.get("platform"),
+            },
+            timeout=10,
+        )
+        logger.info(
+            "Enqueued BATCHED notification %s for pipeline %s execution %s",
+            notification["id"],
+            payload.pipeline_id,
+            payload.execution_id,
+        )
+        return True
+    except Exception as e:
+        logger.error(
+            "Failed to enqueue BATCHED notification %s for pipeline %s: %s",
+            notification["id"],
+            payload.pipeline_id,
+            e,
+        )
+        return False
+
+
+def _route_notification(
+    api_client: Any,
+    notification: dict[str, Any],
+    payload: NotificationPayload,
+) -> None:
+    """IMMEDIATE -> existing worker queue; BATCHED -> backend enqueue endpoint.
+
+    Defaults to IMMEDIATE when delivery_mode is missing so older backend
+    builds (pre-UNS-611) keep working unchanged.
+    """
+    if notification.get("notification_type") != "WEBHOOK":
+        logger.debug(
+            "Skipping non-webhook notification type: %s",
+            notification.get("notification_type"),
+        )
+        return
+
+    if notification.get("delivery_mode") == DELIVERY_MODE_BATCHED:
+        _enqueue_to_buffer(api_client, notification, payload)
+        return
+
+    send_notification_to_worker(
+        url=notification["url"],
+        payload=payload,
+        auth_type=notification.get("authorization_type", "NONE"),
+        auth_key=notification.get("authorization_key"),
+        auth_header=notification.get("authorization_header"),
+        max_retries=notification.get("max_retries", 0),
+        platform=notification.get("platform"),
+    )
+
 
 def get_webhook_headers(
     auth_type: str, auth_key: str | None, auth_header: str | None
@@ -140,20 +224,7 @@ def trigger_notification(
 
         # Send each notification
         for notification in active_notifications:
-            if notification.get("notification_type") == "WEBHOOK":
-                send_notification_to_worker(
-                    url=notification["url"],
-                    payload=notification_payload,
-                    auth_type=notification.get("authorization_type", "NONE"),
-                    auth_key=notification.get("authorization_key"),
-                    auth_header=notification.get("authorization_header"),
-                    max_retries=notification.get("max_retries", 0),
-                    platform=notification.get("platform"),
-                )
-            else:
-                logger.debug(
-                    f"Skipping non-webhook notification type: {notification.get('notification_type')}"
-                )
+            _route_notification(api_client, notification, notification_payload)
 
     except Exception as e:
         logger.error(f"Error triggering pipeline notifications for {pipeline_id}: {e}")
@@ -235,20 +306,7 @@ def trigger_pipeline_notifications(
 
         # Send each notification
         for notification in active_notifications:
-            if notification.get("notification_type") == "WEBHOOK":
-                send_notification_to_worker(
-                    url=notification["url"],
-                    payload=payload,
-                    auth_type=notification.get("authorization_type", "NONE"),
-                    auth_key=notification.get("authorization_key"),
-                    auth_header=notification.get("authorization_header"),
-                    max_retries=notification.get("max_retries", 0),
-                    platform=notification.get("platform"),
-                )
-            else:
-                logger.debug(
-                    f"Skipping non-webhook notification type: {notification.get('notification_type')}"
-                )
+            _route_notification(api_client, notification, payload)
 
     except Exception as e:
         logger.error(f"Error triggering pipeline notifications for {pipeline_id}: {e}")
@@ -316,20 +374,7 @@ def trigger_api_notifications(
 
         # Send each notification
         for notification in active_notifications:
-            if notification.get("notification_type") == "WEBHOOK":
-                send_notification_to_worker(
-                    url=notification["url"],
-                    payload=payload,
-                    auth_type=notification.get("authorization_type", "NONE"),
-                    auth_key=notification.get("authorization_key"),
-                    auth_header=notification.get("authorization_header"),
-                    max_retries=notification.get("max_retries", 0),
-                    platform=notification.get("platform"),
-                )
-            else:
-                logger.debug(
-                    f"Skipping non-webhook notification type: {notification.get('notification_type')}"
-                )
+            _route_notification(api_client, notification, payload)
 
     except Exception as e:
         logger.error(f"Error triggering API notifications for {api_id}: {e}")
