Merge pull request #938 from warmbly/devin/1791655717-preserve-pending-send-conflicts

feat: preserve send-conflict holds and require local recovery authority

.github/workflows/ci.yml

@@ -232,7 +232,7 @@ jobs:
         env:
           WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
         run: |
-          go test -race -v -p 1 ./internal/repository ./internal/app/consumer ./internal/tasks ./internal/app/orgtransfer -run '^TestLive(Warmup(ReservationPreservesOriginalChargeDay|RefundCASProtectsConcurrentAndRestampedTasks|RefundAndResultRollbackTogether|ReplyRefundUsesChargedShapeAfterParentDeletion|ChargeAndRefundUseUTCInNonUTCSession|ChargeProofRollsBackWhenCounterWriteFails|CancellationPathsRefundOnlyProvenCharge|GenericCompletionDoesNotProveChargedAccounting|SendFailureGivesTheDayBack|SendGiveBackIsIdempotent|SafetyHoldReschedulesInsteadOfBlockingTheDispatcher|Recovery(RetiresExistingUnstartedDispatchOnce|HandlesPartialReservations|PreservesExecutionAndReceiptEvidence|KeepsOtherUnknownAndRestrictiveHolds|AndBeginCannotBothWin))|AppliedWarmupFailureRepairsOnlyRefundedChargeDeadLetters|GiveBackDailySend(ReversesADispatchedWarmupSend|LeavesOtherDaysAlone)|AccountDailyUsage(CountsWarmupTasksNotTheStatisticsRow|ScopesWarmupToTypeAndDay)|ImportWarmupStatisticsCannotGrantChargeRefundAuthority|Legacy(FailedStatusCannotProveNonExecution|JSONAndAvroFailurePersistConservativeAdmission)|AuthoritativeDeadLetterResultsReconcileWithoutReplay|DeadLetterResultEffectsAndSettlementRollBackTogether|Campaign(DeadLetterWorkerResultReleasesReservationOnce|ForeignKeyCleanupDoesNotProveWakeupOnly)|UserEmailDeadLetterUsesUnderlyingSendTaskType|UnknownSendOperatorResolutionReconcilesAndAudits|RecordedWorkerOutcomeSurvivesLostEvent|UnreservedLegacyFailedTaskRetainsUnknownHold|UnknownResolutionRejectsMissingEvidenceAndLiveExecution|SendRecoveryResolutionRequiresEvidenceAndRetainsUnknown)$' -count=1
+          go test -race -v -p 1 ./internal/repository ./internal/app/consumer ./internal/tasks ./internal/app/orgtransfer -run '^TestLive(Warmup(ReservationPreservesOriginalChargeDay|RefundCASProtectsConcurrentAndRestampedTasks|RefundAndResultRollbackTogether|ReplyRefundUsesChargedShapeAfterParentDeletion|ChargeAndRefundUseUTCInNonUTCSession|ChargeProofRollsBackWhenCounterWriteFails|CancellationPathsRefundOnlyProvenCharge|GenericCompletionDoesNotProveChargedAccounting|SendFailureGivesTheDayBack|SendGiveBackIsIdempotent|SafetyHoldReschedulesInsteadOfBlockingTheDispatcher|Recovery(RetiresExistingUnstartedDispatchOnce|HandlesPartialReservations|PreservesExecutionAndReceiptEvidence|KeepsOtherUnknownAndRestrictiveHolds|AndBeginCannotBothWin))|AppliedWarmupFailureRepairsOnlyRefundedChargeDeadLetters|GiveBackDailySend(ReversesADispatchedWarmupSend|LeavesOtherDaysAlone)|AccountDailyUsage(CountsWarmupTasksNotTheStatisticsRow|ScopesWarmupToTypeAndDay)|ImportWarmupStatisticsCannotGrantChargeRefundAuthority|Legacy(FailedStatusCannotProveNonExecution|JSONAndAvroFailurePersistConservativeAdmission)|AuthoritativeDeadLetterResultsReconcileWithoutReplay|DeadLetterResultEffectsAndSettlementRollBackTogether|Campaign(DeadLetterWorkerResultReleasesReservationOnce|ForeignKeyCleanupDoesNotProveWakeupOnly)|UserEmailDeadLetterUsesUnderlyingSendTaskType|UnknownSendOperatorResolutionReconcilesAndAudits|RecordedWorkerOutcomeSurvivesLostEvent|UnreservedLegacyFailedTaskRetainsUnknownHold|UnknownResolutionRejectsMissingEvidenceAndLiveExecution|SendRecoveryResolutionRequiresEvidenceAndRetainsUnknown|UnstartedDispatchRetainsDeferredOutcomeConflict)$' -count=1
           go test -race -v -p 1 ./internal/infrastructure/db -run '^TestLiveWarmupChargeUpgradeKeepsLegacyUnverifiedAndPreservesEvidence$' -count=1
 
       - name: Warmup filing provider-floor regressions (PostgreSQL)
@@ -243,6 +243,16 @@ jobs:
           go test -race -v -p 1 ./internal/repository -run '^TestLiveWarmupFilingAdmission(DeniesMissingOwnedRow|KeepsUntrackedLegacyAction)$' -count=1
           go test -race -v -p 1 ./internal/infrastructure/db -run '^TestLiveWarmupFilingProviderFloorUpgradePreservesLegacyLeasesAndRollback$' -count=1
 
+      - name: Authoritative send-conflict and historical-origin regressions (PostgreSQL)
+        env:
+          WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
+        run: |
+          go test -race -v -p 1 ./internal/infrastructure/db -run '^TestLiveSendRecoveryConflict(CLIStartup|UpgradePreservesHistoricalEvidence|Production284UpgradePreservesHistoricalEvidence|HistoricalOnlyDownRefuses|EmptyDownUp)$' -count=1
+          go test -race -v -p 1 ./internal/repository ./internal/app/orgtransfer -run '^TestLive(DefinitiveConflictDirectionsDuplicatesAndRestrictiveHolds|PriorDefinitiveConflictWithoutOperatorResolution|AuthenticationRepairRetainsUnresolvedOutcomeConflict|ConcurrentAuthenticationRepairRetainsNewOutcomeConflict|UnstartedDispatchRetainsDeferredOutcomeConflict|HistoricalConfirmationRequiresFreshOrIndependentAuthority|ExpandedConflictSchemaRequiresUpgradedReadersAndWriters|OrdinarySendResultsNeedNoRecoveryConfirmation|ImportSendRecoveryHistoryCannotReconcileUnknownOutcome|ArchiveSendRecoveryLocalOriginResets|ImportedHistoryDoesNotRepeatAuthoritativeWorkerResult)$' -count=1
+          go test -race -v -p 1 ./internal/app/consumer -run '^TestLiveDelayedPreparationDuplicateMustNotCreateDefinitiveConflict$' -count=1
+
+          go test -race -v -p 1 ./internal/app/consumer -run '^TestLiveAdvancedBounceReconciliationRollbackAndReplayWithOneConnection$' -count=1
+
       - name: Mailbox sync admission, replay, truthful state and upgrade regressions (PostgreSQL)
         env:
           WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable

docs/content/docs/development/instance-health.mdx

@@ -84,6 +84,10 @@ Worker heartbeat totals and up to 20 worker details share one PostgreSQL snapsho
 - A nullable count is **Not measured**, not zero. A successful zero means no matching rows in that metric's scope/window. Zero recent sends means **No matching recent evidence**, not healthy throughput. A completed task, heartbeat, low memory usage or sync tick does not prove delivery or recovery.
 - Recent/persistent failure evidence, recovery unverified, safety holds, backlog and stale telemetry remain separate conditions. An old failure without later confirmed send evidence is **Recovery unverified**, not proof that a new failure is happening now. Unknown warning observations such as missing assignment remain unknown, not fabricated provider outages.
 - Schedule waits, retry waits, budget waits, daily-limit waits and in-flight work are informational. **Next eligible** describes eligibility, not a promise of success. A safety hold is a restriction, not an outage; monitoring never clears it.
