UN-3056 [FEAT] Failure-only notifications with batched dispatch

[kirtimanmishrazipstack]

What

• New "Notify on failures only" option — subscribers stop getting webhooks for successful runs.
• Notifications are now grouped into one webhook per time window instead of one per run.
• New Notifications panel in Platform Settings lets each org set the grouping window (1–120 minutes, default 5).

Why

• Customers asked for failure-only alerts; previously every successful run also fired a webhook.
• High-volume deployments drowned in per-run webhooks; grouping collapses them into one message per window.
• Ops can tune alert cadence per org from the UI without code changes or redeploys.

How

• Three dispatch sites (api_v2, pipeline_v2, worker callback) enqueue into NotificationBuffer; workers/log_consumer/scheduler.sh periodically calls the internal flush endpoint, which dispatches with SELECT … FOR UPDATE OF o SKIP LOCKED.
• Rendering goes through unstract.core.notification_clubbed_renderer — canonical {summary, events} envelope for API webhooks, Slack mrkdwn for platform=SLACK.
• Frontend: redesigned Platform Settings page, added notify_on_failures checkbox in the notification modal, removed pagination from Setup Notifications.

Can this PR break any existing features. If yes, please list possible items. If no, please explain why.

• Webhook payload shape changed — receivers that read the old flat per-run body must now read events[0].
• Notifications no longer fire instantly; they arrive once per grouping window (default 5 minutes).

Database Migrations

• backend/notification_v2/migrations/0002_notification_notify_on_failures.py — adds failure-only flag.
• backend/notification_v2/migrations/0003_add_notification_buffer.py — adds buffer table for grouped dispatch.
• backend/workflow_manager/workflow_v2/migrations/0020_workflowexecution_file_counts.py — adds file-count fields used in payload.

Env Config

Relevant Docs

Related Issues or PRs

Dependencies Versions

• None

Notes on Testing

• Tested on local using slack webhook + api webhook
• Confirm the grouped webhook arrives within the configured window (default 5 min).

Screenshots

• UI screen changes

• Slack Webhook Notification

• API Webhook Notification

Checklist

I have read and understood the Contribution Guidelines.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

Adds org-configurable batched webhook buffering with enqueue/process endpoints, clubbed envelope rendering, worker routing to enqueue events, file-count propagation for failure-only routing, DB buffer model retention/dispatch lifecycle, and frontend settings/UI for interval and failure-only toggles.

Changes

Buffered notification feature

Layer / File(s)	Summary
Workflow execution file-counts backend/workflow_manager/workflow_v2/models/execution.py	Adds successful_files and failed_files fields to WorkflowExecution and uses them in run-status/timeline logic.
Internal API: buffer endpoints & processing backend/notification_v2/internal_api_views.py	Adds enqueue_notification_buffer and process_notification_buffer; implements enqueue validation (including notify_on_failures suppression), grouping/claiming, clubbed rendering, marking DISPATCHED, scheduling Celery sends, and GC of terminal/aged rows.
Worker routing & enqueue client workers/shared/patterns/notification/helper.py	Routes WEBHOOK events to backend enqueue endpoint (v1/webhook/buffer/enqueue/), passes execution_id and file-counts when listing notifications, and adds _enqueue_to_buffer/_route_notification and ENQUEUE_BUFFER_ENDPOINT.
Frontend settings & UI frontend/src/components/settings/platform/PlatformSettings.jsx, frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx, frontend/src/components/settings/platform/PlatformSettings.css, frontend/src/components/pipelines-or-deployments/notification-modal/DisplayNotifications.jsx	Adds org notification-interval UI (minutes ↔ seconds) with GET/PATCH, notify_on_failures checkbox in create modal, disables table pagination, and updates styling.
Supporting: clubbed renderer & payloads unstract/core/src/unstract/core/notification_clubbed_renderer.py, backend/notification_v2/clubbed_renderer.py, backend/pipeline_v2/dto.py, unstract/core/src/unstract/core/data_models.py	Introduces canonical clubbed envelope and Slack renderer, extends pipeline payloads and notification payloads to include total/successful/failed file counts.

sequenceDiagram
  participant Producer as Notifier Source
  participant Backend as Backend API
  participant BufferAPI as Buffer Enqueue API
  participant DB as NotificationBuffer
  participant Scheduler as Scheduler/LogConsumer
  participant Celery as Celery Worker
  participant Webhook as External Webhook

  Producer->>Backend: trigger notification (includes execution_id)
  Backend->>Backend: load execution -> compute file counts
  Backend->>Backend: filter notifications (notify_on_failures)
  Backend->>BufferAPI: POST enqueue_notification_buffer (payload + counts)
  BufferAPI->>DB: persist NotificationBuffer (PENDING, flush_after)
  Scheduler->>DB: process_notification_buffer (groups due flushes)
  DB->>DB: claim rows FOR UPDATE SKIP LOCKED
  DB->>DB: build_envelope() / render_clubbed_message()
  DB->>DB: mark rows DISPATCHED
  DB->>Celery: enqueue single clubbed send task (buffer ids)
  Celery->>Webhook: POST clubbed payload
  Webhook-->>Celery: 200 OK / error
  Celery->>DB: on success keep DISPATCHED / on failure mark DEAD_LETTER or revert to PENDING

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 72.22% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the main feature: failure-only notifications with batched dispatch instead of per-run webhooks.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description check	✅ Passed	The PR description comprehensively covers all required template sections with detailed information about changes, rationale, implementation approach, and breaking changes.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch UN-3056-Notify-on-API-deployment-failures

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 12

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

backend/workflow_manager/internal_serializers.py (1)

176-184: ⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Validate the file aggregates against total_files.

These fields are validated independently right now, so impossible payloads like total_files=1, successful_files=2, failed_files=2 will pass and then be persisted. That will skew the new outcome-based notification logic.

Suggested fix

 class WorkflowExecutionStatusUpdateSerializer(serializers.Serializer):
     """Serializer for updating workflow execution status."""
@@
     failed_files = serializers.IntegerField(required=False, min_value=0)
     attempts = serializers.IntegerField(required=False, min_value=0)
     execution_time = serializers.FloatField(required=False, min_value=0)
+
+    def validate(self, attrs):
+        total_files = attrs.get("total_files")
+        successful_files = attrs.get("successful_files")
+        failed_files = attrs.get("failed_files")
+
+        if (successful_files is not None or failed_files is not None) and total_files is None:
+            raise serializers.ValidationError(
+                {"total_files": "total_files is required when file aggregates are provided."}
+            )
+
+        if total_files is not None:
+            if successful_files is not None and successful_files > total_files:
+                raise serializers.ValidationError(
+                    {"successful_files": "successful_files cannot exceed total_files."}
+                )
+            if failed_files is not None and failed_files > total_files:
+                raise serializers.ValidationError(
+                    {"failed_files": "failed_files cannot exceed total_files."}
+                )
+            if (
+                successful_files is not None
+                and failed_files is not None
+                and successful_files + failed_files > total_files
+            ):
+                raise serializers.ValidationError(
+                    "successful_files + failed_files cannot exceed total_files."
+                )
+
+        return attrs

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/workflow_manager/internal_serializers.py` around lines 176 - 184, The
serializer currently validates total_files, successful_files and failed_files
independently; add a validate(self, data) method in the same serializer (where
status, error_message, total_files, successful_files, failed_files, attempts,
execution_time are defined) that, when total_files is provided, enforces that
successful_files and failed_files are each <= total_files (if present) and that
(successful_files + failed_files) <= total_files; also handle the case where
only one of successful_files/failed_files is present by ensuring it does not
exceed total_files, and raise serializers.ValidationError with a clear message
on violation so invalid aggregates like total_files=1, successful_files=2 are
rejected before persisting.

🧹 Nitpick comments (2)

backend/notification_v2/views.py (1)

56-68: ⚡ Quick win

Use tuple for permission_classes class attribute.

Class attributes that are collections should be immutable (tuples) rather than mutable (lists) to avoid potential issues and follow best practices.

♻️ Proposed fix

-    permission_classes = [IsAuthenticated, IsOrganizationAdmin]
+    permission_classes = (IsAuthenticated, IsOrganizationAdmin)

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/notification_v2/views.py` around lines 56 - 68, The class attribute
permission_classes on NotificationSettingsView is currently a list; change it to
an immutable tuple to follow best practices by replacing the mutable list
[IsAuthenticated, IsOrganizationAdmin] with a tuple (IsAuthenticated,
IsOrganizationAdmin) so permission_classes is not modifiable at runtime and
matches other DRF class-attribute patterns.

backend/notification_v2/tasks.py (1)

