From ba1c6e3f3e48876a74726ccc6acd6c3ab2e7d7ef Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Sat, 10 Oct 2026 13:23:12 +0000
Subject: [PATCH 1/6] feat: bound local task dispatch across polling rounds
 using effective database pool capacity without stamping queued work

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 cmd/backend/main.go                           |  4 +-
 .../docs/development/configuration.mdx        |  2 +
 internal/tasksched/local.go                   | 46 ++++++++++-
 internal/tasksched/local_test.go              | 81 +++++++++++++++++++
 4 files changed, 128 insertions(+), 5 deletions(-)

diff --git a/cmd/backend/main.go b/cmd/backend/main.go
index c6cb9b166..26485e4b3 100644
--- a/cmd/backend/main.go
+++ b/cmd/backend/main.go
@@ -1398,7 +1398,9 @@ func main() {
 			}
 			tasksClient = gclient
 		} else {
-			localTasks = tasksched.NewLocal(taskRepository, localTasksPollInterval(), 0)
+			// Leave pool headroom for HTTP admission and result reconciliation.
+			localTasks = tasksched.NewLocal(taskRepository, localTasksPollInterval(), 0,
+				tasksched.WithPoolCapacity(int(primaryDB.Config().MaxConns)))
 			tasksClient = localTasks
 		}
 
diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx
index 74ffe8077..bf22c0a06 100644
--- a/docs/content/docs/development/configuration.mdx
+++ b/docs/content/docs/development/configuration.mdx
@@ -685,6 +685,8 @@ Per-mailbox `test_mode`, `test_send_enabled` and `test_receive_enabled` separate
 | `BILLING_PROVIDER` | `none` (every feature unlocked, no trial expiry; the dashboard reports the workspace as self-hosted rather than on a free tier and hides billing) or `stripe` | `none` |
 | `STRIPE_SECRET_KEY`, `STRIPE_WEBHOOK_SECRET`, `STRIPE_PUBLISHABLE_KEY` | Required together when `BILLING_PROVIDER=stripe`. The backend exits at boot if any is missing | unset |
 
+The local task dispatcher runs at most eight handlers at once, lowered to one third of the backend's effective Postgres pool ceiling with a minimum of one. This leaves room for API requests and reconciliation. Work beyond that limit stays pending in Postgres without a dispatch attempt or failure stamp. The ceiling applies across polling rounds, not separately to each batch, and does not change sending authority or unknown-outcome protections.
+
 Stripe Tax also needs account-side setup. In **Tax > Settings**, confirm the head office address, use the business-use SaaS product tax code (`txcd_10103001`) as the default, and choose whether prices include tax or have tax added at checkout. Apply the same tax code and an explicit tax behavior to every subscription and credit-pack product and price. A price set to inclusive or exclusive cannot be changed later, so replacing that choice requires a new Price and an update to the matching plan or `STRIPE_CREDIT_PACK_*_PRICE_ID` value.
 
 Add a Tax registration only after the business is registered with that tax authority. Automatic tax returns zero with a `not_collecting` reason in jurisdictions where no active registration applies. Enabling Stripe Tax is not a substitute for registering or filing.
diff --git a/internal/tasksched/local.go b/internal/tasksched/local.go
index 793b5860a..7bc6521cd 100644
--- a/internal/tasksched/local.go
+++ b/internal/tasksched/local.go
@@ -18,7 +18,24 @@ type DueTaskLister interface {
 	RecordDispatchAttempt(ctx context.Context, taskID uuid.UUID, failed bool, retryAt time.Time) error
 }
 
-const dispatchRetryDelay = 5 * time.Minute
+const (
+	dispatchRetryDelay      = 5 * time.Minute
+	defaultLocalConcurrency = 8
+)
+
+type LocalOption func(*Local)
+
+func WithPoolCapacity(maxConns int) LocalOption {
+	return WithConcurrency(max(1, min(defaultLocalConcurrency, maxConns/3)))
+}
+
+func WithConcurrency(limit int) LocalOption {
+	return func(l *Local) {
+		if limit > 0 {
+			l.slots = make(chan struct{}, limit)
+		}
+	}
+}
 
 // Local is the no-cloud Scheduler. Task rows already exist in Postgres (the
 // caller inserts them before CreateTask), and Run polls for due rows and fires
@@ -27,18 +44,23 @@ type Local struct {
 	repo     DueTaskLister
 	interval time.Duration
 	batch    int
+	slots    chan struct{}
 	inflight sync.Map // taskID string -> struct{}, dedups across overlapping ticks
 }
 
 // NewLocal builds a Local poller. interval<=0 defaults to 1s; batch<=0 to 200.