+- A declining unknown-hold count alone does not prove successful sends. Definitive outcomes or nonce-gated retirement of commands that never started can resolve an unknown, but cancellation, deferral and authorized repair keep the mailbox held when another unknown or unresolved outcome conflict remains. Compare task-scoped evidence, not only before-and-after totals.
+- Contradictory definitive results retain the original applied outcome and accounting, record a payload-free conflict outcome, and hold sending without replay. Automatic conflict evidence is separate from operator/provider confirmation. Duplicate results cannot undo a fresh local confirmation of that conflict. Repair and result processing share the same task and mailbox locks so a concurrent conflict cannot disappear during hold recalculation.
+- Workspace archives preserve recovery audit history but reset its local provenance. Imported or older audit history without explicit local provenance cannot execute an operator-confirmed result or lift a conflict hold. Independent authoritative worker results still reconcile unknown sends automatically; only genuinely unresolved evidence needs fresh local administrator/provider confirmation. Two contradictory definitive outcomes are not resolved by choosing whichever arrived last. An upgrade does not infer that historical confirmations originated locally. Recovery-audit rollback is refused while audit rows remain; roll forward instead of deleting evidence to force a downgrade.
+- Normal customer sending, mailbox connection and authentication flows gain no new confirmation step. Administrators handle exceptional uncertain outcomes through the existing scoped recovery API after checking provider evidence; customers do not need to read logs or supply technical task identifiers. Status toggles, credential repair and duplicate results are not shortcuts around another unresolved send. See [send-safety upgrade ordering](/development/send-safety/#conflict-audit-upgrade-ordering) before enabling the audit extension on a mixed-version backend fleet.
 - Positive broker lag/pending/ack-pending/redelivered counts appear prominently on Overview even though the backend deliberately labels them `unknown` with severity `info`. Kafka committed lag excludes uncommitted processing; JetStream pending and acknowledgement-pending are different measurements. Neither is provider-send throughput. Redelivered is not a lifetime retry rate; absent group members can occur during rebalancing.
 - Counts, affected mailboxes, affected organizations and rows with unknown organization are separate quantities. Metrics and per-worker scopes may overlap. The attention total counts recorded signals, not failed sends or unique customers; do not add metric counts together.
 - Query errors, permission denials, unsupported capability, missing dependencies/schema and timeouts do not produce an all-clear. No sources or no metrics means unknown coverage. Raw provider/connection errors, credentials, payloads and mailbox addresses are not rendered by the monitoring panel.

docs/content/docs/development/send-safety.mdx

@@ -127,6 +127,24 @@ The source-instance reconciliation outbox is excluded from organization archives
 
 Rollback refuses unresolved reservations, explicit participation state, diagnostic authorization/proof, pending outbox effects and repair history instead of discarding them. Earlier component migrations also refuse restrictive dispatch or managed consent history. Restore or complete a reviewed reconciliation; do not delete negative evidence simply to make downgrade succeed. Legacy deployments remain readable, but do not resume a mixed-old executor fleet assuming new guarantees.
 
+### Conflict-audit upgrade ordering
+
+The conflict-audit extension adds typed automatic contradiction evidence, local-record provenance and an explicit conflict-resolution projection. All historical rows remain intact, with no inferred provenance backfill. A historical confirmation without trustworthy local origin is history, not permission to apply an outcome or clear an unknown/conflict hold. Independent authoritative worker/provider results continue to reconcile unknown sends automatically. A genuine contradiction between definitive results preserves the original accounting and hold until a validated provider confirmation resolves the known conflict; arrival order and duplicate delivery do not decide the outcome.
+
+Normal customer sending, connection and authentication steps do not change. Exceptional uncertain history is an administrator recovery workflow using the existing `PATCH /emails/:id` recovery payload, workspace boundary and `manage_emails`/`WRITE_EMAILS` permissions. It does not add a customer log-reading or confirmation form. These are existing workspace/API permissions, not a new platform-admin-only authorization gate. Confirming an unknown outcome keeps admission held until the normal result reconciler commits accounting and hold transfer together. Authentication repair cannot resolve an unrelated outcome conflict.
+
+An additive schema alone is **not** a safe active mixed-version rollout. Old readers can still execute historical confirmations without checking provenance; old hold-clearing code cannot recognize automatic conflict rows. New binaries also require the added columns before using these paths. Use this ordering for a separately approved rollout:
+
+1. Preserve the database and audit evidence. Finish all predecessor migrations and review the allocated extension's up/down tests. Do not allocate a colliding migration or delete history to force rollback.
+2. Quiesce old dispatch/scheduling and all old processes that can apply send outcomes, mutate recovery holds, accept recovery confirmations or import workspace data. Preserve queued results and durable evidence. Merely disabling the admin confirmation screen or stopping one result consumer is insufficient.
+3. Apply the reviewed additive extension, then start only compatible backend, result-consumer, scheduler and import/repair binaries. Verify every control-plane replica, including Cloud and independently running consumers, before resuming these operations. Worker message/result formats are unchanged by this extension; it needs no new worker capability, credentials or customer configuration.
+4. Resume automatic stored-result reconciliation first, without resending tasks. Check task-scoped outcome, accounting, dead-letter and hold evidence. Remaining historical confirmations without local provenance require fresh administrator/provider confirmation only when independent authoritative evidence cannot resolve the unknown. Unknown outcomes and genuine conflicts remain restrictive.
+5. Keep the new schema and roll forward if a binary rollout fails. Recovery-audit down refuses whenever audit rows remain, including historical rows; older binaries must not resume recovery mutation against retained new evidence. A validated empty-audit downgrade still requires its matching compatible binaries.
+
+A refused downgrade preserves recovery rows, columns, constraints and indexes, but `golang-migrate` records the attempted down target as dirty before executing SQL. This is not a successful downgrade. Keep mutators quiesced and have an operator verify the retained schema and migration state before a reviewed roll-forward; do not blindly force a version, restart older binaries or delete evidence to bypass the refusal.
+
+This ordering is an operator deployment requirement, not permission to merge, deploy, replay, retire uncertain tasks or manually clear holds. A later applied send or failed/cancelled task is not recovery proof for a different held-task cohort. Local tests cannot establish that a provider executed a send.
+
 ## Verification and limits
 
 ### Research coverage and external gates

docs/content/docs/guides/workspace-export-import.mdx

@@ -95,6 +95,10 @@ The pending warmup-filing queue stays on its source instance. The destination re
 
 Warmup statistics travel as history. Per-task charge and refund markers reset on import so settling an old task cannot refund another send's destination statistics. Whole-instance database backups retain these markers.
 
+Send-recovery audit rows also travel as history, including operator confirmations and payload-free contradictory outcome metadata. Their local provenance resets on import. A source-instance confirmation cannot become an executable result or release a conflict hold on the destination. Unknown outcomes stay restrictive until independent authoritative worker results or a fresh local administrator/provider confirmation resolves them; importing an archive never authorizes a resend. Recovery uses available authoritative results automatically, without a new customer sending or connection step. Only unresolved evidence needs administrator verification, not customer log investigation.
+
+Move recovery history only to a destination with compatible recovery code and schema. An older destination can reject the new conflict evidence type or misinterpret historical confirmations even when it can read the archive. Do not discard audit rows to force an import or downgrade. See [conflict-audit upgrade ordering](/development/send-safety/#conflict-audit-upgrade-ordering).
+
 [Test workspace samples](/development/accounts-and-access/#adding-sample-data) move as ordinary contacts, campaigns and CRM records in their corresponding groups; reply templates are in **Assistant**. Their instance-wide admin audit marker does not travel, and import does not grant managed Test access.
 
 A restricted member stays restricted on the destination. Their campaign and folder grants travel with **Campaigns**, so an import without that group brings them in with no campaign access rather than more.

internal/app/consumer/advanced_reconciliation_live_test.go

@@ -27,6 +27,10 @@ func (d *resultDispatcher) Dispatch(context.Context, uuid.UUID, models.WebhookEv
 func TestLiveAdvancedBounceReconciliationRollbackAndReplayWithOneConnection(t *testing.T) {
 	h := liveDB(t)
 	f := newSendResultFixture(t, h)
+	if _, err := h.Pool.Exec(t.Context(), `INSERT INTO organization_members(organization_id,user_id,role,permissions,accepted_at)
+		SELECT id,owner_user_id,'owner',$2,NOW() FROM organizations WHERE id=$1`, f.org, models.PermViewAnalytics); err != nil {
+		t.Fatal(err)
+	}
 	s := liveJobsService(h)
 	task := f.stampSend(t, s)
 	cfg := h.Pool.Config()
@@ -51,7 +55,9 @@ func TestLiveAdvancedBounceReconciliationRollbackAndReplayWithOneConnection(t *t
 	svc.(interface {
 		WireDispatcher(advanced.EventDispatcher)
 	}).WireDispatcher(dispatcher)
-	svc.WireNotifier(notification.NewService(repository.NewNotificationRepository(pool), nil))
+	notifier := notification.NewService(repository.NewNotificationRepository(pool), nil)
+	notifier.WireDelivery(nil, nil, nil, repository.NewOrganizationRepository(pool))
+	svc.WireNotifier(notifier)
 	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
 	defer cancel()
 	request := &models.IngestDeliverabilityEventRequest{CampaignID: &f.campaign, TaskID: &task, ContactID: &f.contact, EventType: models.DeliverabilityEventBounce, Provider: "smtp_reject", RecipientEmail: "lead-" + f.contact.String()[:8] + "@test.local", Reason: "unknown recipient", IdempotencyKey: "refusal-" + task.String()}

internal/app/consumer/preparation_duplicate_live_test.go

@@ -0,0 +1,88 @@
+package jobs
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func TestLiveDelayedPreparationDuplicateMustNotCreateDefinitiveConflict(t *testing.T) {
+	h := liveDB(t)
+	f := newSendResultFixture(t, h)
+	s := liveJobsService(h)
+	s.WarmupRepo = repository.NewWarmupRepository(h.Pool)
+	s.InitEvents()
+	ctx := t.Context()
+	task, worker := uuid.New(), uuid.New()
+	exec := func(query string, args ...any) {
+		t.Helper()
+		if _, err := h.Exec(ctx, query, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol)VALUES($1,'worker',true,NOW(),1)`, worker)
+	exec(`INSERT INTO workers(id)VALUES($1)`, worker)
+	t.Cleanup(func() { _, _ = h.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+	exec(`UPDATE email_accounts SET worker_id=$1,test_mode='legacy' WHERE id=$2`, worker, f.mailbox)
+	exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id)VALUES($1,$2,'warmup','active','')`, task, f.mailbox)
+	exec(`INSERT INTO warmup_tasks(task_id,target_account_id,lineage_version)VALUES($1,$2,1)`, task, f.mailbox)
+	dispatch := s.TaskRepo.(repository.WarmupDispatchRepository)
+	nonce, err := dispatch.AuthorizeWarmupDispatch(ctx, task, f.mailbox, worker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := dispatch.DeferWarmupDispatch(ctx, task, f.mailbox, worker, nonce, time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	result := models.SendEmailResult{TaskID: task, SentAt: time.Now().UTC(), Error: &models.EmailSendError{Failure: &errx.SendFailure{
+		Protocol: "internal", Stage: "prepare", Disposition: errx.SendRetry, Scope: "mailbox", ObservedAt: time.Now().UTC(),
+	}}}
+	if err := s.HandleEvent(ctx, &models.JobEvent{Type: models.JobEventTypeEmailFailed, Body: result}); err != nil {
+		t.Fatal(err)
+	}
+	claim := s.TaskRepo.(interface {
+		ClaimWarmupTask(context.Context, uuid.UUID, time.Time) (bool, error)
+	})
+	if claimed, err := claim.ClaimWarmupTask(ctx, task, time.Now().Add(time.Second)); err != nil || !claimed {
+		t.Fatalf("claim=%v error=%v", claimed, err)
+	}
+	if next, err := dispatch.AuthorizeWarmupDispatch(ctx, task, f.mailbox, worker); err != nil || next == nonce {
+		t.Fatalf("new authorization nonce=%s error=%v", next, err)
+	}
+	if err := s.HandleEvent(ctx, &models.JobEvent{Type: models.JobEventTypeEmailFailed, Body: result}); err != nil {
+		t.Fatal(err)
+	}
+	var state, reason string
+	var applied, held, started bool
+	var audits int
+	if err := h.QueryRow(ctx, `SELECT t.send_result_state,t.send_result_applied_at IS NOT NULL,ea.send_recovery_hold,
+		COALESCE(ea.send_recovery_reason,''),w.dispatch_started_at IS NOT NULL,
+		(SELECT count(*) FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id AND r.evidence_type='authoritative_result_conflict')
+		FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, task).Scan(&state, &applied, &held, &reason, &started, &audits); err != nil {
+		t.Fatal(err)
+	}
+	if state != "unknown" || !applied || !held || reason != "unknown" || started || audits != 0 {
+		t.Errorf("delayed preparation duplicate after new authorization: state=%s applied=%v held=%v reason=%s started=%v automatic audits=%d", state, applied, held, reason, started, audits)
+	}
+	recovery := s.TaskRepo.(interface {
+		RecoverUnstartedWarmupDispatches(context.Context, time.Time, int) (int, error)
+	})
+	if n, err := recovery.RecoverUnstartedWarmupDispatches(ctx, time.Now().Add(time.Second), 1); err != nil || n != 1 {
+		t.Fatalf("positively nonce-gated unstarted retirement: retired=%d error=%v", n, err)
+	}
+	var status string
+	var sent int
+	if err := h.QueryRow(ctx, `SELECT t.status,ea.send_recovery_hold,COALESCE(ea.send_recovery_reason,''),ws.emails_sent,
+		(SELECT count(*) FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id AND r.evidence_type='authoritative_result_conflict')
+		FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id JOIN warmup_statistics ws ON ws.email_account_id=ea.id AND ws.date=CURRENT_DATE WHERE t.id=$1`, task).Scan(&status, &held, &reason, &sent, &audits); err != nil {
+		t.Fatal(err)
+	}
+	if status != "cancelled" || held || sent != 0 || audits != 0 {
+		t.Fatalf("after safe unstarted retirement: status=%s held=%v reason=%s charged=%d automatic audits=%d", status, held, reason, sent, audits)
+	}
+}

internal/app/orgtransfer/import.go

@@ -252,6 +252,14 @@ func (s *service) importTable(
 	for _, c := range t.ResetOnImport {
 		reset[c] = true
 	}
+	var resetOnOverwrite []string
+	if t.ResetOnOverwrite {
+		for _, c := range t.ResetOnImport {
+			if writable[c] {
+				resetOnOverwrite = append(resetOnOverwrite, c)
+			}
+		}
+	}
 
 	insertCols := make([]string, 0, len(mt.Columns))
 	unknown := 0
@@ -320,7 +328,7 @@ func (s *service) importTable(
 		if foreign > 0 {
 			return fmt.Errorf("%s: %w", t.Name, ErrForeignRows)
 		}
-		n, err := s.repo.InsertBatch(ctx, tx, t.Name, insertCols, batch, ic.conflict, pk, t.OwnerScope(), orgID)
+		n, err := s.repo.InsertBatch(ctx, tx, t.Name, insertCols, batch, ic.conflict, pk, t.OwnerScope(), orgID, resetOnOverwrite)
 		if err != nil {
 			return err
 		}

internal/app/orgtransfer/send_recovery_live_test.go

@@ -0,0 +1,382 @@
+package orgtransfer
+
+import (
+	"archive/zip"
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func TestLiveArchiveSendRecoveryLocalOriginResets(t *testing.T) {
+	svc, pool := liveImportService(t)
+	f := newTenantFixture(t, pool)
+	ctx := t.Context()
+	mailbox := uuid.New()
+	if _, err := pool.Exec(ctx, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider)
+		VALUES($1,$2,$3,$4,'Archive','','','smtp_imap')`, mailbox, f.owner, f.org, mailbox.String()+"@example.test"); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM email_accounts WHERE id=$1`, mailbox) })
+	var heldTask uuid.UUID
+	for _, item := range []struct{ evidence, state, contrary string }{
+		{"operator_confirmed_sent", "unknown", "failed"},
+		{"operator_confirmed_not_sent", "unknown", "sent"},
+		{"authoritative_result_conflict", "sent", "failed"},
+		{"authoritative_result_conflict", "failed", "sent"},
+	} {
+		task := uuid.New()
+		if heldTask == uuid.Nil {
+			heldTask = task
+		}
+		if _, err := pool.Exec(ctx, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_result_applied_at)
+			VALUES($1,$2,'email','completed','',$3,CASE WHEN $3<>'unknown' THEN NOW() END)`, task, mailbox, item.state); err != nil {
+			t.Fatal(err)
+		}
+		reason, reference := "conflict", ""
+		var confirmed any
+		if item.state == "unknown" {
+			reason, reference = "unknown", "Source confirmation fixture"
+			result := models.SendEmailResult{TaskID: task, Success: item.evidence == "operator_confirmed_sent"}
+			if !result.Success {
+				result.Error = &models.EmailSendError{Code: "RECIPIENT_REJECTED"}
+			}
+			raw, err := json.Marshal(result)
+			if err != nil {
+				t.Fatal(err)
+			}
+			confirmed = raw
+		}
+		if _, err := pool.Exec(ctx, `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmation_reference,confirmed_result,conflicting_result_state,recorded_locally_at,conflict_detected_at,conflict_resolved_at)
+			VALUES($1,$2,$3,$4,$5,$6,$7,$8,'2099-01-01Z','2099-01-01Z','2099-01-01Z')`, f.org, mailbox, task, reason, item.evidence, reference, confirmed, item.contrary); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := pool.Exec(ctx, `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, mailbox, heldTask); err != nil {
+		t.Fatal(err)
+	}
+	groups := []models.OrgDataGroup{models.OrgDataGroupCore, models.OrgDataGroupSending}
+	var archive bytes.Buffer
+	if _, err := svc.ExportTo(ctx, f.org, ExportOptions{Groups: groups}, &archive, nil); err != nil {
+		t.Fatal(err)
+	}
+	zr, err := zip.NewReader(bytes.NewReader(archive.Bytes()), int64(archive.Len()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	exported := 0
+	for _, entry := range zr.File {
+		if entry.Name != dataPath("send_recovery_resolutions") {
+			continue
+		}
+		reader, err := entry.Open()
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer reader.Close()
+		decoder := json.NewDecoder(reader)
+		for {
+			var row map[string]json.RawMessage
+			if err := decoder.Decode(&row); err == io.EOF {
+				break
+			} else if err != nil {
+				t.Fatal(err)
+			}
+			if string(row["recorded_locally_at"]) == "null" || string(row["conflict_resolved_at"]) == "null" {
+				t.Fatal("export destroyed source evidence rather than import resetting local authority")
+			}
+			exported++
+		}
+	}
+	if exported != 4 {
+		t.Fatalf("native export lost recovery audit: rows=%d", exported)
+	}
+	if _, err := svc.ImportFrom(ctx, f.org, bytes.NewReader(archive.Bytes()), ImportOptions{Groups: groups, Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.owner}, nil); err != nil {
+		t.Fatal(err)
+	}
+	var count, reset, confirmations, automatic int
+	if err := pool.QueryRow(ctx, `SELECT count(*),count(*) FILTER(WHERE recorded_locally_at IS NULL AND conflict_resolved_at IS NULL AND conflict_detected_at='2099-01-01Z'),
+		count(*) FILTER(WHERE confirmed_result IS NOT NULL AND confirmation_reference='Source confirmation fixture'),
+		count(*) FILTER(WHERE evidence_type='authoritative_result_conflict' AND conflicting_result_state IN('sent','failed') AND confirmed_result IS NULL AND evidence_task_id IS NULL AND confirmation_reference='')
+		FROM send_recovery_resolutions WHERE organization_id=$1`, f.org).Scan(&count, &reset, &confirmations, &automatic); err != nil || count != 4 || reset != 4 || confirmations != 2 || automatic != 2 {
+		t.Errorf("archive lost history or retained source authority: count=%d reset=%d confirmed=%d automatic=%d error=%v", count, reset, confirmations, automatic, err)
+	}
+	results, err := repository.NewTaskRepository(pool).(interface {
+		ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error)
+	}).ListUnappliedSendResults(ctx, 200)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, result := range results {
+		var ours bool
+		if err := pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM tasks WHERE id=$1 AND email_account_id=$2)`, result.TaskID, mailbox).Scan(&ours); err != nil || ours {
+			t.Fatalf("archive confirmation became executable: ours=%v error=%v", ours, err)
+		}
+	}
+	var held bool
+	var heldAfter uuid.UUID
+	if err := pool.QueryRow(ctx, `SELECT send_recovery_hold,send_recovery_task_id FROM email_accounts WHERE id=$1`, mailbox).Scan(&held, &heldAfter); err != nil || !held || heldAfter != heldTask {
+		t.Fatalf("archive roundtrip released restrictive hold: held=%v error=%v", held, err)
+	}
+	var audit uuid.UUID
+	if err := pool.QueryRow(ctx, `SELECT id FROM send_recovery_resolutions WHERE recovery_task_id=$1`, heldTask).Scan(&audit); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE send_recovery_resolutions SET recorded_locally_at=NOW(),conflict_resolved_at=NOW() WHERE id=$1`, audit); err != nil {
+		t.Fatal(err)
+	}
+	for _, conflict := range []models.OrgImportConflict{models.OrgImportConflictSkip, models.OrgImportConflictOverwrite} {
+		t.Run("old_archive_"+string(conflict), func(t *testing.T) {
+			oldArchive := tenantArchive(t, map[string][]map[string]any{"send_recovery_resolutions": {{
+				"id": audit, "organization_id": f.org, "email_account_id": mailbox, "recovery_task_id": heldTask,
+				"previous_reason": "unknown", "evidence_type": "operator_confirmed_not_sent", "confirmation_reference": "Older archive fixture",
+				"confirmed_result": map[string]any{"task_id": heldTask, "success": false, "error": map[string]any{"code": "RECIPIENT_REJECTED"}},
+			}}})
+			if _, err := svc.ImportFrom(ctx, f.org, oldArchive, ImportOptions{Conflict: conflict, ActorUserID: f.owner}, nil); err != nil {
+				t.Fatal(err)
+			}
+			var local, resolved, success bool
+			if err := pool.QueryRow(ctx, `SELECT recorded_locally_at IS NOT NULL,conflict_resolved_at IS NOT NULL,(confirmed_result->>'success')::boolean FROM send_recovery_resolutions WHERE id=$1`, audit).Scan(&local, &resolved, &success); err != nil {
+				t.Fatal(err)
+			}
+			kept := conflict == models.OrgImportConflictSkip
+			if local != kept || resolved != kept || success != kept {
+				t.Fatalf("old archive inherited authority or skip changed local evidence: local=%v resolved=%v success=%v kept=%v", local, resolved, success, kept)
+			}
+		})
+	}
+}
+
+func TestLiveImportSendRecoveryHistoryCannotReconcileUnknownOutcome(t *testing.T) {
+	svc, pool := liveImportService(t)
+	dest := newTenantFixture(t, pool)
+	ctx := t.Context()
+	mailbox, task, audit, sourceRepair := uuid.New(), uuid.New(), uuid.New(), uuid.New()
+	if _, err := pool.Exec(ctx, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider)
+		VALUES($1,$2,$3,$4,'Imported','','','smtp_imap')`, mailbox, dest.owner, dest.org, "import-"+mailbox.String()[:8]+"@example.test"); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM email_accounts WHERE id=$1`, mailbox) })
+	archive := tenantArchive(t, map[string][]map[string]any{
+		"tasks": {{"id": task, "email_account_id": mailbox, "task_type": "email", "status": "completed", "message_id": "", "completed_at": time.Now().UTC(), "send_result_state": "unknown"}},
+		"send_recovery_resolutions": {{"id": audit, "organization_id": dest.org, "email_account_id": mailbox, "recovery_task_id": task,
+			"previous_reason": "unknown", "evidence_type": "operator_confirmed_sent", "confirmation_reference": "Source-instance provider confirmation",
+			"recorded_locally_at":  time.Now().UTC().Add(365 * 24 * time.Hour),
+			"conflict_detected_at": time.Now().UTC().Add(365 * 24 * time.Hour), "conflicting_result_state": "failed",
+			"conflict_resolved_at": time.Now().UTC().Add(365 * 24 * time.Hour),
+			"confirmed_result":     map[string]any{"task_id": task, "success": true, "message_id": "source-result@example.test"}},
+			{"id": sourceRepair, "organization_id": dest.org, "email_account_id": mailbox, "recovery_task_id": task,
+				"previous_reason": "conflict", "evidence_type": "operator_provider_confirmation", "evidence_task_id": task,
+				"confirmation_reference": "Source conflict repair", "recorded_locally_at": time.Now().UTC().Add(365 * 24 * time.Hour)}},
+	})
+	if _, err := svc.ImportFrom(ctx, dest.org, archive, ImportOptions{Conflict: models.OrgImportConflictOverwrite, ActorUserID: dest.owner}, nil); err != nil {
+		t.Fatal(err)
+	}
+	repo := repository.NewTaskRepository(pool).(interface {
+		ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error)
+		ApplySendResult(context.Context, models.SendEmailResult, func(context.Context) error) error
+	})
+	results, err := repo.ListUnappliedSendResults(ctx, 200)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, result := range results {
+		if result.TaskID == task {
+			t.Fatal("imported historical confirmation gained result reconciliation authority")
+		}
+	}
+	var preserved bool
+	if err := pool.QueryRow(ctx, `SELECT recorded_locally_at IS NULL AND conflict_resolved_at IS NULL AND confirmed_result->>'success'='true'
+		AND confirmation_reference='Source-instance provider confirmation' FROM send_recovery_resolutions WHERE id=$1`, audit).Scan(&preserved); err != nil || !preserved {
+		t.Fatalf("portable audit lost or imported proof gained local provenance: preserved=%v error=%v", preserved, err)
+	}
+	if _, err := pool.Exec(ctx, `UPDATE email_accounts SET status='active',send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, mailbox, task); err != nil {
+		t.Fatal(err)
+	}
+	failed := models.SendEmailResult{TaskID: task, Error: &models.EmailSendError{Code: "RECIPIENT_REJECTED"}}
+	if err := repo.ApplySendResult(ctx, failed, func(context.Context) error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if err := repo.ApplySendResult(ctx, models.SendEmailResult{TaskID: task, Success: true}, func(context.Context) error {
+		t.Error("conflict re-applied accounting")
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var held bool
+	var reason string
+	if err := pool.QueryRow(ctx, `SELECT send_recovery_hold,COALESCE(send_recovery_reason,'') FROM email_accounts WHERE id=$1`, mailbox).Scan(&held, &reason); err != nil || !held || reason != "conflict" {
+		t.Fatalf("imported operator repair erased a fresh local conflict: held=%v reason=%s error=%v", held, reason, err)
+	}
+	var count int
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM send_recovery_resolutions WHERE recovery_task_id=$1 AND recorded_locally_at IS NOT NULL AND evidence_type='authoritative_result_conflict'`, task).Scan(&count); err != nil || count != 1 {
+		t.Fatalf("fresh local conflict was merged into imported confirmation: count=%d error=%v", count, err)
+	}
+	emailRepo := repository.NewEmailRepostory(&db.DB{Pool: pool}, nil)
+	resolution := &models.UpdateEmail{SendRecoveryResolution: &models.SendRecoveryResolution{
+		HeldTaskID: task, HeldReason: "conflict", EvidenceType: "operator_provider_confirmation",
+		EvidenceTaskID: &task, ConfirmationReference: "Fresh local provider confirmation",
+	}}
+	if _, xerr := emailRepo.Update(ctx, dest.org.String(), mailbox.String(), resolution); xerr != nil {
+		t.Fatal(xerr)
+	}
+	if err := pool.QueryRow(ctx, `SELECT send_recovery_hold FROM email_accounts WHERE id=$1`, mailbox).Scan(&held); err != nil || held {
+		t.Fatalf("fresh local confirmation could not resolve imported and native conflict regardless of clock order: held=%v error=%v", held, err)
+	}
+	if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM send_recovery_resolutions WHERE recovery_task_id=$1 AND conflict_detected_at IS NOT NULL AND conflict_resolved_at IS NULL`, task).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("explicit confirmation left known conflict rows unresolved: count=%d error=%v", count, err)
+	}
+}
+
+func TestLiveImportedHistoryDoesNotRepeatAuthoritativeWorkerResult(t *testing.T) {
+	for _, source := range []string{"executor", "warmup"} {
+		for _, sent := range []bool{false, true} {
+			for _, fresh := range []bool{false, true} {
+				name := source + map[bool]string{false: "/failed", true: "/sent"}[sent] + map[bool]string{false: "/independent", true: "/after fresh confirmation"}[fresh]
+				t.Run(name, func(t *testing.T) {
+					svc, pool := liveImportService(t)
+					dest := newTenantFixture(t, pool)
+					ctx := t.Context()
+					mailbox, task, audit := uuid.New(), uuid.New(), uuid.New()
+					if _, err := pool.Exec(ctx, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider)
+						VALUES($1,$2,$3,$4,'Imported','','','smtp_imap')`, mailbox, dest.owner, dest.org, mailbox.String()+"@example.test"); err != nil {
+						t.Fatal(err)
+					}
+					t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DELETE FROM email_accounts WHERE id=$1`, mailbox) })
+					kind := "email"
+					if source == "warmup" {
+						kind = "warmup"
+					}
+					failure := &models.EmailSendError{Code: "RECIPIENT_REJECTED"}
+					history := models.SendEmailResult{TaskID: task, Success: !sent, MessageID: "history@example.test"}
+					evidence := "operator_confirmed_sent"
+					if sent {
+						history.Error, evidence = failure, "operator_confirmed_not_sent"
+					}
+					archive := tenantArchive(t, map[string][]map[string]any{
+						"tasks": {{"id": task, "email_account_id": mailbox, "task_type": kind, "status": "completed", "message_id": "", "completed_at": time.Now().UTC().Add(-5 * time.Minute), "send_result_state": "unknown"}},
+						"send_recovery_resolutions": {{"id": audit, "organization_id": dest.org, "email_account_id": mailbox, "recovery_task_id": task,
+							"previous_reason": "unknown", "evidence_type": evidence, "confirmation_reference": "Source confirmation history", "confirmed_result": history,
+							"recorded_locally_at": time.Now().UTC().Add(365 * 24 * time.Hour)}},
+					})
+					if _, err := svc.ImportFrom(ctx, dest.org, archive, ImportOptions{Conflict: models.OrgImportConflictOverwrite, ActorUserID: dest.owner}, nil); err != nil {
+						t.Fatal(err)
+					}
+					repo := repository.NewTaskRepository(pool).(interface {
+						ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error)
+						ApplySendResult(context.Context, models.SendEmailResult, func(context.Context) error) error
+					})
+					results, err := repo.ListUnappliedSendResults(ctx, 1)
+					if err != nil || len(results) != 0 {
+						t.Fatalf("imported history gained execution authority: results=%v error=%v", results, err)
+					}
+					if _, err := pool.Exec(ctx, `UPDATE email_accounts SET status='active',send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, mailbox, task); err != nil {
+						t.Fatal(err)
+					}
+					var historyBefore string
+					if err := pool.QueryRow(ctx, `SELECT to_jsonb(r)::text FROM send_recovery_resolutions r WHERE id=$1`, audit).Scan(&historyBefore); err != nil {
+						t.Fatal(err)
+					}
+					calls := 0
+					apply := func(context.Context) error { calls++; return nil }
+					var appliedBefore time.Time
+					if fresh {
+						resolution := models.SendRecoveryResolution{HeldTaskID: task, HeldReason: "unknown", EvidenceType: evidence, ConfirmationReference: "Fresh local provider confirmation"}
+						if !sent {
+							resolution.MessageID = "fresh@example.test"
+						}
+						emailRepo := repository.NewEmailRepostory(&db.DB{Pool: pool}, nil)
+						if _, xerr := emailRepo.Update(ctx, dest.org.String(), mailbox.String(), &models.UpdateEmail{SendRecoveryResolution: &resolution}); xerr != nil {
+							t.Fatal(xerr)
+						}
+						results, err = repo.ListUnappliedSendResults(ctx, 1)
+						if err != nil || len(results) != 1 || results[0].TaskID != task || results[0].Success != !sent {
+							t.Fatalf("fresh local confirmation unavailable: results=%v error=%v", results, err)
+						}
+						if err := repo.ApplySendResult(ctx, results[0], apply); err != nil {
+							t.Fatal(err)
+						}
+						if err := pool.QueryRow(ctx, `SELECT send_result_applied_at FROM tasks WHERE id=$1`, task).Scan(&appliedBefore); err != nil {
+							t.Fatal(err)
+						}
+					}
+					worker := models.SendEmailResult{TaskID: task, Success: sent, MessageID: "native@example.test"}
+					if !sent {
+						worker.Error = failure
+					}
+					raw, err := json.Marshal(worker)
+					if err != nil {
+						t.Fatal(err)
+					}
+					if source == "executor" {
+						_, err = pool.Exec(ctx, `UPDATE tasks SET send_executor_result=$2 WHERE id=$1`, task, raw)
+					} else {
+						_, err = pool.Exec(ctx, `INSERT INTO warmup_tasks(task_id,dispatch_result)VALUES($1,$2)`, task, raw)
+					}
+					if err != nil {
+						t.Fatal(err)
+					}
+					results, err = repo.ListUnappliedSendResults(ctx, 1)
+					if err != nil || len(results) != 1 || results[0].TaskID != task || results[0].Success != sent {
+						t.Fatalf("independent authoritative worker result unavailable: results=%v error=%v", results, err)
+					}
+					if err := repo.ApplySendResult(ctx, results[0], apply); err != nil {
+						t.Fatal(err)
+					}
+					results, err = repo.ListUnappliedSendResults(ctx, 1)
+					if err != nil || len(results) != 0 {
+						t.Errorf("applied worker result repeats via imported history: results=%v error=%v", results, err)
+					}
+					var state, historyAfter string
+					var appliedAt time.Time
+					var held bool
+					var conflicts int
+					if err := pool.QueryRow(ctx, `SELECT t.send_result_state,t.send_result_applied_at,ea.send_recovery_hold,
+						(SELECT count(*) FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id AND r.recorded_locally_at IS NOT NULL AND r.conflict_detected_at IS NOT NULL AND r.conflict_resolved_at IS NULL),
+						(SELECT to_jsonb(r)::text FROM send_recovery_resolutions r WHERE r.id=$2)
+						FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, task, audit).Scan(&state, &appliedAt, &held, &conflicts, &historyAfter); err != nil {
+						t.Fatal(err)
+					}
+					wantSent, wantConflicts := sent, 0
+					if fresh {
+						wantSent, wantConflicts = !sent, 1
+					}
+					wantState := "failed"
+					if wantSent {
+						wantState = "sent"
+					}
+					if state != wantState || calls != 1 || held != fresh || conflicts != wantConflicts || historyAfter != historyBefore || fresh && !appliedAt.Equal(appliedBefore) {
+						t.Fatalf("state=%s callbacks=%d held=%v conflicts=%d history preserved=%v original application preserved=%v", state, calls, held, conflicts, historyAfter == historyBefore, appliedAt.Equal(appliedBefore))
+					}
+					newer := uuid.New()
+					newRaw, err := json.Marshal(models.SendEmailResult{TaskID: newer, Success: true})
+					if err != nil {
+						t.Fatal(err)
+					}
+					if _, err := pool.Exec(ctx, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_executor_result)
+						VALUES($1,$2,'email','completed','','unknown',$3)`, newer, mailbox, newRaw); err != nil {
+						t.Fatal(err)
+					}
+					results, err = repo.ListUnappliedSendResults(ctx, 1)
+					if err != nil || len(results) != 1 || results[0].TaskID != newer {
+						t.Fatalf("imported history starves newer result in batch of one: results=%v error=%v", results, err)
+					}
+					if err := repo.ApplySendResult(ctx, results[0], apply); err != nil {
+						t.Fatal(err)
+					}
+					results, err = repo.ListUnappliedSendResults(ctx, 1)
+					if err != nil || len(results) != 0 || calls != 2 {
+						t.Fatalf("newer result did not settle idempotently: results=%v callbacks=%d error=%v", results, calls, err)
+					}
+				})
+			}
+		}
+	}
+}

internal/app/orgtransfer/spec.go

@@ -87,6 +87,9 @@ type Table struct {
 	// works for a NOT NULL column (writing NULL into one aborts the import).
 	ResetOnImport []string
 
+	// ResetOnOverwrite also defaults reset columns when replacing destination rows.
+	ResetOnOverwrite bool
+
 	// ImportSkip exports the table for the record but never writes it back.
 	ImportSkip bool
 
@@ -789,9 +792,11 @@ var Tables = []Table{
 	},
 	{
 		Name: "send_recovery_resolutions", Group: models.OrgDataGroupSending,
-		Scope: scopeOrg,
-		Owner: scopeOrg,
-		Note:  "Evidence and operator confirmation used to lift a durable mailbox send hold. Unknown sends cannot be resolved through this history and remain restrictive after import.",
+		Scope:            scopeOrg,
+		Owner:            scopeOrg,
+		ResetOnImport:    []string{"recorded_locally_at", "conflict_resolved_at"},
+		ResetOnOverwrite: true,
+		Note:             "Recovery evidence travels as history. Source-instance results and resolutions cannot execute or release a conflict hold after import; fresh local confirmation or independent authoritative worker evidence is required.",
 	},
 	{
 		// An AI-group table, but it sits here because task_id points at tasks.

internal/infrastructure/db/migrations/000287_send_recovery_conflict_provenance.down.sql

@@ -0,0 +1,20 @@
+DO $$ BEGIN
+    IF EXISTS (SELECT 1 FROM send_recovery_resolutions) THEN
+        RAISE EXCEPTION 'Recovery audit evidence remains; roll forward rather than deleting evidence to downgrade';
+    END IF;
+END $$;
+
+DROP INDEX send_recovery_native_conflict;
+DROP INDEX send_recovery_unknown_resolution;
+CREATE UNIQUE INDEX send_recovery_unknown_resolution ON send_recovery_resolutions (recovery_task_id)
+    WHERE previous_reason = 'unknown';
+ALTER TABLE send_recovery_resolutions
+    DROP CONSTRAINT send_recovery_automatic_conflict_check,
+    DROP CONSTRAINT send_recovery_resolutions_conflicting_result_state_check,
+    DROP CONSTRAINT send_recovery_resolutions_evidence_type_check,
+    ADD CONSTRAINT send_recovery_resolutions_evidence_type_check CHECK (evidence_type IN (
+        'authentication_repaired', 'operator_provider_confirmation',
+        'operator_confirmed_sent', 'operator_confirmed_not_sent')),
+    DROP COLUMN conflicting_result_state,
+    DROP COLUMN recorded_locally_at,
+    DROP COLUMN conflict_resolved_at;

internal/infrastructure/db/migrations/000287_send_recovery_conflict_provenance.up.sql

@@ -0,0 +1,21 @@
+ALTER TABLE send_recovery_resolutions
+    ADD COLUMN conflicting_result_state text,
+    ADD COLUMN recorded_locally_at timestamptz,
+    ADD COLUMN conflict_resolved_at timestamptz,
+    DROP CONSTRAINT send_recovery_resolutions_evidence_type_check,
+    ADD CONSTRAINT send_recovery_resolutions_evidence_type_check CHECK (evidence_type IN (
+        'authentication_repaired', 'operator_provider_confirmation', 'operator_confirmed_sent',
+        'operator_confirmed_not_sent', 'authoritative_result_conflict')),
+    ADD CONSTRAINT send_recovery_resolutions_conflicting_result_state_check
+        CHECK (conflicting_result_state IN ('sent', 'failed')),
+    ADD CONSTRAINT send_recovery_automatic_conflict_check CHECK (
+        evidence_type <> 'authoritative_result_conflict' OR (
+            previous_reason = 'conflict' AND conflicting_result_state IS NOT NULL
+            AND conflict_detected_at IS NOT NULL AND confirmed_result IS NULL
+            AND evidence_task_id IS NULL AND confirmation_reference = ''));
+
+DROP INDEX send_recovery_unknown_resolution;
+CREATE UNIQUE INDEX send_recovery_unknown_resolution ON send_recovery_resolutions (recovery_task_id)
+    WHERE previous_reason = 'unknown' AND recorded_locally_at IS NOT NULL;
+CREATE UNIQUE INDEX send_recovery_native_conflict ON send_recovery_resolutions (recovery_task_id)
+    WHERE evidence_type = 'authoritative_result_conflict' AND recorded_locally_at IS NOT NULL;

internal/infrastructure/db/send_recovery_conflict_upgrade_test.go

@@ -0,0 +1,306 @@
+package db
+
+import (
+	"context"
+	"errors"
+	"net/url"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/golang-migrate/migrate/v4"
+	"github.com/golang-migrate/migrate/v4/source/iofs"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+const conflictPredecessorVersion = 286
+const conflictMigrationVersion = 287
+
+func conflictUpgradeDatabase(t *testing.T) (*migrate.Migrate, *pgx.Conn, uint) {
+	t.Helper()
+	return conflictUpgradeDatabaseAt(t, conflictPredecessorVersion)
+}
+
+func conflictUpgradeDatabaseAt(t *testing.T, start uint) (*migrate.Migrate, *pgx.Conn, uint) {
+	t.Helper()
+	dsn := os.Getenv("WARMBLY_TEST_DB")
+	if dsn == "" {
+		t.Skip("WARMBLY_TEST_DB not set")
+	}
+	u, err := url.Parse(dsn)
+	if err != nil || (u.Hostname() != "localhost" && u.Hostname() != "127.0.0.1") {
+		t.Fatal("upgrade requires local PostgreSQL")
+	}
+	admin, err := pgx.Connect(t.Context(), dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = admin.Close(context.Background()) })
+	name := "send_conflict_upgrade_" + uuid.NewString()[:8]
+	quoted := pgx.Identifier{name}.Sanitize()
+	if _, err := admin.Exec(t.Context(), "CREATE DATABASE "+quoted); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if _, err := admin.Exec(context.Background(), "DROP DATABASE "+quoted); err != nil {
+			t.Error(err)
+		}
+	})
+	u.Path = "/" + name
+	source, err := iofs.New(migrationsFS, "migrations")
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, err := openMigrate(source, u.String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _, _ = m.Close() })
+	if err := m.Migrate(start); err != nil {
+		t.Fatal(err)
+	}
+	conn, err := pgx.Connect(t.Context(), u.String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = conn.Close(context.Background()) })
+	assertConflictMigrationVersion(t, m, start, false)
+	return m, conn, conflictMigrationVersion
+}
+
+func conflictMigrationCommand(t *testing.T, conn *pgx.Conn) *exec.Cmd {
+	t.Helper()
+	root, err := filepath.Abs("../../..")
+	if err != nil {
+		t.Fatal(err)
+	}
+	cmd := exec.CommandContext(t.Context(), "go", "run", "./cmd/migrate")
+	cmd.Dir = root
+	cmd.Env = append(os.Environ(), "PRIMARY_DB="+conn.Config().ConnString())
+	return cmd
+}
+
+func runConflictMigrationCLI(t *testing.T, conn *pgx.Conn) {
+	t.Helper()
+	output, err := conflictMigrationCommand(t, conn).CombinedOutput()
+	if err != nil {
+		t.Fatalf("actual cmd/migrate failed: %v\n%s", err, output)
+	}
+	t.Log("actual cmd/migrate applied embedded production migrations")
+}
+
+func runConflictUpgrade(t *testing.T, m *migrate.Migrate, conn *pgx.Conn) {
+	t.Helper()
+	if conflictLatestMigration(t) > conflictMigrationVersion {
+		if err := m.Migrate(conflictMigrationVersion); err != nil {
+			t.Fatal(err)
+		}
+		t.Log("later migrations exist; native version-specific upgrade remains pinned to 287")
+		return
+	}
+	runConflictMigrationCLI(t, conn)
+}
+
+func conflictLatestMigration(t *testing.T) uint {
+	t.Helper()
+	entries, err := migrationsFS.ReadDir("migrations")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var latest uint
+	for _, entry := range entries {
+		if !strings.HasSuffix(entry.Name(), ".up.sql") {
+			continue
+		}
+		version, err := strconv.ParseUint(strings.SplitN(entry.Name(), "_", 2)[0], 10, 32)
+		if err != nil {
+			t.Fatal(err)
+		}
+		latest = max(latest, uint(version))
+	}
+	return latest
+}
+
+func assertConflictMigrationVersion(t *testing.T, m *migrate.Migrate, want uint, wantDirty bool) {
+	t.Helper()
+	got, dirty, err := m.Version()
+	if err != nil || got != want || dirty != wantDirty {
+		t.Fatalf("version=%d dirty=%v want=%d/%v err=%v", got, dirty, want, wantDirty, err)
+	}
+}
+
+func conflictUpgradeExec(t *testing.T, conn *pgx.Conn, query string, args ...any) {
+	t.Helper()
+	if _, err := conn.Exec(t.Context(), query, args...); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func conflictUpgradeSnapshot(t *testing.T, conn *pgx.Conn, query string) string {
+	t.Helper()
+	var data string
+	if err := conn.QueryRow(t.Context(), query).Scan(&data); err != nil {
+		t.Fatal(err)
+	}
+	return data
+}
+
+func TestLiveSendRecoveryConflictUpgradePreservesHistoricalEvidence(t *testing.T) {
+	testSendRecoveryConflictUpgrade(t, conflictPredecessorVersion, false)
+}
+
+func TestLiveSendRecoveryConflictCLIStartup(t *testing.T) {
+	m, conn, _ := conflictUpgradeDatabaseAt(t, 284)
+	runConflictMigrationCLI(t, conn)
+	assertConflictMigrationVersion(t, m, conflictLatestMigration(t), false)
+}
+
+func TestLiveSendRecoveryConflictProduction284UpgradePreservesHistoricalEvidence(t *testing.T) {
+	testSendRecoveryConflictUpgrade(t, 284, false)
+}
+
+func TestLiveSendRecoveryConflictHistoricalOnlyDownRefuses(t *testing.T) {
+	testSendRecoveryConflictUpgrade(t, conflictPredecessorVersion, true)
+}
+
+func testSendRecoveryConflictUpgrade(t *testing.T, start uint, historicalOnly bool) {
+	t.Helper()
+	m, conn, candidate := conflictUpgradeDatabaseAt(t, start)
+	user, org, mailbox, task := uuid.New(), uuid.New(), uuid.New(), uuid.New()
+	conflictUpgradeExec(t, conn, `SET TIME ZONE 'Pacific/Honolulu'`)
+	conflictUpgradeExec(t, conn, `INSERT INTO users(id,email,first_name,last_name)VALUES($1,$2,'Upgrade','')`, user, user.String()+"@example.test")
+	conflictUpgradeExec(t, conn, `INSERT INTO organizations(id,name,owner_user_id)VALUES($1,'Upgrade',$2)`, org, user)
+	conflictUpgradeExec(t, conn, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,send_recovery_hold,send_recovery_reason,send_recovery_task_id)
+		VALUES($1,$2,$3,$4,'Upgrade','','','smtp_imap',true,'unknown',$5)`, mailbox, user, org, mailbox.String()+"@example.test", task)
+	conflictUpgradeExec(t, conn, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_reserved_at,send_result_state)
+		VALUES($1,$2,'email','completed','',NOW()-INTERVAL '1 hour','unknown')`, task, mailbox)
+	conflictUpgradeExec(t, conn, `INSERT INTO warmup_statistics(email_account_id,date,emails_sent,emails_replied,target_volume)
+		VALUES($1,'2001-01-01',17,4,25)`, mailbox)
+	for _, evidence := range []string{"authentication_repaired", "operator_provider_confirmation", "operator_confirmed_sent"} {
+		reason := "conflict"
+		if evidence == "authentication_repaired" {
+			reason = "authentication"
+		} else if evidence == "operator_confirmed_sent" {
+			reason = "unknown"
+		}
+		conflictUpgradeExec(t, conn, `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,evidence_task_id,previous_reason,evidence_type,confirmation_reference,confirmed_result,created_at,conflict_detected_at)
+			VALUES($1,$2,$3,$3,$4,$5,'Historical fixture',CASE WHEN $4='unknown' THEN jsonb_build_object('task_id',$3::uuid::text,'success',true) END,'2001-01-01Z','2099-01-01Z')`, org, mailbox, task, reason, evidence)
+	}
+	auditBefore := conflictUpgradeSnapshot(t, conn, `SELECT jsonb_agg(to_jsonb(r) ORDER BY id)::text FROM send_recovery_resolutions r`)
+	mailboxBefore := conflictUpgradeSnapshot(t, conn, `SELECT to_jsonb(e)::text FROM email_accounts e LIMIT 1`)
+	taskBefore := conflictUpgradeSnapshot(t, conn, `SELECT to_jsonb(t)::text FROM tasks t LIMIT 1`)
+	accountingBefore := conflictUpgradeSnapshot(t, conn, `SELECT jsonb_agg(to_jsonb(s) ORDER BY email_account_id,date)::text FROM warmup_statistics s`)
+	runConflictUpgrade(t, m, conn)
+	assertConflictMigrationVersion(t, m, candidate, false)
+	auditAfter := conflictUpgradeSnapshot(t, conn, `SELECT jsonb_agg(to_jsonb(r)-'recorded_locally_at'-'conflict_resolved_at'-'conflicting_result_state' ORDER BY id)::text FROM send_recovery_resolutions r`)
+	if auditBefore != auditAfter || mailboxBefore != conflictUpgradeSnapshot(t, conn, `SELECT to_jsonb(e)::text FROM email_accounts e LIMIT 1`) || taskBefore != conflictUpgradeSnapshot(t, conn, `SELECT to_jsonb(t)::text FROM tasks t LIMIT 1`) || accountingBefore != conflictUpgradeSnapshot(t, conn, `SELECT jsonb_agg(to_jsonb(s) ORDER BY email_account_id,date)::text FROM warmup_statistics s`) {
+		t.Fatal("upgrade changed historical evidence, task/accounting or mailbox/hold state")
+	}
+	var trusted, projected, badDefaults int
+	if err := conn.QueryRow(t.Context(), `SELECT count(*) FILTER(WHERE recorded_locally_at IS NOT NULL),count(*) FILTER(WHERE conflicting_result_state IS NOT NULL OR conflict_resolved_at IS NOT NULL) FROM send_recovery_resolutions`).Scan(&trusted, &projected); err != nil || trusted != 0 || projected != 0 {
+		t.Fatalf("historical evidence acquired local authority: trusted=%d projected=%d err=%v", trusted, projected, err)
+	}
+	if err := conn.QueryRow(t.Context(), `SELECT count(*) FROM information_schema.columns WHERE table_name='send_recovery_resolutions' AND column_name IN('recorded_locally_at','conflict_resolved_at','conflicting_result_state') AND (is_nullable<>'YES' OR column_default IS NOT NULL)`).Scan(&badDefaults); err != nil || badDefaults != 0 {
+		t.Fatalf("new fields infer provenance: badDefaults=%d err=%v", badDefaults, err)
+	}
+	if historicalOnly {
+		assertConflictDownRefusal(t, m, conn, candidate)
+		return
+	}
+	// Local-only uniqueness leaves history intact and permits a fresh confirmation.
+	conflictUpgradeExec(t, conn, `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmation_reference,recorded_locally_at)
+		VALUES($1,$2,$3,'unknown','operator_confirmed_not_sent','Fresh fixture',clock_timestamp())`, org, mailbox, task)
+	for _, state := range []string{"sent", "failed"} {
+		t.Run("automatic_"+state, func(t *testing.T) {
+			newTask := uuid.New()
+			conflictUpgradeExec(t, conn, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_result_applied_at)
+				VALUES($1,$2,'email','completed','',$3,NOW())`, newTask, mailbox, map[string]string{"sent": "failed", "failed": "sent"}[state])
+			insert := `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,conflicting_result_state,conflict_detected_at,recorded_locally_at)
+				VALUES($1,$2,$3,'conflict','authoritative_result_conflict',$4,clock_timestamp(),clock_timestamp()) RETURNING id`
+			var audit uuid.UUID
+			if err := conn.QueryRow(t.Context(), insert, org, mailbox, newTask, state).Scan(&audit); err != nil {
+				t.Fatal(err)
+			}
+			for _, mutation := range []string{
+				`confirmed_result='{"success":true}'`, `evidence_task_id=recovery_task_id`, `confirmation_reference='not automatic'`,
+				`previous_reason='unknown'`, `conflicting_result_state=NULL`, `conflicting_result_state='unknown'`, `conflict_detected_at=NULL`,
+			} {
+				_, err := conn.Exec(t.Context(), `UPDATE send_recovery_resolutions SET `+mutation+` WHERE id=$1`, audit)
+				var pe *pgconn.PgError
+				if !errors.As(err, &pe) || pe.Code != "23514" {
+					t.Fatalf("automatic evidence mutation %s accepted or wrong failure: %v", mutation, err)
+				}
+			}
+			if err := conn.QueryRow(t.Context(), insert, org, mailbox, newTask, state).Scan(&audit); err == nil {
+				t.Fatal("duplicate local automatic evidence accepted")
+			}
+		})
+	}
+	t.Logf("real migrations 1..%d -> %d: historical rows/hold/accounting preserved; no provenance backfill; both payload-free conflict states and constraints verified", start, candidate)
+	assertConflictDownRefusal(t, m, conn, candidate)
+}
+
+func assertConflictDownRefusal(t *testing.T, m *migrate.Migrate, conn *pgx.Conn, candidate uint) {
+	t.Helper()
+	rowsQuery := `SELECT jsonb_agg(to_jsonb(r) ORDER BY id)::text FROM send_recovery_resolutions r`
+	columnsQuery := `SELECT jsonb_agg(to_jsonb(c) ORDER BY ordinal_position)::text FROM information_schema.columns c WHERE table_schema='public' AND table_name='send_recovery_resolutions'`
+	constraintsQuery := `SELECT jsonb_agg(jsonb_build_object('name',conname,'definition',pg_get_constraintdef(oid)) ORDER BY conname)::text FROM pg_constraint WHERE conrelid='send_recovery_resolutions'::regclass`
+	indexesQuery := `SELECT jsonb_agg(to_jsonb(i) ORDER BY indexname)::text FROM pg_indexes i WHERE schemaname='public' AND tablename='send_recovery_resolutions'`
+	queries := []string{rowsQuery, columnsQuery, constraintsQuery, indexesQuery}
+	before := make([]string, len(queries))
+	for i, query := range queries {
+		before[i] = conflictUpgradeSnapshot(t, conn, query)
+	}
+	err := m.Steps(-1)
+	if err == nil || !strings.Contains(err.Error(), "Recovery audit evidence remains") {
+		t.Fatalf("evidence-bearing down did not refuse: %v", err)
+	}
+	for i, query := range queries {
+		if before[i] != conflictUpgradeSnapshot(t, conn, query) {
+			t.Fatalf("refused down changed preserved rows or schema projection %d", i)
+		}
+	}
+	assertConflictMigrationVersion(t, m, candidate-1, true)
+	var dirty migrate.ErrDirty
+	if err := m.Up(); !errors.As(err, &dirty) {
+		t.Fatalf("dirty rejected downgrade was silently retried: %v", err)
+	}
+	if output, err := conflictMigrationCommand(t, conn).CombinedOutput(); err == nil || !strings.Contains(string(output), "Dirty database version") {
+		t.Fatalf("actual cmd/migrate silently retried dirty downgrade: %v\n%s", err, output)
+	}
+	for i, query := range queries {
+		if before[i] != conflictUpgradeSnapshot(t, conn, query) {
+			t.Fatalf("rejected CLI restart changed preserved rows or schema projection %d", i)
+		}
+	}
+	t.Log("real runner down refused without row/column/constraint/index changes; schema_migrations is dirty at down target, not a successful downgrade")
+}
+
+func TestLiveSendRecoveryConflictEmptyDownUp(t *testing.T) {
+	m, conn, candidate := conflictUpgradeDatabase(t)
+	before := conflictUpgradeSnapshot(t, conn, `SELECT jsonb_agg(to_jsonb(c) ORDER BY ordinal_position)::text FROM information_schema.columns c WHERE table_schema='public' AND table_name='send_recovery_resolutions'`)
+	constraintsQuery := `SELECT jsonb_agg(jsonb_build_object('name',conname,'definition',pg_get_constraintdef(oid)) ORDER BY conname)::text FROM pg_constraint WHERE conrelid='send_recovery_resolutions'::regclass`
+	constraintsBefore := conflictUpgradeSnapshot(t, conn, constraintsQuery)
+	runConflictUpgrade(t, m, conn)
+	assertConflictMigrationVersion(t, m, candidate, false)
+	if err := m.Steps(-1); err != nil {
+		t.Fatal(err)
+	}
+	assertConflictMigrationVersion(t, m, conflictPredecessorVersion, false)
+	after := conflictUpgradeSnapshot(t, conn, `SELECT jsonb_agg(to_jsonb(c) ORDER BY ordinal_position)::text FROM information_schema.columns c WHERE table_schema='public' AND table_name='send_recovery_resolutions'`)
+	if before != after || constraintsBefore != conflictUpgradeSnapshot(t, conn, constraintsQuery) {
+		t.Fatal("empty downgrade did not restore original columns and constraints")
+	}
+	var indexRestored bool
+	if err := conn.QueryRow(t.Context(), `SELECT pg_get_expr(indpred,indrelid) NOT LIKE '%recorded_locally_at%' FROM pg_index WHERE indexrelid='send_recovery_unknown_resolution'::regclass`).Scan(&indexRestored); err != nil || !indexRestored {
+		t.Fatalf("old unknown uniqueness not restored: %v %v", indexRestored, err)
+	}
+	runConflictUpgrade(t, m, conn)
+	assertConflictMigrationVersion(t, m, candidate, false)
+	t.Log("real runner empty-table candidate down -> 286 -> up restores original schema and re-applies extension")
+}