46-50: ⚡ Quick win

Combine the implicitly concatenated strings.

The two string literals on line 47 are implicitly concatenated. While valid Python, this can be error-prone and less readable.

♻️ Proposed fix

     logger.warning(
-        "metric=notification_batch_dispatched_total result=dead_letter rows=%d " "exc=%r",
+        "metric=notification_batch_dispatched_total result=dead_letter rows=%d exc=%r",
         updated,
         exc,
     )

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@backend/notification_v2/tasks.py` around lines 46 - 50, The logger.warning
call currently uses two implicitly concatenated string literals; replace them
with a single combined format string in the logger.warning invocation so the
message is explicit and readable (keep the format placeholders and the same
arguments: updated and exc), e.g., a single string like
"metric=notification_batch_dispatched_total result=dead_letter rows=%d exc=%r"
passed to logger.warning with updated and exc.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@backend/notification_v2/clubbed_renderer.py`:
- Line 3: Update the docstring in clubbed_renderer.py to replace the ambiguous
multiplication symbol "×" with a plain ASCII "x" so it satisfies Ruff rule
RUF002; locate the module-level or function/class docstring that contains the
sentence "The same envelope shape feeds every channel × mode cell so receivers
never" and change "×" to "x" (i.e., "channel x mode") to avoid the lint failure.

In `@backend/notification_v2/helper.py`:
- Around line 34-41: The current auth_sig is built by joining fields with "|"
which is ambiguous because authorization_key/header may contain "|" — change the
construction in helper.py where the hash is computed (the block that builds raw
from notification.authorization_type/authorization_key/authorization_header and
returns hashlib.sha256(...).hexdigest()) to encode the three parts as an
unambiguous structured value before hashing (e.g., create a fixed-order
list/tuple of the three parts with fallback to _AUTH_SIG_NONE and then
JSON-serialize it with stable separators, or use a length-prefixed
concatenation) and hash that encoded representation so different tuples can
never collide due to delimiter characters.

In `@backend/notification_v2/internal_api_views.py`:
- Around line 423-425: The batching flush currently groups rows by
(organization_id, webhook_url, auth_sig) but then uses rows[0].platform when
calling render_clubbed_message in _dispatch_group, causing mixed-platform
batches; update the flush query that persists/groups batches to include platform
in its grouping key (and add the corresponding DB index change) so grouping is
done by (organization_id, webhook_url, auth_sig, platform), and ensure
_dispatch_group (and any other consumer like the code around lines ~496-500)
reads platform from the grouped key rather than assuming rows[0].platform.

In `@backend/notification_v2/migrations/0002_notification_notify_on_failures.py`:
- Around line 10-21: The add-field migration currently creates a BooleanField
notify_on_failures which only distinguishes ALL vs FAILURES_ONLY; change this to
a tri-state field (e.g. models.CharField with choices or a small IntegerField)
on the Notification model in this migration so it can represent ALL,
FAILURES_ONLY, and SUCCESS_ONLY (use explicit choices like ("all","ALL"),
("failures","FAILURES_ONLY"), ("success","SUCCESS_ONLY")), set a sensible
default (e.g. "all"), and update the db_comment to document the three modes;
ensure the migration operation uses the new field type and name
notify_on_failures so downstream code can read the string/enum value rather than
a boolean.

In `@backend/notification_v2/migrations/0003_add_notification_buffer.py`:
- Around line 14-28: The migration adds a non-null CharField delivery_mode to
the notification model with default="BATCHED", which will change existing rows
to BATCHED on deploy; instead, modify the migration to preserve existing
behavior by performing a two-step change: 1) add delivery_mode as nullable (or
without a DB-level default) and include a RunPython data migration that sets
delivery_mode="IMMEDIATE" for existing Notification rows that should remain
immediate, and 2) then add a subsequent migration to set default="BATCHED" and
make the field non-nullable for future records; reference the migration
module/migration class in 0003_add_notification_buffer.py and the model name
"notification" and field name "delivery_mode" when implementing the nullable
field + RunPython backfill, or alternatively write a single migration that uses
RunPython before altering the field to set existing rows to "IMMEDIATE".

In `@backend/notification_v2/models.py`:
- Around line 58-65: The boolean field notify_on_failures on the model cannot
represent the three states required (ALL / FAILURES_ONLY / SUCCESS_ONLY); change
this to a tri-state enum field (e.g., a CharField or IntegerField with explicit
choices like NOTIFY_ALL, NOTIFY_FAILURES_ONLY, NOTIFY_SUCCESS_ONLY) and rename
the DB column/field to something clearer if helpful (e.g., notify_mode or
notify_condition) so intent is explicit; update the corresponding serializer(s)
and any filters/UI code that read/write notify_on_failures to accept and
validate the new enum values and migrate existing boolean data to the new enum
values in a migration.

In `@backend/notification_v2/provider/webhook/api_webhook.py`:
- Line 15: format_payload() is unconditionally wrapping self.payload which
causes double-enveloping on already-enveloped payloads or repeated send() calls;
change format_payload (and any callers like the constructor assignment
self.payload = self.format_payload() and the send() path at the 24-30 block) to
first detect whether the payload is already in the expected envelope shape
(e.g., check for the envelope root key/structure) and return it unchanged if so,
otherwise wrap it; ensure the envelope-detection logic is deterministic and
idempotent so multiple calls to format_payload/send() do not alter an
already-correctly-enveloped payload.

In `@backend/pipeline_v2/notification.py`:
- Around line 14-15: Remove ExecutionStatus.STOPPED from the set treated as
failures: update the _FAILURE_STATUSES definition to exclude
ExecutionStatus.STOPPED and similarly remove/adjust any other checks that
include ExecutionStatus.STOPPED (the second occurrence around the block handling
audience selection at lines 56-60) so that STOPPED executions are no longer
routed to failure-only subscriptions; ensure only true failure statuses (e.g.,
ExecutionStatus.ERROR) remain in _FAILURE_STATUSES and that any conditional
logic using that set (in notification audience selection) treats STOPPED as
non-failure/catch-all.

In `@backend/workflow_manager/workflow_v2/models/execution.py`:
- Around line 440-441: The current mapping sets successful = e.successful_files
or 0 and failed = e.failed_files or 0 which coerces NULL/None to 0 and can hide
unknown historical counts; change these assignments to preserve NULL/None (e.g.,
successful = e.successful_files if e.successful_files is not None else None, and
likewise for failed) and update any downstream status logic that treats 0 as "no
failures" to explicitly handle None as "unknown" (so PARTIAL_SUCCESS isn't
lost). Ensure references to successful and failed in the execution status
computation explicitly check for None versus integer values.

In
`@frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx`:
- Line 15: The component currently sends a boolean notify_on_failures which
can't represent SUCCESS_ONLY—replace this with a notify_on enum carrying one of
"ALL", "FAILURES_ONLY", or "SUCCESS_ONLY": update the component state/props
default (in CreateNotification.jsx) to hold notify_on instead of
notify_on_failures, map the form inputs (checkboxes/radio/select) to produce the
correct enum value, and change all payload constructions and update/create API
calls (the places around the existing notify_on_failures usage and the other
block referenced later in the file) to include notify_on with the proper enum
string; also adjust any validation/serialization logic that reads
notify_on_failures to use notify_on.

In `@frontend/src/components/settings/platform/PlatformSettings.jsx`:
- Around line 61-77: The effect in useEffect that fetches org-scoped batch
interval should guard on sessionDetails?.orgId and include proper deps and
cancellation: at the top of the effect return early if !sessionDetails?.orgId,
add sessionDetails?.orgId and axiosPrivate to the dependency array, and
implement request cancellation (e.g., AbortController or axios cancel token) so
in-flight responses don't call setBatchIntervalMinutes after unmount or when
orgId changes; update references to axiosPrivate and setBatchIntervalMinutes
accordingly.

In `@workers/shared/patterns/notification/helper.py`:
- Around line 77-84: The except block in _enqueue_to_buffer() is swallowing
enqueue failures (logging then returning False) which causes
_route_notification() to treat BATCHED delivery as successful; instead,
propagate the failure so a retrying caller can act (or implement local
retry/backoff). Replace the logger.error+return False with logger.exception(...)
to include stack context and then re-raise the exception (or raise a specific
EnqueueError) so _route_notification() sees the failure; make the same change
for the other similar block referenced (lines ~104-106) to avoid silent drops.

---

