Merge pull request #930 from warmbly/devin/1791639319-protect-legacy-dlq-replay

feat: protect legacy send dead letters and expose unverified recovery evidence

cmd/warmblyctl/remote_output.go

@@ -102,7 +102,7 @@ arrival_aging arrival_due arrival_leased arrival_pending arrival_stage_0 arrival
 arrival_stage_2 arrival_stage_3 arrival_wait backfill_complete backfill_pending
 backfill_running backfill_stale_running campaign_reserved campaign_reserved_aging
 confirmed_other_1h confirmed_warmup_1h confirmed_warmup_proxy_1h credential_errors
-dispatch_due dispatch_old_failure dispatch_persistent dlq_due dlq_exhausted dlq_replayed
+dispatch_due dispatch_old_failure dispatch_persistent dlq_due dlq_exhausted dlq_replayed dlq_unverified
 dlq_unscheduled dlq_wait effects_leased effects_pending effects_repeated effects_unleased
 failed_results_1h failed_results_24h future_schedule holds_authentication holds_conflict warmup_invalid_window
 holds_permanent holds_unknown inactive_credential_errors job_error_old job_error_recent

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

@@ -324,7 +324,9 @@ Returns a `data` array of dead-letter records. This endpoint is not cursor-pagin
 
 `POST /tasks/dlq/:id/replay`
 
-Re-dispatches a dead-lettered task. Because a replay can transmit real mail, this requires the send permission rather than plain write access.
+Replays only a dead-lettered campaign wakeup with explicit durable wakeup intent, rechecked under its campaign lock. Once a task binds a contact or sequence for sending, its send intent stays restrictive even after either record is deleted. Legacy or imported tasks without verified wakeup intent and all send-capable tasks return `409 Conflict` and retain their dead-letter evidence. A retry date, JSON payload, null contact/sequence, or missing send marker cannot grant replay. This route requires the send permission rather than plain write access.
+
+Database downgrades refuse to discard this provenance while send-intent or unresolved unverified dead-letter rows exist. Keep the schema and roll forward instead.
 
 Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns`
 

docs/content/docs/development/admin-panel.mdx

@@ -71,7 +71,7 @@ Relative ages use your browser's current clock when a snapshot arrives, with per
 Mailbox diagnostic participation, sender/recipient flags and shared/rolling limits are organization-scoped mailbox fields, not new admin-panel switches. Use the existing authorized mailbox PATCH; no new UI is implied. An operator-reviewed `send_recovery_resolution` binds the current held task/reason and definite repair/provider evidence, never an unknown send. See [send safety](/development/send-safety/).
 | **Seed panel** | The instance's placement seed inboxes by provider, adding and removing them, every workspace's recent [placement tests](/guides/placement-tests/) with their results, and an operator test that counts against no allowance |
 | **Campaigns** | Every campaign on the instance with its state and owner |
-| **Sends** | In-flight reservations no worker has answered, and the reclaimer that resolves them after 30 minutes; dead letters with replay; task failures; customer webhook delivery health, with reclaim |
+| **Sends** | In-flight reservations no worker has answered, and the reclaimer that resolves them after 30 minutes; dead letters with replay limited to non-sending campaign wakeups; task failures; customer webhook delivery health, with reclaim |
 
 ## Accounts
 

docs/content/docs/development/instance-health.mdx

@@ -67,7 +67,7 @@ Two neighbouring endpoints round out the surface, both under **Instance** in the
 
 ## Operational monitoring
 
-**Operations** renders the version `"1"` snapshot from `GET /admin/instance/monitoring`. It covers registered workers, worker sample receipt, confirmed sends, campaign reservations, dispatch, safety holds, sync state and fair-use waits, stored mailbox errors, jobs, dead letters, webhooks, customer notification rows, send-result effects, domain checks, warmup mailbox loading and broker queue observations. Unsupported sources remain visible as unavailable, including detailed send-wait attribution and external tracking-broker diagnostics. Optional arrival-outbox measurements are unavailable when their schema is absent; monitoring needs no arrival migration and does not require a new worker wire version.
+**Operations** renders the version `"1"` snapshot from `GET /admin/instance/monitoring`. It covers registered workers, worker sample receipt, confirmed sends, campaign reservations, dispatch, safety holds, sync state and fair-use waits, stored mailbox errors, jobs, dead letters, webhooks, customer notification rows, send-result effects, domain checks, warmup mailbox loading and broker queue observations. Dead letters without safe replay proof remain visible as recovery-unverified even when a retry date has passed; the due, waiting and unscheduled counts only cover non-sending campaign wakeup passes. The consumer does not have a task scheduler for automatic replay. Unsupported sources remain visible as unavailable, including detailed send-wait attribution and external tracking-broker diagnostics. Optional arrival-outbox measurements are unavailable when their schema is absent; monitoring needs no arrival migration and does not require a new worker wire version.
 
 The backend collects with an eight-second overall timeout, three seconds per source and at most four concurrent collectors. It shares a one-minute snapshot across admins, then filters source permissions for each response. HTTP responses are not cacheable. On collection failure it can retain a source's last good measurement for at most five minutes, marked stale and with its original observation/evidence clocks. After that it is unavailable. Source samples are bounded; measured/expected scope counts and incomplete coverage expose truncation. Counts in partial samples may be lower bounds, never extrapolated fleet totals.
 

docs/content/docs/development/warmblyctl.mdx

@@ -172,7 +172,7 @@ warmblyctl admin recovery restart-backfill --id <mailbox-uuid> --server laptop \
 
 Substitute the same real canonical UUID in both places. Clearing a provider throttle can resume provider work; restarting backfill resets the existing sync backfill state. Neither action establishes sending authority.
 
-Dead-letter replay is never part of diagnosis. Inspect the task, mailbox, provider receipts and uncertain-send protections first. Replay additionally requires a separate risk acknowledgement:
+Dead-letter replay is never part of diagnosis. Only non-sending campaign wakeup passes can be replayed; send-capable dead letters remain pending until their outcome is independently reconciled through normal recovery. Inspect the task, mailbox, provider receipts and uncertain-send protections first. Replay additionally requires a separate risk acknowledgement:
 
 ```bash
 warmblyctl admin recovery replay --id <dead-letter-uuid> --server laptop \

docs/content/docs/guides/workspace-export-import.mdx