internal/infrastructure/db/testdata/send_recovery_conflict/README.md

@@ -0,0 +1,11 @@
+# Conflict-audit upgrade regressions
+
+Migration 287 was allocated only after fetching main at `3a10faf0624bd598962207d14b0261eb1bfb6fe5` and independently verifying PR939 merged with migration 286. The regressions use the real embedded production SQL, not virtual migration names or manually applied `ALTER TABLE`.
+
+Disposable local databases are initialized through all production migrations to version 284 or 286. Historical fixture rows are inserted before the actual `go run ./cmd/migrate` upgrades to 287. That command has no down action, so the down steps use the same native `openMigrate` runner with `Steps(-1)`. Empty-table down/up restores the old constraints and index. Both historical-only and new automatic evidence refuse down without altering rows, columns, constraints or indexes.
+
+When later migrations are added, the version-specific cases stay pinned to 287 with the native runner; a separate actual CLI startup test always verifies upgrade from 284 to the latest embedded version. Current source ends at 287, so every version-specific up also executes the actual CLI.
+
+Real refusal leaves `schema_migrations` dirty at the down target even though evidence and schema remain untouched. There is intentionally no automatic force/cleanup recovery for that state. Keep old control-plane mutators quiesced and use reviewed roll-forward rather than deleting evidence or blindly forcing a version.
+
+The normal earlier CI `cmd/migrate` invocation initializes the shared live component database. The dedicated send-conflict step explicitly selects these migration regressions and the native authority, repair, ordinary-send and archive regressions.