Outside diff comments:
In `@backend/workflow_manager/internal_serializers.py`:
- Around line 176-184: The serializer currently validates total_files,
successful_files and failed_files independently; add a validate(self, data)
method in the same serializer (where status, error_message, total_files,
successful_files, failed_files, attempts, execution_time are defined) that, when
total_files is provided, enforces that successful_files and failed_files are
each <= total_files (if present) and that (successful_files + failed_files) <=
total_files; also handle the case where only one of
successful_files/failed_files is present by ensuring it does not exceed
total_files, and raise serializers.ValidationError with a clear message on
violation so invalid aggregates like total_files=1, successful_files=2 are
rejected before persisting.

---

Nitpick comments:
In `@backend/notification_v2/tasks.py`:
- Around line 46-50: The logger.warning call currently uses two implicitly
concatenated string literals; replace them with a single combined format string
in the logger.warning invocation so the message is explicit and readable (keep
the format placeholders and the same arguments: updated and exc), e.g., a single
string like "metric=notification_batch_dispatched_total result=dead_letter
rows=%d exc=%r" passed to logger.warning with updated and exc.

In `@backend/notification_v2/views.py`:
- Around line 56-68: The class attribute permission_classes on
NotificationSettingsView is currently a list; change it to an immutable tuple to
follow best practices by replacing the mutable list [IsAuthenticated,
IsOrganizationAdmin] with a tuple (IsAuthenticated, IsOrganizationAdmin) so
permission_classes is not modifiable at runtime and matches other DRF
class-attribute patterns.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: d8dfdd8e-ab77-48bb-a35e-1d2d9ef6fd9a

📥 Commits

Reviewing files that changed from the base of the PR and between 9678d78 and 33d77e9.

📒 Files selected for processing (38)

• backend/api_v2/notification.py
• backend/backend/settings/base.py
• backend/configuration/enums.py
• backend/notification_v2/clubbed_renderer.py
• backend/notification_v2/enums.py
• backend/notification_v2/helper.py
• backend/notification_v2/internal_api_views.py
• backend/notification_v2/internal_serializers.py
• backend/notification_v2/internal_urls.py
• backend/notification_v2/migrations/0002_notification_notify_on_failures.py
• backend/notification_v2/migrations/0003_add_notification_buffer.py
• backend/notification_v2/models.py
• backend/notification_v2/provider/webhook/api_webhook.py
• backend/notification_v2/provider/webhook/slack_webhook.py
• backend/notification_v2/provider/webhook/webhook.py
• backend/notification_v2/serializers.py
• backend/notification_v2/tasks.py
• backend/notification_v2/urls.py
• backend/notification_v2/views.py
• backend/pipeline_v2/dto.py
• backend/pipeline_v2/notification.py
• backend/workflow_manager/internal_serializers.py
• backend/workflow_manager/internal_views.py
• backend/workflow_manager/workflow_v2/migrations/0020_workflowexecution_file_counts.py
• backend/workflow_manager/workflow_v2/models/execution.py
• frontend/src/components/pipelines-or-deployments/notification-modal/CreateNotification.jsx
• frontend/src/components/settings/platform/PlatformSettings.jsx
• unstract/core/src/unstract/core/data_models.py
• workers/callback/tasks.py
• workers/log_consumer/process_notification_buffer.py
• workers/log_consumer/scheduler.sh
• workers/notification/providers/_clubbed_format.py
• workers/notification/providers/api_webhook.py
• workers/notification/providers/slack_webhook.py
• workers/scheduler/tasks.py
• workers/shared/api/internal_client.py
• workers/shared/clients/execution_client.py
• workers/shared/patterns/notification/helper.py

[greptile-apps]

Greptile Summary

This PR introduces failure-only webhook subscriptions (notify_on_failures) and a batched dispatch pipeline: execution events are buffered in NotificationBuffer, grouped by (org, webhook_url, auth_sig, platform), and flushed in a configurable window (default 5 min) rather than fired per-run. A new Platform Settings panel lets org admins set the grouping window (1–120 min).

• Backend: enqueue_notification_buffer writes buffer rows; process_notification_buffer drives a SELECT … FOR UPDATE SKIP LOCKED flush, SENDING-reaper, and two-sweep GC. Duplicate-dispatch is prevented by the on_commit publish order and the notification__is_active filter on the lock query.
• Worker / shared core: NotificationPayload now carries per-run file counts (total_files, successful_files, failed_files); workers/notification providers render via the shared notification_clubbed_renderer; scheduler.sh runs the buffer-flush script on its own independent cadence.
• Frontend: redesigned Platform Settings page with an "Internal API Keys" section and a new "Notifications" section; CreateNotification modal adds a notify_on_failures checkbox.

Confidence Score: 4/5

Safe to merge with awareness of one transient notification gap for in-flight executions at deploy time.

The flush engine, duplicate-dispatch prevention, is_active filter, and retry-budget refund all look correct. The one concern worth tracking: pre-migration WorkflowExecution rows have null failed_files, so is_failure_run silently returns False for COMPLETED runs with partial file failures until the worker has written updated counts. Runs that were in-flight at deploy time may not trigger failure-only notifications for that window.

backend/workflow_manager/workflow_v2/models/execution.py — the null coalescing of successful_files/failed_files affects is_failure_run for pre-migration rows.

Important Files Changed

Filename	Overview
backend/notification_v2/internal_api_views.py	Core flush engine: GROUP BY → SKIP LOCKED dispatch, SENDING reaper, GC. The is_active guard, max() retry budget, atomic PENDING→SENDING claim, and on_commit publish order all look correct.
backend/notification_v2/helper.py	Buffer enqueue path: computes auth_sig, flush_after, stamps timestamp. Raises ValueError for missing org (correct). Broad except in dispatch_notifications logs and continues — acceptable for failure-alert path.
backend/notification_v2/models.py	Adds notify_on_failures BooleanField to Notification and new NotificationBuffer model with composite partial index on PENDING rows. Cascade delete on notification FK is intentional and documented.
backend/notification_v2/migrations/0002_notification_batching.py	Combined migration adds notify_on_failures (default False, safe for existing rows) and creates NotificationBuffer with partial covering index.
backend/notification_v2/tasks.py	Celery callbacks mark_buffer_dispatched / mark_buffer_dead_letter only flip SENDING rows, preventing stale callbacks from clobbering reaper-reclaimed rows.
unstract/core/src/unstract/core/notification_clubbed_renderer.py	Canonical envelope + Slack renderer. _humanize_timestamp uses dt.day instead of %-d (portability fix). LEGACY_FLAT_KEYS backward-compat shim for single-event batches is well-documented.
workers/shared/patterns/notification/helper.py	Replaced direct Celery send_task with backend enqueue HTTP call. execution_id is now forwarded for failure-filter correctness.
workers/log_consumer/scheduler.sh	Scheduler now runs two independent tasks at their own cadences. Dropped -e from set flags intentionally so one task failure doesn't abort the loop.
backend/workflow_manager/workflow_v2/models/execution.py	Adds successful_files / failed_files nullable fields. list_executions now reads denormalized counts instead of N+1 queries. Pre-migration rows return 0/0 (null coalesced) — documented trade-off with notification filter implications.
backend/backend/settings/base.py	Adds four new notification settings constants. NOTIFICATION_BUFFER_RETENTION_DAYS comment is stale after the inactive-PENDING GC sweep was added.
frontend/src/components/settings/platform/PlatformSettings.jsx	Adds Notifications section with org-scoped batch interval control. Session guard on useEffect. InputNumber bounds (1–120 min) match serializer constraints.
workers/notification/tasks.py	Adds raise_on_final_failure flag so the clubbed dispatch's link_error callback fires on retry exhaustion, enabling dead-letter rows.
backend/notification_v2/views.py	New NotificationSettingsView GET/PATCH for org-scoped club_interval_seconds. Protected by IsOrganizationAdmin. Configuration.get_value_by_organization always returns typed int — int() cast is safe.

Sequence Diagram

sequenceDiagram
    participant W as Worker (callback/scheduler)
    participant B as Backend (enqueue endpoint)
    participant DB as NotificationBuffer (DB)
    participant S as scheduler.sh
    participant F as Flush endpoint
    participant C as Celery (send_webhook_notification)
    participant WH as Webhook receiver

    W->>B: POST /v1/webhook/buffer/enqueue/
    B->>DB: "INSERT NotificationBuffer status=PENDING"
    B-->>W: 201 buffer_row_id

    S->>F: POST /v1/webhook/buffer/process/
    F->>DB: "GROUP BY org+url+auth_sig+platform WHERE MIN(flush_after)<=now"
    loop For each due group
        F->>DB: "SELECT FOR UPDATE SKIP LOCKED is_active=True"
        F->>DB: "UPDATE status=SENDING dispatch_attempts+=1"
        F-->>C: "on_commit send_task link=mark_dispatched link_error=mark_dead_letter"
    end
    F->>DB: Reclaim stale SENDING to PENDING
    F->>DB: GC terminal rows + inactive PENDING past retention

    C->>WH: POST clubbed envelope summary+events
    WH-->>C: 200 OK
    C->>DB: mark_buffer_dispatched DISPATCHED

    alt Retry exhaustion
        C->>DB: mark_buffer_dead_letter DEAD_LETTER
    end

