Merge pull request #988 from warmbly/fix/reply-processing-stale-stored-mail

feat: withdraw reply intents and untouched follow-up tasks left by mail from before its sender's contact, and credit unthreaded mail only to an earlier send (#960)

cmd/consumer/main.go

@@ -606,6 +606,7 @@ func main() {
 	// reply refused by a since-fixed check is still attributed to its lead.
 	go jobsService.StartIncomingReplyRepair(ctx)
 	go jobsService.StartReplyOptOutRecheck(ctx)
+	go jobsService.StartHistoricalReplyCleanup(ctx)
 
 	// Mirrors Warmbly Cloud's warmup verdicts onto the mailboxes it warms, so
 	// a cloud quarantine pauses their campaigns here too. No-op when unlinked.

docs/content/docs/guides/contacts-crm.mdx

@@ -371,6 +371,8 @@ Which replies get one is up to you, in **Settings > Sending > Reply follow-ups**
 
 Only mail that arrives after the sender became a contact opens a task. Importing a list or creating contacts through the API does not turn their older messages already in your mailboxes into replies; a reply in one of your campaign threads always counts.
 
+Before this rule, older mail could open tasks and record reply intents when its sender was imported. Warmbly checks those once per consumer start, but withdraws an effect only when its saved identifiers link it to the exact message and, for a follow-up, the exact task. Older records without those links are left alone, even if their sender, subject or timing matches. A linked task is cancelled only if it is still pending and nobody has edited it. Every cancellation is recorded in the audit log. A task someone already started, edited or completed is left as it is. Temporary cleanup errors are retried before the check advances.
+
 Automated replies are off deliberately. A vacation notice is not follow-up work, and a week of sending makes enough of them to bury the real replies. Tick **Automated** if you do want one per machine reply, or untick the human ones you do not need. Unticking everything is the same as turning the switch off.
 
 The classification does more than decide the task: an automated reply never counts as a reply for **stop on reply** or for a **replied** branch in a sequence, and its notification goes to the **Out-of-office detected** preference rather than **New reply**, so muting one does not mute the other.

docs/content/docs/guides/sequences.mdx

@@ -313,6 +313,8 @@ Only inbound mail from that campaign contact to the sending mailbox counts. A th
 
 When a campaign rotates between sender accounts, a real inbound reply still links to the campaign if the mail client references an earlier step from another mailbox in the same workspace.
 
+A new email from a contact that answers no thread of yours counts toward the last campaign email you sent them before it arrived. Mail that arrived before any of your emails to them counts toward none, so importing older mail never stops a sequence.
+
 A reply that lands in the mailbox a send's reply-to named counts the same way, although that mailbox never wrote to the contact. That is what a [shared reply inbox](/guides/mailboxes/#a-shared-reply-inbox) relies on: the reply stops the sequence, runs the reply branch, and is credited to the mailbox that sent the email.
 
 When an upgrade repairs a reply marker that was provably created by a synchronized Sent message, the contact resumes at the next due step and the campaign reply totals update with it. An upgrade also restores provable human replies that were received by one campaign mailbox but referenced a thread started by another mailbox in the workspace. Warmbly recalculates address verification from the corrected delivery, open, click, reply, and bounce evidence. It leaves ambiguous historical replies untouched.

internal/app/advanced/incoming_reply_test.go

@@ -191,6 +191,7 @@ type incomingReplyProgressRepo struct {
 	claims        int
 	completed     int
 	latest        *repository.CampaignSequencePair
+	latestBefore  time.Time
 	sourceInbound bool
 	sourceClaimed bool
 	replyAccepted bool
@@ -218,7 +219,8 @@ func (r *incomingReplyProgressRepo) CampaignContactSentFromAccount(context.Conte
 	return r.receivingSent, nil
 }
 
-func (r *incomingReplyProgressRepo) GetLatestCampaignSequenceForContact(context.Context, uuid.UUID) (*repository.CampaignSequencePair, error) {
+func (r *incomingReplyProgressRepo) GetLatestCampaignSequenceForContact(_ context.Context, _ uuid.UUID, receivedAt time.Time) (*repository.CampaignSequencePair, error) {
+	r.latestBefore = receivedAt
 	return r.latest, nil
 }
 

internal/app/advanced/reply_history_recheck.go

@@ -0,0 +1,77 @@
+package advanced
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// WithdrawHistoricalReplyEffects removes only effects with exact saved provenance.
+func (s *service) WithdrawHistoricalReplyEffects(ctx context.Context, since time.Time, afterID uuid.UUID) (uuid.UUID, bool, error) {
+	if s.uniboxRepo == nil {
+		return afterID, true, nil
+	}
+	page, err := s.uniboxRepo.ListHistoricalReplies(ctx, since, afterID)
+	if err != nil {
+		return afterID, false, err
+	}
+	for i := range page.Replies {
+		if err := s.withdrawHistoricalReply(ctx, &page.Replies[i]); err != nil {
+			log.Warn().Err(err).Str("message_id", page.Replies[i].ID.String()).Msg("historical reply cleanup failed")
+			return afterID, false, err
+		}
+	}
+	return page.NextCursor, page.Done, nil
+}
+
+func (s *service) withdrawHistoricalReply(ctx context.Context, h *repository.HistoricalReply) error {
+	intents, err := s.repo.ListReplyIntentsForMessage(ctx, h.OrganizationID, h.ID)
+	if err != nil || len(intents) == 0 {
+		return err
+	}
+	ids := make([]uuid.UUID, 0, len(intents))
+	for _, in := range intents {
+		if strings.Contains(in.ActionTaken, "created_crm_task") {
+			rawTaskID, _ := in.Metadata["crm_task_id"].(string)
+			taskID, err := uuid.Parse(rawTaskID)
+			if err != nil || taskID == uuid.Nil {
+				continue // Legacy tasks without an exact link must not be guessed.
+			}
+			if s.crmRepo == nil {
+				return fmt.Errorf("historical reply cleanup: CRM repository unavailable")
+			}
+			cancelled, err := s.crmRepo.CancelUntouchedCRMTask(ctx, h.OrganizationID, h.ContactID, taskID)
+			if err != nil {
+				return err
+			}
+			if cancelled {
+				s.pushCRM(ctx, h.OrganizationID, models.CRMObjectTask, taskID)
+				if s.audit != nil {
+					s.audit.LogAction(ctx, h.OrganizationID, uuid.Nil, models.AuditActionUpdate, models.AuditEntityCRMTask, &taskID, "", "",
+						map[string]string{"status": "cancelled"},
+						map[string]string{"reason": "opened for mail that arrived before its sender was a contact, which is not a reply"})
+				}
+			}
+		}
+		ids = append(ids, in.ID)
+	}
+	if len(ids) == 0 {
+		return nil
+	}
+	if err := s.repo.DeleteReplyIntents(ctx, h.OrganizationID, ids); err != nil {
+		return err
+	}
+	log.Info().
+		Str("organization_id", h.OrganizationID.String()).
+		Str("message_id", h.ID.String()).
+		Int("intents", len(ids)).
+		Msg("historical reply cleanup: withdrew what mail from before its contact produced")
+	return nil
+}

internal/app/advanced/reply_history_recheck_test.go

@@ -0,0 +1,160 @@
+package advanced
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+type historyUnibox struct {
+	repository.UniboxRepository
+	page repository.HistoricalReplyPage
+}
+
+func (u historyUnibox) ListHistoricalReplies(context.Context, time.Time, uuid.UUID) (repository.HistoricalReplyPage, error) {
+	return u.page, nil
+}
+
+type historyAdvancedRepo struct {
+	*incomingReplyAdvancedRepo
+	intents      []models.ReplyIntentRecord
+	deleted      []uuid.UUID
+	failDelete   bool
+	failDeleteID uuid.UUID
+}
+
+func (r *historyAdvancedRepo) ListReplyIntentsForMessage(_ context.Context, orgID, messageID uuid.UUID) ([]models.ReplyIntentRecord, error) {
+	var out []models.ReplyIntentRecord
+	for _, in := range r.intents {
+		if in.OrganizationID == orgID && in.Metadata["unibox_message_id"] == messageID.String() {
+			out = append(out, in)
+		}
+	}
+	return out, nil
+}
+
+func (r *historyAdvancedRepo) DeleteReplyIntents(_ context.Context, _ uuid.UUID, ids []uuid.UUID) error {
+	if r.failDelete && (r.failDeleteID == uuid.Nil || slices.Contains(ids, r.failDeleteID)) {
+		r.failDelete = false
+		return errors.New("temporary database failure")
+	}
+	r.deleted = append(r.deleted, ids...)
+	r.intents = slices.DeleteFunc(r.intents, func(in models.ReplyIntentRecord) bool { return slices.Contains(ids, in.ID) })
+	return nil
+}
+
+type historyCRMRepo struct {
+	repository.CRMRepository
+	cancelled  []uuid.UUID
+	failCancel bool
+}
+
+func (r *historyCRMRepo) CancelUntouchedCRMTask(_ context.Context, _, _, taskID uuid.UUID) (bool, error) {
+	if r.failCancel {
+		r.failCancel = false
+		return false, errors.New("temporary task cancellation failure")
+	}
+	if slices.Contains(r.cancelled, taskID) {
+		return false, nil
+	}
+	r.cancelled = append(r.cancelled, taskID)
+	return true, nil
+}
+
+// A stored newsletter handled as a reply once its sender was imported left an
+// intent and a follow-up task (#960). The cleanup withdraws exactly those: a
+// fresh reply from the same sender, or another message's intent, stays.
+func TestWithdrawHistoricalReplyEffects(t *testing.T) {
+	processed := time.Date(2026, 10, 12, 23, 36, 0, 0, time.UTC)
+	orgID, contactID := uuid.New(), uuid.New()
+	messageID, taskID := uuid.New(), uuid.New()
+	intent := func(subject, action string, at time.Time) models.ReplyIntentRecord {
+		return models.ReplyIntentRecord{ID: uuid.New(), OrganizationID: orgID, ContactEmail: "lead@example.org",
+			Intent: models.ReplyIntentNeutral, ActionTaken: action, CreatedAt: at,
+			Metadata: map[string]interface{}{"subject": subject}}
+	}
+	fromNewsletter := intent("Big sale", "created_crm_task", processed.Add(time.Second))
+	fromNewsletter.Metadata["unibox_message_id"] = messageID.String()
+	fromNewsletter.Metadata["crm_task_id"] = taskID.String()
+	fromFreshReply := intent("Re: Quick question", "created_crm_task", processed.Add(time.Second))
+	sameSubjectLater := intent("Big sale", "created_crm_task", processed.Add(time.Hour))
+
+	repo := &historyAdvancedRepo{incomingReplyAdvancedRepo: &incomingReplyAdvancedRepo{},
+		intents: []models.ReplyIntentRecord{fromNewsletter, fromFreshReply, sameSubjectLater}}
+	crm := &historyCRMRepo{}
+	audit := &recordingAudit{}
+	next := uuid.New()
+	svc := &service{repo: repo, crmRepo: crm, audit: audit, uniboxRepo: historyUnibox{page: repository.HistoricalReplyPage{
+		Replies: []repository.HistoricalReply{{ID: messageID, OrganizationID: orgID, Sender: "lead@example.org",
+			Subject: "Big sale", ContactID: contactID, ProcessedAt: processed}},
+		NextCursor: next, Done: true,
+	}}}
+
+	cursor, done, err := svc.WithdrawHistoricalReplyEffects(context.Background(), processed.AddDate(0, -3, 0), uuid.Nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if cursor != next || !done {
+		t.Errorf("cursor = %v done = %v, want %v and true", cursor, done, next)
+	}
+	if len(repo.deleted) != 1 || repo.deleted[0] != fromNewsletter.ID {
+		t.Fatalf("deleted intents %v, want only %v", repo.deleted, fromNewsletter.ID)
+	}
+	if len(crm.cancelled) != 1 || crm.cancelled[0] != taskID {
+		t.Errorf("cancelled tasks %v, want [%v]", crm.cancelled, taskID)
+	}
+	if len(audit.lifted) != 1 {
+		t.Errorf("audited %d cancellations, want 1", len(audit.lifted))
+	}
+	if _, _, err := svc.WithdrawHistoricalReplyEffects(context.Background(), processed, uuid.Nil); err != nil {
+		t.Fatal(err)
+	}
+	if len(crm.cancelled) != 1 || len(audit.lifted) != 1 || len(repo.deleted) != 1 {
+		t.Fatal("second cleanup repeated an effect")
+	}
+}
+
+// An intent that opened no task cancels none.
+func TestWithdrawHistoricalReplyEffectsWithoutATask(t *testing.T) {
+	processed := time.Now()
+	orgID := uuid.New()
+	messageID := uuid.New()
+	in := models.ReplyIntentRecord{ID: uuid.New(), OrganizationID: orgID, ContactEmail: "lead@example.org",
+		Intent: models.ReplyIntentAutomated, CreatedAt: processed, Metadata: map[string]interface{}{"subject": "Thanks", "unibox_message_id": messageID.String()}}
+	repo := &historyAdvancedRepo{incomingReplyAdvancedRepo: &incomingReplyAdvancedRepo{}, intents: []models.ReplyIntentRecord{in}}
+	crm := &historyCRMRepo{}
+	svc := &service{repo: repo, crmRepo: crm, uniboxRepo: historyUnibox{page: repository.HistoricalReplyPage{
+		Replies: []repository.HistoricalReply{{ID: messageID, OrganizationID: orgID, Sender: "lead@example.org",
+			Subject: "Thanks", ContactID: uuid.New(), ProcessedAt: processed}},
+		Done: true,
+	}}}
+	if _, _, err := svc.WithdrawHistoricalReplyEffects(context.Background(), processed.AddDate(0, -3, 0), uuid.Nil); err != nil {
+		t.Fatal(err)
+	}
+	if len(repo.deleted) != 1 || len(crm.cancelled) != 0 {
+		t.Errorf("deleted %v and cancelled %v, want the intent only", repo.deleted, crm.cancelled)
+	}
+}
+
+// Mail from a contact outside any thread is credited to the last send that
+// went out before it arrived, never to a later one.
+func TestUnthreadedReplyLooksUpTheSendBeforeIt(t *testing.T) {
+	received := time.Date(2026, 9, 1, 8, 0, 0, 0, time.UTC)
+	contact := &models.Contact{ID: uuid.New(), Email: "lead@example.org", CreatedAt: received.AddDate(0, -1, 0)}
+	svc, _, _, accountID := newOptOutService(contact)
+	progress := svc.campaignProgressRepo.(*incomingReplyProgressRepo)
+	msg := inbound(accountID, "Lead <lead@example.org>", "Hello again", "Are you still hiring?", nil)
+	msg.InternalDate = received
+	if xerr := svc.ProcessIncomingReply(context.Background(), accountID, msg); xerr != nil {
+		t.Fatal(xerr)
+	}
+	if !progress.latestBefore.Equal(received) {
+		t.Errorf("latest send looked up before %v, want %v", progress.latestBefore, received)
+	}
+}

internal/app/advanced/reply_history_regression_test.go

@@ -0,0 +1,169 @@
+package advanced
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func TestReview988PreservesFreshSameSubjectIntent(t *testing.T) {
+	processed := time.Now()
+	org, messageID := uuid.New(), uuid.New()
+	stale := models.ReplyIntentRecord{ID: uuid.New(), OrganizationID: org, ContactEmail: "lead@example.org", CreatedAt: processed,
+		Metadata: map[string]interface{}{"subject": "Hello", "unibox_message_id": messageID.String()}}
+	fresh := stale
+	fresh.ID = uuid.New()
+	fresh.CreatedAt = processed.Add(30 * time.Second)
+	fresh.Metadata = map[string]interface{}{"subject": "Hello", "unibox_message_id": uuid.NewString()}
+	repo := &historyAdvancedRepo{incomingReplyAdvancedRepo: &incomingReplyAdvancedRepo{}, intents: []models.ReplyIntentRecord{stale, fresh}}
+	svc := &service{repo: repo, uniboxRepo: historyUnibox{page: repository.HistoricalReplyPage{
+		Replies: []repository.HistoricalReply{{ID: messageID, OrganizationID: org, ContactID: uuid.New(), Sender: stale.ContactEmail, Subject: "Hello", ProcessedAt: processed}}, Done: true,
+	}}}
+	if _, _, err := svc.WithdrawHistoricalReplyEffects(context.Background(), processed.Add(-time.Hour), uuid.Nil); err != nil {
+		t.Fatal(err)
+	}
+	if len(repo.deleted) != 1 || repo.deleted[0] != stale.ID {
+		t.Fatalf("deleted %v, want only stale %v; fresh %v must remain", repo.deleted, stale.ID, fresh.ID)
+	}
+}
+
+func TestWithdrawHistoricalReplyEffectsSkipsUnattributableLegacy(t *testing.T) {
+	messageID, taskID := uuid.New(), uuid.New()
+	for _, tc := range []struct {
+		name          string
+		message, task any
+	}{
+		{"missing message", nil, taskID.String()},
+		{"invalid message", "unknown", taskID.String()},
+		{"zero message", uuid.Nil.String(), taskID.String()},
+		{"different message", uuid.NewString(), taskID.String()},
+		{"missing task", messageID.String(), nil},
+		{"invalid task", messageID.String(), "unknown"},
+		{"zero task", messageID.String(), uuid.Nil.String()},
+		{"non-string task", messageID.String(), 42},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			org := uuid.New()
+			repo := &historyAdvancedRepo{incomingReplyAdvancedRepo: &incomingReplyAdvancedRepo{}, intents: []models.ReplyIntentRecord{{
+				ID: uuid.New(), OrganizationID: org, ContactEmail: "lead@example.org", ActionTaken: "created_crm_task",
+				Metadata: map[string]interface{}{"subject": "Hello", "unibox_message_id": tc.message, "crm_task_id": tc.task},
+			}}}
+			crm := &historyCRMRepo{}
+			svc := &service{repo: repo, crmRepo: crm, uniboxRepo: historyUnibox{page: repository.HistoricalReplyPage{
+				Replies: []repository.HistoricalReply{{ID: messageID, OrganizationID: org, ContactID: uuid.New(), Sender: "lead@example.org", Subject: "Hello", ProcessedAt: time.Now()}}, Done: true,
+			}}}
+			if _, done, err := svc.WithdrawHistoricalReplyEffects(context.Background(), time.Now(), uuid.Nil); err != nil || !done {
+				t.Fatalf("legacy skip: done=%v, err=%v", done, err)
+			}
+			if len(repo.deleted) != 0 || len(crm.cancelled) != 0 {
+				t.Fatalf("unattributable effects changed: deleted=%v cancelled=%v", repo.deleted, crm.cancelled)
+			}
+		})
+	}
+}
+
+func TestReview988DoesNotFinishAfterCleanupWriteFailure(t *testing.T) {
+	for _, failTask := range []bool{false, true} {
+		t.Run(map[bool]string{false: "delete", true: "cancel"}[failTask], func(t *testing.T) {
+			org, messageID, taskID, after, next := uuid.New(), uuid.New(), uuid.New(), uuid.New(), uuid.New()
+			in := models.ReplyIntentRecord{ID: uuid.New(), OrganizationID: org, ActionTaken: "created_crm_task",
+				Metadata: map[string]interface{}{"unibox_message_id": messageID.String(), "crm_task_id": taskID.String()}}
+			repo := &historyAdvancedRepo{incomingReplyAdvancedRepo: &incomingReplyAdvancedRepo{}, intents: []models.ReplyIntentRecord{in}, failDelete: !failTask}
+			crm, audit := &historyCRMRepo{failCancel: failTask}, &recordingAudit{}
+			svc := &service{repo: repo, crmRepo: crm, audit: audit, uniboxRepo: historyUnibox{page: repository.HistoricalReplyPage{
+				Replies: []repository.HistoricalReply{{ID: messageID, OrganizationID: org, ContactID: uuid.New()}}, NextCursor: next, Done: true,
+			}}}
+			cursor, done, err := svc.WithdrawHistoricalReplyEffects(context.Background(), time.Now(), after)
+			if err == nil || done || cursor != after {
+				t.Fatalf("failed page advanced: cursor=%v done=%v err=%v", cursor, done, err)
+			}
+			cursor, done, err = svc.WithdrawHistoricalReplyEffects(context.Background(), time.Now(), cursor)
+			if err != nil || !done || cursor != next {
+				t.Fatalf("retry did not finish: cursor=%v done=%v err=%v", cursor, done, err)
+			}
+			if len(repo.deleted) != 1 || len(crm.cancelled) != 1 || len(audit.lifted) != 1 {
+				t.Fatalf("retry not idempotent: deleted=%v cancelled=%v audit=%v", repo.deleted, crm.cancelled, audit.lifted)
+			}
+		})
+	}
+}
+
+func TestWithdrawHistoricalReplyEffectsRetriesTheSamePage(t *testing.T) {
+	org, after, next := uuid.New(), uuid.New(), uuid.New()
+	repo := &historyAdvancedRepo{incomingReplyAdvancedRepo: &incomingReplyAdvancedRepo{}, failDelete: true}
+	page := repository.HistoricalReplyPage{NextCursor: next, Done: true}
+	for range 2 {
+		messageID := uuid.New()
+		page.Replies = append(page.Replies, repository.HistoricalReply{ID: messageID, OrganizationID: org, ContactID: uuid.New()})
+		repo.intents = append(repo.intents, models.ReplyIntentRecord{ID: uuid.New(), OrganizationID: org, ActionTaken: "created_crm_task",
+			Metadata: map[string]interface{}{"unibox_message_id": messageID.String(), "crm_task_id": uuid.NewString()}})
+	}
+	repo.failDeleteID = repo.intents[1].ID
+	crm, audit := &historyCRMRepo{}, &recordingAudit{}
+	svc := &service{repo: repo, crmRepo: crm, audit: audit, uniboxRepo: historyUnibox{page: page}}
+	cursor, done, err := svc.WithdrawHistoricalReplyEffects(context.Background(), time.Now(), after)
+	if err == nil || done || cursor != after || len(repo.deleted) != 1 {
+		t.Fatalf("partially failed page advanced: cursor=%v done=%v deleted=%v err=%v", cursor, done, repo.deleted, err)
+	}
+	cursor, done, err = svc.WithdrawHistoricalReplyEffects(context.Background(), time.Now(), cursor)
+	if err != nil || !done || cursor != next || len(repo.deleted) != 2 || len(crm.cancelled) != 2 || len(audit.lifted) != 2 {
+		t.Fatalf("page retry lost or repeated effects: cursor=%v done=%v deleted=%v cancelled=%v audit=%v err=%v", cursor, done, repo.deleted, crm.cancelled, audit.lifted, err)
+	}
+}
+
+type replyProvenanceRepo struct {
+	*incomingReplyAdvancedRepo
+	record *models.ReplyIntentRecord
+}
+
+func (r *replyProvenanceRepo) CreateReplyIntent(_ context.Context, record *models.ReplyIntentRecord) error {
+	r.record = record
+	return nil
+}
+
+type replyTaskProvenanceRepo struct {
+	repository.CRMRepository
+	taskID uuid.UUID
+	fail   bool
+}
+
+func (r replyTaskProvenanceRepo) CreateCRMTask(context.Context, uuid.UUID, uuid.UUID, *models.CreateCRMTask) (*models.CRMTask, error) {
+	if r.fail {
+		return nil, errors.New("temporary task creation failure")
+	}
+	return &models.CRMTask{ID: r.taskID}, nil
+}
+
+func TestProcessIncomingReplyRecordsExactEffectProvenance(t *testing.T) {
+	for _, failTask := range []bool{false, true} {
+		t.Run(map[bool]string{false: "created", true: "failed"}[failTask], func(t *testing.T) {
+			org, accountID, taskID := uuid.New(), uuid.New(), uuid.New()
+			account := &models.Email{ID: accountID, OrganizationID: &org, UserID: uuid.NewString(), Email: "sender@example.test"}
+			contact := &models.Contact{ID: uuid.New(), Email: "lead@example.org", CreatedAt: time.Now().Add(-time.Hour)}
+			svc, progress := newIncomingReplyService(account, contact, contact.ID)
+			repo := &replyProvenanceRepo{incomingReplyAdvancedRepo: progress.advanced}
+			svc.repo = repo
+			svc.crmRepo = replyTaskProvenanceRepo{taskID: taskID, fail: failTask}
+			msg := inbound(accountID, contact.Email, "Re: Hello", "Thanks for writing.", nil)
+			msg.InternalDate = time.Now()
+			if err := svc.ProcessIncomingReply(context.Background(), accountID, msg); err != nil {
+				t.Fatal(err)
+			}
+			if repo.record == nil || repo.record.Metadata["unibox_message_id"] != msg.ID.String() {
+				t.Fatalf("missing message provenance: %+v", repo.record)
+			}
+			if failTask {
+				if _, ok := repo.record.Metadata["crm_task_id"]; ok || repo.record.ActionTaken == "created_crm_task" {
+					t.Fatalf("failed task was recorded as created: %+v", repo.record)
+				}
+			} else if repo.record.Metadata["crm_task_id"] != taskID.String() || repo.record.ActionTaken != "created_crm_task" {
+				t.Fatalf("missing task provenance: %+v", repo.record)
+			}
+		})
+	}
+}

internal/app/advanced/service.go

@@ -193,6 +193,10 @@ type Service interface {
 	// RecheckReplyOptOuts re-reads one page of reply opt-outs under the
 	// current rules, lifting the ones no message from the sender supports.
 	RecheckReplyOptOuts(ctx context.Context, afterID uuid.UUID, limit int) (uuid.UUID, bool, error)
+	// WithdrawHistoricalReplyEffects pages mail handled as a reply though it
+	// arrived before its sender was a contact, and withdraws the reply intent
+	// and untouched follow-up task each one left.
+	WithdrawHistoricalReplyEffects(ctx context.Context, since time.Time, afterID uuid.UUID) (uuid.UUID, bool, error)
 	// WireAudit attaches the audit trail a lifted reply opt-out is recorded in.
 	WireAudit(a AuditLogger)
 }
@@ -1390,7 +1394,8 @@ func (s *service) ProcessIncomingReply(ctx context.Context, emailAccountID uuid.
 	}
 
 	if campaignID == nil && contactID != nil && !referencesCampaignThread {
-		latest, err := s.campaignProgressRepo.GetLatestCampaignSequenceForContact(ctx, *contactID)
+		// Only a send that went out before the message arrived can be what it answers.
+		latest, err := s.campaignProgressRepo.GetLatestCampaignSequenceForContact(ctx, *contactID, msg.InternalDate)
 		if err == nil && latest != nil {
 			campaignID = &latest.CampaignID
 			sequenceID = &latest.SequenceID
@@ -1665,6 +1670,15 @@ func (s *service) ProcessIncomingReply(ctx context.Context, emailAccountID uuid.
 		}
 	}
 
+	metadata := map[string]interface{}{
+		"subject":       msg.Subject,
+		"reply_class":   verdict.Class,
+		"classified_by": verdict.Source,
+	}
+	if msg.ID != uuid.Nil {
+		metadata["unibox_message_id"] = msg.ID.String()
+	}
+
 	// Per-intent, so an out-of-office or a bounce does not become a
 	// high-priority follow-up nobody asked for. The default set is human
 	// replies only; a workspace can add the automated ones back.
@@ -1680,12 +1694,13 @@ func (s *service) ProcessIncomingReply(ctx context.Context, emailAccountID uuid.
 				DueDate:    ptrTime(time.Now().UTC().Add(24 * time.Hour)),
 				AssignedTo: &owner,
 			}); terr == nil {
+				metadata["crm_task_id"] = task.ID.String()
 				s.pushCRM(ctx, *account.OrganizationID, models.CRMObjectTask, task.ID)
-			}
-			if actionTaken == "" {
-				actionTaken = "created_crm_task"
-			} else {
-				actionTaken += ",created_crm_task"
+				if actionTaken == "" {
+					actionTaken = "created_crm_task"
+				} else {
+					actionTaken += ",created_crm_task"
+				}
 			}
 		}
 	}
