Merge pull request #2164 from rcourtman/maintainer/20260922T164109Z

Group resolved alert notifications to prevent restart delivery bursts

docs/release-control/v6/internal/subsystems/notifications.md

@@ -43,7 +43,9 @@ Unknown and absent persisted targets preserve historical all-destination
 behavior, and destination tag filters still apply after target selection.
 Grouping is an explicit runtime policy: `grouping.enabled=false` or a zero
 window delivers each alert independently, and disabling grouping flushes any
-pending alerts as individual deliveries. Grouped provider payloads must retain
+in-memory pending firing alerts as individual deliveries. Already admitted
+persistent recovery groups retain their original deadline across a policy edit.
+Grouped provider payloads must retain
 every alert, while live ntfy firing deliveries and webhook tests share the same
 severity-derived title, priority, and tags.
 Email rendering treats `patrol_finding` as a finding rather than a numeric
@@ -139,6 +141,46 @@ stable opaque routing identities and must not expose credentials.
 
 ## Current State
 
+### Persistent resolved-alert grouping
+
+`SendResolvedAlert` applies the configured grouping window to recoveries as
+well as firings. Event types are never mixed. With the persistent queue available,
+`internal/notifications/resolved_grouping.go` persists the first recovery's
+window and coalesces only untouched, not-yet-due pending jobs with identical
+notification type, destination configuration and quiet-hours schedule. Queue
+scheduling metadata identifies these groups; it is not the provider HTTP method.
+Deadlines round up to queue timestamp precision, never shorten the grouping
+window, and do not slide with each added occurrence. Retry batches are immutable.
+No schema or driver modification is needed. Pending groups survive manager
+shutdown/reopen. The queue-unavailable fallback uses a separate in-memory buffer
+and does not promise restart durability.
+
+Admission retains destination-specific successful firing receipts. A destination
+that never received an occurrence, or an unannounced alert, must not receive its
+recovery. Group extension deduplicates by alert ID and occurrence start time,
+retains each resolution timestamp and regenerates the group's operational links.
+Delivery consumes only receipts for the occurrences in the delivered job.
+Grouping disabled or a zero window keeps new recoveries individual; changing
+that policy does not rewrite already-admitted persistent jobs.
+
+Generic payloads and built-in/custom recovery summary data retain all grouped
+alerts and the count, rather than rendering only the first alert. PagerDuty's
+built-in Events v2 resolve path remains one deduplication key per request;
+it is explicitly excluded from coalescing to avoid consuming other incidents'
+receipts. Existing rate limits, retry budgets, destination security checks and
+firing grouping are unchanged.
+
+Verification: `resolved_grouping_contract_test.go` establishes ordinary firing
+receipts then resolves fifteen alerts in one process, after manager restart,
+and across restart with pending recoveries. It checks unseen-destination and
+unannounced-alert suppression, per-occurrence resolution metadata, eleven
+recovery renderers, individual PagerDuty keys, and the disabled-grouping burst's
+rate-limit/dead-letter consequence. The HTTP adapter is in-process; queue and
+receipt persistence are real. Focused and race runs establish this synthetic
+boundary, not a full notification-suite result, installed monitor recovery,
+provider acceptance, reporter confirmation or release qualification. The existing
+singleton delivery test explicitly disables grouping.
+
 ### Webhook retry delay conversion
 
 `parseRetryAfterBackoff` bounds parsed integer seconds before multiplying by

internal/notifications/notifications.go