internal/repository/pg_email.go

@@ -1649,6 +1649,20 @@ func (r *emailRepository) Update(ctx context.Context, orgID, emailAccountID stri
 		return nil, errx.InternalError()
 	}
 	defer tx.Rollback(ctx)
+	if resolution := udata.SendRecoveryResolution; resolution != nil {
+		mailboxID, parseErr := uuid.Parse(emailAccountID)
+		if parseErr != nil || resolution.HeldTaskID == uuid.Nil {
+			return nil, errx.ErrInvalid
+		}
+		if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,268))`, resolution.HeldTaskID.String()); err != nil {
+			db.CaptureError(err, "lock send recovery task", nil, "exec")
+			return nil, errx.InternalError()
+		}
+		if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,269))`, mailboxID.String()); err != nil {
+			db.CaptureError(err, "lock send recovery mailbox", nil, "exec")
+			return nil, errx.InternalError()
+		}
+	}
 	if udata.WarmupStartTime != nil || udata.WarmupEndTime != nil {
 		// Lock the stored half of a partial patch so concurrent changes cannot cross.
 		query := `SELECT warmup_start_time::text, warmup_end_time::text FROM email_accounts WHERE organization_id=$1 AND id=$2 FOR UPDATE`
@@ -1695,8 +1709,8 @@ func (r *emailRepository) Update(ctx context.Context, orgID, emailAccountID stri
 		 AND EXISTS(SELECT 1 FROM tasks e WHERE e.id=$4 AND e.email_account_id=held.id AND e.send_result_state='sent' AND e.send_result_applied_at>held.completed_at AND e.send_executor_started_at>held.completed_at))
 		 OR ($3='operator_provider_confirmation' AND held.send_recovery_reason IN('authentication','permanent','conflict') AND length($5)>0 AND $4::uuid IS NOT NULL
 		 AND EXISTS(SELECT 1 FROM tasks e WHERE e.id=$4 AND e.email_account_id=held.id AND e.send_result_state IN('sent','failed') AND e.send_result_applied_at IS NOT NULL)))),