@@ -1698,13 +1713,7 @@ func (s *service) ProcessIncomingReply(ctx context.Context, emailAccountID uuid.
 		Intent:         intent,
 		Confidence:     confidence,
 		ActionTaken:    actionTaken,
-		Metadata: map[string]interface{}{
-			"subject": msg.Subject,
-			// What decided it, so an intent nobody expected can be explained
-			// without re-running the message through the classifier.
-			"reply_class":   verdict.Class,
-			"classified_by": verdict.Source,
-		},
+		Metadata:       metadata,
 	})
 
 	// Fan the classified reply out to customer webhooks + integration actions

internal/app/consumer/reply_repair.go

@@ -93,3 +93,33 @@ func (s *JobsService) StartReplyOptOutRecheck(ctx context.Context) {
 		return err
 	})
 }
+
+// historicalReplyCleanupWindow covers every message the repair sweep could
+// have handled as a reply before it learned to tell older mail apart.
+const historicalReplyCleanupWindow = 2 * replyRepairWindow
+
+// StartHistoricalReplyCleanup withdraws, once per start, the reply intents
+// and untouched follow-up tasks that mail from before its sender's contact
+// produced. Current processing leaves none, so a later pass finds nothing.
+func (s *JobsService) StartHistoricalReplyCleanup(ctx context.Context) {
+	if s.AdvancedService == nil {
+		return
+	}
+	since := time.Now().Add(-historicalReplyCleanupWindow)
+	var afterID uuid.UUID
+	finished := false
+	jobrun.Loop(ctx, "historical_reply_cleanup", time.Minute, true, func(ctx context.Context) error {
+		if finished {
+			return nil
+		}
+		batchCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+		defer cancel()
+		next, done, err := s.AdvancedService.WithdrawHistoricalReplyEffects(batchCtx, since, afterID)
+		if err != nil {
+			return err
+		}
+		afterID = next
+		finished = done
+		return nil
+	})
+}

