From fa6c935b129695fbaf0e5f3be3e49c5cddf7d02e Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <meszmatew@gmail.com>
Date: Fri, 16 Oct 2026 06:59:40 +0200
Subject: [PATCH 1/2] feat: withdraw the reply intents and untouched follow-up
 tasks that mail from before its sender's contact produced through a one-time
 consumer cleanup, and credit an unthreaded contact message only to a campaign
 email sent before it arrived (#960)

---
 cmd/consumer/main.go                          |   1 +
 docs/content/docs/guides/contacts-crm.mdx     |   2 +
 docs/content/docs/guides/sequences.mdx        |   2 +
 internal/app/advanced/incoming_reply_test.go  |   4 +-
 .../app/advanced/reply_history_recheck.go     |  75 ++++++++++
 .../advanced/reply_history_recheck_test.go    | 133 +++++++++++++++++
 internal/app/advanced/service.go              |   7 +-
 internal/app/consumer/reply_repair.go         |  27 ++++
 .../contact_sent_count_live_test.go           |  19 ++-
 .../repository/historical_reply_live_test.go  | 135 ++++++++++++++++++
 internal/repository/pg_advanced_outreach.go   |  34 +++++
 internal/repository/pg_campaign_progress.go   |  14 +-
 internal/repository/pg_crm.go                 |  26 ++++
 internal/repository/pg_unibox.go              |   4 +
 internal/repository/pg_unibox_warmup.go       |  90 ++++++++++--
 15 files changed, 557 insertions(+), 16 deletions(-)
 create mode 100644 internal/app/advanced/reply_history_recheck.go
 create mode 100644 internal/app/advanced/reply_history_recheck_test.go
 create mode 100644 internal/repository/historical_reply_live_test.go

diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go
index 62a9390a1..f0bd527ee 100644
--- a/cmd/consumer/main.go
+++ b/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.
diff --git a/docs/content/docs/guides/contacts-crm.mdx b/docs/content/docs/guides/contacts-crm.mdx
index f01f6a29c..d1ceb1b34 100644
--- a/docs/content/docs/guides/contacts-crm.mdx
+++ b/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 withdraws those once: each reply intent such a message recorded is removed, and each follow-up task it opened is cancelled 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.
+
 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.
diff --git a/docs/content/docs/guides/sequences.mdx b/docs/content/docs/guides/sequences.mdx
index 563201b7f..4ded17ef4 100644
--- a/docs/content/docs/guides/sequences.mdx
+++ b/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.
diff --git a/internal/app/advanced/incoming_reply_test.go b/internal/app/advanced/incoming_reply_test.go
index 5c47c396a..957e0cc8d 100644
--- a/internal/app/advanced/incoming_reply_test.go
+++ b/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
 }
 
