Merge pull request #162 from warmbly/fix/161-campaign-stuck-processing

Fix campaigns stuck on processing with no email sent

AGENTS.md

@@ -731,6 +731,17 @@ When extending anti-fraud logic in this repo:
 - keep worker-side abuse checks lightweight and infrastructure-backed
 - record enough structured evidence for admin review when a user or mailbox is blocked
 
+## Send Outcome Loop
+
+A campaign step is stamped sent (`campaign_contact_progress.sent_at`, task `completed`, daily counters) the moment the backend hands the `SEND_EMAIL` to a worker, before anything has left the mailbox. What keeps that honest is the worker's per-task result on `jobs.worker-events`: every send is answered with exactly one `EMAIL_SENT` or `EMAIL_FAILED`, and the consumer's `HandleEmailFailed` (`internal/app/consumer/event_send_result.go`) walks the stamp back (clears `sent_at`, counts the attempt, gives back the daily counters, logs to the campaign feed, reopens a campaign that completed meanwhile). Routing retries the step on the next tick and drops the lead as `failed` after `config.CampaignSendMaxAttempts`.
+
+Rules that follow from this:
+
+- a worker must always answer a `SEND_EMAIL` it acked; a failure path that returns without producing `EMAIL_FAILED` leaves the lead at "processing" forever. Use `failSend` in `event_send_email.go`
+- the per-task result is always `EMAIL_FAILED`; the typed account events (`EMAIL_AUTH_ERROR` and friends) are raised in addition and carry an `EmailErrorEvent`, never a `SendEmailResult`
+- the backend refuses to publish a send to a worker that is not heartbeating (`tasks.NewWorkerLiveness`), because a command queued for a dead worker is never executed and never answered
+- the campaign wizard (and `POST /campaigns` with `steps`) connects steps in order at creation. Routing has no implicit "next position": a step with no outgoing connection ends the flow
+
 ## Control Plane vs Execution Plane
 
 Prefer this split:

cmd/backend/main.go

@@ -1317,6 +1317,13 @@ func main() {
 			contactRepostory, aiDraftRepo, streamingPublisher,
 		))
 		emailSender := tasks.NewEmailSender(emailRepostory, eventsPublisher)