internal/repository/contact_sent_count_live_test.go

@@ -102,11 +102,28 @@ func TestLiveSentCountsEmailStepsOnly(t *testing.T) {
 
 	// A reply with no thread to attribute it to lands on the last email, never
 	// on the action step after it.
-	latest, err := progress.GetLatestCampaignSequenceForContact(ctx, finished)
+	latest, err := progress.GetLatestCampaignSequenceForContact(ctx, finished, time.Time{})
 	if err != nil {
 		t.Fatalf("GetLatestCampaignSequenceForContact: %v", err)
 	}
 	if latest == nil || latest.SequenceID != email2 {
 		t.Errorf("latest step = %+v, want the second email %v", latest, email2)
 	}
+
+	// Mail that arrived between the two emails answers the first; mail from
+	// before either answers neither.
+	between, err := progress.GetLatestCampaignSequenceForContact(ctx, finished, time.Now().Add(-59*time.Minute))
+	if err != nil {
+		t.Fatalf("GetLatestCampaignSequenceForContact: %v", err)
+	}
+	if between == nil || between.SequenceID != email1 {
+		t.Errorf("step before the second email = %+v, want the first email %v", between, email1)
+	}
+	earlier, err := progress.GetLatestCampaignSequenceForContact(ctx, finished, time.Now().Add(-2*time.Hour))
+	if err != nil {
+		t.Fatalf("GetLatestCampaignSequenceForContact: %v", err)
+	}
+	if earlier != nil {
+		t.Errorf("mail from before any send credited %+v", earlier)
+	}
 }

