Merge pull request #59 from warmbly/feature/human-send-behavior

Human-like send + warmup engagement timing (no metronome, round-mark, or 3am-read fingerprints)

docs/content/docs/guides/deliverability.mdx

@@ -140,6 +140,10 @@ Campaigns skip suppressed recipients automatically. This is intentional: when a
 
 The **Suppressed** count on the dashboard shows how many recipients are currently held back this way.
 
+### How bounces are detected
+
+When a message can't be delivered, the recipient's mail server sends back a bounce notification (a delivery-status report) that lands in your mailbox. Warmbly reads those automatically and, only for permanent failures (a `5.x.x` status, such as an address that doesn't exist), records the bounce against the exact send and suppresses that recipient. Temporary failures (a `4.x.x` status, such as a full mailbox or a greylisting delay) are never treated as permanent, so a recipient is never suppressed over a transient hiccup. This works the same for every provider, including mailboxes connected over the provider APIs (Gmail and Microsoft Graph), where the send itself reports success and the bounce only shows up later.
+
 ## How to read your dashboard
 
 A practical way to use the page:

docs/content/docs/guides/warmup.mdx

@@ -78,6 +78,8 @@ Real inboxes do not only send. They also reply. Warmup mimics this with a config
 
 A portion of the time, instead of starting a new message, a mailbox will reply to an existing warmup thread it received. The reply stays on the same topic as the original message and threads correctly with a real `Re:` subject and reply headers, so the exchange looks like a genuine back-and-forth rather than a one-way blast. Replies are tracked separately in your warmup stats as a healthy-engagement signal.
 
+Timing follows human behavior throughout. Sends are spread over the day in bursts and lulls rather than a fixed rhythm, never land on round clock marks, and replies only go to messages that arrived long enough ago to have plausibly been read. On the receiving side, opens and reads happen on a natural delay (most within minutes, some up to an hour later) and only during the recipient's waking hours, so no mailbox in the pool "reads" email at 3am or reacts to every message within seconds.
+
 ## Keeping warmup out of your inbox
 
 Warmup traffic should build reputation without cluttering the mailbox you actually read. When a warmup message is received, Warmbly recognizes it by its verification token and files it away automatically: the message is moved out of the inbox into a dedicated `Warmbly` folder and marked as read, and anything that landed in spam is rescued back to the inbox first so providers record a positive placement signal. This works across Gmail, Outlook (via Microsoft Graph), and generic IMAP mailboxes, and warmup mail never appears in your unified inbox. The engagement (open, read, occasional reply, importance) still happens, so providers see the healthy signals while you see a clean inbox.

internal/app/advanced/inbound_bounce.go

@@ -0,0 +1,68 @@
+package advanced
+
+import (
+	"context"
+	"strings"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+// RecordInboundBounce turns a permanent NDR the worker parsed into a bounce
+// deliverability event. The worker only emits permanent (5.x.x) bounces with an
+// original Message-ID, so this resolves that id to the campaign send and routes
+// it through IngestDeliverabilityEvent (suppression + campaign progress +
+// breaker), keyed idempotently so re-delivered NDRs don't double-count.
+func (s *service) RecordInboundBounce(ctx context.Context, emailAccountID uuid.UUID, originalMessageID, failedRecipient, reason string) *errx.Error {
+	originalMessageID = strings.Trim(strings.TrimSpace(originalMessageID), "<>")
+	if originalMessageID == "" {
+		return nil
+	}
+
+	task, err := s.taskRepo.GetTaskByMessageID(ctx, originalMessageID)
+	if err != nil || task == nil {
+		// Unknown message id (warmup mail, non-campaign send, or already
+		// pruned) — nothing to attribute the bounce to.
+		return nil
+	}
+
+	// The NDR should have landed in the same mailbox that sent it; if it didn't
+	// resolve to this account, don't attribute a cross-account bounce.
+	if task.EmailAccountID != emailAccountID {
+		return nil
+	}
+
+	account, aerr := s.emailRepo.GetByID(ctx, emailAccountID)
+	if aerr != nil || account == nil || account.OrganizationID == nil {
+		return nil
+	}
+
+	req := &models.IngestDeliverabilityEventRequest{
+		EventType:      models.DeliverabilityEventBounce,
+		Provider:       "inbound_ndr",
+		TaskID:         &task.ID,
+		RecipientEmail: failedRecipient,
+		Reason:         reason,
+		// Same NDR re-synced (delta re-runs, reconnects) must not double-count.
+		IdempotencyKey: "ndr:" + originalMessageID,
+	}
+
+	if ct, cerr := s.taskRepo.GetCampaignTask(ctx, task.ID); cerr == nil && ct != nil {
+		req.CampaignID = ct.CampaignID
+		req.ContactID = ct.ContactID
+		if req.RecipientEmail == "" && ct.ContactID != nil {
+			if contact, cerr := s.contactRepo.GetByID(ctx, *ct.ContactID); cerr == nil && contact != nil {
+				req.RecipientEmail = contact.Email
+			}
+		}
+	}
+
+	if req.RecipientEmail == "" {
+		// IngestDeliverabilityEvent requires a recipient; without one we can't
+		// suppress or record. Give up rather than guess.
+		return nil
+	}
+
+	return s.IngestDeliverabilityEvent(ctx, *account.OrganizationID, req)
+}

internal/app/advanced/service.go

@@ -37,6 +37,11 @@ type Service interface {
 
 	IngestDeliverabilityEvent(ctx context.Context, organizationID uuid.UUID, req *models.IngestDeliverabilityEventRequest) *errx.Error
 
+	// RecordInboundBounce resolves a permanent NDR (parsed worker-side) back to
+	// the original campaign send via its Message-ID and records a bounce
+	// deliverability event. Best-effort: unresolvable bounces are a no-op.
+	RecordInboundBounce(ctx context.Context, emailAccountID uuid.UUID, originalMessageID, failedRecipient, reason string) *errx.Error
+
 	ShouldSuppressRecipient(ctx context.Context, organizationID uuid.UUID, recipient string) (bool, string, *errx.Error)
 	// Unsubscribe suppresses a contact in response to a List-Unsubscribe action
 	// (one-click POST or the manual link). Always suppresses — it's an explicit

internal/app/consumer/event_inbound_bounce.go

@@ -0,0 +1,26 @@
+package jobs
+
+import (
+	"context"
+
+	"github.com/rs/zerolog/log"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+// HandleInboundBounce records a permanent NDR (parsed worker-side) against the
+// original campaign send: suppress the recipient, mark the send bounced, and
+// feed the deliverability breaker. Best-effort — a bounce we can't attribute is
+// logged, not retried.
+func (s *JobsService) HandleInboundBounce(ctx context.Context, e *models.JobEventInboundBounce) error {
+	if s.AdvancedService == nil {
+		return nil
+	}
+	if xerr := s.AdvancedService.RecordInboundBounce(ctx, e.EmailID, e.OriginalMessageID, e.FailedRecipient, e.Reason); xerr != nil {
+		log.Warn().
+			Str("email_id", e.EmailID.String()).
+			Str("original_message_id", e.OriginalMessageID).
+			Str("error", xerr.Message).
+			Msg("Failed to record inbound bounce")
+	}
+	return nil
+}

internal/app/consumer/event_new_email.go

@@ -192,11 +192,14 @@ func (s *JobsService) performWarmupActions(ctx context.Context, e *models.JobEve
 		RFCMessageID: e.Message.MessageID,
 	}
 
-	// Resolve the receiving mailbox's worker once.
+	// Resolve the receiving mailbox once (worker routing + timezone for the
+	// waking-hours engagement guard).
 	var workerID *uuid.UUID
+	var recipientTZ string
 	if s.EmailRepository != nil {
 		if account, xerr := s.EmailRepository.GetByID(ctx, e.Message.EmailID); xerr == nil && account != nil {
 			workerID = account.WorkerID
+			recipientTZ = account.Timezone
 		}
 	}
 	if workerID == nil {
@@ -238,7 +241,7 @@ func (s *JobsService) performWarmupActions(ctx context.Context, e *models.JobEve
 		s.Publisher.PublishWarmupAction(ctx, *workerID, &act)
 		return
 	}
-	fireAt := time.Now().Add(time.Duration(delaySeconds) * time.Second)
+	fireAt := humanizeFireAt(time.Now().Add(time.Duration(delaySeconds)*time.Second), recipientTZ)
 	if err := s.WarmupEngagementRepo.EnqueuePendingEngagement(ctx, e.Message.EmailID, payload, fireAt); err != nil {
 		log.Warn().Err(err).Str("email_id", e.Message.EmailID.String()).Msg("Failed to enqueue delayed warmup engagement; publishing immediately")
 		s.Publisher.PublishWarmupAction(ctx, *workerID, &act)

internal/app/consumer/events.go

@@ -21,6 +21,7 @@ func (s *JobsService) HandleEvent(ctx context.Context, event *models.JobEvent) e
 func (w *JobsService) InitEvents() {
 	w.eventHandlers = make(map[models.JobEventType]func(ctx context.Context, body any) error)
 	Register(w, models.JobEventTypeNewEmail, w.HandleNewEmail)
+	Register(w, models.JobEventTypeInboundBounce, w.HandleInboundBounce)
 	Register(w, models.JobEventTypeEmailUpdate, w.HandleUpdateEmail)
 	Register(w, models.JobEventTypeRemoveEmail, w.HandleRemoveEmail)
 	Register(w, models.JobEventTypeFlagsAdd, w.HandleFlagsAdd)

internal/app/consumer/warmup_engagement.go

@@ -2,6 +2,7 @@ package jobs
 
 import (
 	"context"
+	"math"
 	"math/rand"
 	"sync"
 	"time"
@@ -62,21 +63,31 @@ func engagementPlan(accountID uuid.UUID, e models.WarmupEngagementSettings) (act
 	// Foldering is organisational, not an engagement fingerprint — always do it.
 	actions = append(actions, "move_to_warmbly")
 
-	if rollPct(e.MarkReadRate, p.Bias("read", 0.85, 1.15)) {
-		actions = append(actions, "mark_read")
-	}
-	// Spam-rescue: the worker only actually moves it if it's in spam.
+	// Spam-rescue is the reputation-critical "not spam" signal warmup exists for,
+	// so it fires regardless of neglect (and the worker only acts if it's really
+	// in spam).
 	if rollPct(e.SpamRescueRate, p.Bias("rescue", 0.8, 1.2)) {
 		actions = append(actions, "remove_from_spam")
 	}
-	if rollPct(e.MarkImportantRate, p.Bias("important", 0.7, 1.3)) {
-		actions = append(actions, "mark_important")
-	}
-	// Starring is a separate, lower-rate positive signal (Gmail STARRED). On
-	// IMAP the worker no-ops it because \Flagged is already covered by
-	// mark_important — so it never double-flags the same message.
-	if rollPct(e.StarRate, p.Bias("star", 0.6, 1.4)) {
-		actions = append(actions, "star")
+
+	// Occasionally a mailbox files a message but never engages with it — real
+	// inboxes are full of read-later-and-forgotten mail. Skip the positive
+	// engagement signals on those so the pool never shows perfect, every-message
+	// engagement. Spam-rescue and foldering above still happen.
+	neglect := rand.Float64() < 0.07*p.Bias("neglect", 0.6, 1.4)
+	if !neglect {
+		if rollPct(e.MarkReadRate, p.Bias("read", 0.85, 1.15)) {
+			actions = append(actions, "mark_read")
+		}
+		if rollPct(e.MarkImportantRate, p.Bias("important", 0.7, 1.3)) {
+			actions = append(actions, "mark_important")
+		}
+		// Starring is a separate, lower-rate positive signal (Gmail STARRED). On
+		// IMAP the worker no-ops it because \Flagged is already covered by
+		// mark_important — so it never double-flags the same message.
+		if rollPct(e.StarRate, p.Bias("star", 0.6, 1.4)) {
+			actions = append(actions, "star")
+		}
 	}
 
 	delaySeconds = dwellSeconds(e.MinDwellSeconds, e.MaxDwellSeconds, p.Bias("dwell", 0.7, 1.3))
@@ -114,6 +125,10 @@ func rollPct(rate int, bias float64) bool {
 }
 
 // dwellSeconds returns a randomised delay within [min,max], nudged by persona.
+// The sample is heavy-tailed (u^2.2 curve), not uniform: most mail gets opened
+// within the first stretch of the range, a long tail waits much longer — the
+// shape of real inbox-checking, where a uniform "always read within N minutes
+// of delivery, around the clock" is a lockstep signature.
 func dwellSeconds(minS, maxS int, bias float64) int {
 	if maxS <= 0 || maxS < minS {
 		return 0
@@ -121,7 +136,8 @@ func dwellSeconds(minS, maxS int, bias float64) int {
 	span := maxS - minS
 	base := minS
 	if span > 0 {
-		base += rand.Intn(span + 1)
+		u := rand.Float64()
+		base += int(float64(span) * math.Pow(u, 2.2))
 	}
 	out := int(float64(base) * bias)
 	if out < minS {
@@ -132,3 +148,29 @@ func dwellSeconds(minS, maxS int, bias float64) int {
 	}
 	return out
 }
+
+// humanizeFireAt keeps recipient-side engagement inside plausible waking hours.
+// A read that fires at 3am local is a bot signature; anything landing in the
+// night window (22:30–07:30) is deferred to the next morning at a randomised
+// 07:30–09:30 moment in the recipient's timezone.
+func humanizeFireAt(fireAt time.Time, timezone string) time.Time {
+	loc, err := time.LoadLocation(timezone)
+	if err != nil || timezone == "" {
+		return fireAt
+	}
+	local := fireAt.In(loc)
+	minutes := local.Hour()*60 + local.Minute()
+
+	const nightStart = 22*60 + 30
+	const morningEnd = 7*60 + 30
+	if minutes < nightStart && minutes >= morningEnd {
+		return fireAt
+	}
+
+	morning := time.Date(local.Year(), local.Month(), local.Day(), 7, 30, 0, 0, loc)
+	if minutes >= nightStart {
+		morning = morning.Add(24 * time.Hour)
+	}
+	offset := time.Duration(rand.Intn(120))*time.Minute + time.Duration(rand.Intn(60))*time.Second
+	return morning.Add(offset)
+}

internal/app/consumer/warmup_engagement_test.go

@@ -0,0 +1,102 @@
+package jobs
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+func TestHumanizeFireAt_DefersNightToMorning(t *testing.T) {
+	// 3am local — a read at this hour is a bot signature.
+	night := time.Date(2026, 1, 1, 3, 0, 0, 0, time.UTC)
+	got := humanizeFireAt(night, "UTC")
+	h := got.Hour()
+	if h < 7 || h > 9 {
+		t.Errorf("night fire time not deferred into 07:30-09:30 window: hour=%d (%v)", h, got)
+	}
+	if !got.After(night) {
+		t.Errorf("deferred time should be later than the night time: %v", got)
+	}
+}
+
+func TestHumanizeFireAt_KeepsDaytime(t *testing.T) {
+	day := time.Date(2026, 1, 1, 14, 0, 0, 0, time.UTC)
+	if got := humanizeFireAt(day, "UTC"); !got.Equal(day) {
+		t.Errorf("daytime fire time should be unchanged, got %v", got)
+	}
+}
+
+func TestHumanizeFireAt_LateNightRollsToNextMorning(t *testing.T) {
+	// 23:00 should roll to the NEXT day's morning window.
+	late := time.Date(2026, 1, 1, 23, 0, 0, 0, time.UTC)
+	got := humanizeFireAt(late, "UTC")
+	if got.Day() != 2 {
+		t.Errorf("23:00 should defer to next day, got day %d (%v)", got.Day(), got)
+	}
+	if got.Hour() < 7 || got.Hour() > 9 {
+		t.Errorf("deferred hour outside morning window: %d", got.Hour())
+	}
+}
+
+func TestDwellSeconds_WithinBounds(t *testing.T) {
+	for i := 0; i < 500; i++ {
+		d := dwellSeconds(45, 3600, 1.0)
+		if d < 45 || d > 3600 {
+			t.Fatalf("dwell out of bounds: %d", d)
+		}
+	}
+}
+
+func TestDwellSeconds_HeavyTailedBias(t *testing.T) {
+	// The u^2.2 curve should concentrate mass low: median well under the mid.
+	var below int
+	const n = 2000
+	for i := 0; i < n; i++ {
+		if dwellSeconds(45, 3600, 1.0) < (45+3600)/2 {
+			below++
+		}
+	}
+	// The u^2.2 curve puts its median around u=0.73, so ~73% of samples fall
+	// below the midpoint (vs 50% for a uniform draw). Assert clearly above
+	// uniform without hugging the exact figure.
+	if below < n*13/20 {
+		t.Errorf("expected heavy-tailed (>=65%% below midpoint), got %d/%d", below, n)
+	}
+}
+
+func TestEngagementPlan_AlwaysFoldersFirst(t *testing.T) {
+	set := models.WarmupEngagementSettings{
+		SpamRescueRate: 85, MarkReadRate: 95, MarkImportantRate: 30, StarRate: 20,
+		MinDwellSeconds: 45, MaxDwellSeconds: 3600,
+	}
+	for i := 0; i < 100; i++ {
+		actions, delay := engagementPlan(uuid.New(), set)
+		if len(actions) == 0 || actions[0] != "move_to_warmbly" {
+			t.Fatalf("foldering must always be first: %v", actions)
+		}
+		if delay < 45 || delay > 3600 {
+			t.Fatalf("delay out of bounds: %d", delay)
+		}
+	}
+}
+
+func TestEngagementPlan_SometimesNeglects(t *testing.T) {
+	set := models.WarmupEngagementSettings{
+		SpamRescueRate: 0, MarkReadRate: 100, MarkImportantRate: 100, StarRate: 100,
+		MinDwellSeconds: 45, MaxDwellSeconds: 3600,
+	}
+	// With all positive rates at 100%, a plan with ONLY move_to_warmbly can only
+	// happen via the neglect path. Over many mailboxes/messages it must occur.
+	neglected := 0
+	for i := 0; i < 2000; i++ {
+		actions, _ := engagementPlan(uuid.New(), set)
+		if len(actions) == 1 {
+			neglected++
+		}
+	}
+	if neglected == 0 {
+		t.Error("engagement is never neglected — every message gets perfect engagement")
+	}
+}

internal/app/email/loader.go

@@ -19,14 +19,22 @@ func (s *emailService) WireGraphDelta(repo repository.EmailGraphDeltaRepository)
 	s.graphDelta = repo
 }
 
+// reconcileRepublishInterval bounds how often the reconciler re-publishes a
+// given account. The immediate onboarding load and any reassignment still fire
+// right away (they call LoadAccountOntoWorker directly); this only throttles the
+// steady-state safety-net loop so the fleet isn't re-shipping every account's
+// decrypted credentials over Kafka every tick. A restarted worker is re-seeded
+// within this window rather than within one tick.
+const reconcileRepublishInterval = 5 * time.Minute
+
 // StartWorkerReconciler periodically ensures every active mailbox is assigned to
 // a worker and loaded onto it. Workers hold accounts in memory only, so this is
-// what makes onboarding, worker restarts, and reassignment converge: a fresh
-// account gets picked up within one interval, and a restarted worker is
-// re-seeded. PublishAddEmail is idempotent worker-side (already-loaded accounts
-// are skipped), so re-publishing every tick is safe.
+// what makes onboarding, worker restarts, and reassignment converge. Each
+// account is republished at most once per reconcileRepublishInterval;
+// PublishAddEmail is idempotent worker-side, so a republish is always safe.
 func (s *emailService) StartWorkerReconciler(ctx context.Context, interval time.Duration) {
-	s.reconcileWorkerAccounts(ctx)
+	lastPublished := map[uuid.UUID]time.Time{}
+	s.reconcileWorkerAccounts(ctx, lastPublished)
 
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -35,20 +43,37 @@ func (s *emailService) StartWorkerReconciler(ctx context.Context, interval time.
 		case <-ctx.Done():
 			return
 		case <-ticker.C:
-			s.reconcileWorkerAccounts(ctx)
+			s.reconcileWorkerAccounts(ctx, lastPublished)
 		}
 	}
 }
 
-func (s *emailService) reconcileWorkerAccounts(ctx context.Context) {
+func (s *emailService) reconcileWorkerAccounts(ctx context.Context, lastPublished map[uuid.UUID]time.Time) {
 	ids, err := s.emailRepository.ListActiveWorkerAccounts(ctx)
 	if err != nil {
 		log.Warn().Err(err).Msg("worker reconciler: list active accounts failed")
 		return
 	}
+
+	active := make(map[uuid.UUID]struct{}, len(ids))
+	now := time.Now()
 	for _, id := range ids {
+		active[id] = struct{}{}
+		if last, ok := lastPublished[id]; ok && now.Sub(last) < reconcileRepublishInterval {
+			continue
+		}
 		if err := s.LoadAccountOntoWorker(ctx, id); err != nil {
 			log.Warn().Err(err).Str("email_id", id.String()).Msg("worker reconciler: load account failed")
+			continue
+		}
+		lastPublished[id] = now
+	}
+
+	// Drop throttle entries for accounts no longer active so the map can't grow
+	// without bound as mailboxes are disconnected.
+	for id := range lastPublished {
+		if _, ok := active[id]; !ok {
+			delete(lastPublished, id)
 		}
 	}
 }

internal/app/worker/wmail/bounce.go

@@ -0,0 +1,62 @@
+package wmail
+
+import (
+	"strings"
+
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/pkg/dsn"
+)
+
+// maybeEmitBounce inspects a freshly synced inbound message and, when it is a
+// permanent delivery-status notification for one of our sends, emits an
+// INBOUND_BOUNCE event so the consumer can suppress the recipient and record the
+// bounce against the campaign. This is where API-sent (Gmail/Graph) mail finally
+// gets bounce tracking: those sends succeed synchronously, so the only bounce
+// signal is the NDR that lands back in the mailbox.
+//
+// Runs on the worker because the full DSN body is in hand here (the consumer has
+// no S3 access). It only PARSES — resolution and suppression stay control-plane.
+// Best-effort and permanent-only: a message that doesn't parse to a permanent
+// failure with a resolvable original id is silently ignored, so a transient
+// (4.x.x) bounce never suppresses a valid recipient.
+func (w *WMail) maybeEmitBounce(msg *models.EmailMessageData) {
+	from := strings.Join(msg.From, " ")
+	if !dsn.Detect(from, msg.Subject, headerFlagValue(msg.Flags, "Content-Type")) {
+		return
+	}
+
+	report := dsn.Parse(msg.BodyPlain + "\n" + msg.BodyHTML)
+	if !report.Permanent {
+		return
+	}
+
+	// Resolve the original outbound Message-ID: the DSN body's returned headers
+	// are the reliable source; fall back to the envelope In-Reply-To.
+	originalID := report.OriginalMessageID
+	if originalID == "" && len(msg.InReplyTo) > 0 {
+		originalID = strings.Trim(msg.InReplyTo[len(msg.InReplyTo)-1], "<>")
+	}
+	if originalID == "" {
+		return // nothing to resolve the campaign send against
+	}
+
+	_ = w.onEvent(models.JobEventTypeInboundBounce, &models.JobEventInboundBounce{
+		UserID:            w.UserID,
+		EmailID:           w.ID,
+		OriginalMessageID: strings.Trim(originalID, "<>"),
+		FailedRecipient:   report.FailedRecipient,
+		Reason:            msg.Subject,
+	})
+}
+
+// headerFlagValue reads a "Header:value" pseudo-flag out of the flag slice (the
+// sidecar the sync mappers use to carry internet headers). Returns "" if absent.
+func headerFlagValue(flags []string, name string) string {
+	prefix := name + ":"
+	for _, f := range flags {
+		if strings.HasPrefix(f, prefix) {
+			return strings.TrimSpace(strings.TrimPrefix(f, prefix))
+		}
+	}
+	return ""
+}

internal/app/worker/wmail/event_google.go

@@ -76,6 +76,8 @@ func (w *WMail) onGoogleMessageAdd(ctx context.Context, msg *models.EmailMessage
 		return err
 	}
 
+	w.maybeEmitBounce(msg)
+
 	if err := w.onEvent(models.JobEventTypeNewEmail, data); err != nil {
 		return err
 	}

internal/app/worker/wmail/event_graph.go

@@ -77,6 +77,7 @@ func (w *WMail) onGraphMessageAdd(ctx context.Context, msg *models.EmailMessageD
 		return err
 	}
 
+	w.maybeEmitBounce(msg)
 	return w.onEvent(models.JobEventTypeNewEmail, data)
 }
 

internal/app/worker/wmail/event_imap.go

@@ -97,6 +97,8 @@ func (w *WMail) onImapEmailUpdate(ctx context.Context, msg *models.EmailMessageD
 			return err
 		}
 
+		w.maybeEmitBounce(msg)
+
 		if err := w.onEvent(models.JobEventTypeNewEmail, data); err != nil {
 			return err
 		}

internal/client/goog/message_test.go

@@ -0,0 +1,81 @@
+package goog
+
+import (
+	"encoding/base64"
+	"slices"
+	"strings"
+	"testing"
+
+	"google.golang.org/api/gmail/v1"
+)
+
+func gh(n, v string) *gmail.MessagePartHeader { return &gmail.MessagePartHeader{Name: n, Value: v} }
+
+func hasPrefix(flags []string, prefix string) bool {
+	for _, f := range flags {
+		if strings.HasPrefix(f, prefix) {
+			return true
+		}
+	}
+	return false
+}
+
+func TestGmailMapping_ReadStateInversionAndLabels(t *testing.T) {
+	m := &gmail.Message{
+		Id: "gid1", ThreadId: "t1",
+		LabelIds: []string{"UNREAD", "STARRED", "SPAM", "CATEGORY_PROMOTIONS", "IMPORTANT"},
+		Payload:  &gmail.MessagePart{Headers: []*gmail.MessagePartHeader{gh("Subject", "hi")}},
+	}
+	d := GmailMessageToEmailData(m)
+	if slices.Contains(d.Flags, "\\Seen") {
+		t.Error("UNREAD present must NOT map to \\Seen")
+	}
+	for _, want := range []string{"\\Flagged", "\\Important", "SPAM", "CATEGORY_PROMOTIONS"} {
+		if !slices.Contains(d.Flags, want) {
+			t.Errorf("missing flag %q in %v", want, d.Flags)
+		}
+	}
+
+	// No UNREAD label => the message is read.
+	seen := GmailMessageToEmailData(&gmail.Message{Id: "x", Payload: &gmail.MessagePart{}})
+	if !slices.Contains(seen.Flags, "\\Seen") {
+		t.Error("absence of UNREAD must map to \\Seen")
+	}
+}
+
+func TestGmailMapping_SinglePartBodyUnpaddedBase64URL(t *testing.T) {
+	raw := base64.RawURLEncoding.EncodeToString([]byte("Hello, world")) // unpadded, as Gmail sends
+	m := &gmail.Message{
+		Id:      "x",
+		Payload: &gmail.MessagePart{MimeType: "text/plain", Body: &gmail.MessagePartBody{Data: raw}},
+	}
+	d := GmailMessageToEmailData(m)
+	if d.BodyPlain != "Hello, world" {
+		t.Errorf("single-part body not extracted: %q", d.BodyPlain)
+	}
+}
+
+func TestGmailMapping_WarmupAndClassificationHeaders(t *testing.T) {
+	m := &gmail.Message{
+		Id: "x",
+		Payload: &gmail.MessagePart{Headers: []*gmail.MessagePartHeader{
+			gh("X-Mailtrace-Verify", "tok123"),
+			gh("auto-submitted", "auto-replied"), // lowercase -> case-insensitive match
+		}},
+	}
+	d := GmailMessageToEmailData(m)
+	if !slices.Contains(d.Flags, "X-Mailtrace-Verify:tok123") {
+		t.Errorf("warmup token pseudo-flag missing: %v", d.Flags)
+	}
+	if !hasPrefix(d.Flags, "Auto-Submitted:") {
+		t.Errorf("classification header not surfaced case-insensitively: %v", d.Flags)
+	}
+}
+
+func TestGmailMapping_NilPayloadSafe(t *testing.T) {
+	// History stubs / deleted messages can arrive without a payload.
+	d := GmailMessageToEmailData(&gmail.Message{Id: "x"})
+	if d.GmailID != "x" {
+		t.Errorf("nil payload should still map id, got %q", d.GmailID)
+	}
+}

internal/client/msgraph/msgraph_test.go

@@ -0,0 +1,97 @@
+package msgraph
+
+import (
+	"slices"
+	"strings"
+	"testing"
+)
+
+func TestBuildMIME_HeadersThreadingAndBodies(t *testing.T) {
+	hdrs := []hdr{
+		{"From", "a@b.com"},
+		{"To", "c@d.com"},
+		{"Subject", "hi"},
+		{"Message-ID", "<mid@x>"},
+		{"In-Reply-To", "<parent@x>"},
+		{"References", "<parent@x>"},
+		{"X-Mailtrace-Verify", "tok"},
+	}
+	raw, err := buildMIME(hdrs, "plain body", "<p>html</p>", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := string(raw)
+	for _, want := range []string{
+		"Message-ID: <mid@x>",
+		"In-Reply-To: <parent@x>",
+		"References: <parent@x>",
+		"X-Mailtrace-Verify: tok",
+		"multipart/alternative",
+		"text/html",
+	} {
+		if !strings.Contains(s, want) {
+			t.Errorf("MIME missing %q", want)
+		}
+	}
+}
+
+func TestBuildMIME_PlainOnly(t *testing.T) {
+	raw, err := buildMIME([]hdr{{"From", "a@b.com"}, {"Subject", "s"}}, "just text", "", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(raw), "multipart") {
+		t.Error("plain-only message should not be multipart")
+	}
+}
+
+func TestBuildMIME_Attachment(t *testing.T) {
+	raw, err := buildMIME(
+		[]hdr{{"From", "a@b.com"}},
+		"body", "",
+		[]Attachment{{Filename: "f.txt", MimeType: "text/plain", Data: []byte("data")}},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	s := string(raw)
+	if !strings.Contains(s, "multipart/mixed") {
+		t.Error("attachment message must be multipart/mixed")
+	}
+	if !strings.Contains(s, `filename="f.txt"`) {
+		t.Error("attachment filename missing")
+	}
+}
+
+func TestToEmailData_MappingAndFlags(t *testing.T) {
+	m := &graphMessage{
+		ID:                "gid",
+		InternetMessageID: "<mid@x>",
+		ConversationID:    "conv",
+		Subject:           "hello",
+		IsRead:            true,
+		From:              &graphRecipient{EmailAddress: graphEmailAddress{Name: "Alice", Address: "a@b.com"}},
+		ToRecipients:      []graphRecipient{{EmailAddress: graphEmailAddress{Address: "c@d.com"}}},
+		Body:              &graphItemBody{ContentType: "text", Content: "hello body"},
+		InternetMessageHeaders: []graphHeader{
+			{Name: "X-Mailtrace-Verify", Value: "tok"},
+			{Name: "Auto-Submitted", Value: "auto-replied"},
+		},
+	}
+	d := m.toEmailData()
+	if d.GmailID != "gid" || d.MessageID != "<mid@x>" || d.ThreadID != "conv" {
+		t.Errorf("ids wrong: %+v", d)
+	}
+	if !slices.Contains(d.Flags, "\\Seen") {
+		t.Error("isRead should map to \\Seen")
+	}
+	if !slices.Contains(d.Flags, "X-Mailtrace-Verify:tok") {
+		t.Errorf("warmup pseudo-flag missing: %v", d.Flags)
+	}
+	if d.BodyPlain != "hello body" {
+		t.Errorf("text body wrong: %q", d.BodyPlain)
+	}
+	if len(d.From) != 1 || d.From[0] != "Alice <a@b.com>" {
+		t.Errorf("from formatting: %v", d.From)
+	}
+}

internal/client/smtpimap/imap/header_flags_test.go

@@ -0,0 +1,43 @@
+package imap
+
+import (
+	"slices"
+	"strings"
+	"testing"
+)
+
+func TestParseHeaderFlags(t *testing.T) {
+	raw := "X-Mailtrace-Verify: tok123\r\n" +
+		"Auto-Submitted: auto-replied\r\n" +
+		"Content-Type: multipart/report; report-type=delivery-status\r\n" +
+		"Subject: not in the fetch list\r\n\r\n"
+
+	got := parseHeaderFlags(strings.NewReader(raw))
+
+	if !slices.Contains(got, "X-Mailtrace-Verify:tok123") {
+		t.Errorf("warmup token flag missing: %v", got)
+	}
+	if !slices.Contains(got, "Auto-Submitted:auto-replied") {
+		t.Errorf("classification flag missing: %v", got)
+	}
+	// Content-Type value contains ':' and ';' — must survive intact (split on
+	// the first colon only, by the consumer's buildReplyHeaders).
+	if !slices.Contains(got, "Content-Type:multipart/report; report-type=delivery-status") {
+		t.Errorf("content-type flag wrong: %v", got)
+	}
+	// Subject is not in headerFetchFields, so it must not appear.
+	for _, f := range got {
+		if strings.HasPrefix(f, "Subject:") {
+			t.Errorf("unexpected Subject flag: %v", got)
+		}
+	}
+}
+
+func TestParseHeaderFlags_NilAndEmpty(t *testing.T) {
+	if got := parseHeaderFlags(nil); got != nil {
+		t.Errorf("nil reader should yield nil, got %v", got)
+	}
+	if got := parseHeaderFlags(strings.NewReader("\r\n")); len(got) != 0 {
+		t.Errorf("empty header block should yield no flags, got %v", got)
+	}
+}

internal/models/event.go

@@ -19,6 +19,7 @@ type JobEventType string
 
 const (
 	JobEventTypeNewEmail      JobEventType = "NEW_EMAIL"
+	JobEventTypeInboundBounce JobEventType = "INBOUND_BOUNCE"
 	JobEventTypeRemoveEmail   JobEventType = "REMOVE_EMAIL"
 	JobEventTypeFlagsAdd      JobEventType = "FLAGS_ADD"
 	JobEventTypeFlagsRemove   JobEventType = "FLAGS_REMOVE"

internal/models/event_w_bounce.go

@@ -0,0 +1,20 @@
+package models
+
+import "github.com/google/uuid"
+
+// JobEventInboundBounce is emitted by the worker when a synced inbound message
+// is a permanent (hard) delivery-status notification for one of our sends. The
+// worker parses the DSN it already has in hand (no SQL); the consumer resolves
+// the original send and records the bounce. Only permanent failures are emitted,
+// so acting on this never over-suppresses a transient failure.
+type JobEventInboundBounce struct {
+	UserID  uuid.UUID `json:"user_id"`
+	EmailID uuid.UUID `json:"email_id"`
+	// OriginalMessageID is the RFC Message-ID of the bounced outbound message,
+	// used to resolve the campaign/contact/task.
+	OriginalMessageID string `json:"original_message_id"`
+	// FailedRecipient is the address that bounced, when the DSN exposed it.
+	FailedRecipient string `json:"failed_recipient"`
+	// Reason is a short human string (the bounce subject) for the event record.
+	Reason string `json:"reason"`
+}

internal/models/warmup_content.go

@@ -143,8 +143,11 @@ func DefaultWarmupGenerationSettings() WarmupGenerationSettings {
 			MarkImportantRate: 30,
 			MarkReadRate:      95,
 			StarRate:          20,
-			MinDwellSeconds:   20,
-			MaxDwellSeconds:   240,
+			// Heavy-tailed sample (see dwellSeconds): most reads land within
+			// minutes, the tail waits up to an hour — "always read within 4
+			// minutes, around the clock" was a lockstep signature.
+			MinDwellSeconds: 45,
+			MaxDwellSeconds: 3600,
 		},
 	}
 }

internal/pkg/dsn/dsn.go

@@ -0,0 +1,110 @@
+// Package dsn parses delivery-status notifications (bounce messages, RFC 3464)
+// well enough to decide whether a permanent failure occurred, who it was for,
+// and which original message it concerns. It is deliberately conservative:
+// callers act (suppress a recipient) only on a confirmed permanent failure, so
+// a soft/transient bounce is never treated as hard.
+package dsn
+
+import (
+	"regexp"
+	"strings"
+)
+
+// Report is the extracted result of parsing a bounce message.
+type Report struct {
+	// IsBounce is true when the message looks like a delivery-status report at
+	// all (worth acting on); false for ordinary mail.
+	IsBounce bool
+	// Permanent is true only when a 5.x.x status or an explicit permanent
+	// "Action: failed" was found. Transient (4.x.x / delayed) stays false.
+	Permanent bool
+	// FailedRecipient is the address that bounced (Final/Original-Recipient).
+	FailedRecipient string
+	// OriginalMessageID is the Message-ID of the original outbound message the
+	// bounce concerns, recovered from the returned headers section.
+	OriginalMessageID string
+}
+
+var (
+	reStatus     = regexp.MustCompile(`(?im)^\s*Status:\s*([245])\.\d{1,3}\.\d{1,3}`)
+	reAction     = regexp.MustCompile(`(?im)^\s*Action:\s*(failed|delayed|delivered|relayed|expanded)`)
+	reFinalRcpt  = regexp.MustCompile(`(?im)^\s*(?:Final|Original)-Recipient:\s*[^;\r\n]*;\s*<?([^\s<>]+@[^\s<>]+?)>?\s*$`)
+	reMessageID  = regexp.MustCompile(`(?im)^\s*Message-ID:\s*<([^>\s]+)>`)
+	reDiagnostic = regexp.MustCompile(`(?im)^\s*Diagnostic-Code:\s*[^;\r\n]*;\s*([245])\d\d`)
+)
+
+// senderMarkers identify a machine bounce source in the From line.
+var senderMarkers = []string{"mailer-daemon", "postmaster@", "mail-daemon"}
+
+// subjectMarkers are common bounce subjects across providers.
+var subjectMarkers = []string{
+	"undeliverable", "undelivered mail", "delivery status notification",
+	"returned mail", "delivery failure", "mail delivery failed",
+	"failure notice", "message not delivered", "delivery incomplete",
+}
+
+// Detect reports whether an inbound message looks like a bounce, using only the
+// cheap envelope signals (no body parse). Callers gate the full Parse on this.
+func Detect(from, subject, contentType string) bool {
+	f := strings.ToLower(from)
+	for _, m := range senderMarkers {
+		if strings.Contains(f, m) {
+			return true
+		}
+	}
+	if strings.Contains(strings.ToLower(contentType), "multipart/report") {
+		return true
+	}
+	s := strings.ToLower(subject)
+	for _, m := range subjectMarkers {
+		if strings.Contains(s, m) {
+			return true
+		}
+	}
+	return false
+}
+
+// Parse extracts bounce details from the message body (the DSN parts). It is
+// safe on non-DSN bodies: fields simply stay empty. Permanence is decided from
+// the machine-readable Status / Diagnostic-Code / Action fields, never from the
+// human-readable prose, so a "delayed" notice is never a permanent bounce.
+func Parse(body string) Report {
+	var r Report
+
+	if m := reStatus.FindStringSubmatch(body); m != nil {
+		r.IsBounce = true
+		r.Permanent = m[1] == "5"
+	}
+	if !r.Permanent {
+		if m := reDiagnostic.FindStringSubmatch(body); m != nil {
+			r.IsBounce = true
+			r.Permanent = m[1] == "5"
+		}
+	}
+	if a := reAction.FindStringSubmatch(body); a != nil {
+		r.IsBounce = true
+		if strings.EqualFold(a[1], "failed") && r.hasNo4xx(body) {
+			// "failed" is permanent per RFC 3464, but a co-present 4.x.x status
+			// (retry-then-fail) means transient — defer to the status code.
+			r.Permanent = true
+		}
+	}
+
+	if m := reFinalRcpt.FindStringSubmatch(body); m != nil {
+		r.FailedRecipient = strings.TrimSpace(m[1])
+	}
+	// The last Message-ID in the body belongs to the returned original message
+	// (the DSN's own id, if present, is a header, not in the body parts).
+	if ids := reMessageID.FindAllStringSubmatch(body, -1); len(ids) > 0 {
+		r.OriginalMessageID = strings.TrimSpace(ids[len(ids)-1][1])
+	}
+
+	return r
+}
+
+func (r Report) hasNo4xx(body string) bool {
+	if m := reStatus.FindStringSubmatch(body); m != nil {
+		return m[1] != "4"
+	}
+	return true
+}

internal/pkg/dsn/dsn_test.go

@@ -0,0 +1,72 @@
+package dsn
+
+import "testing"
+
+const permanentDSN = `This is the mail system at host mail.example.com.
+
+I'm sorry to have to inform you that your message could not
+be delivered to one or more recipients.
+
+Final-Recipient: rfc822; nobody@invalid.example.com
+Action: failed
+Status: 5.1.1
+Diagnostic-Code: smtp; 550 5.1.1 user unknown
+
+--- Original message headers ---
+From: sender@yourdomain.com
+Message-ID: <camp-abc-123@yourdomain.com>
+Subject: Quick question
+`
+
+const transientDSN = `Delivery is delayed.
+
+Final-Recipient: rfc822; busy@example.com
+Action: delayed
+Status: 4.4.1
+Message-ID: <camp-xyz-999@yourdomain.com>
+`
+
+func TestParsePermanent(t *testing.T) {
+	r := Parse(permanentDSN)
+	if !r.IsBounce || !r.Permanent {
+		t.Fatalf("expected permanent bounce, got %+v", r)
+	}
+	if r.FailedRecipient != "nobody@invalid.example.com" {
+		t.Errorf("failed recipient = %q", r.FailedRecipient)
+	}
+	if r.OriginalMessageID != "camp-abc-123@yourdomain.com" {
+		t.Errorf("original message id = %q", r.OriginalMessageID)
+	}
+}
+
+func TestParseTransientNotPermanent(t *testing.T) {
+	r := Parse(transientDSN)
+	if !r.IsBounce {
+		t.Fatal("expected a bounce")
+	}
+	if r.Permanent {
+		t.Error("4.x.x/delayed must not be permanent (would over-suppress)")
+	}
+}
+
+func TestParseOrdinaryBody(t *testing.T) {
+	r := Parse("Hi, thanks for reaching out, let's chat next week.")
+	if r.IsBounce || r.Permanent {
+		t.Errorf("ordinary mail misread as bounce: %+v", r)
+	}
+}
+
+func TestDetect(t *testing.T) {
+	if !Detect("Mail Delivery Subsystem <MAILER-DAEMON@google.com>", "", "") {
+		t.Error("mailer-daemon not detected")
+	}
+	if !Detect("x@y.com", "Undeliverable: Quick question", "") {
+		t.Error("bounce subject not detected")
+	}
+	if !Detect("x@y.com", "", "multipart/report; report-type=delivery-status") {
+		t.Error("multipart/report not detected")
+	}
+	if Detect("alice@example.com", "Re: your email", "text/plain") {
+		t.Error("ordinary reply misdetected as bounce")
+	}
+}

internal/repository/pg_warmup.go

@@ -1167,6 +1167,11 @@ func (r *warmupRepository) GetLatestReplyCandidate(ctx context.Context, senderAc
 		  AND t.status = 'completed'
 		  AND t.message_id <> ''
 		  AND t.completed_at IS NOT NULL
+		  -- Human reply timing: never reply to a message that just arrived (the
+		  -- recipient-side read engagement hasn't plausibly happened yet), and
+		  -- don't necro-reply to threads older than a week.
+		  AND t.completed_at < NOW() - INTERVAL '45 minutes'
+		  AND t.completed_at > NOW() - INTERVAL '7 days'
 		ORDER BY t.completed_at DESC
 		LIMIT 1
 	`

internal/scheduler/campaign_scheduler.go

@@ -2,6 +2,7 @@ package scheduler
 
 import (
 	"context"
+	"math/rand"
 	"time"
 
 	"github.com/google/uuid"
@@ -413,7 +414,10 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 			currentMinutes := nowLocal.Hour()*60 + nowLocal.Minute()
 			remainingMinutes := dayEnd - max(currentMinutes, dayStart)
 			if remainingMinutes > 0 {
-				idealInterval := time.Minute * time.Duration(remainingMinutes/remainingEmails)
+				// Vary the pace multiplicatively (bursts and lulls) — evenly
+				// metronomed sends are a pattern even with additive jitter.
+				varied := float64(remainingMinutes/remainingEmails) * (0.55 + rand.Float64()*0.9)
+				idealInterval := time.Duration(varied * float64(time.Minute))
 				minInterval := time.Second * time.Duration(account.MinWaitTime)
 				if idealInterval < minInterval {
 					idealInterval = minInterval
@@ -443,10 +447,10 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 		}
 	}
 
-	// STEP 11: Add jitter and round to nearest 5 minutes
+	// STEP 11: Add jitter. Deliberately NOT rounded to a 5-minute grid — a
+	// fleet that only ever sends at :x0/:x5 marks is a detectable pattern.
 	jitter := randomJitter(-20, 20)
 	candidateTime = candidateTime.Add(time.Minute * time.Duration(jitter))
-	candidateTime = roundToNearestMinute(candidateTime, 5)
 
 	// STEP 12: Check conflicts with other scheduled tasks
 	dateToCheck := candidateTime
@@ -464,7 +468,8 @@ func (s *schedulerService) CalculateNextCampaignTime(ctx context.Context, campai
 	// (jitter/conflict/distribution can push into a gap between intervals).
 	candidateTime = nextScheduleSlot(candidateTime, windows, campaignTZ)
 
-	return candidateTime, nextPair, account.ID, nil
+	// STEP 15: Randomise the sub-minute component so sends never land on :00.
+	return humanizeSeconds(candidateTime), nextPair, account.ID, nil
 }
 
 // deferToNextDay pushes a candidate time to the next valid campaign day within

internal/scheduler/helpers.go

@@ -158,7 +158,16 @@ func calculateFirstSlotTomorrowAt(timezone, startTime string) time.Time {
 	firstSlot := time.Date(tomorrow.Year(), tomorrow.Month(), tomorrow.Day(),
 		startMinutes/60, startMinutes%60, 0, 0, loc)
 	jitter := randomJitter(0, 60)
-	return firstSlot.Add(time.Minute * time.Duration(jitter))
+	return humanizeSeconds(firstSlot.Add(time.Minute * time.Duration(jitter)))
+}
+
+// humanizeSeconds randomises the sub-minute component of a scheduled time. All
+// jitter above is minute-granular and every time.Date construction lands on
+// second :00, so without this the whole platform sends at second zero — a
+// fleet-wide fingerprint in Received headers. Applied as the last step of the
+// schedulers.
+func humanizeSeconds(t time.Time) time.Time {
+	return t.Truncate(time.Minute).Add(time.Duration(rand.Intn(60)) * time.Second)
 }
 
 // avoidRoundTimes adds randomness to avoid exact round times (10:00, 11:00)

internal/scheduler/humanize_test.go

@@ -0,0 +1,55 @@
+package scheduler
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+func TestHumanizeSeconds_RandomisesSubMinute(t *testing.T) {
+	base := time.Date(2026, 1, 1, 10, 30, 0, 0, time.UTC)
+	sawNonZero := false
+	for i := 0; i < 100; i++ {
+		got := humanizeSeconds(base)
+		if !got.Truncate(time.Minute).Equal(base) {
+			t.Fatalf("humanizeSeconds changed the minute: %v", got)
+		}
+		if got.Second() != 0 {
+			sawNonZero = true
+		}
+	}
+	if !sawNonZero {
+		t.Error("humanizeSeconds never produced a non-zero second — fleet still sends at :00")
+	}
+}
+
+func TestDailyVolumeFactor_StableWithinDayAndRanged(t *testing.T) {
+	id := uuid.New()
+	day := time.Date(2026, 1, 15, 9, 0, 0, 0, time.UTC)
+
+	a := dailyVolumeFactor(id, day)
+	b := dailyVolumeFactor(id, day.Add(6*time.Hour)) // same calendar day
+	if a != b {
+		t.Errorf("factor not stable within a day: %v vs %v", a, b)
+	}
+
+	for i := 0; i < 500; i++ {
+		f := dailyVolumeFactor(uuid.New(), day)
+		if f < 0.55 || f > 1.10 {
+			t.Fatalf("factor out of expected range: %v", f)
+		}
+	}
+}
+
+func TestDailyVolumeFactor_VariesAcrossDays(t *testing.T) {
+	id := uuid.New()
+	seen := map[float64]bool{}
+	for d := 0; d < 30; d++ {
+		day := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC).AddDate(0, 0, d)
+		seen[dailyVolumeFactor(id, day)] = true
+	}
+	if len(seen) < 5 {
+		t.Errorf("daily factor barely varies across a month (%d distinct) — looks fixed", len(seen))
+	}
+}

internal/scheduler/service.go

@@ -127,7 +127,7 @@ func calculateFirstSlotTomorrow(timezone string) time.Time {
 
 	// Add random jitter
 	jitter := randomJitter(0, 60)
-	return firstSlot.Add(time.Minute * time.Duration(jitter))
+	return humanizeSeconds(firstSlot.Add(time.Minute * time.Duration(jitter)))
 }
 
 // randomJitter generates random jitter between min and max minutes
@@ -137,17 +137,3 @@ func randomJitter(min, max int) int {
 	}
 	return min + rand.Intn(max-min)
 }
-
-// roundToNearestMinute rounds time to nearest N minutes
-func roundToNearestMinute(t time.Time, minutes int) time.Time {
-	if minutes <= 0 {
-		return t
-	}
-
-	rounded := t.Truncate(time.Minute * time.Duration(minutes))
-	if t.Sub(rounded) >= time.Minute*time.Duration(minutes)/2 {
-		rounded = rounded.Add(time.Minute * time.Duration(minutes))
-	}
-
-	return rounded
-}

internal/scheduler/warmup_scheduler.go

@@ -2,6 +2,7 @@ package scheduler
 
 import (
 	"context"
+	"math/rand"
 	"time"
 
 	"github.com/google/uuid"
@@ -74,6 +75,31 @@ func warmupRampTarget(activelyWarming bool, base, increase, max, daysWarming int
 	return target
 }
 
+// dailyVolumeFactor returns a stable-per-day multiplier (0.75–1.10, with an
+// occasional lighter day near 0.6) for a mailbox's warmup volume. Real senders
+// don't send exactly N every day; a flat daily count is a pattern. Deterministic
+// on (accountID, localDate) so the scheduler's many intra-day recomputations
+// don't thrash the target.
+func dailyVolumeFactor(accountID uuid.UUID, day time.Time) float64 {
+	y, m, d := day.Date()
+	var seed uint64 = 1469598103934665603 // FNV-1a offset basis
+	mix := func(b []byte) {
+		for _, c := range b {
+			seed ^= uint64(c)
+			seed *= 1099511628211
+		}
+	}
+	id := accountID
+	mix(id[:])
+	mix([]byte{byte(y), byte(y >> 8), byte(m), byte(d)})
+
+	u := float64(seed%1000) / 1000.0 // stable [0,1)
+	if u < 0.15 {
+		return 0.60 // ~15% of days are lighter
+	}
+	return 0.75 + u*0.35 // 0.75–1.10
+}
+
 // resolveHealthState looks up the participant's current health state across
 // the warmup pools they may belong to. Returns Healthy if the account is not
 // in any pool or the lookup fails — failing open keeps warmup running rather
@@ -142,7 +168,7 @@ func (s *schedulerService) CalculateNextWarmupTime(ctx context.Context, accountI
 			firstSlot := time.Date(nextDay.Year(), nextDay.Month(), nextDay.Day(),
 				startMinutes/60, startMinutes%60, 0, 0, loc)
 			jitter := randomJitter(0, 60)
-			return firstSlot.Add(time.Minute * time.Duration(jitter)), nil
+			return humanizeSeconds(firstSlot.Add(time.Minute * time.Duration(jitter))), nil
 		}
 	}
 
@@ -154,6 +180,23 @@ func (s *schedulerService) CalculateNextWarmupTime(ctx context.Context, accountI
 	}
 	targetVolume := warmupRampTarget(activelyWarming, account.WarmupBase, account.WarmupIncrease, account.WarmupMax, daysWarming, inCampaign)
 
+	// Vary the day's target so a mailbox doesn't send an identical count every
+	// day. Deterministic per (account, local day) so it's stable across the
+	// day's reschedules. Actively-warming mailboxes keep a floor of WarmupBase.
+	if activelyWarming && targetVolume > 0 {
+		factor := dailyVolumeFactor(accountID, time.Now().In(loadLocation(account.Timezone)))
+		varied := int(float64(targetVolume)*factor + 0.5)
+		if varied < account.WarmupBase {
+			varied = account.WarmupBase
+		}
+		if varied < 1 {
+			varied = 1
+		}
+		if varied < targetVolume {
+			targetVolume = varied
+		}
+	}
+
 	// STEP 2.1: Cap per-mailbox volume to actual recipient capacity. The
 	// sender should not send multiple warmup messages to the same recipient
 	// in a single day just to hit an arbitrary target; that creates obvious
@@ -240,9 +283,13 @@ func (s *schedulerService) CalculateNextWarmupTime(ctx context.Context, accountI
 		earliestNext = now
 	}
 
-	// STEP 7: Add ideal interval to last email time
+	// STEP 7: Add ideal interval to last email time. The interval is varied
+	// multiplicatively (0.55x–1.45x) so the day reads as bursts and lulls
+	// rather than a metronome: perfectly even spacing is its own signature
+	// even with additive jitter on top.
 	if lastEmailTime != nil && idealIntervalHours > 0 {
-		idealNext := lastEmailTime.Add(time.Duration(idealIntervalHours * float64(time.Hour)))
+		varied := idealIntervalHours * (0.55 + rand.Float64()*0.9)
+		idealNext := lastEmailTime.Add(time.Duration(varied * float64(time.Hour)))
 		if idealNext.After(earliestNext) {
 			earliestNext = idealNext
 		}
@@ -288,5 +335,6 @@ func (s *schedulerService) CalculateNextWarmupTime(ctx context.Context, accountI
 		}
 	}
 
-	return candidateTime, nil
+	// Randomise the sub-minute component so warmup sends never land on :00.
+	return humanizeSeconds(candidateTime), nil
 }