feat: reconcile authoritative dead-letter send results atomically while preserving unknown outcomes and legacy accounting evidence

docs/content/docs/api/reference/deliverability-ops.mdx

@@ -292,13 +292,15 @@ Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns`
 
 | Parameter | In | Type | Description |
 |-----------|----|------|-------------|
-| `status` | query | string | Optional status filter (e.g. `pending`, `replayed`). |
+| `status` | query | string | Optional status filter (e.g. `pending`, `replayed`, `resolved`). |
 | `limit` | query | integer | Max rows to return, 1 to 200 (default 100). |
 
 ### Response
 
 Returns a `data` array of dead-letter records. This endpoint is not cursor-paginated; it returns up to `limit` rows in one response.
 
+Send dead letters become `resolved` only after an authoritative worker or operator-confirmed send outcome is reconciled. Recovery does not resend mail: the original payload, error and attempts stay available for inspection. Unknown outcomes remain pending and held. A missing Sent-folder copy, reservation marker or executor marker is not proof that nothing was sent.
+
 ```json
 {
   "data": [

internal/app/consumer/event_send_result.go

@@ -186,12 +186,12 @@ func (s *JobsService) applyEmailFailed(ctx context.Context, result models.SendEm
 		}
 	}
 	switch task.Status {
-	case "completed":
+	case "completed", "dead_lettered":
 		// The normal case: stamped by the control plane, refused by the worker.
 	case "active":
 		return errSendResultEarly
 	default:
-		// Already walked back (duplicate delivery), cancelled, or dead-lettered.
+		// Already walked back (duplicate delivery) or cancelled.
 		return nil
 	}
 

internal/app/consumer/send_legacy_recovery_live_test.go

@@ -1,11 +1,15 @@
 package jobs
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/iskorotkov/avro/v2"
+	"github.com/warmbly/warmbly/internal/errx"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
 )
@@ -63,6 +67,257 @@ func legacySendResultSchema(t *testing.T) (avro.Schema, avro.Schema) {
 	return writer, resolved
 }
 