+		// Never hand a send to a worker that stopped heartbeating: nothing
+		// would execute it and nothing would report it, so the step would
+		// look sent forever. The worker reconciler re-places the mailbox and
+		// the dead-letter retry replays the task once it has.
+		if liveness, ok := emailSender.(interface{ WireWorkerLiveness(tasks.WorkerLiveness) }); ok {
+			liveness.WireWorkerLiveness(tasks.NewWorkerLiveness(workerRepository, cache))
+		}
 		tasksService = tasks.NewService(
 			tasksClient,
 			generationClient,

cmd/consumer/main.go

@@ -380,6 +380,11 @@ func main() {
 		AdminRepo:                   repository.NewAdminRepository(primaryDB.Pool),
 		AssignmentService:           workerAssignmentSvc,
 		Notifier:                    notificationService,
+		TaskRepo:                    taskRepo,
+		CampaignRepo:                campaignRepo,
+		CampaignProgressRepo:        campaignProgressRepo,
+		CampaignLogRepo:             repository.NewCampaignLogRepository(primaryDB),
+		ContactRepo:                 contactRepo,
 	}
 
 	jobsService.InitEvents()

cmd/worker/main.go

@@ -238,13 +238,19 @@ func runInternalHeartbeat(ctx context.Context, workerID uuid.UUID, bindIP string
 	client := &http.Client{Timeout: 10 * time.Second}
 	// reqCtx is separate from ctx so the farewell beat still sends after ctx is
 	// cancelled by the shutdown signal.
-	send := func(reqCtx context.Context, stopping bool) {
+	send := func(reqCtx context.Context, booted, stopping bool) {
 		payload := map[string]any{
 			"worker_id":   workerID.String(),
 			"bind_ip":     reportedIP,
 			"tier":        os.Getenv("WORKER_TIER"),
 			"egress_kind": os.Getenv("WORKER_EGRESS_KIND"),
 		}
+		if booted {
+			// Mailboxes live in memory only, so a fresh process holds none.
+			// The backend reloads this worker's mailboxes on this beat
+			// instead of leaving them to the reconciler's next pass.
+			payload["booted"] = true
+		}
 		if stopping {
 			payload["stopping"] = true
 		}
@@ -267,7 +273,7 @@ func runInternalHeartbeat(ctx context.Context, workerID uuid.UUID, bindIP string
 		}
 	}
 
-	send(ctx, false)
+	send(ctx, true, false)
 	ticker := time.NewTicker(90 * time.Second)
 	defer ticker.Stop()
 	for {
@@ -277,11 +283,11 @@ func runInternalHeartbeat(ctx context.Context, workerID uuid.UUID, bindIP string
 			// without this the row stays selectable until the heartbeat ages
 			// out, so placement keeps picking a worker that has exited.
 			byeCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
-			send(byeCtx, true)
+			send(byeCtx, false, true)
 			cancel()
 			return
 		case <-ticker.C:
-			send(ctx, false)
+			send(ctx, false, false)
 		}
 	}
 }

docs/content/docs/api/reference/campaigns.mdx

@@ -140,7 +140,7 @@ Create a campaign. Only `name` is required, every other field is optional and ap
 | `max_new_leads_per_day` | integer | no | New-lead throttle, `0` is unlimited. |
 | `prioritize_new_leads` | boolean | no | Prefer new leads in each send window. |
 | `tracking_domain` | string | no | Campaign-scoped tracking domain (honored only once verified). |
-| `sequences` | object[] | no | Initial sequence steps in order (see create sequence input below). |
+| `steps` | object[] | no | Initial sequence steps in order (see create sequence input below). They are connected in order: each step routes unconditionally to the next, waiting that step's `wait_after` days. The first step's `wait_after` defaults to `0`, follow-ups to `3`. |
 | `variants` | object[] | no | A/B variants for the first step (same shape as create A/B variant). |
 | `advanced_overrides` | object | no | Advanced outreach overrides, see [advanced settings](#get-advanced-settings). |
 
@@ -163,8 +163,9 @@ Create a campaign. Only `name` is required, every other field is optional and ap
   "ramp_start": 10,
   "ramp_increment": 2,
   "ramp_ceiling": 40,
-  "sequences": [
-    { "name": "Step 1", "subject": "Quick question, {{first_name}}", "body_plain": "Hi {{first_name}}...", "wait_after": 0 }
+  "steps": [
+    { "name": "Step 1", "subject": "Quick question, {{first_name}}", "body_plain": "Hi {{first_name}}...", "wait_after": 0 },
+    { "name": "Step 2", "subject": "", "body_plain": "Just bumping this...", "wait_after": 3 }
   ]
 }
 ```
@@ -829,8 +830,8 @@ All fields optional.
 | `body_html` | string | no | HTML body template. |
 | `body_sync` | boolean | no | Keep plain and HTML bodies in sync. |
 | `body_code` | boolean | no | Treat the body as raw code (no auto-formatting). |
-| `wait_after` | integer | no | Minutes to wait after this step before the next (the spacing model, there is no standalone wait node for email steps). |
-| `conditions` | object | no | Branching tree (`{branches: [...]}`). Send `{}` or empty branches to clear branching and fall back to linear progression. |
+| `wait_after` | integer | no | Days to wait before this step, counted from the contact's previous step (`0` to `60`). Spacing belongs to the target step, so there is no standalone wait node for email steps. |
+| `conditions` | object | no | The connections out of this step (`{branches: [...]}`), evaluated in order; a branch with no `conditions` is a plain "go there next" link. Routing follows connections only: a step with `{}` or no branches has no outgoing path and ends the flow for the contact. |
 | `kind` | string | no | `email` (default), `action`, or `wait`. |
 | `action` | object | no | Typed config for non-email nodes. `type` is the switch (`wait`, `add_tag`, `remove_tag`, `unsubscribe`, `notify`, `create_task`, `create_deal`, `move_deal_stage`, `run_automation`, `end`), the remaining fields are type-scoped. |
 
@@ -839,7 +840,7 @@ All fields optional.
   "name": "Step 2",
   "subject": "Following up, {{first_name}}",
   "body_plain": "Just bumping this...",
-  "wait_after": 2880,
+  "wait_after": 2,
   "conditions": {
     "branches": [
       {

docs/content/docs/api/reference/contacts.mdx

@@ -85,7 +85,7 @@ Returns a `data` array of contacts plus a `pagination` envelope.
 }
 ```
 
-When the search filters by exactly one campaign, each contact additionally carries a `campaign_lead` object with its processing state inside that campaign (`status`, `sent`, `opened`, `clicked`, `replied`, `bounced`, `current_step`, `last_activity_at`). The `status` derivation, highest priority first, is `unsubscribed` (not subscribed), then `bounced`, `replied`, `active` (at least one email sent), and `pending` (queued, nothing sent). The `lead_status` filter narrows to one of these buckets.
+When the search filters by exactly one campaign, each contact additionally carries a `campaign_lead` object with its processing state inside that campaign (`status`, `sent`, `opened`, `clicked`, `replied`, `bounced`, `current_step`, `last_activity_at`, and `failure_reason` when failed). The `status` derivation, highest priority first, is `unsubscribed` (not subscribed), then `bounced`, `replied`, `failed` (a step could not be sent after every retry; `failure_reason` carries the sending worker's reason), `completed` (every email step sent, no reply), `active` (some steps sent, more to send), and `pending` (queued, nothing sent). A step counts as sent only once the sending worker has delivered it to the mailbox provider; a send the worker could not complete is retried on the campaign's next pass and never shows as sent. The `lead_status` filter narrows to one of these buckets.
 
 When the search filters by exactly one campaign, the first page (no `cursor`) also includes a `lead_counts` object: per-status lead totals for that campaign, independent of the `lead_status` filter so every scope's total is available at once.
 
@@ -94,9 +94,11 @@ When the search filters by exactly one campaign, the first page (no `cursor`) al
   "lead_counts": {
     "total": 3140,
     "queued": 1980,
-    "processing": 940,
+    "processing": 910,
+    "completed": 27,
     "replied": 180,
     "bounced": 22,
+    "failed": 3,
     "unsubscribed": 18
   }
 }

docs/content/docs/development/events.mdx

@@ -61,6 +61,8 @@ Types: `NEW_EMAIL`, `INBOUND_BOUNCE`, `REMOVE_EMAIL`, `FLAGS_ADD`, `FLAGS_REMOVE
 
 The consumer registers one handler per type (`internal/app/consumer/events.go`); unregistered types are logged and acknowledged rather than redelivered.
 
+Every `SEND_EMAIL` is answered with exactly one per-task result: `EMAIL_SENT` (the provider accepted the message; the consumer records the wire Message-ID on the task) or `EMAIL_FAILED` (a `SendEmailResult` carrying the error code and message). The control plane stamps a campaign step sent when it hands the send to the worker, so `EMAIL_FAILED` is what walks that back: the consumer marks the task failed, clears the step's `sent_at` and counts the attempt on `campaign_contact_progress`, gives the send back to the campaign's daily counters, writes the failure to the campaign activity log, and reopens a campaign that completed while the send was in flight. After `CampaignSendMaxAttempts` (5) the lead is marked failed and routing drops it; a `RECIPIENT_REJECTED` result (refused at RCPT) skips the retries and is ingested as a bounce instead. Account-level conditions (`EMAIL_AUTH_ERROR`, `EMAIL_DISABLED`, `EMAIL_RATE_LIMITED`, `EMAIL_SERVER_ERROR`) are raised in addition to, never instead of, the per-task result; they carry an `EmailErrorEvent` and act on the mailbox. A worker that does not hold the mailbox yet leaves the send for a few redeliveries before reporting it failed, and the backend never publishes a send to a worker that is not heartbeating.
+
 `SYNC_STATE` is the worker's relay of a mailbox's sync state (backfill progress and cursor, fair-use throttle, last-synced time). It carries the full state rather than a delta, so a lost event is repaired by the next one; the consumer writes it to `email_sync_state` and the backend hands it back inside `ADD_EMAIL` on the next load, which is what lets a replaced worker resume an import instead of restarting it.
 
 ## Tracking events

docs/content/docs/guides/campaigns.mdx

@@ -11,7 +11,7 @@ You need at least one active mailbox and a contact list. Warm new mailboxes befo
 
 ## Create a campaign
 
-**Campaigns** > **New campaign** opens a four-step wizard: **Basics** (name, description), **Schedule** (timezone, sending days, hours), **Sending** (mailbox tags, daily limit per mailbox, stop on reply, open and click tracking, unsubscribe header), and **First email** (subject, body, optional follow-ups). Only a name is required; the first email can be skipped and written in the full editor on the Steps tab, and everything else can be changed later from settings. Closing a wizard with unsaved edits asks before discarding them.
+**Campaigns** > **New campaign** opens a four-step wizard: **Basics** (name, description), **Schedule** (timezone, sending days, hours), **Sending** (mailbox tags, daily limit per mailbox, stop on reply, open and click tracking, unsubscribe header), and **First email** (subject, body, optional follow-ups). Follow-ups added here are connected in order, each waiting the number of days you set after the previous step; open the Steps tab to rearrange, branch, or change the waits. Only a name is required; the first email can be skipped and written in the full editor on the Steps tab, and everything else can be changed later from settings. Closing a wizard with unsaved edits asks before discarding them.
 
 The campaign starts as a **draft**. Nothing sends until you start it.
 
@@ -68,10 +68,13 @@ The **Leads** tab takes contacts four ways. **From contacts** opens a picker ove
 | Done | Every step sent, no reply or bounce |
 | Replied | Contact replied; sending stops |
 | Bounced | A send hard-bounced |
+| Failed | A step could not be sent after every retry; hover the status for the reason |
 | Unsubscribed | Unsubscribed or suppressed |
 
 A lead is **Processing** only while steps remain, so a finished campaign reads as done rather than stuck mid-flight.
 
+A step counts as sent only once the sending worker has handed it to the mailbox provider. If the worker cannot send (the mailbox was still loading, the provider refused the message, a storage hiccup), the step goes back to the queue, the failure appears in **Needs attention** with the reason, and the next pass retries it. After five failed attempts the lead is marked **Failed** and dropped from the campaign, so a mailbox that can never send does not loop forever. A recipient the mail server refuses outright is not retried: it is recorded as a **Bounced** lead straight away and goes through the same bounce handling (suppression, guardrails, webhooks) as a bounce notice. A campaign that finished while a send was still in flight reopens to retry it.
+
 ## Scheduling
 
 Campaigns send only inside the **weekly sending windows** you define, in the campaign's timezone. Each day is independent, with different hours or several windows per day, set from presets like `Mon-Fri 9-5` or by dragging on the grid (which is Monday-first).
@@ -107,7 +110,7 @@ A paused campaign shows the reason and the numbers behind it on the **Auto-pause
 
 ## Live activity
 
-The detail view streams sends, opens, clicks, replies, bounces, and skips as they happen, for every teammate at once. A **Needs attention** panel above it surfaces failures such as a step that could not be scheduled; these retry automatically and stay visible so nothing fails silently.
+The detail view streams sends, opens, clicks, replies, bounces, and skips as they happen, for every teammate at once. A **Needs attention** panel above it surfaces failures such as a step that could not be scheduled, a send the worker could not complete, or a mailbox whose sending worker is offline; these retry automatically and stay visible so nothing fails silently.
 
 If scheduling stalls on a transient infrastructure hiccup, Warmbly re-seeds within a few minutes. Pausing and resuming forces a fresh start.
 

docs/content/docs/guides/sequences.mdx

@@ -9,7 +9,7 @@ A campaign sends a sequence of steps to each contact: the first email, then foll
 
 Steps lay out top to bottom, the first marked `Start`.
 
-- Steps never connect themselves. You drag a connector dot from one card to another.
+- Steps never connect themselves on the canvas. You drag a connector dot from one card to another. The only exception is the campaign wizard: follow-ups written there arrive already connected in order, with their waits set.
 - A step with no outgoing line ends the sequence for that contact (`Ends here`; unconnected paths show `Stop`).
 - Click a card to edit the step, a line to edit the connection. Up to 50 steps per sequence.
 - Card positions save automatically. The canvas is collaborative: you see teammates' cursors, cards moving as they drag, and their selections outlined in their color. Press `/` to chat from your cursor. See [Collaboration](/guides/collaboration/).

docs/public/openapi.json

@@ -19635,12 +19635,12 @@
           "tracking_domain": {
             "type": "string"
           },
-          "sequences": {
+          "steps": {
             "type": "array",
             "items": {
               "$ref": "#/components/schemas/CampaignStepCreate"
             },
-            "description": "Initial sequence steps in order."
+            "description": "Initial sequence steps in order. They are connected in order: each step routes unconditionally to the next, waiting that step's wait_after days (first step 0, follow-ups 3 unless given)."
           },
           "variants": {
             "type": "array",
@@ -19906,7 +19906,7 @@
           },
           "wait_after": {
             "type": "integer",
-            "description": "Minutes to wait after this step before the next."
+            "description": "Days to wait before this step, counted from the contact's previous step."
           },
           "position": {
             "type": "integer"
@@ -19921,7 +19921,7 @@
           },
           "conditions": {
             "type": "object",
-            "description": "Branching tree ({branches: [...]}).",
+            "description": "The connections out of this step ({branches: [...]}). Empty means no outgoing path: the flow ends here for the contact.",
             "additionalProperties": true
           },
           "action": {
@@ -19992,11 +19992,11 @@
           },
           "wait_after": {
             "type": "integer",
-            "description": "Minutes to wait after this step (spacing model; no standalone wait node for email steps)."
+            "description": "Days to wait before this step, counted from the contact's previous step (0 to 60). Spacing belongs to the target step; there is no standalone wait node for email steps."
           },
           "conditions": {
             "type": "object",
-            "description": "Branching tree ({branches: [...]}). Send {} or empty branches to clear branching.",
+            "description": "The connections out of this step ({branches: [...]}), evaluated in order; a branch with no conditions is a plain go-there-next link. Routing follows connections only: {} or empty branches means no outgoing path, which ends the flow for the contact.",
             "additionalProperties": true
           },
           "kind": {
@@ -20842,10 +20842,13 @@
             "enum": [
               "pending",
               "active",
+              "completed",
               "replied",
               "bounced",
+              "failed",
               "unsubscribed"
-            ]
+            ],
+            "description": "Derived, highest priority first: unsubscribed, bounced, replied, failed (a step could not be sent after every retry), completed (every email step sent), active (some steps sent), pending (nothing sent yet)."
           },
           "sent": {
             "type": "integer"
@@ -20866,6 +20869,10 @@
             "type": "string",
             "description": "Label of the step the contact is on now. Empty when nothing sent yet."
           },
+          "failure_reason": {
+            "type": "string",
+            "description": "Why the last send failed, as reported by the sending worker. Present only when status is failed."
+          },
           "last_activity_at": {
             "type": [
               "string",

internal/api/handler/internal_worker_config.go

@@ -1,10 +1,12 @@
 package handler
 
 import (
+	"context"
 	"encoding/json"
 	"io"
 	"net/http"
 	"os"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
@@ -111,6 +113,10 @@ type HeartbeatPayload struct {
 	// heartbeat ages out. Placement would otherwise keep handing work to a
 	// process that is already gone.
 	Stopping bool `json:"stopping,omitempty"`
+	// Booted is set on the first beat of a fresh process. A worker holds its
+	// mailboxes in memory only, so the backend reloads every mailbox assigned
+	// to it right away; until then each send to it fails with "not found".
+	Booted bool `json:"booted,omitempty"`
 }
 
 func (h *Handler) InternalWorkerHeartbeat(c *gin.Context) {
@@ -161,6 +167,14 @@ func (h *Handler) InternalWorkerHeartbeat(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
+	if p.Booted && h.EmailService != nil {
+		// Off the request: the reload publishes one ADD_EMAIL per mailbox.
+		go func() {
+			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+			defer cancel()
+			h.EmailService.ReloadWorkerAccounts(ctx, id)
+		}()
+	}
 	c.Status(http.StatusNoContent)
 }
 

internal/app/consumer/event_send_result.go

@@ -0,0 +1,327 @@
+package jobs
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+
+	"github.com/warmbly/warmbly/internal/config"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/infrastructure/pubsub"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// errSendResultEarly is returned when a worker result lands before the
+// control plane has finished stamping the task it belongs to. The message is
+// left for redelivery; the stamping finishes within milliseconds.
+var errSendResultEarly = errors.New("send result arrived before the task was stamped; retrying")
+
+// HandleEmailSent confirms a send the worker delivered to the provider. The
+// control plane stamped the task when it handed the send over, so the only
+// thing left to persist is the Message-ID the worker actually put on the wire.
+func (s *JobsService) HandleEmailSent(ctx context.Context, result models.SendEmailResult) error {
+	if s.TaskRepo == nil || result.TaskID == uuid.Nil {
+		return nil
+	}
+	task, err := s.TaskRepo.GetTask(ctx, result.TaskID)
+	if err != nil {
+		return err
+	}
+	if task == nil {
+		log.Warn().Str("task_id", result.TaskID.String()).Msg("email sent result for unknown task")
+		return nil
+	}
+	if result.MessageID != "" && task.MessageID == "" {
+		if err := s.TaskRepo.UpdateTaskMessageID(ctx, task.ID, result.MessageID); err != nil {
+			log.Warn().Err(err).Str("task_id", task.ID.String()).Msg("could not record worker message id")
+		}
+	}
+	return nil
+}
+
+// HandleEmailFailed walks back a send the worker could not deliver. The
+// control plane stamps a task sent the moment it hands it to a worker, so
+// without this the lead would sit at "processing" forever with no email ever
+// leaving. For a campaign send the step's sent_at is cleared so the next tick
+// retries it, the day's counters give the send back, the failure is written to
+// the campaign's activity log, and a campaign that completed while the send
+// was in flight is reopened. Once the retry cap is spent the lead is marked
+// failed and routing drops it.
+func (s *JobsService) HandleEmailFailed(ctx context.Context, result models.SendEmailResult) error {
+	if s.TaskRepo == nil || result.TaskID == uuid.Nil {
+		return nil
+	}
+	task, err := s.TaskRepo.GetTask(ctx, result.TaskID)
+	if err != nil {
+		return err
+	}
+	if task == nil {
+		log.Warn().Str("task_id", result.TaskID.String()).Msg("email failed result for unknown task")
+		return nil
+	}
+
+	if task.Status == "active" {
+		// The worker answered before the control plane finished the tick; the
+		// stamping completes within milliseconds, so look once more before
+		// leaving the result for redelivery.
+		time.Sleep(300 * time.Millisecond)
+		if task, err = s.TaskRepo.GetTask(ctx, result.TaskID); err != nil || task == nil {
+			return err
+		}
+	}
+	switch task.Status {
+	case "completed":
+		// The normal case: stamped by the control plane, refused by the worker.
+	case "active":
+		return errSendResultEarly
+	default:
+		// Already walked back (duplicate delivery), cancelled, or dead-lettered.
+		return nil
+	}
+
+	reason, code := sendFailureReason(result)
+	if err := s.TaskRepo.RecordTaskFailure(ctx, task.ID, "Send failed", reason); err != nil {
+		return err
+	}
+
+	switch task.TaskType {
+	case "campaign":
+		return s.failCampaignSend(ctx, task, reason, code)
+	case "email":
+		s.notifyUserSendFailed(ctx, task, reason)
+	}
+	return nil
+}
+
+// failCampaignSend is the campaign half of HandleEmailFailed.
+func (s *JobsService) failCampaignSend(ctx context.Context, task *repository.Task, reason, code string) error {
+	ct, err := s.TaskRepo.GetCampaignTask(ctx, task.ID)
+	if err != nil {
+		return err
+	}
+	if ct == nil || ct.CampaignID == nil {
+		return nil
+	}
+	campaignID := *ct.CampaignID
+
+	var campaign *models.Campaign
+	if s.CampaignRepo != nil {
+		campaign, _ = s.CampaignRepo.GetByID(ctx, campaignID)
+	}
+
+	recipient := ""
+	if ct.ContactID != nil && s.ContactRepo != nil {
+		if contact, cerr := s.ContactRepo.GetByID(ctx, *ct.ContactID); cerr == nil && contact != nil {
+			recipient = contact.Email
+		}
+	}
+
+	attempts, exhausted, rolledBack := 0, false, false
+	if ct.ContactID != nil && ct.SequenceID != nil && s.CampaignProgressRepo != nil {
+		attempts, exhausted, rolledBack, err = s.CampaignProgressRepo.RecordSendFailure(ctx, campaignID, *ct.ContactID, *ct.SequenceID, reason)
+		if err != nil {
+			return err
+		}
+	}
+
+	if rolledBack && s.CampaignRepo != nil {
+		// With the step walked back, no other stamped step means this was the
+		// lead's first email, which the new-lead cap counted.
+		newLead := true
+		if has, herr := s.CampaignProgressRepo.HasSentSteps(ctx, campaignID, *ct.ContactID); herr == nil {
+			newLead = !has
+		}
+		countedOn := time.Now()
+		if task.CompletedAt != nil {
+			countedOn = *task.CompletedAt
+		}
+		if derr := s.CampaignRepo.DecrementCampaignDailySend(ctx, campaignID, countedOn, newLead); derr != nil {
+			log.Warn().Err(derr).Str("campaign_id", campaignID.String()).Msg("could not give back the failed send's daily count")
+		}
+	}
+
+	// A recipient the mail server refused at RCPT is a hard bounce that simply
+	// arrived synchronously. Retrying it from the same mailbox only spends
+	// reputation, so it goes through the bounce pipeline (progress, optional
+	// suppression, guardrails, warmup health, webhooks) and the lead is
+	// dropped as bounced instead of being offered again.
+	if rolledBack && code == string(errx.MailErrorCodeRecipientRejected) {
+		if s.recordSynchronousBounce(ctx, task, ct, campaign, recipient, reason) {
+			s.logCampaignSendFailure(ctx, campaignID, ct, recipient, reason, code, attempts, false, false, false)
+			s.publishCampaignUpdated(ctx, campaign, campaignID, "")
+			log.Info().Str("task_id", task.ID.String()).Str("campaign_id", campaignID.String()).Msg("campaign send refused at RCPT; recorded as bounce")
+			return nil
+		}
+	}
+
+	reopened := false
+	if rolledBack && !exhausted && campaign != nil && campaign.Status == "completed" && s.CampaignRepo != nil {
+		if ok, rerr := s.CampaignRepo.ReopenAfterSendFailure(ctx, campaignID); rerr != nil {
+			log.Warn().Err(rerr).Str("campaign_id", campaignID.String()).Msg("could not reopen campaign after send failure")
+		} else {
+			reopened = ok
+		}
+	}
+
+	s.logCampaignSendFailure(ctx, campaignID, ct, recipient, reason, code, attempts, exhausted, rolledBack, reopened)
+
+	status := ""
+	if reopened {
+		status = "active"
+	}
+	s.publishCampaignUpdated(ctx, campaign, campaignID, status)
+
+	log.Info().
+		Str("task_id", task.ID.String()).
+		Str("campaign_id", campaignID.String()).
+		Str("code", code).
+		Int("attempts", attempts).
+		Bool("exhausted", exhausted).
+		Bool("rolled_back", rolledBack).
+		Bool("reopened", reopened).
+		Msg("campaign send failed in worker; step walked back")
+	return nil
+}
+
+// recordSynchronousBounce feeds a RCPT-refused send into the deliverability
+// pipeline as a bounce. Returns false when the bounce could not be attributed
+// (no org, no recipient), in which case the caller falls back to the retry
+// path.
+func (s *JobsService) recordSynchronousBounce(ctx context.Context, task *repository.Task, ct *repository.CampaignTask, campaign *models.Campaign, recipient, reason string) bool {
+	if s.AdvancedService == nil || campaign == nil || campaign.OrganizationID == nil || recipient == "" {
+		return false
+	}
+	taskID := task.ID
+	req := &models.IngestDeliverabilityEventRequest{
+		EventType:      models.DeliverabilityEventBounce,
+		Provider:       "smtp_reject",
+		TaskID:         &taskID,
+		CampaignID:     ct.CampaignID,
+		ContactID:      ct.ContactID,
+		RecipientEmail: recipient,
+		Reason:         reason,
+		IdempotencyKey: "reject:" + taskID.String(),
+	}
+	if xerr := s.AdvancedService.IngestDeliverabilityEvent(ctx, *campaign.OrganizationID, req); xerr != nil {
+		log.Warn().Str("task_id", taskID.String()).Str("error", xerr.Message).Msg("could not record refused recipient as a bounce")
+		return false
+	}
+	return true
+}
+
+// publishCampaignUpdated pulses the campaign for every teammate (status "" keeps
+// the dashboard's status as is).
+func (s *JobsService) publishCampaignUpdated(ctx context.Context, campaign *models.Campaign, campaignID uuid.UUID, status string) {
+	if s.StreamingPublisher == nil || campaign == nil {
+		return
+	}
+	s.StreamingPublisher.PublishCampaignEvent(ctx, &pubsub.CampaignEvent{
+		BaseEvent:  pubsub.BaseEvent{EventType: pubsub.EventCampaignUpdated, UserID: campaign.UserID},
+		OrgID:      campaignOrgID(campaign),
+		CampaignID: campaignID.String(),
+		Name:       campaign.Name,
+		Status:     status,
+	})
+}
+
+// logCampaignSendFailure writes the failure to the campaign activity log.
+// metadata.level "error" tints it red in the dashboard's activity feed.
+func (s *JobsService) logCampaignSendFailure(ctx context.Context, campaignID uuid.UUID, ct *repository.CampaignTask, recipient, reason, code string, attempts int, exhausted, rolledBack, reopened bool) {
+	if s.CampaignLogRepo == nil {
+		return
+	}
+	who := recipient
+	if who == "" {
+		who = "the recipient"
+	}
+	var msg string
+	switch {
+	case code == string(errx.MailErrorCodeRecipientRejected):
+		msg = fmt.Sprintf("The mail server refused %s (hard bounce): %s", who, reason)
+	case exhausted:
+		msg = fmt.Sprintf("Gave up on %s after %d failed attempts: %s", who, attempts, reason)
+	case rolledBack:
+		msg = fmt.Sprintf("Could not send to %s (attempt %d of %d), will retry: %s", who, attempts, config.CampaignSendMaxAttempts, reason)
+	default:
+		msg = fmt.Sprintf("Could not send to %s: %s", who, reason)
+	}
+	if reopened {
+		msg += ". Campaign reopened to retry."
+	}
+	meta := map[string]interface{}{
+		"level":        "error",
+		"code":         code,
+		"error":        reason,
+		"attempts":     attempts,
+		"max_attempts": config.CampaignSendMaxAttempts,
+		"will_retry":   rolledBack && !exhausted && code != string(errx.MailErrorCodeRecipientRejected),
+		"task_id":      ct.TaskID.String(),
+	}
+	if ct.ContactID != nil {
+		meta["contact_id"] = ct.ContactID.String()
+	}
+	if ct.SequenceID != nil {
+		meta["sequence_id"] = ct.SequenceID.String()
+	}
+	if reopened {
+		meta["reopened"] = true
+	}
+	_ = s.CampaignLogRepo.CreateLog(ctx, &repository.CampaignLogEntry{
+		CampaignID: campaignID,
+		EventType:  "email_failed",
+		Message:    msg,
+		Metadata:   meta,
+	})
+}
+
+// notifyUserSendFailed tells the mailbox owner that a one-off send (compose,
+// reply, scheduled email) did not leave, since nothing else would.
+func (s *JobsService) notifyUserSendFailed(ctx context.Context, task *repository.Task, reason string) {
+	if s.StreamingPublisher == nil || s.EmailRepository == nil {
+		return
+	}
+	account, xerr := s.EmailRepository.GetByID(ctx, task.EmailAccountID)
+	if xerr != nil || account == nil {
+		return
+	}
+	s.StreamingPublisher.PublishEmailError(ctx, account.UserID, account.ID, task.ID,
+		"Email could not be sent",
+		fmt.Sprintf("%s could not send your email: %s", account.Email, reason))
+}
+
+// sendFailureReason picks the most useful human-readable reason and the
+// machine code out of a worker result.
+func sendFailureReason(result models.SendEmailResult) (reason, code string) {
+	if result.Error != nil {
+		code = result.Error.Code
+		reason = strings.TrimSpace(result.Error.Message)
+		if reason == "" {
+			reason = strings.TrimSpace(result.Error.UserMessage)
+		}
+	}
+	if reason == "" {
+		reason = strings.TrimSpace(result.LegacyErrorMsg)
+	}
+	if reason == "" {
+		reason = "the sending worker reported an unknown error"
+	}
+	if code == "" {
+		code = "SEND_FAILED"
+	}
+	return reason, code
+}
+
+// campaignOrgID returns the campaign's organization id for org-scoped
+// realtime events, or "" for legacy orgless rows.
+func campaignOrgID(campaign *models.Campaign) string {
+	if campaign == nil || campaign.OrganizationID == nil {
+		return ""
+	}
+	return campaign.OrganizationID.String()
+}

internal/app/consumer/event_send_result_live_test.go

@@ -0,0 +1,251 @@
+package jobs
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/warmbly/warmbly/internal/config"
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// Live checks of the send-outcome loop against a real Postgres. Skipped unless
+// WARMBLY_TEST_DB is set (same convention as internal/scheduler):
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \
+//	  go test ./internal/app/consumer/ -run Live -v
+//
+// What is worth proving here is the walk-back itself: that a worker's
+// EMAIL_FAILED clears the step the control plane stamped, gives the day's
+// counters back, keeps the step routable until the attempt cap, and then drops
+// the lead. That only holds against the real queries.
+
+type sendResultFixture struct {
+	user, org, mailbox, campaign, step, contact uuid.UUID
+}
+
+func newSendResultFixture(t *testing.T, handle *db.DB) *sendResultFixture {
+	t.Helper()
+	ctx := context.Background()
+	pool := handle.Pool
+	f := &sendResultFixture{
+		user: uuid.New(), org: uuid.New(), mailbox: uuid.New(),
+		campaign: uuid.New(), step: uuid.New(), contact: uuid.New(),
+	}
+	exec := func(sql string, args ...any) {
+		t.Helper()
+		if _, err := pool.Exec(ctx, sql, args...); err != nil {
+			t.Fatalf("fixture %q: %v", sql[:min(60, len(sql))], err)
+		}
+	}
+	exec(`INSERT INTO users (id, email, first_name, last_name) VALUES ($1, $2, 'Live', 'Test')`,
+		f.user, "live-"+f.user.String()[:8]+"@test.local")
+	exec(`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Live Test', $2, $3)`,
+		f.org, "live-"+f.org.String()[:8], f.user)
+	exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name,
+	          signature_plain, signature_html, provider, status, campaign_limit, min_wait_time, timezone)
+	      VALUES ($1, $2, $3, $4, 'Live', '', '', 'smtp_imap', 'active', 50, 600, 'UTC')`,
+		f.mailbox, f.user, f.org, "live-"+f.mailbox.String()[:8]+"@test.local")
+	exec(`INSERT INTO campaigns (id, user_id, organization_id, name, description, status,
+	          daily_limit, timezone, days, start_time, end_time, rotation_mode, updated_at, created_at)
+	      VALUES ($1, $2, $3, 'Live Test', '', 'completed', 50, 'UTC', 127, '00:00', '23:59',
+	              'least_recently_used', NOW(), NOW())`, f.campaign, f.user, f.org)
+	exec(`INSERT INTO sequences (id, campaign_id, organization_id, name, subject,
+	          body_plain, body_html, wait_after, position, kind)
+	      VALUES ($1, $2, $3, 'Step 1', 'Hi', 'Hello', '<p>Hello</p>', 0, 1, 'email')`, f.step, f.campaign, f.org)
+	exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields)
+	      VALUES ($1, $2, $3, $4, 'Live', 'Contact', '', '', '{}')`,
+		f.contact, f.user, f.org, "lead-"+f.contact.String()[:8]+"@test.local")
+	exec(`INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, 0)`, f.campaign, f.contact)
+
+	t.Cleanup(func() {
+		c := context.Background()
+		for _, step := range []struct {
+			sql string
+			arg any
+		}{
+			{`DELETE FROM task_failures WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox},
+			{`DELETE FROM campaign_tasks WHERE task_id IN (SELECT id FROM tasks WHERE email_account_id = $1)`, f.mailbox},
+			{`DELETE FROM tasks WHERE email_account_id = $1`, f.mailbox},
+			{`DELETE FROM campaign_logs WHERE campaign_id = $1`, f.campaign},
+			{`DELETE FROM campaign_daily_sends WHERE campaign_id = $1`, f.campaign},
+			{`DELETE FROM campaign_contact_progress WHERE campaign_id = $1`, f.campaign},
+			{`DELETE FROM campaign_leads WHERE campaign_id = $1`, f.campaign},
+			{`DELETE FROM sequences WHERE campaign_id = $1`, f.campaign},
+			{`DELETE FROM campaigns WHERE id = $1`, f.campaign},
+			{`DELETE FROM email_accounts WHERE id = $1`, f.mailbox},
+			{`DELETE FROM contacts WHERE organization_id = $1`, f.org},
+			{`DELETE FROM organizations WHERE id = $1`, f.org},
+			{`DELETE FROM users WHERE id = $1`, f.user},
+		} {
+			if _, err := pool.Exec(c, step.sql, step.arg); err != nil {
+				t.Errorf("cleanup %q: %v", step.sql, err)
+			}
+		}
+	})
+	return f
+}
+
+// stampSend does what the backend's campaign tick does after handing a send
+// to the worker: a completed task, the progress stamp, and the day's counters.
+func (f *sendResultFixture) stampSend(t *testing.T, s *JobsService) uuid.UUID {
+	t.Helper()
+	ctx := context.Background()
+	taskID := uuid.New()
+	now := time.Now()
+	created, err := s.TaskRepo.CreateTaskWithLock(ctx, &repository.Task{
+		ID: taskID, TaskType: "campaign", EmailAccountID: f.mailbox, Status: "pending", ScheduledAt: &now,
+	}, &repository.CampaignTask{TaskID: taskID, CampaignID: &f.campaign})
+	if err != nil || !created {
+		t.Fatalf("create task: created=%v err=%v", created, err)
+	}
+	if err := s.TaskRepo.UpdateCampaignTaskTracking(ctx, taskID, f.contact, f.step); err != nil {
+		t.Fatalf("tracking: %v", err)
+	}
+	if err := s.CampaignProgressRepo.RecordEmailSent(ctx, f.campaign, f.contact, f.step); err != nil {
+		t.Fatalf("record sent: %v", err)
+	}
+	if err := s.CampaignRepo.IncrementCampaignDailySend(ctx, f.campaign, true); err != nil {
+		t.Fatalf("increment daily: %v", err)
+	}
+	if err := s.TaskRepo.UpdateTaskStatusWithLock(ctx, taskID, "completed"); err != nil {
+		t.Fatalf("complete task: %v", err)
+	}
+	return taskID
+}
+
+func TestLiveHandleEmailFailedWalksBackAndRetriesUntilCap(t *testing.T) {
+	dsn := os.Getenv("WARMBLY_TEST_DB")
+	if dsn == "" {
+		t.Skip("WARMBLY_TEST_DB not set")
+	}
+	ctx := context.Background()
+	handle, err := db.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("connect: %v", err)
+	}
+	t.Cleanup(func() { handle.Pool.Close() })
+
+	f := newSendResultFixture(t, handle)
+	s := &JobsService{
+		TaskRepo:             repository.NewTaskRepository(handle.Pool),
+		CampaignRepo:         repository.NewCampaignRepostory(handle),
+		CampaignProgressRepo: repository.NewCampaignProgressRepository(handle.Pool),
+		CampaignLogRepo:      repository.NewCampaignLogRepository(handle),
+		ContactRepo:          repository.NewContactRepostory(handle),
+	}
+
+	var sentAt *time.Time
+	var attempts int
+	readProgress := func() {
+		t.Helper()
+		if err := handle.Pool.QueryRow(ctx, `SELECT sent_at, send_attempts FROM campaign_contact_progress
+			WHERE campaign_id = $1 AND contact_id = $2 AND sequence_id = $3`, f.campaign, f.contact, f.step).
+			Scan(&sentAt, &attempts); err != nil {
+			t.Fatalf("read progress: %v", err)
+		}
+	}
+	nextPair := func() *repository.ContactSequencePair {
+		t.Helper()
+		pair, _, err := s.CampaignProgressRepo.FindNextRoutedPair(ctx, f.campaign, "created_at", "asc", "", false, false)
+		if err != nil {
+			t.Fatalf("next pair: %v", err)
+		}
+		return pair
+	}
+
+	// First failure: the stamped step is walked back, the day's counters give
+	// the send back, the task is failed, the completed campaign reopens, and
+	// the step is routable again.
+	taskID := f.stampSend(t, s)
+	readProgress()
+	if sentAt == nil {
+		t.Fatal("precondition: step should be stamped sent")
+	}
+	if nextPair() != nil {
+		t.Fatal("precondition: a stamped single-step lead has nothing left to route")
+	}
+	if err := s.HandleEmailFailed(ctx, models.SendEmailResult{
+		TaskID: taskID, Success: false,
+		Error: &models.EmailSendError{Code: "UNSUPPORTED", Message: "worker could not send"},
+	}); err != nil {
+		t.Fatalf("handle failed: %v", err)
+	}
+	readProgress()
+	if sentAt != nil || attempts != 1 {
+		t.Fatalf("after first failure: sent_at=%v attempts=%d, want NULL/1", sentAt, attempts)
+	}
+	var taskStatus string
+	if err := handle.Pool.QueryRow(ctx, `SELECT status FROM tasks WHERE id = $1`, taskID).Scan(&taskStatus); err != nil {
+		t.Fatal(err)
+	}
+	if taskStatus != "failed" {
+		t.Fatalf("task status = %s, want failed", taskStatus)
+	}
+	var sent, newLeads int
+	if err := handle.Pool.QueryRow(ctx, `SELECT emails_sent, new_leads_started FROM campaign_daily_sends
+		WHERE campaign_id = $1 AND send_date = CURRENT_DATE`, f.campaign).Scan(&sent, &newLeads); err != nil {
+		t.Fatal(err)
+	}
+	if sent != 0 || newLeads != 0 {
+		t.Fatalf("daily counters after walk-back: sent=%d new=%d, want 0/0", sent, newLeads)
+	}
+	var campaignStatus string
+	if err := handle.Pool.QueryRow(ctx, `SELECT status FROM campaigns WHERE id = $1`, f.campaign).Scan(&campaignStatus); err != nil {
+		t.Fatal(err)
+	}
+	if campaignStatus != "active" {
+		t.Fatalf("campaign status = %s, want active (reopened)", campaignStatus)
+	}
+	if pair := nextPair(); pair == nil || pair.SequenceID != f.step || !pair.IsNewLead {
+		t.Fatalf("after walk-back the step should be routable as a new lead, got %+v", pair)
+	}
+	var logs int
+	if err := handle.Pool.QueryRow(ctx, `SELECT COUNT(*) FROM campaign_logs
+		WHERE campaign_id = $1 AND event_type = 'email_failed' AND metadata->>'level' = 'error'`, f.campaign).Scan(&logs); err != nil {
+		t.Fatal(err)
+	}
+	if logs != 1 {
+		t.Fatalf("campaign log entries = %d, want 1", logs)
+	}
+
+	// A duplicate result for the same task is a no-op.
+	if err := s.HandleEmailFailed(ctx, models.SendEmailResult{TaskID: taskID, LegacyErrorMsg: "again"}); err != nil {
+		t.Fatalf("duplicate: %v", err)
+	}
+	readProgress()
+	if attempts != 1 {
+		t.Fatalf("duplicate result changed attempts to %d", attempts)
+	}
+
+	// Keep failing: the step stays routable until the cap, then the lead is
+	// dropped and marked failed.
+	for i := 2; i <= config.CampaignSendMaxAttempts; i++ {
+		id := f.stampSend(t, s)
+		if err := s.HandleEmailFailed(ctx, models.SendEmailResult{TaskID: id, LegacyErrorMsg: "still broken"}); err != nil {
+			t.Fatalf("attempt %d: %v", i, err)
+		}
+		readProgress()
+		if attempts != i {
+			t.Fatalf("attempt %d recorded as %d", i, attempts)
+		}
+		if i < config.CampaignSendMaxAttempts && nextPair() == nil {
+			t.Fatalf("after attempt %d the step should still be routable", i)
+		}
+	}
+	if nextPair() != nil {
+		t.Fatal("after the attempt cap the lead should be dropped from routing")
+	}
+	counts, xerr := s.ContactRepo.CampaignLeadCounts(ctx, f.org.String(), f.campaign.String())
+	if xerr != nil {
+		t.Fatalf("lead counts: %v", xerr)
+	}
+	if counts.Failed != 1 || counts.Queued != 0 || counts.Total != 1 {
+		t.Fatalf("lead counts = %+v, want failed=1", counts)
+	}
+}

internal/app/consumer/events.go

@@ -39,6 +39,11 @@ func (w *JobsService) InitEvents() {
 	Register(w, models.JobEventTypeSyncState, w.HandleSyncState)
 	Register(w, models.JobEventTypeTokenUpdate, w.HandleTokenUpdate)
 
+	// Send outcomes. A worker reports every send it was handed; a failure
+	// walks back what the control plane stamped at hand-off.
+	Register(w, models.JobEventTypeEmailSent, w.HandleEmailSent)
+	Register(w, models.JobEventTypeEmailFailed, w.HandleEmailFailed)
+
 	// Email error handlers
 	Register(w, models.JobEventTypeEmailAuthError, w.HandleEmailAuthError)
 	Register(w, models.JobEventTypeEmailDisabled, w.HandleEmailDisabled)

internal/app/consumer/service.go

@@ -59,6 +59,16 @@ type JobsService struct {
 	// dead-worker job moves or strands their mailboxes. Nil disables it.
 	Notifier OrgNotifier
 
+	// Send-outcome handling (EMAIL_SENT / EMAIL_FAILED from workers). The task
+	// and campaign progress the control plane stamped at hand-off are walked
+	// back here when a worker reports it could not send. Nil TaskRepo disables
+	// the handlers.
+	TaskRepo             repository.TaskRepository
+	CampaignRepo         repository.CampaignRepository
+	CampaignProgressRepo repository.CampaignProgressRepository
+	CampaignLogRepo      repository.CampaignLogRepository
+	ContactRepo          repository.ContactRepository
+
 	eventHandlers map[models.JobEventType]func(ctx context.Context, body any) error
 }
 

internal/app/email/loader.go

@@ -86,6 +86,29 @@ func (s *emailService) reconcileWorkerAccounts(ctx context.Context, lastPublishe
 	}
 }
 
+// ReloadWorkerAccounts publishes every active mailbox assigned to workerID
+// back onto it. Called from the boot heartbeat, so a restarted worker is
+// sending and syncing again within seconds rather than after the reconciler's
+// next republish window.
+func (s *emailService) ReloadWorkerAccounts(ctx context.Context, workerID uuid.UUID) {
+	ids, err := s.emailRepository.ListActiveAccountsByWorker(ctx, workerID)
+	if err != nil {
+		log.Warn().Err(err).Str("worker_id", workerID.String()).Msg("worker boot reload: list accounts failed")
+		return
+	}
+	loaded := 0
+	for _, id := range ids {
+		if err := s.LoadAccountOntoWorker(ctx, id); err != nil {
+			log.Warn().Err(err).Str("email_id", id.String()).Str("worker_id", workerID.String()).Msg("worker boot reload: load account failed")
+			continue
+		}
+		loaded++
+	}
+	if len(ids) > 0 {
+		log.Info().Str("worker_id", workerID.String()).Int("loaded", loaded).Int("assigned", len(ids)).Msg("worker boot reload: mailboxes re-shipped")
+	}
+}
+
 // loadAccountBestEffort loads a freshly onboarded account onto its worker without
 // blocking or failing the onboarding response; the reconciler is the safety net.
 func (s *emailService) loadAccountBestEffort(ctx context.Context, accountID uuid.UUID) {

internal/app/email/service.go

@@ -70,6 +70,9 @@ type EmailService interface {
 	// StartWorkerReconciler periodically ensures every active mailbox is
 	// assigned to a worker and loaded onto it (blocks until ctx is cancelled).
 	StartWorkerReconciler(ctx context.Context, interval time.Duration)
+	// ReloadWorkerAccounts re-ships every active mailbox assigned to one
+	// worker, for a worker that just booted and holds none in memory.
+	ReloadWorkerAccounts(ctx context.Context, workerID uuid.UUID)
 }
 
 type emailService struct {

internal/app/worker/event_send_email.go

@@ -3,6 +3,7 @@ package worker
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"time"
@@ -29,9 +30,13 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se
 	w.mailManager.RUnlock()
 
 	if !exists {
+		// The mailbox is not loaded here: it is still being added (its
+		// ADD_EMAIL is queued behind this send), or this worker restarted and
+		// the reconciler has not re-shipped it yet. Leave the send for
+		// redelivery a few times so a queued ADD_EMAIL gets processed first,
+		// then report the failure so the control plane retries the step.
 		err := fmt.Errorf("email account %s not found in worker", sendEmail.EmailID.String())
-		w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, err.Error())
-		return err
+		return w.failSend(ctx, sendEmail, err.Error(), true)
 	}
 
 	// Decrypt subject
@@ -50,8 +55,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se
 	bodyPlain, bodyHTML, attachmentRefs, err := w.fetchEmailBody(ctx, sendEmail.OrgID, sendEmail.BodyS3Key)
 	if err != nil {
 		log.Error().Err(err).Str("s3_key", sendEmail.BodyS3Key).Msg("Failed to fetch email body from S3")
-		w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch email body: %v", err))
-		return err
+		return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch email body: %v", err), true)
 	}
 
 	// Fetch each attachment's bytes from object storage by key. A fetch failure
@@ -59,8 +63,7 @@ func (w *WorkerService) HandleSendEmail(ctx context.Context, sendEmail models.Se
 	attachments, err := w.fetchAttachments(ctx, attachmentRefs)
 	if err != nil {
 		log.Error().Err(err).Str("task_id", sendEmail.TaskID.String()).Msg("Failed to fetch attachment bytes from S3")
-		w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, mail, fmt.Sprintf("failed to fetch attachment: %v", err))
-		return err
+		return w.failSend(ctx, sendEmail, fmt.Sprintf("failed to fetch attachment: %v", err), true)
 	}
 
 	// Use unified Send method
@@ -220,7 +223,30 @@ func (w *WorkerService) sendEmailSuccess(taskID uuid.UUID, messageID, providerMs
 	}
 }
 
-// sendEmailError sends a structured error result back to the jobs service
+// sendNotLoadedRedeliveries is how many bus deliveries a send may burn waiting
+// for a transient condition (mailbox not loaded yet, object storage blip)
+// before the worker reports it failed. Each redelivery is a second apart, and
+// the bus stops redelivering at ten, so this stays well inside that.
+const sendNotLoadedRedeliveries = 5
+
+// failSend reports a send the worker could not attempt. A retryable condition
+// is first left for bus redelivery (returning an error naks the message) so a
+// queued ADD_EMAIL or a storage blip can clear; once the redeliveries are
+// spent, when the bus does not redeliver, or when the condition is not
+// retryable, the failure result is produced and the message is acked, handing
+// the retry to the control plane.
+func (w *WorkerService) failSend(ctx context.Context, sendEmail models.SendEmail, reason string, retryable bool) error {
+	if d := deliveryOf(ctx); retryable && d.redelivers && d.attempt < sendNotLoadedRedeliveries {
+		return errors.New(reason)
+	}
+	w.sendEmailFailure(sendEmail.TaskID, sendEmail.EmailID, nil, reason)
+	return nil
+}
+
+// sendEmailError reports a failed send attempt. The per-task result is always
+// an EMAIL_FAILED so the control plane has one result channel to walk the
+// send back on; account-level conditions (auth, disabled, rate limit, server
+// error) additionally raise their own typed event carrying the full context.
 func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail *wmail.WMail, mailErr *errx.MailError) {
 	// Determine the appropriate event type based on error
 	eventType := wmail.DetermineErrorEventType(mailErr)
@@ -236,14 +262,15 @@ func (w *WorkerService) sendEmailError(taskID uuid.UUID, emailID uuid.UUID, mail
 		SentAt:         time.Now(),
 	}
 
-	if err := w.Produce(eventType, taskID.String(), result); err != nil {
-		log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email error event")
+	if err := w.Produce(models.JobEventTypeEmailFailed, taskID.String(), result); err != nil {
+		log.Error().Err(err).Str("task_id", taskID.String()).Msg("Failed to produce email failed event")
 	}
 
-	// For critical auth/disabled errors, also send a separate error event with full context
+	// Account-level conditions also raise their typed event with full context
 	if eventType == models.JobEventTypeEmailAuthError ||
 		eventType == models.JobEventTypeEmailDisabled ||
-		eventType == models.JobEventTypeEmailRateLimited {
+		eventType == models.JobEventTypeEmailRateLimited ||
+		eventType == models.JobEventTypeEmailServerError {
 
 		userInfo := mailErr.GetUserErrorInfo()
 		errorEvent := models.EmailErrorEvent{

internal/app/worker/heartbeat.go

@@ -7,6 +7,13 @@ import (
 )
 
 func (s *WorkerService) Heartbeat(ctx context.Context) {
+	// Beat at once: the control plane refuses to hand sends to a worker with
+	// no heartbeat key, so waiting a full interval would make a fresh worker
+	// unusable for its first 90 seconds.
+	if err := s.heartbeat(ctx); err != nil {
+		log.Println("Failed to do heartbeat", err)
+	}
+
 	ticker := time.NewTicker(90 * time.Second)
 	defer ticker.Stop()
 

internal/app/worker/receive.go

@@ -8,6 +8,21 @@ import (
 	"github.com/warmbly/warmbly/internal/models"
 )
 
+type deliveryKey struct{}
+
+// delivery is what a handler may know about the bus message it is processing.
+type delivery struct {
+	attempt    int  // 1-based delivery count; 0 when unknown
+	redelivers bool // a handler error gets the message delivered again
+}
+
+// deliveryOf returns the bus delivery details for the message a handler is
+// processing (zero value when the handler was not invoked from Receive).
+func deliveryOf(ctx context.Context) delivery {
+	d, _ := ctx.Value(deliveryKey{}).(delivery)
+	return d
+}
+
 // Receive is the eventbus.Handler that drives the worker's event loop. It
 // decodes the wire payload via the injected codec.Codec and dispatches to
 // HandleEvent.
@@ -17,7 +32,7 @@ func (w *WorkerService) Receive(ctx context.Context, msg eventbus.Message) error
 		return err
 	}
 
-	hctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	hctx, cancel := context.WithTimeout(context.WithValue(ctx, deliveryKey{}, delivery{attempt: msg.Attempt, redelivers: msg.Redelivers}), 30*time.Second)
 	defer cancel()
 	return w.HandleEvent(hctx, &event)
 }

internal/app/worker/wmail/send.go

@@ -422,7 +422,7 @@ func DetermineErrorEventType(err *errx.MailError) models.JobEventType {
 	}
 
 	switch err.Code {
-	case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed:
+	case errx.MailErrorCodeGoogleAuth, errx.MailErrorCodeAuthenticationFailed, errx.MailErrorCodeInvalidCredentials:
 		return models.JobEventTypeEmailAuthError
 
 	case errx.MailErrorCodeAccountSuspended, errx.MailErrorCodeAuthorizationFailed:

internal/config/constants.go

@@ -73,6 +73,12 @@ const (
 	// delay that the scheduler would then turn into an unreachable send time.
 	SequenceWaitAfterMax = 60
 
+	// CampaignSendMaxAttempts is how many times one (contact, step) may be
+	// handed to a worker before the lead is marked failed and dropped from the
+	// campaign. A worker-reported failure clears the step's sent_at so the next
+	// tick retries it; this bounds that loop for a mailbox that can never send.
+	CampaignSendMaxAttempts = 5
+
 	// Webhook/integration fan-out throttle. Caps how many events of a single
 	// type one org can fan out to its webhooks + integration sinks
 	// (Slack/Discord/CRM) per minute — the backstop against a campaign "notify"

internal/events/publisher.go

@@ -92,6 +92,14 @@ func (p *publisher) PublishSendEmail(ctx context.Context, workerID uuid.UUID, pa
 	// Store email body (and attachment refs) in S3. The attachment refs ride
 	// inside the emsg blob so the worker receives them via BodyS3Key without any
 	// change to the Avro event contract.
+	if p.bus == nil {
+		return fmt.Errorf("event bus not configured; cannot hand send %s to a worker", params.TaskID)
+	}
+	if p.storageClient == nil {
+		// The worker reads the body from object storage; without it the send
+		// would be published body-less and fail there.
+		return fmt.Errorf("object storage not configured; cannot hand send %s to a worker", params.TaskID)
+	}
 	s3Key, err := p.storeEmailBody(ctx, params.TaskID, params.OrgID, params.BodyPlain, params.BodyHTML, params.Attachments)
 	if err != nil {
 		return fmt.Errorf("failed to store email body: %w", err)

internal/infrastructure/db/migrations/000089_campaign_send_outcome.down.sql

@@ -0,0 +1,4 @@
+ALTER TABLE campaign_contact_progress
+    DROP COLUMN IF EXISTS failure_reason,
+    DROP COLUMN IF EXISTS failed_at,
+    DROP COLUMN IF EXISTS send_attempts;

internal/infrastructure/db/migrations/000089_campaign_send_outcome.up.sql

@@ -0,0 +1,8 @@
+-- Worker send outcomes close the loop on campaign sends. A step is stamped
+-- sent_at when the control plane hands it to a worker; when the worker reports
+-- that the send failed, sent_at is cleared again so the step is retried, and the
+-- failure is kept here so the lead can be shown as failed once retries run out.
+ALTER TABLE campaign_contact_progress
+    ADD COLUMN IF NOT EXISTS send_attempts integer NOT NULL DEFAULT 0,
+    ADD COLUMN IF NOT EXISTS failed_at timestamptz,
+    ADD COLUMN IF NOT EXISTS failure_reason text NOT NULL DEFAULT '';

internal/infrastructure/db/migrations/000090_connect_wizard_steps.down.sql

@@ -0,0 +1 @@
+-- Connections are ordinary step data the user can edit; there is nothing to undo.

internal/infrastructure/db/migrations/000090_connect_wizard_steps.up.sql

@@ -0,0 +1,28 @@
+-- Campaigns created by the wizard before connections were written at creation
+-- hold their follow-ups as disconnected steps, so only the first email ever
+-- sent. Link the steps of every campaign that has two or more email steps and
+-- no connection anywhere, in position order, which is the sequence the wizard
+-- showed ("Follow-up 1, after N days").
+WITH candidates AS (
+    SELECT campaign_id
+    FROM sequences
+    GROUP BY campaign_id
+    HAVING COUNT(*) >= 2
+       AND bool_and(kind = 'email')
+       AND bool_and(NOT COALESCE(
+               jsonb_typeof(conditions->'branches') = 'array'
+               AND jsonb_array_length(conditions->'branches') > 0, false))
+),
+ordered AS (
+    SELECT s.id,
+           LEAD(s.id) OVER (PARTITION BY s.campaign_id ORDER BY s.position ASC, s.created_at ASC) AS next_id
+    FROM sequences s
+    JOIN candidates c ON c.campaign_id = s.campaign_id
+)
+UPDATE sequences s
+SET conditions = jsonb_build_object('branches', jsonb_build_array(jsonb_build_object(
+        'branch_id', gen_random_uuid()::text,
+        'target_step_id', o.next_id::text,
+        'conditions', '[]'::jsonb)))
+FROM ordered o
+WHERE o.id = s.id AND o.next_id IS NOT NULL;

internal/infrastructure/eventbus/bus.go

@@ -76,6 +76,15 @@ type Message struct {
 	Topic   string
 	Key     string
 	Payload []byte
+	// Attempt is the 1-based delivery count of this message (NATS reports it
+	// from the consumer's redelivery metadata; Kafka always reports 1). A
+	// handler that retries by returning an error can read it to know when the
+	// broker is about to stop redelivering and give up cleanly instead.
+	Attempt int
+	// Redelivers is true when a handler error leaves the message for another
+	// delivery (NATS). Kafka commits regardless, so a handler must not count
+	// on a retry there and should finish what it can on this delivery.
+	Redelivers bool
 }
 
 // Subject normalises a topic name to the dot-separated form by replacing any

internal/infrastructure/eventbus/kafka.go

@@ -140,6 +140,7 @@ func (b *KafkaBus) Subscribe(ctx context.Context, topics []string, group string,
 			Topic:   topic,
 			Key:     string(msg.Key),
 			Payload: msg.Value,
+			Attempt: 1,
 		}); err != nil {
 			log.Error().Err(err).Str("topic", topic).Msg("eventbus kafka handler error")
 			return err

internal/infrastructure/eventbus/nats.go

@@ -249,10 +249,16 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string,
 		// on the exact separator and should use Subject() if it needs to
 		// compare against a Kafka-style topic name.
 		topic := strings.TrimPrefix(m.Subject(), b.prefix+".")
+		attempt := 1
+		if meta, merr := m.Metadata(); merr == nil && meta.NumDelivered > 0 {
+			attempt = int(meta.NumDelivered)
+		}
 		if err := invokeHandler(hctx, handler, Message{
-			Topic:   topic,
-			Key:     key,
-			Payload: m.Data(),
+			Topic:      topic,
+			Key:        key,
+			Payload:    m.Data(),
+			Attempt:    attempt,
+			Redelivers: true,
 		}); err != nil {
 			log.Error().Err(err).Str("subject", m.Subject()).Msg("eventbus nats handler error")
 			// Nak with a short delay so transient errors don't hot-loop.

internal/models/contact.go

@@ -65,6 +65,7 @@ type ContactCampaignProgress struct {
 	//   completed    — every sequence step was sent, no reply (done, nothing left to send)
 	//   replied      — the contact has replied (terminal/positive)
 	//   bounced      — a send hard-bounced (terminal/negative)
+	//   failed       — the mailbox could not send a step after every retry (terminal/negative)
 	//   unsubscribed — the contact is unsubscribed/suppressed (terminal)
 	Status         string     `json:"status"`
 	Sent           int        `json:"sent"`
@@ -77,6 +78,9 @@ type ContactCampaignProgress struct {
 	// step actually sent ("Email 2", a custom step name, or an action label).
 	// Empty when nothing has been sent yet (status "pending").
 	CurrentStep string `json:"current_step,omitempty"`
+	// FailureReason is the worker's reason for the last failed send, set only
+	// when Status is "failed".
+	FailureReason string `json:"failure_reason,omitempty"`
 }
 
 // Lead status constants for ContactCampaignProgress.Status.
@@ -86,6 +90,7 @@ const (
 	LeadStatusCompleted    = "completed"
 	LeadStatusReplied      = "replied"
 	LeadStatusBounced      = "bounced"
+	LeadStatusFailed       = "failed"
 	LeadStatusUnsubscribed = "unsubscribed"
 )
 
@@ -93,7 +98,7 @@ const (
 // Used to gate the single-campaign Leads-view `lead_status` search filter.
 func ValidLeadStatus(s string) bool {
 	switch s {
-	case LeadStatusPending, LeadStatusActive, LeadStatusCompleted, LeadStatusReplied, LeadStatusBounced, LeadStatusUnsubscribed:
+	case LeadStatusPending, LeadStatusActive, LeadStatusCompleted, LeadStatusReplied, LeadStatusBounced, LeadStatusFailed, LeadStatusUnsubscribed:
 		return true
 	default:
 		return false
@@ -118,8 +123,8 @@ type ContactsResult struct {
 
 // CampaignLeadCounts are per-status lead totals within a single campaign,
 // derived the same way as ContactCampaignProgress.Status (unsubscribed >
-// bounced > replied > completed > processing > queued). Drives the Leads-view
-// scope chips.
+// bounced > replied > failed > completed > processing > queued). Drives the
+// Leads-view scope chips.
 type CampaignLeadCounts struct {
 	Total        int `json:"total"`
 	Queued       int `json:"queued"`     // pending: a lead, no email sent yet
@@ -127,6 +132,7 @@ type CampaignLeadCounts struct {
 	Completed    int `json:"completed"`  // done: every step sent, no reply
 	Replied      int `json:"replied"`
 	Bounced      int `json:"bounced"`
+	Failed       int `json:"failed"` // a step could not be sent after every retry
 	Unsubscribed int `json:"unsubscribed"`
 }
 

internal/models/sequence.go

@@ -27,9 +27,11 @@ type Sequence struct {
 	X float64 `json:"x"`
 	Y float64 `json:"y"`
 
-	// Conditions is the per-step branching tree. When empty (`{}` / no
-	// branches), the step keeps the default linear behaviour (advance to the
-	// next position). When populated, the scheduler evaluates the contact's
+	// Conditions is the per-step routing: the connections out of this step.
+	// Routing follows connections only. When empty (`{}` / no branches) the
+	// step has no outgoing path and the contact's flow ends there; position
+	// orders the canvas and picks the entry step, it never advances a contact
+	// by itself. When populated, the scheduler evaluates the contact's
 	// engagement against these branches at schedule time to decide which step
 	// (or stop) comes next. Stored as a single jsonb column on `sequences`.
 	Conditions json.RawMessage `json:"conditions,omitempty"`
@@ -220,8 +222,9 @@ type SequencePosition struct {
 // `conditions` jsonb column. Branches are evaluated in declared order; the first
 // branch whose conditions ALL match wins. A winning branch routes the contact to
 // its TargetSequenceID (any step in the campaign), or stops them when the target
-// is nil. When no branch matches (or Branches is empty) the scheduler keeps the
-// default linear progression (advance to the next step by position).
+// is nil. When no branch matches (or Branches is empty) the contact's flow
+// ends at this step; a plain "go there next" link is a branch with no
+// conditions.
 type BranchConditions struct {
 	Branches []Branch `json:"branches,omitempty"`
 }

internal/repository/pg_campaign.go

@@ -87,6 +87,14 @@ type CampaignRepository interface {
 	// CountNewLeadsStartedToday returns new_leads_started for the current UTC
 	// day (0 when no row exists yet).
 	CountNewLeadsStartedToday(ctx context.Context, campaignID uuid.UUID) (int, error)
+	// DecrementCampaignDailySend gives back a counted send the worker could not
+	// deliver, against the day it was counted on (sentAt), so the new-lead cap
+	// and the campaign daily limit do not charge for mail that never left.
+	DecrementCampaignDailySend(ctx context.Context, campaignID uuid.UUID, sentAt time.Time, newLead bool) error
+	// ReopenAfterSendFailure flips a campaign that completed while a send was
+	// still in flight back to active, so the failed step is retried instead of
+	// being finalised as done. Returns true when the status changed.
+	ReopenAfterSendFailure(ctx context.Context, campaignID uuid.UUID) (bool, error)
 
 	// ── Campaign-scoped tracking domain (feature 5) ─────────────────────
 	// SetCampaignTrackingDomainVerified flips the verified flag / timestamp on
@@ -482,8 +490,12 @@ func (r *campaignRepository) Create(ctx context.Context, userID string, orgID *u
 
 	// Initial sequences. Position is the array index; wait_after defaults
 	// to 0 for the first step and 3 days for any follow-ups so a default
-	// wizard run still produces something usable.
+	// wizard run still produces something usable. Steps given together are a
+	// linear sequence, so each one is connected to the next: routing follows
+	// connections only (a step with no outgoing connection ends the flow), and
+	// a follow-up that is listed but not connected would never send.
 	if len(data.Sequences) > 0 {
+		stepIDs := make([]uuid.UUID, 0, len(data.Sequences))
 		for i, seq := range data.Sequences {
 			waitAfter := 0
 			if i > 0 {
@@ -513,16 +525,22 @@ func (r *campaignRepository) Create(ctx context.Context, userID string, orgID *u
 					body_plain, body_html, body_sync, body_code,
 					wait_after, position
 				) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+				RETURNING id
 			`
 			seqParams := []any{
 				campaign.ID, orgID, seq.Name, seq.Subject,
 				seq.BodyPlain, bodyHTML, bodySync, bodyCode,
 				waitAfter, i + 1,
 			}
-			if _, err := tx.Exec(ctx, seqInsert, seqParams...); err != nil {
-				db.CaptureError(err, seqInsert, seqParams, "exec")
+			var stepID uuid.UUID
+			if err := tx.QueryRow(ctx, seqInsert, seqParams...).Scan(&stepID); err != nil {
+				db.CaptureError(err, seqInsert, seqParams, "queryrow")
 				return nil, errx.InternalError()
 			}
+			stepIDs = append(stepIDs, stepID)
+		}
+		if err := connectLinearSequenceTx(ctx, tx, stepIDs); err != nil {
+			return nil, errx.InternalError()
 		}
 	}
 
@@ -1798,6 +1816,37 @@ func (r *campaignRepository) AdvanceRampLevel(ctx context.Context, campaignID uu
 	return err
 }
 
+// DecrementCampaignDailySend reverses IncrementCampaignDailySend for a send
+// that failed in the worker. Clamped at zero; a missing row is left alone.
+func (r *campaignRepository) DecrementCampaignDailySend(ctx context.Context, campaignID uuid.UUID, sentAt time.Time, newLead bool) error {
+	newLeadDec := 0
+	if newLead {
+		newLeadDec = 1
+	}
+	_, err := r.DB.Exec(ctx, `
+		UPDATE campaign_daily_sends
+		SET emails_sent = GREATEST(emails_sent - 1, 0),
+		    new_leads_started = GREATEST(new_leads_started - $3, 0)
+		WHERE campaign_id = $1 AND send_date = $2::date
+	`, campaignID, sentAt.UTC(), newLeadDec)
+	return err
+}
+
+// ReopenAfterSendFailure moves a completed campaign back to active. Only the
+// completed state is touched: a paused campaign stays paused and picks the
+// retry up when it is resumed.
+func (r *campaignRepository) ReopenAfterSendFailure(ctx context.Context, campaignID uuid.UUID) (bool, error) {
+	tag, err := r.DB.Exec(ctx, `
+		UPDATE campaigns
+		SET status = 'active', last_status_change_at = NOW(), updated_at = NOW()
+		WHERE id = $1 AND status = 'completed'
+	`, campaignID)
+	if err != nil {
+		return false, err
+	}
+	return tag.RowsAffected() > 0, nil
+}
+
 // IncrementCampaignDailySend bumps today's per-campaign send counters. newLead
 // also increments new_leads_started (a position-1 send) so the new-lead cap can
 // read it back.
@@ -1877,3 +1926,27 @@ func (r *campaignRepository) UpdateStatusWithLock(ctx context.Context, campaignI
 	}
 	return err
 }
+
+// connectLinearSequenceTx links steps in order with an unconditional
+// connection from each to the next, the same shape the sequence canvas writes
+// when a step is dragged onto another ("just go there after the wait").
+func connectLinearSequenceTx(ctx context.Context, tx pgx.Tx, stepIDs []uuid.UUID) error {
+	for i := 0; i+1 < len(stepIDs); i++ {
+		next := stepIDs[i+1]
+		conditions, err := json.Marshal(models.BranchConditions{
+			Branches: []models.Branch{{
+				BranchID:         uuid.New().String(),
+				TargetSequenceID: &next,
+			}},
+		})
+		if err != nil {
+			return err
+		}
+		const q = `UPDATE sequences SET conditions = $2 WHERE id = $1`
+		if _, err := tx.Exec(ctx, q, stepIDs[i], conditions); err != nil {
+			db.CaptureError(err, q, []any{stepIDs[i]}, "exec")
+			return err
+		}
+	}
+	return nil
+}

internal/repository/pg_campaign_progress.go

@@ -4,10 +4,13 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/config"
 	"github.com/warmbly/warmbly/internal/models"
 )
 
@@ -74,6 +77,16 @@ type CampaignSequencePair struct {
 type CampaignProgressRepository interface {
 	// Record email status
 	RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error
+	// RecordSendFailure walks back a step the worker could not send: sent_at is
+	// cleared so routing offers the step again, and the attempt is counted. It
+	// returns the attempts so far and whether the lead has now exhausted
+	// config.CampaignSendMaxAttempts. rolledBack is false when there was
+	// nothing to walk back (a duplicate result, or the send was already retried).
+	RecordSendFailure(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, reason string) (attempts int, exhausted bool, rolledBack bool, err error)
+	// HasSentSteps reports whether the contact has any other step of the
+	// campaign stamped sent, which is what decides if a failed send was the
+	// lead's first step (a "new lead" for the daily new-lead counter).
+	HasSentSteps(ctx context.Context, campaignID, contactID uuid.UUID) (bool, error)
 	RecordEmailOpened(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, machine bool) error
 	RecordEmailClicked(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error
 	RecordEmailReplied(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error
@@ -149,19 +162,62 @@ func NewCampaignProgressRepository(db *pgxpool.Pool) CampaignProgressRepository
 	return &campaignProgressRepository{db: db}
 }
 
-// RecordEmailSent records that an email was sent
+// RecordEmailSent stamps a step as handed to the worker. A retry after a
+// worker-reported failure clears the failure marker again; send_attempts is
+// kept as history so the retry cap still holds.
 func (r *campaignProgressRepository) RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error {
 	query := `
 		INSERT INTO campaign_contact_progress (campaign_id, contact_id, sequence_id, sent_at)
 		VALUES ($1, $2, $3, NOW())
 		ON CONFLICT (campaign_id, contact_id, sequence_id)
-		DO UPDATE SET sent_at = NOW()
+		DO UPDATE SET sent_at = NOW(), failed_at = NULL, failure_reason = ''
 	`
 
 	_, err := r.db.Exec(ctx, query, campaignID, contactID, sequenceID)
 	return err
 }
 
+// RecordSendFailure clears sent_at on a stamped step and counts the attempt.
+// Only a row that is currently stamped sent is touched, so a duplicate worker
+// result after the step was already walked back (or re-sent) is a no-op.
+func (r *campaignProgressRepository) RecordSendFailure(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, reason string) (int, bool, bool, error) {
+	if len(reason) > 500 {
+		reason = reason[:500]
+	}
+	query := `
+		UPDATE campaign_contact_progress
+		SET sent_at = NULL,
+		    send_attempts = send_attempts + 1,
+		    failed_at = NOW(),
+		    failure_reason = $4
+		WHERE campaign_id = $1 AND contact_id = $2 AND sequence_id = $3
+		  AND sent_at IS NOT NULL
+		RETURNING send_attempts
+	`
+	var attempts int
+	err := r.db.QueryRow(ctx, query, campaignID, contactID, sequenceID, reason).Scan(&attempts)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return 0, false, false, nil
+		}
+		return 0, false, false, err
+	}
+	return attempts, attempts >= config.CampaignSendMaxAttempts, true, nil
+}
+
+// HasSentSteps reports whether any step of the campaign is stamped sent for
+// the contact.
+func (r *campaignProgressRepository) HasSentSteps(ctx context.Context, campaignID, contactID uuid.UUID) (bool, error) {
+	var has bool
+	err := r.db.QueryRow(ctx, `
+		SELECT EXISTS (
+			SELECT 1 FROM campaign_contact_progress
+			WHERE campaign_id = $1 AND contact_id = $2 AND sent_at IS NOT NULL
+		)
+	`, campaignID, contactID).Scan(&has)
+	return has, err
+}
+
 // RecordEmailOpened records that an email was opened. machine marks automated
 // fetches (Apple MPP prefetch, UA-less clients): the first open stamps
 // opened_at with the flag, and a later HUMAN open upgrades a machine open to
@@ -538,16 +594,20 @@ func (r *campaignProgressRepository) GetLatestCampaignSequenceForContact(ctx con
 }
 
 // FindNextRoutedPair selects the next (contact, step) to send by FOLLOWING THE
-// FLOW graph instead of linear position order. For each contact, the next step
-// is the route out of their last-sent step:
+// FLOW graph. For each contact, the next step is the route out of their
+// last-sent step:
 //  1. conditional branches (first match wins, evaluated against engagement),
-//  2. then the explicit "else" catch-all branch (empty conditions; target nil = STOP),
-//  3. then linear position+1 — but ONLY when the step defines no branches at all
-//     (so plain linear campaigns keep working unchanged).
+//  2. then the explicit "else" catch-all branch (empty conditions; target nil = STOP).
+//
+// There is no implicit advance by position: a step with no outgoing connection
+// ends the contact's flow. The campaign wizard connects the steps it creates
+// in order, and the canvas connects steps as they are dragged.
 //
 // A contact who has never been sent starts at the entry step (position 1). A
 // step is sent only if the route reaches it, so branch-only steps are never sent
 // linearly, and a routed step that was already sent (a loop) stops the contact.
+// A contact whose step failed in the worker more than CampaignSendMaxAttempts
+// times is dropped, the same way a bounced or suppressed contact is.
 //
 // Conditions are evaluated SEND-RELATIVE with a three-valued result: a contact
 // whose next step isn't decidable yet (an engagement window still open) is not
@@ -787,6 +847,12 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam
 		    SELECT 1 FROM campaign_contact_progress b
 		    WHERE b.contact_id = cl.contact_id AND b.bounced_at IS NOT NULL
 		  )
+		  AND NOT EXISTS (
+		    SELECT 1 FROM campaign_contact_progress f
+		    WHERE f.campaign_id = $1 AND f.contact_id = cl.contact_id
+		      AND f.sent_at IS NULL AND f.failed_at IS NOT NULL
+		      AND f.send_attempts >= $2
+		  )
 		  AND NOT EXISTS (
 		    SELECT 1 FROM suppressed_recipients sr
 		    JOIN campaigns camp ON camp.organization_id = sr.organization_id
@@ -797,7 +863,7 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam
 		ORDER BY ` + orderPrefix + contactOrder + ` ` + dir + `
 	`
 
-	rows, err := r.db.Query(ctx, query, campaignID)
+	rows, err := r.db.Query(ctx, query, campaignID, config.CampaignSendMaxAttempts)
 	if err != nil {
 		return nil, nil, err
 	}

internal/repository/pg_contact.go

@@ -705,8 +705,9 @@ func (r *contactRepository) Search(
 	// Lead status filter (single-campaign Leads view only)
 	// -----------------------------
 	// The derived status has no stored column, so reproduce the same priority
-	// chain used on read (unsubscribed > bounced > replied > processing >
-	// queued) as a boolean predicate over the campaign's progress rows. Only
+	// chain used on read (unsubscribed > bounced > replied > failed >
+	// processing > queued) as a boolean predicate over the campaign's progress
+	// rows. Only
 	// meaningful with exactly one campaign bound; ignored otherwise.
 	if filters.LeadStatus != "" && singleCampaignPlaceholder != "" {
 		if clause := leadStatusClause(filters.LeadStatus, singleCampaignPlaceholder); clause != "" {
@@ -828,7 +829,14 @@ func (r *contactRepository) Search(
 				'clicked', COUNT(*) FILTER (WHERE p.clicked_at IS NOT NULL),
 				'replied', COUNT(*) FILTER (WHERE p.replied_at IS NOT NULL),
 				'bounced', COUNT(*) FILTER (WHERE p.bounced_at IS NOT NULL),
-				'last_at', MAX(GREATEST(p.sent_at, p.opened_at, p.clicked_at, p.replied_at, p.bounced_at)),
+				-- Steps the mailbox could not send after every retry.
+				'failed',  COUNT(*) FILTER (WHERE p.sent_at IS NULL AND p.failed_at IS NOT NULL AND p.send_attempts >= %[2]d),
+				'failure_reason', (
+					SELECT fp.failure_reason FROM campaign_contact_progress fp
+					WHERE fp.campaign_id = %[1]s AND fp.contact_id = c.id AND fp.failed_at IS NOT NULL
+					ORDER BY fp.failed_at DESC LIMIT 1
+				),
+				'last_at', MAX(GREATEST(p.sent_at, p.opened_at, p.clicked_at, p.replied_at, p.bounced_at, p.failed_at)),
 				-- Total email steps in the sequence, to tell "still sending" (active)
 				-- apart from "every step sent" (completed/done).
 				'total_steps', (SELECT COUNT(*) FROM sequences st WHERE st.campaign_id = %[1]s AND st.kind = 'email'),
@@ -861,7 +869,7 @@ func (r *contactRepository) Search(
 			)
 			FROM campaign_contact_progress p
 			WHERE p.campaign_id = %[1]s AND p.contact_id = c.id
-		)`, singleCampaignPlaceholder)
+		)`, singleCampaignPlaceholder, config.CampaignSendMaxAttempts)
 	}
 
 	// Main query.
@@ -969,6 +977,8 @@ func (r *contactRepository) Search(
 				Clicked    int        `json:"clicked"`
 				Replied    int        `json:"replied"`
 				Bounced    int        `json:"bounced"`
+				Failed     int        `json:"failed"`
+				FailReason *string    `json:"failure_reason"`
 				TotalSteps int        `json:"total_steps"`
 				LastAt     *time.Time `json:"last_at"`
 				Step       *string    `json:"step"`
@@ -985,6 +995,8 @@ func (r *contactRepository) Search(
 				status = models.LeadStatusBounced
 			case lp.Replied > 0:
 				status = models.LeadStatusReplied
+			case lp.Failed > 0:
+				status = models.LeadStatusFailed
 			case lp.Sent > 0 && lp.TotalSteps > 0 && lp.Sent >= lp.TotalSteps:
 				// Every email step has been sent and the contact hasn't replied
 				// or bounced: the sequence is exhausted, so the lead is done.
@@ -996,6 +1008,10 @@ func (r *contactRepository) Search(
 			if lp.Step != nil {
 				currentStep = *lp.Step
 			}
+			failureReason := ""
+			if status == models.LeadStatusFailed && lp.FailReason != nil {
+				failureReason = *lp.FailReason
+			}
 			c.CampaignLead = &models.ContactCampaignProgress{
 				Status:         status,
 				Sent:           lp.Sent,
@@ -1005,6 +1021,7 @@ func (r *contactRepository) Search(
 				Bounced:        lp.Bounced,
 				LastActivityAt: lp.LastAt,
 				CurrentStep:    currentStep,
+				FailureReason:  failureReason,
 			}
 		}
 
@@ -1151,7 +1168,7 @@ func (r *contactRepository) DistinctCustomFieldKeys(ctx context.Context, orgID u
 
 // leadStatusClause builds the WHERE predicate for a derived lead status inside
 // ONE campaign, matching pg_contact Search's read-time derivation exactly:
-// unsubscribed > bounced > replied > completed > processing(active) >
+// unsubscribed > bounced > replied > failed > completed > processing(active) >
 // queued(pending). `cp` is the already-bound placeholder for that campaign id
 // (e.g. "$5"). Returns "" for an unknown status (the caller then applies no lead
 // filter).
@@ -1164,6 +1181,13 @@ func leadStatusClause(status, cp string) string {
 		)
 	}
 	sent, replied, bounced := has("sent_at"), has("replied_at"), has("bounced_at")
+	// failed: a step the mailbox could not send after every retry (sent_at was
+	// walked back and the attempt cap is spent).
+	failed := fmt.Sprintf(
+		"EXISTS (SELECT 1 FROM campaign_contact_progress p WHERE p.campaign_id = %s AND p.contact_id = c.id "+
+			"AND p.sent_at IS NULL AND p.failed_at IS NOT NULL AND p.send_attempts >= %d)",
+		cp, config.CampaignSendMaxAttempts,
+	)
 	// allSent: every email step of the campaign has been sent to this contact.
 	allSent := fmt.Sprintf(
 		"((SELECT COUNT(*) FROM sequences st WHERE st.campaign_id = %[1]s AND st.kind = 'email') > 0 "+
@@ -1178,12 +1202,14 @@ func leadStatusClause(status, cp string) string {
 		return fmt.Sprintf("(c.subscribed AND %s)", bounced)
 	case models.LeadStatusReplied:
 		return fmt.Sprintf("(c.subscribed AND NOT %s AND %s)", bounced, replied)
+	case models.LeadStatusFailed:
+		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND %s)", bounced, replied, failed)
 	case models.LeadStatusCompleted:
-		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND %s AND %s)", bounced, replied, sent, allSent)
+		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s AND %s AND %s)", bounced, replied, failed, sent, allSent)
 	case models.LeadStatusActive:
-		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND %s AND NOT %s)", bounced, replied, sent, allSent)
+		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s AND %s AND NOT %s)", bounced, replied, failed, sent, allSent)
 	case models.LeadStatusPending:
-		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s)", bounced, replied, sent)
+		return fmt.Sprintf("(c.subscribed AND NOT %s AND NOT %s AND NOT %s AND NOT %s)", bounced, replied, failed, sent)
 	default:
 		return ""
 	}
@@ -1192,22 +1218,24 @@ func leadStatusClause(status, cp string) string {
 // CampaignLeadCounts returns per-status lead totals for one campaign (the
 // campaign Leads view scope chips). A single aggregate over the campaign's
 // leads joined to their contact and a rolled-up view of their progress, so the
-// buckets follow the same unsubscribed > bounced > replied > completed >
-// processing > queued priority as the row-level derived status. Scoped to the
-// org through the contacts join.
+// buckets follow the same unsubscribed > bounced > replied > failed >
+// completed > processing > queued priority as the row-level derived status.
+// Scoped to the org through the contacts join.
 func (r *contactRepository) CampaignLeadCounts(ctx context.Context, orgID, campaignID string) (*models.CampaignLeadCounts, *errx.Error) {
 	// A lead is "done" (completed) when every email step has been sent and it
 	// hasn't replied or bounced; "processing" when some but not all steps sent.
 	const done = "ts.total_steps > 0 AND COALESCE(pr.sent_steps, 0) >= ts.total_steps"
+	const live = "c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND NOT COALESCE(pr.has_failed, false)"
 	query := fmt.Sprintf(`
 		SELECT
 			COUNT(*) AS total,
 			COUNT(*) FILTER (WHERE NOT c.subscribed) AS unsubscribed,
 			COUNT(*) FILTER (WHERE c.subscribed AND COALESCE(pr.has_bounced, false)) AS bounced,
 			COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND COALESCE(pr.has_replied, false)) AS replied,