Loading

Prompt To Fix All With AI

Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 2
backend/backend/settings/base.py:228-230
The inline comment says "PENDING rows are never GC'd regardless of age," but `_gc_terminal_rows()` in `internal_api_views.py` now also deletes PENDING rows whose source notification is inactive and whose `flush_after` is past the retention cutoff. The comment should be updated to match the actual behaviour.

```suggestion
# Retention window for NotificationBuffer GC. Two sweeps run on each flush tick:
# 1. Terminal rows (DISPATCHED / DEAD_LETTER) older than this window.
# 2. PENDING rows whose source notification is inactive and whose flush_after
#    is past this window (see _gc_terminal_rows in internal_api_views.py).
#    PENDING rows for *active* notifications are never touched.
NOTIFICATION_BUFFER_RETENTION_DAYS = int(
```

### Issue 2 of 2
backend/workflow_manager/workflow_v2/models/execution.py:443-447
**Historical executions show 0/0 file counts — affects failure filter for pre-migration rows**

For any `WorkflowExecution` row created before this migration, `successful_files` and `failed_files` are `null`. `e.successful_files or 0` coalesces to `0`, so `is_failure_run(status, 0)` returns `False` for COMPLETED runs where files actually failed. This means failure-only subscribers won't receive notifications for partial failures on historical re-triggers until the worker has written the new column values at least once for that execution.

The list view cosmetic impact (showing 0/0 for old rows) is a known trade-off per the migration comment, but the silent miss on `notify_on_failures` filtering is worth calling out so operators know to expect a brief gap for runs that were already in-flight at deploy time.

_{Reviews (26): Last reviewed commit: "UN-3056 [FIX] Single-event webhook compa..." | Re-trigger Greptile}

[kirtimanmishrazipstack]

Review triage summary

Thanks to @coderabbitai and @greptile-apps for the thorough pass. Status of each item:

Fixed in this branch

• CR clubbed_renderer.py × → x (RUF002)
• CR auth_sig JSON-encoded before SHA-256 (no delimiter-collision)
• CR + Greptile P1: platform added to flush grouping and folded into the original idx_notif_buffer_pending partial index in migration 0003 (in lockstep with Meta.indexes)
• CR + Greptile P2: PlatformSettings.jsx useEffect guards on orgId + adds it to deps
• CR: _enqueue_to_buffer now logger.exception/raises; caller catches per-row so sibling notifications still iterate
• Greptile P1 duplicate-delivery race: dispatch order in _dispatch_group reversed — UPDATE first inside transaction.atomic(), broker send_task runs via transaction.on_commit(...); broker failure post-commit reverts rows to PENDING for next-tick retry
• Greptile P2: _FAILURE_STATUSES consolidated into a single notification_v2.enums.FAILURE_STATUSES
• CR outside-diff: WorkflowExecutionStatusUpdateSerializer.validate(...) added for cross-field aggregate checks
• CR nitpick (tasks.py): implicit string concat merged

Stale / wont fix (replies posted on each thread with reasoning)\n- Tri-state notify_on (CR migration 0002 / model / CreateNotification.jsx) — PR scope cut to boolean notify_on_failures; PR description updated to match\n- BATCHED-as-default migration backfill (CR + Greptile P1) — deliberate v2.7 design choice; operator note added to PR description\n- STOPPED as failure (CR) — db_comment in migration 0002 is the spec\n- APIWebhook double-wrap (CR) — fresh instance per dispatch in current dispatcher loop; not reachable today\n- Historical NULL-aggregate PARTIAL_SUCCESS in last_run_statuses tooltip (CR) — display-only path outside UN-3056 scope; deferred\n\n### Nitpick — not adopting\n- permission_classes as tuple (CR views.py:56-68): project convention is list across every DRF view (verified across 10+ files including pipeline_v2/execution_view.py, dashboard_metrics/views.py, platform_api/views.py). Keeping consistent with house style.

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

@kirtimanmishrazipstack Thanks for the thorough triage — really appreciate the clear breakdown.

All the fixes look solid:

• The transaction.on_commit(...) broker dispatch ordering is the right approach for the duplicate-delivery race.
• _FAILURE_STATUSES consolidation into notification_v2.enums is a clean improvement.
• The WorkflowExecutionStatusUpdateSerializer.validate(...) cross-field check is a good catch to have in place.

The reasoning on the won't-fix items is sound — the db_comment as spec for STOPPED, the deliberate BATCHED-as-default design decision, and the deferred display-only PARTIAL_SUCCESS path are all reasonable calls for this PR's scope.

Noted on the permission_classes convention — keeping it as a list is the right call for project consistency.

---

📝 Learnings were identified and not saved (knowledge base disabled). Enable

[greptile-apps]

Thanks for the thorough triage write-up. All of my P1 and P2 items are confirmed addressed:

• Duplicate-delivery race → UPDATE inside transaction.atomic() + on_commit broker dispatch with PENDING revert on post-commit broker failure. Correct pattern.
• platform flush grouping → migration 0004 rebuilds the partial index in lockstep.
• FAILURE_STATUSES consolidation → single source of truth in notification_v2.enums.
• _enqueue_to_buffer error handling → logger.exception/raise with per-row catch in the caller so siblings still iterate.
• useEffect orgId guard → dependency array fixed.

The "wont fix" reasoning all holds up:

• Boolean notify_on_failures scope cut is reasonable and the PR description is updated to match.
• BATCHED-as-default is a deliberate v2.7 design choice with operator notes documented.
• STOPPED as failure is spec'd in the migration db_comment.
• APIWebhook double-wrap is not reachable with the current dispatcher loop structure.
• permission_classes as list — consistent with the project's house style across DRF views.

LGTM from my side. The implementation is solid and the edge cases I flagged are all properly handled.

Tip: You can customize Greptile's behavior for this repo with .greptile/rules.md and .greptile/config.json.

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

@kirtimanmishrazipstack Thanks for the detailed triage — really well organized. Everything looks good:

Fixes confirmed:

• transaction.on_commit(...) for broker dispatch is exactly the right pattern for the duplicate-delivery race.
• _FAILURE_STATUSES consolidation into notification_v2.enums is a clean improvement for future consumers.
• The WorkflowExecutionStatusUpdateSerializer.validate(...) cross-field check is a solid guard to have.
• Per-row exception handling in _enqueue_to_buffer preserving sibling iteration is the right resilience model.

Won't-fix reasoning is sound:

• Boolean notify_on_failures is a cleaner scope for this PR; the tri-state can come in a follow-up if needed.
• BATCHED-as-default as a deliberate v2.7 design choice with an operator note in the PR description is the right way to handle that.
• db_comment as the spec for STOPPED-as-failure is clear and self-documenting.
• The PARTIAL_SUCCESS display path being deferred is a reasonable call — it's isolated and not on the critical notification path.

permission_classes convention noted — keeping list for consistency with the rest of the project is the right call.

---

📝 Learnings were identified and not saved (knowledge base disabled). Enable

[johnyrahul]

🔴 The dead-letter path never fires — failed webhook deliveries are silently lost

While tracing the dispatch/flush concurrency I hit a correctness gap in the dead-letter wiring that I don't think is covered by the existing review threads. Flagging it because it defeats the reliability guarantee the buffer design is built around.

Symptom: if a subscriber's webhook endpoint is down (or returns 4xx/5xx) for its whole retry budget, the run produces zero notifications, zero DEAD_LETTER rows, and zero error signal — and the buffer rows are eventually GC'd as if they'd been delivered.

Root cause — the link_error contract doesn't match the worker task's behavior:

• _send_clubbed wires mark_buffer_dead_letter as a Celery link_error on send_webhook_notification:

  # backend/notification_v2/internal_api_views.py:433
  link_error=celery_app.signature(
      "notification_v2.mark_buffer_dead_letter",
      kwargs={"buffer_row_ids": buffer_ids},
  ),

• But send_webhook_notification returns None on every terminal-failure branch instead of raising:

  # workers/notification/tasks.py:259 / :262 / :278 / :281
  return None  # Final failure - matches original behavior

  Celery only invokes link_error when a task ends in the FAILURE state (i.e. it raised). A task that swallows the error and returns None ends in SUCCESS, so mark_buffer_dead_letter is never called.

