Merge remote-tracking branch 'origin/main' into fix/main-ci-failure-and-issue-400

2026-09-12 00:05:09 +00:00 · 2026-09-09 09:05:45 -07:00
parent 92b811773e 128c2261e3
commit cab27fac62
30 changed files with 1224 additions and 239 deletions
@@ -94,7 +94,7 @@ Returns a `data` array of contacts plus a `pagination` envelope. Paginate by pas

 Every contact carries its address verification: `verification_status` (`valid`, `risky`, `invalid`, or `unknown`), `verification_sub_status` (`catch_all`, `disposable`, `role`, `spamtrap`, `mailbox_full`, `no_mx`, `syntax`, `undisclosed`, or empty), `verification_source` (`probe` for the built-in check, `provider` for a connected verification service, `imported` for a verdict that came with the contact, `manual` for one a member set, empty when never checked), `verification_provider` (who produced it), `verification_reason`, `verification_checked_at`, and `verification_confidence` (0 to 100, scored from the check plus what real mail to the address showed; see [what real mail teaches the check](/guides/deliverability/#what-real-mail-teaches-the-check)). Campaigns never send to `invalid`, and send to `risky` only when their `risky_emails` setting is on.

-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`, `machine_opened`, `clicked`, `replied`, `bounced`, `current_step`, `last_activity_at`, and `failure_reason` when failed). `opened` counts steps opened by a person; steps fetched automatically by a mail client (Apple Mail Privacy Protection and similar) are in `machine_opened` instead, matching the machine opens the analytics summary reports. 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), `undeliverable` (pre-send verification refused the address, so the campaign skips the lead and never sends to it), 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, each contact additionally carries a `campaign_lead` object with its processing state inside that campaign (`status`, `sent`, `opened`, `machine_opened`, `clicked`, `replied`, `bounced`, `current_step`, `sender`, `last_activity_at`, and `failure_reason` when failed). `sender` is the mailbox address the lead's whole sequence sends from, fixed when its first email went out and absent until then. `opened` counts steps opened by a person; steps fetched automatically by a mail client (Apple Mail Privacy Protection and similar) are in `machine_opened` instead, matching the machine opens the analytics summary reports. 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), `undeliverable` (pre-send verification refused the address, so the campaign skips the lead and never sends to it), 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` and `engagement` filters so every scope's total is available at once. Alongside the status buckets it carries engagement totals that match the `engagement` filter: `contacted` (leads sent at least one step), `opened` (a human open on any step), `clicked`, and `replied_any` (a reply on any step, whatever the derived status).

@@ -812,6 +812,8 @@ Auth: **Scope** `READ_CONTACTS` · **Org permission** `view_contacts`
      "campaign_name": "Q3 Outbound",
      "campaign_status": "active",
      "lead_status": "active",
+      "sender_id": "m1...",
+      "sender_email": "alex@acme.com",
      "steps": [
        {
          "id": "s1...",
@@ -843,6 +845,8 @@ Auth: **Scope** `READ_CONTACTS` · **Org permission** `view_contacts`
 }
 ```

+`sender_id` and `sender_email` are the mailbox this lead's whole sequence sends from. Rotation picks it when the first email goes out and every follow-up keeps it, so the contact only ever hears from one address; both fields are absent until that first email. They change only when that mailbox can no longer send for the campaign.
+
 `lead_status` uses the same values as the campaign Leads view: `pending`, `active`, `completed`, `replied`, `bounced`, `failed`, `unsubscribed`, or `undeliverable`. Each step carries whichever of `sent_at`, `opened_at`, `clicked_at`, `replied_at`, `bounced_at` and `failed_at` apply, plus `attempts` and `in_flight` (reserved for a worker whose result has not come back). `opened_at` is a person's open, as it is in the Leads view: a step a mail client prefetched or a security gateway scanned carries no `opened_at`. While a branch condition is undecided, `next.step_id` is absent and `next.step_label` says the step depends on the contact's response.

 ## List a contact's activities
@@ -40,7 +40,7 @@ Pick mailboxes in **Sending accounts** three ways, and the first two combine:
 Sending accounts always resolve inside the campaign's own workspace. If you belong to more than one workspace, a tag you reuse across them still only picks up mailboxes belonging to the workspace that owns the campaign, so one workspace's sending reputation, daily caps, and warmup are never spent on another's traffic.

 <Callout type="info" title="Follow-ups stay on the same mailbox">
-Each mailbox stays within its own daily limit, and follow-ups always send from the mailbox that sent the first email to that contact, so threads stay consistent.
+Rotation chooses a mailbox for a lead's **first** email, and that mailbox sends the rest of that lead's sequence. Three emails from three addresses is not a conversation, and a reply belongs in the mailbox the contact was written to from.
 </Callout>

 | Rotation | Behavior |
@@ -49,7 +49,20 @@ Each mailbox stays within its own daily limit, and follow-ups always send from t
 | Round-robin | Cycles in order: `A`, `B`, `C`, back to `A` |
 | Weighted | Sends more from healthier, higher-limit mailboxes |

-**ESP matching** optionally aligns sender and recipient providers (Gmail to Gmail). **Prefer same** falls back to any mailbox when no same-provider one is free; **Strict same** makes the contact wait instead. An unknown recipient provider never blocks a send.
+Whichever mode you pick, it decides which mailbox **starts** a lead. Every mailbox still stays inside its own daily cap and minimum gap, so a campaign's volume is the sum of its mailboxes' budgets as always.
+
+The Leads tab shows each lead's mailbox in its **Sender** column, and a contact's Activity tab shows it beside their progress. A lead has none until its first email goes out.
+
+**ESP matching** optionally aligns sender and recipient providers (Gmail to Gmail). **Prefer same** falls back to any mailbox when no same-provider one is free; **Strict same** makes the contact wait instead. An unknown recipient provider never blocks a send. Like rotation, it applies to a lead's first email: a follow-up goes out from the mailbox the contact already knows.
+
+### When a lead's mailbox is busy, or gone
+
+Two things can happen to a lead's mailbox, and Warmbly treats them differently:
+
+- **Busy**: it has used its daily budget, or it is outside its own sending hours. The lead waits for it, and the rest of the campaign keeps sending from the mailboxes that are free. Nothing is lost: the lead sends when the mailbox is next available, and the activity log notes the wait once a day.
+- **Gone**: it was disconnected, removed from the campaign's sending accounts, sent to rest, held by warmup health, or its domain is failing authentication. Waiting would leave the contact silent for days, so the lead moves to another mailbox in the pool and the campaign's activity log records the change.
+
+Removing a mailbox from **Sending accounts** therefore hands its in-flight leads to the remaining mailboxes as each of them reaches its next step. Adding a mailbox never moves an existing lead; it picks up new ones.

 ## Sending limits

@@ -219,11 +219,13 @@ A mailbox can be set inactive (`PATCH /emails/{id}` with `status`) when you want

 Switching it off takes effect immediately: the machine syncing it is told to drop it, so it stops importing mail and stops being picked for campaign sends and warmup within seconds rather than at that machine's next restart. Warmup pool membership is dropped, and any warmup chain it had winds down on its next step. A campaign email already handed over is answered as a failure and the step is retried later on a mailbox that is still active, so nobody receives it twice and no lead is stranded.

+Leads mid-sequence on that mailbox move to another one in their campaign as they reach their next step, rather than going silent waiting for it. Each campaign's activity log records the change. See [senders and rotation](/guides/campaigns/) for how a lead's mailbox is chosen and when it changes.
+
 It keeps its worker assignment while off, so switching it back on puts it back on the same machine, sending from the same IP, and it resumes syncing from where it stopped instead of re-importing.

 **Disconnecting** removes the mailbox for good, from the Accounts selection bar. The machine syncing it is told to drop it before the record is removed, because afterwards there is nothing left to tell. If that instruction cannot be delivered, the disconnect fails with a `503` and nothing is removed, so retry it in a moment rather than assuming it worked.

-Everything belonging to that mailbox goes with it: its imported mail in the unibox, its warmup history and pool membership, its credentials, its sender links, and any send still scheduled for it. A campaign that was using it keeps running on its remaining senders. Export the workspace first if you want a copy. Disable the mailbox instead when you only want it to stop.
+Everything belonging to that mailbox goes with it: its imported mail in the unibox, its warmup history and pool membership, its credentials, its sender links, and any send still scheduled for it. A campaign that was using it keeps running on its remaining senders, and the leads it had been writing to move onto them at their next step. Export the workspace first if you want a copy. Disable the mailbox instead when you only want it to stop.

 ## Worker assignment

@@ -254,7 +256,7 @@ A mailbox now has a cold-rotation state, separate from its health:
 | State | Meaning |
 |-------|---------|
 | `active` | In cold rotation. The default, and where every mailbox starts |
-| `resting` | Out of cold rotation to recover. Warmup keeps running, so its reputation stays alive |
+| `resting` | Out of cold rotation to recover. Warmup keeps running, so its reputation stays alive. Leads mid-sequence on it move to another mailbox at their next step |
 | `reserve` | Held back by you. Never entered or left automatically |

 A mailbox rests when its warmup health reaches `throttled` or worse, and returns on its own once it is healthy again **and** has held steady for three days. One good hour does not put it back at full cold volume.
@@ -526,6 +526,10 @@ func (d Deps) listCampaignLeads(ctx context.Context, inv Invocation, args json.R
 			row["opened"] = lp.Opened
 			row["replied"] = lp.Replied
 			row["current_step"] = lp.CurrentStep
+			if lp.Sender != "" {
+				// The mailbox this lead's whole sequence sends from.
+				row["sender"] = lp.Sender
+			}
 			if lp.LastActivityAt != nil {
 				row["last_activity_at"] = lp.LastActivityAt
 			}
@@ -109,7 +109,7 @@ func (f *sendResultFixture) dispatch(t *testing.T, s *JobsService) uuid.UUID {
 	if err := s.TaskRepo.UpdateCampaignTaskTracking(ctx, taskID, f.contact, f.step); err != nil {
 		t.Fatalf("tracking: %v", err)
 	}
-	reserved, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, taskID, true)
+	reserved, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, taskID, uuid.Nil, true)
 	if err != nil || !reserved {
 		t.Fatalf("reserve send: reserved=%v err=%v", reserved, err)
 	}
@@ -164,7 +164,7 @@ func TestLiveHandleEmailFailedWalksBackAndRetriesUntilCap(t *testing.T) {
 	}
 	nextPair := func() *repository.ContactSequencePair {
 		t.Helper()
-		pair, _, err := s.CampaignProgressRepo.FindNextRoutedPair(ctx, f.campaign, "created_at", "asc", "", false, false)
+		pair, _, _, err := s.CampaignProgressRepo.FindNextRoutedPair(ctx, f.campaign, "created_at", "asc", "", false, false, nil)
 		if err != nil {
 			t.Fatalf("next pair: %v", err)
 		}
@@ -53,7 +53,7 @@ func liveJobsService(handle *db.DB) *JobsService {
 // nextPair asks routing what it would send next, exactly as the scheduler does.
 func (f *sendResultFixture) nextPair(t *testing.T, s *JobsService) *repository.ContactSequencePair {
 	t.Helper()
-	pair, _, err := s.CampaignProgressRepo.FindNextRoutedPair(context.Background(), f.campaign, "created_at", "asc", "", false, false)
+	pair, _, _, err := s.CampaignProgressRepo.FindNextRoutedPair(context.Background(), f.campaign, "created_at", "asc", "", false, false, nil)
 	if err != nil {
 		t.Fatalf("next pair: %v", err)
 	}
@@ -147,11 +147,11 @@ func TestLiveReserveSendClaimsAStepExactlyOnce(t *testing.T) {
 	s := liveJobsService(handle)
 	f := newSendResultFixture(t, handle)

-	first, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), true)
+	first, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), uuid.Nil, true)
 	if err != nil || !first {
 		t.Fatalf("first reservation: claimed=%v err=%v", first, err)
 	}
-	second, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), true)
+	second, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), uuid.Nil, true)
 	if err != nil {
 		t.Fatalf("second reservation: %v", err)
 	}
@@ -173,7 +173,7 @@ func TestLiveReserveSendClaimsAStepExactlyOnce(t *testing.T) {
 	if err := s.CampaignProgressRepo.RecordEmailSent(ctx, f.campaign, f.contact, f.step); err != nil {
 		t.Fatalf("record sent: %v", err)
 	}
-	third, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), true)
+	third, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), uuid.Nil, true)
 	if err != nil {
 		t.Fatalf("third reservation: %v", err)
 	}
@@ -191,7 +191,7 @@ func TestLiveReleaseSendReturnsTheStep(t *testing.T) {
 	s := liveJobsService(handle)
 	f := newSendResultFixture(t, handle)

-	if _, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), true); err != nil {
+	if _, err := s.CampaignProgressRepo.ReserveSend(ctx, f.campaign, f.contact, f.step, uuid.New(), uuid.Nil, true); err != nil {
 		t.Fatalf("reserve: %v", err)
 	}
 	if err := s.CampaignProgressRepo.ReleaseSend(ctx, f.campaign, f.contact, f.step, true); err != nil {
@@ -162,6 +162,13 @@ func constraintCopy(c scheduler.ContactSendConstraint, st *models.ContactCampaig
 		return "Today's new-lead limit is reached"
 	case scheduler.ConstraintCapacity:
 		return "No mailbox can take it right now (daily cap, spacing, or health)"
+	case scheduler.ConstraintSenderBusy:
+		// One mailbox is the gate, not the pool: this contact's whole sequence
+		// sends from the address they first heard from.
+		if st.SenderEmail != "" {
+			return "Waiting for " + st.SenderEmail + ", which sends the rest of this sequence"
+		}
+		return "Waiting for the mailbox that sends this contact's sequence"
 	case scheduler.ConstraintCampaignInactive:
 		switch st.CampaignStatus {
 		case "draft":
@@ -0,0 +1,6 @@
+DROP INDEX IF EXISTS idx_campaign_leads_sender;
+DROP INDEX IF EXISTS idx_campaign_tasks_lead;
+
+ALTER TABLE campaign_leads
+    DROP COLUMN IF EXISTS email_account_id,
+    DROP COLUMN IF EXISTS sender_assigned_at;
@@ -0,0 +1,47 @@
+-- A lead's sending mailbox is fixed for the whole sequence. Rotation picks the
+-- mailbox for a lead's FIRST email and that mailbox sends every follow-up, so a
+-- contact never hears from three different addresses in one conversation and
+-- their replies always land in the mailbox that wrote to them (issue #401).
+-- Rotation still spreads NEW leads across the pool.
+--
+-- Nullable: a lead has no sender until its first email is reserved. ON DELETE
+-- SET NULL because a disconnected mailbox must not take its leads with it — the
+-- scheduler moves them to another mailbox on their next step.
+ALTER TABLE campaign_leads
+    ADD COLUMN email_account_id uuid REFERENCES email_accounts (id) ON DELETE SET NULL,
+    ADD COLUMN sender_assigned_at timestamptz;
+
+-- Which mailbox a lead was last actually sent from. The backfill below reads
+-- it, and so does the scheduler when a lead has steps but no binding (removed
+-- from the campaign and added back keeps the steps, not the row).
+CREATE INDEX IF NOT EXISTS idx_campaign_tasks_lead
+    ON campaign_tasks (campaign_id, contact_id, task_id)
+    WHERE campaign_id IS NOT NULL AND contact_id IS NOT NULL;
+
+-- Leads already mid-sequence keep the address they have been writing from. The
+-- earliest completed campaign task for the pair is the mailbox the contact saw
+-- first; a mailbox that has since been deleted leaves the lead unassigned, and
+-- its next step picks one by rotation exactly as a new lead does.
+WITH first_send AS (
+    SELECT DISTINCT ON (ct.campaign_id, ct.contact_id)
+           ct.campaign_id, ct.contact_id, t.email_account_id, t.created_at
+    FROM campaign_tasks ct
+    JOIN tasks t ON t.id = ct.task_id
+    WHERE ct.campaign_id IS NOT NULL
+      AND ct.contact_id IS NOT NULL
+      AND t.task_type = 'campaign'
+      AND t.status = 'completed'
+    ORDER BY ct.campaign_id, ct.contact_id, t.created_at ASC
+)
+UPDATE campaign_leads cl
+SET email_account_id   = fs.email_account_id,
+    sender_assigned_at = fs.created_at
+FROM first_send fs
+WHERE cl.campaign_id = fs.campaign_id
+  AND cl.contact_id = fs.contact_id
+  AND EXISTS (SELECT 1 FROM email_accounts ea WHERE ea.id = fs.email_account_id);
+
+-- Read by the mailbox drawer's "leads pinned here" count and by the FK's own
+-- ON DELETE SET NULL sweep.
+CREATE INDEX IF NOT EXISTS idx_campaign_leads_sender
+    ON campaign_leads (email_account_id) WHERE email_account_id IS NOT NULL;
@@ -100,6 +100,9 @@ type ContactCampaignProgress struct {
 	// FailureReason is the worker's reason for the last failed send, set only
 	// when Status is "failed".
 	FailureReason string `json:"failure_reason,omitempty"`
+	// Sender is the mailbox address this lead's whole sequence sends from,
+	// fixed when its first email went out. Empty until then.
+	Sender string `json:"sender,omitempty"`
 }

 // Lead status constants for ContactCampaignProgress.Status.
@@ -14,6 +14,12 @@ type ContactCampaignState struct {
 	CampaignStatus string    `json:"campaign_status"`
 	// LeadStatus follows ContactCampaignProgress.Status exactly.
 	LeadStatus string `json:"lead_status"`
+
+	// SenderID / SenderEmail are the mailbox this lead's whole sequence sends
+	// from, fixed when its first email went out. Empty until then: rotation
+	// picks it for the first step and every follow-up follows.
+	SenderID    *uuid.UUID `json:"sender_id,omitempty"`
+	SenderEmail string     `json:"sender_email,omitempty"`
 	// FailureReason is the worker's reason for the last failed send.
 	FailureReason string `json:"failure_reason,omitempty"`

@@ -72,6 +72,10 @@ type ContactSequencePair struct {
 	// "due now". The placer floors its base time with it, so a wait routing
 	// honours can never be dropped by the placement pass.
 	NotBefore *time.Time
+	// AssignedSender is the mailbox this lead's sequence is bound to, set when
+	// its first email was reserved. The placer sends from it rather than
+	// rotating, so every step of one conversation comes from one address.
+	AssignedSender *uuid.UUID
 }

 type CampaignSequencePair struct {
@@ -100,11 +104,17 @@ type CampaignProgressRepository interface {
 	// the caller must NOT dispatch. Resolve every reservation exactly once:
 	// RecordEmailSent on a successful hand-off, ReleaseSend when the command
 	// provably never left, RecordSendFailure on a worker failure.
-	ReserveSend(ctx context.Context, campaignID, contactID, sequenceID, taskID uuid.UUID, newLead bool) (bool, error)
+	//
+	// senderID binds the lead to the mailbox this send goes out from, in the
+	// same transaction, so the rest of its sequence follows the address the
+	// contact has already seen. Pass uuid.Nil to leave the binding alone.
+	ReserveSend(ctx context.Context, campaignID, contactID, sequenceID, taskID, senderID uuid.UUID, newLead bool) (bool, error)
 	// ReleaseSend gives a reservation back when the send provably never reached
 	// the bus (no worker assigned, worker offline), so the step is retried on the
 	// next tick without spending an attempt. Only an unstamped reservation is
-	// touched, so it can never undo a real send.
+	// touched, so it can never undo a real send. A lead whose binding to a
+	// mailbox came from this reservation alone is unbound again, so a mailbox
+	// that never actually wrote to the contact cannot keep their sequence.
 	ReleaseSend(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, newLead bool) error
 	// Record email status
 	RecordEmailSent(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error
@@ -123,7 +133,16 @@ type CampaignProgressRepository interface {
 	// 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).
+	// A lead with nothing else delivered is unbound from the mailbox that
+	// failed, so rotation can offer it a working one.
 	RecordSendFailure(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID, reason string) (attempts int, exhausted bool, rolledBack bool, err error)
+	// LastSenderForLead is the mailbox a lead was LAST actually sent from,
+	// read from the campaign tasks that dispatched its steps. It answers the
+	// case campaign_leads.email_account_id cannot: a lead removed from the
+	// campaign and added back keeps every step it was sent but starts a new
+	// row, and the address it last heard from is still the one to keep. Nil
+	// when nothing was ever sent to it in this campaign.
+	LastSenderForLead(ctx context.Context, campaignID, contactID uuid.UUID) (*uuid.UUID, 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).
@@ -199,7 +218,15 @@ type CampaignProgressRepository interface {
 	// while follow-ups keep flowing. The second return value, when the pair is
 	// nil, is the soonest time a waiting contact's condition window elapses — the
 	// scheduler should defer and re-check then rather than completing.
-	FindNextRoutedPair(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool) (*ContactSequencePair, *time.Time, error)
+	// paced names the pool mailboxes that cannot take a send right now but will
+	// be able to on their own, mapped to when; a lead already bound to one is
+	// skipped for this pass and reported through the next-due time instead of
+	// parking every lead behind it. Nil applies no sender gate.
+	//
+	// waitingOnSender reports that the returned next-due moment is a lead
+	// waiting for its own mailbox rather than for a step's wait or a condition
+	// window, so the caller can say which.
+	FindNextRoutedPair(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool, paced PacedSenders) (pair *ContactSequencePair, nextDue *time.Time, waitingOnSender bool, err error)
 	// RouteContact runs the same routing for ONE contact and reports where
 	// their flow goes next, plus the pre-send gate that excludes them, so a
 	// per-contact preview reads the facts the send path reads.
@@ -230,7 +257,7 @@ func NewCampaignProgressRepository(db *pgxpool.Pool) CampaignProgressRepository
 // flight (dispatched_at set) or already sent updates nothing and returns no
 // row, so two ticks that picked the same pair cannot both dispatch. A step
 // walked back after a worker failure has both cleared and is claimable again.
-func (r *campaignProgressRepository) ReserveSend(ctx context.Context, campaignID, contactID, sequenceID, taskID uuid.UUID, newLead bool) (bool, error) {
+func (r *campaignProgressRepository) ReserveSend(ctx context.Context, campaignID, contactID, sequenceID, taskID, senderID uuid.UUID, newLead bool) (bool, error) {
 	tx, err := r.db.Begin(ctx)
 	if err != nil {
 		return false, err
@@ -254,6 +281,22 @@ func (r *campaignProgressRepository) ReserveSend(ctx context.Context, campaignID
 		return false, err
 	}

+	// Bind the lead to the mailbox this send leaves from. Written with the
+	// claim, so the binding exists for every send that was ever dispatched and
+	// for none that was not. Re-stated on every step: the placer only hands a
+	// different mailbox when the bound one is gone, and that decision is what
+	// this records.
+	if senderID != uuid.Nil {
+		if _, err := tx.Exec(ctx, `
+			UPDATE campaign_leads
+			SET email_account_id = $3, sender_assigned_at = NOW()
+			WHERE campaign_id = $1 AND contact_id = $2
+			  AND email_account_id IS DISTINCT FROM $3
+		`, campaignID, contactID, senderID); err != nil {
+			return false, err
+		}
+	}
+
 	newLeadInc := 0
 	if newLead {
 		newLeadInc = 1
@@ -295,6 +338,23 @@ func (r *campaignProgressRepository) ReleaseSend(ctx context.Context, campaignID
 		return err
 	}

+	// The mailbox never wrote to this contact, so it must not keep their
+	// sequence: a lead bound by this reservation alone is unbound again, and
+	// the next tick picks by rotation as it would for any new lead. A lead with
+	// an earlier delivered step keeps its binding.
+	if _, err := tx.Exec(ctx, `
+		UPDATE campaign_leads cl
+		SET email_account_id = NULL, sender_assigned_at = NULL
+		WHERE cl.campaign_id = $1 AND cl.contact_id = $2
+		  AND cl.email_account_id IS NOT NULL
+		  AND NOT EXISTS (
+		    SELECT 1 FROM campaign_contact_progress p
+		    WHERE p.campaign_id = $1 AND p.contact_id = $2 AND p.sent_at IS NOT NULL
+		  )
+	`, campaignID, contactID); err != nil {
+		return err
+	}
+
 	newLeadDec := 0
 	if newLead {
 		newLeadDec = 1
@@ -384,17 +444,36 @@ func (r *campaignProgressRepository) RecordSendFailure(ctx context.Context, camp
 	if len(reason) > 500 {
 		reason = reason[:500]
 	}
+	// The unbind rides along in the same statement so a failure can never leave
+	// a lead pinned to a mailbox that delivered it nothing. Its NOT EXISTS reads
+	// the pre-statement snapshot, so the step being walked back is excluded by
+	// hand — otherwise a step this call is clearing would still count as
+	// delivered and hold the binding.
 	query := `
-		UPDATE campaign_contact_progress
-		SET sent_at = NULL,
-		    dispatched_at = NULL,
-		    dispatch_task_id = 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 OR dispatched_at IS NOT NULL)
-		RETURNING send_attempts
+		WITH walked AS (
+			UPDATE campaign_contact_progress
+			SET sent_at = NULL,
+			    dispatched_at = NULL,
+			    dispatch_task_id = 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 OR dispatched_at IS NOT NULL)
+			RETURNING send_attempts
+		), unbound AS (
+			UPDATE campaign_leads cl
+			SET email_account_id = NULL, sender_assigned_at = NULL
+			WHERE EXISTS (SELECT 1 FROM walked)
+			  AND cl.campaign_id = $1 AND cl.contact_id = $2
+			  AND cl.email_account_id IS NOT NULL
+			  AND NOT EXISTS (
+			    SELECT 1 FROM campaign_contact_progress p
+			    WHERE p.campaign_id = $1 AND p.contact_id = $2
+			      AND p.sequence_id <> $3 AND p.sent_at IS NOT NULL
+			  )
+		)
+		SELECT send_attempts FROM walked
 	`
 	var attempts int
 	err := r.db.QueryRow(ctx, query, campaignID, contactID, sequenceID, reason).Scan(&attempts)
@@ -407,6 +486,30 @@ func (r *campaignProgressRepository) RecordSendFailure(ctx context.Context, camp
 	return attempts, attempts >= config.CampaignSendMaxAttempts, true, nil
 }

+// LastSenderForLead reads the mailbox of the most recent campaign task that
+// dispatched a step to this lead. Only completed email tasks carry a contact
+// id (action nodes return before the tracking stamp), so this is exactly the
+// set of sends the contact actually received.
+func (r *campaignProgressRepository) LastSenderForLead(ctx context.Context, campaignID, contactID uuid.UUID) (*uuid.UUID, error) {
+	var id uuid.UUID
+	err := r.db.QueryRow(ctx, `
+		SELECT t.email_account_id
+		FROM campaign_tasks ct
+		JOIN tasks t ON t.id = ct.task_id
+		WHERE ct.campaign_id = $1 AND ct.contact_id = $2
+		  AND t.task_type = 'campaign' AND t.status = 'completed'
+		ORDER BY t.created_at DESC
+		LIMIT 1
+	`, campaignID, contactID).Scan(&id)
+	if errors.Is(err, pgx.ErrNoRows) || errors.Is(err, sql.ErrNoRows) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return &id, 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) {
@@ -916,14 +1019,23 @@ func undeliverableClause(cp string) string {
 // every other lead's first email sits queued. When nothing is due, the second
 // value is the soonest moment something will be (a step becoming due or a
 // condition window closing) so the scheduler defers exactly until then.
-// Returns (nil, nil, nil) when the campaign is genuinely complete.
-func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool) (*ContactSequencePair, *time.Time, error) {
+// A lead already bound to a mailbox (its first email went out from it) is only
+// offered while that mailbox can take a send: one that is merely out of budget
+// for today or outside its hours is listed in `paced`, and the lead waits for
+// it rather than switching addresses mid-conversation. Its reopening is
+// reported as a next-due time, so the leads behind it keep sending and the
+// campaign defers instead of completing when they are all waiting. A mailbox
+// that is NOT coming back on its own is deliberately absent from `paced`: the
+// lead is offered, and the placer moves it to another mailbox.
+//
+// Returns a nil pair and a nil next-due when the campaign is genuinely complete.
+func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, campaignID uuid.UUID, orderBy, orderDir, orderField string, prioritizeNewLeads, excludeNewLeads bool, paced PacedSenders) (*ContactSequencePair, *time.Time, bool, error) {
 	router, err := r.loadRouter(ctx, campaignID)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
 	}
 	if router == nil {
-		return nil, nil, nil
+		return nil, nil, false, nil
 	}

 	// 2. Ordered candidate contacts + their last-sent step (with engagement) + sent set.
@@ -955,7 +1067,7 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam
 	}

 	query := `
-		SELECT cl.contact_id, cl.added_at,
+		SELECT cl.contact_id, cl.added_at, cl.email_account_id,
 		       lp.sequence_id, lp.sent_at, lp.opened_at, lp.clicked_at, lp.replied_at, COALESCE(lp.reply_class, ''), COALESCE(lp.ai_label, ''),
 		       COALESCE(ss.ids, '{}') AS sent_ids,
 		       EXISTS (
@@ -1006,23 +1118,28 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam

 	rows, err := r.db.Query(ctx, query, campaignID, config.CampaignSendMaxAttempts)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, false, err
 	}
 	defer rows.Close()
 	// nextDue is the soonest moment anything becomes sendable: a step whose
 	// wait elapses or a condition window that closes. Only reported when no
 	// contact is due right now.
 	var nextDue *time.Time
-	noteDue := func(at time.Time) {
+	// waitingOnSender is true while the soonest moment belongs to a lead
+	// waiting for its own mailbox, so the caller can say "waiting for its
+	// mailbox" rather than "waiting for a step's delay".
+	waitingOnSender := false
+	noteDue := func(at time.Time, onSender bool) {
 		if nextDue == nil || at.Before(*nextDue) {
 			nextDue = &at
+			waitingOnSender = onSender
 		}
 	}
 	for rows.Next() {
 		var in routeInput
 		var contactID uuid.UUID
-		if serr := rows.Scan(&contactID, &in.addedAt, &in.lastSeq, &in.sentAt, &in.openedAt, &in.clickedAt, &in.repliedAt, &in.replyClass, &in.aiLabel, &in.sentIDs, &in.hasReplied); serr != nil {
-			return nil, nil, serr
+		if serr := rows.Scan(&contactID, &in.addedAt, &in.sender, &in.lastSeq, &in.sentAt, &in.openedAt, &in.clickedAt, &in.repliedAt, &in.replyClass, &in.aiLabel, &in.sentIDs, &in.hasReplied); serr != nil {
+			return nil, nil, false, serr
 		}
 		if in.lastSeq == nil && excludeNewLeads {
 			continue
@@ -1031,7 +1148,7 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam
 		if res.WaitUntil != nil {
 			// Not decidable yet — remember the soonest window so the scheduler
 			// can re-check exactly then instead of guessing or completing.
-			noteDue(*res.WaitUntil)
+			noteDue(*res.WaitUntil, false)
 			continue
 		}
 		if res.Target == nil {
@@ -1040,17 +1157,26 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam
 		// When is this step due? Skip it (but remember when) if not yet: the
 		// contacts behind this one may be sendable right now.
 		if res.DueAt != nil && res.DueAt.After(router.dueBy) {
-			noteDue(*res.DueAt)
+			noteDue(*res.DueAt, false)
 			continue
 		}
-		return &ContactSequencePair{ContactID: contactID, SequenceID: *res.Target, IsNewLead: res.IsNewLead, NotBefore: res.DueAt}, nil, nil
+		// Due, but the mailbox this lead's conversation belongs to has nothing
+		// left today. Wait for it — the address a contact has already heard
+		// from is worth more than sending the follow-up a few hours earlier
+		// from a stranger. Only an email step waits: an action or wait node
+		// sends nothing, so no mailbox has to be free for it.
+		if back, ok := paced.reopening(in.sender); ok && router.isEmailStep(*res.Target) {
+			noteDue(back, true)
+			continue
+		}
+		return &ContactSequencePair{ContactID: contactID, SequenceID: *res.Target, IsNewLead: res.IsNewLead, NotBefore: res.DueAt, AssignedSender: in.sender}, nil, false, nil
 	}
 	if rerr := rows.Err(); rerr != nil {
-		return nil, nil, rerr
+		return nil, nil, false, rerr
 	}
 	// Nobody sendable now. Hand back the soonest moment somebody will be so the
 	// scheduler defers until then rather than completing.
-	return nil, nextDue, nil
+	return nil, nextDue, waitingOnSender, nil
 }

 // ContactRoute is where a contact's flow goes next inside one campaign.
@@ -1072,6 +1198,9 @@ type ContactRoute struct {
 	Excluded string
 	// LastSentStep is the step the contact is on now (nil before any send).
 	LastSentStep *uuid.UUID
+	// AssignedSender is the mailbox this lead's sequence is bound to, nil
+	// until its first email was reserved.
+	AssignedSender *uuid.UUID
 }

 // RouteContact runs the campaign's routing for ONE contact and reports where
@@ -1084,7 +1213,7 @@ func (r *campaignProgressRepository) RouteContact(ctx context.Context, campaignI
 		return nil, err
 	}
 	query := `
-		SELECT cl.added_at,
+		SELECT cl.added_at, cl.email_account_id,
 		       lp.sequence_id, lp.sent_at, lp.opened_at, lp.clicked_at, lp.replied_at, COALESCE(lp.reply_class, ''), COALESCE(lp.ai_label, ''),
 		       COALESCE(ss.ids, '{}') AS sent_ids,
 		       EXISTS (
@@ -1125,7 +1254,7 @@ func (r *campaignProgressRepository) RouteContact(ctx context.Context, campaignI
 	var in routeInput
 	var bounced, failed, suppressed, undeliverable bool
 	err = r.db.QueryRow(ctx, query, campaignID, config.CampaignSendMaxAttempts, contactID).Scan(
-		&in.addedAt, &in.lastSeq, &in.sentAt, &in.openedAt, &in.clickedAt, &in.repliedAt, &in.replyClass, &in.aiLabel, &in.sentIDs, &in.hasReplied,
+		&in.addedAt, &in.sender, &in.lastSeq, &in.sentAt, &in.openedAt, &in.clickedAt, &in.repliedAt, &in.replyClass, &in.aiLabel, &in.sentIDs, &in.hasReplied,
 		&bounced, &failed, &suppressed, &undeliverable,
 	)
 	if errors.Is(err, pgx.ErrNoRows) {
@@ -1134,7 +1263,7 @@ func (r *campaignProgressRepository) RouteContact(ctx context.Context, campaignI
 	if err != nil {
 		return nil, err
 	}
-	out := &ContactRoute{LastSentStep: in.lastSeq}
+	out := &ContactRoute{LastSentStep: in.lastSeq, AssignedSender: in.sender}
 	switch {
 	case bounced:
 		out.Excluded = "bounced"
@@ -1159,12 +1288,34 @@ type routeInput struct {
 	sentAt, openedAt, clickedAt, repliedAt *time.Time
 	// addedAt is when the contact entered the campaign; the anchor the entry
 	// delay counts from. Nil for a lead added before the column existed.
-	addedAt             *time.Time
+	addedAt *time.Time
+	// sender is the mailbox this lead's sequence is bound to, nil until its
+	// first email was reserved.
+	sender              *uuid.UUID
 	replyClass, aiLabel string
 	sentIDs             []uuid.UUID
 	hasReplied          bool
 }

+// PacedSenders maps a mailbox that cannot take a send right now, but will be
+// able to on its own (its daily budget resets, its sending hours reopen), to
+// when it is expected back. It is deliberately NOT the set of every unusable
+// mailbox: one that is disconnected, failing authentication, resting or held
+// by warmup health is absent, so the leads bound to it are routed and moved
+// off it rather than left waiting for a return that is not coming.
+type PacedSenders map[uuid.UUID]time.Time
+
+// reopening reports when a lead's bound mailbox is expected back, and whether
+// it is paced at all. An unassigned lead is never paced: rotation is still
+// free to pick for it.
+func (p PacedSenders) reopening(sender *uuid.UUID) (time.Time, bool) {
+	if p == nil || sender == nil {
+		return time.Time{}, false
+	}
+	at, ok := p[*sender]
+	return at, ok
+}
+
 // campaignRouter is a campaign's flow graph loaded once per pass, with the
 // routing rules that decide where a contact goes out of their current step.
 type campaignRouter struct {
@@ -1187,6 +1338,7 @@ type campaignRouter struct {

 type routeStep struct {
 	id          uuid.UUID
+	kind        string
 	bc          models.BranchConditions
 	waitAfter   int
 	waitMinutes int // a "wait" node's own delay, gating the step after it
@@ -1212,15 +1364,14 @@ func (r *campaignProgressRepository) loadRouter(ctx context.Context, campaignID
 	for srows.Next() {
 		var si routeStep
 		var raw, action []byte
-		var kind string
-		if serr := srows.Scan(&si.id, &raw, &si.waitAfter, &kind, &action); serr != nil {
+		if serr := srows.Scan(&si.id, &raw, &si.waitAfter, &si.kind, &action); serr != nil {
 			srows.Close()
 			return nil, serr
 		}
 		if len(raw) > 0 {
 			_ = json.Unmarshal(raw, &si.bc)
 		}
-		if kind != "email" && len(action) > 0 {
+		if si.kind != "email" && len(action) > 0 {
 			var cfg models.ActionConfig
 			if json.Unmarshal(action, &cfg) == nil && cfg.Type == "wait" && cfg.WaitMinutes != nil && *cfg.WaitMinutes > 0 {
 				si.waitMinutes = *cfg.WaitMinutes
@@ -1456,6 +1607,13 @@ func (cr *campaignRouter) route(campaignID, contactID uuid.UUID, in routeInput)
 	return out
 }

+// isEmailStep reports whether a step sends mail. Only those need a mailbox, so
+// only those wait for one.
+func (cr *campaignRouter) isEmailStep(id uuid.UUID) bool {
+	idx, ok := cr.idxByID[id]
+	return ok && cr.steps[idx].kind == "email"
+}
+
 // enteredAt is when a lead entered the campaign. campaign_leads.added_at is
 // nullable and deliberately not backfilled, so a lead that predates the column
 // falls back to the campaign's own creation time — which keeps turning an entry
@@ -1376,6 +1376,13 @@ func (r *contactRepository) Search(
 				-- 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'),
+				-- The mailbox this lead's whole sequence sends from, fixed when
+				-- its first email went out. Null until then.
+				'sender', (
+					SELECT ea.email FROM campaign_leads cls
+					JOIN email_accounts ea ON ea.id = cls.email_account_id
+					WHERE cls.campaign_id = %[1]s AND cls.contact_id = c.id
+				),
 				-- The step the contact is on now = the latest step actually sent.
 				-- Labelled the same way the canvas does: custom name, else
 				-- "Email N" (Nth email-kind step by position), else action label.
@@ -1536,6 +1543,7 @@ func (r *contactRepository) Search(
 				TotalSteps int        `json:"total_steps"`
 				LastAt     *time.Time `json:"last_at"`
 				Step       *string    `json:"step"`
+				Sender     *string    `json:"sender"`

 				Undeliverable bool `json:"undeliverable"`
 			}
@@ -1572,8 +1580,13 @@ func (r *contactRepository) Search(
 			if status == models.LeadStatusFailed && lp.FailReason != nil {
 				failureReason = *lp.FailReason
 			}
+			sender := ""
+			if lp.Sender != nil {
+				sender = *lp.Sender
+			}
 			c.CampaignLead = &models.ContactCampaignProgress{
 				Status:         status,
+				Sender:         sender,
 				Sent:           lp.Sent,
 				Opened:         lp.Opened,
 				MachineOpened:  lp.MachineOpn,
@@ -19,10 +19,12 @@ import (
 // the service.
 func (r *contactRepository) ListCampaignStates(ctx context.Context, orgID, contactID uuid.UUID) ([]models.ContactCampaignState, *errx.Error) {
 	campQuery := `
-		SELECT cam.id, cam.name, cam.status, c.subscribed, ` + undeliverableClause("cam.id") + `
+		SELECT cam.id, cam.name, cam.status, c.subscribed, ` + undeliverableClause("cam.id") + `,
+		       cl.email_account_id, COALESCE(sender.email, '')
 		FROM campaign_leads cl
 		JOIN campaigns cam ON cam.id = cl.campaign_id AND cam.organization_id = $2
 		JOIN contacts c ON c.id = cl.contact_id AND c.organization_id = $2
+		LEFT JOIN email_accounts sender ON sender.id = cl.email_account_id
 		WHERE cl.contact_id = $1
 		ORDER BY cam.created_at DESC
 	`
@@ -39,7 +41,8 @@ func (r *contactRepository) ListCampaignStates(ctx context.Context, orgID, conta
 	var camps []campRow
 	for rows.Next() {
 		var cr campRow
-		if err := rows.Scan(&cr.state.CampaignID, &cr.state.CampaignName, &cr.state.CampaignStatus, &cr.subscribed, &cr.undeliverable); err != nil {
+		if err := rows.Scan(&cr.state.CampaignID, &cr.state.CampaignName, &cr.state.CampaignStatus, &cr.subscribed, &cr.undeliverable,
+			&cr.state.SenderID, &cr.state.SenderEmail); err != nil {
 			rows.Close()
 			db.CaptureError(err, "", nil, "ListCampaignStates campaigns scan")
 			return nil, errx.InternalError()
@@ -48,6 +48,14 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 		return time.Time{}, nil, uuid.Nil, ErrNoEmailAccounts
 	}

+	// STEP 2.5: What the pool can do this pass, resolved once. `paced` is the
+	// mailboxes that are simply busy right now (today's budget spent, hours
+	// closed); routing holds back the leads already bound to one so the rest of
+	// the campaign keeps sending, instead of parking every lead behind the
+	// first one whose mailbox is full.
+	pass := s.newCampaignPass(ctx, campaign, accounts)
+	paced := s.pacedSenders(ctx, pass, accounts)
+
 	// STEP 3: Get campaign progress - find next contact/sequence to send.
 	// Honor the new-lead-per-day cap and the prioritize-new-leads ordering.
 	orderField := ""
@@ -64,7 +72,7 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 			excludeNewLeads = true
 		}
 	}
-	nextPair, recheckAt, err := s.campaignProgressRepo.FindNextRoutedPair(
+	nextPair, recheckAt, senderWait, err := s.campaignProgressRepo.FindNextRoutedPair(
 		ctx,
 		campaignID,
 		campaign.ContactOrderBy,
@@ -72,6 +80,7 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 		orderField,
 		campaign.PrioritizeNewLeads,
 		excludeNewLeads,
+		paced,
 	)
 	if err != nil {
 		return time.Time{}, nil, uuid.Nil, err
@@ -83,9 +92,9 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 		// without it. In that case defer to the next day so follow-ups keep
 		// progressing and new leads resume tomorrow — do NOT complete.
 		if excludeNewLeads {
-			again, againDue, aerr := s.campaignProgressRepo.FindNextRoutedPair(
+			again, againDue, againSenderWait, aerr := s.campaignProgressRepo.FindNextRoutedPair(
 				ctx, campaignID, campaign.ContactOrderBy, campaign.ContactOrderDir, orderField,
-				campaign.PrioritizeNewLeads, false,
+				campaign.PrioritizeNewLeads, false, paced,
 			)
 			switch {
 			case aerr != nil:
@@ -110,7 +119,7 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 				// re-check time to `recheckAt`. Take the unexcluded pass's, or a
 				// campaign whose only remaining leads are delayed ones completes
 				// while they are still waiting to be sent.
-				recheckAt = againDue
+				recheckAt, senderWait = againDue, againSenderWait
 			}
 		}
 		// Nothing is due yet: every remaining contact is inside a step's wait
@@ -123,7 +132,13 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 			// days would otherwise write two hundred identical activity lines.
 			message := "No step is due yet; re-checking when the next wait elapses"
 			metadata := map[string]interface{}{"recheck_at": recheckAt.UTC().Format(time.RFC3339)}
-			if campaign.EntryDelayMinutes > 0 {
+			if senderWait {
+				// The steps ARE due; their mailboxes are not free. Say that,
+				// rather than blaming a wait nobody configured.
+				message = "Every lead that is due is waiting for its own mailbox: each contact keeps the address they first heard from, and those mailboxes have nothing left for now"
+				metadata["waiting_on_sender"] = true
+			}
+			if !senderWait && campaign.EntryDelayMinutes > 0 {
 				message += fmt.Sprintf(" (this campaign holds the first email for %s after a contact enters it)",
 					humanizeMinutes(campaign.EntryDelayMinutes))
 				metadata["entry_delay_minutes"] = campaign.EntryDelayMinutes
@@ -141,7 +156,7 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 	// no branches. A step is sent only if the flow reaches it; STOP/end and
 	// already-sent loops drop the contact in the finder. Conditions are evaluated
 	// at schedule time (a known, accepted race vs. last-moment engagement).
-	return s.placeCampaignSend(ctx, campaign, accounts, senderMetaByID, nextPair, false)
+	return s.placeCampaignSend(ctx, campaign, accounts, senderMetaByID, nextPair, pass, false)
 }

 // humanizeMinutes renders a delay as the largest whole unit it divides into
@@ -196,8 +211,11 @@ func (s *schedulerService) campaignSenders(ctx context.Context, campaign *models
 // slot, exactly as the send path uses it. preview makes it read-only (no
 // decision logs, no cached writes) so the contact drawer can ask "when would
 // this step go" through the same rules the scheduler applies.
-func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *models.Campaign, accounts []models.Email, senderMetaByID map[uuid.UUID]senderMeta, nextPair *repository.ContactSequencePair, preview bool) (time.Time, *repository.ContactSequencePair, uuid.UUID, error) {
+func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *models.Campaign, accounts []models.Email, senderMetaByID map[uuid.UUID]senderMeta, nextPair *repository.ContactSequencePair, pass *campaignPass, preview bool) (time.Time, *repository.ContactSequencePair, uuid.UUID, error) {
 	campaignID := campaign.ID
+	if pass == nil {
+		pass = s.newCampaignPass(ctx, campaign, accounts)
+	}
 	logDecision := func(eventType, message string, metadata map[string]interface{}) {
 		if !preview {
 			s.logCampaignDecision(ctx, campaignID, eventType, message, metadata)
@@ -288,51 +306,15 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 	// intervals per day).
 	candidateTime = nextScheduleSlot(candidateTime, windows, campaignTZ)

-	// effectiveCap is the per-mailbox cold cap for THIS campaign, after the ramp
-	// clamp. It is min(per-mailbox cold cap, campaign daily limit) further min()'d
-	// with the day's ramp ceiling. Applied via min() only — it can never RAISE a
-	// mailbox above its cold cap (the mailbox-first safety invariant).
-	// Graduation state for the whole pool in one round trip, so the per-mailbox
-	// ceiling below costs no query inside the candidate loop.
-	coldRamp := s.coldRampStates(ctx, accounts)
-
-	// The organization's own posture, resolved once. A restricted organization
-	// keeps sending at a fraction of the volume; a suspended one does not send
-	// at all, and is stopped here because campaign sends never pass through the
-	// manual send gate.
-	riskState := s.orgRiskState(ctx, campaign.OrganizationID)
-	if riskState.BlocksSending() {
+	// The organization's own posture. A restricted organization keeps sending at
+	// a fraction of the volume (folded into pass.effectiveCap); a suspended one
+	// does not send at all, and is stopped here because campaign sends never
+	// pass through the manual send gate.
+	if pass.risk.BlocksSending() {
 		logDecision("org_suspended",
 			"Sending is paused for this workspace while it is under review", nil)
 		return s.deferToNextDay(campaign), nil, accounts[0].ID, ErrCampaignDeferred
 	}
-	riskMultiplier := riskState.CapMultiplier()
-
-	// Which mailboxes are in cold rotation at all. A resting mailbox keeps its
-	// warmup traffic and its reputation; it just is not offered cold sends.
-	lifecycles, lifecyclesKnown := s.sendLifecycles(ctx, accounts)
-
-	effectiveCap := func(acct models.Email) int {
-		lim := min(acct.CampaignLimit, campaign.DailyLimit)
-		if campaign.RampEnabled {
-			lim = min(lim, campaignRampCeiling(true, campaign.RampStart, campaign.RampIncrement, campaign.RampCeiling, campaign.RampLevel))
-		}
-		// Graduation ceiling: a mailbox at its warmup ceiling must not reach the
-		// full cold cap the day it joins a campaign. min() only, so it can lower
-		// a mailbox but never raise one.
-		lim = min(lim, coldCeilingFor(coldRamp[acct.ID], lim))
-		if riskMultiplier < 1 {
-			risked := int(float64(lim)*riskMultiplier + 0.5)
-			// A restricted organization still sends, just far less. Zeroing it
-			// here would stop the campaign without ever saying why; suspension
-			// is the band that stops sending, and it does so at the send gate.
-			if risked < 1 {
-				risked = 1
-			}
-			lim = min(lim, risked)
-		}
-		return lim
-	}

 	// providerMatches reports whether a mailbox's provider satisfies the
 	// recipient ESP under the current match mode. An unknown recipient provider
@@ -351,15 +333,11 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 		return acctProvider == recipientProvider
 	}

-	// STEP 8: Build weighted account candidates, gating each mailbox on when it
-	// may next send in its OWN timezone: its rolled workday when it has a
-	// sending-behaviour profile, and the historical 8am-8pm business-hours band
-	// when it does not.
+	// STEP 8: Build weighted account candidates. Each mailbox goes through the
+	// same two-part gate routing's availability pre-pass used: what it can do at
+	// all (authentication, cold rotation, warmup health, today's budget), then
+	// when its own calendar next lets it send.
 	//
-	// Profiles are resolved for the whole pool in one query, so adding
-	// mailboxes to a campaign does not add a query per mailbox to every pass.
-	behaviors := s.behaviorForAll(ctx, accounts)
-
 	// Rotation fallback for mailboxes with no campaign_senders row (tag-resolved
 	// pools and the "all active mailboxes" default). One query for the whole
 	// pool; a failure just leaves the map empty and rotation degrades to its
@@ -377,125 +355,89 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 		}
 	}

-	// The sending-domain authentication gate, resolved once for the whole pass.
-	// authGated counts mailboxes dropped by it so an empty candidate set can be
-	// reported as the DNS problem it is rather than as a scheduling one.
-	enforceAuth, authGrace := s.domainAuthGate(ctx)
+	// Why mailboxes were left out. A reason that clears on its own (the budget
+	// resets at midnight, the mailbox's hours reopen, a health hold expires)
+	// makes an empty pool a deferral; only one that never clears makes it a
+	// pause. authGated is counted apart so an empty pool can be reported as the
+	// DNS problem it is rather than as a scheduling one, and reopensAt is the
+	// earliest reopening among hours-closed mailboxes.
+	//
+	// gates keeps the same answer per mailbox, so the lead's own mailbox can be
+	// told apart from the rest: busy today (wait for it) or not sending for this
+	// campaign at all (move the lead off it).
 	authGated := 0
 	lifecycleGated := 0
-
-	// Why the other mailboxes were left out. A reason that clears on its own
-	// (the budget resets at midnight, the mailbox's hours reopen, a health hold
-	// expires) makes an empty pool a deferral; only one that never clears makes
-	// it a pause. reopensAt is the earliest reopening among hours-closed
-	// mailboxes.
 	budgetSpent := 0
 	hoursClosed := 0
 	healthHeld := 0
 	var reopensAt time.Time
+	gates := map[uuid.UUID]mailboxGate{}
+
+	// The mailbox this lead's conversation belongs to, when it is still one this
+	// campaign can send from. Its gates are the same as everyone else's; what
+	// differs is that a closed hour moves the send rather than dropping the
+	// mailbox, because there is no second address to fall back to.
+	bound := boundSender(accounts, nextPair.AssignedSender)
+
+	// A lead can have steps but no recorded binding: it was removed from the
+	// campaign and added back, which keeps its progress and starts a new lead
+	// row. The address it last actually heard from is still the one to keep, so
+	// it is PREFERRED — used when that mailbox is free, and given up when it is
+	// not. It cannot be waited for the way a recorded binding is: routing does
+	// not know about it, so a lead waiting on one would hold up every lead
+	// behind it. The next send records it, and from then on it is a rule.
+	prefer := bound
+	if bound == nil && nextPair.AssignedSender == nil && !nextPair.IsNewLead {
+		if last, lerr := s.campaignProgressRepo.LastSenderForLead(ctx, campaignID, nextPair.ContactID); lerr == nil {
+			prefer = boundSender(accounts, last)
+		}
+	}

 	var candidates []AccountCandidate
 	for _, acct := range accounts {
-		// Sending-domain authentication. This runs before the daily-count query
-		// so a gated mailbox costs nothing, and before every other gate because
-		// unauthenticated mail is rejected outright by Gmail/Yahoo/Outlook: no
-		// amount of budget, window, or rotation makes it deliverable.
-		//
-		// Only a SUSTAINED failure gates. "unknown" (never checked, or DNS
-		// could not answer) and a failure inside the grace window both pass
-		// through, so a resolver hiccup can never stop a campaign.
-		if enforceAuth && acct.DomainAuthBlocked(time.Now(), authGrace) {
-			authGated++
-			continue
-		}
-
-		// Not in cold rotation. Checked here, beside the authentication gate,
-		// so a resting mailbox costs no capacity query. Applied only when the
-		// states were actually read.
-		if lifecyclesKnown && !lifecycles[acct.ID].State.SendsCold() {
-			lifecycleGated++
-			continue
-		}
-
-		sentToday, err := s.taskRepo.CountCampaignEmailsSentToday(ctx, acct.ID)
+		sentToday, err := s.sentTodayFor(ctx, pass, acct.ID)
 		if err != nil {
 			return time.Time{}, nil, uuid.Nil, err
 		}

-		acctLimit := effectiveCap(acct)
-		remaining := acctLimit - sentToday
-
-		// Skip accounts that have reached their daily limit
-		if remaining <= 0 {
-			budgetSpent++
+		acctLimit := pass.effectiveCap(acct)
+		// Authentication, cold rotation, warmup health and today's budget, in
+		// the order the gate applies them. Same answer routing's availability
+		// pre-pass got, from the same reads.
+		gate, remaining := s.gateFor(ctx, pass, acct, acctLimit-sentToday)
+		if !gate.open() {
+			gates[acct.ID] = gate
+			switch gate.reason {
+			case gateAuth:
+				authGated++
+			case gateResting:
+				lifecycleGated++
+			case gateHealth:
+				healthHeld++
+			case gateBudget:
+				budgetSpent++
+			}
 			continue
 		}

-		bhv := behaviors[acct.ID]
-
-		// Health-gate cold sends on the SAME warmup health state used for pool
-		// selection, so a mailbox in deliverability trouble doesn't keep blasting
-		// cold volume (the concentration risk the safety policy warns about):
-		//   - quarantined/blocked (still within blocked_until) → don't send at all
-		//   - watch/throttled → dampen today's budget by the same multiplier warmup
-		//     uses for that band (watch 0.7x, throttled 0.5x), via adjustmentFor so
-		//     the warmup and cold schedulers can't drift; the wider min-gap still applies
-		// This gate runs FIRST, before any rotation/ESP logic, so a degraded
-		// mailbox is always dropped regardless of weighting.
-		if state, blockedUntil, herr := s.warmupRepo.GetHealthState(ctx, acct.ID); herr == nil {
-			switch state {
-			case models.WarmupHealthQuarantined, models.WarmupHealthBlocked:
-				if blockedUntil == nil || blockedUntil.After(time.Now()) {
-					healthHeld++
-					continue
-				}
-			case models.WarmupHealthWatch, models.WarmupHealthThrottled:
-				remaining = int(float64(remaining) * adjustmentFor(state).volumeMultiplier)
-				if remaining <= 0 {
-					budgetSpent++
-					continue
-				}
-			}
-		}
-
-		// Where the mailbox may next send, in its OWN timezone. With a
-		// behaviour profile this is its rolled workday (start, lunch, end,
-		// working weekdays) with the hourly ceiling already applied; without
-		// one it falls back to the historical 8am-8pm business-hours gate.
-		var behaviorOpenAt *time.Time
-		if bhv.Enabled {
-			openAt, ok := s.placeWithinBehavior(ctx, bhv, candidateTime)
-			if !ok {
-				continue // profile has no working days at all
-			}
-			behaviorOpenAt = &openAt
-
-			// Budget the send against the day it will actually land on. When
-			// today is spent, placeWithinBehavior has already walked to a later
-			// day, and that day starts with the mailbox's full cold cap.
-			if !sameLocalDay(openAt, time.Now(), bhv.Loc) {
-				remaining = acctLimit
-			}
-			// The day's rolled cold budget, folded in by min(): a persona can
-			// only lower a mailbox's remaining sends, never lift it above the
-			// cold cap.
-			remaining = s.behaviorDailyCap(ctx, bhv, remaining, openAt)
-			if remaining <= 0 {
+		// Where the mailbox may next send, on its OWN calendar. The lead's own
+		// mailbox is allowed to WAIT for its next opening rather than drop out
+		// of the pass: there is no second address this conversation could come
+		// from, so a closed hour moves the send instead of refusing it.
+		openAt, openLoc, remaining, wgate := s.windowFor(
+			ctx, pass, acct, candidateTime, acctLimit, remaining, bound != nil && bound.ID == acct.ID)
+		if !wgate.open() {
+			gates[acct.ID] = wgate
+			switch wgate.reason {
+			case gateBudget:
 				budgetSpent++
-				continue
-			}
-		} else if acct.Timezone != "" && acct.Timezone != campaign.Timezone {
-			acctTZ := loadLocation(acct.Timezone)
-			acctLocal := candidateTime.In(acctTZ)
-			acctHour := acctLocal.Hour()
-			if acctHour < 8 || acctHour >= 20 {
-				// Outside the account's business hours.
+			case gateHours:
 				hoursClosed++
-				if open := businessHoursReopen(candidateTime, acctTZ); reopensAt.IsZero() || open.Before(reopensAt) {
-					reopensAt = open
+				if reopensAt.IsZero() || wgate.reopensAt.Before(reopensAt) {
+					reopensAt = wgate.reopensAt
 				}
-				continue
 			}
+			continue
 		}

 		warmupAgeDays := 0
@@ -509,8 +451,9 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 			WarmupAgeDays:  warmupAgeDays,
 			Weight:         computeWeight(remaining, warmupAgeDays),
 			ProviderMatch:  providerMatches(acct.Provider),
-			Behavior:       bhv,
-			BehaviorOpenAt: behaviorOpenAt,
+			Behavior:       pass.behaviors[acct.ID],
+			OpenAt:         openAt,
+			OpenLoc:        openLoc,
 		}
 		if meta, ok := senderMetaByID[acct.ID]; ok {
 			cand.HasSenderMetadata = true
@@ -606,12 +549,22 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 		return time.Time{}, nil, uuid.Nil, ErrNoEligibleMailbox
 	}

+	// STEP 8.2: The lead's own mailbox, if this pass left it able to send. Every
+	// step of one conversation leaves from the address the contact first heard
+	// from, so once this is set there is nothing left to choose: no rotation, no
+	// ESP matching, no weighting. Those decide which mailbox STARTS a lead.
+	boundCand := pickBound(candidates, prefer)
+
 	// STEP 8.25: Apply ESP matching to the under-budget candidate set.
 	//   strict → only matching mailboxes are eligible; if none, DEFER (never
 	//            send cross-provider).
 	//   prefer → restrict to matching mailboxes when at least one has capacity,
 	//            otherwise fall back to the full eligible set (never starves).
-	if campaign.ESPMatchMode != "off" && recipientProvider != "" {
+	//
+	// Skipped for a lead that already has its mailbox: matching picks a sender
+	// for a first email, and re-applying it to a follow-up could only refuse
+	// the one address this contact is allowed to hear from.
+	if boundCand == nil && campaign.ESPMatchMode != "off" && recipientProvider != "" {
 		matching := make([]AccountCandidate, 0, len(candidates))
 		for _, c := range candidates {
 			if c.ProviderMatch {
@@ -638,12 +591,35 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 		}
 	}

-	// STEP 8.5: Select best account per the campaign's rotation mode. pool is
-	// the set selection actually ran over, kept for the pacing maths below.
-	// Every candidate has budget left today, so it has weight and the selector
-	// always picks one; the guard only keeps a nil from being dereferenced.
+	// STEP 8.5: Select the mailbox. A lead already bound to one keeps it; only a
+	// lead starting its sequence goes through the campaign's rotation mode. pool
+	// is the set selection ran over, kept for the pacing maths below. Every
+	// candidate has budget left today, so it has weight and the selector always
+	// picks one; the guard only keeps a nil from being dereferenced.
 	pool := candidates
-	selected := selectAccountByRotationMode(campaign.RotationMode, candidates)
+	selected := boundCand
+	if selected == nil {
+		// The lead is bound to a mailbox that did not survive this pass. Waiting
+		// is right only while the mailbox is coming back: a spent budget resets
+		// at midnight, closed hours reopen. A mailbox that is disconnected,
+		// failing authentication, resting or held by warmup health is not going
+		// to write to this contact again soon, and a week of silence mid-sequence
+		// is worse than a change of address, so the lead moves.
+		if bound != nil {
+			gate := gates[bound.ID]
+			if gate.paced {
+				logDecisionOnce("sender_busy",
+					"Some leads are waiting for their own mailbox: every contact keeps the address they first heard from, and that mailbox has nothing left for now",
+					map[string]interface{}{"mailbox": bound.Email, "reason": gate.reason})
+				resume := gate.reopensAt
+				if resume.IsZero() {
+					resume = s.deferToNextDay(campaign)
+				}
+				return resume, nil, bound.ID, ErrSenderBusy
+			}
+		}
+		selected = selectAccountByRotationMode(campaign.RotationMode, candidates)
+	}
 	if selected == nil {
 		return time.Time{}, nil, uuid.Nil, ErrNoEligibleMailbox
 	}
@@ -658,8 +634,8 @@ func (s *schedulerService) placeCampaignSend(ctx context.Context, campaign *mode
 	// Move the candidate onto the mailbox's own workday before the spacing
 	// maths below runs, so distribution is computed against the window the send
 	// will actually land in.
-	if selected.BehaviorOpenAt != nil && selected.BehaviorOpenAt.After(candidateTime) {
-		candidateTime = *selected.BehaviorOpenAt
+	if selected.OpenAt != nil && selected.OpenAt.After(candidateTime) {
+		candidateTime = *selected.OpenAt
 	}

 	// hardFloor is the earliest moment this send is ALLOWED: wait_after,
@@ -0,0 +1,314 @@
+package scheduler
+
+import (
+	"context"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/app/behavior"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// A lead's sending mailbox is fixed for its whole sequence: rotation picks the
+// mailbox for the first email and every follow-up leaves from the same address
+// (issue #401). Three emails from three addresses is not a conversation, and
+// the reply lands in a mailbox the contact was never written to from.
+//
+// That only works if the two passes agree on what "this mailbox can send" means:
+// routing skips a lead whose mailbox is busy, and placement then has to be
+// willing to send from it. Both read gateFor and windowFor, over facts this
+// file collects once per pass, so a lead can never be told to wait for a
+// mailbox placement would have refused, nor moved off one it would have used.
+
+// mailboxGate is why a mailbox cannot take a cold send right now.
+type mailboxGate struct {
+	// reason is "" when the mailbox can send. Otherwise it names the gate, in
+	// the vocabulary the placer's empty-pool activity lines already count.
+	reason string
+	// paced is true when the gate lifts on its own and soon: today's budget
+	// resets at midnight, the mailbox's hours reopen in the morning. A lead
+	// bound to a paced mailbox waits for it.
+	//
+	// A gate that is NOT paced (disconnected, failing domain authentication,
+	// resting, held by warmup health) is a mailbox that is not going to serve
+	// this campaign in the near future, so its leads are moved to another one
+	// rather than left silent for a week.
+	paced bool
+	// reopensAt is when a paced mailbox is expected back; zero when unknown,
+	// which the caller reads as "tomorrow".
+	reopensAt time.Time
+}
+
+func (g mailboxGate) open() bool { return g.reason == "" }
+
+// Gate reasons. They double as the metadata the empty-pool activity-log lines
+// count, so keep them stable.
+const (
+	gateAuth      = "domain_auth"
+	gateResting   = "resting"
+	gateHealth    = "health"
+	gateBudget    = "budget"
+	gateHours     = "hours"
+	gateNoWorkday = "no_working_day"
+)
+
+// campaignPass holds everything about the mailbox pool that one scheduling pass
+// resolves once: the batch lookups, and the per-mailbox reads memoized so the
+// availability pre-pass and the placement that follows it cost one query each
+// rather than two.
+type campaignPass struct {
+	campaign        *models.Campaign
+	coldRamp        map[uuid.UUID]repository.ColdRampState
+	lifecycles      map[uuid.UUID]models.SendLifecycleState
+	lifecyclesKnown bool
+	behaviors       map[uuid.UUID]behavior.Resolved
+	enforceAuth     bool
+	authGrace       time.Duration
+	risk            models.OrgRiskState
+
+	sentToday map[uuid.UUID]int
+	health    map[uuid.UUID]healthRead
+}
+
+// healthRead is one mailbox's warmup health, as the gate reads it.
+type healthRead struct {
+	state        models.WarmupHealthState
+	blockedUntil *time.Time
+	known        bool
+}
+
+// newCampaignPass resolves the pool-wide state for one scheduling pass.
+func (s *schedulerService) newCampaignPass(ctx context.Context, campaign *models.Campaign, accounts []models.Email) *campaignPass {
+	enforceAuth, authGrace := s.domainAuthGate(ctx)
+	lifecycles, known := s.sendLifecycles(ctx, accounts)
+	return &campaignPass{
+		campaign:        campaign,
+		coldRamp:        s.coldRampStates(ctx, accounts),
+		lifecycles:      lifecycles,
+		lifecyclesKnown: known,
+		behaviors:       s.behaviorForAll(ctx, accounts),
+		enforceAuth:     enforceAuth,
+		authGrace:       authGrace,
+		risk:            s.orgRiskState(ctx, campaign.OrganizationID),
+		sentToday:       map[uuid.UUID]int{},
+		health:          map[uuid.UUID]healthRead{},
+	}
+}
+
+// effectiveCap is the per-mailbox cold cap for THIS campaign, after the ramp,
+// graduation and org-risk clamps. Applied via min() only — it can never RAISE a
+// mailbox above its own cold cap (the mailbox-first safety invariant).
+func (p *campaignPass) effectiveCap(acct models.Email) int {
+	c := p.campaign
+	lim := min(acct.CampaignLimit, c.DailyLimit)
+	if c.RampEnabled {
+		lim = min(lim, campaignRampCeiling(true, c.RampStart, c.RampIncrement, c.RampCeiling, c.RampLevel))
+	}
+	// Graduation ceiling: a mailbox at its warmup ceiling must not reach the
+	// full cold cap the day it joins a campaign.
+	lim = min(lim, coldCeilingFor(p.coldRamp[acct.ID], lim))
+	if m := p.risk.CapMultiplier(); m < 1 {
+		risked := int(float64(lim)*m + 0.5)
+		// A restricted organization still sends, just far less. Zeroing it here
+		// would stop the campaign without ever saying why; suspension is the
+		// band that stops sending, and it does so at the send gate.
+		if risked < 1 {
+			risked = 1
+		}
+		lim = min(lim, risked)
+	}
+	return lim
+}
+
+// sentTodayFor is the mailbox's campaign sends so far today, read once per pass.
+func (s *schedulerService) sentTodayFor(ctx context.Context, p *campaignPass, id uuid.UUID) (int, error) {
+	if n, ok := p.sentToday[id]; ok {
+		return n, nil
+	}
+	n, err := s.taskRepo.CountCampaignEmailsSentToday(ctx, id)
+	if err != nil {
+		return 0, err
+	}
+	p.sentToday[id] = n
+	return n, nil
+}
+
+// healthFor is the mailbox's warmup health, read once per pass. An unreadable
+// state is "unknown" and gates nothing, exactly as the inline read it replaces.
+func (s *schedulerService) healthFor(ctx context.Context, p *campaignPass, id uuid.UUID) healthRead {
+	if h, ok := p.health[id]; ok {
+		return h
+	}
+	h := healthRead{}
+	if s.warmupRepo != nil {
+		if state, until, err := s.warmupRepo.GetHealthState(ctx, id); err == nil {
+			h = healthRead{state: state, blockedUntil: until, known: true}
+		}
+	}
+	p.health[id] = h
+	return h
+}
+
+// gateFor is the ONE answer to "can this mailbox take a cold send for this
+// campaign right now", over the gates that depend on the mailbox alone and not
+// on which lead is being placed. The time-of-day gates (a mailbox's own hours,
+// a behaviour profile's workday) are deliberately not here: they move a send
+// forward rather than refusing it, and the placer applies them against the slot
+// it is actually building.
+//
+// remaining is today's budget left, already read; it is returned adjusted for a
+// health band that dampens rather than blocks.
+func (s *schedulerService) gateFor(ctx context.Context, p *campaignPass, acct models.Email, remaining int) (mailboxGate, int) {
+	// Sending-domain authentication first, because unauthenticated mail is
+	// rejected outright by Gmail/Yahoo/Outlook: no amount of budget, window or
+	// rotation makes it deliverable. Only a SUSTAINED failure gates — "unknown"
+	// and a failure inside the grace window both pass, so a resolver hiccup
+	// never stops a campaign.
+	if p.enforceAuth && acct.DomainAuthBlocked(time.Now(), p.authGrace) {
+		return mailboxGate{reason: gateAuth}, 0
+	}
+	// Not in cold rotation. A resting mailbox keeps its warmup traffic and its
+	// reputation; it just is not offered cold sends, and its leads move on.
+	if p.lifecyclesKnown && !p.lifecycles[acct.ID].State.SendsCold() {
+		return mailboxGate{reason: gateResting}, 0
+	}
+	// The SAME warmup health state pool selection uses, so a mailbox in
+	// deliverability trouble does not keep blasting cold volume:
+	//   - quarantined/blocked (still within blocked_until) → no sends at all
+	//   - watch/throttled → dampen today's budget by the band's multiplier
+	if h := s.healthFor(ctx, p, acct.ID); h.known {
+		switch h.state {
+		case models.WarmupHealthQuarantined, models.WarmupHealthBlocked:
+			if h.blockedUntil == nil || h.blockedUntil.After(time.Now()) {
+				return mailboxGate{reason: gateHealth}, 0
+			}
+		case models.WarmupHealthWatch, models.WarmupHealthThrottled:
+			remaining = int(float64(remaining) * adjustmentFor(h.state).volumeMultiplier)
+		}
+	}
+	if remaining <= 0 {
+		return mailboxGate{reason: gateBudget, paced: true}, 0
+	}
+	return mailboxGate{}, remaining
+}
+
+// windowFor places a mailbox on its OWN calendar for a send aimed at `at`, and
+// is the second half of the gate: the part that depends on the clock rather
+// than on the mailbox's standing. With a behaviour profile that is its rolled
+// workday (start, lunch, end, working weekdays, hourly ceiling and the day's
+// rolled volume); without one it is the historical 8am-8pm business-hours band.
+//
+// wait asks it to move the send to the mailbox's next opening instead of
+// refusing it. Only the lead's own mailbox gets that: for everyone else the
+// pass simply picks another mailbox, but a bound lead has no other address.
+//
+// It returns the opening instant to place the send at (nil when the mailbox is
+// free now), the timezone that instant's DAY is counted in, the day's remaining
+// budget re-based onto the day the send will land on, and the gate.
+func (s *schedulerService) windowFor(ctx context.Context, p *campaignPass, acct models.Email, at time.Time, acctLimit, remaining int, wait bool) (*time.Time, *time.Location, int, mailboxGate) {
+	bhv := p.behaviors[acct.ID]
+	if bhv.Enabled {
+		openAt, ok := s.placeWithinBehavior(ctx, bhv, at)
+		if !ok {
+			// A profile with no working days at all: nothing to wait for.
+			return nil, nil, 0, mailboxGate{reason: gateNoWorkday}
+		}
+		// Budget the send against the day it will actually land on. When today
+		// is spent, placeWithinBehavior has already walked to a later day, and
+		// that day starts with the mailbox's full cold cap.
+		if !sameLocalDay(openAt, time.Now(), bhv.Loc) {
+			remaining = acctLimit
+		}
+		// The day's rolled cold budget, folded in by min(): a persona can only
+		// lower a mailbox's remaining sends, never lift it above the cold cap.
+		remaining = s.behaviorDailyCap(ctx, bhv, remaining, openAt)
+		if remaining <= 0 {
+			return nil, nil, 0, mailboxGate{reason: gateBudget, paced: true}
+		}
+		return &openAt, bhv.Loc, remaining, mailboxGate{}
+	}
+	if acct.Timezone == "" || acct.Timezone == p.campaign.Timezone {
+		return nil, nil, remaining, mailboxGate{}
+	}
+	loc := loadLocation(acct.Timezone)
+	if h := at.In(loc).Hour(); h >= 8 && h < 20 {
+		return nil, nil, remaining, mailboxGate{}
+	}
+	open := businessHoursReopen(at, loc)
+	if !wait {
+		return nil, nil, 0, mailboxGate{reason: gateHours, paced: true, reopensAt: open}
+	}
+	if !sameLocalDay(open, time.Now(), loc) {
+		remaining = acctLimit
+	}
+	return &open, loc, remaining, mailboxGate{}
+}
+
+// pacedSenders is the availability pre-pass routing consults: the pool
+// mailboxes that cannot take a send right now but will be able to on their own.
+// A lead already bound to one waits for it instead of parking every lead behind
+// it; a lead bound to a mailbox that is NOT here is offered to the placer,
+// which moves it.
+//
+// A mailbox outside its own sending hours is paced too, so a campaign spanning
+// timezones keeps serving the leads whose mailbox is awake.
+func (s *schedulerService) pacedSenders(ctx context.Context, p *campaignPass, accounts []models.Email) repository.PacedSenders {
+	paced := repository.PacedSenders{}
+	now := time.Now()
+	for _, acct := range accounts {
+		sentToday, err := s.sentTodayFor(ctx, p, acct.ID)
+		if err != nil {
+			// Unknown budget is not a reason to strand a lead: leave the
+			// mailbox unpaced and let the placer decide with a real read.
+			continue
+		}
+		acctLimit := p.effectiveCap(acct)
+		gate, remaining := s.gateFor(ctx, p, acct, acctLimit-sentToday)
+		if gate.open() {
+			// Same clock gates the placement pass applies, asked about now.
+			_, _, _, gate = s.windowFor(ctx, p, acct, now, acctLimit, remaining, false)
+		}
+		// Held, not paced, means its leads are moved rather than left waiting,
+		// so only a paced mailbox is recorded here.
+		if gate.paced {
+			back := gate.reopensAt
+			if back.IsZero() {
+				back = s.deferToNextDay(p.campaign)
+			}
+			paced[acct.ID] = back
+		}
+	}
+	return paced
+}
+
+// boundSender finds the lead's bound mailbox in the campaign's pool. Not found
+// means the mailbox is no longer one this campaign can send from at all — it
+// was disconnected, or taken off the campaign's sending accounts — and the lead
+// is moved to another one.
+func boundSender(accounts []models.Email, id *uuid.UUID) *models.Email {
+	if id == nil {
+		return nil
+	}
+	for i := range accounts {
+		if accounts[i].ID == *id {
+			return &accounts[i]
+		}
+	}
+	return nil
+}
+
+// pickBound returns the candidate for the lead's own mailbox, or nil when it
+// did not survive this pass's gates. Rotation is not consulted: the address the
+// contact has already heard from is not a choice to re-make on every step.
+func pickBound(candidates []AccountCandidate, acct *models.Email) *AccountCandidate {
+	if acct == nil {
+		return nil
+	}
+	for i := range candidates {
+		if candidates[i].Account.ID == acct.ID && candidates[i].Weight > 0 {
+			return &candidates[i]
+		}
+	}
+	return nil
+}
@@ -27,6 +27,9 @@ const (
 	ConstraintNoMailbox        ContactSendConstraint = "no_mailbox"
 	ConstraintDomainAuth       ContactSendConstraint = "domain_auth"
 	ConstraintCampaignEnded    ContactSendConstraint = "campaign_ended"
+	// ConstraintSenderBusy is the whole pool being fine and one mailbox not:
+	// the lead's sequence belongs to a mailbox that has nothing left today.
+	ConstraintSenderBusy ContactSendConstraint = "sender_busy"
 )

 // ContactSendPreview is a read-only "what happens next" for one contact in
@@ -85,8 +88,11 @@ func (s *schedulerService) PreviewContactSend(ctx context.Context, campaignID, c
 		return pv, nil
 	}

-	pair := &repository.ContactSequencePair{ContactID: contactID, SequenceID: *route.Target, IsNewLead: route.IsNewLead, NotBefore: route.DueAt}
-	at, sendable, _, perr := s.placeCampaignSend(ctx, campaign, accounts, meta, pair, true)
+	pair := &repository.ContactSequencePair{
+		ContactID: contactID, SequenceID: *route.Target, IsNewLead: route.IsNewLead,
+		NotBefore: route.DueAt, AssignedSender: route.AssignedSender,
+	}
+	at, sendable, _, perr := s.placeCampaignSend(ctx, campaign, accounts, meta, pair, nil, true)
 	switch {
 	case perr == nil && sendable != nil:
 		pv.State = models.NextActionDue
@@ -101,6 +107,11 @@ func (s *schedulerService) PreviewContactSend(ctx context.Context, campaignID, c
 		}
 		pv.NotBefore = &slot
 		pv.Constraint = s.deferralConstraint(ctx, campaign, route)
+		// The pool is not the gate here, one mailbox is: say which, rather than
+		// blaming "no mailbox can take it" while two others sit idle.
+		if errors.Is(perr, ErrSenderBusy) && pv.Constraint == ConstraintCapacity {
+			pv.Constraint = ConstraintSenderBusy
+		}
 		return pv, nil
 	case errors.Is(perr, ErrCampaignEnded):
 		pv.State = models.NextActionBlocked
@@ -60,6 +60,16 @@ var (
 	// contact; the returned accountID is a nominal pool mailbox for the wakeup
 	// task only (the next invocation re-evaluates selection from scratch).
 	ErrCampaignDeferred = errors.New("campaign send deferred - no eligible mailbox for this contact right now")
+
+	// ErrSenderBusy is the narrower deferral: this lead's sequence belongs to
+	// one mailbox, and that mailbox has nothing left today. Every step of a
+	// conversation comes from the address the contact first heard from, so the
+	// lead waits for it rather than being written to by a stranger.
+	//
+	// It wraps ErrCampaignDeferred, so every caller that reschedules on a
+	// deferral behaves exactly as before; only the contact drawer, which words
+	// the reason, tests for it.
+	ErrSenderBusy = fmt.Errorf("%w: the mailbox this lead's sequence belongs to has no capacity left today", ErrCampaignDeferred)
 )

 // DeferSlot is the wakeup time a caller must use after CalculateNextCampaignTime
@@ -313,10 +313,13 @@ type AccountCandidate struct {
 	// Behavior is the mailbox's resolved sending-behaviour profile for this
 	// pass. Zero value (Enabled false) means the mailbox has not opted in.
 	Behavior behavior.Resolved
-	// BehaviorOpenAt is the earliest instant the mailbox's rolled workday and
-	// hourly ceiling allow, computed while filtering candidates. nil when
-	// behaviour is off for this mailbox.
-	BehaviorOpenAt *time.Time
+	// OpenAt is the earliest instant the mailbox's own calendar allows: its
+	// rolled workday and hourly ceiling under a behaviour profile, otherwise
+	// the reopening of its 8am-8pm band, and it is only set for a mailbox the
+	// pass decided to wait for rather than drop. OpenLoc is the timezone that
+	// instant's DAY is counted in. Both nil when the mailbox can send now.
+	OpenAt  *time.Time
+	OpenLoc *time.Location
 }

 // remainingSendMinutes returns how much sending time is left in the day the
@@ -363,10 +366,10 @@ func remainingSendMinutes(c *AccountCandidate, at time.Time, sw models.ScheduleW
 // have between them on the day `at` falls on — the denominator the even-
 // distribution step paces the day across.
 //
-// Only mailboxes actually landing on that day are counted. A behaviour-profiled
-// mailbox whose today is spent has already been walked to a later day by
-// placeWithinBehavior, and counting tomorrow's allowance as if it were available
-// now would pace the campaign faster than the mailboxes that can send today can
+// Only mailboxes actually landing on that day are counted. A mailbox whose
+// today is spent (or whose hours have closed) has already been walked to a
+// later day, and counting tomorrow's allowance as if it were available now
+// would pace the campaign faster than the mailboxes that can send today can
 // keep up with. A mailbox that opened BEFORE `at` is available whatever day it
 // opened on, which is what keeps the count right when the whole pass has been
 // pushed to tomorrow because every mailbox was at capacity today.
@@ -377,8 +380,8 @@ func poolRemainingOn(pool []AccountCandidate, at time.Time) int {
 		if c.RemainingToday <= 0 {
 			continue
 		}
-		if c.Behavior.Enabled && c.BehaviorOpenAt != nil &&
-			c.BehaviorOpenAt.After(at) && !sameLocalDay(*c.BehaviorOpenAt, at, c.Behavior.Loc) {
+		if c.OpenAt != nil && c.OpenLoc != nil &&
+			c.OpenAt.After(at) && !sameLocalDay(*c.OpenAt, at, c.OpenLoc) {
 			continue
 		}
 		total += c.RemainingToday
@@ -707,7 +707,7 @@ func TestLiveInFlightSendIsNotOfferedAgain(t *testing.T) {

 	// A's send was reserved and dispatched; the stamp never landed.
 	progress := repository.NewCampaignProgressRepository(pool)
-	reserved, err := progress.ReserveSend(ctx, f.campaign, leadA, step1, uuid.New(), true)
+	reserved, err := progress.ReserveSend(ctx, f.campaign, leadA, step1, uuid.New(), uuid.Nil, true)
 	if err != nil || !reserved {
 		t.Fatalf("reserve A's send: reserved=%v err=%v", reserved, err)
 	}
@@ -723,7 +723,7 @@ func TestLiveInFlightSendIsNotOfferedAgain(t *testing.T) {

 	// With B dispatched too, nothing is left: the campaign completes rather than
 	// re-offering either in-flight step.
-	if reserved, err := progress.ReserveSend(ctx, f.campaign, leadB, step1, uuid.New(), true); err != nil || !reserved {
+	if reserved, err := progress.ReserveSend(ctx, f.campaign, leadB, step1, uuid.New(), uuid.Nil, true); err != nil || !reserved {
 		t.Fatalf("reserve B's send: reserved=%v err=%v", reserved, err)
 	}
 	if _, pair, _, err = s.CalculateNextCampaignTime(ctx, f.campaign); !errors.Is(err, ErrCampaignCompleted) || pair != nil {
@@ -777,7 +777,7 @@ func TestLiveInFlightFollowUpIsNotOfferedAgain(t *testing.T) {
 		t.Fatalf("precondition: step 2 should be due, got pair=%v err=%v", pair, err)
 	}
 	// ...so dispatch it, and lose the stamp.
-	if reserved, err := progress.ReserveSend(ctx, f.campaign, contact, step2, uuid.New(), false); err != nil || !reserved {
+	if reserved, err := progress.ReserveSend(ctx, f.campaign, contact, step2, uuid.New(), uuid.Nil, false); err != nil || !reserved {
 		t.Fatalf("reserve the follow-up: reserved=%v err=%v", reserved, err)
 	}
 	if _, pair, _, err := s.CalculateNextCampaignTime(ctx, f.campaign); !errors.Is(err, ErrCampaignCompleted) || pair != nil {
@@ -73,8 +73,8 @@ func TestPoolRemainingSkipsMailboxesWaitingForTomorrow(t *testing.T) {
 	b.Enabled = true

 	pool := []AccountCandidate{
-		{RemainingToday: 30, Behavior: b, BehaviorOpenAt: &now},
-		{RemainingToday: 40, Behavior: b, BehaviorOpenAt: &tomorrow},
+		{RemainingToday: 30, Behavior: b, OpenAt: &now, OpenLoc: b.Loc},
+		{RemainingToday: 40, Behavior: b, OpenAt: &tomorrow, OpenLoc: b.Loc},
 	}
 	if got := poolRemainingOn(pool, now); got != 30 {
 		t.Fatalf("pool remaining = %d, want 30 (only the mailbox open today)", got)
@@ -0,0 +1,327 @@
+package tasks
+
+import (
+	"context"
+	"os"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+	"github.com/warmbly/warmbly/internal/pkg/encrypt"
+	"github.com/warmbly/warmbly/internal/repository"
+	"github.com/warmbly/warmbly/internal/scheduler"
+	"github.com/warmbly/warmbly/internal/tasks/proto"
+)
+
+// Issue #401: rotation chose a mailbox per SEND, so a six-step sequence reached
+// one contact from three different addresses. Rotation belongs to the lead, not
+// to the step: the mailbox that sends a contact's first email sends the rest of
+// their sequence, and rotation spreads NEW leads across the pool.
+//
+// Skipped unless WARMBLY_TEST_DB is set:
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/warmbly_dev?sslmode=disable \
+//	  go test ./internal/tasks/ -run Live -v
+
+type stickyFixture struct {
+	pool      *pgxpool.Pool
+	user, org uuid.UUID
+	mailboxes []uuid.UUID
+	campaign  uuid.UUID
+	steps     []uuid.UUID
+	leads     []uuid.UUID
+	svc       *tasksService
+	sender    *attributingSender
+}
+
+// newStickyFixture builds a round-robin campaign over two mailboxes with a
+// chained multi-step sequence and no waits, so consecutive ticks walk one lead
+// through its whole sequence and rotation is the only thing choosing mailboxes.
+func newStickyFixture(t *testing.T, steps, leads int) *stickyFixture {
+	t.Helper()
+	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() })
+	pool := handle.Pool
+
+	f := &stickyFixture{
+		pool: pool, user: uuid.New(), org: uuid.New(), campaign: uuid.New(),
+		mailboxes: []uuid.UUID{uuid.New(), uuid.New()},
+		sender:    &attributingSender{},
+	}
+	for i := 0; i < steps; i++ {
+		f.steps = append(f.steps, 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', 'Sticky')`,
+		f.user, "sticky-"+f.user.String()[:8]+"@test.local")
+	exec(`INSERT INTO organizations (id, name, slug, owner_user_id) VALUES ($1, 'Live Sticky', $2, $3)`,
+		f.org, "sticky-"+f.org.String()[:8], f.user)
+	for _, mb := range f.mailboxes {
+		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, 0, 'UTC')`,
+			mb, f.user, f.org, "sticky-"+mb.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 Sticky', '', 'active', 50, 'UTC', 127, '00:00', '23:59',
+	              'round_robin', NOW(), NOW())`, f.campaign, f.user, f.org)
+
+	// Each step connects to the next with an unconditional branch; the last one
+	// ends the flow. Nothing waits, so every step is due the moment the one
+	// before it was sent.
+	for i, id := range f.steps {
+		conditions := `{"branches":[]}`
+		if i+1 < len(f.steps) {
+			conditions = `{"branches":[{"branch_id":"b` + string(rune('1'+i)) + `","target_step_id":"` + f.steps[i+1].String() + `"}]}`
+		}
+		exec(`INSERT INTO sequences (id, campaign_id, organization_id, name, subject,
+		          body_plain, body_html, wait_after, position, conditions, kind)
+		      VALUES ($1, $2, $3, $4, 'Hi', 'Hello', '<p>Hello</p>', 0, $5, $6::jsonb, 'email')`,
+			id, f.campaign, f.org, "Step "+string(rune('1'+i)), i+1, conditions)
+	}
+	for i := 0; i < leads; i++ {
+		id := uuid.New()
+		f.leads = append(f.leads, id)
+		exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields, verification_status, created_at)
+		      VALUES ($1, $2, $3, $4, 'Live', 'Contact', '', '', '{}', 'valid', NOW() + make_interval(secs => $5))`,
+			id, f.user, f.org, "sticky-lead-"+id.String()[:8]+"@test.local", float64(i))
+		exec(`INSERT INTO campaign_leads (campaign_id, contact_id, position) VALUES ($1, $2, $3)`, f.campaign, id, i)
+	}
+
+	t.Cleanup(func() {
+		c := context.Background()
+		for _, step := range []struct {
+			sql string
+			arg any
+		}{
+			{`DELETE FROM task_failures WHERE task_id IN (SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign},
+			{`DELETE FROM task_execution_keys WHERE task_id IN (SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign},
+			{`DELETE FROM tasks WHERE id IN (SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign},
+			{`DELETE FROM campaign_tasks WHERE campaign_id = $1`, f.campaign},
+			{`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 organization_id = $1`, f.org},
+			{`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)
+			}
+		}
+	})
+
+	enc, err := encrypt.NewEncrypter([]byte("0123456789abcdef0123456789abcdef"))
+	if err != nil {
+		t.Fatalf("encrypter: %v", err)
+	}
+	emailRepo := repository.NewEmailRepostory(handle, enc)
+	campaignRepo := repository.NewCampaignRepostory(handle)
+	contactRepo := repository.NewContactRepostory(handle)
+	logRepo := repository.NewCampaignLogRepository(handle)
+	progressRepo := repository.NewCampaignProgressRepository(pool)
+	taskRepo := repository.NewTaskRepository(pool)
+	f.svc = &tasksService{
+		tasksClient:          noopTaskScheduler{},
+		scheduler:            scheduler.NewSchedulerService(taskRepo, repository.NewWarmupRepository(pool), progressRepo, emailRepo, campaignRepo, contactRepo, logRepo),
+		cipherService:        noopCipher{},
+		emailSender:          f.sender,
+		taskRepo:             taskRepo,
+		campaignProgressRepo: progressRepo,
+		emailRepo:            emailRepo,
+		campaignRepo:         campaignRepo,
+		contactRepo:          contactRepo,
+		campaignLogRepo:      logRepo,
+		trackedLinkRepo:      repository.NewTrackedLinkRepository(pool),
+	}
+	return f
+}
+
+// tick runs one real campaign tick, seeded the way the chain seeds it.
+func (f *stickyFixture) tick(t *testing.T) uuid.UUID {
+	t.Helper()
+	ctx := context.Background()
+	if _, err := f.pool.Exec(ctx,
+		`UPDATE tasks SET status = 'cancelled' WHERE status = 'pending' AND id IN (
+			SELECT task_id FROM campaign_tasks WHERE campaign_id = $1)`, f.campaign); err != nil {
+		t.Fatalf("clear pending tasks: %v", err)
+	}
+	taskID := uuid.New()
+	if _, err := f.pool.Exec(ctx, `
+		INSERT INTO tasks (id, task_type, email_account_id, status, message_id, scheduled_at, created_at, updated_at)
+		VALUES ($1, 'campaign', $2, 'pending', '', NOW(), NOW(), NOW())`, taskID, f.mailboxes[0]); err != nil {
+		t.Fatalf("create task: %v", err)
+	}
+	if _, err := f.pool.Exec(ctx, `INSERT INTO campaign_tasks (task_id, campaign_id) VALUES ($1, $2)`,
+		taskID, f.campaign); err != nil {
+		t.Fatalf("create campaign task: %v", err)
+	}
+	if xerr := f.svc.HandleCampaignTask(&proto.ProcessTask{TaskId: taskID.String()}); xerr != nil {
+		t.Fatalf("campaign tick: %v", xerr)
+	}
+	return taskID
+}
+
+// sendsByLead maps each contact to the mailboxes its steps were sent from, in
+// order, read from the tasks that actually dispatched them.
+func (f *stickyFixture) sendsByLead(t *testing.T) map[uuid.UUID][]uuid.UUID {
+	t.Helper()
+	rows, err := f.pool.Query(context.Background(), `
+		SELECT ct.contact_id, t.email_account_id
+		FROM campaign_tasks ct
+		JOIN tasks t ON t.id = ct.task_id
+		WHERE ct.campaign_id = $1 AND ct.contact_id IS NOT NULL AND t.status = 'completed'
+		ORDER BY t.created_at ASC`, f.campaign)
+	if err != nil {
+		t.Fatalf("read sends: %v", err)
+	}
+	defer rows.Close()
+	out := map[uuid.UUID][]uuid.UUID{}
+	for rows.Next() {
+		var contact, mailbox uuid.UUID
+		if err := rows.Scan(&contact, &mailbox); err != nil {
+			t.Fatalf("scan sends: %v", err)
+		}
+		out[contact] = append(out[contact], mailbox)
+	}
+	return out
+}
+
+func (f *stickyFixture) boundSender(t *testing.T, contact uuid.UUID) *uuid.UUID {
+	t.Helper()
+	var id *uuid.UUID
+	if err := f.pool.QueryRow(context.Background(),
+		`SELECT email_account_id FROM campaign_leads WHERE campaign_id = $1 AND contact_id = $2`,
+		f.campaign, contact).Scan(&id); err != nil {
+		t.Fatalf("read lead sender: %v", err)
+	}
+	return id
+}
+
+// TestLiveSequenceKeepsOneSenderPerLead is issue #401: every step a contact
+// receives comes from the mailbox that sent them the first one, while rotation
+// still spreads the two leads across the two mailboxes.
+func TestLiveSequenceKeepsOneSenderPerLead(t *testing.T) {
+	f := newStickyFixture(t, 3, 2)
+
+	// Six ticks: three steps for each of the two leads.
+	for i := 0; i < 6; i++ {
+		f.tick(t)
+	}
+
+	sends := f.sendsByLead(t)
+	if len(sends) != 2 {
+		t.Fatalf("expected both leads to be emailed, got sends for %d", len(sends))
+	}
+	used := map[uuid.UUID]bool{}
+	for _, contact := range f.leads {
+		got := sends[contact]
+		if len(got) != 3 {
+			t.Fatalf("lead %s received %d steps, want 3", contact, len(got))
+		}
+		for _, mb := range got[1:] {
+			if mb != got[0] {
+				t.Fatalf("lead %s heard from %v: a sequence must stay on the mailbox that sent its first email", contact, got)
+			}
+		}
+		if bound := f.boundSender(t, contact); bound == nil || *bound != got[0] {
+			t.Fatalf("lead %s sent from %s but is recorded against %v", contact, got[0], bound)
+		}
+		used[got[0]] = true
+	}
+	if len(used) != 2 {
+		t.Fatalf("both leads were started on the same mailbox; rotation must still spread NEW leads")
+	}
+}
+
+// TestLiveLeadMovesOffAMailboxThatCanNoLongerSend: staying on one address is a
+// rule while the mailbox can send. A mailbox that is disconnected, or taken off
+// the campaign, is not coming back, and a lead left waiting for it would go
+// silent mid-sequence — so it moves, and the activity log says so.
+func TestLiveLeadMovesOffAMailboxThatCanNoLongerSend(t *testing.T) {
+	f := newStickyFixture(t, 2, 1)
+
+	f.tick(t)
+	first := f.sendsByLead(t)[f.leads[0]]
+	if len(first) != 1 {
+		t.Fatalf("first step did not send: %v", first)
+	}
+	if _, err := f.pool.Exec(context.Background(),
+		`UPDATE email_accounts SET status = 'inactive' WHERE id = $1`, first[0]); err != nil {
+		t.Fatalf("disconnect mailbox: %v", err)
+	}
+
+	f.tick(t)
+	got := f.sendsByLead(t)[f.leads[0]]
+	if len(got) != 2 {
+		t.Fatalf("the follow-up did not send after its mailbox was disconnected: %v", got)
+	}
+	if got[1] == got[0] {
+		t.Fatalf("the follow-up left from the disconnected mailbox %s", got[0])
+	}
+	if bound := f.boundSender(t, f.leads[0]); bound == nil || *bound != got[1] {
+		t.Fatalf("the lead is recorded against %v, want its new mailbox %s", bound, got[1])
+	}
+	var logged int
+	if err := f.pool.QueryRow(context.Background(),
+		`SELECT COUNT(*) FROM campaign_logs WHERE campaign_id = $1 AND event_type = 'sender_reassigned'`,
+		f.campaign).Scan(&logged); err != nil {
+		t.Fatalf("read logs: %v", err)
+	}
+	if logged != 1 {
+		t.Fatalf("moving a lead to another mailbox wrote %d activity lines, want 1", logged)
+	}
+}
+
+// TestLiveABusySenderHoldsItsOwnLeadOnly: a mailbox that has used its budget is
+// coming back tomorrow, so its leads wait rather than switching address. The
+// leads behind them must NOT wait with them — that would drop a campaign's
+// whole throughput to the first mailbox that fills up.
+func TestLiveABusySenderHoldsItsOwnLeadOnly(t *testing.T) {
+	f := newStickyFixture(t, 2, 2)
+
+	f.tick(t)
+	first := f.sendsByLead(t)[f.leads[0]]
+	if len(first) != 1 {
+		t.Fatalf("first step did not send: %v", first)
+	}
+	// That mailbox has now sent its one email for the day.
+	if _, err := f.pool.Exec(context.Background(),
+		`UPDATE email_accounts SET campaign_limit = 1 WHERE id = $1`, first[0]); err != nil {
+		t.Fatalf("cap mailbox: %v", err)
+	}
+
+	f.tick(t)
+	sends := f.sendsByLead(t)
+	if got := sends[f.leads[0]]; len(got) != 1 {
+		t.Fatalf("lead 1 was emailed again from %v while its own mailbox was out of budget", got)
+	}
+	got := sends[f.leads[1]]
+	if len(got) != 1 {
+		t.Fatalf("lead 2 did not send while a free mailbox was available: %v", got)
+	}
+	if got[0] == first[0] {
+		t.Fatalf("lead 2 sent from the mailbox that has no budget left")
+	}
+}
@@ -685,7 +685,11 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error {
 	// and the next tick emails the same person again (issue #169). The
 	// reservation also counts the send against the day's counters, so a lost
 	// stamp can never let the daily cap over-send either.
-	reserved, rerr := s.campaignProgressRepo.ReserveSend(ctx, campaign.ID, contact.ID, sequence.ID, taskID, nextPair.IsNewLead)
+	//
+	// It also binds the lead to this mailbox, in the same transaction, so every
+	// remaining step of this contact's sequence leaves from the address they
+	// are about to hear from (issue #401).
+	reserved, rerr := s.campaignProgressRepo.ReserveSend(ctx, campaign.ID, contact.ID, sequence.ID, taskID, account.ID, nextPair.IsNewLead)
 	if rerr != nil {
 		// The attempt could not be made durable, so it must not be made at all.
 		// Retry the whole task rather than sending something nothing remembers.
@@ -857,6 +861,29 @@ func (s *tasksService) HandleCampaignTask(task *proto.ProcessTask) *errx.Error {
 		})
 	}

+	// This contact's sequence changed address. It only happens when the mailbox
+	// they had been hearing from stopped being one this campaign can send from
+	// (disconnected, taken off its sending accounts, resting, or held by warmup
+	// health), and it is the kind of thing an owner should find in the activity
+	// log rather than in a confused reply.
+	if prev := nextPair.AssignedSender; prev != nil && *prev != account.ID && s.campaignLogRepo != nil {
+		from := "its previous mailbox"
+		if old, oerr := s.emailRepo.GetByID(ctx, *prev); oerr == nil && old != nil {
+			from = old.Email
+		}
+		s.campaignLogRepo.CreateLog(ctx, &repository.CampaignLogEntry{
+			CampaignID: campaign.ID,
+			EventType:  "sender_reassigned",
+			Message: fmt.Sprintf("%s now hears from %s: %s can no longer send for this campaign",
+				contact.Email, account.Email, from),
+			Metadata: map[string]interface{}{
+				"contact_id":      contact.ID.String(),
+				"account_id":      account.ID.String(),
+				"prev_account_id": prev.String(),
+			},
+		})
+	}
+
 	// STEP 18: Mark task as completed (with advisory lock)
 	if err := s.taskRepo.UpdateTaskStatusWithLock(ctx, taskID, "completed"); err != nil {
 		errs.CaptureException(err)
@@ -99,7 +99,7 @@ const faq = [
  ['How do A/B variants get chosen?',
   'Each contact is assigned a variant at the step by a stable weighted split, so a retry always lands on the same copy. We report which variant earned the reply.'],
  ['Where does a sequence actually send from?',
-   'A pool of your mailboxes, not one. The scheduler round-robins across the pool under each mailbox cap, the 600s spacing rule, and its current health band, so no single mailbox carries the campaign.'],
+   'A pool of your mailboxes, not one. The scheduler round-robins leads across the pool under each mailbox cap, the 600s spacing rule, and its current health band, so no single mailbox carries the campaign. A lead keeps the mailbox that sent its first email, so every follow-up comes from the same address.'],
  ['What stops a sequence for a contact?',
   'A reply, a met goal, a bounce, an unsubscribe, a STOP reply, or a manual suppression. Suppression is workspace-wide, so another sequence never picks the contact back up by accident.'],
 ];
@@ -593,7 +593,7 @@ const totalCap = senderPool.reduce((a, m) => a + m.cap, 0);
          One sequence, many mailboxes.
        </h2>
        <p class="mt-4 text-[15px] text-foreground/70 leading-relaxed">
-          A sequence sends through a pool of mailboxes, not one. The scheduler round-robins contacts across the pool with an eye on each mailbox daily cap and current health band, so no single mailbox ever carries the campaign.
+          A sequence sends through a pool of mailboxes, not one. The scheduler round-robins contacts across the pool with an eye on each mailbox daily cap and current health band, so no single mailbox ever carries the campaign. Rotation happens per lead, not per email: whichever mailbox sends a contact their first email sends the rest of the sequence too.
        </p>
      </div>

@@ -622,7 +622,7 @@ const totalCap = senderPool.reduce((a, m) => a + m.cap, 0);
          );
        })}
        <div class="px-5 sm:px-7 py-3 border-t border-[color:var(--border)] bg-[color:var(--surface-1)]/60 flex flex-wrap items-center justify-between gap-2 text-[11.5px] font-mono text-muted-foreground">
-          <span>Round-robin &middot; health-aware &middot; cap-aware</span>
+          <span>Round-robin per lead &middot; health-aware &middot; cap-aware</span>
          <span>Pool headroom: {totalCap - totalToday} sends today</span>
        </div>
      </div>
@@ -21,7 +21,7 @@ const groups = [
    title: 'Sending',
    items: [
      ['What is the per-mailbox cap?', 'Default 50 cold emails / day with a 10-minute minimum gap. You can lower it. You can raise it, but anything above 50 / day requires positive reputation signals.'],
-      ['How do you spread sending across IPs?', 'One worker per machine, each with its own IP. We distribute mailboxes across workers and assign campaign sends round-robin across the chosen sender pool.'],
+      ['How do you spread sending across IPs?', 'One worker per machine, each with its own IP. We distribute mailboxes across workers and assign campaign leads round-robin across the chosen sender pool, with each lead keeping the mailbox that sent its first email.'],
      ['Do you offer dedicated IPs?', 'Business and Enterprise run your sending on a worker machine allocated to your organisation alone, so your reputation is not shared with other customers. We do not sell or manage IPs as a product, and we do not let you bring your own.'],
      ['What providers do you support?', 'Google Workspace / Gmail, Microsoft 365 / Outlook, and any SMTP / IMAP provider. iCloud and Zoho are first-class too.'],
    ],
@@ -115,8 +115,9 @@ export function RotationRampSection({
                    }
                />
                <p className="text-[11px] text-slate-400 mt-2 leading-relaxed">
-                    Each mailbox stays within its own daily limit, and follow-ups always come from the mailbox that
-                    sent the first email — so every thread stays consistent.
+                    Rotation picks the mailbox for a lead's first email, and every follow-up to that lead comes from
+                    the same address, so the conversation stays consistent. Each mailbox stays within its own daily
+                    limit, and a lead only changes address if its mailbox stops being able to send for this campaign.
                </p>
            </div>

@@ -1233,7 +1233,20 @@ function ContactsTableBody({
                            </>
                        )}
                        {embedded ? (
-                            <Th className="w-28 hidden md:table-cell">Current step</Th>
+                            <>
+                                <Th className="w-28 hidden md:table-cell">Current step</Th>
+                                <Th className="w-36 hidden xl:table-cell">
+                                    <span className="inline-flex items-center gap-1">
+                                        Sender
+                                        <span
+                                            className="inline-flex cursor-help text-slate-300 hover:text-slate-500"
+                                            title="The mailbox this lead's whole sequence sends from. It is picked when the first email goes out and every follow-up keeps it, so the contact always hears from one address."
+                                        >
+                                            <InfoIcon className="w-3 h-3" aria-label="How the sender is chosen" />
+                                        </span>
+                                    </span>
+                                </Th>
+                            </>
                        ) : (
                            <Th className="w-24 text-right hidden md:table-cell">Campaigns</Th>
                        )}
@@ -1369,6 +1382,7 @@ function ContactsTableBody({
                                    </>
                                )}
                                {embedded ? (
+                                    <>
                                    <td className="px-3 hidden md:table-cell">
                                        {lead?.current_step ? (
                                            <span
@@ -1385,6 +1399,19 @@ function ContactsTableBody({
                                            <span className="text-[11px] text-slate-300">Not started</span>
                                        )}
                                    </td>
+                                    <td className="px-3 hidden xl:table-cell">
+                                        {lead?.sender ? (
+                                            <span
+                                                title={`Every step of this lead's sequence sends from ${lead.sender}`}
+                                                className="block truncate text-[11.5px] text-slate-600"
+                                            >
+                                                {lead.sender}
+                                            </span>
+                                        ) : (
+                                            <span className="text-[11px] text-slate-300">Not assigned</span>
+                                        )}
+                                    </td>
+                                    </>
                                ) : (
                                    <td className="px-3 text-right font-mono text-[12px] text-slate-600 tabular-nums hidden md:table-cell">
                                        {c.campaigns?.length ?? 0}
@@ -335,6 +335,20 @@ function CampaignCard({ state }: { state: ContactCampaignState }) {
                            )
                        }
                    />
+                    <PanelFact
+                        label="Sender"
+                        value={
+                            state.sender_email ? (
+                                <span title={`Every step of this sequence sends from ${state.sender_email}`}>
+                                    {state.sender_email}
+                                </span>
+                            ) : (
+                                <span title="Picked when the first email goes out; every follow-up then keeps it.">
+                                    Chosen at the first email
+                                </span>
+                            )
+                        }
+                    />
                    <div className="sm:col-span-2">
                        <NextActionFact
                            next={state.next}
@@ -44,6 +44,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 mailbox this lead's whole sequence sends from, fixed when its first
+    // email went out. Empty until then.
+    sender?: string;
    // The worker's reason for the last failed send; set only when status is
    // "failed".
    failure_reason?: string;
@@ -48,6 +48,12 @@ export default interface ContactCampaignState {
    lead_status: LeadStatus;
    failure_reason?: string;

+    // The mailbox this lead's whole sequence sends from. Rotation picks it for
+    // the first email and every follow-up keeps it, so the contact always hears
+    // from one address. Absent until the first email goes out.
+    sender_id?: string | null;
+    sender_email?: string;
+
    steps: ContactCampaignStep[];
    completed_steps: number;
    total_steps: number;