-			COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND COALESCE(pr.has_sent, false) AND (%[1]s)) AS completed,
-			COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND COALESCE(pr.has_sent, false) AND NOT (%[1]s)) AS processing,
-			COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND NOT COALESCE(pr.has_sent, false)) AS queued
+			COUNT(*) FILTER (WHERE c.subscribed AND NOT COALESCE(pr.has_bounced, false) AND NOT COALESCE(pr.has_replied, false) AND COALESCE(pr.has_failed, false)) AS failed,
+			COUNT(*) FILTER (WHERE %[2]s AND COALESCE(pr.has_sent, false) AND (%[1]s)) AS completed,
+			COUNT(*) FILTER (WHERE %[2]s AND COALESCE(pr.has_sent, false) AND NOT (%[1]s)) AS processing,
+			COUNT(*) FILTER (WHERE %[2]s AND NOT COALESCE(pr.has_sent, false)) AS queued
 		FROM campaign_leads cl
 		JOIN contacts c ON c.id = cl.contact_id AND c.organization_id = $2
 		CROSS JOIN (SELECT COUNT(*) AS total_steps FROM sequences st WHERE st.campaign_id = $1 AND st.kind = 'email') ts
@@ -1216,15 +1244,16 @@ func (r *contactRepository) CampaignLeadCounts(ctx context.Context, orgID, campa
 				bool_or(p.sent_at IS NOT NULL)    AS has_sent,
 				bool_or(p.replied_at IS NOT NULL) AS has_replied,
 				bool_or(p.bounced_at IS NOT NULL) AS has_bounced,
+				bool_or(p.sent_at IS NULL AND p.failed_at IS NOT NULL AND p.send_attempts >= $3) AS has_failed,
 				COUNT(*) FILTER (WHERE p.sent_at IS NOT NULL) AS sent_steps
 			FROM campaign_contact_progress p
 			WHERE p.campaign_id = cl.campaign_id AND p.contact_id = cl.contact_id
 		) pr ON true
 		WHERE cl.campaign_id = $1