internal/repository/historical_reply_live_test.go

@@ -0,0 +1,179 @@
+package repository
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/warmbly/warmbly/internal/models"
+)
+
+// The historical reply cleanup finds mail handled as a reply after its
+// sender became a contact though it arrived before, outside every campaign
+// thread, and withdraws only what that message left (#960).
+//
+//	WARMBLY_TEST_DB=postgres://warmbly:warmbly@localhost:15432/<db>?sslmode=disable \
+//	  go test ./internal/repository/ -run LiveHistoricalReply -v
+func TestLiveHistoricalReplyCleanupQueries(t *testing.T) {
+	handle, pool := liveContactDB(t)
+	f := newSharedOrgFixture(t, pool)
+	ctx := context.Background()
+	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)
+		}
+	}
+
+	mailbox := uuid.New()
+	exec(`INSERT INTO email_accounts (id, user_id, organization_id, email, name, signature_plain, signature_html, provider)
+	      VALUES ($1, $2, $3, $4, 'History', '', '', 'smtp_imap')`, mailbox, f.owner, f.org, "history-"+mailbox.String()[:6]+"@test.local")
+	lead := "lead-" + uuid.New().String()[:6] + "@example.org"
+	imported := time.Now().Add(-48 * time.Hour).UTC().Truncate(time.Second)
+	contact := uuid.New()
+	exec(`INSERT INTO contacts (id, user_id, organization_id, email, first_name, last_name, company, phone, custom_fields, created_at)
+	      VALUES ($1, $2, $3, $4, 'Lead', '', '', '', '{}', $5)`, contact, f.owner, f.org, lead, imported)
+	sendID := "<send-" + uuid.New().String()[:8] + "@test.local>"
+	exec(`INSERT INTO tasks (id, task_type, email_account_id, status, message_id) VALUES ($1, 'campaign', $2, 'completed', $3)`, uuid.New(), mailbox, sendID)
+
+	processed := imported.Add(12 * time.Hour)
+	message := func(from, subject string, received time.Time, inReplyTo []string, processedAt *time.Time) uuid.UUID {
+		id := uuid.New()
+		if inReplyTo == nil {
+			inReplyTo = []string{}
+		}
+		exec(`INSERT INTO unibox_emails (id, user_id, email_id, folder, provider_folder, from_addr, subject, in_reply_to, internal_date, campaign_reply_processed_at)
+		      VALUES ($1, $2, $3, 'inbox', 'inbox', ARRAY[$4], $5, $6, $7, $8)`, id, f.owner, mailbox, from, subject, inReplyTo, received, processedAt)
+		return id
+	}
+	newsletter := message("Lead ("+lead+")", "Big sale", imported.AddDate(0, -5, 0), nil, &processed)
+	message("Lead <"+lead+">", "Re: Quick question", imported.AddDate(0, -5, 0), []string{sendID}, &processed)
+	message("Lead <"+lead+">", "Fresh", imported.Add(time.Hour), nil, &processed)
+	message("Lead <"+lead+">", "Unclaimed", imported.AddDate(0, -5, 0), nil, nil)
+	message("Lead <"+lead+">", "Undated", time.Time{}, nil, &processed)
+	t.Cleanup(func() {
+		_, _ = pool.Exec(context.Background(), `DELETE FROM unibox_emails WHERE email_id = $1`, mailbox)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM tasks WHERE email_account_id = $1`, mailbox)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM email_accounts WHERE id = $1`, mailbox)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM reply_intents WHERE organization_id = $1`, f.org)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM crm_tasks WHERE contact_id = $1`, contact)
+		_, _ = pool.Exec(context.Background(), `DELETE FROM contacts WHERE id = $1`, contact)
+	})
+
+	unibox := &uniboxRepository{db: handle}
+	var found []HistoricalReply
+	cursor := uuid.Nil
+	for range 10_000 {
+		page, err := unibox.ListHistoricalReplies(ctx, time.Now().AddDate(0, 0, -1), cursor)
+		if err != nil {
+			t.Fatalf("ListHistoricalReplies: %v", err)
+		}
+		for _, h := range page.Replies {
+			if h.ContactID == contact {
+				found = append(found, h)
+			}
+		}
+		cursor = page.NextCursor
+		if page.Done {
+			break
+		}
+	}
+	if len(found) != 1 || found[0].ID != newsletter {
+		t.Fatalf("historical replies = %+v, want only the newsletter %v", found, newsletter)
+	}
+	if h := found[0]; h.OrganizationID != f.org || h.Sender != lead || h.Subject != "Big sale" || !h.ProcessedAt.Equal(processed) {
+		t.Errorf("historical reply read as %+v", h)
+	}
+
+	adv := &advancedOutreachRepository{db: pool}
+	for _, subject := range []string{"Big sale", "Fresh"} {
+		if err := adv.CreateReplyIntent(ctx, &models.ReplyIntentRecord{OrganizationID: f.org, ContactEmail: lead,
+			Intent: models.ReplyIntentNeutral, ActionTaken: "created_crm_task", Metadata: map[string]interface{}{"subject": subject}}); err != nil {
+			t.Fatalf("CreateReplyIntent: %v", err)
+		}
+	}
+	if intents, err := adv.ListReplyIntentsForMessage(ctx, f.org, newsletter); err != nil || len(intents) != 0 {
+		t.Fatalf("legacy intents matched without provenance: %+v, %v", intents, err)
+	}
+	if err := adv.CreateReplyIntent(ctx, &models.ReplyIntentRecord{OrganizationID: f.org, ContactEmail: lead,
+		Intent: models.ReplyIntentNeutral, Metadata: map[string]interface{}{"unibox_message_id": newsletter.String()}}); err != nil {
+		t.Fatal(err)
+	}
+	intents, err := adv.ListReplyIntentsForMessage(ctx, f.org, newsletter)
+	if err != nil || len(intents) != 1 || intents[0].Metadata["unibox_message_id"] != newsletter.String() {
+		t.Fatalf("ListReplyIntentsForMessage = %+v, %v; want the attributable intent", intents, err)
+	}
+	for _, scope := range [][2]uuid.UUID{{uuid.New(), newsletter}, {f.org, uuid.New()}, {f.org, uuid.Nil}} {
+		if intents, err := adv.ListReplyIntentsForMessage(ctx, scope[0], scope[1]); err != nil || len(intents) != 0 {
+			t.Fatalf("intents crossed scope %v: %+v, %v", scope, intents, err)
+		}
+	}
+	if err := adv.DeleteReplyIntents(ctx, f.org, []uuid.UUID{intents[0].ID}); err != nil {
+		t.Fatalf("DeleteReplyIntents: %v", err)
+	}
+	var left int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM reply_intents WHERE organization_id = $1`, f.org).Scan(&left); err != nil || left != 2 {
+		t.Fatalf("intents left = %d, %v; want both unattributable legacy intents", left, err)
+	}
+
+	crm := &crmRepository{db: pool}
+	title := "Follow up: reply from " + lead
+	task := func(status string) uuid.UUID {
+		created, err := crm.CreateCRMTask(ctx, f.org, f.owner, &models.CreateCRMTask{ContactID: &contact, Title: title, Priority: "high"})
+		if err != nil {
+			t.Fatalf("CreateCRMTask: %v", err)
+		}
+		if status != "pending" {
+			exec(`UPDATE crm_tasks SET status = $2::crm_task_status, updated_at = NOW() + interval '1 second' WHERE id = $1`, created.ID, status)
+		}
+		return created.ID
+	}
+	untouched, worked := task("pending"), task("in_progress")
+	edited := task("pending")
+	exec(`UPDATE crm_tasks SET description = 'called them', updated_at = NOW() + interval '1 second' WHERE id = $1`, edited)
+	for _, scope := range [][3]uuid.UUID{{uuid.New(), contact, untouched}, {f.org, uuid.New(), untouched}, {f.org, contact, worked}, {f.org, contact, edited}} {
+		if cancelled, err := crm.CancelUntouchedCRMTask(ctx, scope[0], scope[1], scope[2]); err != nil || cancelled {
+			t.Fatalf("changed out-of-scope or touched task %v: %v, %v", scope, cancelled, err)
+		}
+	}
+	cancelled, err := crm.CancelUntouchedCRMTask(ctx, f.org, contact, untouched)
+	if err != nil {
+		t.Fatalf("CancelUntouchedCRMTask: %v", err)
+	}
+	if !cancelled {
+		t.Fatalf("did not cancel the untouched task %v", untouched)
+	}
+	if again, err := crm.CancelUntouchedCRMTask(ctx, f.org, contact, untouched); err != nil || again {
+		t.Fatalf("second cancel = %v, %v; want nothing", again, err)
+	}
+}
+
+func TestReview988PreservesFreshDifferentSubjectTask(t *testing.T) {
+	_, pool := liveContactDB(t)
+	f := newSharedOrgFixture(t, pool)
+	ctx := context.Background()
+	crm := &crmRepository{db: pool}
+	create := func() *models.CRMTask {
+		t.Helper()
+		task, err := crm.CreateCRMTask(ctx, f.org, f.owner, &models.CreateCRMTask{
+			ContactID: &f.contact, Title: "Follow up: reply from lead@example.org", Priority: "high",
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return task
+	}
+	staleNewsletterTask, freshReplyTask := create(), create()
+	if cancelled, err := crm.CancelUntouchedCRMTask(ctx, f.org, f.contact, staleNewsletterTask.ID); err != nil || !cancelled {
+		t.Fatalf("attributed task cancellation = %v, %v", cancelled, err)
+	}
+	var freshStatus string
+	if err := pool.QueryRow(ctx, `SELECT status FROM crm_tasks WHERE id=$1`, freshReplyTask.ID).Scan(&freshStatus); err != nil {
+		t.Fatal(err)
+	}
+	if freshStatus != "pending" {
+		t.Fatalf("fresh reply task %v became %s", freshReplyTask.ID, freshStatus)
+	}
+}

internal/repository/pg_advanced_outreach.go

@@ -65,6 +65,9 @@ type AdvancedOutreachRepository interface {
 	MarkTaskDeadLetterReplayed(ctx context.Context, id uuid.UUID) error
 
 	CreateReplyIntent(ctx context.Context, record *models.ReplyIntentRecord) error
+	// ListReplyIntentsForMessage requires the stored message's exact provenance.
+	ListReplyIntentsForMessage(ctx context.Context, organizationID, messageID uuid.UUID) ([]models.ReplyIntentRecord, error)
+	DeleteReplyIntents(ctx context.Context, organizationID uuid.UUID, ids []uuid.UUID) error
 	CreatePreflightReport(ctx context.Context, report *models.PreflightReport) error
 
 	GetABVariantStats(ctx context.Context, campaignID uuid.UUID) ([]models.ABVariantStats, error)
@@ -1255,6 +1258,37 @@ func (r *advancedOutreachRepository) CreateReplyIntent(ctx context.Context, reco
 	return err
 }
 
+func (r *advancedOutreachRepository) ListReplyIntentsForMessage(ctx context.Context, organizationID, messageID uuid.UUID) ([]models.ReplyIntentRecord, error) {
+	if messageID == uuid.Nil {
+		return nil, nil
+	}
+	rows, err := resultDB(ctx, r.db).Query(ctx, `
+		SELECT id, organization_id, contact_email, intent, action_taken, created_at, metadata
+		FROM reply_intents
+		WHERE organization_id = $1 AND metadata->>'unibox_message_id' = $2`, organizationID, messageID.String())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []models.ReplyIntentRecord
+	for rows.Next() {
+		var rec models.ReplyIntentRecord
+		if err := rows.Scan(&rec.ID, &rec.OrganizationID, &rec.ContactEmail, &rec.Intent, &rec.ActionTaken, &rec.CreatedAt, &rec.Metadata); err != nil {
+			return nil, err
+		}
+		out = append(out, rec)
+	}
+	return out, rows.Err()
+}
+
+func (r *advancedOutreachRepository) DeleteReplyIntents(ctx context.Context, organizationID uuid.UUID, ids []uuid.UUID) error {
+	if len(ids) == 0 {
+		return nil
+	}
+	_, err := resultDB(ctx, r.db).Exec(ctx, `DELETE FROM reply_intents WHERE organization_id = $1 AND id = ANY($2)`, organizationID, ids)
+	return err
+}
+
 func (r *advancedOutreachRepository) CreatePreflightReport(ctx context.Context, report *models.PreflightReport) error {
 	checks, err := marshalJSON(report.Checks)
 	if err != nil {

internal/repository/pg_campaign_progress.go

@@ -266,7 +266,10 @@ type CampaignProgressRepository interface {
 	GetContactLastSequenceTime(ctx context.Context, contactID, campaignID uuid.UUID) (*time.Time, error)
 	CheckContactHasReplied(ctx context.Context, contactID, campaignID uuid.UUID) (bool, error)
 	CountEmailsSentTodayByOrganization(ctx context.Context, organizationID uuid.UUID) (int, error)
-	GetLatestCampaignSequenceForContact(ctx context.Context, contactID uuid.UUID) (*CampaignSequencePair, error)
+	// GetLatestCampaignSequenceForContact returns the contact's latest sent
+	// email step sent at or before receivedAt; a zero receivedAt is unknown and
+	// bounds nothing.
+	GetLatestCampaignSequenceForContact(ctx context.Context, contactID uuid.UUID, receivedAt time.Time) (*CampaignSequencePair, error)
 
 	// FindRoutedPairs selects the (contact, step) pairs that are due now by
 	// following each contact's step rules (the branching tree) rather than a
@@ -1295,18 +1298,23 @@ func (r *campaignProgressRepository) CountEmailsSentTodayByOrganization(ctx cont
 	return count, err
 }
 
-func (r *campaignProgressRepository) GetLatestCampaignSequenceForContact(ctx context.Context, contactID uuid.UUID) (*CampaignSequencePair, error) {
+func (r *campaignProgressRepository) GetLatestCampaignSequenceForContact(ctx context.Context, contactID uuid.UUID, receivedAt time.Time) (*CampaignSequencePair, error) {
+	var before *time.Time
+	if !receivedAt.IsZero() {
+		before = &receivedAt
+	}
 	query := `
 		SELECT campaign_id, sequence_id
 		FROM campaign_contact_progress p
 		WHERE contact_id = $1
 		  AND sent_at IS NOT NULL
