feat: persist monotonic campaign dispatch intent and require locked verified wakeup provenance for every dead-letter replay path

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

@@ -324,7 +324,7 @@ Returns a `data` array of dead-letter records. This endpoint is not cursor-pagin
 
 `POST /tasks/dlq/:id/replay`
 
-Replays only a non-sending campaign wakeup pass, using its campaign lock. Send-capable tasks, including legacy warmup and campaign contact sends, return `409 Conflict` and remain in the dead-letter queue: a retry date or missing send marker is not proof that the provider did not send. This route 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.
 
 Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns`
 

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

@@ -38,6 +38,8 @@ type replayTasks struct {
 	campaignID *uuid.UUID
 	contactID  *uuid.UUID
 	sequenceID *uuid.UUID
+	intent     repository.CampaignDispatchIntent
+	replayErr  error
 	ctErr      error
 	created    bool
 	inserted   int
@@ -58,25 +60,28 @@ 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, ContactID: t.contactID, SequenceID: t.sequenceID}, 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}, campaignID: &campaignID, created: true}
+			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
 			}
@@ -104,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
 }
@@ -131,6 +146,7 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) {
 	for _, tc := range []struct {
 		name         string
 		ctErr        error
+		replayErr    error
 		created      bool
 		queueFails   bool
 		wantErr      bool
@@ -142,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()}
+			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, created: tc.created}
+			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

@@ -2311,25 +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.ContactID != nil || ct.SequenceID != nil {
+	if ct == nil || ct.DispatchIntent != repository.CampaignDispatchWakeup || ct.ContactID != nil || ct.SequenceID != nil {
 		return false, false, nil
 	}
 	if ct.CampaignID == nil {
-		// The campaign is gone; there is nothing to replay into.
-		return false, true, 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
 	}

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

@@ -1317,13 +1317,11 @@ func (r *advancedOutreachRepository) ListRetryableDeadLetters(ctx context.Contex
 		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
-		JOIN campaign_tasks ct ON ct.task_id = t.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 d.task_type = 'campaign' AND t.task_type = 'campaign'
-		  AND ct.contact_id IS NULL AND ct.sequence_id IS NULL
+		  AND ` + safeDLQCampaignPass + `
 		ORDER BY d.next_retry_at ASC
 		LIMIT $1
 	`

internal/repository/pg_monitoring.go

@@ -29,8 +29,6 @@ type monitoringSection struct {
 	queries     []monitoringQuery
 }
 
-const safeDLQCampaignPass = `d.task_type='campaign' AND t.task_type='campaign' AND EXISTS (SELECT 1 FROM campaign_tasks ct WHERE ct.task_id=t.id AND ct.contact_id IS NULL AND ct.sequence_id IS NULL)`
-
 func (r *MonitoringRepository) Sources() []monitoring.Source {
 	out := make([]monitoring.Source, 0, len(monitoringSections)+3)
 	for _, id := range monitoringSectionOrder {

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)
@@ -31,8 +31,11 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) {
 	}
 	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM task_dead_letters WHERE task_id=$1`, id) })
 	passID := uuid.New()
-	f.exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id)VALUES($1,'campaign',$2,'dead_lettered','')`, passID, f.account)
-	f.exec(`INSERT INTO campaign_tasks(task_id) VALUES($1)`, passID)
+	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)
@@ -41,14 +44,20 @@ func TestLiveMonitoringQueueSettlementAndFailureSemantics(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if len(listed) == 0 || listed[0].TaskID != passID {
-		t.Fatal("replayable campaign pass was missing or an unverified send was chosen")
-	}
+	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) })

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