-func NewLocal(repo DueTaskLister, interval time.Duration, batch int) *Local {
+func NewLocal(repo DueTaskLister, interval time.Duration, batch int, options ...LocalOption) *Local {
 	if interval <= 0 {
 		interval = time.Second
 	}
 	if batch <= 0 {
 		batch = 200
 	}
-	return &Local{repo: repo, interval: interval, batch: batch}
+	l := &Local{repo: repo, interval: interval, batch: batch, slots: make(chan struct{}, defaultLocalConcurrency)}
+	for _, option := range options {
+		option(l)
+	}
+	return l
 }
 
 // CreateTask is a no-op enqueue: the row already exists with status=pending and
@@ -60,7 +82,7 @@ func (l *Local) DeleteTask(_ context.Context, _ string) error { return nil }
 func (l *Local) Run(ctx context.Context, handle func(taskID string) error) {
 	ticker := time.NewTicker(l.interval)
 	defer ticker.Stop()
-	log.Info().Dur("interval", l.interval).Int("batch", l.batch).Msg("tasksched: local dispatcher started")
+	log.Info().Dur("interval", l.interval).Int("batch", l.batch).Int("concurrency", cap(l.slots)).Msg("tasksched: local dispatcher started")
 	for {
 		select {
 		case <-ctx.Done():
@@ -72,18 +94,34 @@ func (l *Local) Run(ctx context.Context, handle func(taskID string) error) {
 }
 
 func (l *Local) tick(ctx context.Context, handle func(taskID string) error) {
+	if ctx.Err() != nil || len(l.slots) == cap(l.slots) {
+		return
+	}
 	ids, err := l.repo.ListDuePendingTaskIDs(ctx, l.batch)
 	if err != nil {
 		log.Error().Err(err).Msg("tasksched: list due tasks")
 		return
 	}
 	for _, id := range ids {
+		if ctx.Err() != nil {
+			return
+		}
+		select {
+		case l.slots <- struct{}{}:
+		default:
+			return
+		}
 		key := id.String()
 		if _, busy := l.inflight.LoadOrStore(key, struct{}{}); busy {
+			<-l.slots
 			continue
 		}
 		go func(taskID string) {
+			defer func() { <-l.slots }()
 			defer l.inflight.Delete(taskID)
+			if ctx.Err() != nil {
+				return
+			}
 			err := handle(taskID)
 			if ctx.Err() != nil {
 				return
diff --git a/internal/tasksched/local_test.go b/internal/tasksched/local_test.go
index 454dad648..f92685337 100644
--- a/internal/tasksched/local_test.go
+++ b/internal/tasksched/local_test.go
@@ -95,3 +95,84 @@ func TestLocalInflightIsNotDispatchedTwice(t *testing.T) {
 		t.Fatal("attempt did not finish")
 	}
 }
+
+func TestLocalConcurrencyUsesEffectivePoolCapacity(t *testing.T) {
+	for _, tc := range []struct {
+		pool, want int
+	}{{25, 8}, {4, 1}, {12, 4}, {100, 8}, {1, 1}} {
+		l := NewLocal(nil, 0, 0, WithPoolCapacity(tc.pool))
+		if got := cap(l.slots); got != tc.want {
+			t.Fatalf("pool %d: concurrency %d, want %d", tc.pool, got, tc.want)
+		}
+	}
+}
+
+func TestLocalExcessWorkStaysQueuedAcrossTicks(t *testing.T) {
+	ids := []uuid.UUID{uuid.New(), uuid.New(), uuid.New()}
+	r := &localTestRepo{ids: ids, retries: make(map[uuid.UUID]bool), attempts: make(chan uuid.UUID, 3)}
+	l := NewLocal(r, time.Millisecond, 200, WithConcurrency(2))
+	entered, release := make(chan string, 3), make(chan struct{})
+	var closeOnce sync.Once
+	t.Cleanup(func() { closeOnce.Do(func() { close(release) }) })
+	handle := func(id string) error { entered <- id; <-release; return nil }
+	for round := 0; round < 3; round++ {
+		l.tick(t.Context(), handle)
+	}
+	for range 2 {
+		select {
+		case <-entered:
+		case <-time.After(time.Second):
+			t.Fatal("available dispatch slot was not used")
+		}
+	}
+	select {
+	case <-entered:
+		t.Fatal("excess task dispatched while both slots were occupied")
+	case <-r.attempts:
+		t.Fatal("queued task stamped before a handler completed")
+	default:
+	}
+	closeOnce.Do(func() { close(release) })
+	for range 2 {
+		select {
+		case <-r.attempts:
+		case <-time.After(time.Second):
+			t.Fatal("dispatch attempt did not finish")
+		}
+	}
+	r.mu.Lock()
+	r.ids = ids[2:]
+	_, stamped := r.retries[ids[2]]
+	r.mu.Unlock()
+	if stamped {
+		t.Fatal("task that never started received an attempt stamp")
+	}
+	deadline := time.NewTimer(time.Second)
+	defer deadline.Stop()
+	tick := time.NewTicker(time.Millisecond)
+	defer tick.Stop()
+	for {
+		l.tick(t.Context(), handle)
+		select {
+		case id := <-entered:
+			if id != ids[2].String() {
+				t.Fatalf("wrong queued task started: %s", id)
+			}
+			return
+		case <-deadline.C:
+			t.Fatal("queued task did not use a released slot")
+		case <-tick.C:
+		}
+	}
+}
+
+func TestLocalCancelledTickDoesNotDispatchOrStamp(t *testing.T) {
+	r := &localTestRepo{ids: []uuid.UUID{uuid.New()}, retries: make(map[uuid.UUID]bool), attempts: make(chan uuid.UUID, 1)}
+	l := NewLocal(r, 0, 0)
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	l.tick(ctx, func(string) error { t.Error("cancelled dispatcher started a task"); return nil })
+	if len(l.slots) != 0 || len(r.retries) != 0 {
+		t.Fatal("cancelled tick claimed or stamped pending work")
+	}
+}

From 2db20352e7438f759f68e69c4738a90598b333be Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Sat, 10 Oct 2026 13:31:57 +0000
Subject: [PATCH 2/6] feat: bound campaign reply repair to raw message pages
 without losing eligible replies

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 internal/app/consumer/reply_repair.go         |  7 +-
 internal/app/consumer/reply_repair_test.go    | 36 +++++++++-
 internal/repository/pg_unibox.go              |  2 +-
 .../pg_unibox_reply_repair_live_test.go       | 65 ++++++++++++++++++
 internal/repository/pg_unibox_warmup.go       | 68 +++++++++++++++----
 5 files changed, 157 insertions(+), 21 deletions(-)
 create mode 100644 internal/repository/pg_unibox_reply_repair_live_test.go

diff --git a/internal/app/consumer/reply_repair.go b/internal/app/consumer/reply_repair.go
index 23f07ed11..041ff398c 100644
--- a/internal/app/consumer/reply_repair.go
+++ b/internal/app/consumer/reply_repair.go
@@ -47,11 +47,11 @@ func (s *JobsService) StartIncomingReplyRepair(ctx context.Context) {
 
 func (s *JobsService) repairIncomingReplyBatch(ctx context.Context, afterID uuid.UUID) (uuid.UUID, bool, error) {
 	const batchSize = 100
-	events, err := s.UniboxRepository.ListUnprocessedCampaignReplies(ctx, time.Now().Add(-replyRepairWindow), afterID, batchSize)
+	page, err := s.UniboxRepository.ListUnprocessedCampaignReplies(ctx, time.Now().Add(-replyRepairWindow), afterID, batchSize)
 	if err != nil {
 		return afterID, false, err
 	}
-	for _, e := range events {
+	for _, e := range page.Events {
 		if e.Message == nil {
 			continue
 		}
@@ -63,9 +63,8 @@ func (s *JobsService) repairIncomingReplyBatch(ctx context.Context, afterID uuid
 				Str("message_id", e.Message.MessageID).
 				Msg("incoming reply repair: reply processing failed")
 		}
-		afterID = e.Message.ID
 	}
-	return afterID, len(events) < batchSize, nil
+	return page.NextCursor, page.Done, nil
 }
 
 // StartReplyOptOutRecheck re-reads, once, every suppression a reply opt-out
diff --git a/internal/app/consumer/reply_repair_test.go b/internal/app/consumer/reply_repair_test.go
index ad39df9d1..4759a148a 100644
--- a/internal/app/consumer/reply_repair_test.go
+++ b/internal/app/consumer/reply_repair_test.go
@@ -16,11 +16,17 @@ import (
 type repairInbox struct {
 	repository.UniboxRepository
 	events []models.JobEventNewEmail
+	pages  []repository.CampaignReplyRepairPage
 	since  time.Time
 }
 
-func (r *repairInbox) ListUnprocessedCampaignReplies(_ context.Context, since time.Time, afterID uuid.UUID, limit int) ([]models.JobEventNewEmail, error) {
+func (r *repairInbox) ListUnprocessedCampaignReplies(_ context.Context, since time.Time, afterID uuid.UUID, limit int) (repository.CampaignReplyRepairPage, error) {
 	r.since = since
+	if len(r.pages) > 0 {
+		page := r.pages[0]
+		r.pages = r.pages[1:]
+		return page, nil
+	}
 	// Ordered by id like the query, so the cursor is deterministic.
 	var out []models.JobEventNewEmail
 	for _, e := range r.events {
@@ -29,7 +35,12 @@ func (r *repairInbox) ListUnprocessedCampaignReplies(_ context.Context, since ti
 		}
 	}
 	sort.Slice(out, func(i, j int) bool { return out[i].Message.ID.String() < out[j].Message.ID.String() })
-	return out[:min(len(out), limit)], nil
+	out = out[:min(len(out), limit)]
+	page := repository.CampaignReplyRepairPage{Events: out, NextCursor: afterID, Done: len(out) < limit}
+	if len(out) > 0 {
+		page.NextCursor = out[len(out)-1].Message.ID
+	}
+	return page, nil
 }
 
 type repairAdvanced struct {
@@ -87,3 +98,24 @@ func TestIncomingReplyRepairNeedsBothCollaborators(t *testing.T) {
 	(&JobsService{UniboxRepository: &repairInbox{}}).StartIncomingReplyRepair(context.Background())
 	_ = errors.New
 }
+
+func TestIncomingReplyRepairKeepsScanningWhenAFullRawPageHasNoMatches(t *testing.T) {
+	first := uuid.New()
+	second := uuid.New()
+	event := models.JobEventNewEmail{Message: &models.EmailMessageStoreData{ID: second, EmailID: uuid.New()}}
+	inbox := &repairInbox{pages: []repository.CampaignReplyRepairPage{
+		{NextCursor: first},
+		{Events: []models.JobEventNewEmail{event}, NextCursor: second, Done: true},
+	}}
+	adv := &repairAdvanced{}
+	s := &JobsService{UniboxRepository: inbox, AdvancedService: adv}
+
+	next, done, err := s.repairIncomingReplyBatch(context.Background(), uuid.Nil)
+	if err != nil || done || next != first || len(adv.seen) != 0 {
+		t.Fatalf("first raw page: next=%v done=%t seen=%d err=%v", next, done, len(adv.seen), err)
+	}
+	next, done, err = s.repairIncomingReplyBatch(context.Background(), next)
+	if err != nil || !done || next != second || len(adv.seen) != 1 || adv.seen[0] != second {
+		t.Fatalf("second raw page: next=%v done=%t seen=%v err=%v", next, done, adv.seen, err)
+	}
+}
diff --git a/internal/repository/pg_unibox.go b/internal/repository/pg_unibox.go
index 095176c26..f5c3533f6 100644
--- a/internal/repository/pg_unibox.go
+++ b/internal/repository/pg_unibox.go
@@ -87,7 +87,7 @@ type UniboxRepository interface {
 	// ListUnprocessedCampaignReplies pages inbound messages that reply
 	// processing never claimed and that look like campaign replies: they
 	// answer a campaign send, or come from one of the workspace's contacts.
-	ListUnprocessedCampaignReplies(ctx context.Context, since time.Time, afterID uuid.UUID, limit int) ([]models.JobEventNewEmail, error)
+	ListUnprocessedCampaignReplies(ctx context.Context, since time.Time, afterID uuid.UUID, limit int) (CampaignReplyRepairPage, error)
 	// ListInboundFrom returns a workspace's inbound mail from one address,
 	// newest first: the evidence behind an opt-out that address triggered.
 	ListInboundFrom(ctx context.Context, orgID uuid.UUID, address string, limit int) ([]InboundMessage, error)
diff --git a/internal/repository/pg_unibox_reply_repair_live_test.go b/internal/repository/pg_unibox_reply_repair_live_test.go
new file mode 100644
index 000000000..dddab115e
--- /dev/null
+++ b/internal/repository/pg_unibox_reply_repair_live_test.go
@@ -0,0 +1,65 @@
+package repository
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+func TestLiveCampaignReplyRepairDoesNotSkipMatchingMessagesAcrossRawPages(t *testing.T) {
+	handle := liveUniboxFolderDB(t)
+	f := newUniboxFolderFixture(t, handle.Pool)
+	repo := NewUniboxRepository(handle)
+	ctx := context.Background()
+	contactID := uuid.New()
+	if _, err := handle.Pool.Exec(ctx, `INSERT INTO contacts (id, user_id, organization_id, first_name, last_name, email, company, phone, custom_fields)
+		VALUES ($1, $2, $3, '', '', 'eligible@test.local', '', '', '{}'::jsonb)`, contactID, f.user, f.org); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = handle.Pool.Exec(context.Background(), `DELETE FROM contacts WHERE id=$1`, contactID) })
+
+	prefix := uuid.New().String()[:14] + "4000-8000-"
+	id := func(i int) uuid.UUID { return uuid.MustParse(prefix + fmt.Sprintf("%012x", i)) }
+	var values [][]any
+	for i := 1; i <= 502; i++ {
+		folder := "inbox"
+		from := "not-a-contact@test.local"
+		if i == 501 || i == 502 {
+			from = "Name (eligible@test.local)"
+		}
+		if i == 502 {
+			folder = "sent"
+		}
+		values = append(values, []any{id(i), f.user, f.mailbox, time.Now().UTC(), folder, folder, []string{from}})
+	}
+	if _, err := handle.Pool.CopyFrom(ctx, pgx.Identifier{"unibox_emails"},
+		[]string{"id", "user_id", "email_id", "created_at", "folder", "provider_folder", "from_addr"}, pgx.CopyFromRows(values)); err != nil {
+		t.Fatal(err)
+	}
+	since := time.Now().Add(-time.Hour)
+	first, err := repo.ListUnprocessedCampaignReplies(ctx, since, id(0), 100)
+	if err != nil || first.Done || first.NextCursor != id(500) || len(first.Events) != 0 {
+		t.Fatalf("first page: next=%v done=%t matches=%d err=%v", first.NextCursor, first.Done, len(first.Events), err)
+	}
+	second, err := repo.ListUnprocessedCampaignReplies(ctx, since, first.NextCursor, 100)
+	if err != nil || !second.Done || second.NextCursor != id(501) || len(second.Events) != 1 || second.Events[0].Message.ID != id(501) {
+		t.Fatalf("second page: next=%v done=%t matches=%d err=%v", second.NextCursor, second.Done, len(second.Events), err)
+	}
+
+	if _, err := handle.Pool.Exec(ctx, `UPDATE unibox_emails SET from_addr=ARRAY['eligible@test.local']
+		WHERE email_id=$1 AND id >= $2 AND id <= $3`, f.mailbox, id(1), id(110)); err != nil {
+		t.Fatal(err)
+	}
+	dense, err := repo.ListUnprocessedCampaignReplies(ctx, since, id(0), 100)
+	if err != nil || dense.Done || dense.NextCursor != id(100) || len(dense.Events) != 100 {
+		t.Fatalf("full result page: next=%v done=%t matches=%d err=%v", dense.NextCursor, dense.Done, len(dense.Events), err)
+	}
+	remainder, err := repo.ListUnprocessedCampaignReplies(ctx, since, dense.NextCursor, 100)
+	if err != nil || !remainder.Done || remainder.NextCursor != id(501) || len(remainder.Events) != 11 {
+		t.Fatalf("remaining page: next=%v done=%t matches=%d err=%v", remainder.NextCursor, remainder.Done, len(remainder.Events), err)
+	}
+}
diff --git a/internal/repository/pg_unibox_warmup.go b/internal/repository/pg_unibox_warmup.go
index 6ab444074..2bd92d736 100644
--- a/internal/repository/pg_unibox_warmup.go
+++ b/internal/repository/pg_unibox_warmup.go
@@ -141,13 +141,45 @@ func (r *uniboxRepository) UpdatePendingEmail(ctx context.Context, userID, id uu
 	return true, tx.Commit(ctx)
 }
 
-// ListUnprocessedCampaignReplies feeds the repair sweep in the consumer. A
-// reply that reply processing refused before claiming (an address check that
-// failed, the automation switch) has campaign_reply_processed_at NULL, so
-// this is exactly the set that can still be attributed; everything already
-// decided is excluded by that column. Bounded to `since` because older mail
-// was processed by the code of its day.
-func (r *uniboxRepository) ListUnprocessedCampaignReplies(ctx context.Context, since time.Time, afterID uuid.UUID, limit int) ([]models.JobEventNewEmail, error) {
+type CampaignReplyRepairPage struct {
+	Events     []models.JobEventNewEmail
+	NextCursor uuid.UUID
+	Done       bool
+}
+
+// ListUnprocessedCampaignReplies bounds the raw scan as well as the replies returned.
+func (r *uniboxRepository) ListUnprocessedCampaignReplies(ctx context.Context, since time.Time, afterID uuid.UUID, limit int) (CampaignReplyRepairPage, error) {
+	const scanSize = 500
+	page := CampaignReplyRepairPage{NextCursor: afterID}
+	if limit <= 0 {
+		return page, fmt.Errorf("campaign reply repair requires a positive result limit")
+	}
+	scan, err := r.db.Query(ctx, `SELECT u.id FROM unibox_emails u
+		WHERE u.id > $1 AND u.created_at >= $2 AND u.campaign_reply_processed_at IS NULL
+		AND u.folder NOT IN ('sent', 'drafts') AND u.provider_folder NOT IN ('sent', 'drafts')
+		ORDER BY u.id LIMIT $3`, afterID, since, scanSize)
+	if err != nil {
+		return page, err
+	}
+	var scanned int
+	var lastScanned uuid.UUID
+	for scan.Next() {
+		if err := scan.Scan(&lastScanned); err != nil {
+			scan.Close()
+			return page, err
+		}
+		scanned++
+	}
+	err = scan.Err()
+	scan.Close()
+	if err != nil {
+		return page, err
+	}
+	if scanned == 0 {
+		page.Done = true
+		return page, nil
+	}
+
 	// The sender in whichever form the sync stored it: "Name <addr>",
 	// "Name (addr)" or bare.
 	const bareFrom = `lower(COALESCE(
@@ -159,7 +191,7 @@ func (r *uniboxRepository) ListUnprocessedCampaignReplies(ctx context.Context, s
 		       u.reply_to, u.in_reply_to, u.subject, u.snippet, u.body_text, u.folder, u.provider_folder,
 		       u.gmail_id, u.uid, u.mailbox, u.folder_path, u.internal_date
 		FROM unibox_emails u
-		WHERE u.id > $1
+		WHERE u.id > $1 AND u.id <= $3
 		  AND u.created_at >= $2
 		  AND u.campaign_reply_processed_at IS NULL
 		  AND u.folder NOT IN ('sent', 'drafts')
@@ -180,23 +212,31 @@ func (r *uniboxRepository) ListUnprocessedCampaignReplies(ctx context.Context, s
 		        AND lower(co.email) = `+bareFrom+`
 		    )
 		  )
-		ORDER BY u.id LIMIT $3`, afterID, since, limit)
+		ORDER BY u.id LIMIT $4`, afterID, since, lastScanned, limit)
 	if err != nil {
-		return nil, err
+		return page, err
 	}
 	defer rows.Close()
-	var events []models.JobEventNewEmail
 	for rows.Next() {
 		e := models.JobEventNewEmail{Message: &models.EmailMessageStoreData{}}
 		m := e.Message
 		if err := rows.Scan(&e.UserID, &m.ID, &m.EmailID, &m.MessageID, &m.ThreadID, &m.Flags, &m.FromAddr, &m.ToAddr, &m.CC, &m.BCC,
 			&m.ReplyTo, &m.InReplyTo, &m.Subject, &m.Snippet, &m.BodyText, &m.Folder, &m.ProviderFolder,
 			&m.GmailID, &m.UID, &m.Mailbox, &m.FolderPath, &m.InternalDate); err != nil {
-			return nil, err
+			return page, err
 		}
-		events = append(events, e)
+		page.Events = append(page.Events, e)
 	}
-	return events, rows.Err()
+	if err := rows.Err(); err != nil {
+		return page, err
+	}
+	page.Done = scanned < scanSize && len(page.Events) < limit
+	if len(page.Events) == limit {
+		page.NextCursor = page.Events[len(page.Events)-1].Message.ID
+	} else {
+		page.NextCursor = lastScanned
+	}
+	return page, nil
 }
 
 // InboundMessage is a stored inbound message and when reply processing

From 3f8e951bd6a2e979348156ee5575e59e42b9da9f Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Sat, 10 Oct 2026 13:38:43 +0000
Subject: [PATCH 3/6] feat: guard legacy send dead-letter replay and expose
 unverified recovery evidence

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 .../docs/api/reference/deliverability-ops.mdx |  2 +-
 docs/content/docs/development/admin-panel.mdx |  2 +-
 .../docs/development/instance-health.mdx      |  2 +-
 docs/content/docs/development/warmblyctl.mdx  |  2 +-
 internal/app/advanced/campaign_replay_test.go | 46 ++++++++++++-
 internal/app/advanced/service.go              | 64 ++++++-------------
 internal/repository/pg_advanced_outreach.go   | 18 ++++--
 internal/repository/pg_monitoring.go          |  9 ++-
 .../pg_monitoring_queues_live_test.go         | 25 +++++++-
 9 files changed, 107 insertions(+), 63 deletions(-)

diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx
index 2ec34e0e7..82eb08043 100644
--- a/docs/content/docs/api/reference/deliverability-ops.mdx
+++ b/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`
 
-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 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.
 
 Auth: **Scope** `SEND_CAMPAIGNS` · **Org permission** `send_campaigns`
 
diff --git a/docs/content/docs/development/admin-panel.mdx b/docs/content/docs/development/admin-panel.mdx
index 794d109cd..80599c185 100644
--- a/docs/content/docs/development/admin-panel.mdx
+++ b/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
 
diff --git a/docs/content/docs/development/instance-health.mdx b/docs/content/docs/development/instance-health.mdx
index a413b3ac3..832ba11ec 100644
--- a/docs/content/docs/development/instance-health.mdx
+++ b/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.
 
diff --git a/docs/content/docs/development/warmblyctl.mdx b/docs/content/docs/development/warmblyctl.mdx
index 083f681f3..c01f173a9 100644
--- a/docs/content/docs/development/warmblyctl.mdx
+++ b/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 \
diff --git a/internal/app/advanced/campaign_replay_test.go b/internal/app/advanced/campaign_replay_test.go
index da8827668..d5144fc36 100644
--- a/internal/app/advanced/campaign_replay_test.go
+++ b/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,8 @@ type replayTasks struct {
 	repository.TaskRepository
 	task       *repository.Task
 	campaignID *uuid.UUID
+	contactID  *uuid.UUID
+	sequenceID *uuid.UUID
 	ctErr      error
 	created    bool
 	inserted   int
@@ -52,7 +58,43 @@ 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}, nil
+}
+
+func TestDeadLetterNeverReplaysSendTasksWithoutAuthority(t *testing.T) {
+	for _, tc := range []struct {
+		name, dlqType, taskType string
+		contact, sequence       bool
+	}{
+		{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"},
+	} {
+		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}
+			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) {
@@ -103,7 +145,7 @@ func TestReplayDeadLetterForACampaignPass(t *testing.T) {
 	} {
 		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}}
+			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}
 			svc := &service{repo: repo, taskRepo: tasks, tasksClient: replayClient{fail: tc.queueFails}}
 
diff --git a/internal/app/advanced/service.go b/internal/app/advanced/service.go
index fe830727b..bb2d1eb64 100644
--- a/internal/app/advanced/service.go
+++ b/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)
@@ -2329,7 +2318,10 @@ func (s *service) replayCampaignPass(ctx context.Context, task *repository.Task)
 	if err != nil {
 		return false, true, err
 	}
-	if ct == nil || ct.CampaignID == nil {
+	if ct == nil || 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
 	}
@@ -2680,10 +2672,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 +2696,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
diff --git a/internal/repository/pg_advanced_outreach.go b/internal/repository/pg_advanced_outreach.go
index 90375a124..624f66e7a 100644
--- a/internal/repository/pg_advanced_outreach.go
+++ b/internal/repository/pg_advanced_outreach.go
@@ -1314,13 +1314,17 @@ 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
+		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
+		ORDER BY d.next_retry_at ASC
 		LIMIT $1
 	`
 	rows, err := resultDB(ctx, r.db).Query(ctx, query, limit)
diff --git a/internal/repository/pg_monitoring.go b/internal/repository/pg_monitoring.go
index 080e9e093..9841eeb35 100644
--- a/internal/repository/pg_monitoring.go
+++ b/internal/repository/pg_monitoring.go
@@ -29,6 +29,8 @@ 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 {
@@ -238,10 +240,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`},
 	}},
diff --git a/internal/repository/pg_monitoring_queues_live_test.go b/internal/repository/pg_monitoring_queues_live_test.go
index 3cb2d4862..d3e4ecbc7 100644
--- a/internal/repository/pg_monitoring_queues_live_test.go
+++ b/internal/repository/pg_monitoring_queues_live_test.go
@@ -30,6 +30,25 @@ 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()
+	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)
+	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)
+	}
+	if len(listed) == 0 || listed[0].TaskID != passID {
+		t.Fatal("replayable campaign pass was missing or an unverified send was chosen")
+	}
+	for _, letter := range listed {
+		if letter.TaskID == id {
+			t.Fatal("legacy send was selected for automatic replay")
+		}
+	}
 	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 +69,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 +81,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")
 	}

From ab3f0c0ca4d4e593d4822adad128277dbe06f4b1 Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Sat, 10 Oct 2026 13:53:18 +0000
Subject: [PATCH 4/6] feat: preserve the unverified dead-letter monitoring
 identity in redacted administrator CLI output

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 cmd/warmblyctl/remote_output.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/cmd/warmblyctl/remote_output.go b/cmd/warmblyctl/remote_output.go
index e7ad9e866..5c3c13484 100644
--- a/cmd/warmblyctl/remote_output.go
+++ b/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

From a53de41c92589e10ab49f4eaa6cb77f139d73b38 Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Sat, 10 Oct 2026 14:16:01 +0000
Subject: [PATCH 5/6] feat: persist monotonic campaign dispatch intent and
 require locked verified wakeup provenance for every dead-letter replay path

---
 .../docs/api/reference/deliverability-ops.mdx |   2 +-
 .../docs/guides/workspace-export-import.mdx   |   2 +
 internal/app/advanced/campaign_replay_test.go |  25 ++-
 internal/app/advanced/service.go              |  14 +-
 internal/app/orgtransfer/export_test.go       |  13 ++
 internal/app/orgtransfer/spec.go              |   3 +-
 .../campaign_dispatch_intent_upgrade_test.go  | 124 ++++++++++++
 .../000282_campaign_dispatch_intent.down.sql  |  17 ++
 .../000282_campaign_dispatch_intent.up.sql    |  24 +++
 .../campaign_replay_intent_live_test.go       | 186 ++++++++++++++++++
 internal/repository/pg_advanced_outreach.go   |   4 +-
 internal/repository/pg_monitoring.go          |   2 -
 .../pg_monitoring_queues_live_test.go         |  23 ++-
 internal/repository/pg_task.go                |  69 ++++++-
 internal/tasks/campaign_send_live_test.go     |  41 ++++
 internal/tasks/campaign_task.go               |   6 +-
 16 files changed, 521 insertions(+), 34 deletions(-)
 create mode 100644 internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go
 create mode 100644 internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql
 create mode 100644 internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql
 create mode 100644 internal/repository/campaign_replay_intent_live_test.go

diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx
index 82eb08043..ff41364ba 100644
--- a/docs/content/docs/api/reference/deliverability-ops.mdx
+++ b/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`
 
diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx
index 316273817..abeea2c2a 100644
--- a/docs/content/docs/guides/workspace-export-import.mdx
+++ b/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>
diff --git a/internal/app/advanced/campaign_replay_test.go b/internal/app/advanced/campaign_replay_test.go
index d5144fc36..f8525b321 100644
--- a/internal/app/advanced/campaign_replay_test.go
+++ b/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)
diff --git a/internal/app/advanced/service.go b/internal/app/advanced/service.go
index bb2d1eb64..fc3078151 100644
--- a/internal/app/advanced/service.go
+++ b/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
 	}
diff --git a/internal/app/orgtransfer/export_test.go b/internal/app/orgtransfer/export_test.go
index 16dd91e41..7c5d636c8 100644
--- a/internal/app/orgtransfer/export_test.go
+++ b/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.
diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go
index 424ab46d9..59e04042c 100644
--- a/internal/app/orgtransfer/spec.go
+++ b/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,
diff --git a/internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go b/internal/infrastructure/db/campaign_dispatch_intent_upgrade_test.go
new file mode 100644
index 000000000..998c81237
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql b/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.down.sql
new file mode 100644
index 000000000..f09fad647
--- /dev/null
+++ b/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;
diff --git a/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql b/internal/infrastructure/db/migrations/000282_campaign_dispatch_intent.up.sql
new file mode 100644
index 000000000..dc63c36af
--- /dev/null
+++ b/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();
diff --git a/internal/repository/campaign_replay_intent_live_test.go b/internal/repository/campaign_replay_intent_live_test.go
new file mode 100644
index 000000000..8a83e0bca
--- /dev/null
+++ b/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)
+	}
+}
diff --git a/internal/repository/pg_advanced_outreach.go b/internal/repository/pg_advanced_outreach.go
index 624f66e7a..d28520ed3 100644
--- a/internal/repository/pg_advanced_outreach.go
+++ b/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
 	`
diff --git a/internal/repository/pg_monitoring.go b/internal/repository/pg_monitoring.go
index 9841eeb35..028f785d9 100644
--- a/internal/repository/pg_monitoring.go
+++ b/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 {
diff --git a/internal/repository/pg_monitoring_queues_live_test.go b/internal/repository/pg_monitoring_queues_live_test.go
index d3e4ecbc7..e8450d008 100644
--- a/internal/repository/pg_monitoring_queues_live_test.go
+++ b/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) })
diff --git a/internal/repository/pg_task.go b/internal/repository/pg_task.go
index e4a772ea2..13d60205b 100644
--- a/internal/repository/pg_task.go
+++ b/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
 }
 