-		 history AS(INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,evidence_task_id,previous_reason,evidence_type,confirmation_reference)
-		 SELECT organization_id,id,send_recovery_task_id,$4,send_recovery_reason,$3,$5 FROM valid RETURNING email_account_id)
+		 history AS(INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,evidence_task_id,previous_reason,evidence_type,confirmation_reference,recorded_locally_at)
+		 SELECT organization_id,id,send_recovery_task_id,$4,send_recovery_reason,$3,$5,clock_timestamp() FROM valid RETURNING email_account_id)
 		 UPDATE email_accounts ea SET send_recovery_hold=false,send_recovery_reason=NULL,send_recovery_task_id=NULL
 		 FROM history WHERE ea.id=history.email_account_id`, orgID, emailAccountID, resolution.EvidenceType, evidenceTask, strings.TrimSpace(resolution.ConfirmationReference), resolution.HeldTaskID, resolution.HeldReason)
 			if resolveErr != nil {
@@ -1706,6 +1720,23 @@ func (r *emailRepository) Update(ctx context.Context, orgID, emailAccountID stri
 			if tag.RowsAffected() != 1 {
 				return nil, errx.ErrInvalid
 			}
+			if resolution.HeldReason == "conflict" {
+				if _, resolveErr := tx.Exec(ctx, `UPDATE send_recovery_resolutions SET conflict_resolved_at=clock_timestamp()
+					WHERE organization_id=$1 AND email_account_id=$2 AND recovery_task_id=$3
+					AND conflict_detected_at IS NOT NULL AND conflict_resolved_at IS NULL
+					AND (previous_reason='unknown' OR evidence_type='authoritative_result_conflict')`, orgID, emailAccountID, resolution.HeldTaskID); resolveErr != nil {
+					db.CaptureError(resolveErr, "resolve outcome conflict", nil, "exec")
+					return nil, errx.InternalError()
+				}
+			}
+			mailboxID, parseErr := uuid.Parse(emailAccountID)
+			if parseErr != nil {
+				return nil, errx.ErrInvalid
+			}
+			if resolveErr := refreshSendHold(ctx, tx, mailboxID, nil, nil); resolveErr != nil {
+				db.CaptureError(resolveErr, "refresh send recovery", nil, "exec")
+				return nil, errx.InternalError()
+			}
 		}
 	}
 	query := fmt.Sprintf(`

internal/repository/pg_orgtransfer.go