@@ -57,6 +57,8 @@ A workspace archive is a single file holding everything one workspace owns. It e
 
 Diagnostic participation, shared/rolling limits, restrictive send evidence and audited repair-resolution history travel with their registered mailbox/task/organization data. Source-worker executor handles reset; exact provider-message DKIM authorization/nonces are excluded. Portable minimal receipt proof remains historical evidence, never an execution grant. The reconciliation webhook/notification outbox is source-instance-local and excluded to prevent replay on import; drain pending effects before moving. Whole-instance database backups retain it. See [send-safety retention and rollback](/development/send-safety/#upgrade-rollback-and-portability).
 
+Campaign task replay intent resets to unverified on workspace import. The destination cannot replay an imported dead letter as a verified non-sending wakeup. A contact or sequence binding remains restrictive send intent even if that record is later removed. Whole-instance backups retain the durable intent; migration rollback refuses to discard retained send intent or unverified pending campaign dead letters.
+
 <Callout title="Moving a whole self-hosted instance is a different tool">
 This page moves **one workspace** between two running instances, re-sealing its secrets for the destination's keys. To move an entire self-hosted install (every workspace, its users, its platform admins) use `warmblyctl backup` and `warmblyctl restore`, which carry the database, the blob root and the encryption keys as one bundle. See [data control](/development/data-control/#backups). The two are not interchangeable: a bundle cannot be applied to a single workspace, and this archive cannot restore an instance.
 </Callout>

internal/app/advanced/campaign_replay_test.go

@@ -19,6 +19,10 @@ type replayRepo struct {
 	replayed int
 }
 
+func (r *replayRepo) ListRetryableDeadLetters(context.Context, int) ([]models.TaskDeadLetter, error) {
+	return []models.TaskDeadLetter{*r.dlq}, nil
+}
+
 func (r *replayRepo) GetTaskDeadLetter(context.Context, uuid.UUID, uuid.UUID) (*models.TaskDeadLetter, error) {
 	return r.dlq, nil
 }
@@ -32,6 +36,10 @@ type replayTasks struct {
 	repository.TaskRepository
 	task       *repository.Task
 	campaignID *uuid.UUID
+	contactID  *uuid.UUID
+	sequenceID *uuid.UUID
+	intent     repository.CampaignDispatchIntent
+	replayErr  error
 	ctErr      error
 	created    bool
 	inserted   int
@@ -52,7 +60,46 @@ func (t *replayTasks) GetCampaignTask(context.Context, uuid.UUID) (*repository.C
 	if t.ctErr != nil {
 		return nil, t.ctErr
 	}
-	return &repository.CampaignTask{TaskID: t.task.ID, CampaignID: t.campaignID}, nil
+	return &repository.CampaignTask{TaskID: t.task.ID, CampaignID: t.campaignID, ContactID: t.contactID, SequenceID: t.sequenceID, DispatchIntent: t.intent}, nil
+}
+
+func TestDeadLetterNeverReplaysSendTasksWithoutAuthority(t *testing.T) {
+	for _, tc := range []struct {
+		name, dlqType, taskType string
+		contact, sequence       bool
+		intent                  repository.CampaignDispatchIntent
+	}{
+		{name: "legacy warmup send", dlqType: "warmup", taskType: "warmup"},
+		{name: "user email send", dlqType: "user_email", taskType: "user_email"},
+		{name: "campaign contact send", dlqType: "campaign", taskType: "campaign", contact: true},
+		{name: "campaign sequence send", dlqType: "campaign", taskType: "campaign", sequence: true},
+		{name: "mismatched dead letter", dlqType: "warmup", taskType: "campaign"},
+		{name: "legacy null FKs", dlqType: "campaign", taskType: "campaign", intent: repository.CampaignDispatchUnverified},
+		{name: "send after FK cleanup", dlqType: "campaign", taskType: "campaign", intent: repository.CampaignDispatchSend},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			id := uuid.New()
+			campaignID := uuid.New()
+			repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: id, TaskType: tc.dlqType}}
+			tasks := &replayTasks{task: &repository.Task{ID: id, TaskType: tc.taskType, Status: "dead_lettered"}, campaignID: &campaignID, created: true, intent: tc.intent}
+			if tc.contact {
+				tasks.contactID = &id
+			}
+			if tc.sequence {
+				tasks.sequenceID = &id
+			}
+			svc := &service{repo: repo, taskRepo: tasks, tasksClient: replayClient{}}
+			if xerr := svc.ReplayDeadLetter(t.Context(), uuid.New(), repo.dlq.ID); xerr == nil {
+				t.Fatal("send dead letter was allowed through manual replay")
+			}
+			if n, xerr := svc.ProcessRetryableDeadLetters(t.Context()); xerr != nil || n != 0 {
+				t.Fatalf("auto replay result = %d, %v", n, xerr)
+			}
+			if tasks.inserted != 0 || tasks.repended != 0 || repo.replayed != 0 {
+				t.Fatalf("send mutated: inserted=%d pending=%d replayed=%d", tasks.inserted, tasks.repended, repo.replayed)
+			}
+		})
+	}
 }
 
 func (t *replayTasks) CreateTaskWithLock(context.Context, *repository.Task, *repository.CampaignTask) (bool, error) {
@@ -62,6 +109,16 @@ func (t *replayTasks) CreateTaskWithLock(context.Context, *repository.Task, *rep
 	return t.created, nil
 }
 
+func (t *replayTasks) CreateCampaignReplayTask(ctx context.Context, task *repository.Task, ct *repository.CampaignTask, source uuid.UUID) (bool, error) {
+	if source != t.task.ID {
+		return false, errors.New("wrong replay source")
+	}
+	if t.replayErr != nil {
+		return false, t.replayErr
+	}
+	return t.CreateTaskWithLock(ctx, task, ct)
+}
+
 func (t *replayTasks) UpdateTaskScheduledAt(context.Context, uuid.UUID, time.Time, string) error {
 	return nil
 }
@@ -89,6 +146,7 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) {
 	for _, tc := range []struct {
 		name         string
 		ctErr        error
+		replayErr    error
 		created      bool
 		queueFails   bool
 		wantErr      bool
@@ -100,11 +158,12 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) {
 		{name: "chain already has its next pass", created: false, wantReplayed: 1},
 		{name: "campaign task unreadable", ctErr: errors.New("connection reset"), wantErr: true},
 		{name: "queue refuses the new pass", created: true, queueFails: true, wantErr: true, wantInserted: 1, wantDeleted: 1},
+		{name: "intent changed before lock", replayErr: repository.ErrCampaignReplayUnverified, wantErr: true},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
-			task := &repository.Task{ID: uuid.New(), TaskType: "campaign", EmailAccountID: uuid.New()}
-			repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: task.ID}}
-			tasks := &replayTasks{task: task, campaignID: &campaignID, ctErr: tc.ctErr, created: tc.created}
+			task := &repository.Task{ID: uuid.New(), TaskType: "campaign", EmailAccountID: uuid.New(), Status: "dead_lettered"}
+			repo := &replayRepo{dlq: &models.TaskDeadLetter{ID: uuid.New(), TaskID: task.ID, TaskType: "campaign"}}
+			tasks := &replayTasks{task: task, campaignID: &campaignID, ctErr: tc.ctErr, replayErr: tc.replayErr, created: tc.created, intent: repository.CampaignDispatchWakeup}
 			svc := &service{repo: repo, taskRepo: tasks, tasksClient: replayClient{fail: tc.queueFails}}
 
 			xerr := svc.ReplayDeadLetter(context.Background(), uuid.New(), repo.dlq.ID)

internal/app/advanced/service.go