+		  AND ($2::timestamptz IS NULL OR sent_at <= $2)
 		  AND ` + progressIsEmailStep("p") + `
 		ORDER BY sent_at DESC
 		LIMIT 1
 	`
 	out := &CampaignSequencePair{}
-	if err := resultDB(ctx, r.db).QueryRow(ctx, query, contactID).Scan(&out.CampaignID, &out.SequenceID); err != nil {
+	if err := resultDB(ctx, r.db).QueryRow(ctx, query, contactID, before).Scan(&out.CampaignID, &out.SequenceID); err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, nil
 		}

internal/repository/pg_crm.go

@@ -56,6 +56,8 @@ type CRMRepository interface {
 	TasksSummary(ctx context.Context, orgID uuid.UUID, filters models.SearchTasks) (*models.TasksSummary, error)
 	UpdateCRMTask(ctx context.Context, orgID, taskID uuid.UUID, data *models.UpdateCRMTask) (*models.CRMTask, error)
 	DeleteCRMTask(ctx context.Context, orgID, taskID uuid.UUID) error
+	// CancelUntouchedCRMTask cancels only the attributed task if nobody edited it.
+	CancelUntouchedCRMTask(ctx context.Context, orgID, contactID, taskID uuid.UUID) (bool, error)
 	BulkDeleteCRMTasks(ctx context.Context, orgID uuid.UUID, sel models.TaskSelection, cap int) (matched, affected int64, err error)
 	// SelectTaskIDs resolves a bulk selection to ids, at most cap+1 of them.
 	SelectTaskIDs(ctx context.Context, orgID uuid.UUID, sel models.TaskSelection, cap int) ([]uuid.UUID, error)
@@ -1887,6 +1889,17 @@ func (r *crmRepository) DeleteTaskType(ctx context.Context, orgID, typeID uuid.U
 	return nil
 }
 
+func (r *crmRepository) CancelUntouchedCRMTask(ctx context.Context, orgID, contactID, taskID uuid.UUID) (bool, error) {
+	result, err := r.db.Exec(ctx, `
+		UPDATE crm_tasks SET status = 'cancelled', updated_at = NOW()
+		WHERE organization_id = $1 AND contact_id = $2 AND id = $3
+		  AND status = 'pending' AND completed_at IS NULL AND updated_at = created_at`, orgID, contactID, taskID)
+	if err != nil {
+		return false, err
+	}
+	return result.RowsAffected() == 1, nil
+}
+
 func (r *crmRepository) DeleteCRMTask(ctx context.Context, orgID, taskID uuid.UUID) error {
 	cmd, err := r.db.Exec(ctx, `DELETE FROM crm_tasks WHERE organization_id = $1 AND id = $2`, orgID, taskID)
 	if err != nil {

internal/repository/pg_unibox.go

@@ -88,6 +88,10 @@ type UniboxRepository interface {
 	// processing never claimed and that look like campaign replies: they
 	// answer a campaign send, or come from one of the workspace's contacts.
 	ListUnprocessedCampaignReplies(ctx context.Context, since time.Time, afterID uuid.UUID, limit int) (CampaignReplyRepairPage, error)
+	// ListHistoricalReplies pages inbound mail reply processing handled after
+	// its sender became a contact, though it arrived before, outside any
+	// campaign thread.
+	ListHistoricalReplies(ctx context.Context, since time.Time, afterID uuid.UUID) (HistoricalReplyPage, error)
 	// ListInboundFrom returns a workspace's inbound mail from one address,
 	// newest first: the evidence behind an opt-out that address triggered.
 	ListInboundFrom(ctx context.Context, orgID uuid.UUID, address string, limit int) ([]InboundMessage, error)

internal/repository/pg_unibox_warmup.go

@@ -141,6 +141,13 @@ func (r *uniboxRepository) UpdatePendingEmail(ctx context.Context, userID, id uu
 	return true, tx.Commit(ctx)
 }
 
+// bareFrom is the sender in whichever form the sync stored it: "Name <addr>",
+// "Name (addr)" or bare.
+const bareFrom = `lower(COALESCE(
+		(regexp_match(COALESCE(u.from_addr[1], ''), '<([^<>]+)>\s*$'))[1],
+		(regexp_match(COALESCE(u.from_addr[1], ''), '\(([^()]+)\)\s*$'))[1],
+		u.from_addr[1]))`
+
 type CampaignReplyRepairPage struct {
 	Events     []models.JobEventNewEmail
 	NextCursor uuid.UUID
@@ -180,12 +187,6 @@ func (r *uniboxRepository) ListUnprocessedCampaignReplies(ctx context.Context, s
 		return page, nil
 	}
 
-	// The sender in whichever form the sync stored it: "Name <addr>",
-	// "Name (addr)" or bare.
-	const bareFrom = `lower(COALESCE(
-		(regexp_match(COALESCE(u.from_addr[1], ''), '<([^<>]+)>\s*$'))[1],
-		(regexp_match(COALESCE(u.from_addr[1], ''), '\(([^()]+)\)\s*$'))[1],
-		u.from_addr[1]))`
 	rows, err := r.db.Query(ctx, `
 		SELECT u.user_id, u.id, u.email_id, u.message_id, u.thread_id, u.flags, u.from_addr, u.to_addr, u.cc, u.bcc,
 		       u.reply_to, u.in_reply_to, u.subject, u.snippet, u.body_text, u.folder, u.provider_folder,