@@ -66,7 +66,7 @@ type OrgTransferRepository interface {
 	CountForeignRows(ctx context.Context, tx pgx.Tx, table, owner string, pk []string, refs []TenantReference, orgID uuid.UUID, rows []json.RawMessage) (int64, error)
 	// InsertBatch writes rows into table, honouring the conflict strategy. An
 	// overwrite only updates an existing row that owner selects for orgID.
-	InsertBatch(ctx context.Context, tx pgx.Tx, table string, cols []string, rows []json.RawMessage, conflict models.OrgImportConflict, pk []string, owner string, orgID uuid.UUID) (int64, error)
+	InsertBatch(ctx context.Context, tx pgx.Tx, table string, cols []string, rows []json.RawMessage, conflict models.OrgImportConflict, pk []string, owner string, orgID uuid.UUID, resetOnOverwrite []string) (int64, error)
 	// DeveloperBlocked reports an operator's block on building apps for this
 	// workspace or person.
 	DeveloperBlocked(ctx context.Context, tx pgx.Tx, orgID, userID uuid.UUID) (bool, error)
@@ -475,6 +475,7 @@ func (r *orgTransferRepository) InsertBatch(
 	pk []string,
 	owner string,
 	orgID uuid.UUID,
+	resetOnOverwrite []string,
 ) (int64, error) {
 	if len(rows) == 0 || len(cols) == 0 {
 		return 0, nil
@@ -503,6 +504,9 @@ func (r *orgTransferRepository) InsertBatch(
 			}
 			sets = append(sets, quoteIdent(c)+" = EXCLUDED."+quoteIdent(c))
 		}
+		for _, c := range resetOnOverwrite {
+			sets = append(sets, quoteIdent(c)+" = DEFAULT")
+		}
 		if len(sets) == 0 {
 			// Every column is part of the key, so there is nothing to update.
 			q += ` ON CONFLICT DO NOTHING`

internal/repository/pg_send_recovery.go

@@ -186,14 +186,27 @@ func (r *taskRepository) ApplySendResult(ctx context.Context, result models.Send
 	// Receipt stamping is idempotent; an already-applied failure cannot be refunded again.
 	repairDeadLetter := status == "dead_lettered" && previous != nil && *previous == state && state == "sent"
 	if !repairDeadLetter && (applied != nil || (previous != nil && *previous != "unknown")) {
-		if previous != nil && state != "unknown" && state != *previous {
-			if _, err = tx.Exec(ctx, `UPDATE send_recovery_resolutions SET conflict_detected_at=NOW()
-				WHERE recovery_task_id=$1 AND previous_reason='unknown' AND confirmed_result IS NOT NULL`, result.TaskID); err != nil {
+		if previous != nil && *previous != "unknown" && state != "unknown" && state != *previous {
+			tag, err := tx.Exec(ctx, `UPDATE send_recovery_resolutions
+				SET conflict_detected_at=COALESCE(conflict_detected_at,clock_timestamp()),conflicting_result_state=COALESCE(conflicting_result_state,$2)
+				WHERE recovery_task_id=$1 AND recorded_locally_at IS NOT NULL
+				AND (previous_reason='unknown' OR evidence_type='authoritative_result_conflict')`, result.TaskID, state)
+			if err != nil {
 				return err
 			}
-			_, err = tx.Exec(ctx, `UPDATE email_accounts SET send_recovery_hold = true, send_recovery_reason = 'conflict', send_recovery_task_id = $2
-				WHERE id = $1 AND (NOT send_recovery_hold OR (send_recovery_reason = 'unknown' AND send_recovery_task_id = $2))`, mailboxID, result.TaskID)
-			if err != nil {
+			if tag.RowsAffected() == 0 {
+				_, err = tx.Exec(ctx, `INSERT INTO send_recovery_resolutions
+					(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,conflict_detected_at,conflicting_result_state,recorded_locally_at)
+					SELECT ea.organization_id,ea.id,t.id,'conflict','authoritative_result_conflict',clock_timestamp(),$2,clock_timestamp()
+					FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, result.TaskID, state)
+				if err != nil {
+					return err
+				}
+			}
+			if err = refreshUnknownSendHold(ctx, tx, mailboxID, result.TaskID); err != nil {
+				return err
+			}
+			if err = refreshSendHold(ctx, tx, mailboxID, nil, nil); err != nil {
 				return err
 			}
 			return tx.Commit(ctx)
@@ -278,21 +291,33 @@ func resolveSendDeadLetter(ctx context.Context, tx pgx.Tx, taskID uuid.UUID) err
 	return err
 }
 
-func persistSendHold(ctx context.Context, tx pgx.Tx, mailboxID uuid.UUID, result models.SendEmailResult, state string) error {
-	if state != "unknown" {
-		_, err := tx.Exec(ctx, `WITH next_hold AS (
+func refreshUnknownSendHold(ctx context.Context, tx pgx.Tx, mailboxID, taskID uuid.UUID) error {
+	reason := "unknown"
+	return refreshSendHold(ctx, tx, mailboxID, &taskID, &reason)
+}
+
+func refreshSendHold(ctx context.Context, tx pgx.Tx, mailboxID uuid.UUID, taskID *uuid.UUID, reason *string) error {
+	_, err := tx.Exec(ctx, `WITH next_hold AS (
 			SELECT id,reason FROM (
 			 SELECT t.id,'unknown' AS reason,1 AS priority,t.created_at FROM tasks t
 			 WHERE t.email_account_id=$1 AND t.send_result_state='unknown'
 			 UNION ALL
 			 SELECT t.id,'conflict',2,t.created_at FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
-			 WHERE t.email_account_id=$1 AND r.previous_reason='unknown' AND r.conflict_detected_at IS NOT NULL
-			 AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions resolved WHERE resolved.recovery_task_id=t.id AND resolved.previous_reason='conflict')
+			 WHERE t.email_account_id=$1 AND (r.previous_reason='unknown' OR r.evidence_type='authoritative_result_conflict')
+			 AND r.conflict_detected_at IS NOT NULL AND r.conflict_resolved_at IS NULL
 			) candidates ORDER BY priority,created_at,id LIMIT 1)
 			UPDATE email_accounts ea SET send_recovery_task_id=n.id,send_recovery_hold=n.id IS NOT NULL,send_recovery_reason=n.reason
 			FROM (SELECT 1) singleton LEFT JOIN next_hold n ON true
-			WHERE ea.id=$1 AND ea.send_recovery_task_id=$2 AND ea.send_recovery_reason='unknown'`, mailboxID, result.TaskID)
-		if err != nil {
+			WHERE ea.id=$1 AND (
+			 ($3::text IS NOT NULL AND ea.send_recovery_task_id IS NOT DISTINCT FROM $2::uuid
+			  AND ea.send_recovery_reason IS NOT DISTINCT FROM $3::text)
+			 OR ($2::uuid IS NULL AND $3::text IS NULL AND NOT ea.send_recovery_hold))`, mailboxID, taskID, reason)
+	return err
+}
+
+func persistSendHold(ctx context.Context, tx pgx.Tx, mailboxID uuid.UUID, result models.SendEmailResult, state string) error {
+	if state != "unknown" {
+		if err := refreshUnknownSendHold(ctx, tx, mailboxID, result.TaskID); err != nil {
 			return err
 		}
 	}

internal/repository/send_admission.go

@@ -503,7 +503,7 @@ func (r *taskRepository) cancelOutbound(ctx context.Context, task, mailbox, work
 			return err
 		}
 	}
-	if _, err = tx.Exec(ctx, `UPDATE email_accounts SET send_recovery_hold=false,send_recovery_task_id=NULL,send_recovery_reason=NULL WHERE id=$1 AND send_recovery_reason='unknown' AND send_recovery_task_id=$2 AND NOT EXISTS(SELECT 1 FROM tasks WHERE email_account_id=$1 AND send_result_state='unknown')`, mailbox, task); err != nil {
+	if err = refreshUnknownSendHold(ctx, tx, mailbox, task); err != nil {
 		return err
 	}
 	return tx.Commit(ctx)

internal/repository/send_recovery_resolution_live_test.go

@@ -3,10 +3,14 @@ package repository
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/warmbly/warmbly/internal/errx"
 	"github.com/warmbly/warmbly/internal/infrastructure/db"
 	"github.com/warmbly/warmbly/internal/models"
 )
@@ -127,6 +131,170 @@ func TestLiveUnknownResolutionRejectsMissingEvidenceAndLiveExecution(t *testing.
 	}
 }
 
+func TestLiveOrdinarySendResultsNeedNoRecoveryConfirmation(t *testing.T) {
+	for _, success := range []bool{true, false} {
+		t.Run(fmt.Sprintf("success_%t", success), func(t *testing.T) {
+			f, r := lineageFixture(t)
+			ctx := t.Context()
+			if _, err := f.pool.Exec(ctx, `UPDATE email_accounts SET status='active',send_recovery_hold=false,send_recovery_task_id=NULL,send_recovery_reason=NULL,send_cooldown_until=NULL,provider='smtp_imap' WHERE id=$1`, f.sender); err != nil {
+				t.Fatal(err)
+			}
+			result := models.SendEmailResult{TaskID: f.task, Success: success}
+			want := "sent"
+			if !success {
+				want = "failed"
+				result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient"}}
+			}
+			calls := 0
+			for range 2 {
+				if err := r.ApplySendResult(ctx, result, func(context.Context) error { calls++; return nil }); err != nil {
+					t.Fatal(err)
+				}
+			}
+			var state string
+			var applied bool
+			if err := f.pool.QueryRow(ctx, `SELECT send_result_state,send_result_applied_at IS NOT NULL FROM tasks WHERE id=$1`, f.task).Scan(&state, &applied); err != nil || state != want || !applied || calls != 1 {
+				t.Fatalf("ordinary result did not apply once: state=%s applied=%v callbacks=%d error=%v", state, applied, calls, err)
+			}
+			admission, err := r.GetSendAdmission(ctx, f.org, f.sender, models.InboxProviderSMTPIMAP, time.Now())
+			if err != nil || !admission.Allowed || admission.RecoveryHold {
+				t.Fatalf("ordinary result required admin recovery: admission=%+v error=%v", admission, err)
+			}
+			var audits int
+			if err := f.pool.QueryRow(ctx, `SELECT count(*) FROM send_recovery_resolutions WHERE email_account_id=$1`, f.sender).Scan(&audits); err != nil || audits != 0 {
+				t.Fatalf("ordinary result fabricated recovery evidence: rows=%d error=%v", audits, err)
+			}
+		})
+	}
+}
+
+func TestLiveDefinitiveConflictDirectionsDuplicatesAndRestrictiveHolds(t *testing.T) {
+	for _, sentFirst := range []bool{false, true} {
+		for _, hold := range []string{"none", "unheld_pointer", "unheld_reason", "unheld_pointer_and_reason", "unknown", "authentication", "permanent", "conflict", "unclassified"} {
+			t.Run(fmt.Sprintf("sent_first_%t/%s", sentFirst, hold), func(t *testing.T) {
+				f, r := lineageFixture(t)
+				ctx := t.Context()
+				exec := func(q string, args ...any) {
+					t.Helper()
+					if _, err := f.pool.Exec(ctx, q, args...); err != nil {
+						t.Fatal(err)
+					}
+				}
+				exec(`UPDATE tasks SET status='completed',send_result_state='unknown' WHERE id=$1`, f.task)
+				failed := models.SendEmailResult{TaskID: f.task, LegacyErrorMsg: "raw-sentinel", Error: &models.EmailSendError{
+					Message: "raw-sentinel", Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient"},
+				}}
+				sent := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "raw-sentinel@example.test", ProviderMsgID: "raw-sentinel"}
+				first, conflicting := failed, sent
+				prior, outcome := "failed", "sent"
+				if sentFirst {
+					first, conflicting, prior, outcome = sent, failed, "sent", "failed"
+				}
+				var callbacks atomic.Int64
+				apply := func(context.Context) error { callbacks.Add(1); return nil }
+				if err := r.ApplySendResult(ctx, first, apply); err != nil {
+					t.Fatal(err)
+				}
+				var appliedAt time.Time
+				if err := f.pool.QueryRow(ctx, `SELECT send_result_applied_at FROM tasks WHERE id=$1`, f.task).Scan(&appliedAt); err != nil {
+					t.Fatal(err)
+				}
+				other := uuid.New()
+				exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id,send_result_state,send_result_applied_at)VALUES($1,'email',$2,'completed','','sent',NOW())`, other, f.sender)
+				switch hold {
+				case "none":
+					exec(`UPDATE email_accounts SET send_recovery_hold=false,send_recovery_reason=NULL,send_recovery_task_id=NULL WHERE id=$1`, f.sender)
+				case "unheld_pointer":
+					exec(`UPDATE email_accounts SET send_recovery_hold=false,send_recovery_reason=NULL,send_recovery_task_id=$2 WHERE id=$1`, f.sender, other)
+				case "unheld_reason":
+					exec(`UPDATE email_accounts SET send_recovery_hold=false,send_recovery_reason='permanent',send_recovery_task_id=NULL WHERE id=$1`, f.sender)
+				case "unheld_pointer_and_reason":
+					exec(`UPDATE email_accounts SET send_recovery_hold=false,send_recovery_reason='permanent',send_recovery_task_id=$2 WHERE id=$1`, f.sender, other)
+				case "unclassified":
+					exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason=NULL,send_recovery_task_id=NULL WHERE id=$1`, f.sender)
+				case "unknown":
+					exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task)
+				default:
+					exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason=$2,send_recovery_task_id=$3 WHERE id=$1`, f.sender, hold, other)
+				}
+				duplicate := func() {
+					t.Helper()
+					var wg sync.WaitGroup
+					results := make(chan error, 8)
+					start := make(chan struct{})
+					for range 8 {
+						wg.Add(1)
+						go func() {
+							defer wg.Done()
+							<-start
+							results <- r.ApplySendResult(ctx, conflicting, apply)
+						}()
+					}
+					close(start)
+					wg.Wait()
+					close(results)
+					for err := range results {
+						if err != nil {
+							t.Fatal(err)
+						}
+					}
+				}
+				duplicate()
+				var count int
+				var clean bool
+				if err := f.pool.QueryRow(ctx, `SELECT COUNT(*),BOOL_AND(conflicting_result_state=$2 AND confirmed_result IS NULL
+					AND evidence_type='authoritative_result_conflict' AND previous_reason='conflict' AND recorded_locally_at IS NOT NULL
+					AND confirmation_reference='' AND evidence_task_id IS NULL AND to_jsonb(r)::text NOT LIKE '%raw-sentinel%')
+					FROM send_recovery_resolutions r WHERE recovery_task_id=$1`, f.task, outcome).Scan(&count, &clean); err != nil || count != 1 || !clean {
+					t.Fatalf("automatic evidence duplicated, leaked payload or gained operator authority: count=%d clean=%v error=%v", count, clean, err)
+				}
+				var held bool
+				var reason string
+				var task uuid.UUID
+				readHold := func() {
+					t.Helper()
+					if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,COALESCE(send_recovery_reason,''),COALESCE(send_recovery_task_id,'00000000-0000-0000-0000-000000000000'::uuid) FROM email_accounts WHERE id=$1`, f.sender).Scan(&held, &reason, &task); err != nil {
+						t.Fatal(err)
+					}
+				}
+				readHold()
+				wantReason, wantTask := hold, other
+				unheld := hold == "none" || hold == "unheld_pointer" || hold == "unheld_reason" || hold == "unheld_pointer_and_reason"
+				if unheld || hold == "unknown" {
+					wantReason, wantTask = "conflict", f.task
+				} else if hold == "unclassified" {
+					wantReason, wantTask = "", uuid.Nil
+				}
+				if !held || reason != wantReason || task != wantTask {
+					t.Fatalf("restrictive hold lost or replaced: held=%v reason=%s task=%s", held, reason, task)
+				}
+				if unheld || hold == "unknown" {
+					if hold == "none" {
+						exec(`UPDATE send_recovery_resolutions SET conflict_detected_at=clock_timestamp()+INTERVAL '10 years' WHERE recovery_task_id=$1 AND evidence_type='authoritative_result_conflict'`, f.task)
+					}
+					repo := NewEmailRepostory(&db.DB{Pool: f.pool}, nil)
+					resolution := &models.UpdateEmail{SendRecoveryResolution: &models.SendRecoveryResolution{
+						HeldTaskID: f.task, HeldReason: "conflict", EvidenceType: "operator_provider_confirmation", EvidenceTaskID: &f.task, ConfirmationReference: "Fresh local provider confirmation",
+					}}
+					if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), resolution); xerr != nil {
+						t.Fatal(xerr)
+					}
+					duplicate()
+					readHold()
+					if held {
+						t.Fatal("duplicate contradiction undid fresh local operator confirmation")
+					}
+				}
+				var state string
+				var lastApplied time.Time
+				if err := f.pool.QueryRow(ctx, `SELECT send_result_state,send_result_applied_at FROM tasks WHERE id=$1`, f.task).Scan(&state, &lastApplied); err != nil || state != prior || !lastApplied.Equal(appliedAt) || callbacks.Load() != 1 {
+					t.Fatalf("duplicate/conflict changed original accounting: state=%s callbacks=%d error=%v", state, callbacks.Load(), err)
+				}
+			})
+		}
+	}
+}
+
 func TestLiveSendRecoveryResolutionRequiresEvidenceAndRetainsUnknown(t *testing.T) {
 	f, _ := lineageFixture(t)
 	ctx := t.Context()
@@ -196,3 +364,486 @@ func TestLiveSendRecoveryResolutionRequiresEvidenceAndRetainsUnknown(t *testing.
 		t.Fatal("operator evidence reference was not retained", history)
 	}
 }
+
+func TestLiveUnstartedDispatchRetainsDeferredOutcomeConflict(t *testing.T) {
+	for _, action := range []string{"cancel outbound", "defer warmup"} {
+		for _, anotherUnknown := range []bool{false, true} {
+			t.Run(action+map[bool]string{false: "/conflict", true: "/another unknown then conflict"}[anotherUnknown], func(t *testing.T) {
+				f, r := lineageFixture(t)
+				ctx := t.Context()
+				exec := func(query string, args ...any) {
+					t.Helper()
+					if _, err := f.pool.Exec(ctx, query, args...); err != nil {
+						t.Fatal(err)
+					}
+				}
+				exec(`UPDATE tasks SET status='completed',completed_at=NOW()-INTERVAL '5 minutes',send_result_state='unknown' WHERE id=$1`, f.task)
+				exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task)
+				resolution := models.SendRecoveryResolution{HeldTaskID: f.task, HeldReason: "unknown", EvidenceType: "operator_confirmed_not_sent", ConfirmationReference: "Provider confirmation fixture"}
+				repo := NewEmailRepostory(&db.DB{Pool: f.pool}, nil)
+				if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), &models.UpdateEmail{SendRecoveryResolution: &resolution}); xerr != nil {
+					t.Fatal(xerr)
+				}
+				results, err := r.ListUnappliedSendResults(ctx, 200)
+				if err != nil || len(results) != 1 || results[0].TaskID != f.task {
+					t.Fatal("confirmed outcome unavailable", err, len(results))
+				}
+				if err := r.ApplySendResult(ctx, results[0], func(context.Context) error { return nil }); err != nil {
+					t.Fatal(err)
+				}
+				task, worker, nonce := uuid.New(), uuid.New(), uuid.New()
+				exec(`INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol)VALUES($1,'worker',true,NOW(),2)`, worker)
+				t.Cleanup(func() { _, _ = f.pool.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+				lane := "email"
+				var executorNonce any = nonce
+				if action == "defer warmup" {
+					lane = "warmup"
+					executorNonce = nil
+				}
+				exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_executor_worker,send_executor_nonce)
+				VALUES($1,$2,$3,'completed','','unknown',$4,$5)`, task, f.sender, lane, worker, executorNonce)
+				if action == "defer warmup" {
+					exec(`INSERT INTO warmup_tasks(task_id,dispatch_worker_id,dispatch_nonce)VALUES($1,$2,$3)`, task, worker, nonce)
+				}
+				exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, task)
+				if err := r.ApplySendResult(ctx, models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "late-native@example.test"}, func(context.Context) error {
+					t.Fatal("contradictory result changed committed accounting")
+					return nil
+				}); err != nil {
+					t.Fatal(err)
+				}
+				var recorded bool
+				if err := f.pool.QueryRow(ctx, `SELECT conflict_detected_at IS NOT NULL FROM send_recovery_resolutions WHERE recovery_task_id=$1 AND previous_reason='unknown'`, f.task).Scan(&recorded); err != nil || !recorded {
+					t.Fatal("native contradiction not retained", recorded, err)
+				}
+				other := uuid.New()
+				if anotherUnknown {
+					exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state)VALUES($1,$2,'email','completed','','unknown')`, other, f.sender)
+				}
+				if action == "defer warmup" {
+					err = r.DeferWarmupDispatch(ctx, task, f.sender, worker, nonce, time.Now().Add(time.Hour))
+				} else {
+					err = r.CancelOutbound(ctx, task, f.sender, worker, nonce)
+				}
+				if err != nil {
+					t.Fatal(err)
+				}
+				var held bool
+				var reason *string
+				var heldTask *uuid.UUID
+				if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,send_recovery_reason,send_recovery_task_id FROM email_accounts WHERE id=$1`, f.sender).Scan(&held, &reason, &heldTask); err != nil {
+					t.Fatal(err)
+				}
+				if anotherUnknown {
+					if !held || reason == nil || *reason != "unknown" || heldTask == nil || *heldTask != other {
+						t.Fatal("unstarted dispatch did not transfer the remaining unknown hold", held, reason, heldTask)
+					}
+					if err := r.ApplySendResult(ctx, models.SendEmailResult{TaskID: other, Success: true}, func(context.Context) error { return nil }); err != nil {
+						t.Fatal(err)
+					}
+					if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,send_recovery_reason,send_recovery_task_id FROM email_accounts WHERE id=$1`, f.sender).Scan(&held, &reason, &heldTask); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if !held || reason == nil || *reason != "conflict" || heldTask == nil || *heldTask != f.task {
+					t.Fatalf("unstarted dispatch erased authoritative conflict: held=%v reason=%v task=%v", held, reason, heldTask)
+				}
+				admission, err := r.GetSendAdmission(ctx, f.org, f.sender, models.InboxProviderSMTPIMAP, time.Now())
+				if err != nil || admission.Allowed || admission.HoldReason != "conflict" {
+					t.Fatal("conflicting outcomes allowed another provider send", admission, err)
+				}
+				if action == "defer warmup" {
+					state, err := r.BeginWarmupDispatch(ctx, task, f.sender, worker, nonce)
+					if err != nil || state.State == "execute" {
+						t.Fatal("deferred nonce still executable", state, err)
+					}
+				} else {
+					state, err := r.InspectOutbound(ctx, task, f.sender, worker)
+					if err != nil || state.State != "denied" {
+						t.Fatal("cancelled nonce still executable", state, err)
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestLivePriorDefinitiveConflictWithoutOperatorResolution(t *testing.T) {
+	f, r := lineageFixture(t)
+	ctx := t.Context()
+	exec := func(q string, args ...any) {
+		t.Helper()
+		if _, err := f.pool.Exec(ctx, q, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`UPDATE tasks SET status='completed',send_result_state='unknown' WHERE id=$1`, f.task)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task)
+	failure := models.SendEmailResult{TaskID: f.task, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient"}}}
+	if err := r.ApplySendResult(ctx, failure, func(context.Context) error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	task, worker, nonce := uuid.New(), uuid.New(), uuid.New()
+	exec(`INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol)VALUES($1,'worker',true,NOW(),2)`, worker)
+	t.Cleanup(func() { _, _ = f.pool.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, worker) })
+	exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state,send_executor_worker,send_executor_nonce)
+		VALUES($1,$2,'email','completed','','unknown',$3,$4)`, task, f.sender, worker, nonce)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, task)
+	if err := r.ApplySendResult(ctx, models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "late-native@example.test"}, func(context.Context) error {
+		t.Fatal("contradictory result changed committed accounting")
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var recorded, initial int
+	var evidenceType, outcome string
+	if err := f.pool.QueryRow(ctx, `SELECT COUNT(*),COUNT(*) FILTER(WHERE confirmed_result IS NOT NULL),
+		MAX(evidence_type),MAX(conflicting_result_state)
+		FROM send_recovery_resolutions WHERE recovery_task_id=$1 AND evidence_type='authoritative_result_conflict' AND conflict_detected_at IS NOT NULL`, f.task).
+		Scan(&recorded, &initial, &evidenceType, &outcome); err != nil {
+		t.Fatal(err)
+	}
+	if recorded != 1 || initial != 0 || evidenceType != "authoritative_result_conflict" || outcome != "sent" {
+		t.Fatal("contradiction provenance not durably retained", recorded, initial, evidenceType, outcome)
+	}
+	if err := r.CancelOutbound(ctx, task, f.sender, worker, nonce); err != nil {
+		t.Fatal(err)
+	}
+	var held bool
+	var reason string
+	var heldTask uuid.UUID
+	if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,send_recovery_reason,send_recovery_task_id FROM email_accounts WHERE id=$1`, f.sender).
+		Scan(&held, &reason, &heldTask); err != nil || !held || reason != "conflict" || heldTask != f.task {
+		t.Fatal("ordinary authoritative contradiction was lost", held, reason, heldTask, err)
+	}
+	confirmation := &models.UpdateEmail{SendRecoveryResolution: &models.SendRecoveryResolution{
+		HeldTaskID: f.task, HeldReason: "conflict", EvidenceType: "operator_provider_confirmation", EvidenceTaskID: &f.task,
+		ConfirmationReference: "Provider evidence for the previously failed native result",
+	}}
+	repo := NewEmailRepostory(&db.DB{Pool: f.pool}, nil)
+	if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), confirmation); xerr != nil {
+		t.Fatal("conflict cannot be confirmed by operator", xerr)
+	}
+}
+
+func TestLiveAuthenticationRepairRetainsUnresolvedOutcomeConflict(t *testing.T) {
+	f, r := lineageFixture(t)
+	ctx := t.Context()
+	exec := func(q string, args ...any) {
+		t.Helper()
+		if _, err := f.pool.Exec(ctx, q, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`UPDATE tasks SET status='completed',completed_at=NOW()-INTERVAL '5 minutes',send_result_state='unknown' WHERE id=$1`, f.task)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task)
+	repo := NewEmailRepostory(&db.DB{Pool: f.pool}, nil)
+	resolution := models.SendRecoveryResolution{HeldTaskID: f.task, HeldReason: "unknown", EvidenceType: "operator_confirmed_not_sent", ConfirmationReference: "Provider ticket fixture"}
+	if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), &models.UpdateEmail{SendRecoveryResolution: &resolution}); xerr != nil {
+		t.Fatal(xerr)
+	}
+	results, err := r.ListUnappliedSendResults(ctx, 200)
+	if err != nil || len(results) != 1 || results[0].TaskID != f.task {
+		t.Fatal("operator confirmation not available", err, len(results))
+	}
+	if err := r.ApplySendResult(ctx, results[0], func(context.Context) error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.ApplySendResult(ctx, models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "late-native@example.test"}, func(context.Context) error {
+		t.Fatal("contradiction re-applied accounting")
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+	evidenceTask := uuid.New()
+	exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id,send_result_state,send_result_applied_at,send_executor_started_at)
+		VALUES($1,'email',$2,'completed','later@example.test','sent',NOW(),NOW())`, evidenceTask, f.sender)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='authentication',send_recovery_task_id=$2,last_synced_at=NOW() WHERE id=$1`, f.sender, f.task)
+	repair := &models.UpdateEmail{SendRecoveryResolution: &models.SendRecoveryResolution{
+		HeldTaskID: f.task, HeldReason: "authentication", EvidenceType: "authentication_repaired", EvidenceTaskID: &evidenceTask,
+	}}
+	if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), repair); xerr != nil {
+		t.Fatal(xerr)
+	}
+	var held bool
+	var reason string
+	var task uuid.UUID
+	if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,COALESCE(send_recovery_reason,''),COALESCE(send_recovery_task_id,'00000000-0000-0000-0000-000000000000'::uuid) FROM email_accounts WHERE id=$1`, f.sender).
+		Scan(&held, &reason, &task); err != nil || !held || reason != "conflict" || task != f.task {
+		t.Fatal("authentication repair erased contradictory outcome hold", held, reason, task, err)
+	}
+}
+
+func TestLiveConcurrentAuthenticationRepairRetainsNewOutcomeConflict(t *testing.T) {
+	f, r := lineageFixture(t)
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	exec := func(q string, args ...any) {
+		t.Helper()
+		if _, err := f.pool.Exec(ctx, q, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	exec(`UPDATE tasks SET status='completed',completed_at=NOW()-INTERVAL '5 minutes',send_result_state='unknown' WHERE id=$1`, f.task)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task)
+	repo := NewEmailRepostory(&db.DB{Pool: f.pool}, nil)
+	confirmation := &models.UpdateEmail{SendRecoveryResolution: &models.SendRecoveryResolution{
+		HeldTaskID: f.task, HeldReason: "unknown", EvidenceType: "operator_confirmed_not_sent", ConfirmationReference: "Provider ticket fixture",
+	}}
+	if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), confirmation); xerr != nil {
+		t.Fatal(xerr)
+	}
+	failure := models.SendEmailResult{TaskID: f.task, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient"}}}
+	if err := r.ApplySendResult(ctx, failure, func(context.Context) error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	evidenceTask := uuid.New()
+	exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id,send_result_state,send_result_applied_at,send_executor_started_at)
+		VALUES($1,'email',$2,'completed','','sent',NOW(),NOW())`, evidenceTask, f.sender)
+	authTask := uuid.New()
+	exec(`INSERT INTO tasks(id,task_type,email_account_id,status,message_id,completed_at,send_result_state,send_result_applied_at)
+		VALUES($1,'email',$2,'completed','',NOW()-INTERVAL '5 minutes','failed',NOW()-INTERVAL '5 minutes')`, authTask, f.sender)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='authentication',send_recovery_task_id=$2 WHERE id=$1`, f.sender, authTask)
+	key := int32(f.task.ID() & 0x7fffffff)
+	name := fmt.Sprintf("hold_repair_%x", f.task.ID())
+	exec(fmt.Sprintf(`CREATE FUNCTION %s() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN
+		IF NEW.id='%s'::uuid AND NOT OLD.send_recovery_hold AND NOT NEW.send_recovery_hold
+		AND OLD.send_recovery_reason IS NULL AND NEW.send_recovery_reason IS NULL THEN
+		PERFORM pg_advisory_xact_lock(938,%d); END IF; RETURN NEW; END $$`, name, f.sender, key))
+	exec(fmt.Sprintf(`CREATE TRIGGER %s BEFORE UPDATE ON email_accounts FOR EACH ROW EXECUTE FUNCTION %s()`, name, name))
+	t.Cleanup(func() {
+		_, _ = f.pool.Exec(context.Background(), fmt.Sprintf(`DROP TRIGGER IF EXISTS %s ON email_accounts; DROP FUNCTION IF EXISTS %s()`, name, name))
+	})
+	barrier, err := f.pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer barrier.Rollback(context.Background())
+	if _, err := barrier.Exec(ctx, `SELECT pg_advisory_xact_lock(938,$1)`, key); err != nil {
+		t.Fatal(err)
+	}
+	repair := &models.UpdateEmail{SendRecoveryResolution: &models.SendRecoveryResolution{
+		HeldTaskID: authTask, HeldReason: "authentication", EvidenceType: "authentication_repaired", EvidenceTaskID: &evidenceTask,
+	}}
+	repaired := make(chan error, 1)
+	go func() {
+		_, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), repair)
+		if xerr != nil {
+			repaired <- xerr
+		} else {
+			repaired <- nil
+		}
+	}()
+	deadline := time.Now().Add(3 * time.Second)
+	waiting := false
+	for time.Now().Before(deadline) {
+		if err := f.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_locks WHERE locktype='advisory' AND NOT granted AND classid=938 AND objid=$1::oid)`, key).Scan(&waiting); err != nil {
+			t.Fatal(err)
+		}
+		if waiting {
+			break
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	if !waiting {
+		t.Fatal("repair did not reach its hold-refresh snapshot barrier")
+	}
+	result := make(chan error, 1)
+	go func() {
+		result <- r.ApplySendResult(ctx, models.SendEmailResult{TaskID: f.task, Success: true}, func(context.Context) error {
+			return fmt.Errorf("contradiction re-applied accounting")
+		})
+	}()
+	committedBeforeRefresh := false
+	select {
+	case err := <-result:
+		if err != nil {
+			t.Fatal(err)
+		}
+		committedBeforeRefresh = true
+	case <-time.After(time.Second):
+	}
+	if err := barrier.Rollback(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if !committedBeforeRefresh {
+		if err := <-result; err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := <-repaired; err != nil {
+		t.Fatal(err)
+	}
+	var held bool
+	var reason string
+	var task uuid.UUID
+	if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold,COALESCE(send_recovery_reason,''),COALESCE(send_recovery_task_id,'00000000-0000-0000-0000-000000000000'::uuid) FROM email_accounts WHERE id=$1`, f.sender).Scan(&held, &reason, &task); err != nil || !held || reason != "conflict" || task != f.task {
+		t.Fatalf("repair lost a concurrent native conflict: held=%v reason=%s task=%s result_committed_before_refresh=%v error=%v", held, reason, task, committedBeforeRefresh, err)
+	}
+}
+
+func TestLiveHistoricalConfirmationRequiresFreshOrIndependentAuthority(t *testing.T) {
+	for _, sent := range []bool{false, true} {
+		for _, authority := range []string{"history only", "fresh confirmation", "executor result", "warmup result"} {
+			t.Run(fmt.Sprintf("sent=%v/%s", sent, authority), func(t *testing.T) {
+				f, r := lineageFixture(t)
+				ctx := t.Context()
+				exec := func(query string, args ...any) {
+					t.Helper()
+					if _, err := f.pool.Exec(ctx, query, args...); err != nil {
+						t.Fatal(err)
+					}
+				}
+				exec(`UPDATE tasks SET status='completed',completed_at=NOW()-INTERVAL '5 minutes',send_result_state='unknown' WHERE id=$1`, f.task)
+				exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, f.task)
+				failure := &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient"}}
+				history := models.SendEmailResult{TaskID: f.task, Success: !sent, Error: failure}
+				rawHistory, err := json.Marshal(history)
+				if err != nil {
+					t.Fatal(err)
+				}
+				exec(`INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmation_reference,confirmed_result)
+					VALUES($1,$2,$3,'unknown','operator_confirmed_sent','Historical fixture',$4)`, f.org, f.sender, f.task, rawHistory)
+				results, err := r.ListUnappliedSendResults(ctx, 200)
+				if err != nil || len(results) != 0 {
+					t.Fatalf("history gained authority: results=%v error=%v", results, err)
+				}
+				if authority != "history only" {
+					actual := models.SendEmailResult{TaskID: f.task, Success: sent, Error: failure, MessageID: "native@example.test"}
+					raw, err := json.Marshal(actual)
+					if err != nil {
+						t.Fatal(err)
+					}
+					switch authority {
+					case "fresh confirmation":
+						resolution := models.SendRecoveryResolution{HeldTaskID: f.task, HeldReason: "unknown", EvidenceType: "operator_confirmed_not_sent", ConfirmationReference: "Fresh provider fixture"}
+						if sent {
+							resolution.EvidenceType, resolution.MessageID = "operator_confirmed_sent", "fresh@example.test"
+						}
+						repo := NewEmailRepostory(&db.DB{Pool: f.pool}, nil)
+						if _, xerr := repo.Update(ctx, f.org.String(), f.sender.String(), &models.UpdateEmail{SendRecoveryResolution: &resolution}); xerr != nil {
+							t.Fatal(xerr)
+						}
+					case "executor result":
+						exec(`UPDATE tasks SET send_executor_result=$2 WHERE id=$1`, f.task, raw)
+					case "warmup result":
+						exec(`INSERT INTO warmup_tasks(task_id,dispatch_result)VALUES($1,$2)`, f.task, raw)
+					}
+					results, err = r.ListUnappliedSendResults(ctx, 200)
+					if err != nil || len(results) != 1 || results[0].TaskID != f.task || results[0].Success != sent {
+						t.Fatalf("fresh authority unavailable or replaced by history: results=%v error=%v", results, err)
+					}
+				}
+				var held bool
+				if err := f.pool.QueryRow(ctx, `SELECT send_recovery_hold FROM email_accounts WHERE id=$1`, f.sender).Scan(&held); err != nil || !held {
+					t.Fatal("authority released hold before accounting", held, err)
+				}
+				applied := 0
+				for _, result := range results {
+					for range 2 {
+						if err := r.ApplySendResult(ctx, result, func(context.Context) error { applied++; return nil }); err != nil {
+							t.Fatal(err)
+						}
+					}
+				}
+				var state string
+				var historyCount int
+				if err := f.pool.QueryRow(ctx, `SELECT t.send_result_state,ea.send_recovery_hold,
+					(SELECT COUNT(*) FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id AND r.recorded_locally_at IS NULL AND r.confirmed_result=$2::jsonb)
+					FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, f.task, rawHistory).Scan(&state, &held, &historyCount); err != nil {
+					t.Fatal(err)
+				}
+				wantState, wantApplied := "unknown", 0
+				if authority != "history only" {
+					wantApplied, wantState = 1, "failed"
+					if sent {
+						wantState = "sent"
+					}
+				}
+				if state != wantState || applied != wantApplied || held != (authority == "history only") || historyCount != 1 {
+					t.Fatalf("state=%s applied=%d held=%v preserved history=%d", state, applied, held, historyCount)
+				}
+			})
+		}
+	}
+}
+
+func TestLiveExpandedConflictSchemaRequiresUpgradedReadersAndWriters(t *testing.T) {
+	f, r := lineageFixture(t)
+	ctx := t.Context()
+	exec := func(query string, args ...any) {
+		t.Helper()
+		if _, err := f.pool.Exec(ctx, query, args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+	failure := models.SendEmailResult{TaskID: f.task, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient"}}}
+	if err := r.ApplySendResult(ctx, failure, func(context.Context) error { return nil }); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.ApplySendResult(ctx, models.SendEmailResult{TaskID: f.task, Success: true}, func(context.Context) error { t.Fatal("contradiction re-applied accounting"); return nil }); err != nil {
+		t.Fatal(err)
+	}
+	other := uuid.New()
+	exec(`INSERT INTO tasks(id,email_account_id,task_type,status,message_id,send_result_state)VALUES($1,$2,'email','failed','','failed')`, other, f.sender)
+	exec(`UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='unknown',send_recovery_task_id=$2 WHERE id=$1`, f.sender, other)
+	old, err := f.pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer old.Rollback(ctx)
+	// Pre-extension hold selection does not recognize typed automatic conflicts.
+	if _, err := old.Exec(ctx, `WITH next_hold AS (
+		SELECT id,reason FROM (
+		 SELECT t.id,'unknown' AS reason,1 AS priority,t.created_at FROM tasks t
+		 WHERE t.email_account_id=$1 AND t.send_result_state='unknown'
+		 UNION ALL
+		 SELECT t.id,'conflict',2,t.created_at FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
+		 WHERE t.email_account_id=$1 AND r.previous_reason='unknown' AND r.conflict_detected_at IS NOT NULL
+		 AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions resolved WHERE resolved.recovery_task_id=t.id AND resolved.previous_reason='conflict')
+		) candidates ORDER BY priority,created_at,id LIMIT 1)
+		UPDATE email_accounts ea SET send_recovery_task_id=n.id,send_recovery_hold=n.id IS NOT NULL,send_recovery_reason=n.reason
+		FROM (SELECT 1) singleton LEFT JOIN next_hold n ON true
+		WHERE ea.id=$1 AND ea.send_recovery_task_id=$2 AND ea.send_recovery_reason='unknown'`, f.sender, other); err != nil {
+		t.Fatal(err)
+	}
+	var held bool
+	if err := old.QueryRow(ctx, `SELECT send_recovery_hold FROM email_accounts WHERE id=$1`, f.sender).Scan(&held); err != nil || held {
+		t.Fatal("fixture no longer demonstrates old writer losing the conflict", held, err)
+	}
+	if err := old.Rollback(ctx); err != nil {
+		t.Fatal(err)
+	}
+	updated, err := f.pool.Begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer updated.Rollback(ctx)
+	if err := refreshUnknownSendHold(ctx, updated, f.sender, other); err != nil {
+		t.Fatal(err)
+	}
+	var reason string
+	var heldTask uuid.UUID
+	if err := updated.QueryRow(ctx, `SELECT send_recovery_hold,send_recovery_reason,send_recovery_task_id FROM email_accounts WHERE id=$1`, f.sender).Scan(&held, &reason, &heldTask); err != nil || !held || reason != "conflict" || heldTask != f.task {
+		t.Fatal("updated writer lost conflict", held, reason, heldTask, err)
+	}
+	if err := updated.Rollback(ctx); err != nil {
+		t.Fatal(err)
+	}
+	raw, err := json.Marshal(models.SendEmailResult{TaskID: other, Success: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	exec(`UPDATE tasks SET status='completed',send_result_state='unknown',send_result_applied_at=NULL WHERE id=$1`, other)
+	exec(`INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmed_result)
+		VALUES($1,$2,$3,'unknown','operator_confirmed_sent',$4)`, f.org, f.sender, other, raw)
+	var oldReadable bool
+	if err := f.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
+		WHERE t.id=$1 AND r.confirmed_result IS NOT NULL AND t.send_result_applied_at IS NULL AND t.status<>'cancelled')`, other).Scan(&oldReadable); err != nil || !oldReadable {
+		t.Fatal("fixture no longer demonstrates ungated historical authority", oldReadable, err)
+	}
+	results, err := r.ListUnappliedSendResults(ctx, 200)
+	if err != nil || len(results) != 0 {
+		t.Fatal("updated reader accepted historical authority", results, err)
+	}
+}

internal/repository/unknown_send_resolution.go

@@ -73,11 +73,11 @@ func queueUnknownSendResolution(ctx context.Context, tx pgx.Tx, org, mailbox str
 		return errx.InternalError()
 	}
 	// Keep admission held until the normal result consumer commits all accounting effects.
-	tag, err := tx.Exec(ctx, `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmation_reference,confirmed_result)
-	 SELECT ea.organization_id,ea.id,t.id,'unknown',$4,$5,$6 FROM email_accounts ea JOIN tasks t ON t.email_account_id=ea.id
+	tag, err := tx.Exec(ctx, `INSERT INTO send_recovery_resolutions(organization_id,email_account_id,recovery_task_id,previous_reason,evidence_type,confirmation_reference,confirmed_result,recorded_locally_at)
+	 SELECT ea.organization_id,ea.id,t.id,'unknown',$4,$5,$6,clock_timestamp() FROM email_accounts ea JOIN tasks t ON t.email_account_id=ea.id
 	 WHERE ea.organization_id=$1 AND ea.id=$2 AND ea.status='active' AND ea.send_recovery_hold AND ea.send_recovery_reason='unknown'
 	 AND ea.send_recovery_task_id=$3 AND t.id=$3 AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL
-	 AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id AND r.previous_reason='unknown')`,
+	 AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id AND r.previous_reason='unknown' AND r.recorded_locally_at IS NOT NULL)`,
 		org, mailbox, resolution.HeldTaskID, resolution.EvidenceType, strings.TrimSpace(resolution.ConfirmationReference), raw)
 	if err != nil {
 		return errx.InternalError()
@@ -95,7 +95,7 @@ func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int
 	 AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`))
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,t.send_executor_result,1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
-	 WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL
+	 WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL AND r.recorded_locally_at IS NOT NULL
 	 AND t.send_executor_result IS NOT NULL AND t.send_result_applied_at IS NOT NULL AND t.status<>'cancelled'
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,w.dispatch_result,1 FROM warmup_tasks w JOIN tasks t ON t.id=w.task_id
@@ -103,11 +103,12 @@ func (r *taskRepository) ListUnappliedSendResults(ctx context.Context, limit int
 	 AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`))
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,w.dispatch_result,1 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id JOIN warmup_tasks w ON w.task_id=t.id
-	 WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL
+	 WHERE r.previous_reason='unknown' AND r.conflict_detected_at IS NULL AND r.confirmed_result IS NOT NULL AND r.recorded_locally_at IS NOT NULL
 	 AND w.dispatch_result IS NOT NULL AND t.send_result_applied_at IS NOT NULL AND t.status<>'cancelled'
 	 UNION ALL
 	 SELECT t.id,t.created_at,t.send_result_state,t.send_result_applied_at,r.confirmed_result,2 FROM send_recovery_resolutions r JOIN tasks t ON t.id=r.recovery_task_id