@@ -2288,26 +2288,15 @@ func (s *service) ReplayDeadLetter(ctx context.Context, organizationID, deadLett
 	if task == nil {
 		return errx.ErrNotFound
 	}
-	if _, handled, rerr := s.replayCampaignPass(ctx, task); handled {
-		if rerr != nil {
-			return toErrx(rerr)
-		}
-		if err := s.repo.MarkTaskDeadLetterReplayed(ctx, deadLetterID); err != nil {
-			return toErrx(err)
-		}
-		return nil
+	if dlq.TaskType != "campaign" || task.TaskType != "campaign" {
+		return errx.New(errx.Conflict, "send dead letters require authoritative outcome evidence and cannot be replayed")
 	}
-
-	scheduleAt := time.Now().UTC().Add(10 * time.Second)
-	cloudTaskName, err := s.tasksClient.CreateTask(ctx, &proto.ProcessTask{TaskId: task.ID.String()}, scheduleAt)
-	if err != nil {
-		return toErrx(err)
+	_, handled, rerr := s.replayCampaignPass(ctx, task)
+	if rerr != nil {
+		return toErrx(rerr)
 	}
-	if err := s.taskRepo.UpdateTaskScheduledAt(ctx, task.ID, scheduleAt, cloudTaskName); err != nil {
-		return toErrx(err)
-	}
-	if err := s.taskRepo.UpdateTaskStatus(ctx, task.ID, "pending"); err != nil {
-		return toErrx(err)
+	if !handled {
+		return errx.New(errx.Conflict, "only non-sending campaign passes can be replayed")
 	}
 	if err := s.repo.MarkTaskDeadLetterReplayed(ctx, deadLetterID); err != nil {
 		return toErrx(err)
@@ -2322,22 +2311,27 @@ func (s *service) ReplayDeadLetter(ctx context.Context, organizationID, deadLett
 // campaign pass; replayed, that a new pass was queued. An error means nothing
 // was queued and the dead letter stays for another try.
 func (s *service) replayCampaignPass(ctx context.Context, task *repository.Task) (replayed, handled bool, err error) {
-	if task == nil || task.TaskType != "campaign" {
+	if task == nil || task.TaskType != "campaign" || task.Status != "dead_lettered" {
 		return false, false, nil
 	}
 	ct, err := s.taskRepo.GetCampaignTask(ctx, task.ID)
 	if err != nil {
 		return false, true, err
 	}
-	if ct == nil || ct.CampaignID == nil {
-		// The campaign is gone; there is nothing to replay into.
-		return false, true, nil
+	if ct == nil || ct.DispatchIntent != repository.CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil {
+		return false, false, nil
+	}
+	if ct.CampaignID == nil {
+		return false, false, nil
 	}
 	at := time.Now().UTC().Add(10 * time.Second)
 	id := uuid.New()
-	created, err := s.taskRepo.CreateTaskWithLock(ctx,
+	created, err := s.taskRepo.CreateCampaignReplayTask(ctx,
 		&repository.Task{ID: id, TaskType: "campaign", EmailAccountID: task.EmailAccountID, Status: "pending", ScheduledAt: &at},
-		&repository.CampaignTask{TaskID: id, CampaignID: ct.CampaignID})
+		&repository.CampaignTask{TaskID: id, CampaignID: ct.CampaignID}, task.ID)
+	if errors.Is(err, repository.ErrCampaignReplayUnverified) {
+		return false, false, nil
+	}
 	if err != nil {
 		return false, true, err
 	}
@@ -2680,10 +2674,14 @@ func (s *service) ProcessRetryableDeadLetters(ctx context.Context) (int, *errx.E
 
 	retried := 0
 	for _, dlq := range items {
+		if dlq.TaskType != "campaign" {
+			continue
+		}
 		task, err := s.taskRepo.GetTask(ctx, dlq.TaskID)
-		if err != nil || task == nil {
-			// Mark as exhausted if the task no longer exists
-			_ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID)
+		if err != nil {
+			return retried, toErrx(err)
+		}
+		if task == nil || task.TaskType != "campaign" {
 			continue
 		}
 
@@ -2700,30 +2698,6 @@ func (s *service) ProcessRetryableDeadLetters(ctx context.Context) (int, *errx.E
 			_ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID)
 			continue
 		}
-
-		// Check if attempts exceeded
-		if dlq.Attempts >= dlq.MaxAttempts {
-			_ = s.repo.IncrementDeadLetterAttempt(ctx, dlq.ID, nil)
-			continue
-		}
-
-		// Schedule retry via Cloud Tasks
-		scheduleAt := time.Now().UTC().Add(10 * time.Second)
-		cloudTaskName, err := s.tasksClient.CreateTask(ctx, &proto.ProcessTask{TaskId: task.ID.String()}, scheduleAt)
-		if err != nil {
-			// Compute next retry with exponential backoff
-			backoff := time.Duration(30*(1<<uint(dlq.Attempts+1))) * time.Second
-			nextRetry := time.Now().UTC().Add(backoff)
-			_ = s.repo.IncrementDeadLetterAttempt(ctx, dlq.ID, &nextRetry)
-			continue
-		}
-
-		if err := s.taskRepo.UpdateTaskScheduledAt(ctx, task.ID, scheduleAt, cloudTaskName); err != nil {
-			continue
-		}
-		_ = s.taskRepo.UpdateTaskStatus(ctx, task.ID, "pending")
-		_ = s.repo.MarkTaskDeadLetterReplayed(ctx, dlq.ID)
-		retried++
 	}
 
 	return retried, nil

internal/app/orgtransfer/export_test.go

@@ -2,9 +2,22 @@ package orgtransfer
 
 import (
 	"encoding/json"
+	"slices"
 	"testing"
 )
 
+func TestCampaignReplayIntentResetsOnWorkspaceImport(t *testing.T) {
+	for _, tbl := range Tables {
+		if tbl.Name == "campaign_tasks" {
+			if !slices.Contains(tbl.ResetOnImport, "dispatch_intent") {
+				t.Fatal("workspace import grants source-instance replay intent")
+			}
+			return
+		}
+	}
+	t.Fatal("campaign tasks missing from portable data registry")
+}
+
 // An archive is a customer download. The risk verdict and the evidence behind
 // it are operator-facing, so they must not ride along in the manifest, and the
 // same list must refuse them on the way back in.

internal/app/orgtransfer/spec.go

@@ -814,7 +814,8 @@ var Tables = []Table{
 	},
 	{
 		Name: "campaign_tasks", Group: models.OrgDataGroupSending,
-		Scope: `task_id IN ` + orgTasks,
+		Scope:         `task_id IN ` + orgTasks,
+		ResetOnImport: []string{"dispatch_intent"},
 	},
 	{
 		Name: "warmup_tasks", Group: models.OrgDataGroupSending,

internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go

@@ -0,0 +1,124 @@
+package db
+
+import (
+	"context"
+	"net/url"
+	"os"
+	"strings"
+	"testing"
+
+	"github.com/golang-migrate/migrate/v4/source/iofs"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+func TestLiveCampaignDispatchIntentUpgradePreservesUnverifiedHistoryAndRollback(t *testing.T) {
+	dsn := os.Getenv("WARMBLY_TEST_DB")
+	if dsn == "" {
+		t.Skip("WARMBLY_TEST_DB not set")
+	}
+	u, err := url.Parse(dsn)
+	if err != nil || (u.Hostname() != "localhost" && u.Hostname() != "127.0.0.1") {
+		t.Fatal("upgrade requires local PostgreSQL")
+	}
+	admin, err := pgx.Connect(t.Context(), dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer admin.Close(context.Background())
+	name := "campaign_intent_upgrade_" + uuid.New().String()[:8]
+	quoted := pgx.Identifier{name}.Sanitize()
+	if _, err := admin.Exec(t.Context(), "CREATE DATABASE "+quoted); err != nil {
+		t.Fatal(err)
+	}
+	defer func() {
+		if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil {
+			t.Error(err)
+		}
+	}()
+	u.Path = "/" + name
+	source, err := iofs.New(migrationsFS, "migrations")
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, err := openMigrate(source, u.String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer m.Close()
+	if err := m.Migrate(281); err != nil {
+		t.Fatal(err)
+	}
+	conn, err := pgx.Connect(t.Context(), u.String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close(context.Background())
+	exec := func(sql string, args ...any) {
+		t.Helper()
+		if _, err := conn.Exec(t.Context(), sql, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	user, org, mailbox, campaign, contact, step := uuid.New(), uuid.New(), uuid.New(), uuid.New(), uuid.New(), uuid.New()
+	legacy, bound, wakeup := uuid.New(), uuid.New(), uuid.New()
+	exec(`INSERT INTO users(id,email,first_name,last_name) VALUES($1,$2,'Upgrade','')`, user, user.String()+"@example.test")
+	exec(`INSERT INTO organizations(id,name,owner_user_id) VALUES($1,'Upgrade',$2)`, org, user)
+	exec(`INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,send_recovery_hold,send_recovery_reason,send_recovery_task_id) VALUES($1,$2,$3,$4,'Upgrade','','','smtp_imap',true,'unknown',$5)`, mailbox, user, org, mailbox.String()+"@example.test", legacy)
+	exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at) VALUES($1,$2,$3,'Upgrade','','active',127,'UTC',NOW(),NOW())`, campaign, user, org)
+	exec(`INSERT INTO contacts(id,user_id,organization_id,email,first_name,last_name,company,phone,custom_fields) VALUES($1,$2,$3,'lead@example.test','','','','','{}')`, contact, user, org)
+	exec(`INSERT INTO sequences(id,campaign_id,organization_id,name,subject,body_plain,body_html,wait_after,position) VALUES($1,$2,$3,'Step','Subject','Text','',0,1)`, step, campaign, org)
+	for _, task := range []uuid.UUID{legacy, bound} {
+		exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, task, mailbox)
+		exec(`INSERT INTO campaign_tasks(task_id,campaign_id) VALUES($1,$2)`, task, campaign)
+		exec(`INSERT INTO task_dead_letters(task_id,task_type,payload,last_error,attempts) VALUES($1,'campaign','{"dispatch_intent":"wakeup"}','original',1)`, task)
+	}
+	exec(`UPDATE campaign_tasks SET contact_id=$2,sequence_id=$3 WHERE task_id=$1`, bound, contact, step)
+	if err := m.Up(); err != nil {
+		t.Fatal(err)
+	}
+	var unchanged bool
+	if err := conn.QueryRow(t.Context(), `SELECT COUNT(*)=2 AND bool_and(ct.dispatch_intent='unverified' AND t.send_result_state IS NULL AND t.send_result_applied_at IS NULL AND d.status='pending' AND d.attempts=1 AND d.last_error='original' AND d.payload='{"dispatch_intent":"wakeup"}'::jsonb AND ea.send_recovery_hold AND ea.send_recovery_reason='unknown') FROM campaign_tasks ct JOIN tasks t ON t.id=ct.task_id JOIN task_dead_letters d ON d.task_id=t.id JOIN email_accounts ea ON ea.id=t.email_account_id`).Scan(&unchanged); err != nil || !unchanged {
+		t.Fatal("upgrade granted authority or changed unknown evidence", unchanged, err)
+	}
+	exec(`UPDATE campaign_tasks SET dispatch_intent='wakeup' WHERE task_id=$1`, legacy)
+	var intent string
+	if err := conn.QueryRow(t.Context(), `SELECT dispatch_intent::text FROM campaign_tasks WHERE task_id=$1`, legacy).Scan(&intent); err != nil || intent != "unverified" {
+		t.Fatal("legacy promoted to verified wakeup", intent, err)
+	}
+	exec(`DELETE FROM contacts WHERE id=$1`, contact)
+	exec(`DELETE FROM sequences WHERE id=$1`, step)
+	exec(`UPDATE campaign_tasks SET dispatch_intent='wakeup' WHERE task_id=$1`, bound)
+	if err := conn.QueryRow(t.Context(), `SELECT dispatch_intent FROM campaign_tasks WHERE task_id=$1`, bound).Scan(&intent); err != nil || intent != "send" {
+		t.Fatal("FK cleanup forgot prior bound send", intent, err)
+	}
+	exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, wakeup, mailbox)
+	exec(`INSERT INTO campaign_tasks(task_id,campaign_id,dispatch_intent) VALUES($1,$2,'wakeup')`, wakeup, campaign)
+	if _, err := conn.Exec(t.Context(), `UPDATE campaign_tasks SET dispatch_intent='arbitrary' WHERE task_id=$1`, wakeup); err == nil {
+		t.Fatal("intent accepted arbitrary value")
+	}
+	down, err := migrationsFS.ReadFile("migrations/000282_campaign_dispatch_intent.down.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := conn.Exec(t.Context(), string(down)); err == nil || !strings.Contains(err.Error(), "restrictive recovery evidence") {
+		t.Fatal("rollback discarded retained send evidence", err)
+	}
+	exec(`DELETE FROM tasks WHERE id=$1`, bound)
+	if _, err := conn.Exec(t.Context(), string(down)); err == nil || !strings.Contains(err.Error(), "restrictive recovery evidence") {
+		t.Fatal("rollback discarded unverified pending dead letter", err)
+	}
+	// Only this isolated fixture is removed to exercise a clean down/up pair.
+	exec(`DELETE FROM tasks WHERE id IN($1,$2)`, legacy, wakeup)
+	exec(string(down))
+	up, err := migrationsFS.ReadFile("migrations/000282_campaign_dispatch_intent.up.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	exec(string(up))
+	exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, legacy, mailbox)
+	exec(`INSERT INTO campaign_tasks(task_id,campaign_id) VALUES($1,$2)`, legacy, campaign)
+	if err := conn.QueryRow(t.Context(), `SELECT dispatch_intent::text FROM campaign_tasks WHERE task_id=$1`, legacy).Scan(&intent); err != nil || intent != "unverified" {
+		t.Fatal("clean down/up lost restrictive default", intent, err)
+	}
+}

internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql

@@ -0,0 +1,17 @@
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM campaign_tasks ct
+        LEFT JOIN task_dead_letters d ON d.task_id = ct.task_id AND d.status = 'pending'
+        WHERE ct.dispatch_intent = 'send'
+           OR (ct.dispatch_intent = 'unverified' AND d.id IS NOT NULL)
+    ) THEN
+        RAISE EXCEPTION 'campaign dispatch intent is restrictive recovery evidence; retain it and roll forward';
+    END IF;
+END;
+$$;
+
+DROP TRIGGER campaign_dispatch_intent_monotonic ON campaign_tasks;
+DROP FUNCTION retain_campaign_dispatch_intent();
+ALTER TABLE campaign_tasks DROP COLUMN dispatch_intent;
+DROP TYPE campaign_dispatch_intent;

internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql

@@ -0,0 +1,24 @@
+CREATE TYPE campaign_dispatch_intent AS ENUM ('unverified', 'wakeup', 'send');
+
+ALTER TABLE campaign_tasks
+    ADD COLUMN dispatch_intent campaign_dispatch_intent NOT NULL DEFAULT 'unverified';
+
+CREATE FUNCTION retain_campaign_dispatch_intent() RETURNS trigger LANGUAGE plpgsql AS $$
+BEGIN
+    IF TG_OP = 'UPDATE' THEN
+        IF OLD.dispatch_intent = 'send' OR OLD.contact_id IS NOT NULL OR OLD.sequence_id IS NOT NULL THEN
+            NEW.dispatch_intent := 'send';
+        ELSIF OLD.dispatch_intent = 'unverified' AND NEW.dispatch_intent = 'wakeup' THEN
+            NEW.dispatch_intent := 'unverified';
+        END IF;
+    END IF;
+    IF NEW.contact_id IS NOT NULL OR NEW.sequence_id IS NOT NULL THEN
+        NEW.dispatch_intent := 'send';
+    END IF;
+    RETURN NEW;
+END;
+$$;
+
+CREATE TRIGGER campaign_dispatch_intent_monotonic
+    BEFORE INSERT OR UPDATE ON campaign_tasks
+    FOR EACH ROW EXECUTE FUNCTION retain_campaign_dispatch_intent();

internal/repository/campaign_replay_intent_live_test.go

@@ -0,0 +1,186 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"net/url"
+	"os"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+)
+
+func liveCampaignReplayDB(t *testing.T) *pgxpool.Pool {
+	t.Helper()
+	dsn := os.Getenv("WARMBLY_TEST_DB")
+	if dsn == "" {
+		t.Skip("WARMBLY_TEST_DB not set")
+	}
+	u, err := url.Parse(dsn)
+	if err != nil || (u.Hostname() != "localhost" && u.Hostname() != "127.0.0.1") {
+		t.Fatal("replay fixture requires local PostgreSQL")
+	}
+	admin, err := pgx.Connect(t.Context(), dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	name := "campaign_replay_" + uuid.New().String()[:8]
+	quoted := pgx.Identifier{name}.Sanitize()
+	if _, err := admin.Exec(t.Context(), "CREATE DATABASE "+quoted); err != nil {
+		_ = admin.Close(context.Background())
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil {
+			t.Error(err)
+		}
+		_ = admin.Close(context.Background())
+	})
+	u.Path = "/" + name
+	if err := db.RunMigrations(u.String()); err != nil {
+		t.Fatal(err)
+	}
+	pool, err := pgxpool.New(t.Context(), u.String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(pool.Close)
+	return pool
+}
+
+func TestLiveCampaignReplayIntentTransitionsFilterBeforeLimitAndLock(t *testing.T) {
+	pool := liveCampaignReplayDB(t)
+	f := newWarmupUsageFixture(t, pool)
+	r := NewTaskRepository(pool)
+	campaign, contact, sequence := uuid.New(), uuid.New(), uuid.New()
+	f.exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at) VALUES($1,$2,$3,'Replay','','active',127,'UTC',NOW(),NOW())`, campaign, f.user, f.org)
+	f.exec(`INSERT INTO contacts(id,user_id,organization_id,email,first_name,last_name,company,phone,custom_fields) VALUES($1,$2,$3,'lead@example.test','','','','','{}')`, contact, f.user, f.org)
+	f.exec(`INSERT INTO sequences(id,campaign_id,organization_id,name,subject,body_plain,body_html,wait_after,position) VALUES($1,$2,$3,'Step','Subject','Text','',0,1)`, sequence, campaign, f.org)
+	if err := r.UpdateCampaignTaskTracking(t.Context(), uuid.New(), contact, sequence); !errors.Is(err, pgx.ErrNoRows) {
+		t.Fatal("missing tracking row accepted as durable intent", err)
+	}
+	for range 60 {
+		id := uuid.New()
+		f.exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'campaign','dead_lettered','')`, id, f.account)
+		f.exec(`INSERT INTO campaign_tasks(task_id,campaign_id) VALUES($1,$2)`, id, campaign)
+		f.exec(`INSERT INTO task_dead_letters(task_id,task_type,next_retry_at,payload) VALUES($1,'campaign',NOW()-INTERVAL '2 hours','{"dispatch_intent":"wakeup","safe":true}')`, id)
+	}
+	newWakeup := func() uuid.UUID {
+		t.Helper()
+		id := uuid.New()
+		created, err := r.CreateTaskWithLock(t.Context(), &Task{ID: id, TaskType: "campaign", Status: "dead_lettered", EmailAccountID: f.account}, &CampaignTask{TaskID: id, CampaignID: &campaign})
+		if err != nil || !created {
+			t.Fatalf("create wakeup: %t %v", created, err)
+		}
+		f.exec(`INSERT INTO task_dead_letters(task_id,task_type,next_retry_at,payload,last_error,attempts) VALUES($1,'campaign',NOW()-INTERVAL '1 hour','{"original":true}','retained',1)`, id)
+		return id
+	}
+	send := newWakeup()
+	if err := r.UpdateCampaignTaskTracking(t.Context(), send, contact, sequence); err != nil {
+		t.Fatal(err)
+	}
+	f.exec(`DELETE FROM contacts WHERE id=$1`, contact)
+	f.exec(`DELETE FROM sequences WHERE id=$1`, sequence)
+	f.exec(`UPDATE campaign_tasks SET dispatch_intent='wakeup' WHERE task_id=$1`, send)
+	ct, err := r.GetCampaignTask(t.Context(), send)
+	if err != nil || ct.DispatchIntent != CampaignDispatchSend || ct.ContactID != nil || ct.SequenceID != nil {
+		t.Fatalf("send capability lost after FK cleanup: %+v %v", ct, err)
+	}
+	wakeup := newWakeup()
+	listed, err := NewAdvancedOutreachRepository(pool).ListRetryableDeadLetters(t.Context(), 50)
+	if err != nil || len(listed) != 1 || listed[0].TaskID != wakeup {
+		t.Fatalf("unsafe legacy rows filled LIMIT before wakeup: %+v %v", listed, err)
+	}
+	for _, row := range listed {
+		ct, err := r.GetCampaignTask(t.Context(), row.TaskID)
+		if err != nil || ct.DispatchIntent != CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil {
+			t.Fatalf("unverified replay candidate: %+v %v", ct, err)
+		}
+	}
+	for _, source := range []uuid.UUID{send, uuid.New()} {
+		id := uuid.New()
+		if _, err := r.CreateCampaignReplayTask(t.Context(), &Task{ID: id, TaskType: "campaign", Status: "pending", EmailAccountID: f.account}, &CampaignTask{TaskID: id, CampaignID: &campaign}, source); !errors.Is(err, ErrCampaignReplayUnverified) {
+			t.Fatalf("unverified source replay: %v", err)
+		}
+	}
+	var created atomic.Int32
+	var wg sync.WaitGroup
+	errs := make(chan error, 8)
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			id := uuid.New()
+			ok, err := r.CreateCampaignReplayTask(t.Context(), &Task{ID: id, TaskType: "campaign", Status: "pending", EmailAccountID: f.account}, &CampaignTask{TaskID: id, CampaignID: &campaign}, wakeup)
+			if err != nil {
+				errs <- err
+			}
+			if ok {
+				created.Add(1)
+			}
+		}()
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		t.Fatal(err)
+	}
+	if created.Load() != 1 {
+		t.Fatal("per-campaign replay created duplicate chains", created.Load())
+	}
+	var intact bool
+	if err := pool.QueryRow(t.Context(), `SELECT status='pending' AND attempts=1 AND last_error='retained' AND payload='{"original":true}'::jsonb FROM task_dead_letters WHERE task_id=$1`, wakeup).Scan(&intact); err != nil || !intact {
+		t.Fatal("constructor changed original evidence", intact, err)
+	}
+}
+
+func TestLiveCampaignReplayRechecksIntentAfterRowLockWait(t *testing.T) {
+	pool := liveCampaignReplayDB(t)
+	f := newWarmupUsageFixture(t, pool)
+	r := NewTaskRepository(pool)
+	campaign, source, replay := uuid.New(), uuid.New(), uuid.New()
+	f.exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at) VALUES($1,$2,$3,'Replay','','active',127,'UTC',NOW(),NOW())`, campaign, f.user, f.org)
+	if ok, err := r.CreateTaskWithLock(t.Context(), &Task{ID: source, TaskType: "campaign", Status: "dead_lettered", EmailAccountID: f.account}, &CampaignTask{TaskID: source, CampaignID: &campaign}); err != nil || !ok {
+		t.Fatal(ok, err)
+	}
+	f.exec(`INSERT INTO task_dead_letters(task_id,task_type,next_retry_at) VALUES($1,'campaign',NOW())`, source)
+	tx, err := pool.Begin(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer tx.Rollback(context.Background())
+	if _, err := tx.Exec(t.Context(), `UPDATE campaign_tasks SET dispatch_intent='send' WHERE task_id=$1`, source); err != nil {
+		t.Fatal(err)
+	}
+	result := make(chan error, 1)
+	go func() {
+		_, err := r.CreateCampaignReplayTask(t.Context(), &Task{ID: replay, TaskType: "campaign", Status: "pending", EmailAccountID: f.account}, &CampaignTask{TaskID: replay, CampaignID: &campaign}, source)
+		result <- err
+	}()
+	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
+	defer cancel()
+	for {
+		var waiting bool
+		if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_stat_activity WHERE datname=current_database() AND wait_event_type='Lock' AND query LIKE 'SELECT true FROM task_dead_letters%')`).Scan(&waiting); err != nil {
+			t.Fatal(err)
+		}
+		if waiting {
+			break
+		}
+		if ctx.Err() != nil {
+			t.Fatal(ctx.Err())
+		}
+	}
+	if err := tx.Commit(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	if err := <-result; !errors.Is(err, ErrCampaignReplayUnverified) {
+		t.Fatalf("stale wakeup snapshot allowed replay: %v", err)
+	}
+}

internal/repository/pg_advanced_outreach.go

@@ -1314,13 +1314,15 @@ func (r *advancedOutreachRepository) ListRetryableDeadLetters(ctx context.Contex
 		limit = 10
 	}
 	query := `
-		SELECT id, task_id, task_type, payload, last_error, attempts, max_attempts, status, next_retry_at, replayed_at, created_at, updated_at
-		FROM task_dead_letters
-		WHERE status = 'pending'
-		  AND next_retry_at IS NOT NULL
-		  AND next_retry_at <= NOW()
-		  AND attempts < max_attempts
-		ORDER BY next_retry_at ASC
+		SELECT d.id, d.task_id, d.task_type, d.payload, d.last_error, d.attempts, d.max_attempts, d.status, d.next_retry_at, d.replayed_at, d.created_at, d.updated_at
+		FROM task_dead_letters d
+		JOIN tasks t ON t.id = d.task_id
+		WHERE d.status = 'pending'
+		  AND d.next_retry_at IS NOT NULL
+		  AND d.next_retry_at <= NOW()
+		  AND d.attempts < d.max_attempts
+		  AND ` + safeDLQCampaignPass + `
+		ORDER BY d.next_retry_at ASC
 		LIMIT $1
 	`
 	rows, err := resultDB(ctx, r.db).Query(ctx, query, limit)

internal/repository/pg_monitoring.go

@@ -238,10 +238,11 @@ var monitoringSections = map[string]monitoringSection{
 		{id: "job_ok", title: "Jobs with latest successful wrapper completion", unit: "jobs", condition: "healthy", severity: "info", note: "Name-keyed instance rows, not replica heartbeats; wrapper return does not prove subsystem throughput.", window: 0 * time.Second, query: `SELECT COUNT(*), NULL::bigint, NULL::bigint, NULL::bigint, MIN(j.last_finished_at::timestamptz), MAX(j.last_finished_at::timestamptz), MIN(j.next_run_at::timestamptz) FROM scheduled_job_runs j WHERE j.last_status='ok'`},
 	}},
 	"dead_letters": {permission: models.AdminPermViewCampaigns, investigate: "/sends?tab=dead-letters", queries: []monitoringQuery{
-		{id: "dlq_due", title: "Dead letters due for retry", unit: "records", condition: "backlog", severity: "warning", note: "Durable queue state; attempts are delivery/replay attempts, replay is not provider confirmation.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.next_retry_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at<=$1 AND d.attempts<d.max_attempts`},
-		{id: "dlq_wait", title: "Dead letters awaiting retry slot", unit: "records", condition: "waiting_retry", severity: "info", note: "Durable queue state; attempts are delivery/replay attempts, replay is not provider confirmation.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at>$1 AND d.attempts<d.max_attempts`},
+		{id: "dlq_due", title: "Campaign wakeup dead letters due for replay", unit: "records", condition: "backlog", severity: "warning", note: "Only non-sending campaign passes; the consumer's automatic replay is not configured. Requires manual recovery, not provider delivery proof.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.next_retry_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at<=$1 AND d.attempts<d.max_attempts AND ` + safeDLQCampaignPass},
+		{id: "dlq_wait", title: "Campaign wakeup dead letters awaiting retry slot", unit: "records", condition: "waiting_retry", severity: "info", note: "Only non-sending campaign passes; the consumer's automatic replay is not configured.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at>$1 AND d.attempts<d.max_attempts AND ` + safeDLQCampaignPass},
+		{id: "dlq_unverified", title: "Dead letters without safe replay proof", unit: "records", condition: "recovery_unverified", severity: "warning", note: "Includes legacy send tasks. Retry dates and absent send markers do not prove the worker never sent; these rows are retained, not replayed.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.attempts<d.max_attempts AND NOT (` + safeDLQCampaignPass + `)`},
 		{id: "dlq_exhausted", title: "Dead letters with exhausted attempts", unit: "records", condition: "recovery_unverified", severity: "warning", note: "Durable queue state; attempts are delivery/replay attempts, replay is not provider confirmation.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status IN ('pending','failed') AND d.attempts>=d.max_attempts`},
-		{id: "dlq_unscheduled", title: "Pending dead letters without retry slot", unit: "records", condition: "unknown", severity: "warning", note: "Durable queue state; attempts are delivery/replay attempts, replay is not provider confirmation.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at IS NULL AND d.attempts<d.max_attempts`},
+		{id: "dlq_unscheduled", title: "Campaign wakeup dead letters without retry slot", unit: "records", condition: "unknown", severity: "warning", note: "Only non-sending campaign passes; the consumer's automatic replay is not configured.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='pending' AND d.next_retry_at IS NULL AND d.attempts<d.max_attempts AND ` + safeDLQCampaignPass},
 		{id: "dlq_replayed", title: "Replayed dead letters", unit: "records", condition: "unknown", severity: "info", note: "Durable queue state; attempts are delivery/replay attempts, replay is not provider confirmation.", window: 0 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(d.created_at::timestamptz), MAX(d.updated_at::timestamptz), MIN(d.next_retry_at::timestamptz) FROM task_dead_letters d LEFT JOIN tasks t ON t.id=d.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE d.status='replayed'`},
 		{id: "task_failures_recent", title: "Currently failed tasks updated within 1h", unit: "tasks", condition: "recent_failure", severity: "warning", note: "Mutable task update proxy; not immutable failure-event time.", window: 3600 * time.Second, query: `SELECT COUNT(*), COUNT(DISTINCT t.email_account_id), COUNT(DISTINCT e.organization_id), COUNT(*) FILTER(WHERE e.organization_id IS NULL), MIN(t.updated_at::timestamptz), MAX(t.updated_at::timestamptz), MIN(NULL::timestamptz) FROM task_failures f JOIN tasks t ON t.id=f.task_id LEFT JOIN email_accounts e ON e.id=t.email_account_id WHERE t.status='failed' AND t.updated_at BETWEEN $1-INTERVAL '1 hour' AND $1`},
 	}},

internal/repository/pg_monitoring_queues_live_test.go

@@ -11,8 +11,8 @@ import (
 )
 
 func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) {
-	_, pool := liveContactDB(t)
-	requireSchemaVersion(t, pool, 277)
+	pool := liveCampaignReplayDB(t)
+	requireSchemaVersion(t, pool, 282)
 	f := newWarmupUsageFixture(t, pool)
 	now := time.Now().UTC().Truncate(time.Microsecond)
 	r := NewMonitoringRepository(pool)
@@ -30,6 +30,34 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) {
 		f.exec(`INSERT INTO task_dead_letters(task_id,task_type,status,next_retry_at,attempts,max_attempts,payload,last_error)VALUES($1,'email',$2,$3,$4,5,'{"private":"body"}','private cursor')`, id, row.status, row.at, row.attempts)
 	}
 	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM task_dead_letters WHERE task_id=$1`, id) })
+	passID := uuid.New()
+	campaignID := uuid.New()
+	f.exec(`INSERT INTO campaigns(id,user_id,organization_id,name,description,status,days,timezone,created_at,updated_at)VALUES($1,$2,$3,'Replay','','active',127,'UTC',NOW(),NOW())`, campaignID, f.user, f.org)
+	if created, err := NewTaskRepository(pool).CreateTaskWithLock(t.Context(), &Task{ID: passID, TaskType: "campaign", EmailAccountID: f.account, Status: "dead_lettered"}, &CampaignTask{TaskID: passID, CampaignID: &campaignID}); err != nil || !created {
+		t.Fatal(created, err)
+	}
+	f.exec(`INSERT INTO task_dead_letters(task_id,task_type,status,next_retry_at,attempts,max_attempts,payload,last_error)VALUES($1,'campaign','pending',$2,1,5,'{}','test')`, passID, now.Add(-2*time.Minute))
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM task_dead_letters WHERE task_id=$1`, passID)
+	})
+	listed, err := NewAdvancedOutreachRepository(pool).ListRetryableDeadLetters(t.Context(), 50)
+	if err != nil {
+		t.Fatal(err)
+	}
+	found := false
+	for _, letter := range listed {
+		found = found || letter.TaskID == passID
+		ct, err := NewTaskRepository(pool).GetCampaignTask(t.Context(), letter.TaskID)
+		if err != nil || ct == nil || ct.DispatchIntent != CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil {
+			t.Fatal("candidate lacks positive wakeup intent", ct, err)
+		}
+		if letter.TaskID == id {
+			t.Fatal("legacy send was selected for automatic replay")
+		}
+	}
+	if !found {
+		t.Fatal("replayable campaign pass missing from isolated fixture")
+	}
 	job := uuid.NewString()
 	f.exec(`INSERT INTO scheduled_job_runs(name,last_status,last_finished_at,error_count,run_count,last_error)VALUES($1,'error',$2,999,1000,'private error')`, job, now.Add(-time.Minute))
 	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM scheduled_job_runs WHERE name=$1`, job) })
@@ -50,7 +78,7 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) {
 	}
 	f.exec(`INSERT INTO send_result_effects(task_id,effect_key,organization_id,kind,payload,attempts)VALUES($1,'monitoring',$2,'notification','{"private":"body"}',2)`, id, f.org)
 	for source, deltas := range map[string]map[string]int64{
-		"jobs": {"job_error_recent": 1}, "dead_letters": {"dlq_due": 1, "dlq_wait": 1, "dlq_exhausted": 1, "dlq_replayed": 1}, "webhooks": {"webhook_due": 1, "webhook_wait": 1, "webhook_stale_claim": 1, "webhook_abandoned_24h": 1, "webhook_endpoint_persistent": 1}, "notifications": {"notification_aging": 1, "notification_stale": 1, "notification_settled_24h": 1, "notification_exhausted": 1}, "result_effects": {"effects_pending": 1, "effects_repeated": 1},
+		"jobs": {"job_error_recent": 1}, "dead_letters": {"dlq_due": 1, "dlq_unverified": 2, "dlq_exhausted": 1, "dlq_replayed": 1}, "webhooks": {"webhook_due": 1, "webhook_wait": 1, "webhook_stale_claim": 1, "webhook_abandoned_24h": 1, "webhook_endpoint_persistent": 1}, "notifications": {"notification_aging": 1, "notification_stale": 1, "notification_settled_24h": 1, "notification_exhausted": 1}, "result_effects": {"effects_pending": 1, "effects_repeated": 1},
 	} {
 		got := collectMonitoring(t, NewMonitoringRepository(monitoringReadOnlyPool(t, pool)), source, now)
 		for id, delta := range deltas {
@@ -62,6 +90,10 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) {
 		}
 	}
 	settled := measuredMonitoring(t, collectMonitoring(t, r, "notifications", now), "notification_settled_24h")
+	dlq := collectMonitoring(t, r, "dead_letters", now)
+	if *measuredMonitoring(t, dlq, "dlq_wait").Count != *measuredMonitoring(t, baseline["dead_letters"], "dlq_wait").Count {
+		t.Fatal("unverified send was reported as eligible for replay")
+	}
 	if settled.Condition == models.MonitoringHealthy {
 		t.Fatal("notification settlement claimed delivery")
 	}

internal/repository/pg_task.go

@@ -30,12 +30,29 @@ type Task struct {
 
 // CampaignTask represents campaign-specific task data
 type CampaignTask struct {
-	TaskID     uuid.UUID
-	CampaignID *uuid.UUID
-	ContactID  *uuid.UUID
-	SequenceID *uuid.UUID
+	TaskID         uuid.UUID
+	CampaignID     *uuid.UUID
+	ContactID      *uuid.UUID
+	SequenceID     *uuid.UUID
+	DispatchIntent CampaignDispatchIntent
 }
 
+type CampaignDispatchIntent string
+
+const (
+	CampaignDispatchUnverified CampaignDispatchIntent = "unverified"
+	CampaignDispatchWakeup     CampaignDispatchIntent = "wakeup"
+	CampaignDispatchSend       CampaignDispatchIntent = "send"
+)
+
+var ErrCampaignReplayUnverified = errors.New("campaign replay requires verified wakeup intent")
+
+const campaignWakeupReplayPredicate = `ct.dispatch_intent='wakeup' AND ct.campaign_id IS NOT NULL AND ct.contact_id IS NULL AND ct.sequence_id IS NULL`
+
+const campaignReplayTaskPredicate = `d.task_type='campaign' AND t.task_type='campaign' AND t.status='dead_lettered'`
+
+const safeDLQCampaignPass = campaignReplayTaskPredicate + ` AND EXISTS (SELECT 1 FROM campaign_tasks ct WHERE ct.task_id=t.id AND ` + campaignWakeupReplayPredicate + `)`
+
 // WarmupTask represents warmup-specific task data
 type WarmupTask struct {
 	TaskID          uuid.UUID
@@ -170,6 +187,7 @@ type TaskRepository interface {
 
 	// Task locking
 	CreateTaskWithLock(ctx context.Context, task *Task, campaignTask *CampaignTask) (bool, error)
+	CreateCampaignReplayTask(ctx context.Context, task *Task, campaignTask *CampaignTask, sourceTaskID uuid.UUID) (bool, error)
 	CreateWarmupTaskWithLock(ctx context.Context, task *Task, warmupTask *WarmupTask) (bool, error)
 	// DirectPendingWarmupTask points a mailbox's pending warmup task at one
 	// partner and pulls it forward. Returns false when there is no pending task
@@ -385,7 +403,7 @@ func (r *taskRepository) GetTaskByMessageID(ctx context.Context, messageID strin
 // GetCampaignTask retrieves campaign task data
 func (r *taskRepository) GetCampaignTask(ctx context.Context, taskID uuid.UUID) (*CampaignTask, error) {
 	query := `
-		SELECT task_id, campaign_id, contact_id, sequence_id
+		SELECT task_id, campaign_id, contact_id, sequence_id, dispatch_intent
 		FROM campaign_tasks
 		WHERE task_id = $1
 	`
@@ -396,6 +414,7 @@ func (r *taskRepository) GetCampaignTask(ctx context.Context, taskID uuid.UUID)
 		&campaignTask.CampaignID,
 		&campaignTask.ContactID,
 		&campaignTask.SequenceID,
+		&campaignTask.DispatchIntent,
 	)
 
 	if errors.Is(err, sql.ErrNoRows) {
@@ -945,6 +964,20 @@ func (r *taskRepository) DeleteTask(ctx context.Context, taskID uuid.UUID) error
 // CreateTaskWithLock creates a campaign task with a PostgreSQL advisory lock.
 // It returns false when the campaign already has a pending wakeup task.
 func (r *taskRepository) CreateTaskWithLock(ctx context.Context, task *Task, campaignTask *CampaignTask) (bool, error) {
+	return r.createCampaignTaskWithLock(ctx, task, campaignTask, uuid.Nil)
+}
+
+func (r *taskRepository) CreateCampaignReplayTask(ctx context.Context, task *Task, campaignTask *CampaignTask, sourceTaskID uuid.UUID) (bool, error) {
+	if sourceTaskID == uuid.Nil || task == nil || task.TaskType != "campaign" || campaignTask == nil || campaignTask.CampaignID == nil || campaignTask.ContactID != nil || campaignTask.SequenceID != nil {
+		return false, ErrCampaignReplayUnverified
+	}
+	return r.createCampaignTaskWithLock(ctx, task, campaignTask, sourceTaskID)
+}
+
+func (r *taskRepository) createCampaignTaskWithLock(ctx context.Context, task *Task, campaignTask *CampaignTask, sourceTaskID uuid.UUID) (bool, error) {
+	if task == nil || (campaignTask != nil && campaignTask.TaskID != task.ID) {
+		return false, ErrCampaignReplayUnverified
+	}
 	tx, err := r.db.Begin(ctx)
 	if err != nil {
 		return false, err
@@ -957,6 +990,19 @@ func (r *taskRepository) CreateTaskWithLock(ctx context.Context, task *Task, cam
 		if err != nil {
 			return false, err
 		}
+		if sourceTaskID != uuid.Nil {
+			var verified bool
+			err = tx.QueryRow(ctx, `SELECT true FROM task_dead_letters d JOIN tasks t ON t.id=d.task_id
+			 JOIN campaign_tasks ct ON ct.task_id=t.id WHERE t.id=$1 AND ct.campaign_id=$2
+			 AND t.email_account_id=$3 AND d.status='pending' AND `+campaignReplayTaskPredicate+` AND `+campaignWakeupReplayPredicate+`
+			 FOR UPDATE OF d,t,ct`, sourceTaskID, *campaignTask.CampaignID, task.EmailAccountID).Scan(&verified)
+			if errors.Is(err, pgx.ErrNoRows) {
+				return false, ErrCampaignReplayUnverified
+			}
+			if err != nil {
+				return false, err
+			}
+		}
 
 		var existing uuid.UUID
 		err = tx.QueryRow(ctx, `
@@ -987,8 +1033,12 @@ func (r *taskRepository) CreateTaskWithLock(ctx context.Context, task *Task, cam
 
 	// Create campaign task entry if provided
 	if campaignTask != nil {
-		ctQuery := `INSERT INTO campaign_tasks (task_id, campaign_id, contact_id, sequence_id) VALUES ($1, $2, $3, $4)`
-		_, err = tx.Exec(ctx, ctQuery, campaignTask.TaskID, campaignTask.CampaignID, campaignTask.ContactID, campaignTask.SequenceID)
+		intent := CampaignDispatchUnverified
+		if task.TaskType == "campaign" && campaignTask.CampaignID != nil && campaignTask.ContactID == nil && campaignTask.SequenceID == nil {
+			intent = CampaignDispatchWakeup
+		}
+		ctQuery := `INSERT INTO campaign_tasks (task_id, campaign_id, contact_id, sequence_id, dispatch_intent) VALUES ($1, $2, $3, $4, $5)`
+		_, err = tx.Exec(ctx, ctQuery, campaignTask.TaskID, campaignTask.CampaignID, campaignTask.ContactID, campaignTask.SequenceID, intent)
 		if err != nil {
 			return false, err
 		}
@@ -1181,7 +1231,10 @@ func (r *taskRepository) UpdateCampaignTaskTracking(ctx context.Context, taskID,
 		WHERE task_id = $1
 	`
 
-	_, err := r.db.Exec(ctx, query, taskID, contactID, sequenceID)
+	tag, err := r.db.Exec(ctx, query, taskID, contactID, sequenceID)
+	if err == nil && tag.RowsAffected() != 1 {
+		return pgx.ErrNoRows
+	}
 	return err
 }
 

internal/tasks/campaign_send_live_test.go

@@ -37,6 +37,47 @@ type breakingProgressRepo struct {
 	breakStamp bool
 }
 
+type rejectingCampaignTrackingRepo struct {
+	repository.TaskRepository
+}
+
+func (r rejectingCampaignTrackingRepo) UpdateCampaignTaskTracking(context.Context, uuid.UUID, uuid.UUID, uuid.UUID) error {
+	return errors.New("simulated durable tracking write failure")
+}
+
+func TestLiveCampaignSendRequiresDurableIntentBeforeReservation(t *testing.T) {
+	f := newSendFixture(t)
+	id := uuid.New()
+	at := time.Now().Add(-time.Minute)
+	if ok, err := f.svc.taskRepo.CreateTaskWithLock(t.Context(), &Task{ID: id, TaskType: "campaign", EmailAccountID: f.mailbox, Status: "pending", ScheduledAt: &at}, &CampaignTask{TaskID: id, CampaignID: &f.campaign}); err != nil || !ok {
+		t.Fatal(ok, err)
+	}
+	original := f.svc.taskRepo
+	f.svc.taskRepo = rejectingCampaignTrackingRepo{TaskRepository: original}
+	if xerr := f.svc.HandleCampaignTask(&proto.ProcessTask{TaskId: id.String()}); xerr == nil {
+		t.Fatal("failed durable intent write allowed execution")
+	}
+	if f.sender.count() != 0 {
+		t.Fatal("send dispatched without durable send intent")
+	}
+	var dispatched int
+	if err := f.pool.QueryRow(t.Context(), `SELECT COUNT(*) FROM campaign_contact_progress WHERE campaign_id=$1 AND dispatched_at IS NOT NULL`, f.campaign).Scan(&dispatched); err != nil || dispatched != 0 {
+		t.Fatal("send reserved without durable intent", dispatched, err)
+	}
+	ct, err := original.GetCampaignTask(t.Context(), id)
+	if err != nil || ct.DispatchIntent != repository.CampaignDispatchWakeup {
+		t.Fatal("unsent wakeup intent changed", ct, err)
+	}
+	f.svc.taskRepo = original
+	if xerr := f.svc.HandleCampaignTask(&proto.ProcessTask{TaskId: id.String()}); xerr != nil {
+		t.Fatal(xerr)
+	}
+	ct, err = original.GetCampaignTask(t.Context(), id)
+	if err != nil || ct.DispatchIntent != repository.CampaignDispatchSend || f.sender.count() != 1 {
+		t.Fatal("durable send intent not recorded before successful retry", ct, err)
+	}
+}
+
 func (b *breakingProgressRepo) RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error {
 	if b.breakStamp {
 		return errors.New("simulated transient database error on the progress write")

internal/tasks/campaign_task.go

@@ -504,8 +504,10 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) (result *errx
 	// This allows the tracking consumer to find the correct contact/sequence when
 	// processing open/click events from the tracking pixel service
 	if err := s.taskRepo.UpdateCampaignTaskTracking(ctx, taskID, contact.ID, sequence.ID); err != nil {
-		// Log but don't fail - tracking can still work via fallback methods
-		log.Warn().Err(err).Str("campaign_id", campaign.ID.String()).Str("task_id", taskID.String()).Msg("Failed to update campaign task tracking")
+		errs.CaptureException(err)
+		s.retryCampaignTickLater(ctx, taskRecord)
+		executionStatus = "failed"
+		return errx.InternalError()
 	}
 
 	// stop_on_reply is enforced inside FindRoutedPairs (STEP 6), and it is now