@@ -241,6 +242,79 @@ func (r *uniboxRepository) ListUnprocessedCampaignReplies(ctx context.Context, s
 	return page, nil
 }
 
+// HistoricalReply is a message that predates its sender's contact and was
+// still handled as a reply once the contact existed.
+type HistoricalReply struct {
+	ID             uuid.UUID
+	OrganizationID uuid.UUID
+	Sender         string
+	Subject        string
+	ContactID      uuid.UUID
+	ProcessedAt    time.Time
+}
+
+type HistoricalReplyPage struct {
+	Replies    []HistoricalReply
+	NextCursor uuid.UUID
+	Done       bool
+}
+
+func (r *uniboxRepository) ListHistoricalReplies(ctx context.Context, since time.Time, afterID uuid.UUID) (HistoricalReplyPage, error) {
+	const scanSize = 500
+	page := HistoricalReplyPage{NextCursor: afterID}
+	var scanned int
+	var lastScanned *uuid.UUID
+	if err := r.db.QueryRow(ctx, `SELECT COUNT(*), MAX(id::text)::uuid FROM (
+		SELECT u.id FROM unibox_emails u
+		WHERE u.id > $1 AND u.created_at >= $2 AND u.campaign_reply_processed_at IS NOT NULL
+		ORDER BY u.id LIMIT $3) s`, afterID, since, scanSize).Scan(&scanned, &lastScanned); err != nil {
+		return page, err
+	}
+	if scanned == 0 || lastScanned == nil {
+		page.Done = true
+		return page, nil
+	}
+	rows, err := r.db.Query(ctx, `
+		SELECT u.id, ea.organization_id, `+bareFrom+`, COALESCE(u.subject, ''), co.id, u.campaign_reply_processed_at
+		FROM unibox_emails u
+		JOIN email_accounts ea ON ea.id = u.email_id
+		JOIN contacts co ON co.organization_id = ea.organization_id AND lower(co.email) = `+bareFrom+`
+		WHERE u.id > $1 AND u.id <= $3
+		  AND u.created_at >= $2
+		  AND u.campaign_reply_processed_at IS NOT NULL
+		  AND u.folder NOT IN ('sent', 'drafts')
+		  AND u.provider_folder NOT IN ('sent', 'drafts')
+		  -- Received before the contact existed, handled after: a zero date is unknown and never matches.
+		  AND u.internal_date > '0001-01-02 00:00:00+00'::timestamptz
+		  AND u.internal_date < co.created_at
+		  AND co.created_at <= u.campaign_reply_processed_at
+		  AND NOT EXISTS (
+		    SELECT 1 FROM tasks t
+		    WHERE t.message_id <> '' AND t.task_type = 'campaign'
+		      AND t.message_id = ANY(ARRAY(
+		        SELECT v FROM unnest(u.in_reply_to) AS x,
+		          LATERAL (VALUES ('<' || btrim(x, '<>') || '>'), (btrim(x, '<>'))) AS f(v)))
+		  )
+		ORDER BY u.id`, afterID, since, *lastScanned)
+	if err != nil {
+		return page, err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var h HistoricalReply
+		if err := rows.Scan(&h.ID, &h.OrganizationID, &h.Sender, &h.Subject, &h.ContactID, &h.ProcessedAt); err != nil {
+			return page, err
+		}
+		page.Replies = append(page.Replies, h)
+	}
+	if err := rows.Err(); err != nil {
+		return page, err
+	}
+	page.NextCursor = *lastScanned
+	page.Done = scanned < scanSize
+	return page, nil
+}
+
 // InboundMessage is a stored inbound message and when reply processing
 // handled it (its arrival, for mail processing never claimed).
 type InboundMessage struct {
@@ -249,10 +323,6 @@ type InboundMessage struct {
 }
 
 func (r *uniboxRepository) ListInboundFrom(ctx context.Context, orgID uuid.UUID, address string, limit int) ([]InboundMessage, error) {
-	const bareFrom = `lower(COALESCE(
-		(regexp_match(COALESCE(u.from_addr[1], ''), '<([^<>]+)>\s*$'))[1],
-		(regexp_match(COALESCE(u.from_addr[1], ''), '\(([^()]+)\)\s*$'))[1],
-		u.from_addr[1]))`
 	rows, err := r.db.Query(ctx, `
 		SELECT u.id, u.email_id, u.message_id, u.thread_id, u.flags, u.from_addr, u.to_addr, u.cc, u.bcc,
 		       u.reply_to, u.in_reply_to, u.subject, u.snippet, u.body_text, u.folder, u.provider_folder,