diff --git a/internal/tasks/campaign_send_live_test.go b/internal/tasks/campaign_send_live_test.go
index ec5be6ae5..ebe6b1ac2 100644
--- a/internal/tasks/campaign_send_live_test.go
+++ b/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")
diff --git a/internal/tasks/campaign_task.go b/internal/tasks/campaign_task.go
index 2d4cf2a5c..6ee4d36ed 100644
--- a/internal/tasks/campaign_task.go
+++ b/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

From 5bf8bf6e5176689e2da192b89c898da3a503a14e Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Sat, 10 Oct 2026 14:25:16 +0000
Subject: [PATCH 6/6] feat: document the campaign dispatch provenance downgrade
 guard and roll-forward recovery contract

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
---
 docs/content/docs/api/reference/deliverability-ops.mdx | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/docs/content/docs/api/reference/deliverability-ops.mdx b/docs/content/docs/api/reference/deliverability-ops.mdx
index ff41364ba..46c84bbd8 100644
--- a/docs/content/docs/api/reference/deliverability-ops.mdx
+++ b/docs/content/docs/api/reference/deliverability-ops.mdx
@@ -326,6 +326,8 @@ Returns a `data` array of dead-letter records. This endpoint is not cursor-pagin
 
 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`
 
 | Parameter | In | Type | Description |