diff --git a/internal/app/advanced/reply_history_recheck.go b/internal/app/advanced/reply_history_recheck.go
new file mode 100644
index 000000000..548e7d929
--- /dev/null
+++ b/internal/app/advanced/reply_history_recheck.go
@@ -0,0 +1,75 @@
+package advanced
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/rs/zerolog/log"
+
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+// replyEffectWindow is how soon after a message is processed the intent and
+// follow-up task it produced are written.
+const replyEffectWindow = 2 * time.Minute
+
+// WithdrawHistoricalReplyEffects reads one page of mail that predates its
+// sender's contact and was still handled as a reply, deleting the intent each
+// one recorded and cancelling the follow-up task it opened while nobody has
+// touched it. Opt-outs are the reply opt-out recheck's.
+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 {
+			// Left for the next pass, which reads the same message again.
+			log.Warn().Err(err).Str("message_id", page.Replies[i].ID.String()).Msg("historical reply cleanup failed")
+		}
+	}
+	return page.NextCursor, page.Done, nil
+}
+
+func (s *service) withdrawHistoricalReply(ctx context.Context, h *repository.HistoricalReply) error {
+	// The subject pins an intent to its message; the window to its processing.
+	intents, err := s.repo.ListReplyIntentsFor(ctx, h.OrganizationID, h.Sender, h.Subject,
+		h.ProcessedAt.Add(-time.Minute), h.ProcessedAt.Add(replyEffectWindow))
+	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") && s.crmRepo != nil {
+			cancelled, err := s.crmRepo.CancelUntouchedCRMTasks(ctx, h.OrganizationID, h.ContactID, replyTaskTitle(in.Intent, in.ContactEmail),
+				h.ProcessedAt.Add(-time.Minute), h.ProcessedAt.Add(replyEffectWindow))
+			if err != nil {
+				return err
+			}
+			for _, id := range cancelled {
+				s.pushCRM(ctx, h.OrganizationID, models.CRMObjectTask, id)
+				if s.audit != nil {
+					s.audit.LogAction(ctx, h.OrganizationID, uuid.Nil, models.AuditActionUpdate, models.AuditEntityCRMTask, &id, "", "",
+						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 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
+}
diff --git a/internal/app/advanced/reply_history_recheck_test.go b/internal/app/advanced/reply_history_recheck_test.go
new file mode 100644
index 000000000..7ad4616eb
--- /dev/null
+++ b/internal/app/advanced/reply_history_recheck_test.go
@@ -0,0 +1,133 @@
+package advanced
+
+import (
+	"context"
+	"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
+}
+
+func (r *historyAdvancedRepo) ListReplyIntentsFor(_ context.Context, _ uuid.UUID, email, subject string, from, to time.Time) ([]models.ReplyIntentRecord, error) {
+	var out []models.ReplyIntentRecord
+	for _, in := range r.intents {
+		if in.ContactEmail == email && in.Metadata["subject"] == subject && !in.CreatedAt.Before(from) && !in.CreatedAt.After(to) {
+			out = append(out, in)
+		}
+	}
+	return out, nil
+}
+
+func (r *historyAdvancedRepo) DeleteReplyIntents(_ context.Context, _ uuid.UUID, ids []uuid.UUID) error {
+	r.deleted = append(r.deleted, ids...)
+	return nil
+}
+
+type historyCRMRepo struct {
+	repository.CRMRepository
+	titles []string
+}
+
+func (r *historyCRMRepo) CancelUntouchedCRMTasks(_ context.Context, _, _ uuid.UUID, title string, _, _ time.Time) ([]uuid.UUID, error) {
+	r.titles = append(r.titles, title)
+	return []uuid.UUID{uuid.New()}, 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()
+	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))
+	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: uuid.New(), 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 want := "Follow up: reply from lead@example.org"; len(crm.titles) != 1 || crm.titles[0] != want {
+		t.Errorf("cancelled tasks titled %v, want [%q]", crm.titles, want)
+	}
+	if len(audit.lifted) != 1 {
+		t.Errorf("audited %d cancellations, want 1", len(audit.lifted))
+	}
+}
+
+// An intent that opened no task cancels none.
+func TestWithdrawHistoricalReplyEffectsWithoutATask(t *testing.T) {
+	processed := time.Now()
+	orgID := 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"}}
+	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: uuid.New(), 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.titles) != 0 {
+		t.Errorf("deleted %v and cancelled %v, want the intent only", repo.deleted, crm.titles)
+	}
+}
+
+// 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)
+	}
+}
diff --git a/internal/app/advanced/service.go b/internal/app/advanced/service.go
index 36012d0b0..be337f636 100644
--- a/internal/app/advanced/service.go
+++ b/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)
 }
@@ -1384,7 +1388,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
diff --git a/internal/app/consumer/reply_repair.go b/internal/app/consumer/reply_repair.go
index 041ff398c..ab38588d8 100644
--- a/internal/app/consumer/reply_repair.go
+++ b/internal/app/consumer/reply_repair.go
@@ -93,3 +93,30 @@ 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)
+		afterID = next
+		finished = err == nil && done
+		return err
+	})
+}
diff --git a/internal/repository/contact_sent_count_live_test.go b/internal/repository/contact_sent_count_live_test.go
index e34eb3805..743774fdc 100644
--- a/internal/repository/contact_sent_count_live_test.go
+++ b/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)
+	}
 }