+func TestLiveLegacyFailedStatusCannotProveNonExecution(t *testing.T) {
+	for _, status := range []string{"failed", "dead_lettered"} {
+		t.Run(status, func(t *testing.T) {
+			h := liveDB(t)
+			f := newSendResultFixture(t, h)
+			s := liveJobsService(h)
+			id := f.stampSend(t, s)
+			if _, err := h.Exec(t.Context(), `UPDATE tasks SET status=$2::task_status,send_result_state='unknown' WHERE id=$1`, id, status); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := h.Exec(t.Context(), `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.mailbox, id); err != nil {
+				t.Fatal(err)
+			}
+			reader := s.TaskRepo.(interface {
+				ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error)
+			})
+			results, err := reader.ListUnappliedSendResults(t.Context(), 200)
+			if err != nil {
+				t.Fatal(err)
+			}
+			for _, result := range results {
+				if result.TaskID == id {
+					t.Fatal("task status and missing modern markers manufactured provider failure evidence")
+				}
+			}
+			ambiguous := models.SendEmailResult{TaskID: id, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendAmbiguous}}}
+			if err := s.HandleEmailFailed(t.Context(), ambiguous); err != nil {
+				t.Fatal(err)
+			}
+			var state string
+			var applied, released, hold bool
+			if err := h.QueryRow(t.Context(), `SELECT t.send_result_state,t.send_result_applied_at IS NOT NULL,t.send_released_at IS NOT NULL,e.send_recovery_hold FROM tasks t JOIN email_accounts e ON e.id=t.email_account_id WHERE t.id=$1`, id).Scan(&state, &applied, &released, &hold); err != nil {
+				t.Fatal(err)
+			}
+			if state != "unknown" || applied || released || !hold {
+				t.Fatalf("legacy outcome guessed: state=%s applied=%t released=%t hold=%t", state, applied, released, hold)
+			}
+		})
+	}
+}
+
+func TestLiveAuthoritativeDeadLetterResultsReconcileWithoutReplay(t *testing.T) {
+	for _, outcome := range []string{"sent", "failed", "ambiguous", "previously applied sent", "previously applied failed", "unstamped failure", "legacy failure"} {
+		t.Run(outcome, func(t *testing.T) {
+			s, h := liveWarmupService(t)
+			f := newWarmupFixture(t, h)
+			token := f.mintToken(t, s.WarmupRepo)
+			var day time.Time
+			if err := h.QueryRow(t.Context(), `SELECT completed_at FROM tasks WHERE id=$1`, f.task).Scan(&day); err != nil {
+				t.Fatal(err)
+			}
+			if err := s.WarmupRepo.IncrementDailyCount(t.Context(), f.sender, day); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := h.Exec(t.Context(), `UPDATE tasks SET status='dead_lettered',send_result_state='unknown' WHERE id=$1`, f.task); err != nil {
+				t.Fatal(err)
+			}
+			if outcome != "legacy failure" {
+				if _, err := h.Exec(t.Context(), `UPDATE tasks SET send_reserved_at=completed_at,send_executor_nonce=$2 WHERE id=$1`, f.task, uuid.New()); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type,payload,last_error,attempts,next_retry_at) VALUES($1,'warmup','{"task_id":"preserved"}','legacy dispatch error',1,NOW())`, f.task); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := h.Exec(t.Context(), `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task); err != nil {
+				t.Fatal(err)
+			}
+			if outcome == "unstamped failure" {
+				if _, err := h.Exec(t.Context(), `UPDATE tasks SET completed_at=NULL WHERE id=$1`, f.task); err != nil {
+					t.Fatal(err)
+				}
+			}
+			result := models.SendEmailResult{TaskID: f.task, Success: outcome == "sent" || outcome == "previously applied sent", MessageID: f.sentMsgID, SentAt: day}
+			if !result.Success {
+				disposition := errx.SendRetry
+				if outcome == "ambiguous" {
+					disposition = errx.SendAmbiguous
+				}
+				result.MessageID = ""
+				result.Error = &models.EmailSendError{Code: "SERVER_UNREACHABLE", Failure: &errx.SendFailure{Disposition: disposition, Stage: "dial", Protocol: "smtp", Scope: "mailbox"}}
+			}
+			if outcome == "previously applied sent" || outcome == "previously applied failed" {
+				raw, err := json.Marshal(result)
+				if err != nil {
+					t.Fatal(err)
+				}
+				state := "failed"
+				if result.Success {
+					state = "sent"
+				}
+				if _, err := h.Exec(t.Context(), `UPDATE tasks SET send_result_state=$2,send_result_applied_at=NOW(),send_executor_result=$3 WHERE id=$1`, f.task, state, raw); err != nil {
+					t.Fatal(err)
+				}
+				s.reconcileRecordedSendResults(t.Context())
+			}
+			for range 2 {
+				var err error
+				if result.Success {
+					err = s.HandleEmailSent(t.Context(), result)
+				} else {
+					err = s.HandleEmailFailed(t.Context(), result)
+				}
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			var status, state, dlq, message, lastError string
+			var applied, released, hold, replayed, evidenceKept bool
+			var count, attempts int
+			if err := h.QueryRow(t.Context(), `SELECT t.status,t.send_result_state,t.send_result_applied_at IS NOT NULL,t.send_released_at IS NOT NULL,e.send_recovery_hold,d.status,d.replayed_at IS NOT NULL,d.payload='{"task_id":"preserved"}'::jsonb,d.last_error,d.attempts,wt.sent_message_id,ws.emails_sent FROM tasks t JOIN email_accounts e ON e.id=t.email_account_id JOIN task_dead_letters d ON d.task_id=t.id JOIN warmup_tokens wt ON wt.task_id=t.id JOIN warmup_statistics ws ON ws.email_account_id=e.id AND ws.date=DATE($3::timestamptz) WHERE t.id=$1 AND wt.token=$2`, f.task, token, day).Scan(&status, &state, &applied, &released, &hold, &dlq, &replayed, &evidenceKept, &lastError, &attempts, &message, &count); err != nil {
+				t.Fatal(err)
+			}
+			if replayed || !evidenceKept || lastError != "legacy dispatch error" || attempts != 1 {
+				t.Fatal("recovery replayed or erased original dead-letter evidence")
+			}
+			switch outcome {
+			case "sent", "previously applied sent":
+				if status != "completed" || state != "sent" || !applied || released || hold || dlq != "resolved" || message != f.sentMsgID || count != 1 {
+					t.Fatalf("sent recovery incomplete: status=%s state=%s applied=%t released=%t hold=%t dlq=%s count=%d", status, state, applied, released, hold, dlq, count)
+				}
+			case "failed", "unstamped failure", "legacy failure":
+				wantCount := 0
+				if outcome == "unstamped failure" || outcome == "legacy failure" {
+					wantCount = 1
+				}
+				if status != "failed" || state != "failed" || !applied || !released || hold || dlq != "resolved" || message != "" || count != wantCount {
+					t.Fatalf("failure recovery incomplete: status=%s state=%s applied=%t released=%t hold=%t dlq=%s count=%d", status, state, applied, released, hold, dlq, count)
+				}
+			case "previously applied failed":
+				if status != "dead_lettered" || state != "failed" || !applied || released || !hold || dlq != "pending" || message != "" || count != 1 {
+					t.Fatal("already-applied failure refunded without accounting provenance")
+				}
+			default:
+				if status != "dead_lettered" || state != "unknown" || applied || released || !hold || dlq != "pending" || message != "" || count != 1 {
+					t.Fatal("ambiguous dead letter was guessed resolved")
+				}
+			}
+		})
+	}
+}
+
+func TestLiveDeadLetterResultEffectsAndSettlementRollBackTogether(t *testing.T) {
+	s, h := liveWarmupService(t)
+	f := newWarmupFixture(t, h)
+	f.mintToken(t, s.WarmupRepo)
+	if _, err := h.Exec(t.Context(), `UPDATE tasks SET status='dead_lettered',send_result_state='unknown' WHERE id=$1`, f.task); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type) VALUES($1,'warmup')`, f.task); err != nil {
+		t.Fatal(err)
+	}
+	want := errors.New("effect failed after task transition")
+	result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: f.sentMsgID}
+	recovery := s.TaskRepo.(repository.SendResultRecovery)
+	err := recovery.ApplySendResult(t.Context(), result, func(ctx context.Context) error {
+		if err := s.applyEmailSent(ctx, result); err != nil {
+			return err
+		}
+		return want
+	})
+	if !errors.Is(err, want) {
+		t.Fatal(err)
+	}
+	var status, state, dlq, message string
+	var applied bool
+	if err := h.QueryRow(t.Context(), `SELECT t.status,t.send_result_state,t.send_result_applied_at IS NOT NULL,d.status,wt.sent_message_id FROM tasks t JOIN task_dead_letters d ON d.task_id=t.id JOIN warmup_tokens wt ON wt.task_id=t.id WHERE t.id=$1`, f.task).Scan(&status, &state, &applied, &dlq, &message); err != nil {
+		t.Fatal(err)
+	}
+	if status != "dead_lettered" || state != "unknown" || applied || dlq != "pending" || message != "" {
+		t.Fatal("partial result effects escaped rollback")
+	}
+}
+
+func TestLiveCampaignDeadLetterWorkerResultReleasesReservationOnce(t *testing.T) {
+	h := liveDB(t)
+	f := newSendResultFixture(t, h)
+	s := liveJobsService(h)
+	id := f.stampSend(t, s)
+	if _, err := h.Exec(t.Context(), `UPDATE tasks SET status='dead_lettered',send_result_state='unknown' WHERE id=$1`, id); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type) VALUES($1,'campaign')`, id); err != nil {
+		t.Fatal(err)
+	}
+	result := models.SendEmailResult{TaskID: id, SentAt: time.Now().UTC(), Error: &models.EmailSendError{Code: "SERVER_UNREACHABLE", Failure: &errx.SendFailure{Protocol: "smtp", Stage: "dial", Disposition: errx.SendRetry, Scope: "mailbox"}}}
+	raw, err := json.Marshal(result)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := h.Exec(t.Context(), `UPDATE tasks SET send_executor_result=$2 WHERE id=$1`, id, raw); err != nil {
+		t.Fatal(err)
+	}
+	for range 2 {
+		s.reconcileRecordedSendResults(t.Context())
+		if err := s.HandleEmailFailed(t.Context(), result); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var status, dlq string
+	var sent, reserved, applied bool
+	var attempts int
+	if err := h.QueryRow(t.Context(), `SELECT t.status,d.status,p.sent_at IS NOT NULL,p.dispatched_at IS NOT NULL,t.send_result_applied_at IS NOT NULL,p.send_attempts FROM tasks t JOIN task_dead_letters d ON d.task_id=t.id JOIN campaign_contact_progress p ON p.campaign_id=$2 AND p.contact_id=$3 AND p.sequence_id=$4 WHERE t.id=$1`, id, f.campaign, f.contact, f.step).Scan(&status, &dlq, &sent, &reserved, &applied, &attempts); err != nil {
+		t.Fatal(err)
+	}
+	if status != "failed" || dlq != "resolved" || sent || reserved || !applied || attempts != 1 {
+		t.Fatalf("definitive campaign refusal not reconciled once: status=%s dlq=%s sent=%t reserved=%t applied=%t attempts=%d", status, dlq, sent, reserved, applied, attempts)
+	}
+}
+
+func TestLiveUserEmailDeadLetterUsesUnderlyingSendTaskType(t *testing.T) {
+	h := liveDB(t)
+	f := newSendResultFixture(t, h)
+	s := liveJobsService(h)
+	id := uuid.New()
+	if _, err := h.Exec(t.Context(), `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','dead_lettered','')`, id, f.mailbox); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := h.Exec(t.Context(), `INSERT INTO task_dead_letters(task_id,task_type) VALUES($1,'user_email')`, id); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.HandleEmailSent(t.Context(), models.SendEmailResult{TaskID: id, MessageID: "confirmed@example.test"}); err != nil {
+		t.Fatal(err)
+	}
+	var status string
+	if err := h.QueryRow(t.Context(), `SELECT status FROM task_dead_letters WHERE task_id=$1`, id).Scan(&status); err != nil || status != "resolved" {
+		t.Fatal("user_email dead letter was not settled", status, err)
+	}
+}
+
+func TestLiveCampaignForeignKeyCleanupDoesNotProveWakeupOnly(t *testing.T) {
+	h := liveDB(t)
+	f := newSendResultFixture(t, h)
+	s := liveJobsService(h)
+	id := f.stampSend(t, s)
+	before, err := s.TaskRepo.GetCampaignTask(t.Context(), id)
+	if err != nil || before == nil || before.ContactID == nil || before.SequenceID == nil {
+		t.Fatal("send-capable campaign fixture missing", err)
+	}
+	if _, err := h.Exec(t.Context(), `DELETE FROM contacts WHERE id=$1`, f.contact); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := h.Exec(t.Context(), `DELETE FROM sequences WHERE id=$1`, *before.SequenceID); err != nil {
+		t.Fatal(err)
+	}
+	after, err := s.TaskRepo.GetCampaignTask(t.Context(), id)
+	if err != nil || after == nil || after.ContactID != nil || after.SequenceID != nil || after.CampaignID == nil {
+		t.Fatal("FK-null former send no longer matches the unsafe wakeup-only predicate", err)
+	}
+}
+
 func TestLiveLegacyJSONAndAvroFailurePersistConservativeAdmission(t *testing.T) {
 	writer, reader := legacySendResultSchema(t)
 	for _, format := range []string{"json", "avro"} {

internal/app/consumer/send_recovery_resolution_live_test.go

@@ -113,9 +113,9 @@ func TestLiveRecordedWorkerOutcomeSurvivesLostEvent(t *testing.T) {
 	}
 }
 
-func TestLiveUnreservedFailedTaskCannotHoldMailbox(t *testing.T) {
+func TestLiveUnreservedLegacyFailedTaskRetainsUnknownHold(t *testing.T) {
 	for _, existing := range []bool{false, true} {
-		t.Run(map[bool]string{false: "new failure", true: "existing invalid hold"}[existing], func(t *testing.T) {
+		t.Run(map[bool]string{false: "new ambiguous result", true: "existing unknown hold"}[existing], func(t *testing.T) {
 			h := liveDB(t)
 			f := newSendResultFixture(t, h)
 			s := liveJobsService(h)
@@ -140,8 +140,8 @@ func TestLiveUnreservedFailedTaskCannotHoldMailbox(t *testing.T) {
 			if err := h.Pool.QueryRow(ctx, `SELECT ea.send_recovery_hold,t.send_result_applied_at IS NOT NULL,t.send_result_state FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, id).Scan(&hold, &applied, &state); err != nil {
 				t.Fatal(err)
 			}
-			if hold || !applied || state != "failed" {
-				t.Fatal("unreserved failure held mailbox", hold, applied, state)
+			if !hold || applied || state != "unknown" {
+				t.Fatal("missing authority markers resolved a legacy unknown", hold, applied, state)
 			}
 		})
 	}

internal/repository/pg_send_recovery.go

@@ -163,11 +163,8 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 	var applied *time.Time
 	var mailboxID uuid.UUID
 	var status string
-	var reserved, warmupAuthorized bool
-	err = tx.QueryRow(ctx, `SELECT send_result_state, send_result_applied_at, email_account_id, status::text,
-	 send_reserved_at IS NOT NULL OR send_executor_nonce IS NOT NULL OR send_executor_started_at IS NOT NULL OR send_executor_result IS NOT NULL,
-	 EXISTS(SELECT 1 FROM warmup_tasks WHERE task_id=tasks.id AND dispatch_nonce IS NOT NULL)
-	 FROM tasks WHERE id = $1`, result.TaskID).Scan(&previous, &applied, &mailboxID, &status, &reserved, &warmupAuthorized)
+	err = tx.QueryRow(ctx, `SELECT send_result_state, send_result_applied_at, email_account_id, status::text
+	 FROM tasks WHERE id = $1`, result.TaskID).Scan(&previous, &applied, &mailboxID, &status)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return nil
 	}
@@ -177,12 +174,10 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 	if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 269))`, mailboxID.String()); err != nil {
 		return err
 	}
-	// A pre-dispatch failure cannot leave ambiguous provider execution behind.
-	if resultState(result) == "unknown" && status == "failed" && !reserved && !warmupAuthorized {
-		result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Protocol: "internal", Stage: "prepare", Scope: "mailbox", Disposition: errx.SendRetry}}
-	}
-	if applied != nil || (previous != nil && *previous != "unknown") {
-		state := resultState(result)
+	state := resultState(result)
+	// Receipt stamping is idempotent; an already-applied failure cannot be refunded again.
+	repairDeadLetter := status == "dead_lettered" && previous != nil && *previous == state && state == "sent"
+	if !repairDeadLetter && (applied != nil || (previous != nil && *previous != "unknown")) {
 		if previous != nil && state != "unknown" && state != *previous {
 			if _, err = tx.Exec(ctx, `UPDATE send_recovery_resolutions SET conflict_detected_at=NOW()
 				WHERE recovery_task_id=$1 AND previous_reason='unknown' AND confirmed_result IS NOT NULL`, result.TaskID); err != nil {
@@ -195,12 +190,17 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 			}
 			return tx.Commit(ctx)
 		}
+		if applied != nil && status != "dead_lettered" && previous != nil && state == *previous && state != "unknown" {
+			if err = resolveSendDeadLetter(ctx, tx, result.TaskID); err != nil {
+				return err
+			}
+			return tx.Commit(ctx)
+		}
 		return nil
 	}
 	if status == "cancelled" {
 		return nil
 	}
-	state := resultState(result)
 	if state == "failed" && status == "active" {
 		return errors.New("send result arrived before task stamping")
 	}
@@ -217,6 +217,15 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 	var effects []func(context.Context)
 	var effectError error
 	if state != "unknown" {
+		if status == "dead_lettered" && state == "sent" {
+			sentAt := result.SentAt
+			if sentAt.IsZero() {
+				sentAt = time.Now().UTC()
+			}
+			if _, err = tx.Exec(ctx, `UPDATE tasks SET status='completed',completed_at=COALESCE(completed_at,$2),updated_at=NOW() WHERE id=$1`, result.TaskID, sentAt); err != nil {
+				return err
+			}
+		}
 		inner := context.WithValue(ctx, sendResultKey{}, sendResultContext{tx: tx, taskID: result.TaskID, effects: &effects, effectError: &effectError})
 		if err = apply(inner); err != nil {
 			return err
@@ -227,6 +236,9 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 		if _, err = tx.Exec(ctx, `UPDATE tasks SET send_result_applied_at = NOW(), send_released_at=CASE WHEN send_result_state='failed' THEN NOW() ELSE send_released_at END WHERE id = $1`, result.TaskID); err != nil {
 			return err
 		}
+		if err = resolveSendDeadLetter(ctx, tx, result.TaskID); err != nil {
+			return err
+		}
 	}
 	if err = persistSendHold(ctx, tx, mailboxID, result, state); err != nil {
 		return err
@@ -240,6 +252,13 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 	return nil
 }
 
+func resolveSendDeadLetter(ctx context.Context, tx pgx.Tx, taskID uuid.UUID) error {
+	_, err := tx.Exec(ctx, `UPDATE task_dead_letters SET status='resolved',next_retry_at=NULL,updated_at=NOW()
+		WHERE task_id=$1 AND status IN ('pending','failed','replayed')
+		AND EXISTS(SELECT 1 FROM tasks t WHERE t.id=$1 AND t.task_type IN ('warmup','campaign','email','placement'))`, taskID)
+	return err
+}
+
 func persistSendHold(ctx context.Context, tx pgx.Tx, mailboxID uuid.UUID, result models.SendEmailResult, state string) error {
 	if state != "unknown" {
 		_, err := tx.Exec(ctx, `WITH next_hold AS (

internal/repository/pg_warmup.go

@@ -1094,23 +1094,32 @@ func (r *warmupRepository) FailWarmupSend(ctx context.Context, accountID, taskID
 	defer tx.Rollback(ctx)
 
 	var status string
+	var completedAt *time.Time
+	var dispatchAuthorized bool
 	err = tx.QueryRow(ctx, `
-		SELECT status::text
+		SELECT status::text, completed_at,
+		 send_reserved_at IS NOT NULL OR send_executor_nonce IS NOT NULL
+		 OR EXISTS(SELECT 1 FROM warmup_tasks WHERE task_id=tasks.id AND dispatch_nonce IS NOT NULL)
 		FROM tasks
 		WHERE id = $1 AND email_account_id = $2 AND task_type = 'warmup'
 		FOR UPDATE
-	`, taskID, accountID).Scan(&status)
+	`, taskID, accountID).Scan(&status, &completedAt, &dispatchAuthorized)
 	if errors.Is(err, pgx.ErrNoRows) {
 		return nil
 	}
 	if err != nil {
 		return err
 	}
-	if status != "completed" {
+	if status != "completed" && status != "dead_lettered" {
 		return tx.Commit(ctx)
 	}
 
-	if _, err = tx.Exec(ctx, `
+	// Legacy dead-letter accounting cannot be reconstructed from status alone.
+	if status == "completed" || (dispatchAuthorized && completedAt != nil) {
+		if completedAt != nil {
+			date = *completedAt
+		}
+		if _, err = tx.Exec(ctx, `
 		UPDATE warmup_statistics ws
 		SET emails_sent = GREATEST(ws.emails_sent - 1, 0),
 		    emails_replied = CASE
@@ -1122,8 +1131,9 @@ func (r *warmupRepository) FailWarmupSend(ctx context.Context, accountID, taskID
 		    END
 		WHERE ws.email_account_id = $1
 		  AND ws.date = DATE($3)
-	`, accountID, taskID, date); err != nil {
-		return err
+		`, accountID, taskID, date); err != nil {
+			return err
+		}
 	}
 	if _, err = tx.Exec(ctx, `UPDATE tasks SET status = 'failed', updated_at = NOW() WHERE id = $1`, taskID); err != nil {
 		return err

internal/repository/unknown_send_resolution.go

@@ -91,33 +91,31 @@ func queueUnknownSendResolution(ctx context.Context, tx pgx.Tx, org, mailbox str
 func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int) ([]models.SendEmailResult, error) {
 	rows, err := r.db.Query(ctx, `WITH recorded AS (
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,t.send_executor_result AS result,1 AS priority
-	 FROM tasks t WHERE t.send_executor_result IS NOT NULL AND t.status<>'cancelled' AND t.send_result_applied_at IS NULL
+	 FROM tasks t WHERE t.send_executor_result IS NOT NULL AND t.status<>'cancelled'
+	 AND (t.send_result_applied_at IS NULL OR (t.status='dead_lettered' AND t.send_result_state='sent'))
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,t.send_executor_result,1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
 	 WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL
 	 AND t.send_executor_result IS NOT NULL AND t.send_result_applied_at IS NOT NULL AND t.status<>'cancelled'
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,w.dispatch_result,1 FROM warmup_tasks w JOIN tasks t ON t.id=w.task_id
-	 WHERE w.dispatch_result IS NOT NULL AND t.status<>'cancelled' AND t.send_result_applied_at IS NULL
+	 WHERE w.dispatch_result IS NOT NULL AND t.status<>'cancelled'
+	 AND (t.send_result_applied_at IS NULL OR (t.status='dead_lettered' AND t.send_result_state='sent'))
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,w.dispatch_result,1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id JOIN warmup_tasks w ON w.task_id=t.id
 	 WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL
 	 AND w.dispatch_result IS NOT NULL AND t.send_result_applied_at IS NOT NULL AND t.status<>'cancelled'
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,r.confirmed_result,2 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
-	 WHERE r.confirmed_result IS NOT NULL AND t.send_result_applied_at IS NULL AND t.status<>'cancelled'
-	 UNION ALL
-	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,jsonb_build_object('task_id',t.id,'success',false,'legacy_error','Send failed before reservation',
-	 'error',jsonb_build_object('failure',jsonb_build_object('protocol','internal','stage','prepare','scope','mailbox','disposition','retry'))),3
-	 FROM tasks t WHERE t.status='failed' AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL
-	 AND t.send_reserved_at IS NULL AND t.send_executor_nonce IS NULL AND t.send_executor_started_at IS NULL
-	 AND NOT EXISTS(SELECT 1 FROM warmup_tasks w WHERE w.task_id=t.id AND w.dispatch_nonce IS NOT NULL)),
+	 WHERE r.confirmed_result IS NOT NULL AND (t.send_result_applied_at IS NULL OR (t.status='dead_lettered' AND t.send_result_state='sent')) AND t.status<>'cancelled'
+	 ),
 	 definitive AS (SELECT *,CASE WHEN result->>'success'='true' THEN 'sent' ELSE 'failed' END AS outcome FROM recorded
 	 WHERE result->>'success'='true' OR result->'error'->'failure'->>'disposition' IN('retry','throttle','permanent','authentication')
 	 OR (result->'error'->>'failure' IS NULL AND result->'error'->>'code' IN
 	 ('RECIPIENT_REJECTED','SEND_REJECTED','DOMAIN_AUTH_REJECTED','AUTHENTICATION_FAILED','INVALID_CREDENTIALS','GOOGLE_AUTHENTICATION_FAILED','SENDING_TOO_FAST','QUOTA_EXCEEDED','UNSUPPORTED'))),
 	 pending AS (SELECT DISTINCT ON(id) id,created_at,result FROM definitive
-	 WHERE send_result_applied_at IS NULL OR outcome IS DISTINCT FROM send_result_state ORDER BY id,priority)
+	 WHERE send_result_applied_at IS NULL OR outcome IS DISTINCT FROM send_result_state
+	 OR EXISTS(SELECT 1 FROM tasks t WHERE t.id=definitive.id AND t.status='dead_lettered' AND t.send_result_state='sent') ORDER BY id,priority)
 	 SELECT id,result FROM pending ORDER BY created_at LIMIT $1`, limit)
 	if err != nil {
 		return nil, err