@@ -245,6 +245,8 @@ type NotificationManager struct {
 	groupingEnabled    bool
 	pendingAlerts      []*alerts.Alert
 	groupTimer         *time.Timer
+	pendingResolved    []*alerts.Alert
+	resolvedGroupTimer *time.Timer
 	groupByNode        bool
 	publicURL          string // Full URL to access Pulse
 	groupByGuest       bool
@@ -887,6 +889,9 @@ func (n *NotificationManager) SetNotifyOnResolve(enabled bool) {
 	n.mu.Lock()
 	was := n.notifyOnResolve
 	n.notifyOnResolve = enabled
+	if !enabled {
+		n.takePendingResolvedLocked()
+	}
 	n.mu.Unlock()
 
 	if was != enabled {
@@ -949,12 +954,14 @@ func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNod
 	n.groupByGuest = byGuest
 
 	var pending []*alerts.Alert
+	var resolved []*alerts.Alert
 	var emailConfig EmailConfig
 	var webhooks []WebhookConfig
 	var appriseConfig AppriseConfig
 	var initialTarget notificationDeliveryTarget
 	var queue *NotificationQueue
 	if !enabled || seconds == 0 {
+		resolved = n.takePendingResolvedLocked()
 		pending = append(pending, n.pendingAlerts...)
 		n.pendingAlerts = n.pendingAlerts[:0]
 		if n.groupTimer != nil {
@@ -969,6 +976,9 @@ func (n *NotificationManager) SetGroupingConfig(enabled bool, seconds int, byNod
 	}
 	n.mu.Unlock()
 
+	for _, alert := range resolved {
+		n.dispatchResolvedAlerts([]*alerts.Alert{alert})
+	}
 	for _, alert := range pending {
 		n.dispatchFiringAlerts(emailConfig, webhooks, appriseConfig, []*alerts.Alert{alert}, initialTarget, nil, queue)
 	}
@@ -1063,6 +1073,7 @@ func (n *NotificationManager) SetEnabled(enabled bool) {
 	changed = n.enabled != enabled
 	n.enabled = enabled
 	if !enabled {
+		n.takePendingResolvedLocked()
 		for i := range n.pendingAlerts {
 			n.pendingAlerts[i] = nil
 		}
@@ -1408,7 +1419,10 @@ func (n *NotificationManager) filterResolvedJobsByDeliveryReceipt(jobs []notific
 	return filtered
 }
 
-// SendResolvedAlert delivers notifications for a resolved alert immediately.
+// SendResolvedAlert applies the same grouping window to recoveries as firings.
+// The two event types have separate buffers; a recovery can never be rendered
+// as a firing alert. Normal operation persists the grouping window in the
+// delivery queue; the in-memory buffer is only the queue-unavailable fallback.
 func (n *NotificationManager) SendResolvedAlert(resolved *alerts.ResolvedAlert) {
 	if resolved == nil || resolved.Alert == nil {
 		return
@@ -1426,6 +1440,52 @@ func (n *NotificationManager) SendResolvedAlert(resolved *alerts.ResolvedAlert)
 	}
 	annotateResolvedMetadata(alertCopy, resolvedAt)
 
+	n.mu.Lock()
+	if !n.enabled || !n.notifyOnResolve {
+		n.mu.Unlock()
+		return
+	}
+	if n.queue == nil && n.groupingEnabled && n.groupWindow > 0 {
+		// Repeated reconciliation of one occurrence must not duplicate batch rows.
+		for _, pending := range n.pendingResolved {
+			if pending.ID == alertCopy.ID && pending.StartTime.Equal(alertCopy.StartTime) {
+				n.mu.Unlock()
+				return
+			}
+		}
+		n.pendingResolved = append(n.pendingResolved, alertCopy)
+		if n.resolvedGroupTimer == nil {
+			n.resolvedGroupTimer = time.AfterFunc(n.groupWindow, n.sendGroupedResolvedAlerts)
+		}
+		n.mu.Unlock()
+		return
+	}
+	n.mu.Unlock()
+	n.dispatchResolvedAlerts([]*alerts.Alert{alertCopy})
+}
+
+// takePendingResolvedLocked drains only the recovery buffer. Call with mu held.
+func (n *NotificationManager) takePendingResolvedLocked() []*alerts.Alert {
+	pending := n.pendingResolved
+	n.pendingResolved = nil
+	if n.resolvedGroupTimer != nil {
+		n.resolvedGroupTimer.Stop()
+		n.resolvedGroupTimer = nil
+	}
+	return pending
+}
+
+func (n *NotificationManager) sendGroupedResolvedAlerts() {
+	n.mu.Lock()
+	pending := n.takePendingResolvedLocked()
+	n.mu.Unlock()
+	n.dispatchResolvedAlerts(pending)
+}
+
+func (n *NotificationManager) dispatchResolvedAlerts(alertList []*alerts.Alert) {
+	if len(alertList) == 0 {
+		return
+	}
 	n.mu.RLock()
 	enabled := n.enabled && n.notifyOnResolve
 	emailConfig := copyEmailConfig(n.emailConfig)
@@ -1433,18 +1493,13 @@ func (n *NotificationManager) SendResolvedAlert(resolved *alerts.ResolvedAlert)
 	appriseConfig := copyAppriseConfig(n.appriseConfig)
 	queue := n.queue
 	n.mu.RUnlock()
-
 	if !enabled {
-		log.Debug().
-			Str("alertID", alertCopy.ID).
-			Msg("resolved notifications disabled, skipping")
 		return
 	}
-
-	alertsToSend := []*alerts.Alert{alertCopy}
-	jobs := buildNotificationDeliveryJobs(emailConfig, webhooks, appriseConfig, alertsToSend, eventResolved, resolvedAt)
+	jobs := buildNotificationDeliveryJobs(emailConfig, webhooks, appriseConfig, alertList, eventResolved, resolvedTimeFromAlerts(alertList))
+	// Filter each destination independently at dispatch, not merely when the
+	// alert enters the window. Never announce a recovery for an undelivered firing.
 	jobs = n.filterResolvedJobsByDeliveryReceipt(jobs)
-
 	if queue != nil {
 		n.enqueueNotificationJobs(queue, jobs)
 	} else {
@@ -1628,6 +1683,14 @@ func buildNotificationDeliveryJobsForSelection(
 			routedAlerts := routeNotificationAlerts(alertsToSend, webhook.TagFilter, webhook.TagMode, webhook.MinimumSeverity, event)
 			if len(routedAlerts) > 0 {
 				webhookCopy := webhook
+				// PagerDuty Events v2 resolves one dedup_key, not a list.
+				// Never consume other occurrences' receipts for that request.
+				if event == eventResolved && webhook.Service == "pagerduty" && strings.TrimSpace(webhook.Template) == "" {
+					for _, alert := range routedAlerts {
+						jobs = append(jobs, notificationDeliveryJob{Type: "webhook", Event: event, Alerts: []*alerts.Alert{alert}, ResolvedAt: resolvedAt, WebhookConfig: &webhookCopy})
+					}
+					continue
+				}
 				jobs = append(jobs, notificationDeliveryJob{
 					Type:          "webhook",
 					Event:         event,
@@ -1753,6 +1816,12 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue,
 		return false
 	}
 
+	n.mu.RLock()
+	groupWindow := n.groupWindow
+	if !n.groupingEnabled {
+		groupWindow = 0
+	}
+	n.mu.RUnlock()
 	anyFailed := false
 	for _, job := range jobs {
 		configJSON, err := configJSONForNotificationDeliveryJob(job)
@@ -1770,7 +1839,13 @@ func (n *NotificationManager) enqueueNotificationJobs(queue *NotificationQueue,
 				MaxAttempts:   3,
 				NextRetryAt:   bucket.nextRetryAt,
 			}
-			if err := queue.Enqueue(notif); err != nil {
+			var enqueueErr error
+			if job.Event == eventResolved && groupWindow > 0 && !(job.WebhookConfig != nil && job.WebhookConfig.Service == "pagerduty" && strings.TrimSpace(job.WebhookConfig.Template) == "") {
+				enqueueErr = queue.enqueueResolvedGroup(notif, groupWindow)
+			} else {
+				enqueueErr = queue.Enqueue(notif)
+			}
+			if err := enqueueErr; err != nil {
 				anyFailed = true
 				n.logNotificationJobError(bucket.job, err, "failed to enqueue notification - falling back to direct send")
 				n.dispatchNotificationJobAsync(bucket.job, "failed to send notification after queue enqueue failure")
@@ -2685,6 +2760,17 @@ func (n *NotificationManager) sendResolvedWebhook(webhook WebhookConfig, alertLi
 	data.ResolvedAtISO = resolvedAt.Format(time.RFC3339)
 	data.Duration = formatWebhookDuration(resolvedAt.Sub(alert.StartTime))
 	data.Message = fmt.Sprintf("%s on %s is now healthy", alert.ResourceName, alert.Node)
+	data.AlertCount = len(alertList)
+	data.Alerts = alertList
+	if len(alertList) > 1 {
+		names := make([]string, 0, len(alertList))
+		for _, a := range alertList {
+			if a != nil {
+				names = append(names, fmt.Sprintf("%s on %s", a.ResourceName, a.Node))
+			}
+		}
+		data.Message = fmt.Sprintf("%d alerts resolved: %s", len(alertList), strings.Join(names, "; "))
+	}
 
 	var err error
 	webhook, data, err = n.prepareWebhookDeliveryContext(webhook, data)
@@ -4020,6 +4106,7 @@ func (n *NotificationManager) Stop() {
 	n.stopOnce.Do(func() {
 		n.mu.Lock()
 		n.enabled = false
+		n.takePendingResolvedLocked()
 		queue := n.queue
 		cleanupDone := n.cleanupDone
 		client := n.webhookClient

internal/notifications/notifications_test.go

@@ -4044,6 +4044,7 @@ func TestSendResolvedAlert(t *testing.T) {
 	nm.AddWebhook(webhook)
 
 	nm.SetNotifyOnResolve(true)
+	nm.SetGroupingConfig(false, 0, false, false)
 
 	alert := &alerts.Alert{
 		ID:           "test-alert",

internal/notifications/queue.go

@@ -705,6 +705,12 @@ func normalizeNotificationLinks(
 
 // Enqueue adds a notification to the queue
 func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error {
+	nq.mu.Lock()
+	defer nq.mu.Unlock()
+	return nq.enqueueLocked(notif)
+}
+
+func (nq *NotificationQueue) enqueueLocked(notif *QueuedNotification) error {
 	if notif == nil {
 		return fmt.Errorf("notification cannot be nil")
 	}
@@ -780,9 +786,6 @@ func (nq *NotificationQueue) Enqueue(notif *QueuedNotification) error {
 		return fmt.Errorf("failed to marshal operational links: %w", err)
 	}
 
-	nq.mu.Lock()
-	defer nq.mu.Unlock()
-
 	query := `
 		INSERT INTO notification_queue
 		(id, type, method, status, alerts, operational_links, config, attempts, max_attempts, created_at, next_retry_at)

internal/notifications/resolved_grouping.go

@@ -0,0 +1,83 @@
+package notifications
+
+import (
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+)
+
+// enqueueResolvedGroup persists the window on the first resolution and appends
+// subsequent occurrences to that not-yet-due delivery. Only untouched pending
+// groups with identical destination configuration and quiet-hours schedule can
+// merge. Attempts/retries are immutable batches. The queue mutex also serializes
+// claims, so no delivery can be claimed while its payload is being extended.
+// Method is queue scheduling metadata, not the HTTP method in Config.
+func (nq *NotificationQueue) enqueueResolvedGroup(notif *QueuedNotification, window time.Duration) error {
+	if notif == nil || window <= 0 {
+		return nq.Enqueue(notif)
+	}
+	nq.mu.Lock()
+	defer nq.mu.Unlock()
+	now := time.Now()
+	method := "resolved-group"
+	if notif.NextRetryAt != nil {
+		method += ":" + notif.NextRetryAt.UTC().Format(time.RFC3339Nano)
+	}
+	var id string
+	var raw []byte
+	err := nq.db.QueryRow(`SELECT id, alerts FROM notification_queue
+  WHERE type = ? AND config = ? AND method = ? AND status = 'pending'
+   AND attempts = 0 AND next_retry_at > ? AND created_at >= ?
+  ORDER BY created_at, id LIMIT 1`, notif.Type, string(notif.Config), method, now.Unix(), now.Add(-window).Unix()).Scan(&id, &raw)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return fmt.Errorf("read pending resolved group: %w", err)
+	}
+	if errors.Is(err, sql.ErrNoRows) {
+		// Queue deadlines have second precision: round up rather than sending early.
+		deadline := now.Add(window).Truncate(time.Second).Add(time.Second)
+		if notif.NextRetryAt == nil || notif.NextRetryAt.Before(deadline) {
+			notif.NextRetryAt = &deadline
+		}
+		notif.Method = method
+		return nq.enqueueLocked(notif)
+	}
+	var batch []*alerts.Alert
+	if err := json.Unmarshal(raw, &batch); err != nil {
+		return fmt.Errorf("decode pending resolved group: %w", err)
+	}
+	for _, incoming := range notif.Alerts {
+		if incoming == nil {
+			continue
+		}
+		duplicate := false
+		for _, existing := range batch {
+			if existing != nil && existing.ID == incoming.ID && existing.StartTime.Equal(incoming.StartTime) {
+				duplicate = true
+				break
+			}
+		}
+		if !duplicate {
+			batch = append(batch, incoming)
+		}
+	}
+	links, err := normalizeNotificationLinks(id, notif.DestinationID, notificationLinksForAlerts(batch, notif.DestinationID), notificationDeliveryStateForQueueStatus(QueueStatusPending), nil, nil)
+	if err != nil {
+		return err
+	}
+	encoded, err := json.Marshal(batch)
+	if err != nil {
+		return err
+	}
+	encodedLinks, err := json.Marshal(links)
+	if err != nil {
+		return err
+	}
+	if _, err := nq.db.Exec(`UPDATE notification_queue SET alerts = ?, operational_links = ? WHERE id = ?`, string(encoded), string(encodedLinks), id); err != nil {
+		return fmt.Errorf("extend pending resolved group: %w", err)
+	}
+	return nil
+}

internal/notifications/resolved_grouping_contract_test.go

@@ -0,0 +1,245 @@
+package notifications
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
+)
+
+func waitResolvedContract(t *testing.T, predicate func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(12 * time.Second)
+	for !predicate() {
+		if time.Now().After(deadline) {
+			t.Fatal("notification condition not reached")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+// Ordinary firing establishes real receipts. Restart reloads those receipts;
+// no receipt or delivery job is manufactured by this test.
+func TestResolvedGroupingOrdinaryRestart(t *testing.T) {
+	for _, stage := range []string{"same_process", "after_firing", "pending_recovery"} {
+		t.Run(stage, func(t *testing.T) {
+			type payload struct {
+				Event  string          `json:"event"`
+				Alerts []*alerts.Alert `json:"alerts"`
+			}
+			received := make(chan payload, 64)
+			server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				var p payload
+				if err := json.NewDecoder(r.Body).Decode(&p); err != nil {
+					t.Error(err)
+				}
+				received <- p
+				w.WriteHeader(http.StatusOK)
+			}))
+			defer server.Close()
+			dir := t.TempDir()
+			hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL, Enabled: true}
+			open := func() *NotificationManager {
+				m := NewNotificationManagerWithDataDir("", dir)
+				m.webhookClient = server.Client()
+				if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+					t.Fatal(err)
+				}
+				m.AddWebhook(hook)
+				m.SetNotifyOnResolve(true)
+				m.SetGroupingConfig(true, 1, false, false)
+				t.Cleanup(m.Stop)
+				return m
+			}
+			m := open()
+			batch := make([]*alerts.Alert, 15)
+			for i := range batch {
+				batch[i] = &alerts.Alert{ID: fmt.Sprintf("burst-%02d", i), ResourceName: fmt.Sprintf("guest-%02d", i), Type: "cpu", Level: alerts.AlertLevelWarning, StartTime: time.Now().Add(-time.Minute)}
+				m.SendAlert(batch[i])
+			}
+			job := notificationDeliveryJob{Type: "webhook", Event: eventResolved, Alerts: batch, WebhookConfig: &hook}
+			waitResolvedContract(t, func() bool {
+				jobs := m.filterResolvedJobsByDeliveryReceipt([]notificationDeliveryJob{job})
+				return len(jobs) == 1 && len(jobs[0].Alerts) == 15
+			})
+			firing := <-received
+			if len(firing.Alerts) != 15 {
+				t.Fatalf("firing batch = %d", len(firing.Alerts))
+			}
+			if stage == "after_firing" {
+				m.Stop()
+				m = open()
+			}
+			// A newly configured destination must not receive recoveries for firings
+			// delivered only to ops, even when sharing its endpoint.
+			unseenHook := hook
+			unseenHook.ID = "never-delivered"
+			m.AddWebhook(unseenHook)
+			resolvedAt := time.Now().Truncate(time.Second)
+			for i, a := range batch {
+				if m.CancelAlert(a.ID) {
+					t.Fatal("delivered occurrence cancelled as unannounced")
+				}
+				m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: a, ResolvedTime: resolvedAt.Add(time.Duration(i) * time.Second)})
+			}
+			m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: &alerts.Alert{ID: "unannounced", ResourceName: "unannounced", StartTime: time.Now()}, ResolvedTime: resolvedAt})
+			if stage == "pending_recovery" {
+				m.Stop()
+				m = open()
+				m.AddWebhook(unseenHook)
+			}
+			select {
+			case p := <-received:
+				t.Fatalf("resolved delivery bypassed grouping window: event=%s alerts=%d", p.Event, len(p.Alerts))
+			case <-time.After(150 * time.Millisecond):
+			}
+			var recovery payload
+			select {
+			case recovery = <-received:
+			case <-time.After(5 * time.Second):
+				t.Fatal("no grouped recovery")
+			}
+			if recovery.Event != "resolved" || len(recovery.Alerts) != 15 {
+				t.Fatalf("recovery event=%s alerts=%d", recovery.Event, len(recovery.Alerts))
+			}
+			for i, a := range recovery.Alerts {
+				if a.ID != batch[i].ID || a.Metadata[metadataResolvedAt] != resolvedAt.Add(time.Duration(i)*time.Second).Format(time.RFC3339) {
+					t.Fatalf("lost occurrence/resolution metadata: %+v", a)
+				}
+			}
+			waitResolvedContract(t, func() bool { return len(m.filterResolvedJobsByDeliveryReceipt([]notificationDeliveryJob{job})) == 0 })
+			stats, err := m.GetQueue().GetQueueStats()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if stats["dlq"] != 0 || stats["failed"] != 0 {
+				t.Fatalf("unexpected delivery failures: %v", stats)
+			}
+			m.Stop()
+			select {
+			case p := <-received:
+				t.Fatalf("extra recovery: %+v", p)
+			default:
+			}
+		})
+	}
+}
+
+func TestResolvedGroupedWebhookNamesEveryAlert(t *testing.T) {
+	for _, service := range []string{"generic", "discord", "slack", "telegram", "teams", "teams-adaptive", "pushover", "gotify", "ntfy", "mattermost", "custom"} {
+		t.Run(service, func(t *testing.T) {
+			var body string
+			server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				b, _ := io.ReadAll(r.Body)
+				body = string(b)
+				w.WriteHeader(http.StatusOK)
+			}))
+			defer server.Close()
+			m := NewNotificationManagerWithDataDir("", t.TempDir())
+			defer m.Stop()
+			m.webhookClient = server.Client()
+			if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+				t.Fatal(err)
+			}
+			hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL + "?chat_id=1234", Enabled: true, Service: service}
+			if service == "custom" {
+				hook.Service = "generic"
+				hook.Template = `{"message":"{{.Message}}","count":{{.AlertCount}}}`
+			}
+			batch := contractGroupedAlerts()
+			if err := m.sendResolvedWebhook(hook, batch, time.Now()); err != nil {
+				t.Fatal(err)
+			}
+			for _, a := range batch {
+				if !strings.Contains(body, a.ResourceName) {
+					t.Errorf("%s missing %s: %s", service, a.ResourceName, body)
+				}
+			}
+		})
+	}
+}
+
+// Exercise HTTP payloads without a listener or external network. Persistent
+// queue operations remain real; this adapter replaces only provider transport.
+type resolvedContractTransport struct{ handler http.Handler }
+
+func (tr resolvedContractTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+	recorder := httptest.NewRecorder()
+	tr.handler.ServeHTTP(recorder, r)
+	return recorder.Result(), nil
+}
+
+type resolvedContractServer struct {
+	URL    string
+	client *http.Client
+}
+
+func newResolvedContractServer(handler http.Handler) *resolvedContractServer {
+	return &resolvedContractServer{URL: "http://127.0.0.1:12345", client: &http.Client{Transport: resolvedContractTransport{handler}}}
+}
+func (s *resolvedContractServer) Client() *http.Client { return s.client }
+func (s *resolvedContractServer) Close()               {}
+
+// Disabled grouping remains immediate. With the same 15-alert workload this
+// deliberately hits the unchanged internal 10/minute destination limit; grouped
+// recovery above sends one request instead and creates no dead letters.
+func TestResolvedGroupingDisabledBurstRetainsRateLimit(t *testing.T) {
+	server := newResolvedContractServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) }))
+	m := NewNotificationManagerWithDataDir("", t.TempDir())
+	defer m.Stop()
+	m.webhookClient = server.Client()
+	if err := m.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil {
+		t.Fatal(err)
+	}
+	hook := WebhookConfig{ID: "ops", Name: "ops", URL: server.URL, Enabled: true}
+	m.AddWebhook(hook)
+	m.SetNotifyOnResolve(true)
+	m.SetGroupingConfig(true, 1, false, false)
+	batch := make([]*alerts.Alert, 15)
+	for i := range batch {
+		batch[i] = &alerts.Alert{ID: fmt.Sprintf("limit-%d", i), ResourceName: fmt.Sprintf("limit-%d", i), StartTime: time.Now().Add(-time.Minute), Level: alerts.AlertLevelWarning}
+		m.SendAlert(batch[i])
+	}
+	waitResolvedContract(t, func() bool { stats, err := m.GetQueue().GetQueueStats(); return err == nil && stats["sent"] == 1 })
+	m.SetGroupingConfig(false, 1, false, false)
+	for _, a := range batch {
+		m.CancelAlert(a.ID)
+		m.SendResolvedAlert(&alerts.ResolvedAlert{Alert: a, ResolvedTime: time.Now()})
+	}
+	waitResolvedContract(t, func() bool {
+		stats, err := m.GetQueue().GetQueueStats()
+		return err == nil && stats["dlq"] == 6 && stats["sent"] == 10
+	})
+	dlq, err := m.GetQueue().GetDLQ(20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(dlq) != 6 {
+		t.Fatalf("dead letters=%d", len(dlq))
+	}
+	for _, job := range dlq {
+		if job.Attempts != 3 || (job.LastError == nil || !strings.Contains(*job.LastError, "rate limit exceeded")) {
+			t.Errorf("wrong retry outcome: %+v", job)
+		}
+	}
+}
+
+func TestResolvedGroupingPagerDutyKeepsIndividualKeys(t *testing.T) {
+	hook := WebhookConfig{ID: "pd", Service: "pagerduty", Enabled: true}
+	batch := contractGroupedAlerts()
+	jobs := buildNotificationDeliveryJobs(EmailConfig{}, []WebhookConfig{hook}, AppriseConfig{}, batch, eventResolved, time.Now())
+	if len(jobs) != len(batch) {
+		t.Fatalf("PagerDuty jobs=%d, want %d independent dedup keys", len(jobs), len(batch))
+	}
+	for i, job := range jobs {
+		if len(job.Alerts) != 1 || job.Alerts[0].ID != batch[i].ID {
+			t.Fatalf("wrong PagerDuty occurrence: %+v", job)
+		}
+	}
+}