diff --git a/internal/repository/historical_reply_live_test.go b/internal/repository/historical_reply_live_test.go
new file mode 100644
index 000000000..390ffc017
--- /dev/null
+++ b/internal/repository/historical_reply_live_test.go
@@ -0,0 +1,135 @@
+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)
+		}
+	}
+	now := time.Now()
+	intents, err := adv.ListReplyIntentsFor(ctx, f.org, lead, "Big sale", now.Add(-time.Minute), now.Add(time.Minute))
+	if err != nil || len(intents) != 1 || intents[0].ActionTaken != "created_crm_task" || intents[0].Intent != models.ReplyIntentNeutral {
+		t.Fatalf("ListReplyIntentsFor = %+v, %v; want the newsletter's one intent", 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 != 1 {
+		t.Fatalf("intents left = %d, %v; want the fresh reply's", 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)
+	cancelled, err := crm.CancelUntouchedCRMTasks(ctx, f.org, contact, title, now.Add(-time.Minute), time.Now().Add(time.Minute))
+	if err != nil {
+		t.Fatalf("CancelUntouchedCRMTasks: %v", err)
+	}
+	if len(cancelled) != 1 || cancelled[0] != untouched {
+		t.Fatalf("cancelled %v, want only the untouched task %v (kept %v and %v)", cancelled, untouched, worked, edited)
+	}
+	if again, err := crm.CancelUntouchedCRMTasks(ctx, f.org, contact, title, now.Add(-time.Minute), time.Now().Add(time.Minute)); err != nil || len(again) != 0 {
+		t.Fatalf("second cancel = %v, %v; want nothing", again, err)
+	}
+}
diff --git a/internal/repository/pg_advanced_outreach.go b/internal/repository/pg_advanced_outreach.go
index 7dbdd829f..6995d7783 100644
--- a/internal/repository/pg_advanced_outreach.go
+++ b/internal/repository/pg_advanced_outreach.go
@@ -65,6 +65,10 @@ type AdvancedOutreachRepository interface {
 	MarkTaskDeadLetterReplayed(ctx context.Context, id uuid.UUID) error
 
 	CreateReplyIntent(ctx context.Context, record *models.ReplyIntentRecord) error
+	// ListReplyIntentsFor returns the intents recorded for one address and
+	// subject between from and to.
+	ListReplyIntentsFor(ctx context.Context, organizationID uuid.UUID, email, subject string, from, to time.Time) ([]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 +1259,36 @@ func (r *advancedOutreachRepository) CreateReplyIntent(ctx context.Context, reco
 	return err
 }
 
+func (r *advancedOutreachRepository) ListReplyIntentsFor(ctx context.Context, organizationID uuid.UUID, email, subject string, from, to time.Time) ([]models.ReplyIntentRecord, error) {
+	rows, err := resultDB(ctx, r.db).Query(ctx, `
+		SELECT id, organization_id, contact_email, intent, action_taken, created_at
+		FROM reply_intents
+		WHERE organization_id = $1 AND contact_email = lower(btrim($2))
+		  AND created_at BETWEEN $3 AND $4
+		  AND COALESCE(metadata->>'subject', '') = $5`, organizationID, email, from, to, subject)
+	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); 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 {
diff --git a/internal/repository/pg_campaign_progress.go b/internal/repository/pg_campaign_progress.go
index 17a754571..b7cc71a8d 100644
--- a/internal/repository/pg_campaign_progress.go
+++ b/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
 		}
diff --git a/internal/repository/pg_crm.go b/internal/repository/pg_crm.go
index 1cef70298..1cb89af63 100644
--- a/internal/repository/pg_crm.go
+++ b/internal/repository/pg_crm.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
@@ -56,6 +57,9 @@ 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
+	// CancelUntouchedCRMTasks cancels a contact's pending tasks with this
+	// title created between from and to that nobody has edited since.
+	CancelUntouchedCRMTasks(ctx context.Context, orgID, contactID uuid.UUID, title string, from, to time.Time) ([]uuid.UUID, 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 +1891,28 @@ func (r *crmRepository) DeleteTaskType(ctx context.Context, orgID, typeID uuid.U
 	return nil
 }
 
+func (r *crmRepository) CancelUntouchedCRMTasks(ctx context.Context, orgID, contactID uuid.UUID, title string, from, to time.Time) ([]uuid.UUID, error) {
+	rows, err := r.db.Query(ctx, `
+		UPDATE crm_tasks SET status = 'cancelled', updated_at = NOW()
+		WHERE organization_id = $1 AND contact_id = $2 AND lower(title) = lower($3)
+		  AND created_at BETWEEN $4 AND $5
+		  AND status = 'pending' AND completed_at IS NULL AND updated_at = created_at
+		RETURNING id`, orgID, contactID, title, from, to)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var ids []uuid.UUID
+	for rows.Next() {
+		var id uuid.UUID
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, rows.Err()
+}
+
 func (r *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 {
diff --git a/internal/repository/pg_unibox.go b/internal/repository/pg_unibox.go
index e13c37c31..dd3b65dcc 100644
--- a/internal/repository/pg_unibox.go
+++ b/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)
diff --git a/internal/repository/pg_unibox_warmup.go b/internal/repository/pg_unibox_warmup.go
index ad6d51212..b5b76afb3 100644
--- a/internal/repository/pg_unibox_warmup.go
+++ b/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,

From a4d9ca5f20ab894ca1a41814b2ffcea1ef5638dd Mon Sep 17 00:00:00 2001
From: Matthew Meszaros <matt@warmbly.com>
Date: Fri, 16 Oct 2026 07:20:52 +0000
Subject: [PATCH 2/2] feat: require exact message and CRM task provenance for
 historical reply cleanup, preserve unlinked legacy effects and retry failed
 pages idempotently

---
 docs/content/docs/guides/contacts-crm.mdx     |   2 +-
 .../app/advanced/reply_history_recheck.go     |  38 ++--
 .../advanced/reply_history_recheck_test.go    |  57 ++++--
 .../advanced/reply_history_regression_test.go | 169 ++++++++++++++++++
 internal/app/advanced/service.go              |  28 +--
 internal/app/consumer/reply_repair.go         |   7 +-
 .../repository/historical_reply_live_test.go  |  66 +++++--
 internal/repository/pg_advanced_outreach.go   |  18 +-
 internal/repository/pg_crm.go                 |  29 +--
 9 files changed, 325 insertions(+), 89 deletions(-)
 create mode 100644 internal/app/advanced/reply_history_regression_test.go

diff --git a/docs/content/docs/guides/contacts-crm.mdx b/docs/content/docs/guides/contacts-crm.mdx
index d1ceb1b34..215c3a20f 100644
--- a/docs/content/docs/guides/contacts-crm.mdx
+++ b/docs/content/docs/guides/contacts-crm.mdx
@@ -371,7 +371,7 @@ 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 withdraws those once: each reply intent such a message recorded is removed, and each follow-up task it opened is cancelled 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.
+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.
 
diff --git a/internal/app/advanced/reply_history_recheck.go b/internal/app/advanced/reply_history_recheck.go
index 548e7d929..9689dc034 100644
--- a/internal/app/advanced/reply_history_recheck.go
+++ b/internal/app/advanced/reply_history_recheck.go
@@ -2,6 +2,7 @@ package advanced
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 
@@ -12,14 +13,7 @@ import (
 	"github.com/warmbly/warmbly/internal/repository"
 )
 
-// replyEffectWindow is how soon after a message is processed the intent and
-// follow-up task it produced are written.
-const replyEffectWindow = 2 * time.Minute
-
-// WithdrawHistoricalReplyEffects reads one page of mail that predates its
-// sender's contact and was still handled as a reply, deleting the intent each
-// one recorded and cancelling the follow-up task it opened while nobody has
-// touched it. Opt-outs are the reply opt-out recheck's.
+// 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
@@ -30,32 +24,37 @@ func (s *service) WithdrawHistoricalReplyEffects(ctx context.Context, since time
 	}
 	for i := range page.Replies {
 		if err := s.withdrawHistoricalReply(ctx, &page.Replies[i]); err != nil {
-			// Left for the next pass, which reads the same message again.
 			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 {
-	// The subject pins an intent to its message; the window to its processing.
-	intents, err := s.repo.ListReplyIntentsFor(ctx, h.OrganizationID, h.Sender, h.Subject,
-		h.ProcessedAt.Add(-time.Minute), h.ProcessedAt.Add(replyEffectWindow))
+	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") && s.crmRepo != nil {
-			cancelled, err := s.crmRepo.CancelUntouchedCRMTasks(ctx, h.OrganizationID, h.ContactID, replyTaskTitle(in.Intent, in.ContactEmail),
-				h.ProcessedAt.Add(-time.Minute), h.ProcessedAt.Add(replyEffectWindow))
+		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
 			}
-			for _, id := range cancelled {
-				s.pushCRM(ctx, h.OrganizationID, models.CRMObjectTask, id)
+			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, &id, "", "",
+					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"})
 				}
@@ -63,6 +62,9 @@ func (s *service) withdrawHistoricalReply(ctx context.Context, h *repository.His
 		}
 		ids = append(ids, in.ID)
 	}
+	if len(ids) == 0 {
+		return nil
+	}
 	if err := s.repo.DeleteReplyIntents(ctx, h.OrganizationID, ids); err != nil {
 		return err
 	}
diff --git a/internal/app/advanced/reply_history_recheck_test.go b/internal/app/advanced/reply_history_recheck_test.go
index 7ad4616eb..ef8d5635a 100644
--- a/internal/app/advanced/reply_history_recheck_test.go
+++ b/internal/app/advanced/reply_history_recheck_test.go
@@ -2,6 +2,8 @@ package advanced
 
 import (
 	"context"
+	"errors"
+	"slices"
 	"testing"
 	"time"
 
@@ -21,14 +23,16 @@ func (u historyUnibox) ListHistoricalReplies(context.Context, time.Time, uuid.UU
 
 type historyAdvancedRepo struct {
 	*incomingReplyAdvancedRepo
-	intents []models.ReplyIntentRecord
-	deleted []uuid.UUID
+	intents      []models.ReplyIntentRecord
+	deleted      []uuid.UUID
+	failDelete   bool
+	failDeleteID uuid.UUID
 }
 
-func (r *historyAdvancedRepo) ListReplyIntentsFor(_ context.Context, _ uuid.UUID, email, subject string, from, to time.Time) ([]models.ReplyIntentRecord, error) {
+func (r *historyAdvancedRepo) ListReplyIntentsForMessage(_ context.Context, orgID, messageID uuid.UUID) ([]models.ReplyIntentRecord, error) {
 	var out []models.ReplyIntentRecord
 	for _, in := range r.intents {
-		if in.ContactEmail == email && in.Metadata["subject"] == subject && !in.CreatedAt.Before(from) && !in.CreatedAt.After(to) {
+		if in.OrganizationID == orgID && in.Metadata["unibox_message_id"] == messageID.String() {
 			out = append(out, in)
 		}
 	}
@@ -36,18 +40,31 @@ func (r *historyAdvancedRepo) ListReplyIntentsFor(_ context.Context, _ uuid.UUID
 }
 
 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
-	titles []string
+	cancelled  []uuid.UUID
+	failCancel bool
 }
 
-func (r *historyCRMRepo) CancelUntouchedCRMTasks(_ context.Context, _, _ uuid.UUID, title string, _, _ time.Time) ([]uuid.UUID, error) {
-	r.titles = append(r.titles, title)
-	return []uuid.UUID{uuid.New()}, nil
+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
@@ -56,12 +73,15 @@ func (r *historyCRMRepo) CancelUntouchedCRMTasks(_ context.Context, _, _ uuid.UU
 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))
 
@@ -71,7 +91,7 @@ func TestWithdrawHistoricalReplyEffects(t *testing.T) {
 	audit := &recordingAudit{}
 	next := uuid.New()
 	svc := &service{repo: repo, crmRepo: crm, audit: audit, uniboxRepo: historyUnibox{page: repository.HistoricalReplyPage{
-		Replies: []repository.HistoricalReply{{ID: uuid.New(), OrganizationID: orgID, Sender: "lead@example.org",
+		Replies: []repository.HistoricalReply{{ID: messageID, OrganizationID: orgID, Sender: "lead@example.org",
 			Subject: "Big sale", ContactID: contactID, ProcessedAt: processed}},
 		NextCursor: next, Done: true,
 	}}}
@@ -86,32 +106,39 @@ func TestWithdrawHistoricalReplyEffects(t *testing.T) {
 	if len(repo.deleted) != 1 || repo.deleted[0] != fromNewsletter.ID {
 		t.Fatalf("deleted intents %v, want only %v", repo.deleted, fromNewsletter.ID)
 	}
-	if want := "Follow up: reply from lead@example.org"; len(crm.titles) != 1 || crm.titles[0] != want {
-		t.Errorf("cancelled tasks titled %v, want [%q]", crm.titles, want)
+	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"}}
+		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: uuid.New(), OrganizationID: orgID, Sender: "lead@example.org",
+		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.titles) != 0 {
-		t.Errorf("deleted %v and cancelled %v, want the intent only", repo.deleted, crm.titles)
+	if len(repo.deleted) != 1 || len(crm.cancelled) != 0 {
+		t.Errorf("deleted %v and cancelled %v, want the intent only", repo.deleted, crm.cancelled)
 	}
 }
 
diff --git a/internal/app/advanced/reply_history_regression_test.go b/internal/app/advanced/reply_history_regression_test.go
new file mode 100644
index 000000000..7160f1797
--- /dev/null
+++ b/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)
+			}
+		})
+	}
+}
diff --git a/internal/app/advanced/service.go b/internal/app/advanced/service.go
index be337f636..420e3d909 100644
--- a/internal/app/advanced/service.go
+++ b/internal/app/advanced/service.go
@@ -1664,6 +1664,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.
@@ -1679,12 +1688,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"
+				}
 			}
 		}
 	}
@@ -1697,13 +1707,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
diff --git a/internal/app/consumer/reply_repair.go b/internal/app/consumer/reply_repair.go
index ab38588d8..b2333dc6c 100644
--- a/internal/app/consumer/reply_repair.go
+++ b/internal/app/consumer/reply_repair.go
@@ -115,8 +115,11 @@ func (s *JobsService) StartHistoricalReplyCleanup(ctx context.Context) {
 		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 = err == nil && done
-		return err
+		finished = done
+		return nil
 	})
 }
diff --git a/internal/repository/historical_reply_live_test.go b/internal/repository/historical_reply_live_test.go
index 390ffc017..c5892c58a 100644
--- a/internal/repository/historical_reply_live_test.go
+++ b/internal/repository/historical_reply_live_test.go
@@ -94,17 +94,28 @@ func TestLiveHistoricalReplyCleanupQueries(t *testing.T) {
 			t.Fatalf("CreateReplyIntent: %v", err)
 		}
 	}
-	now := time.Now()
-	intents, err := adv.ListReplyIntentsFor(ctx, f.org, lead, "Big sale", now.Add(-time.Minute), now.Add(time.Minute))
-	if err != nil || len(intents) != 1 || intents[0].ActionTaken != "created_crm_task" || intents[0].Intent != models.ReplyIntentNeutral {
-		t.Fatalf("ListReplyIntentsFor = %+v, %v; want the newsletter's one intent", intents, 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 != 1 {
-		t.Fatalf("intents left = %d, %v; want the fresh reply's", left, err)
+	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}
@@ -122,14 +133,47 @@ func TestLiveHistoricalReplyCleanupQueries(t *testing.T) {
 	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)
-	cancelled, err := crm.CancelUntouchedCRMTasks(ctx, f.org, contact, title, now.Add(-time.Minute), time.Now().Add(time.Minute))
+	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("CancelUntouchedCRMTasks: %v", err)
+		t.Fatalf("CancelUntouchedCRMTask: %v", err)
 	}
-	if len(cancelled) != 1 || cancelled[0] != untouched {
-		t.Fatalf("cancelled %v, want only the untouched task %v (kept %v and %v)", cancelled, untouched, worked, edited)
+	if !cancelled {
+		t.Fatalf("did not cancel the untouched task %v", untouched)
 	}
-	if again, err := crm.CancelUntouchedCRMTasks(ctx, f.org, contact, title, now.Add(-time.Minute), time.Now().Add(time.Minute)); err != nil || len(again) != 0 {
+	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)
+	}
+}
diff --git a/internal/repository/pg_advanced_outreach.go b/internal/repository/pg_advanced_outreach.go
index 6995d7783..c55b6aad3 100644
--- a/internal/repository/pg_advanced_outreach.go
+++ b/internal/repository/pg_advanced_outreach.go
@@ -65,9 +65,8 @@ type AdvancedOutreachRepository interface {
 	MarkTaskDeadLetterReplayed(ctx context.Context, id uuid.UUID) error
 
 	CreateReplyIntent(ctx context.Context, record *models.ReplyIntentRecord) error
-	// ListReplyIntentsFor returns the intents recorded for one address and
-	// subject between from and to.
-	ListReplyIntentsFor(ctx context.Context, organizationID uuid.UUID, email, subject string, from, to time.Time) ([]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
 
@@ -1259,13 +1258,14 @@ func (r *advancedOutreachRepository) CreateReplyIntent(ctx context.Context, reco
 	return err
 }
 
-func (r *advancedOutreachRepository) ListReplyIntentsFor(ctx context.Context, organizationID uuid.UUID, email, subject string, from, to time.Time) ([]models.ReplyIntentRecord, error) {
+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
+		SELECT id, organization_id, contact_email, intent, action_taken, created_at, metadata
 		FROM reply_intents
-		WHERE organization_id = $1 AND contact_email = lower(btrim($2))
-		  AND created_at BETWEEN $3 AND $4
-		  AND COALESCE(metadata->>'subject', '') = $5`, organizationID, email, from, to, subject)
+		WHERE organization_id = $1 AND metadata->>'unibox_message_id' = $2`, organizationID, messageID.String())
 	if err != nil {
 		return nil, err
 	}
@@ -1273,7 +1273,7 @@ func (r *advancedOutreachRepository) ListReplyIntentsFor(ctx context.Context, or
 	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); err != nil {
+		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)
diff --git a/internal/repository/pg_crm.go b/internal/repository/pg_crm.go
index 1cb89af63..023cd5cc0 100644
--- a/internal/repository/pg_crm.go
+++ b/internal/repository/pg_crm.go
@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"strings"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
@@ -57,9 +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
-	// CancelUntouchedCRMTasks cancels a contact's pending tasks with this
-	// title created between from and to that nobody has edited since.
-	CancelUntouchedCRMTasks(ctx context.Context, orgID, contactID uuid.UUID, title string, from, to time.Time) ([]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)
@@ -1891,26 +1889,15 @@ func (r *crmRepository) DeleteTaskType(ctx context.Context, orgID, typeID uuid.U
 	return nil
 }
 
-func (r *crmRepository) CancelUntouchedCRMTasks(ctx context.Context, orgID, contactID uuid.UUID, title string, from, to time.Time) ([]uuid.UUID, error) {
-	rows, err := r.db.Query(ctx, `
+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 lower(title) = lower($3)
-		  AND created_at BETWEEN $4 AND $5
-		  AND status = 'pending' AND completed_at IS NULL AND updated_at = created_at
-		RETURNING id`, orgID, contactID, title, from, to)
+		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 nil, err
+		return false, err
 	}
-	defer rows.Close()
-	var ids []uuid.UUID
-	for rows.Next() {
-		var id uuid.UUID
-		if err := rows.Scan(&id); err != nil {
-			return nil, err
-		}
-		ids = append(ids, id)
-	}
-	return ids, rows.Err()
+	return result.RowsAffected() == 1, nil
 }
 
 func (r *crmRepository) DeleteCRMTask(ctx context.Context, orgID, taskID uuid.UUID) error {