-	`, done)
+	`, done, live)
 	out := &models.CampaignLeadCounts{}
-	if err := r.DB.QueryRow(ctx, query, campaignID, orgID).Scan(
-		&out.Total, &out.Unsubscribed, &out.Bounced, &out.Replied, &out.Completed, &out.Processing, &out.Queued,
+	if err := r.DB.QueryRow(ctx, query, campaignID, orgID, config.CampaignSendMaxAttempts).Scan(
+		&out.Total, &out.Unsubscribed, &out.Bounced, &out.Replied, &out.Failed, &out.Completed, &out.Processing, &out.Queued,
 	); err != nil {
 		if err == pgx.ErrNoRows {
 			return out, nil

internal/repository/pg_email.go

@@ -102,6 +102,9 @@ type EmailRepository interface {
 	// worker reconciler uses it to (re)load accounts onto their assigned workers
 	// after onboarding, worker restarts, or reassignment.
 	ListActiveWorkerAccounts(ctx context.Context) ([]uuid.UUID, error)
+	// ListActiveAccountsByWorker returns the ids of the active mailboxes
+	// assigned to one worker, for reloading them after that worker restarts.
+	ListActiveAccountsByWorker(ctx context.Context, workerID uuid.UUID) ([]uuid.UUID, error)
 }
 
 type emailRepository struct {
@@ -220,6 +223,25 @@ func (r *emailRepository) ListActiveWorkerAccounts(ctx context.Context) ([]uuid.
 	return ids, rows.Err()
 }
 
+func (r *emailRepository) ListActiveAccountsByWorker(ctx context.Context, workerID uuid.UUID) ([]uuid.UUID, error) {
+	const query = `SELECT id FROM email_accounts WHERE status = 'active' AND worker_id = $1`
+	rows, err := r.DB.Query(ctx, query, workerID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var ids []uuid.UUID
+	for rows.Next() {
+		var id uuid.UUID
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, rows.Err()
+}
+
 func (r *emailRepository) CountForOrganization(ctx context.Context, orgID uuid.UUID) (int, *errx.Error) {
 	var count int
 	query := `SELECT COUNT(*) FROM email_accounts WHERE organization_id = $1`

internal/tasks/campaign_task.go

@@ -564,16 +564,23 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error {
 	}
 
 	if err := s.emailSender.Send(ctx, taskID, emailMsg, *account); err != nil {
-		// Failed to send to worker, record failure
+		// The send never reached a worker (none assigned, worker offline, bus
+		// or storage down). Nothing is stamped sent; the task is dead-lettered
+		// for the retry loop and the chain is re-seeded by the reconciler.
 		s.taskRepo.RecordTaskFailure(ctx, taskID, "Send failed", err.Error())
 		if s.campaignLogRepo != nil {
 			s.campaignLogRepo.CreateLog(ctx, &repository.CampaignLogEntry{
 				CampaignID: campaign.ID,
 				EventType:  "email_failed",
-				Message:    fmt.Sprintf("Failed to send to %s", contact.Email),
+				Message:    fmt.Sprintf("Could not hand %s's email to a sending worker, will retry: %s", contact.Email, err.Error()),
 				Metadata: map[string]interface{}{
-					"contact_id": contact.ID.String(),
-					"error":      err.Error(),
+					"level":       "error",
+					"code":        "WORKER_UNAVAILABLE",
+					"contact_id":  contact.ID.String(),
+					"sequence_id": sequence.ID.String(),
+					"account_id":  account.ID.String(),
+					"error":       err.Error(),
+					"will_retry":  true,
 				},
 			})
 		}

internal/tasks/email_sender.go

@@ -2,10 +2,12 @@ package tasks
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"strings"
 
 	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
 	"github.com/warmbly/warmbly/internal/events"
 	"github.com/warmbly/warmbly/internal/models"
 	"github.com/warmbly/warmbly/internal/repository"
@@ -41,9 +43,23 @@ type EmailSender interface {
 	Send(ctx context.Context, taskID uuid.UUID, msg EmailMessage, account models.Email) error
 }
 
+// WorkerLiveness reports whether a worker is still heartbeating. Satisfied by
+// repository.WorkerRepository.
+type WorkerLiveness interface {
+	IsWorkerLive(ctx context.Context, workerID uuid.UUID) (bool, error)
+}
+
+// ErrWorkerOffline is returned by Send when the mailbox's worker has stopped
+// heartbeating. The send is not published: a command queued for a worker that
+// is gone is never executed and never answered, which would leave the step
+// looking sent forever. The worker reconciler moves the mailbox to a live
+// worker and the dead-letter retry replays the task.
+var ErrWorkerOffline = errors.New("the mailbox's sending worker is offline")
+
 type emailSender struct {
 	emailRepo repository.EmailRepository
 	publisher events.Publisher
+	liveness  WorkerLiveness
 }
 
 // NewEmailSender creates a new email sender
@@ -54,6 +70,12 @@ func NewEmailSender(emailRepo repository.EmailRepository, publisher events.Publi
 	}
 }
 
+// WireWorkerLiveness makes Send refuse to publish to a worker that is not
+// heartbeating. Optional; without it Send trusts the assignment.
+func (s *emailSender) WireWorkerLiveness(l WorkerLiveness) {
+	s.liveness = l
+}
+
 // Send publishes an email to the worker service for sending
 func (s *emailSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessage, account models.Email) error {
 	// Get worker ID for this email account
@@ -61,6 +83,15 @@ func (s *emailSender) Send(ctx context.Context, taskID uuid.UUID, msg EmailMessa
 	if workerID == nil {
 		return fmt.Errorf("no worker assigned to email account %s", account.ID)
 	}
+	if s.liveness != nil {
+		live, err := s.liveness.IsWorkerLive(ctx, *workerID)
+		if err != nil {
+			return fmt.Errorf("check worker %s liveness: %w", workerID, err)
+		}
+		if !live {
+			return fmt.Errorf("%w (worker %s, mailbox %s)", ErrWorkerOffline, workerID, account.Email)
+		}
+	}
 
 	// Email content is sealed with the organization DEK; an account without an
 	// organization cannot be encrypted for transport.
@@ -115,3 +146,39 @@ func generateMessageID(fromEmail string) string {
 	// Generate unique ID
 	return fmt.Sprintf("<%s@%s>", uuid.New().String(), domain)
 }
+
+// HeartbeatChecker reports whether a worker's short-lived heartbeat key is
+// present. Satisfied by *cache.Cache (go-redis Exists).
+type HeartbeatChecker interface {
+	Exists(ctx context.Context, keys ...string) *redis.IntCmd
+}
+
+// workerLiveness combines the registry's view of a worker (active, seen in the
+// last ten minutes) with its three-minute heartbeat key, so a crashed worker
+// stops receiving sends within minutes rather than at the end of the registry
+// window. A heartbeat lookup error fails open to the registry's answer: a
+// cache blip must not stop every send.
+type workerLiveness struct {
+	repo      WorkerLiveness
+	heartbeat HeartbeatChecker
+}
+
+// NewWorkerLiveness builds the liveness check Send uses. heartbeat may be nil.
+func NewWorkerLiveness(repo WorkerLiveness, heartbeat HeartbeatChecker) WorkerLiveness {
+	return &workerLiveness{repo: repo, heartbeat: heartbeat}
+}
+
+func (l *workerLiveness) IsWorkerLive(ctx context.Context, workerID uuid.UUID) (bool, error) {
+	live, err := l.repo.IsWorkerLive(ctx, workerID)
+	if err != nil || !live {
+		return live, err
+	}
+	if l.heartbeat == nil {
+		return true, nil
+	}
+	n, herr := l.heartbeat.Exists(ctx, "worker:heartbeat:"+workerID.String()).Result()
+	if herr != nil {
+		return true, nil
+	}
+	return n > 0, nil
+}

web/src/components/app/contacts/ContactsTable.tsx

@@ -990,6 +990,7 @@ const LEAD_META: Record<
     completed: { label: "Done", dot: "bg-indigo-500", text: "text-indigo-700", Icon: CheckIcon },
     replied: { label: "Replied", dot: "bg-emerald-500", text: "text-emerald-700", Icon: CornerUpLeftIcon },
     bounced: { label: "Bounced", dot: "bg-rose-500", text: "text-rose-600", Icon: AlertTriangleIcon },
+    failed: { label: "Failed", dot: "bg-rose-500", text: "text-rose-600", Icon: AlertTriangleIcon },
     unsubscribed: { label: "Unsubscribed", dot: "bg-slate-300", text: "text-slate-400", Icon: BanIcon },
 };
 
@@ -997,9 +998,16 @@ function LeadStatusPill({ lead }: { lead?: ContactCampaignProgress | null }) {
     const status: LeadStatus = lead?.status ?? "pending";
     const meta = LEAD_META[status];
     const Icon = meta.Icon;
+    // A failed lead carries the worker's reason; surface it on hover since the
+    // pill itself only has room for the word.
+    const title =
+        status === "failed" && lead?.failure_reason
+            ? `Could not send: ${lead.failure_reason}`
+            : undefined;
     return (
         <span
             className={`inline-flex items-center gap-1.5 text-[10.5px] font-medium uppercase tracking-[0.08em] ${meta.text}`}
+            title={title}
         >
             {status === "active" ? (
                 <span className="campaign-grid text-sky-600 shrink-0" aria-hidden />
@@ -1030,6 +1038,7 @@ function LeadProgressStrip({
             completed: 0,
             replied: 0,
             bounced: 0,
+            failed: 0,
             unsubscribed: 0,
         };
         for (const ct of contacts) c[ct.campaign_lead?.status ?? "pending"]++;
@@ -1045,6 +1054,7 @@ function LeadProgressStrip({
         { key: "replied", color: "bg-emerald-500" },
         { key: "pending", color: "bg-slate-300" },
         { key: "bounced", color: "bg-rose-400" },
+        { key: "failed", color: "bg-rose-500" },
         { key: "unsubscribed", color: "bg-slate-200" },
     ];
 
@@ -1069,6 +1079,7 @@ function LeadProgressStrip({
                 <StripChip dot="bg-emerald-500" label="Replied" n={counts.replied} />
                 <StripChip dot="bg-slate-300" label="Queued" n={counts.pending} />
                 <StripChip dot="bg-rose-400" label="Bounced" n={counts.bounced} />
+                {counts.failed > 0 && <StripChip dot="bg-rose-500" label="Failed" n={counts.failed} />}
                 <StripChip dot="bg-slate-300" label="Unsub" n={counts.unsubscribed} />
             </div>
             <div className="ml-auto flex items-center gap-2 text-[10.5px] text-slate-400 tabular-nums">

web/src/hooks/useRealtimeEvents.ts

@@ -171,6 +171,24 @@ export function useRealtimeEvents() {
         return
       }
 
+      // A user-addressed mailbox/send error (a compose or reply the worker
+      // could not send, a mailbox that needs re-authorizing). Nothing else
+      // would tell the user, so toast it, then refresh the mailbox views.
+      if (event === 'ERROR') {
+        const data = payload.data as Record<string, unknown> | undefined
+        const title =
+          typeof data?.title === 'string' && data.title
+            ? data.title
+            : (getString('message') ?? 'Something went wrong')
+        const detail = typeof data?.message === 'string' ? data.message : ''
+        toast.error(detail ? `${title}: ${detail}` : title, {
+          id: `email-error-${getString('task_id') ?? getString('email_id') ?? 'general'}`,
+          duration: 8000,
+        })
+        invalidate([['emails', 'list'], ['unibox']])
+        return
+      }
+
       if (includes('ACCOUNT', 'EMAIL_STATUS', 'EMAIL_ERROR', 'WARMUP')) {
         // ACCOUNT_SYNC_STATE: the mailbox's import finished or fair use
         // started/stopped holding it; the drawer's sync card refetches.

web/src/lib/api/models/app/contacts/Contact.ts

@@ -1,15 +1,17 @@
 import type MiniCampaign from "../campaigns/MiniCampaign";
 import type MiniCategory from "./MiniCategory";
 
-// LeadStatus mirrors models.ContactCampaignProgress.Status — a contact's
+// LeadStatus mirrors models.ContactCampaignProgress.Status, a contact's
 // processing state inside a single campaign. "completed" = every step sent, no
-// reply (done); "active" = some but not all steps sent (still processing).
+// reply (done); "active" = some but not all steps sent (still processing);
+// "failed" = the mailbox could not send a step after every retry.
 export type LeadStatus =
     | "pending"
     | "active"
     | "completed"
     | "replied"
     | "bounced"
+    | "failed"
     | "unsubscribed";
 
 // ContactCampaignProgress is set only on contacts returned by a single-campaign
@@ -25,6 +27,9 @@ export interface ContactCampaignProgress {
     // Label of the step the lead is on now (latest step sent). Empty when the
     // lead hasn't been contacted yet.
     current_step?: string;
+    // The worker's reason for the last failed send; set only when status is
+    // "failed".
+    failure_reason?: string;
 }
 
 export default interface Contact {