-	 WHERE r.confirmed_result IS NOT NULL AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`)) AND t.status<>'cancelled'
+	 WHERE r.confirmed_result IS NOT NULL AND r.recorded_locally_at IS NOT NULL
+	 AND (t.send_result_applied_at IS NULL OR (`+repairableSendDeadLetterPredicate+`)) AND t.status<>'cancelled'
 	 ),
 	 definitive AS (SELECT *,CASE WHEN result->>'success'='true' THEN 'sent' ELSE 'failed' END AS outcome FROM recorded
 	 WHERE result->>'success'='true' OR result->'error'->'failure'->>'disposition' IN('retry','throttle','permanent','authentication')

internal/repository/warmup_dispatch.go

@@ -192,8 +192,7 @@ func (r *taskRepository) DeferWarmupDispatch(ctx context.Context, task, mailbox,
 	if _, err = tx.Exec(ctx, `UPDATE tasks SET status='pending',completed_at=NULL,scheduled_at=$2,updated_at=NOW(),send_result_state=NULL,message_id='' WHERE id=$1`, task, at); err != nil {
 		return err
 	}
-	if _, err = tx.Exec(ctx, `UPDATE email_accounts SET send_recovery_hold=false,send_recovery_reason=NULL,send_recovery_task_id=NULL
-        WHERE id=$1 AND send_recovery_reason='unknown' AND send_recovery_task_id=$2 AND NOT EXISTS(SELECT 1 FROM tasks WHERE email_account_id=$1 AND send_result_state='unknown')`, mailbox, task); err != nil {
+	if err = refreshUnknownSendHold(ctx, tx, mailbox, task); err != nil {
 		return err
 	}
 	return tx.Commit(ctx)

internal/repository/warmup_preparation_recovery_live_test.go

@@ -67,7 +67,9 @@ func TestLiveWarmupRecoversOldPreparationResultAfterDeferral(t *testing.T) {
 	if _, err := r.AuthorizeWarmupDispatch(ctx, f.task, f.sender, worker); err != nil {
 		t.Fatal(err)
 	}
-	if err := r.ApplySendResult(ctx, result, apply); err != nil {
+	// Reproduce the released backend's unaudited conflict projection, not the current writer.
+	if _, err := f.pool.Exec(ctx, `UPDATE email_accounts SET send_recovery_hold=true,send_recovery_reason='conflict',send_recovery_task_id=$2
+		WHERE id=$1 AND (NOT send_recovery_hold OR (send_recovery_reason='unknown' AND send_recovery_task_id=$2))`, f.sender, f.task); err != nil {
 		t.Fatal(err)
 	}
 	var reason string