Consequence chain:

1. _dispatch_group has already flipped the rows to DISPATCHED (internal_api_views.py:514-517) before the on_commit publish.
2. Delivery fails through all retries → task returns None → link_error never fires → no row ever reaches DEAD_LETTER.
3. The rows stay DISPATCHED forever (the flush query filters status=PENDING, so they're never revisited).
4. _gc_terminal_rows (internal_api_views.py:393-396) deletes aged DISPATCHED rows — i.e. failed deliveries are cleaned up exactly as if they had succeeded.

For a failure-alerting feature, this means the failure alert itself can be lost with no trace.

Suggested fix: make the retry-exhaustion branches of send_webhook_notification re-raise so the task ends FAILURE and link_error runs:

else:
    logger.error("Failed to send webhook to %s after %s attempts: %s", url, max_retries, e)
    raise  # let link_error → mark_buffer_dead_letter run

If the legacy "return None" contract must be preserved for other callers, gate the re-raise behind a flag passed from _send_clubbed (e.g. dead_letter_on_failure=True). Worth adding a test that asserts rows reach DEAD_LETTER on delivery exhaustion.

---

Related, lower-severity hardening (same end-state, different trigger): transaction.on_commit(_send_clubbed) is not crash-durable. If the backend process dies in the window between the transaction committing (rows now DISPATCHED) and the on_commit lambda executing, the lambda is lost — no broker publish, no revert — and the rows are stranded in DISPATCHED with nothing to retry them. Same outcome as above: a row that says "dispatched" but was never delivered.

Both stem from the same root: DISPATCHED currently means "claimed for sending," but is treated as "delivered." Beyond the link_error fix, a periodic sweep that reclaims rows stuck in DISPATCHED with no terminal outcome past a timeout would close the crash-window case too.

Note: the broker-publish failure path inside _send_clubbed itself (RabbitMQ unreachable → except reverts rows to PENDING, retried next tick) is handled correctly — this issue is specifically about delivery failure after a successful publish.

[johnyrahul]

🔴 The dead-letter path never fires — failed webhook deliveries are silently lost

While tracing the dispatch/flush concurrency I hit a correctness gap in the dead-letter wiring that I don't think is covered by the existing review threads. Flagging it because it defeats the reliability guarantee the buffer design is built around.

Symptom: if a subscriber's webhook endpoint is down (or returns 4xx/5xx) for its whole retry budget, the run produces zero notifications, zero DEAD_LETTER rows, and zero error signal — and the buffer rows are eventually GC'd as if they'd been delivered.

Root cause — the link_error contract doesn't match the worker task's behavior:

* `_send_clubbed` wires `mark_buffer_dead_letter` as a Celery `link_error` on `send_webhook_notification`:
  ```python
  # backend/notification_v2/internal_api_views.py:433
  link_error=celery_app.signature(
      "notification_v2.mark_buffer_dead_letter",
      kwargs={"buffer_row_ids": buffer_ids},
  ),
  ```

* But `send_webhook_notification` **returns `None`** on every terminal-failure branch instead of raising:
  ```python
  # workers/notification/tasks.py:259 / :262 / :278 / :281
  return None  # Final failure - matches original behavior
  ```
  
  
      
        
      
  
        
      
  
      
    
  Celery only invokes `link_error` when a task ends in the **FAILURE** state (i.e. it raised). A task that swallows the error and returns `None` ends in **SUCCESS**, so `mark_buffer_dead_letter` is never called.

Consequence chain:

1. `_dispatch_group` has already flipped the rows to `DISPATCHED` (`internal_api_views.py:514-517`) before the `on_commit` publish.

2. Delivery fails through all retries → task returns `None` → `link_error` never fires → no row ever reaches `DEAD_LETTER`.

3. The rows stay `DISPATCHED` forever (the flush query filters `status=PENDING`, so they're never revisited).

4. `_gc_terminal_rows` (`internal_api_views.py:393-396`) deletes aged `DISPATCHED` rows — i.e. failed deliveries are cleaned up exactly as if they had succeeded.

For a failure-alerting feature, this means the failure alert itself can be lost with no trace.

Suggested fix: make the retry-exhaustion branches of send_webhook_notification re-raise so the task ends FAILURE and link_error runs:

else:
    logger.error("Failed to send webhook to %s after %s attempts: %s", url, max_retries, e)
    raise  # let link_error → mark_buffer_dead_letter run

If the legacy "return None" contract must be preserved for other callers, gate the re-raise behind a flag passed from _send_clubbed (e.g. dead_letter_on_failure=True). Worth adding a test that asserts rows reach DEAD_LETTER on delivery exhaustion.

Related, lower-severity hardening (same end-state, different trigger): transaction.on_commit(_send_clubbed) is not crash-durable. If the backend process dies in the window between the transaction committing (rows now DISPATCHED) and the on_commit lambda executing, the lambda is lost — no broker publish, no revert — and the rows are stranded in DISPATCHED with nothing to retry them. Same outcome as above: a row that says "dispatched" but was never delivered.

Both stem from the same root: DISPATCHED currently means "claimed for sending," but is treated as "delivered." Beyond the link_error fix, a periodic sweep that reclaims rows stuck in DISPATCHED with no terminal outcome past a timeout would close the crash-window case too.

Note: the broker-publish failure path inside _send_clubbed itself (RabbitMQ unreachable → except reverts rows to PENDING, retried next tick) is handled correctly — this issue is specifically about delivery failure after a successful publish.

@kirtimanmishrazipstack Please check this scenrio , by configuring a webhook which is not working

[kirtimanmishrazipstack]

@johnyrahul good catch on the dead-letter path — fixed in c2dd5ee03.

Root cause confirmed exactly as you described: send_webhook_notification returned None on retry exhaustion, so the task ended SUCCESS and link_error → mark_buffer_dead_letter never fired; rows stayed in the (then) terminal dispatched state and got GC'd as if delivered.

Fix:

1. send_webhook_notification now takes raise_on_final_failure and re-raises on exhaustion; _send_clubbed passes it True, so the buffered path ends FAILURE → link_error fires → rows reach DEAD_LETTER.
2. Also closed the on_commit crash-window you flagged: added a SENDING claim state. _dispatch_group marks rows SENDING (not "dispatched"), the success callback (mark_buffer_dispatched) moves them to DISPATCHED, the error callback to DEAD_LETTER, and a new _reclaim_stale_sending reaper returns rows stuck in SENDING past NOTIFICATION_DISPATCH_LEASE_SECONDS (default 900s) back to PENDING. So a backend crash between commit and publish no longer strands rows — they retry on a later flush. DISPATCHED now genuinely means "delivered".

Could you re-run your broken-webhook scenario (point a notification at a dead URL) and confirm the rows land in DEAD_LETTER after the retry budget? That's the one path I'd like a live confirmation on.

[kirtimanmishrazipstack]

🔎 Comprehensive review — Failure-only notifications + batched dispatch

Reviewed the full feature surface (38 files: notification_v2, workflow_v2, pipeline_v2, api_v2, unstract/core, workers/, frontend Platform Settings) across five lenses: general correctness, error-handling/silent-failures, type design, comments, and test coverage. The design is genuinely strong — the on-commit dispatch ordering, SELECT … FOR UPDATE SKIP LOCKED group flush, SENDING-guarded terminal callbacks, the partial PENDING index, and the single shared renderer are all the right calls. Findings below are filtered for signal (two agent "criticals" were downgraded after verifying the code — see notes).

✅ Strengths worth keeping

• _dispatch_group claims rows SENDING inside the txn and publishes via transaction.on_commit, so a rolled-back claim never orphans a broker message (no DB-vs-broker duplicate send).
• mark_buffer_dispatched / mark_buffer_dead_letter both gate their UPDATE on status=SENDING, so a reaper-reclaimed row can't be clobbered by a stale callback.
• Org is a mandatory grouping key everywhere; auth_sig JSON-encodes the tuple before hashing to avoid delimiter collisions across tenants.
• WorkflowExecutionStatusUpdateSerializer.validate rejects impossible aggregates (successful + failed > total, etc.).
• Thorough db_comments and the shared unstract.core.notification_clubbed_renderer give byte-identical receiver payloads on both dispatch sides.

---

🟠 Important

1. The failure-filter rule has drifted across its 3 sites.

• backend/api_v2/notification.py:28-30 → is_failure(status) or failed_files > 0
• backend/notification_v2/internal_api_views.py _apply_failure_filter → same two terms
• backend/pipeline_v2/notification.py:53-56 → adds a third term or pipeline.last_run_status == FAILURE

The worker callback path for pipelines uses _apply_failure_filter (two terms), while the backend PipelineNotification.send path uses three. A pipeline run with non-failure status/failed_files==0 but last_run_status==FAILURE would notify failure-only subscribers on one path and not the other. The _apply_failure_filter docstring even claims parity with the pipeline site, which isn't true. Please confirm whether the third term is intentional and, either way, mirror the rule (or extract one shared helper) and fix the docstring. This is exactly the "notifications fire from worker callback and backend; mirror both paths" hazard.

2. The backend webhook-provider cluster is dead code — and this PR modified it.
backend/notification_v2/provider/webhook/{api,slack}_webhook.py gained envelope-wrapping + "IMMEDIATE" docstrings, but get_notification_provider, NotificationHelper, and provider/registry.py have zero callers in backend/ (verified). Every live backend dispatch now goes dispatch_notifications → enqueue → NotificationBuffer → flush → worker. Recommend deleting the provider/webhook/ + registry.py cluster (or confirming it's retained for a planned non-buffered path). Either way the "IMMEDIATE" comments are stale — the platform is BATCHED-only.

3. Buffer redelivery is unbounded if a terminal callback is ever lost.
(Downgraded from a reported "Critical": the link/link_error signatures have no explicit queue=, so they route to the default celery queue — which is a backend-consumed queue per celery_config._BACKEND_QUEUES. So the common path works.) The residual risk: if a terminal callback is genuinely lost (worker for celery down, broker hiccup), _reclaim_stale_sending recycles SENDING → PENDING with no attempt counter, so the batch is re-dispatched every lease window forever — duplicate webhook spam, and a legitimately dead-lettered batch never reaches DEAD_LETTER. Suggest a reclaim counter that transitions to DEAD_LETTER after N reclaims, and an alert on sustained nonzero notification_buffer_reclaimed_total. Pinning the callback signatures to an explicit backend queue would also remove the implicit-routing assumption.

4. Dropped notifications aren't observable enough for a failure-alerting feature.
dispatch_notifications (helper.py:121-129) and _route_notification (workers/shared/patterns/notification/helper.py) swallow per-row failures into logger.exception/logger.warning with no metric= counter, unlike the success path (metric=notification_buffer_enqueued_total). A silently dropped failure alert is precisely what this feature exists to prevent. Add metric=notification_*_failed_total counters at those sites, and thread org_id/webhook_url_hash/buffer-id sample into the mark_buffer_dead_letter log so a delivered-never event is actionable.

5. process_notification_buffer.py can raise despite its "never raises" contract.
result = response.json() (line ~70) sits outside the try/except. A 200 with a non-JSON body (proxy/error page) raises JSONDecodeError, bypassing the clean return False the docstring promises. Wrap response.json() + the .gets and return False on decode error.

---

🟡 Suggestions

6. No tests for highly-testable logic. The PR adds zero tests; the pure functions in unstract/core/notification_clubbed_renderer.py need no DB/Django. Highest value (follow the existing unstract/core/tests/test_pubsub_helper.py unittest pattern):

• compute_auth_sig — equality + the classic collision pair (key="a",header="b,c" vs key="a,b",header="c") → distinct sigs (multi-tenant safety).
• Cross-path failure-filter truth table (ties into fix: Prompt studio index and run issues #1).
• _is_effective_success / build_envelope summary counts (COMPLETED + failed_files>0 ⇒ counts as failed).
• WorkflowExecutionStatusUpdateSerializer.validate five branches.
• render_slack_text overflow/pluralization (1, 25, 26 events), _humanize_timestamp fallback, build_envelope 501-payload cap.

7. Single-source the batch cap. _PROCESS_BUFFER_CAP = 500 (internal_api_views) and MAX_BATCH_SIZE = 500 (renderer) are independent literals; the comment claims "lock-step" but nothing enforces it — raising the cap above MAX_BATCH_SIZE would silently mark un-rendered rows DISPATCHED. Have internal_api_views import MAX_BATCH_SIZE.

8. NotificationBuffer.status db_comment omits SENDING (models.py:176 + the duplicate in migration 0002). It reads PENDING -> DISPATCHED / DEAD_LETTER, hiding the real PENDING → SENDING → … machine the enum docstring documents correctly. Fix both (a db_comment change needs a migration).

9. Type encapsulation (optional, no behavior change):

• Promote the {summary, events} envelope from dict[str, Any] to a frozen dataclass in unstract/core — it crosses the backend↔worker JSON boundary, has a real arithmetic invariant (total == succeeded + failed), and is read in 5+ places; replaces the fragile "events" not in payload heuristic.
• Add BufferStatus.terminal() (mirroring the ExecutionStatus.failure_statuses() this PR already adds) and centralize the filter(status=SENDING) transition guard into one helper.
• Derive NotificationSettingsSerializer bounds from ConfigKey.NOTIFICATION_CLUB_INTERVAL's ConfigSpec instead of repeating 60/7200; hoist the frontend 1/120 to one constant.

10. Stale comments: worker providers/{api,slack}_webhook.py docstrings attribute the flat-payload path to "worker-callback payloads", but callbacks now go through the backend buffer pre-rendered — the flat-wrap branch is only hit by the generic internal webhook-send endpoints. workers/shared/patterns/notification/helper.py:35-37 and internal_api_views.py:354 describe a "Type: — / Additional Data" KV layout the single-line renderer no longer produces.

---

Review assisted by Claude Code (multi-agent: code-review, silent-failure, type-design, comment-accuracy, test-coverage). Findings were verified against the code; the two highest-severity agent flags were downgraded after confirming queue routing and the lock-step cap behavior.

[kirtimanmishrazipstack]

✅ Self-review dispositions — addressed in 8b05977a9

Worked through all 10 findings. Fixes landed in 8b05977a9; three items are deliberately deferred (rationale below).

#	Finding	Disposition
1	Failure-filter rule drift	Fixed — extracted notification_v2.helper.is_failure_run; api_v2, pipeline_v2 and _apply_failure_filter all use it. The pipeline path keeps its last_run_status==FAILURE backstop (now commented as intentional, for the no-loadable-execution case), and the false "parity" docstring is corrected.
2	Dead provider/ cluster	Fixed (comments only) — scrubbed the stale IMMEDIATE/envelope docstrings this PR added. Kept the files: they pre-exist on main and may be referenced by the cloud repo, so deletion is out of scope here.
3	Unbounded redelivery on lost callback	Deferred — real gap, not yet implemented. The SENDING lease bounds the common path; the unbounded case only triggers on a genuinely lost terminal callback. Proper fix needs a reclaim_count column → DEAD_LETTER after N reclaims + an alert on sustained notification_buffer_reclaimed_total. Tracking as a follow-up to avoid schema churn in this PR.
4	Dropped notifications not observable	Fixed — added metric=notification_enqueue_failed_total / notification_dropped_total at the backend (dispatch_notifications) and worker (_route_notification) drop sites, plus a row-id sample= on the dead-letter log.
5	process_notification_buffer.py can raise	Fixed — response.json() is now inside the try/except; a non-JSON 200 returns False, honoring the "never raises" contract.
6	No tests	Deferred — per the standing "manual test only" decision for this PR. Pure-function targets (compute_auth_sig, renderer, validate) noted for a follow-up.
7	Batch cap not single-sourced	Fixed — _PROCESS_BUFFER_CAP = MAX_BATCH_SIZE (imported via clubbed_renderer), so rows and rendered events stay lock-step by construction.
8	status db_comment omits SENDING	Fixed — documents the PENDING → SENDING → DISPATCHED/DEAD_LETTER lifecycle in both the model and migration 0002.
9	Type-design polish	Deferred — optional, no behavior change; follow-up.
10	Stale comments	Fixed — corrected the worker-callback / KV-layout / IMMEDIATE comments in the worker and backend provider docstrings.

Note for testing: the worker image needs a rebuild (worker containers don't hot-reload) before the manual run; the migration 0002 db_comment change applies cleanly (model + migration agree, no drift).

[kirtimanmishrazipstack]

✅ Self-review resolution

Addressing my review above. All 5 Important findings are resolved; 3 of 5 suggestions done, 2 consciously deferred (both non-behavioral). Fixes landed in 8b05977a9 (in-scope self-review items) and 170ec43d8 (dead-code removal + redelivery cap).

🟠 Important — all resolved

#	Finding	Resolution
1	Failure-filter rule drift + false parity docstring	8b05977a9 — single-sourced as notification_v2.helper.is_failure_run, used by api_v2 / pipeline_v2 / _apply_failure_filter. The pipeline's third last_run_status term is kept as a documented backstop (only reached when the WorkflowExecution can't be loaded — legacy caller / missing execution_id), and the misleading "parity" docstring is corrected.
2	Dead backend webhook-provider cluster	170ec43d8 — deleted the entire notification_v2/provider/ package. Verified zero callers on-branch (the batched dispatch_notifications path replaced it) and zero references in cloud. Sending now lives solely in the worker's own providers/ registry.
3	Unbounded redelivery if a terminal callback is lost	170ec43d8 — added NotificationBuffer.dispatch_attempts + NOTIFICATION_MAX_DISPATCH_ATTEMPTS; _dispatch_group increments on each SENDING claim and dead-letters rows past the cap, so a lost callback can't redeliver forever. metric=notification_buffer_reclaimed_total is emitted for alerting. (The optional explicit-queue pinning on the link/link_error callbacks was intentionally skipped — celery is a backend-consumed queue per _BACKEND_QUEUES, so the attempt cap addresses the actual risk.)
4	Dropped notifications not observable	8b05977a9 — metric=…_failed_total counters at both drop sites (dispatch_notifications, worker _route_notification); the dead-letter log now carries a row-id sample= so a delivered-never event is traceable.
5	process_notification_buffer.py can raise despite "never raises"	8b05977a9 — response.json() is wrapped; a non-JSON 200 returns False instead of raising.

🟡 Suggestions

#	Suggestion	Status
7	Single-source the batch cap	✅ 8b05977a9 — _PROCESS_BUFFER_CAP = MAX_BATCH_SIZE.
8	status db_comment omits SENDING	✅ 8b05977a9 + 170ec43d8 — model + migration db_comment now document PENDING → SENDING → DISPATCHED / DEAD_LETTER (incl. the attempt-cap path).
10	Stale worker / renderer comments	✅ 8b05977a9 — provider docstrings and the KV-layout references scrubbed.
6	Unit tests for the pure functions	⏳ Deferred — not in this PR. Highest-value targets remain compute_auth_sig (incl. the collision pair), the renderer (build_envelope / render_slack_text overflow), and the serializer branches.
9	Type encapsulation (envelope dataclass, BufferStatus.terminal(), serializer bounds dedup)	⏳ Deferred — optional, no behavior change; left out to keep this PR's diff focused.

Net: every correctness / observability / resilience finding is in; the two deferred items are non-behavioral (test coverage + ergonomics).

[github-actions]

Frontend Lint Report (Biome)

✅ All checks passed! No linting or formatting issues found.

[github-actions]

Unstract test results

Per-group results

Status	Group	Tier	Passed	Failed	Errors	Skipped	Duration (s)
❌	unit-connectors	unit	64	12	0	3	16.8
❌	unit-core	unit	0	0	2	0	1.2
❌	unit-platform-service	unit	9	0	1	0	1.3
✅	unit-prompt-service	unit	15	0	0	0	18.7
✅	unit-rig	unit	53	0	0	0	3.3
✅	unit-runner	unit	11	0	0	0	3.4
✅	unit-sdk1	unit	354	0	0	0	20.0
❌	unit-tool-registry	unit	0	0	1	0	1.4
⚪	unit-workers	unit	0	0	0	0	17.6
	TOTAL		506	12	4	3	83.6

Critical paths

⚠️ Critical paths not yet covered

• auth-login — User can log in and obtain a session cookie. (entry: POST /api/v1/auth/login; declared coverage: no groups declared)
• adapter-register-llm — Register and validate an LLM adapter. (entry: POST /api/v1/adapter/; declared coverage: no groups declared)
• workflow-create-execute — Create a workflow, configure source+destination, execute, poll, fetch result. (entry: POST /api/v1/workflow/{id}/execute/; declared coverage: e2e-workflow)
• api-deployment-run — Deploy a workflow as an API, POST a document, receive structured JSON. (entry: POST /deployment/api/{org}/{name}/; declared coverage: e2e-api-deployment)
• prompt-studio-fetch-response — Prompt Studio: create project, add prompt, run single-pass, get response. (entry: POST /api/v1/prompt-studio/prompt-studio-tool/{id}/fetch_response/; declared coverage: e2e-prompt-studio)
• pipeline-etl-execute — Run an ETL pipeline from source connector to destination. (entry: POST /api/v1/pipeline/{id}/execute/; declared coverage: no groups declared)
• usage-token-tracking — Per-execution token usage is recorded and retrievable. (entry: GET /api/v1/usage/get_token_usage/; declared coverage: no groups declared)
• workflow-execution-fan-out — Multi-file workflow execution fans out to file-processing workers and rejoins. (entry: internal: backend → rabbitmq → workers/file_processing; declared coverage: no groups declared)
• callback-result-delivery — Async results are posted back via the callback worker. (entry: internal: workers/callback → backend /internal endpoints; declared coverage: no groups declared)

✅ Covered critical paths

• tool-sandbox-exec — covered by unit-runner

[chandrasekharan-zipstack]

The flush poll cadence reuses LOG_HISTORY_CONSUMER_INTERVAL (fallback 5 here), but the Django setting of the same name defaults to 60 (backend/backend/settings/base.py). So the "5s tick is cheap" assumption in the comment block below only holds if nothing exports the var — in any deployment that sets it (e.g. 60 for the log-history consumer), the buffer flush silently polls every 60s instead, and one env knob now governs two unrelated tasks.

Suggest a dedicated var for the flush (e.g. NOTIFICATION_BUFFER_POLL_INTERVAL), decoupled from the log-history consumer, and reconcile the default with the comment. Functional impact is bounded (worst case a due group waits one extra tick), but the coupling + the now-misleading "5s" comment are a maintenance trap.

[kirtimanmishrazipstack]

Good catch — fixed in 600eade02. Added a dedicated NOTIFICATION_BUFFER_POLL_INTERVAL (default 10s) for the flush, decoupled from LOG_HISTORY_CONSUMER_INTERVAL. scheduler.sh now wakes at the min of the two cadences and fires each task on its own elapsed interval, so setting the log-history knob no longer changes the flush cadence. Rewrote the misleading "5s" comment (real dispatch cadence is still gated by NOTIFICATION_CLUB_INTERVAL). Kept the single loop + single trap — no subshell/signal rewrite — since the functional impact you noted is bounded.

[chandrasekharan-zipstack]

This re-implements the failure rule that notification_v2.helper.is_failure_run already owns (status ∈ {ERROR, STOPPED} OR failed_files > 0). The docstring even flags it as "mirrors the failure-filter contract" — but two copies of one rule across two packages will drift, and a drift here means the rendered summary counts disagree with the very reason the alert fired.

Since unstract.core is the lower layer, the canonical predicate should live here and the backend is_failure_run import it (not the reverse mirror). One definition, no drift — same class of issue as the hardcoded-list drift flagged on the group-sharing PR.

[kirtimanmishrazipstack]

Agreed — fixed in 600eade02. Since the canonical ExecutionStatus already lives in unstract.core.data_models, I added the canonical is_failure_run(status, failed_files) right beside it and made both consumers use it:

• the renderer now derives succeeded/failed and the per-event emoji from not is_failure_run(...) (dropped the duplicated _SUCCESS_STATUSES / _is_success / _is_effective_success);
• notification_v2.helper.is_failure_run delegates to the core function (all existing imports unchanged).

One definition, lower layer, no reverse mirror — routing filter and rendered summary can't drift. Verified summary counts identical to before across single/multi, COMPLETED/ERROR/partial-failure payloads.

[chandrasekharan-zipstack]

⚠️ Webhook payload is a hard break for existing receivers — needs a rollout/compat plan

The clubbed renderer always emits the new {summary, events} envelope (even for a single event), with no version field and no compat shape — by design ("one envelope shape so receivers parse a single schema, not two"). That means every existing external webhook receiver breaks: they parse the old flat per-run body, get the new envelope, and silently fail or drop alerts. The "Relevant Docs" section is also empty, so today this ships as an undocumented contract change to a customer-facing surface.

Flagging two decisions to settle before merge:

1. Payload shape (flat → {summary, events}) — options, cheapest-safe first:

• Single-event superset (suggested): when a group has exactly one event, also emit the old flat top-level fields alongside events. Existing receivers keep working, new ones read events; trivial payload bloat, removable after a deprecation window. Self-healing, no customer action required.
• Versioned envelope ("version": 2): future-proofs but doesn't save receivers that don't yet branch on it.
• Opt-in clubbing (per-notification format flag, old shape default): cleanest rollout, more config surface.
• Hard break + comms: acceptable only if we know who consumes these webhooks and warn them — and the docs/changelog land with the PR.

2. Alerts are no longer instant. Window min is 60s, so there's no instant path even for failures — a critical failure alert is delayed up to window + one poll tick. Is a 1-min+ floor acceptable for failure/paging use-cases? Worth considering a low-latency path (send failures immediately, club successes) so the failures-only feature and the clubbing feature don't work against each other — and surfacing the expected latency in the notification UI so users aren't surprised.

Neither is a code bug — both are product/rollout calls that need an owner. Happy to help implement whichever direction we pick (the single-event superset is a small change to build_envelope).

[kirtimanmishrazipstack]

Thanks @chandrasekharan-zipstack — addressed both:

1. Payload break → fixed in 600eade02 (single-event superset). Went with your cheapest-safe option. build_envelope now, when a batch holds exactly one event, also spreads the legacy flat fields onto the envelope alongside summary/events, reproducing main's pre-clubbing shape: {type, pipeline_id, pipeline_name, status, execution_id?, error_message?}. Existing API webhook receivers parsing the flat body keep working; new ones read events. Multi-event batches stay envelope-only (there was never a flat shape for them), and Slack is untouched (it only ever gets {text}). No key collision — summary/events are new. Self-healing, removable after a deprecation window. Documented in the design notes.

2. Latency floor → keeping batched-only (by design), not adding an instant path. Batched-only is a deliberate post-v2.8 decision (no IMMEDIATE path), so a low-latency carve-out would reverse a settled architecture rather than tune it. The window is per-org configurable down to 60s, so an org that wants fast failure paging can set the floor; failures and clubbing don't actually fight here because the failures-only filter runs at enqueue, independent of the window. I'd treat "surface expected latency in the notification UI" as a small, worthwhile follow-up rather than part of this PR — agree it's worth setting the expectation for users. Happy to file it.

Relevant docs gap noted — I'll make sure the changelog/customer-facing note lands with this (the compat superset means it's no longer a hard break, but the new events shape should still be documented).

[sonarqubecloud]

Quality Gate passed

Issues
4 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.3% Duplication on New Code

See analysis details on SonarQube Cloud

[jaseemjaskp]

PR Review Toolkit — consolidated findings

Ran Code Reviewer, Silent Failure Hunter, Type Design Analyzer, PR Test Analyzer, Comment Analyzer, and Code Simplifier over the diff (43 files, ~2.4k insertions). The architecture is sound — on-commit dispatch ordering, SKIP-LOCKED grouping, the dispatch-attempt cap with broker-failure refund, and the lease-based reaper are all correctly reasoned. Inline comments below carry the new, deduplicated findings (existing review threads were cross-checked and not reposted).

Two findings that can't be anchored to diff lines:

P0 — Existing test suite is broken by this PR. workers/tests/test_dispatch_sites_characterisation.py (lines 42-127, 267) imports/exercises send_notification_to_worker and get_webhook_headers in workers/shared/patterns/notification/helper.py — both deleted by this PR. Those tests will fail at collection, and they are currently the only coverage of the worker dispatch path (now rewritten as _route_notification/_enqueue_to_buffer). They must be rewritten to characterise the new HTTP enqueue contract, or removed.

P1 — The feature ships with zero new tests. No test file is touched in the diff and there is no backend/notification_v2/tests/ directory. The highest-ROI gaps are pure functions that decide whether the right alert is sent with correct content and need no DB:

• is_failure_run (unstract/core/.../data_models.py) — the single routing predicate: ERROR/STOPPED, COMPLETED + failed_files>0 (partial failure), failed_files=None, unknown-status.
• clubbed renderer (notification_clubbed_renderer.py) — _humanize_timestamp portability (the %-d regression that was already fixed once), MAX_BATCH_SIZE=500 cap, Slack SLACK_MAX_DISPLAY_EVENTS=25 overflow footer, and the legacy single-event flat-key envelope shape (reworked twice in 577c881c/600eade0).
• dispatch_attempts refund on broker-publish failure (internal_api_views._send_clubbed) — the exact invariant commit 16c2dbac had to fix; a regression silently dead-letters never-delivered failure alerts. Test with TestCase.captureOnCommitCallbacks(execute=True) + a mocked send_task that raises vs succeeds.

Priority order: address the broken test (P0) and the two correctness comments below (_load_execution fail-open, _update_file_aggregates invariant); the rest are accuracy/cleanup/simplification.

[jaseemjaskp]

[Correctness / P1] _load_execution returns None both when no execution_id was supplied and when an execution_id was supplied but the row isn't found. _apply_failure_filter then treats None as "no filter" and returns every active row. For a missing-but-requested execution (replication lag / race between the status write and this fetch), this fails open: notify_on_failures=True subscribers get notified on successful runs — the opposite of the feature's intent. Suggest distinguishing the two cases: when execution_id is present but the lookup misses, retry briefly or fail closed (skip notify_on_failures rows) and emit a metric, rather than silently sending success alerts. The same pattern exists in pipeline_v2/notification.py (partially backstopped there by last_run_status).

[jaseemjaskp]

[Validation invariant / P2] This setattrs total/successful/failed_files individually and saves only the present fields. The cross-field invariant (successful + failed <= total) is checked in WorkflowExecutionStatusUpdateSerializer.validate() against the submitted payload only. A partial update sending just failed_files (no total_files) is validated with total_files=None (check skipped) and persisted against a previously-stored, possibly-smaller total_files — so the row can end up with failed_files > total_files. Since is_failure_run keys on failed_files, a bad aggregate mis-routes the failure alerts this feature exists to deliver. Suggest re-validating against the merged/persisted values here (or in WorkflowExecution.save()/clean()).

[jaseemjaskp]

[Docstring accuracy / P2] The module docstring says this task converts rows from PENDING/DISPATCHED to terminal DEAD_LETTER, but both mark_buffer_dead_letter (line 47) and mark_buffer_dispatched (line 80) only transition rows still in SENDING (.filter(status=BufferStatus.SENDING.value)). The inline comment at line 43 already states this correctly. Suggest: ...converts the buffered rows from SENDING to terminal DEAD_LETTER....

[jaseemjaskp]

[Comment accuracy / P2] The db_comment states created_at + NOTIFICATION_CLUB_INTERVAL, but enqueue computes flush_after = timezone.now() + timedelta(seconds=interval) using the per-org interval from get_org_club_interval_seconds (org override, with NOTIFICATION_CLUB_INTERVAL only as the fallback default) — not created_at, and not the bare setting. Suggest: enqueue-time now() + the org's effective club interval (NOTIFICATION_CLUB_INTERVAL default, per-org override), precomputed at enqueue. Apply the same fix to the matching db_comment in migration 0002_notification_batching.py.

[jaseemjaskp]

[Docstring accuracy / P2] The docstring says the endpoint Rejects rows whose source notification is not BATCHED, but there is no notification_type/BATCHED gate anywhere in the function body — every active notification is buffered. Either add the gate or drop this sentence so the docstring matches behavior (misleads maintainers into assuming a guard that isn't there).

[jaseemjaskp]

[Cleanup / P2] Leftover JIRA reference — reviewers previously asked to scrub JIRA keys from comments/docstrings, and this UNS-611 was missed. Suggest: # Task 2: notification buffer flush (clubbed dispatch).

[jaseemjaskp]

[Cleanup / P2] Leftover UNS-611 v2 JIRA reference — please scrub per the earlier review request (other JIRA refs were removed in c2dd5ee03; this one was missed). Suggest: // Load org-scoped batch interval. Falls back silently to ...

[jaseemjaskp]

[Simplification / P3] _dispatch_group returns len(rows), len(rows) — the two elements are always identical, the early returns are all 0, 0, and the sole caller discards the second (rows, _succeeded = _dispatch_group(...) at line 629). succeeded never diverges from rows. Collapse to a single int return, update the caller to rows = _dispatch_group(...), and fix the (rows, succeeded) wording in the docstring at line 503.

[jaseemjaskp]

[Dead code / P3] build_envelope is imported (line 19) and re-exported in __all__, but nothing imports it from this shim — internal_api_views.py and both worker providers import build_envelope directly from unstract.core.notification_clubbed_renderer. Only MAX_BATCH_SIZE and render_clubbed_message are consumed here. Drop build_envelope from the import and __all__ (or add a one-line comment if the re-export is a deliberate public surface).

[jaseemjaskp]

[Simplification / P3] _render_for_slack is a single-use one-liner (return {"text": render_slack_text(envelope)}) called once at line 44. Consider inlining it at the call site and deleting the helper — minor, keep if you value the named intent.