Merge pull request #950 from warmbly/devin/1791718013-archive-send-preservation

feat: preserve destination-local send protection during workspace archive overwrites
2026-10-12 16:02:44 +00:00 · 2026-10-12 02:48:14 +00:00
parent 180964f2d9 0c413e49bc
commit 2aedb7ea0c
11 changed files with 2146 additions and 23 deletions
@@ -293,6 +293,41 @@ jobs:

          go test -race -v -p 1 ./internal/app/consumer -run '^TestLiveAdvancedBounceReconciliationRollbackAndReplayWithOneConnection$' -count=1

+      - name: Workspace archive send-preservation regressions (PostgreSQL)
+        env:
+          WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
+        run: |
+          set -o pipefail
+          go test -race -v -p 1 ./internal/app/orgtransfer -run '^TestLive(ReviewSelectiveExport|ImportLocal|ImportFresh(WarmupReservationDuringMailboxImport|LegacyWarmupHistoryRetainsPayload)|ImportLegacy|ImportRefundedWarmup|ImportExistingReceiptExcluded|ArchiveSendRecoveryLocalOriginResets|ImportMailTransportWholeArchiveSendResultLockOrder|ImportSendRecoveryHistoryCannotReconcileUnknownOutcome|ImportedHistoryDoesNotRepeatAuthoritativeWorkerResult)' -count=1 | tee archive-send-preservation.log
+          if grep -q -- '--- SKIP:' archive-send-preservation.log; then
+            echo "Workspace archive send-preservation regressions must execute without skips"
+            exit 1
+          fi
+          for test in TestLiveReviewSelectiveExportMustPreserveLocalUnknownHold \
+            TestLiveReviewSelectiveExportLateLocalResultRemainsAccepted \
+            TestLiveReviewSelectiveExportNewImportIsHistoryNotAuthority \
+            TestLiveImportLocalSendIdentityPayloadAndUnrelated \
+            TestLiveImportLocalSendRollbackForeignAndRestrictiveConflicts \
+            TestLiveImportLocalWarmupChargeReplyReceiptAndLateAccounting \
+            TestLiveImportLocalResultPreservationLockOrder \
+            TestLiveImportLegacyUnknownPreservesOtherOutcome \
+            TestLiveImportLegacyAppliedResultRemainsIdempotent \
+            TestLiveImportLegacyStampedTaskAcceptsLaterResult \
+            TestLiveImportRefundedWarmupChargeWithoutNonces \
+            TestLiveImportLegacyWarmupAppliedAccountingWithoutNonces \
+            TestLiveImportLegacyWarmupAppliedMarkers \
+            TestLiveImportLegacyUnknownLockOrder \
+            TestLiveImportLegacyRestrictivePreservation \
+            TestLiveImportExistingReceiptExcluded \
+            TestLiveImportFreshWarmupReservationDuringMailboxImport \
+            TestLiveImportFreshLegacyWarmupHistoryRetainsPayload \
+            TestLiveArchiveSendRecoveryLocalOriginResets \
+            TestLiveImportMailTransportWholeArchiveSendResultLockOrder \
+            TestLiveImportSendRecoveryHistoryCannotReconcileUnknownOutcome \
+            TestLiveImportedHistoryDoesNotRepeatAuthoritativeWorkerResult; do
+            grep -q "^--- PASS: $test " archive-send-preservation.log
+          done
+
      - name: Typed mailbox history, archive checkpoint and real-driver upgrade regressions (PostgreSQL)
        env:
          WARMBLY_TEST_DB: postgres://warmbly:warmbly@127.0.0.1:5432/warmbly_test?sslmode=disable
@@ -47,7 +47,7 @@ Read the report, pick the groups to apply, and confirm.
 - Archives are kept for 7 days and then deleted automatically.
 - The passphrase is never stored. If you lose it, the credentials in that archive cannot be recovered.
 - An import runs as one transaction. If anything fails, nothing lands.
- Mailbox overwrite preserves report history, send audit and restrictive holds. Recovery is checked automatically against the destination's newly loaded mailbox transport; an old or in-flight result cannot clear restored transport warnings. Skipped mailboxes are unaffected.
+- Mailbox overwrite preserves report history, send audit and existing restrictive holds. Destination tasks with send-protection evidence retain their identity, typed payload and send evidence, including older completed tasks, uncertain or applied results without a worker nonce, and known warmup charges. An unprotected task can accept newer imported history without that history blocking its own later task payloads. Source-only history does not grant permission to send. If the archive and destination hold different tasks or have different restrictive hold reasons, overwrite fails and the whole import rolls back rather than discarding either protection. Skipped mailboxes are unaffected. Transport recovery is checked against the destination's newly loaded mailbox transport; an old or in-flight result cannot clear restored transport warnings.
 - To move a whole self-hosted install, use `warmblyctl backup` and `warmblyctl restore` instead.

 <Fold>
@@ -94,13 +94,13 @@ Inbox, send history, delivery events, and logs are the ones that grow without li

 The pending warmup-filing queue stays on its source instance. The destination resyncs mail and files recognized warmup messages using its own workers and the mailbox's placement setting.

-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.
+Warmup statistics travel as history. Source-only per-task charge and refund markers reset on import so settling an old task cannot refund another send's destination statistics. Existing destination charge dates and older completed-day counters are protected on overwrite; preserving older counts does not create charge or refund proof. Whole-instance database backups retain these markers.

 Mailbox transport-error reports travel as history, retaining every report ID, observation timestamp, task reference and recorded resolution. Their operation and observation provenance reset to unknown, including overwrite imports and older archives without those fields. Imported and legacy historical reports do not share the uniqueness rule for new, locally observed reports and are not automatically cleared by a successful sync or send. Unknown history means recovery is unverified, not that a provider is still failing.

 Instance-local transport-success clocks do not travel with an archive. The mailbox transport generation is omitted from exported rows, manifest columns and stored-result/confirmation JSON envelopes; other result and audit evidence remains intact. Importing a new mailbox or overwriting an existing one creates its own destination generation and invalidates only that mailbox's derived transport clocks in the same transaction. Skipped mailboxes and mailboxes not in the archive keep their generation and clocks. This does not clear restrictive send holds, confirmed-task evidence or accounting history.

-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.
+Send-recovery audit rows also travel as history, including operator confirmations and payload-free contradictory outcome metadata. Source-only rows import without local provenance; an existing destination row with locally recorded provenance is retained on overwrite. 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. Legitimate destination-local late results can still settle and apply accounting once. 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).

@@ -119,6 +119,7 @@ func (f *outboundStartFixture) snapshot(t *testing.T) string {

 func (f *outboundStartFixture) importPendingSnapshot(t *testing.T, status string) {
 	t.Helper()
+	before := f.snapshot(t)
 	rows := map[string][]map[string]any{
 		"tasks":         {{"id": f.task, "task_type": "warmup", "email_account_id": f.sender, "status": status, "message_id": "", "send_result_state": "unknown"}},
 		"warmup_tasks":  {{"task_id": f.task, "target_account_id": f.recipient, "subject": f.source.Subject}},
@@ -159,9 +160,36 @@ func (f *outboundStartFixture) importPendingSnapshot(t *testing.T, status string
 	}
 	svc := orgtransfer.NewService(repository.NewOrgTransferRepository(f.db), nil, nil, nil, orgtransfer.InstanceInfo{})
 	result, err := svc.ImportFrom(t.Context(), f.org, bytes.NewReader(buf.Bytes()), orgtransfer.ImportOptions{Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.user}, nil)
-	if err != nil || result.RowCounts["tasks"] != 1 || result.RowCounts["warmup_tasks"] != 1 || result.RowCounts["warmup_tokens"] != 1 {
+	if err != nil || result == nil || result.RowCounts["tasks"] != 0 || result.RowCounts["warmup_tasks"] != 0 || result.RowCounts["warmup_tokens"] != 0 {
 		t.Fatalf("real partial archive import: result=%+v error=%v", result, err)
 	}
+	if after := f.snapshot(t); after != before {
+		t.Fatalf("protected partial import changed local authority/history\nbefore=%s\nafter=%s", before, after)
+	}
+}
+
+func (f *outboundStartFixture) prepareDeferredWarmup(t *testing.T) {
+	t.Helper()
+	lineage := f.repo.(repository.WarmupLineageRepository)
+	if err := lineage.RescheduleWarmupTask(t.Context(), f.task, time.Now().Add(-time.Minute)); err != nil {
+		t.Fatal(err)
+	}
+	if claimed, err := lineage.ClaimWarmupTask(t.Context(), f.task, time.Now()); err != nil || !claimed {
+		t.Fatalf("claim deferred warmup: claimed=%v error=%v", claimed, err)
+	}
+	if err := lineage.SaveWarmupLineage(t.Context(), f.task, repository.WarmupLineage{
+		Subject: f.source.Subject, ScenarioVersion: f.source.Version,
+		RenderingVersion: generation.CanonicalRenderingVersion, MaxTurns: len(f.source.Messages) + 1,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := repository.NewWarmupRepository(f.db.Pool).CreateWarmupToken(t.Context(), &models.WarmupToken{
+		Token: uuid.New(), TaskID: f.task, SenderAccountID: f.sender, RecipientAccountID: f.recipient,
+		Subject: f.source.Subject, ContentSource: "static", ConversationID: &f.source.ID,
+		ExpiresAt: time.Now().Add(24 * time.Hour),
+	}); err != nil {
+		t.Fatal(err)
+	}
 }

 type pausedOutboundValidation struct {
@@ -250,6 +278,7 @@ func TestLiveWarmupGenericStartAfterPartialImport(t *testing.T) {
 				status = "active"
 			}
 			f.importPendingSnapshot(t, status)
+			f.prepareDeferredWarmup(t)
 			if mode == "replacement_nonce" || mode == "cancel_replacement_nonce" {
 				nonce, err := f.dispatch.AuthorizeWarmupDispatch(t.Context(), f.task, f.sender, f.worker)
 				if err != nil || nonce == f.nonce {
@@ -265,7 +294,7 @@ func TestLiveWarmupGenericStartAfterPartialImport(t *testing.T) {
 				ValidateOutboundExecution(context.Context, uuid.UUID) error
 			})
 			if err := validator.ValidateOutboundExecution(t.Context(), f.task); err != nil {
-				t.Fatalf("imported warmup token is not executable: %v", err)
+				t.Fatalf("fresh local warmup preparation is not executable: %v", err)
 			}
 			if pause.refuse {
 				if err := warmupapp.NewService(repository.NewWarmupRepository(f.db.Pool)).RemoveFromAllPools(t.Context(), f.recipient); err != nil {
@@ -382,6 +411,99 @@ func TestLiveWarmupGenericStartAfterCompatibleOverwrite(t *testing.T) {
 	}
 }

+func TestLiveWarmupGenericStartAfterMailboxArchiveImport(t *testing.T) {
+	for _, sourceOnly := range []bool{false, true} {
+		name := "local_mailbox_rotation"
+		if sourceOnly {
+			name = "source_history"
+		}
+		t.Run(name, func(t *testing.T) {
+			f := newOutboundStartFixture(t)
+			enc, err := encrypt.NewEncrypter([]byte("0123456789abcdef0123456789abcdef"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			svc := orgtransfer.NewService(repository.NewOrgTransferRepository(f.db), nil, enc, nil, orgtransfer.InstanceInfo{})
+			const passphrase = "synthetic-mailbox-import-control"
+			var buf bytes.Buffer
+			if _, err := svc.ExportTo(t.Context(), f.org, orgtransfer.ExportOptions{Groups: []models.OrgDataGroup{models.OrgDataGroupSending}, Passphrase: passphrase}, &buf, nil); err != nil {
+				t.Fatal(err)
+			}
+			var generationBefore uuid.UUID
+			if err := f.db.QueryRow(t.Context(), `SELECT transport_generation FROM email_accounts WHERE id=$1`, f.sender).Scan(&generationBefore); err != nil {
+				t.Fatal(err)
+			}
+			if sourceOnly {
+				f.exec(t, `DELETE FROM tasks WHERE id=$1`, f.task)
+			}
+			result, err := svc.ImportFrom(t.Context(), f.org, bytes.NewReader(buf.Bytes()), orgtransfer.ImportOptions{Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.user, Passphrase: passphrase}, nil)
+			if err != nil || result == nil {
+				t.Fatalf("mailbox archive import: result=%+v error=%v", result, err)
+			}
+			var count int64
+			if sourceOnly {
+				count = 1
+			}
+			for _, table := range []string{"tasks", "warmup_tasks", "warmup_tokens"} {
+				if result.RowCounts[table] != count {
+					t.Fatalf("mailbox archive %s count=%d want=%d", table, result.RowCounts[table], count)
+				}
+			}
+			var generationAfter uuid.UUID
+			var worker *uuid.UUID
+			if err := f.db.QueryRow(t.Context(), `SELECT transport_generation,worker_id FROM email_accounts WHERE id=$1`, f.sender).Scan(&generationAfter, &worker); err != nil || generationAfter == generationBefore || worker == nil || *worker != f.worker {
+				t.Fatalf("mailbox import failed to rotate generation or preserve destination placement: before=%s after=%s worker=%v error=%v", generationBefore, generationAfter, worker, err)
+			}
+			var generic, dispatch *uuid.UUID
+			var lineage *int
+			if err := f.db.QueryRow(t.Context(), `SELECT t.send_executor_nonce,w.dispatch_nonce,w.lineage_version FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, f.task).Scan(&generic, &dispatch, &lineage); err != nil {
+				t.Fatal(err)
+			}
+			if sourceOnly {
+				if generic != nil || dispatch != nil || lineage != nil {
+					t.Fatalf("imported history retained source execution authority: generic=%v dispatch=%v lineage=%v", generic, dispatch, lineage)
+				}
+				var unknown bool
+				if err := f.db.QueryRow(t.Context(), `SELECT ea.send_recovery_hold AND t.send_result_state='unknown' AND t.send_result_applied_at IS NULL FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, f.task).Scan(&unknown); err != nil || !unknown {
+					t.Fatalf("imported uncertainty was cleared: unknown=%v error=%v", unknown, err)
+				}
+			} else if generic == nil || *generic != f.nonce || dispatch == nil || *dispatch != f.nonce || lineage == nil || *lineage != 1 {
+				t.Fatalf("mailbox rotation erased local identity: generic=%v dispatch=%v lineage=%v", generic, dispatch, lineage)
+			}
+			oldWorker := f.worker
+			before := f.snapshot(t)
+			f.worker = uuid.New()
+			status, state := f.request(t, f.svc, true, nil)
+			f.worker = oldWorker
+			wantStatus := http.StatusOK
+			if sourceOnly {
+				wantStatus = http.StatusServiceUnavailable
+			}
+			if status.Code != wantStatus || state == "execute" || state == "started" || f.snapshot(t) != before {
+				t.Fatalf("foreign worker acquired imported/local authority: status=%d state=%s", status.Code, state)
+			}
+			status, state = f.request(t, f.svc, true, nil)
+			t.Logf("sourceOnly=%v old command: status=%d state=%s", sourceOnly, status.Code, state)
+			if sourceOnly {
+				if status.Code != wantStatus || state == "execute" || state == "started" || f.snapshot(t) != before {
+					t.Fatalf("source history acquired execution authority: status=%d state=%s", status.Code, state)
+				}
+				result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "<unstarted-source@example.test>"}
+				response, _ := f.request(t, f.svc, false, &result)
+				if response.Code != http.StatusConflict || f.snapshot(t) != before {
+					t.Fatal("unstarted source history gained result authority", response.Code)
+				}
+			} else if status.Code != http.StatusOK || state != "execute" {
+				t.Fatalf("preserved local command lost execution authority: status=%d state=%s", status.Code, state)
+			}
+			var started bool
+			if err := f.db.QueryRow(t.Context(), `SELECT t.send_executor_started_at IS NOT NULL OR w.dispatch_started_at IS NOT NULL OR EXISTS(SELECT 1 FROM outbound_attempts WHERE task_id=t.id) FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, f.task).Scan(&started); err != nil || started == sourceOnly {
+				t.Fatalf("execution markers disagree with local/source authority: started=%v sourceOnly=%v error=%v", started, sourceOnly, err)
+			}
+		})
+	}
+}
+
 func TestLiveWarmupGenericCannotRepeatStartedWarmupClaim(t *testing.T) {
 	f := newOutboundStartFixture(t)
 	started, err := f.dispatch.BeginWarmupDispatch(t.Context(), f.task, f.sender, f.worker, f.nonce)
@@ -242,8 +242,8 @@ func TestLiveImportMailTransportWholeArchiveSendResultLockOrder(t *testing.T) {
 			if err := pool.QueryRow(ctx, `SELECT send_recovery_hold AND send_recovery_reason='authentication' AND send_recovery_task_id=$2 FROM email_accounts WHERE id=$1`, mailbox, task).Scan(&held); err != nil || !held {
 				t.Fatalf("restrictive hold changed %v %v", held, err)
 			}
-			if err := pool.QueryRow(ctx, `SELECT recorded_locally_at IS NULL AND confirmation_reference='Preserve archive lock proof' AND confirmed_result->>'message_id'='native-lock-proof' FROM send_recovery_resolutions WHERE id=$1`, audit).Scan(&historical); err != nil || !historical {
-				t.Fatalf("historical evidence/local provenance changed %v %v", historical, err)
+			if err := pool.QueryRow(ctx, `SELECT recorded_locally_at IS NOT NULL AND confirmation_reference='Preserve archive lock proof' AND confirmed_result->>'message_id'='native-lock-proof' FROM send_recovery_resolutions WHERE id=$1`, audit).Scan(&historical); err != nil || !historical {
+				t.Fatalf("existing local evidence/provenance changed %v %v", historical, err)
 			}
 			if lane != "result_first" {
 				var confirmed bool
@@ -35,12 +35,17 @@ func assertExportJSONEqual(t *testing.T, actual, expected []byte) {
 }

 func TestLiveImportMailTransportWholeArchiveOmitsGeneration(t *testing.T) {
-	for _, encrypted := range []bool{false, true} {
+	for _, mode := range []struct {
+		name             string
+		encrypted, local bool
+	}{
+		{"plaintext/source_only", false, false},
+		{"plaintext/destination_local", false, true},
+		{"encrypted/source_only", true, false},
+		{"encrypted/destination_local", true, true},
+	} {
 		for _, sent := range []bool{false, true} {
-			label := "plaintext"
-			if encrypted {
-				label = "encrypted"
-			}
+			label := mode.name
 			if sent {
 				label += "/sent"
 			} else {
@@ -99,7 +104,7 @@ func TestLiveImportMailTransportWholeArchiveOmitsGeneration(t *testing.T) {
 					t.Fatal(err)
 				}
 				passphrase := ""
-				if encrypted {
+				if mode.encrypted {
 					passphrase = "whole archive export regression passphrase"
 				}
 				var archive bytes.Buffer
@@ -107,7 +112,7 @@ func TestLiveImportMailTransportWholeArchiveOmitsGeneration(t *testing.T) {
 				if err != nil {
 					t.Fatal(err)
 				}
-				if (manifest.Secrets != nil) != encrypted {
+				if (manifest.Secrets != nil) != mode.encrypted {
 					t.Fatal("wrong export mode")
 				}
 				if len(manifest.Groups) != len(models.AllOrgDataGroups) {
@@ -212,6 +217,15 @@ func TestLiveImportMailTransportWholeArchiveOmitsGeneration(t *testing.T) {
 				if importTransportGeneration(t, pool, mailbox) != source {
 					t.Fatal("export changed source generation")
 				}
+				if !mode.local {
+					// Remove source send evidence so the import cannot reuse destination-local provenance.
+					if _, err := pool.Exec(t.Context(), `DELETE FROM send_recovery_resolutions WHERE id=$1`, audit); err != nil {
+						t.Fatal(err)
+					}
+					if _, err := pool.Exec(t.Context(), `DELETE FROM tasks WHERE id=ANY($1::uuid[])`, []uuid.UUID{task, warmTask}); err != nil {
+						t.Fatal(err)
+					}
+				}
 				if _, err := svc.ImportFrom(t.Context(), f.org, bytes.NewReader(archive.Bytes()), ImportOptions{Passphrase: passphrase, Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.owner}, nil); err != nil {
 					t.Fatal(err)
 				}
@@ -229,9 +243,31 @@ func TestLiveImportMailTransportWholeArchiveOmitsGeneration(t *testing.T) {
 				if err != nil {
 					t.Fatal(err)
 				}
-				assertExportJSONEqual(t, historical, portable)
-				if local || importTransportGeneration(t, pool, mailbox) == source {
-					t.Fatal("history gained authority or destination retained source proof")
+				want := portable
+				if mode.local {
+					want = raw
+				}
+				assertExportJSONEqual(t, historical, want)
+				if local != mode.local || importTransportGeneration(t, pool, mailbox) == source {
+					t.Fatal("send evidence lost its origin or mailbox retained old transport authority")
+				}
+				for _, item := range []struct {
+					query    string
+					id       uuid.UUID
+					expected []byte
+				}{
+					{`SELECT send_executor_result FROM tasks WHERE id=$1`, task, raw},
+					{`SELECT dispatch_result FROM warmup_tasks WHERE task_id=$1`, warmTask, warmRaw},
+				} {
+					var current []byte
+					if err := pool.QueryRow(t.Context(), item.query, item.id).Scan(&current); err != nil {
+						t.Fatal(err)
+					}
+					if mode.local {
+						assertExportJSONEqual(t, current, item.expected)
+					} else if len(current) != 0 {
+						t.Fatal("imported result retained source execution authority")
+					}
 				}
 			})
 		}
@@ -0,0 +1,388 @@
+package orgtransfer
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func (f *selectiveHoldReviewFixture) localRows(t *testing.T) string {
+	t.Helper()
+	var snapshot string
+	if err := f.pool.QueryRow(t.Context(), `SELECT jsonb_build_object('task',to_jsonb(t),
+		'email',(SELECT to_jsonb(e) FROM email_tasks e WHERE e.task_id=t.id),
+		'warmup',(SELECT to_jsonb(w) FROM warmup_tasks w WHERE w.task_id=t.id),
+		'tokens',(SELECT jsonb_agg(to_jsonb(w) ORDER BY w.token) FROM warmup_tokens w WHERE w.task_id=t.id),
+		'attempts',(SELECT jsonb_agg(to_jsonb(a) ORDER BY a.nonce) FROM outbound_attempts a WHERE a.task_id=t.id),
+		'resolutions',(SELECT jsonb_agg(to_jsonb(r) ORDER BY r.id) FROM send_recovery_resolutions r WHERE r.recovery_task_id=t.id),
+		'failures',(SELECT jsonb_agg(to_jsonb(f)) FROM task_failures f WHERE f.task_id=t.id),
+		'dead_letters',(SELECT jsonb_agg(to_jsonb(d) ORDER BY d.id) FROM task_dead_letters d WHERE d.task_id=t.id),
+		'execution_keys',(SELECT jsonb_agg(to_jsonb(k)) FROM task_execution_keys k WHERE k.task_id=t.id),
+		'receipts',(SELECT jsonb_agg(to_jsonb(r) ORDER BY r.internal_id) FROM warmup_received r WHERE r.email_account_id=t.email_account_id))::text
+		FROM tasks t WHERE t.id=$1`, f.task).Scan(&snapshot); err != nil {
+		t.Fatal(err)
+	}
+	return snapshot
+}
+
+func TestLiveImportLocalSendIdentityPayloadAndUnrelated(t *testing.T) {
+	f := newSelectiveHoldReviewFixture(t)
+	other := uuid.New()
+	f.exec(t, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,status) VALUES($1,$2,$3,$4,'Other','','','smtp_imap','active')`, other, f.tenant.owner, f.tenant.org, other.String()+"@fixture.invalid")
+	if err := f.repo.UpdateTaskEmailAccount(t.Context(), f.task, other); err != nil {
+		t.Fatal(err)
+	}
+	groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+	archive := f.archive(t, groups)
+	if err := f.repo.UpdateTaskEmailAccount(t.Context(), f.task, f.mailbox); err != nil {
+		t.Fatal(err)
+	}
+	f.exec(t, `UPDATE email_tasks SET subject='Current payload',body='Current body',body_plain='Current body' WHERE task_id=$1`, f.task)
+	f.exec(t, `UPDATE email_accounts SET name='Current unrelated metadata' WHERE id=$1`, other)
+	f.reserve(t)
+	f.begin(t)
+	before := f.localRows(t)
+	f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+	if after := f.localRows(t); after != before {
+		t.Error("older native archive changed current task identity, type, reservation or typed payload")
+	}
+	var name string
+	if err := f.pool.QueryRow(t.Context(), `SELECT name FROM email_accounts WHERE id=$1`, other).Scan(&name); err != nil || name != "Other" {
+		t.Fatalf("unrelated mailbox overwrite was improperly suppressed: %q %v", name, err)
+	}
+	f.finish(t)
+	calls := 0
+	if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), f.result(), func(inner context.Context) error {
+		calls++
+		return f.repo.UpdateTaskMessageID(inner, f.task, f.result().MessageID)
+	}); err != nil || calls != 1 {
+		t.Fatalf("matching original mailbox/payload result refused after reparent archive: %d %v", calls, err)
+	}
+}
+
+type failLocalImportRepository struct {
+	repository.OrgTransferRepository
+	table string
+}
+
+var localImportFailure = errors.New("synthetic local import failure")
+
+func (r *failLocalImportRepository) InsertBatch(ctx context.Context, tx pgx.Tx, table string, cols []string, rows []json.RawMessage, conflict models.OrgImportConflict, pk []string, owner string, org uuid.UUID, resets []string, invalidations []repository.ImportInvalidation) (int64, error) {
+	if table == r.table {
+		return 0, localImportFailure
+	}
+	return r.OrgTransferRepository.InsertBatch(ctx, tx, table, cols, rows, conflict, pk, owner, org, resets, invalidations)
+}
+
+func TestLiveImportLocalSendRollbackForeignAndRestrictiveConflicts(t *testing.T) {
+	for _, kind := range []string{"rollback", "foreign", "conflicting_holds", "local_conflict", "local_authentication"} {
+		t.Run(kind, func(t *testing.T) {
+			f := newSelectiveHoldReviewFixture(t)
+			groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+			archive := f.archive(t, groups)
+			f.reserve(t)
+			f.begin(t)
+			if kind == "conflicting_holds" {
+				archive = f.archive(t, groups)
+			}
+			result := f.result()
+			if kind == "local_authentication" {
+				result.Success = false
+				result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendAuth, Scope: "mailbox", ObservedAt: time.Now()}}
+			}
+			if kind == "local_authentication" || kind == "local_conflict" || kind == "conflicting_holds" {
+				if err := f.admission.FinishOutbound(t.Context(), f.task, f.mailbox, f.worker, result); err != nil {
+					t.Fatal(err)
+				}
+				if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, func(context.Context) error { return nil }); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if kind == "local_conflict" {
+				contrary := result
+				contrary.Success = false
+				contrary.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient", ObservedAt: time.Now()}}
+				if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), contrary, func(context.Context) error { t.Error("conflict repeated accounting"); return nil }); err != nil {
+					t.Fatal(err)
+				}
+				archive = f.archive(t, groups)
+			}
+			if kind == "conflicting_holds" {
+				newTask := uuid.New()
+				f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','active','')`, newTask, f.mailbox)
+				f.exec(t, `INSERT INTO email_tasks(task_id,to_addrs,subject,body,body_plain,encrypted) VALUES($1,ARRAY['recipient@fixture.invalid'],'New local','Synthetic body','Synthetic body',false)`, newTask)
+				if _, err := f.admission.ReserveOutbound(t.Context(), repository.OutboundReservation{TaskID: newTask, MailboxID: f.mailbox, OrganizationID: f.tenant.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{"recipient@fixture.invalid"}}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			before, beforeRows := f.snapshot(t), f.localRows(t)
+			var generation string
+			if err := f.pool.QueryRow(t.Context(), `SELECT transport_generation::text FROM email_accounts WHERE id=$1`, f.mailbox).Scan(&generation); err != nil {
+				t.Fatal(err)
+			}
+			org, actor := f.tenant.org, f.tenant.owner
+			var want error
+			if kind == "rollback" {
+				service := f.svc.(*service)
+				service.repo = &failLocalImportRepository{OrgTransferRepository: service.repo, table: "email_tasks"}
+				want = localImportFailure
+			} else if kind == "foreign" {
+				other := newTenantFixture(t, f.pool)
+				org, actor = other.org, other.owner
+				want = ErrForeignRows
+			} else if kind == "conflicting_holds" {
+				want = repository.ErrImportSendRecoveryConflict
+			}
+			_, err := f.svc.ImportFrom(t.Context(), org, bytes.NewReader(archive), ImportOptions{Groups: groups, Passphrase: "synthetic-review-passphrase", Conflict: models.OrgImportConflictOverwrite, ActorUserID: actor}, nil)
+			if !errors.Is(err, want) {
+				t.Fatalf("import result %v, want %v", err, want)
+			}
+			if f.snapshot(t) != before || f.localRows(t) != beforeRows {
+				t.Error("failed import or restrictive archive overwrite changed local recovery projection/evidence")
+			}
+			if want != nil {
+				var after string
+				if err := f.pool.QueryRow(t.Context(), `SELECT transport_generation::text FROM email_accounts WHERE id=$1`, f.mailbox).Scan(&after); err != nil || after != generation {
+					t.Fatalf("failed import did not roll back mailbox generation: %v", err)
+				}
+			}
+		})
+	}
+}
+
+func nativeWarmupArchiveFixture(t *testing.T, reply bool) (*selectiveHoldReviewFixture, uuid.UUID) {
+	t.Helper()
+	f := newSelectiveHoldReviewFixture(t)
+	recipient := uuid.New()
+	f.exec(t, `UPDATE organizations SET risk_state='trusted' WHERE id=$1`, f.tenant.org)
+	f.exec(t, `UPDATE email_accounts SET warmup=NOW(),warmup_days=127 WHERE id=$1`, f.mailbox)
+	f.exec(t, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,status,worker_id,timezone,warmup,warmup_days,campaign_limit,warmup_max,shared_daily_limit,rolling_recipient_limit) VALUES($1,$2,$3,$4,'Native recipient','','','smtp_imap','active',$5,'UTC',NOW(),127,50,50,100,100)`, recipient, f.tenant.owner, f.tenant.org, recipient.String()+"@fixture.invalid", f.worker)
+	f.exec(t, `INSERT INTO email_accounts_smtp_imap(email_account_id,smtp_host,smtp_port,smtp_user,smtp_password,imap_host,imap_port,imap_user,imap_password) SELECT $1,smtp_host,smtp_port,smtp_user,smtp_password,imap_host,imap_port,imap_user,imap_password FROM email_accounts_smtp_imap WHERE email_account_id=$2`, recipient, f.mailbox)
+	f.exec(t, `UPDATE email_accounts SET test_mode=NULL WHERE id IN($1,$2)`, recipient, f.mailbox)
+	f.exec(t, `INSERT INTO warmup_pool_participants(pool_id,email_account_id,participant_role,health_state) SELECT p.id,ea.id,'sender_receiver','healthy' FROM warmup_pools p CROSS JOIN email_accounts ea WHERE p.pool_type='free' AND ea.id IN($1,$2)`, recipient, f.mailbox)
+	f.exec(t, `DELETE FROM email_tasks WHERE task_id=$1`, f.task)
+	f.exec(t, `UPDATE tasks SET task_type='warmup',status='pending',scheduled_at=NOW()-INTERVAL '1 minute' WHERE id=$1`, f.task)
+	f.exec(t, `INSERT INTO warmup_tasks(task_id,subject) VALUES($1,'Archive opening')`, f.task)
+	lineage := f.repo.(repository.WarmupLineageRepository)
+	if reply {
+		parent := uuid.New()
+		f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'warmup','active','')`, parent, recipient)
+		f.exec(t, `INSERT INTO warmup_tasks(task_id,target_account_id,subject) VALUES($1,$2,'Native parent')`, parent, f.mailbox)
+		if err := lineage.SaveWarmupLineage(t.Context(), parent, repository.WarmupLineage{Subject: "Native parent", ScenarioVersion: "diagnostic-v1", RenderingVersion: "canonical-v1", MaxTurns: 3}); err != nil {
+			t.Fatal(err)
+		}
+		f.exec(t, `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id,subject,conversation_id,conversation_turn,content_source) VALUES($1,$2,$3,$4,'Native parent',$5,0,'static')`, uuid.New(), parent, recipient, f.mailbox, uuid.New())
+		if _, err := f.repo.(repository.WarmupDispatchRepository).AuthorizeWarmupDispatch(t.Context(), parent, recipient, f.worker); err != nil {
+			t.Fatal(err)
+		}
+		nonce, err := f.admission.ReserveOutbound(t.Context(), repository.OutboundReservation{TaskID: parent, MailboxID: recipient, OrganizationID: f.tenant.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{f.mailbox.String() + "@example.test"}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		start, err := f.admission.BeginOutbound(t.Context(), parent, recipient, f.worker, nonce)
+		if err != nil || start == nil || start.State != "execute" {
+			t.Fatalf("real parent start denied: %+v %v", start, err)
+		}
+		result := models.SendEmailResult{TaskID: parent, Success: true, MessageID: "<parent@fixture.invalid>", SentAt: time.Now()}
+		if err := f.admission.FinishOutbound(t.Context(), parent, recipient, f.worker, result); err != nil {
+			t.Fatal(err)
+		}
+		warmup := repository.NewWarmupRepository(f.pool)
+		if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, func(inner context.Context) error {
+			return warmup.RecordWarmupTokenDelivery(inner, parent, result.MessageID)
+		}); err != nil {
+			t.Fatal(err)
+		}
+		receipt := uuid.New()
+		f.exec(t, `INSERT INTO warmup_received(email_account_id,internal_id,sender_account_id,message_id) VALUES($1,$2,$3,'parent@fixture.invalid')`, f.mailbox, receipt, recipient)
+		if err := lineage.RecordVerifiedWarmupParent(t.Context(), parent, f.mailbox, receipt, "native-local-thread", repository.WarmupReceiptProof{SenderAddress: recipient.String() + "@fixture.invalid", MessageID: result.MessageID}); err != nil {
+			t.Fatal(err)
+		}
+		if bound, err := lineage.BindWarmupSuccessor(t.Context(), parent, f.mailbox, receipt, time.Now().Add(-time.Second)); err != nil || !bound {
+			t.Fatalf("native receipt did not bind current reply: %v %v", bound, err)
+		}
+		if err := lineage.RescheduleWarmupTask(t.Context(), f.task, time.Now().Add(-time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+	} else {
+		f.exec(t, `UPDATE warmup_tasks SET target_account_id=$2 WHERE task_id=$1`, f.task, recipient)
+	}
+	if err := lineage.SaveWarmupLineage(t.Context(), f.task, repository.WarmupLineage{Subject: "Pinned local", ScenarioVersion: "diagnostic-v1", RenderingVersion: "canonical-v1", MaxTurns: 3}); err != nil {
+		t.Fatal(err)
+	}
+	f.exec(t, `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id,subject,conversation_id,conversation_turn,content_source) VALUES($1,$2,$3,$4,'Pinned local',$5,$6,'static')`, uuid.New(), f.task, f.mailbox, recipient, uuid.New(), map[bool]int{false: 0, true: 1}[reply])
+	f.exec(t, `INSERT INTO warmup_statistics(email_account_id,date,emails_sent,emails_replied,target_volume) VALUES($1,(NOW() AT TIME ZONE 'UTC')::date,0,0,10)`, f.mailbox)
+	return f, recipient
+}
+
+func TestLiveImportLocalWarmupChargeReplyReceiptAndLateAccounting(t *testing.T) {
+	for _, reply := range []bool{false, true} {
+		for _, success := range []bool{false, true} {
+			for _, mode := range []string{"core", "sending", "sending_warmup"} {
+				label := mode + "/" + map[bool]string{false: "opening", true: "reply"}[reply] + "/" + map[bool]string{false: "failed", true: "sent"}[success]
+				t.Run(label, func(t *testing.T) {
+					f, recipient := nativeWarmupArchiveFixture(t, reply)
+					groups := []models.OrgDataGroup{models.OrgDataGroupCore}
+					if mode != "core" {
+						groups = []models.OrgDataGroup{models.OrgDataGroupSending}
+					}
+					if mode == "sending_warmup" {
+						groups = append(groups, models.OrgDataGroupWarmup)
+					}
+					archive := f.archive(t, groups)
+					f.exec(t, `UPDATE warmup_tokens SET token=gen_random_uuid(),subject='Current pinned token' WHERE task_id=$1`, f.task)
+					if claimed, err := f.repo.(repository.WarmupLineageRepository).ClaimWarmupTask(t.Context(), f.task, time.Now()); err != nil || !claimed {
+						t.Fatalf("real warmup claim failed: %v %v", claimed, err)
+					}
+					if _, err := f.repo.(repository.WarmupDispatchRepository).AuthorizeWarmupDispatch(t.Context(), f.task, f.mailbox, f.worker); err != nil {
+						t.Fatal(err)
+					}
+					var err error
+					f.nonce, err = f.admission.ReserveOutbound(t.Context(), repository.OutboundReservation{TaskID: f.task, MailboxID: f.mailbox, OrganizationID: f.tenant.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{recipient.String() + "@fixture.invalid"}})
+					if err != nil {
+						t.Fatal(err)
+					}
+					start, err := f.admission.BeginOutbound(t.Context(), f.task, f.mailbox, f.worker, f.nonce)
+					if err != nil || start == nil || start.State != "execute" {
+						t.Fatalf("real current native warmup start denied: %+v %v", start, err)
+					}
+					before := f.localRows(t)
+					booked, err := f.repo.GetBookedSendsForAccount(t.Context(), f.mailbox, time.Now())
+					if err != nil {
+						t.Fatal(err)
+					}
+					f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+					if after := f.localRows(t); after != before {
+						t.Error("native import changed warmup task/binding/payload/token/lineage/charge/attempt/receipt evidence")
+					}
+					bookedAfter, err := f.repo.GetBookedSendsForAccount(t.Context(), f.mailbox, time.Now())
+					if err != nil || len(bookedAfter) != len(booked) {
+						t.Errorf("native import changed booked sends: %d -> %d %v", len(booked), len(bookedAfter), err)
+					}
+					var sent, replies int
+					checkCounts := func(wantSent, wantReply int) {
+						t.Helper()
+						if err := f.pool.QueryRow(t.Context(), `SELECT emails_sent,emails_replied FROM warmup_statistics WHERE email_account_id=$1 AND date=(NOW() AT TIME ZONE 'UTC')::date`, f.mailbox).Scan(&sent, &replies); err != nil || sent != wantSent || replies != wantReply {
+							t.Errorf("current known charge/refund wrong: sent/replies=%d/%d want=%d/%d err=%v", sent, replies, wantSent, wantReply, err)
+						}
+					}
+					checkCounts(1, map[bool]int{false: 0, true: 1}[reply])
+					result := f.result()
+					result.Success = success
+					if !success {
+						result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient", ObservedAt: time.Now()}}
+					}
+					if err := f.admission.FinishOutbound(t.Context(), f.task, f.mailbox, f.worker, result); err != nil {
+						t.Fatal(err)
+					}
+					if err := f.repo.(repository.WarmupDispatchRepository).FinishWarmupDispatch(t.Context(), f.task, f.mailbox, f.worker, result); err != nil {
+						t.Fatal(err)
+					}
+					warmup := repository.NewWarmupRepository(f.pool)
+					calls, effects := 0, 0
+					apply := func(inner context.Context) error {
+						calls++
+						repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+						if !success {
+							return warmup.FailWarmupSend(inner, f.mailbox, f.task, time.Now(), "Native refused", "Synthetic local refusal")
+						}
+						if err := f.repo.UpdateTaskMessageID(inner, f.task, result.MessageID); err != nil {
+							return err
+						}
+						return warmup.RecordWarmupTokenDelivery(inner, f.task, result.MessageID)
+					}
+					for range 2 {
+						if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, apply); err != nil {
+							t.Fatal(err)
+						}
+					}
+					if calls != 1 || effects != 1 {
+						t.Errorf("late result accounting/postcommit effects repeated: %d/%d", calls, effects)
+					}
+					checkCounts(map[bool]int{false: 0, true: 1}[success], map[bool]int{false: 0, true: 1}[success && reply])
+					var correct bool
+					if err := f.pool.QueryRow(t.Context(), `SELECT t.send_result_applied_at IS NOT NULL AND t.send_result_state=$2 AND w.warmup_charged_date IS NOT NULL AND w.warmup_reply_charged=$3 AND (w.warmup_refunded_at IS NULL)=$4 AND (wt.sent_message_id<>'')=$4 FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id JOIN warmup_tokens wt ON wt.task_id=t.id WHERE t.id=$1`, f.task, map[bool]string{false: "failed", true: "sent"}[success], reply, success).Scan(&correct); err != nil || !correct {
+						t.Errorf("late native result lost applied state, original known charge or token delivery/refund: %v %v", correct, err)
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestLiveImportFreshWarmupReservationDuringMailboxImport(t *testing.T) {
+	f, recipient := nativeWarmupArchiveFixture(t, false)
+	groups := []models.OrgDataGroup{models.OrgDataGroupSending, models.OrgDataGroupWarmup}
+	archive := f.archive(t, groups)
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+	ready, resume := make(chan struct{}), make(chan struct{})
+	t.Cleanup(func() {
+		select {
+		case <-resume:
+		default:
+			close(resume)
+		}
+	})
+	done := make(chan error, 1)
+	go func() {
+		_, err := f.svc.ImportFrom(ctx, f.tenant.org, bytes.NewReader(archive), ImportOptions{Groups: groups, Passphrase: "synthetic-review-passphrase", Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.tenant.owner}, func(_ int, stage string) {
+			if stage == "importing email_accounts" {
+				close(ready)
+				select {
+				case <-resume:
+				case <-ctx.Done():
+				}
+			}
+		})
+		done <- err
+	}()
+	select {
+	case <-ready:
+	case err := <-done:
+		t.Fatalf("import ended before mailbox barrier: %v", err)
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	f.task = uuid.New()
+	f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,scheduled_at,message_id) VALUES($1,$2,'warmup','pending',NOW()-INTERVAL '1 minute','')`, f.task, f.mailbox)
+	if err := f.repo.CreateWarmupTask(ctx, &repository.WarmupTask{TaskID: f.task, TargetAccountID: &recipient}); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.repo.(repository.WarmupLineageRepository).SaveWarmupLineage(ctx, f.task, repository.WarmupLineage{Subject: "Native late creation", ScenarioVersion: "diagnostic-v1", RenderingVersion: "canonical-v1", MaxTurns: 3}); err != nil {
+		t.Fatal(err)
+	}
+	f.exec(t, `INSERT INTO warmup_tokens(token,task_id,sender_account_id,recipient_account_id,subject,conversation_id,conversation_turn,content_source) VALUES($1,$2,$3,$4,'Native late creation',$5,0,'static')`, uuid.New(), f.task, f.mailbox, recipient, uuid.New())
+	if claimed, err := f.repo.(repository.WarmupLineageRepository).ClaimWarmupTask(ctx, f.task, time.Now()); err != nil || !claimed {
+		t.Fatalf("new warmup task not claimed while import waits: %v %v", claimed, err)
+	}
+	if _, err := f.repo.(repository.WarmupDispatchRepository).AuthorizeWarmupDispatch(ctx, f.task, f.mailbox, f.worker); err != nil {
+		t.Fatal(err)
+	}
+	var err error
+	f.nonce, err = f.admission.ReserveOutbound(ctx, repository.OutboundReservation{TaskID: f.task, MailboxID: f.mailbox, OrganizationID: f.tenant.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{recipient.String() + "@fixture.invalid"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	before, beforeRows := f.snapshot(t), f.localRows(t)
+	close(resume)
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	if after := f.snapshot(t); after != before || f.localRows(t) != beforeRows {
+		t.Error("import's mailbox table inventory lost a new local warmup reservation, charge, hold or typed state")
+	}
+	start, err := f.admission.BeginOutbound(ctx, f.task, f.mailbox, f.worker, f.nonce)
+	if err != nil || start == nil || start.State != "execute" {
+		t.Errorf("fresh new local reservation did not remain executable: %+v %v", start, err)
+	}
+}
@@ -0,0 +1,798 @@
+package orgtransfer
+
+import (
+	"archive/zip"
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/errx"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func TestLiveImportLegacyUnknownPreservesOtherOutcome(t *testing.T) {
+	for _, group := range []models.OrgDataGroup{models.OrgDataGroupCore, models.OrgDataGroupSending} {
+		for _, mode := range []models.OrgImportConflict{models.OrgImportConflictSkip, models.OrgImportConflictOverwrite} {
+			t.Run(string(group)+"/"+string(mode), func(t *testing.T) {
+				f := newSelectiveHoldReviewFixture(t)
+				other := uuid.New()
+				f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','active','')`, other, f.mailbox)
+				f.exec(t, `INSERT INTO email_tasks(task_id,to_addrs,subject,body,body_plain,encrypted) VALUES($1,ARRAY['other@fixture.invalid'],'Other','Other','Other',false)`, other)
+				groups := []models.OrgDataGroup{group}
+				archive := f.archive(t, groups)
+				recovery := f.repo.(repository.SendResultRecovery)
+				calls, effects := 0, 0
+				apply := func(inner context.Context) error {
+					calls++
+					repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+					return nil
+				}
+				for _, task := range []uuid.UUID{f.task, other} {
+					if err := f.repo.UpdateTaskStatus(t.Context(), task, "completed"); err != nil {
+						t.Fatal(err)
+					}
+					if err := recovery.ApplySendResult(t.Context(), models.SendEmailResult{TaskID: task}, apply); err != nil {
+						t.Fatal(err)
+					}
+					assertLegacyArchiveTask(t, f, task)
+				}
+				if calls != 0 || effects != 0 {
+					t.Fatalf("unknown outcomes invoked effects: %d %d", calls, effects)
+				}
+				otherFixture := *f
+				otherFixture.task = other
+				before, beforeOther := f.localRows(t), otherFixture.localRows(t)
+				var heldTask uuid.UUID
+				if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_task_id FROM email_accounts WHERE id=$1 AND send_recovery_hold`, f.mailbox).Scan(&heldTask); err != nil {
+					t.Fatal(err)
+				}
+				assertLegacyArchiveUnknowns(t, f, 2)
+				f.importArchive(t, archive, groups, mode)
+				assertLegacyArchiveUnknowns(t, f, 2)
+				if f.localRows(t) != before || otherFixture.localRows(t) != beforeOther {
+					t.Error("import rewound one of two legacy unknown tasks or typed payloads")
+				}
+				result := models.SendEmailResult{TaskID: heldTask, Success: true, MessageID: "<fresh-legacy@fixture.invalid>", SentAt: time.Now()}
+				if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil {
+					t.Fatal(err)
+				}
+				assertLegacyArchiveUnknowns(t, f, 1)
+				assertHold := func() {
+					t.Helper()
+					var held bool
+					if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_hold AND send_recovery_reason='unknown' AND send_recovery_task_id<>$2 FROM email_accounts WHERE id=$1`, f.mailbox, heldTask).Scan(&held); err != nil || !held {
+						t.Fatalf("settling known result lost other unknown hold: %v %v", held, err)
+					}
+				}
+				assertHold()
+				f.importArchive(t, archive, groups, mode)
+				for range 2 {
+					if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if calls != 1 || effects != 1 {
+					t.Fatalf("repeat archive/result repeated effects: %d %d", calls, effects)
+				}
+				assertLegacyArchiveUnknowns(t, f, 1)
+				assertHold()
+			})
+		}
+	}
+}
+func TestLiveImportLegacyAppliedResultRemainsIdempotent(t *testing.T) {
+	for _, group := range []models.OrgDataGroup{models.OrgDataGroupCore, models.OrgDataGroupSending} {
+		for _, success := range []bool{false, true} {
+			t.Run(string(group)+"/"+map[bool]string{false: "failed", true: "sent"}[success], func(t *testing.T) {
+				f := newSelectiveHoldReviewFixture(t)
+				groups := []models.OrgDataGroup{group}
+				archive := f.archive(t, groups)
+				if err := f.repo.UpdateTaskStatus(t.Context(), f.task, "completed"); err != nil {
+					t.Fatal(err)
+				}
+				result := models.SendEmailResult{TaskID: f.task, Success: success, MessageID: "<legacy-local@fixture.invalid>", SentAt: time.Now()}
+				if !success {
+					result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient", ObservedAt: time.Now()}}
+				}
+				calls, effects := 0, 0
+				apply := func(inner context.Context) error {
+					calls++
+					repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+					if success {
+						return f.repo.UpdateTaskMessageID(inner, f.task, result.MessageID)
+					}
+					return f.repo.UpdateTaskStatus(inner, f.task, "failed")
+				}
+				recovery := f.repo.(repository.SendResultRecovery)
+				if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil {
+					t.Fatal(err)
+				}
+				var noNonce, applied bool
+				if err := f.pool.QueryRow(t.Context(), `SELECT send_executor_nonce IS NULL,send_result_applied_at IS NOT NULL FROM tasks WHERE id=$1`, f.task).Scan(&noNonce, &applied); err != nil || !noNonce || !applied || calls != 1 || effects != 1 {
+					t.Fatalf("native nonce-less applied control invalid: nonceNull=%v applied=%v calls=%d effects=%d err=%v", noNonce, applied, calls, effects, err)
+				}
+				before := f.localRows(t)
+				f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+				if after := f.localRows(t); after != before {
+					t.Error("archive rewound the applied legacy result, task identity, status, or typed payload")
+				}
+				if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil || calls != 1 || effects != 1 {
+					t.Errorf("applied result effects repeated: calls=%d effects=%d err=%v", calls, effects, err)
+				}
+				assertLegacyArchiveUnknowns(t, f, 0)
+				assertLegacyArchiveNoHold(t, f)
+			})
+		}
+	}
+}
+
+func TestLiveImportLegacyStampedTaskAcceptsLaterResult(t *testing.T) {
+	t.Run("completion_during_source_hold_import", func(t *testing.T) { testLegacyCompletionDuringSourceHoldImport(t, false) })
+	t.Run("completion_after_insert_snapshot", func(t *testing.T) { testLegacyCompletionDuringSourceHoldImport(t, true) })
+	t.Run("completed_before_import", testLegacyStampedBeforeImport)
+}
+
+func testLegacyStampedBeforeImport(t *testing.T) {
+	f := newSelectiveHoldReviewFixture(t)
+	groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+	archive := f.archive(t, groups)
+	if err := f.repo.UpdateTaskStatus(t.Context(), f.task, "completed"); err != nil {
+		t.Fatal(err)
+	}
+	before := f.localRows(t)
+	f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+	if after := f.localRows(t); after != before {
+		t.Error("import rewound a locally stamped legacy task before its result arrived")
+	}
+	result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "<late-legacy@fixture.invalid>", SentAt: time.Now()}
+	calls, effects := 0, 0
+	apply := func(inner context.Context) error {
+		calls++
+		repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+		return f.repo.UpdateTaskMessageID(inner, f.task, result.MessageID)
+	}
+	for range 2 {
+		if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, apply); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var correctlySettled bool
+	if err := f.pool.QueryRow(t.Context(), `SELECT status='completed' AND send_result_state='sent' AND send_result_applied_at IS NOT NULL AND send_executor_nonce IS NULL AND message_id=$2 FROM tasks WHERE id=$1`, f.task, result.MessageID).Scan(&correctlySettled); err != nil || !correctlySettled || calls != 1 || effects != 1 {
+		t.Errorf("valid late legacy result not applied once: state=%v calls=%d effects=%d err=%v", correctlySettled, calls, effects, err)
+	}
+	assertLegacyArchiveUnknowns(t, f, 0)
+	assertLegacyArchiveNoHold(t, f)
+}
+
+func testLegacyCompletionDuringSourceHoldImport(t *testing.T, afterSnapshot bool) {
+	f := newSelectiveHoldReviewFixture(t)
+	groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+	older := f.archive(t, groups)
+	if err := f.repo.UpdateTaskStatus(t.Context(), f.task, "completed"); err != nil {
+		t.Fatal(err)
+	}
+	if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), models.SendEmailResult{TaskID: f.task}, nil); err != nil {
+		t.Fatal(err)
+	}
+	newer := f.archive(t, groups)
+	f.exec(t, `DELETE FROM email_accounts WHERE id=$1`, f.mailbox)
+	f.importArchive(t, older, groups, models.OrgImportConflictOverwrite)
+	ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
+	defer cancel()
+	importPID := make(chan uint32, 1)
+	if afterSnapshot {
+		service := f.svc.(*service)
+		service.repo = &observedWholeImportRepository{OrgTransferRepository: service.repo, pid: importPID}
+		name := "fixture_completion_" + strings.ReplaceAll(f.task.String(), "-", "")
+		f.exec(t, fmt.Sprintf(`CREATE FUNCTION %s() RETURNS trigger LANGUAGE plpgsql AS $$
+			BEGIN PERFORM pg_advisory_xact_lock(hashtextextended(NEW.id::text,9900001)); RETURN NEW; END $$`, name))
+		f.exec(t, fmt.Sprintf(`CREATE TRIGGER %s BEFORE INSERT ON tasks FOR EACH ROW WHEN (NEW.id='%s') EXECUTE FUNCTION %s()`, name, f.task, name))
+		defer f.exec(t, fmt.Sprintf(`DROP TRIGGER %s ON tasks; DROP FUNCTION %s()`, name, name))
+	}
+	atTasks, resume := make(chan struct{}), make(chan struct{})
+	done := make(chan error, 1)
+	go func() {
+		_, err := f.svc.ImportFrom(ctx, f.tenant.org, bytes.NewReader(newer), ImportOptions{
+			Groups: groups, Passphrase: "synthetic-review-passphrase", Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.tenant.owner,
+		}, func(_ int, stage string) {
+			if stage == "importing tasks" {
+				close(atTasks)
+				select {
+				case <-resume:
+				case <-ctx.Done():
+				}
+			}
+		})
+		done <- err
+	}()
+	select {
+	case <-atTasks:
+	case err := <-done:
+		t.Fatalf("import failed before barrier: %v", err)
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	var release func()
+	if afterSnapshot {
+		gate, err := f.pool.Begin(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer gate.Rollback(context.Background())
+		if _, err := gate.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1::text,9900001))`, f.task); err != nil {
+			t.Fatal(err)
+		}
+		close(resume)
+		pid := <-importPID
+		ticker := time.NewTicker(10 * time.Millisecond)
+		defer ticker.Stop()
+		for {
+			var blocked bool
+			if err := f.pool.QueryRow(ctx, `SELECT $2::int=ANY(pg_blocking_pids($1))`, pid, gate.Conn().PgConn().PID()).Scan(&blocked); err != nil {
+				t.Fatal(err)
+			}
+			if blocked {
+				break
+			}
+			select {
+			case <-ticker.C:
+			case <-ctx.Done():
+				t.Fatal("import did not reach the before-insert gate", ctx.Err())
+			}
+		}
+		release = func() {
+			if err := gate.Commit(ctx); err != nil {
+				t.Fatal(err)
+			}
+		}
+	} else {
+		release = func() { close(resume) }
+	}
+	if err := f.repo.UpdateTaskStatus(ctx, f.task, "completed"); err != nil {
+		t.Fatal(err)
+	}
+	var before string
+	if err := f.pool.QueryRow(ctx, `SELECT completed_at::text FROM tasks WHERE id=$1`, f.task).Scan(&before); err != nil {
+		t.Fatal(err)
+	}
+	release()
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	var retained bool
+	if err := f.pool.QueryRow(ctx, `SELECT status='completed' AND completed_at::text=$2 AND send_result_state IS NULL AND send_executor_nonce IS NULL
+		FROM tasks WHERE id=$1`, f.task, before).Scan(&retained); err != nil || !retained {
+		t.Errorf("import ignored genuinely new local completion after source-hold classification: retained=%v err=%v", retained, err)
+	}
+	result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "<late-during-import@fixture.invalid>", SentAt: time.Now()}
+	calls, effects := 0, 0
+	for range 2 {
+		if err := f.repo.(repository.SendResultRecovery).ApplySendResult(ctx, result, func(inner context.Context) error {
+			calls++
+			repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+			return f.repo.UpdateTaskMessageID(inner, f.task, result.MessageID)
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if calls != 1 || effects != 1 {
+		t.Fatalf("late local result effects repeated: %d %d", calls, effects)
+	}
+	assertLegacyArchiveUnknowns(t, f, 0)
+	assertLegacyArchiveNoHold(t, f)
+}
+
+func TestLiveImportRefundedWarmupChargeWithoutNonces(t *testing.T) {
+	f, _ := nativeWarmupArchiveFixture(t, false)
+	groups := []models.OrgDataGroup{models.OrgDataGroupSending, models.OrgDataGroupWarmup}
+	archive := f.archive(t, groups)
+	lineage := f.repo.(repository.WarmupLineageRepository)
+	if claimed, err := lineage.ClaimWarmupTask(t.Context(), f.task, time.Now()); err != nil || !claimed {
+		t.Fatalf("real warmup claim failed: %v %v", claimed, err)
+	}
+	dispatch := f.repo.(repository.WarmupDispatchRepository)
+	nonce, err := dispatch.AuthorizeWarmupDispatch(t.Context(), f.task, f.mailbox, f.worker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := dispatch.DeferWarmupDispatch(t.Context(), f.task, f.mailbox, f.worker, nonce, time.Now().Add(time.Hour)); err != nil {
+		t.Fatal(err)
+	}
+	var noNonces, charged, refunded bool
+	if err := f.pool.QueryRow(t.Context(), `SELECT t.send_executor_nonce IS NULL AND w.dispatch_nonce IS NULL,
+		w.warmup_charged_date IS NOT NULL,w.warmup_refunded_at IS NOT NULL FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, f.task).Scan(&noNonces, &charged, &refunded); err != nil || !noNonces || !charged || !refunded {
+		t.Fatalf("native refunded-charge control invalid: noNonces=%v charged=%v refunded=%v err=%v", noNonces, charged, refunded, err)
+	}
+	before := f.localRows(t)
+	booked, err := f.repo.GetBookedSendsForAccount(t.Context(), f.mailbox, time.Now())
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+	if after := f.localRows(t); after != before {
+		t.Error("import discarded the native refunded charge or rescheduled task while both nonces were absent")
+	}
+	bookedAfter, err := f.repo.GetBookedSendsForAccount(t.Context(), f.mailbox, time.Now())
+	if err != nil || len(booked) != len(bookedAfter) {
+		t.Errorf("import changed booked sends: %d -> %d %v", len(booked), len(bookedAfter), err)
+	}
+}
+
+func TestLiveImportLegacyWarmupAppliedAccountingWithoutNonces(t *testing.T) {
+	for _, reply := range []bool{false, true} {
+		for _, success := range []bool{false, true} {
+			t.Run(map[bool]string{false: "opening", true: "reply"}[reply]+"/"+map[bool]string{false: "failed", true: "sent"}[success], func(t *testing.T) {
+				f, _ := nativeWarmupArchiveFixture(t, reply)
+				groups := []models.OrgDataGroup{models.OrgDataGroupSending, models.OrgDataGroupWarmup}
+				archive := f.archive(t, groups)
+				warmup := repository.NewWarmupRepository(f.pool)
+				if err := f.repo.UpdateTaskStatus(t.Context(), f.task, "completed"); err != nil {
+					t.Fatal(err)
+				}
+				if err := warmup.IncrementDailyCount(t.Context(), f.mailbox, time.Now()); err != nil {
+					t.Fatal(err)
+				}
+				if reply {
+					if err := warmup.IncrementReplyCount(t.Context(), f.mailbox, time.Now()); err != nil {
+						t.Fatal(err)
+					}
+				}
+				result := models.SendEmailResult{TaskID: f.task, Success: success, MessageID: "<legacy-warmup@fixture.invalid>", SentAt: time.Now()}
+				if !success {
+					result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient", ObservedAt: time.Now()}}
+				}
+				calls, effects := 0, 0
+				apply := func(inner context.Context) error {
+					calls++
+					repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+					if !success {
+						return warmup.FailWarmupSend(inner, f.mailbox, f.task, time.Now(), "Legacy native refusal", "Synthetic local refusal")
+					}
+					return warmup.RecordWarmupTokenDelivery(inner, f.task, result.MessageID)
+				}
+				recovery := f.repo.(repository.SendResultRecovery)
+				if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil {
+					t.Fatal(err)
+				}
+				var legacy bool
+				if err := f.pool.QueryRow(t.Context(), `SELECT t.send_executor_nonce IS NULL AND w.dispatch_nonce IS NULL AND w.warmup_charged_date IS NULL AND t.send_result_applied_at IS NOT NULL FROM tasks t JOIN warmup_tasks w ON w.task_id=t.id WHERE t.id=$1`, f.task).Scan(&legacy); err != nil || !legacy {
+					t.Fatalf("legacy native warmup control invalid: %v %v", legacy, err)
+				}
+				checkCount := func() {
+					t.Helper()
+					var count, replies int
+					if err := f.pool.QueryRow(t.Context(), `SELECT emails_sent,emails_replied FROM warmup_statistics WHERE email_account_id=$1 AND date=(NOW() AT TIME ZONE 'UTC')::date`, f.mailbox).Scan(&count, &replies); err != nil || count != 1 || replies != map[bool]int{false: 0, true: 1}[reply] {
+						t.Errorf("import/result rewound or repeated legacy accounted send: count=%d replies=%d err=%v", count, replies, err)
+					}
+				}
+				checkCount()
+				before := f.localRows(t)
+				f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+				if after := f.localRows(t); after != before {
+					t.Error("import rewound the nonce-less applied warmup outcome, token or failed-send record")
+				}
+				checkCount()
+				if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil || calls != 1 || effects != 1 {
+					t.Errorf("legacy accounting or postcommit effects repeated: calls=%d effects=%d err=%v", calls, effects, err)
+				}
+				checkCount()
+				assertLegacyArchiveUnknowns(t, f, 0)
+				assertLegacyArchiveNoHold(t, f)
+			})
+		}
+	}
+}
+func TestLiveImportLegacyWarmupAppliedMarkers(t *testing.T) {
+	for _, kind := range []string{"sent", "refunded", "reply_refunded"} {
+		for _, mode := range []models.OrgImportConflict{models.OrgImportConflictSkip, models.OrgImportConflictOverwrite} {
+			t.Run(kind+"/"+string(mode), func(t *testing.T) {
+				f, _ := nativeWarmupArchiveFixture(t, kind == "reply_refunded")
+				ctx := t.Context()
+				if err := f.repo.UpdateTaskStatus(ctx, f.task, "completed"); err != nil {
+					t.Fatal("native stamp failed")
+				}
+				// Historical accepted-data fixture, not a native nonce-free reservation creation.
+				f.exec(t, `UPDATE warmup_tasks SET warmup_charged_date=(NOW() AT TIME ZONE 'UTC')::date,warmup_reply_charged=$2 WHERE task_id=$1`, f.task, kind == "reply_refunded")
+				f.exec(t, `UPDATE warmup_statistics SET emails_sent=1,emails_replied=$2 WHERE email_account_id=$1 AND date=(NOW() AT TIME ZONE 'UTC')::date`, f.mailbox, map[bool]int{false: 0, true: 1}[kind == "reply_refunded"])
+				groups := []models.OrgDataGroup{models.OrgDataGroupSending, models.OrgDataGroupWarmup}
+				archive := f.archive(t, groups)
+				warmup := repository.NewWarmupRepository(f.pool)
+				recovery := f.repo.(repository.SendResultRecovery)
+				result := models.SendEmailResult{TaskID: f.task, Success: kind == "sent", MessageID: "<legacy-result@fixture.invalid>", SentAt: time.Now()}
+				if !result.Success {
+					result.Error = &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendPermanent, Scope: "recipient", ObservedAt: time.Now()}}
+				}
+				calls, effects := 0, 0
+				apply := func(inner context.Context) error {
+					calls++
+					if result.Success {
+						if err := warmup.RecordWarmupTokenDelivery(inner, f.task, result.MessageID); err != nil {
+							return err
+						}
+						if err := f.repo.UpdateTaskMessageID(inner, f.task, result.MessageID); err != nil {
+							return err
+						}
+					} else if err := warmup.FailWarmupSend(inner, f.mailbox, f.task, time.Now(), "Synthetic refusal", "Synthetic fixture"); err != nil {
+						return err
+					}
+					repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+					return nil
+				}
+				if err := recovery.ApplySendResult(ctx, result, apply); err != nil {
+					t.Fatal("native application failed")
+				}
+				assertLegacyArchiveTask(t, f, f.task)
+				before := f.localRows(t)
+				var sent, replied int
+				if err := f.pool.QueryRow(ctx, `SELECT emails_sent,emails_replied FROM warmup_statistics WHERE email_account_id=$1 AND date=(NOW() AT TIME ZONE 'UTC')::date`, f.mailbox).Scan(&sent, &replied); err != nil {
+					t.Fatal("counter read failed")
+				}
+				if sent != map[bool]int{false: 0, true: 1}[result.Success] || replied != 0 {
+					t.Fatal("native accounting precondition failed")
+				}
+				f.importArchive(t, archive, groups, mode)
+				if f.localRows(t) != before {
+					t.Error("import rewound legacy warmup application markers or payload")
+				}
+				for range 2 {
+					if err := recovery.ApplySendResult(ctx, result, apply); err != nil {
+						t.Fatal("native redelivery failed")
+					}
+				}
+				f.importArchive(t, archive, groups, mode)
+				if err := recovery.ApplySendResult(ctx, result, apply); err != nil {
+					t.Fatal("native post-import redelivery failed")
+				}
+				if calls != 1 || effects != 1 {
+					t.Error("import repeated legacy warmup application or postcommit effects")
+				}
+				var chargeRetained bool
+				if err := f.pool.QueryRow(ctx, `SELECT warmup_charged_date IS NOT NULL AND warmup_reply_charged=$2 AND ($3::bool OR warmup_refunded_at IS NOT NULL) FROM warmup_tasks WHERE task_id=$1`, f.task, kind == "reply_refunded", result.Success).Scan(&chargeRetained); err != nil || !chargeRetained {
+					t.Error("import changed historical warmup charge/refund proof")
+				}
+				var afterSent, afterReply int
+				if err := f.pool.QueryRow(ctx, `SELECT emails_sent,emails_replied FROM warmup_statistics WHERE email_account_id=$1 AND date=(NOW() AT TIME ZONE 'UTC')::date`, f.mailbox).Scan(&afterSent, &afterReply); err != nil || afterSent != sent || afterReply != replied {
+					t.Error("import changed already settled warmup counters")
+				}
+				assertLegacyArchiveUnknowns(t, f, 0)
+				var held bool
+				if err := f.pool.QueryRow(ctx, "SELECT send_recovery_hold FROM email_accounts WHERE id=$1", f.mailbox).Scan(&held); err != nil || held {
+					t.Fatalf("settled native legacy warmup gained unknown hold: %v %v", held, err)
+				}
+			})
+		}
+	}
+}
+
+func TestLiveImportLegacyUnknownLockOrder(t *testing.T) {
+	for _, order := range []string{"import_first", "result_first"} {
+		t.Run(order, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
+			defer cancel()
+			f := newSelectiveHoldReviewFixture(t)
+			groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+			archive := f.archive(t, groups)
+			cfg, err := pgxpool.ParseConfig(os.Getenv("WARMBLY_TEST_DB"))
+			if err != nil {
+				t.Fatal("fixture config unavailable")
+			}
+			cfg.MaxConns, cfg.MinConns = 1, 0
+			rpool, err := pgxpool.NewWithConfig(ctx, cfg)
+			if err != nil {
+				t.Fatal("result pool unavailable")
+			}
+			defer rpool.Close()
+			conn, err := rpool.Acquire(ctx)
+			if err != nil {
+				t.Fatal("result connection unavailable")
+			}
+			rpid := conn.Conn().PgConn().PID()
+			conn.Release()
+			recovery := repository.NewTaskRepository(rpool).(repository.SendResultRecovery)
+			importPID := make(chan uint32, 1)
+			svc := f.svc.(*service)
+			svc.repo = &observedWholeImportRepository{OrgTransferRepository: svc.repo, pid: importPID}
+			importDone, resultDone := make(chan error, 1), make(chan error, 1)
+			atTasks, resume := make(chan struct{}, 1), make(chan struct{})
+			startImport := func() {
+				go func() {
+					_, err := f.svc.ImportFrom(ctx, f.tenant.org, bytes.NewReader(archive), ImportOptions{Groups: groups, Passphrase: "synthetic-review-passphrase", Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.tenant.owner}, func(_ int, stage string) {
+						if stage == "importing tasks" && order == "import_first" {
+							atTasks <- struct{}{}
+							select {
+							case <-resume:
+							case <-ctx.Done():
+							}
+						}
+					})
+					importDone <- err
+				}()
+			}
+			calls, effects := 0, 0
+			startResult := func() {
+				go func() {
+					resultDone <- recovery.ApplySendResult(ctx, models.SendEmailResult{TaskID: f.task}, func(inner context.Context) error {
+						calls++
+						repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+						return nil
+					})
+				}()
+			}
+			waitBlocked := func(waiter, holder uint32) {
+				t.Helper()
+				ticker := time.NewTicker(10 * time.Millisecond)
+				defer ticker.Stop()
+				for {
+					var yes bool
+					if err := f.pool.QueryRow(ctx, `SELECT $2::int=ANY(pg_blocking_pids($1))`, waiter, holder).Scan(&yes); err != nil {
+						t.Fatal("lock inspection failed")
+					}
+					if yes {
+						return
+					}
+					select {
+					case <-ticker.C:
+					case <-ctx.Done():
+						t.Fatal("native blocker not reached")
+					}
+				}
+			}
+			if order == "result_first" {
+				gate, err := f.pool.Begin(ctx)
+				if err != nil {
+					t.Fatal("gate unavailable")
+				}
+				defer gate.Rollback(context.Background())
+				if _, err := gate.Exec(ctx, `SELECT id FROM email_accounts WHERE id=$1 FOR UPDATE`, f.mailbox); err != nil {
+					t.Fatal("gate failed")
+				}
+				startResult()
+				waitBlocked(rpid, gate.Conn().PgConn().PID())
+				startImport()
+				waitBlocked(<-importPID, rpid)
+				if err := gate.Rollback(ctx); err != nil {
+					t.Fatal("gate release failed")
+				}
+			} else {
+				startImport()
+				select {
+				case <-atTasks:
+				case <-ctx.Done():
+					t.Fatal("import barrier unavailable")
+				}
+				startResult()
+				waitBlocked(rpid, <-importPID)
+				close(resume)
+			}
+			if err := <-resultDone; err != nil {
+				t.Fatal("native unknown result failed")
+			}
+			if err := <-importDone; err != nil {
+				t.Fatal("native concurrent import failed")
+			}
+			assertLegacyArchiveTask(t, f, f.task)
+			var retained bool
+			if err := f.pool.QueryRow(ctx, `SELECT t.send_result_state='unknown' AND ea.send_recovery_hold AND ea.send_recovery_task_id=t.id FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, f.task).Scan(&retained); err != nil || !retained {
+				t.Error("serialized import rewound a native legacy unknown result or hold")
+			}
+			if calls != 0 || effects != 0 {
+				t.Fatalf("unknown outcome invoked effects: %d %d", calls, effects)
+			}
+			assertLegacyArchiveUnknowns(t, f, 1)
+		})
+	}
+}
+
+func assertLegacyArchiveTask(t *testing.T, f *selectiveHoldReviewFixture, task uuid.UUID) {
+	t.Helper()
+	var absent bool
+	err := f.pool.QueryRow(t.Context(), `SELECT send_executor_nonce IS NULL
+		AND NOT EXISTS(SELECT 1 FROM warmup_tasks WHERE task_id=t.id AND dispatch_nonce IS NOT NULL)
+		AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions WHERE recovery_task_id=t.id AND recorded_locally_at IS NOT NULL)
+		FROM tasks t WHERE id=$1`, task).Scan(&absent)
+	if err != nil || !absent {
+		t.Fatal("fixture did not produce accepted nonce-free local state")
+	}
+}
+
+func TestLiveImportLegacyRestrictivePreservation(t *testing.T) {
+	for _, mode := range []models.OrgImportConflict{models.OrgImportConflictSkip, models.OrgImportConflictOverwrite} {
+		t.Run(string(mode), func(t *testing.T) {
+			f := newSelectiveHoldReviewFixture(t)
+			groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+			archive := f.archive(t, groups)
+			if err := f.repo.UpdateTaskStatus(t.Context(), f.task, "completed"); err != nil {
+				t.Fatal("native task stamp failed")
+			}
+			result := models.SendEmailResult{TaskID: f.task, Error: &models.EmailSendError{Failure: &errx.SendFailure{Disposition: errx.SendAuth, Scope: "mailbox", ObservedAt: time.Now()}}}
+			calls, effects := 0, 0
+			apply := func(inner context.Context) error {
+				calls++
+				repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+				return nil
+			}
+			recovery := f.repo.(repository.SendResultRecovery)
+			if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil {
+				t.Fatal("native restrictive callback failed")
+			}
+			assertLegacyArchiveTask(t, f, f.task)
+			before := f.localRows(t)
+			f.importArchive(t, archive, groups, mode)
+			if f.localRows(t) != before {
+				t.Error("import rewound native legacy authentication markers")
+			}
+			var held bool
+			if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_hold AND send_recovery_reason='authentication' FROM email_accounts WHERE id=$1`, f.mailbox).Scan(&held); err != nil || !held {
+				t.Fatal("restrictive mailbox hold lost")
+			}
+			for range 2 {
+				if err := recovery.ApplySendResult(t.Context(), result, apply); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if calls != 1 || effects != 1 {
+				t.Fatalf("authentication callback/effects repeated: %d %d", calls, effects)
+			}
+			assertLegacyArchiveUnknowns(t, f, 0)
+		})
+	}
+}
+
+func TestLiveImportExistingReceiptExcluded(t *testing.T) {
+	f := newSelectiveHoldReviewFixture(t)
+	warmup := repository.NewWarmupRepository(f.pool)
+	receipt := uuid.New()
+	if err := warmup.RecordWarmupReceived(t.Context(), f.mailbox, receipt, "<native-receipt@fixture.invalid>", f.mailbox, false); err != nil {
+		t.Fatal("native receipt creation failed")
+	}
+	groups := []models.OrgDataGroup{models.OrgDataGroupSending, models.OrgDataGroupWarmup}
+	archive := f.archive(t, groups)
+	zr, err := zip.NewReader(bytes.NewReader(archive), int64(len(archive)))
+	if err != nil {
+		t.Fatal("native archive unreadable")
+	}
+	for _, entry := range zr.File {
+		if entry.Name == dataPath("warmup_received") {
+			t.Fatal("native archive exported forbidden source receipt")
+		}
+	}
+	if ExcludedTables["warmup_received"] == "" {
+		t.Fatal("receipt exclusion policy absent")
+	}
+	if err := warmup.RetireWarmupReceived(t.Context(), f.mailbox, receipt); err != nil {
+		t.Fatal("native receipt update failed")
+	}
+	var before, after string
+	q := `SELECT to_jsonb(r)::text FROM warmup_received r WHERE email_account_id=$1 AND internal_id=$2`
+	if err := f.pool.QueryRow(t.Context(), q, f.mailbox, receipt).Scan(&before); err != nil {
+		t.Fatal("local receipt unavailable")
+	}
+	f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+	if err := f.pool.QueryRow(t.Context(), q, f.mailbox, receipt).Scan(&after); err != nil || after != before {
+		t.Fatal("excluded local receipt changed")
+	}
+}
+
+func assertLegacyArchiveUnknowns(t *testing.T, f *selectiveHoldReviewFixture, want int) {
+	t.Helper()
+	var got int
+	if err := f.pool.QueryRow(t.Context(), "SELECT COUNT(*) FROM tasks WHERE email_account_id=$1 AND send_result_state='unknown'", f.mailbox).Scan(&got); err != nil || got != want {
+		t.Fatalf("legacy unknown candidates changed: got=%d want=%d err=%v", got, want, err)
+	}
+}
+
+func assertLegacyArchiveNoHold(t *testing.T, f *selectiveHoldReviewFixture) {
+	t.Helper()
+	var held bool
+	if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_hold FROM email_accounts WHERE id=$1`, f.mailbox).Scan(&held); err != nil || held {
+		t.Fatalf("settled legacy result gained a restrictive hold: %v %v", held, err)
+	}
+}
+
+func TestLiveImportFreshLegacyWarmupHistoryRetainsPayload(t *testing.T) {
+	for _, existing := range []bool{false, true} {
+		for _, sourceHold := range []bool{false, true} {
+			t.Run(fmt.Sprintf("existing_%v/source_hold_%v", existing, sourceHold), func(t *testing.T) {
+				f, _ := nativeWarmupArchiveFixture(t, false)
+				warmup := repository.NewWarmupRepository(f.pool)
+				groups := []models.OrgDataGroup{models.OrgDataGroupSending, models.OrgDataGroupWarmup}
+				other := uuid.New()
+				if sourceHold {
+					f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','active','')`, other, f.mailbox)
+					f.exec(t, `INSERT INTO email_tasks(task_id,to_addrs,subject,body,body_plain,encrypted) VALUES($1,ARRAY['recipient@fixture.invalid'],'Portable fixture','Synthetic body','Synthetic body',false)`, other)
+				}
+				older := f.archive(t, groups)
+				if err := f.repo.UpdateTaskStatus(t.Context(), f.task, "completed"); err != nil {
+					t.Fatal(err)
+				}
+				if err := warmup.IncrementDailyCount(t.Context(), f.mailbox, time.Now()); err != nil {
+					t.Fatal(err)
+				}
+				calls, effects := 0, 0
+				result := models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "<portable-legacy@fixture.invalid>", SentAt: time.Now()}
+				if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), result, func(inner context.Context) error {
+					calls++
+					repository.AfterSendResultCommit(inner, func(context.Context) { effects++ })
+					return warmup.RecordWarmupTokenDelivery(inner, f.task, result.MessageID)
+				}); err != nil {
+					t.Fatal(err)
+				}
+				if sourceHold {
+					if err := f.repo.UpdateTaskStatus(t.Context(), other, "completed"); err != nil {
+						t.Fatal(err)
+					}
+					if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), models.SendEmailResult{TaskID: other}, func(context.Context) error {
+						t.Error("unknown source outcome invoked effects")
+						return nil
+					}); err != nil {
+						t.Fatal(err)
+					}
+				}
+				archive := f.archive(t, groups)
+				f.exec(t, `DELETE FROM email_accounts WHERE id=$1`, f.mailbox)
+				if existing {
+					f.importArchive(t, older, groups, models.OrgImportConflictOverwrite)
+					var unprotected bool
+					if err := f.pool.QueryRow(t.Context(), `SELECT t.send_executor_nonce IS NULL AND t.send_result_state IS NULL AND t.send_result_applied_at IS NULL
+						AND t.completed_at IS NULL AND t.send_reserved_at IS NULL AND t.send_executor_started_at IS NULL
+						AND t.send_executor_result IS NULL AND t.send_result_evidence IS NULL AND t.send_released_at IS NULL
+						AND NOT ea.send_recovery_hold AND NOT EXISTS(SELECT 1 FROM outbound_attempts WHERE task_id=t.id)
+						AND NOT EXISTS(SELECT 1 FROM send_recovery_resolutions WHERE recovery_task_id=t.id AND recorded_locally_at IS NOT NULL)
+						AND EXISTS(SELECT 1 FROM warmup_tasks w WHERE w.task_id=t.id AND w.dispatch_nonce IS NULL
+						AND w.dispatch_started_at IS NULL AND w.dispatch_result IS NULL AND w.warmup_charged_date IS NULL
+						AND w.warmup_refunded_at IS NULL) FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id
+						WHERE t.id=$1`, f.task).Scan(&unprotected); err != nil || !unprotected {
+						t.Fatalf("older native import unexpectedly protected existing task: %v %v", unprotected, err)
+					}
+				}
+				f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+				var history, payload, token, authority bool
+				if err := f.pool.QueryRow(t.Context(), `SELECT t.status='completed' AND t.send_result_state='sent',
+					EXISTS(SELECT 1 FROM warmup_tasks w WHERE w.task_id=t.id AND w.subject IS NOT NULL),
+					EXISTS(SELECT 1 FROM warmup_tokens k WHERE k.task_id=t.id AND k.sent_message_id=$2),
+					t.send_executor_nonce IS NOT NULL OR EXISTS(SELECT 1 FROM warmup_tasks w WHERE w.task_id=t.id AND
+					(w.dispatch_nonce IS NOT NULL OR w.warmup_charged_date IS NOT NULL OR w.lineage_version<>0))
+					FROM tasks t WHERE t.id=$1`, f.task, result.MessageID).Scan(&history, &payload, &token, &authority); err != nil || !history || !payload || !token || authority {
+					t.Errorf("native historical overwrite lost payload/token or gained authority: history=%v payload=%v token=%v authority=%v err=%v", history, payload, token, authority, err)
+				}
+				var count int
+				if err := f.pool.QueryRow(t.Context(), `SELECT emails_sent FROM warmup_statistics WHERE email_account_id=$1 AND date=(NOW() AT TIME ZONE 'UTC')::date`, f.mailbox).Scan(&count); err != nil || count != 1 {
+					t.Errorf("historical count not imported: %d %v", count, err)
+				}
+				if sourceHold {
+					var historicalUnknown, held, noAuthority bool
+					if err := f.pool.QueryRow(t.Context(), `SELECT COALESCE(t.send_result_state='unknown',false),ea.send_recovery_hold AND ea.send_recovery_task_id=t.id,
+						t.send_executor_nonce IS NULL AND t.send_executor_started_at IS NULL
+						FROM tasks t JOIN email_accounts ea ON ea.id=t.email_account_id WHERE t.id=$1`, other).Scan(&historicalUnknown, &held, &noAuthority); err != nil || !historicalUnknown || !held || !noAuthority {
+						t.Errorf("source hold suppressed its own task history or gained authority: unknown=%v held=%v noAuthority=%v err=%v", historicalUnknown, held, noAuthority, err)
+					}
+					assertLegacyArchiveUnknowns(t, f, 1)
+				} else {
+					assertLegacyArchiveUnknowns(t, f, 0)
+					assertLegacyArchiveNoHold(t, f)
+				}
+				if calls != 1 || effects != 1 {
+					t.Fatalf("import invoked application/postcommit effects: %d %d", calls, effects)
+				}
+				results, err := f.repo.(interface {
+					ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error)
+				}).ListUnappliedSendResults(t.Context(), 200)
+				if err != nil {
+					t.Fatal(err)
+				}
+				for _, pending := range results {
+					if pending.TaskID == f.task || pending.TaskID == other {
+						t.Error("imported historical outcome became pending reconciliation")
+					}
+				}
+			})
+		}
+	}
+}
@@ -0,0 +1,197 @@
+package orgtransfer
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/repository"
+)
+
+func TestLiveImportLocalResultPreservationLockOrder(t *testing.T) {
+	for _, order := range []string{"import_first", "import_first_external_row", "result_first"} {
+		t.Run(order, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(t.Context(), 25*time.Second)
+			defer cancel()
+			f := newSelectiveHoldReviewFixture(t)
+			groups := []models.OrgDataGroup{models.OrgDataGroupSending}
+			archive := f.archive(t, groups)
+			f.reserve(t)
+			f.begin(t)
+			f.finish(t)
+			result := f.result()
+			var generation string
+			if err := f.pool.QueryRow(ctx, `SELECT transport_generation::text FROM email_accounts WHERE id=$1`, f.mailbox).Scan(&generation); err != nil {
+				t.Fatal(err)
+			}
+			cfg, err := pgxpool.ParseConfig(os.Getenv("WARMBLY_TEST_DB"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			cfg.MaxConns, cfg.MinConns = 1, 0
+			resultPool, err := pgxpool.NewWithConfig(ctx, cfg)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer resultPool.Close()
+			conn, err := resultPool.Acquire(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rpid := conn.Conn().PgConn().PID()
+			conn.Release()
+			rrepo := repository.NewTaskRepository(resultPool)
+			recovery := rrepo.(repository.SendResultRecovery)
+			reports := repository.NewEmailAccountErrorRepository(&db.DB{Pool: resultPool})
+			importPID := make(chan uint32, 1)
+			service := f.svc.(*service)
+			service.repo = &observedWholeImportRepository{OrgTransferRepository: service.repo, pid: importPID}
+			atTasks, atResult := make(chan struct{}, 1), make(chan struct{}, 1)
+			resumeImport, resumeResult := make(chan struct{}), make(chan struct{})
+			importDone, resultDone := make(chan error, 1), make(chan error, 1)
+			calls := 0
+			apply := func(inner context.Context) error {
+				calls++
+				if err := rrepo.UpdateTaskMessageID(inner, f.task, result.MessageID); err != nil {
+					return err
+				}
+				if err := rrepo.UpdateTaskStatus(inner, f.task, "completed"); err != nil {
+					return err
+				}
+				gen, err := uuid.Parse(generation)
+				if err != nil {
+					return err
+				}
+				if xerr := reports.RecordTransportSuccess(inner, f.mailbox, models.MailTransportSend, time.Now().Add(-time.Second), &f.task, gen); xerr != nil {
+					return xerr
+				}
+				if order == "result_first" && calls == 1 {
+					atResult <- struct{}{}
+					select {
+					case <-resumeResult:
+					case <-ctx.Done():
+						return ctx.Err()
+					}
+				}
+				return nil
+			}
+			startImport := func() {
+				go func() {
+					_, err := f.svc.ImportFrom(ctx, f.tenant.org, bytes.NewReader(archive), ImportOptions{Groups: groups, Passphrase: "synthetic-review-passphrase", Conflict: models.OrgImportConflictOverwrite, ActorUserID: f.tenant.owner}, func(_ int, stage string) {
+						if stage == "importing tasks" && order != "result_first" {
+							atTasks <- struct{}{}
+							select {
+							case <-resumeImport:
+							case <-ctx.Done():
+							}
+						}
+					})
+					importDone <- err
+				}()
+			}
+			startResult := func() { go func() { resultDone <- recovery.ApplySendResult(ctx, result, apply) }() }
+			waitBlocked := func(waiter, holder uint32) {
+				t.Helper()
+				ticker := time.NewTicker(10 * time.Millisecond)
+				defer ticker.Stop()
+				for {
+					var yes bool
+					if err := f.pool.QueryRow(ctx, `SELECT $2::int=ANY(pg_blocking_pids($1))`, waiter, holder).Scan(&yes); err != nil {
+						t.Fatal(err)
+					}
+					if yes {
+						return
+					}
+					select {
+					case <-ticker.C:
+					case <-ctx.Done():
+						t.Fatal("expected lock blocker not reached", ctx.Err())
+					}
+				}
+			}
+			if order == "result_first" {
+				startResult()
+				select {
+				case <-atResult:
+				case err := <-resultDone:
+					t.Fatalf("result did not reach callback: %v", err)
+				case <-ctx.Done():
+					t.Fatal(ctx.Err())
+				}
+				startImport()
+				ip := <-importPID
+				waitBlocked(ip, rpid)
+				t.Log("actual result owns268/269/285 and task/mailbox rows; import waits before its row writes")
+				close(resumeResult)
+			} else {
+				startImport()
+				select {
+				case <-atTasks:
+				case err := <-importDone:
+					t.Fatalf("import did not reach tasks: %v", err)
+				case <-ctx.Done():
+					t.Fatal(ctx.Err())
+				}
+				ip := <-importPID
+				var gate pgx.Tx
+				if order == "import_first_external_row" {
+					gate, err = f.pool.Begin(ctx)
+					if err != nil {
+						t.Fatal(err)
+					}
+					defer gate.Rollback(context.Background())
+					if _, err := gate.Exec(ctx, `SELECT id FROM tasks WHERE id=$1 FOR UPDATE`, f.task); err != nil {
+						t.Fatal(err)
+					}
+				}
+				startResult()
+				waitBlocked(rpid, ip)
+				var rowBeforeMailbox bool
+				if err := f.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM pg_locks WHERE pid=$1 AND (locktype='tuple' OR relation='tasks'::regclass AND mode IN('RowShareLock','RowExclusiveLock')))`, rpid).Scan(&rowBeforeMailbox); err != nil {
+					t.Fatal(err)
+				}
+				if rowBeforeMailbox {
+					t.Fatal("real ApplySendResult took a task row while waiting for mailbox269")
+				}
+				t.Log("actual import owns269/285; result owns268 but no task row and waits269")
+				close(resumeImport)
+				if gate != nil {
+					waitBlocked(ip, gate.Conn().PgConn().PID())
+					if err := gate.Rollback(ctx); err != nil {
+						t.Fatal(err)
+					}
+				}
+			}
+			for _, done := range []<-chan error{importDone, resultDone} {
+				select {
+				case err := <-done:
+					if err != nil {
+						t.Fatal(err)
+					}
+				case <-ctx.Done():
+					t.Fatal("import/result did not settle", ctx.Err())
+				}
+			}
+			var preserved bool
+			if err := f.pool.QueryRow(ctx, `SELECT COALESCE(send_result_state='sent' AND send_result_applied_at IS NOT NULL AND status='completed' AND message_id=$2,false) FROM tasks WHERE id=$1`, f.task, result.MessageID).Scan(&preserved); err != nil {
+				t.Fatal(err)
+			}
+			if !preserved {
+				t.Error("older archive rewound real committed result state or result application message")
+			}
+			if err := recovery.ApplySendResult(ctx, result, apply); err != nil {
+				t.Fatal(err)
+			}
+			if calls != 1 {
+				t.Errorf("result application repeated after archive overwrite: %d", calls)
+			}
+		})
+	}
+}
@@ -0,0 +1,291 @@
+package orgtransfer
+
+import (
+	"archive/zip"
+	"bytes"
+	"context"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/warmbly/warmbly/internal/infrastructure/db"
+	"github.com/warmbly/warmbly/internal/models"
+	"github.com/warmbly/warmbly/internal/pkg/encrypt"
+	"github.com/warmbly/warmbly/internal/repository"
+	"github.com/warmbly/warmbly/internal/tasks"
+)
+
+type selectiveHoldReviewFixture struct {
+	pool                         *pgxpool.Pool
+	svc                          Service
+	tenant                       tenantFixture
+	mailbox, task, worker, nonce uuid.UUID
+	repo                         repository.TaskRepository
+	admission                    repository.OutboundAdmissionRepository
+	validator                    interface {
+		ValidateOutboundExecution(context.Context, uuid.UUID) error
+	}
+}
+
+func newSelectiveHoldReviewFixture(t *testing.T) *selectiveHoldReviewFixture {
+	t.Helper()
+	_, pool := liveImportService(t)
+	f := &selectiveHoldReviewFixture{pool: pool, tenant: newTenantFixture(t, pool), mailbox: uuid.New(), task: uuid.New(), worker: uuid.New()}
+	f.exec(t, `INSERT INTO fleet_nodes(id,role,active,last_seen_at,warmup_send_protocol) VALUES($1,'worker',true,NOW(),2)`, f.worker)
+	f.exec(t, `INSERT INTO workers(id) VALUES($1)`, f.worker)
+	f.exec(t, `INSERT INTO email_accounts(id,user_id,organization_id,email,name,signature_plain,signature_html,provider,status,worker_id,timezone,campaign_limit,warmup_max,shared_daily_limit,rolling_recipient_limit)
+	 VALUES($1,$2,$3,$4,'Local archive fixture','','','smtp_imap','active',$5,'UTC',50,50,100,100)`, f.mailbox, f.tenant.owner, f.tenant.org, f.mailbox.String()+"@example.test", f.worker)
+	f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','active','')`, f.task, f.mailbox)
+	f.exec(t, `INSERT INTO email_tasks(task_id,to_addrs,subject,body,body_plain,encrypted) VALUES($1,ARRAY['recipient@fixture.invalid'],'Local fixture','Synthetic local body','Synthetic local body',false)`, f.task)
+	enc, err := encrypt.NewEncrypter([]byte("0123456789abcdef0123456789abcdef"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	password, err := enc.Encrypt("unused-local-fixture-password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.exec(t, `INSERT INTO email_accounts_smtp_imap(email_account_id,smtp_host,smtp_port,smtp_user,smtp_password,imap_host,imap_port,imap_user,imap_password)
+	 VALUES($1,'smtp.fixture.invalid',587,'synthetic',$2,'imap.fixture.invalid',993,'synthetic',$2)`, f.mailbox, password)
+	f.svc = NewService(repository.NewOrgTransferRepository(&db.DB{Pool: pool}), nil, enc, nil, InstanceInfo{})
+	f.repo = repository.NewTaskRepository(pool)
+	f.admission = f.repo.(repository.OutboundAdmissionRepository)
+	emailRepo := repository.NewEmailRepostory(&db.DB{Pool: pool}, enc)
+	f.validator = tasks.NewService(nil, nil, nil, nil, nil, nil, nil, nil, nil, f.repo, nil, nil, nil, nil, emailRepo, nil, nil, nil, nil, nil, nil, nil).(interface {
+		ValidateOutboundExecution(context.Context, uuid.UUID) error
+	})
+	t.Cleanup(func() {
+		if _, err := pool.Exec(context.Background(), `DELETE FROM email_accounts WHERE organization_id=$1`, f.tenant.org); err != nil {
+			t.Error(err)
+		}
+		if _, err := pool.Exec(context.Background(), `DELETE FROM fleet_nodes WHERE id=$1`, f.worker); err != nil {
+			t.Error(err)
+		}
+	})
+	return f
+}
+
+func (f *selectiveHoldReviewFixture) exec(t *testing.T, q string, args ...any) {
+	t.Helper()
+	if _, err := f.pool.Exec(t.Context(), q, args...); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *selectiveHoldReviewFixture) archive(t *testing.T, groups []models.OrgDataGroup) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	manifest, err := f.svc.ExportTo(t.Context(), f.tenant.org, ExportOptions{Groups: groups, Passphrase: "synthetic-review-passphrase"}, &buf, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(manifest.Groups) == 0 || manifest.Groups[0] != models.OrgDataGroupCore {
+		t.Fatal("actual export did not force core")
+	}
+	zr, err := zip.NewReader(bytes.NewReader(buf.Bytes()), int64(buf.Len()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	core := false
+	for _, entry := range zr.File {
+		if entry.Name == dataPath("email_accounts") {
+			core = true
+		}
+	}
+	if !core {
+		t.Fatal("native selective archive omitted mailbox core entry")
+	}
+	return buf.Bytes()
+}
+
+func (f *selectiveHoldReviewFixture) importArchive(t *testing.T, raw []byte, groups []models.OrgDataGroup, conflict models.OrgImportConflict) {
+	t.Helper()
+	result, err := f.svc.ImportFrom(t.Context(), f.tenant.org, bytes.NewReader(raw), ImportOptions{Groups: groups, Passphrase: "synthetic-review-passphrase", Conflict: conflict, ActorUserID: f.tenant.owner}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !result.SecretsApplied {
+		t.Fatal("native encrypted credentials were not applied")
+	}
+	t.Logf("actual ExportTo/ImportFrom groups=%v conflict=%s applied=%v warnings=%v", groups, conflict, result.RowCounts, result.Warnings)
+}
+
+func (f *selectiveHoldReviewFixture) reserve(t *testing.T) {
+	t.Helper()
+	var err error
+	f.nonce, err = f.admission.ReserveOutbound(t.Context(), repository.OutboundReservation{TaskID: f.task, MailboxID: f.mailbox, OrganizationID: f.tenant.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{"recipient@fixture.invalid"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *selectiveHoldReviewFixture) begin(t *testing.T) {
+	t.Helper()
+	if err := f.validator.ValidateOutboundExecution(t.Context(), f.task); err != nil {
+		t.Fatalf("real generic execution validation refused fixture: %v", err)
+	}
+	state, err := f.admission.BeginOutbound(t.Context(), f.task, f.mailbox, f.worker, f.nonce)
+	if err != nil || state.State != "execute" {
+		t.Fatalf("native start=%+v error=%v", state, err)
+	}
+}
+
+func (f *selectiveHoldReviewFixture) result() models.SendEmailResult {
+	return models.SendEmailResult{TaskID: f.task, Success: true, MessageID: "<synthetic-result@fixture.invalid>"}
+}
+
+func (f *selectiveHoldReviewFixture) finish(t *testing.T) {
+	t.Helper()
+	if err := f.admission.FinishOutbound(t.Context(), f.task, f.mailbox, f.worker, f.result()); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func (f *selectiveHoldReviewFixture) snapshot(t *testing.T) string {
+	t.Helper()
+	var raw string
+	if err := f.pool.QueryRow(t.Context(), `SELECT jsonb_build_object('mailbox_held',ea.send_recovery_hold,'mailbox_reason',ea.send_recovery_reason,'held_task',ea.send_recovery_task_id,'mailbox_status',ea.status,'worker',ea.worker_id,'task_status',t.status,'task_state',t.send_result_state,'reserved',t.send_reserved_at,'released',t.send_released_at,'started',t.send_executor_started_at,'nonce',t.send_executor_nonce,'durable_result',t.send_executor_result IS NOT NULL,'applied',t.send_result_applied_at,'attempts',(SELECT count(*) FROM outbound_attempts WHERE task_id=t.id))::text FROM email_accounts ea LEFT JOIN tasks t ON t.id=$2 WHERE ea.id=$1`, f.mailbox, f.task).Scan(&raw); err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+func TestLiveReviewSelectiveExportMustPreserveLocalUnknownHold(t *testing.T) {
+	for _, group := range []models.OrgDataGroup{models.OrgDataGroupCore, models.OrgDataGroupSending} {
+		for _, conflict := range []models.OrgImportConflict{models.OrgImportConflictSkip, models.OrgImportConflictOverwrite} {
+			for _, phase := range []string{"reserved", "started", "durable_result"} {
+				t.Run(string(group)+"/"+string(conflict)+"/"+phase, func(t *testing.T) {
+					f := newSelectiveHoldReviewFixture(t)
+					groups := []models.OrgDataGroup{group}
+					archive := f.archive(t, groups)
+					f.reserve(t)
+					if phase != "reserved" {
+						f.begin(t)
+					}
+					if phase == "durable_result" {
+						f.finish(t)
+					}
+					before := f.snapshot(t)
+					beforeRows := f.localRows(t)
+					f.importArchive(t, archive, groups, conflict)
+					after := f.snapshot(t)
+					if beforeRows != f.localRows(t) {
+						t.Error("archive changed the complete local task, typed payload, reservation, durable outcome or evidence")
+					}
+					t.Logf("before=%s after=%s", before, after)
+					var held, hasNonce, started bool
+					var outcome *string
+					if err := f.pool.QueryRow(t.Context(), `SELECT ea.send_recovery_hold,t.send_executor_nonce=$3,t.send_executor_started_at IS NOT NULL,t.send_result_state::text FROM email_accounts ea JOIN tasks t ON t.id=$2 WHERE ea.id=$1`, f.mailbox, f.task, f.nonce).Scan(&held, &hasNonce, &started, &outcome); err != nil {
+						t.Fatal(err)
+					}
+					if !held {
+						t.Error("actual native archive overwrite removed current locally reserved/started unknown hold without evidence")
+					}
+					if !hasNonce || started != (phase != "reserved") {
+						t.Error("local executor authority unexpectedly lost")
+					}
+					if outcome == nil || *outcome != "unknown" {
+						t.Error("actual native archive overwrote locally unresolved task outcome")
+					}
+					admit, err := f.repo.(repository.SendResultRecovery).GetSendAdmission(t.Context(), f.tenant.org, f.mailbox, models.InboxProviderSMTPIMAP, time.Now())
+					if err != nil {
+						t.Fatal(err)
+					}
+					if admit.Allowed {
+						t.Error("mailbox send admission reopened while local unresolved executor authority remains")
+					}
+					if !held {
+						newTask := uuid.New()
+						f.exec(t, `INSERT INTO tasks(id,email_account_id,task_type,status,message_id) VALUES($1,$2,'email','active','')`, newTask, f.mailbox)
+						f.exec(t, `INSERT INTO email_tasks(task_id,to_addrs,subject,body,body_plain,encrypted) VALUES($1,ARRAY['another@fixture.invalid'],'Another local fixture','Synthetic body','Synthetic body',false)`, newTask)
+						_, err := f.admission.ReserveOutbound(t.Context(), repository.OutboundReservation{TaskID: newTask, MailboxID: f.mailbox, OrganizationID: f.tenant.org, WorkerID: f.worker, Provider: models.InboxProviderSMTPIMAP, Recipients: []string{"another@fixture.invalid"}})
+						t.Logf("new outbound reservation after import err=%v", err)
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestLiveReviewSelectiveExportLateLocalResultRemainsAccepted(t *testing.T) {
+	for _, group := range []models.OrgDataGroup{models.OrgDataGroupCore, models.OrgDataGroupSending} {
+		t.Run(string(group), func(t *testing.T) {
+			f := newSelectiveHoldReviewFixture(t)
+			groups := []models.OrgDataGroup{group}
+			archive := f.archive(t, groups)
+			f.reserve(t)
+			f.begin(t)
+			f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+			f.finish(t)
+			recovery := f.repo.(repository.SendResultRecovery)
+			calls := 0
+			for range 2 {
+				if err := recovery.ApplySendResult(t.Context(), f.result(), func(context.Context) error { calls++; return nil }); err != nil {
+					t.Fatal(err)
+				}
+			}
+			var outcome string
+			var applied bool
+			if err := f.pool.QueryRow(t.Context(), `SELECT send_result_state::text,send_result_applied_at IS NOT NULL FROM tasks WHERE id=$1`, f.task).Scan(&outcome, &applied); err != nil {
+				t.Fatal(err)
+			}
+			if outcome != "sent" || !applied || calls != 1 {
+				t.Fatalf("valid local started callback lost: outcome=%s applied=%v calls=%d", outcome, applied, calls)
+			}
+			t.Logf("synthetic typed native callback accepted exactly once: %s", f.snapshot(t))
+		})
+	}
+}
+
+func TestLiveReviewSelectiveExportNewImportIsHistoryNotAuthority(t *testing.T) {
+	for _, phase := range []string{"started_unknown", "applied_sent"} {
+		for _, group := range []models.OrgDataGroup{models.OrgDataGroupCore, models.OrgDataGroupSending} {
+			t.Run(phase+"/"+string(group), func(t *testing.T) {
+				f := newSelectiveHoldReviewFixture(t)
+				f.reserve(t)
+				f.begin(t)
+				if phase == "applied_sent" {
+					f.finish(t)
+					if err := f.repo.(repository.SendResultRecovery).ApplySendResult(t.Context(), f.result(), func(context.Context) error { return nil }); err != nil {
+						t.Fatal(err)
+					}
+				}
+				groups := []models.OrgDataGroup{group}
+				archive := f.archive(t, groups)
+				f.exec(t, `DELETE FROM email_accounts WHERE id=$1`, f.mailbox)
+				f.importArchive(t, archive, groups, models.OrgImportConflictOverwrite)
+				var held, exists, local bool
+				if err := f.pool.QueryRow(t.Context(), `SELECT ea.send_recovery_hold,EXISTS(SELECT 1 FROM tasks WHERE id=$2),EXISTS(SELECT 1 FROM tasks WHERE id=$2 AND (send_executor_nonce IS NOT NULL OR send_executor_started_at IS NOT NULL OR send_executor_result IS NOT NULL)) FROM email_accounts ea WHERE ea.id=$1`, f.mailbox, f.task).Scan(&held, &exists, &local); err != nil {
+					t.Fatal(err)
+				}
+				if held != (phase == "started_unknown") || exists != (group == models.OrgDataGroupSending) || local {
+					t.Fatalf("new import confused history with local proof: held=%v task_exists=%v executor=%v", held, exists, local)
+				}
+				var payload bool
+				if err := f.pool.QueryRow(t.Context(), `SELECT EXISTS(SELECT 1 FROM email_tasks WHERE task_id=$1)`, f.task).Scan(&payload); err != nil || payload != exists {
+					t.Fatalf("fresh historical task lost its portable typed payload: payload=%v task=%v err=%v", payload, exists, err)
+				}
+				results, err := f.repo.(interface {
+					ListUnappliedSendResults(context.Context, int) ([]models.SendEmailResult, error)
+				}).ListUnappliedSendResults(t.Context(), 200)
+				if err != nil {
+					t.Fatal(err)
+				}
+				for _, result := range results {
+					if result.TaskID == f.task {
+						t.Fatal("imported stored source result gained reconciliation authority")
+					}
+				}
+				if phase == "started_unknown" && group == models.OrgDataGroupCore {
+					var heldTask *uuid.UUID
+					if err := f.pool.QueryRow(t.Context(), `SELECT send_recovery_task_id FROM email_accounts WHERE id=$1`, f.mailbox).Scan(&heldTask); err != nil {
+						t.Fatal(err)
+					}
+					t.Logf("core-only unknown history retained hold but has no task row: held_task=%v", heldTask)
+				}
+				t.Logf("new imported snapshot=%s", f.snapshot(t))
+			})
+		}
+	}
+}
@@ -98,6 +98,10 @@ func TestLiveArchiveSendRecoveryLocalOriginResets(t *testing.T) {
 	if exported != 4 {
 		t.Fatalf("native export lost recovery audit: rows=%d", exported)
 	}
+	// Import source-only evidence as new history, not over an existing local confirmation.
+	if _, err := pool.Exec(ctx, `DELETE FROM send_recovery_resolutions WHERE organization_id=$1`, f.org); err != nil {
+		t.Fatal(err)
+	}
 	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)
 	}
@@ -146,7 +150,7 @@ func TestLiveArchiveSendRecoveryLocalOriginResets(t *testing.T) {
 			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
+			kept := true
 			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)
 			}
@@ -12,6 +12,7 @@ import (

 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"

 	"github.com/warmbly/warmbly/internal/infrastructure/db"
 	"github.com/warmbly/warmbly/internal/models"
@@ -151,6 +152,20 @@ type orgTransferRepository struct {
 	DB *db.DB
 }

+type orgImportTx struct {
+	pgx.Tx
+	indexedOrigin         bool
+	importedSendTasks     []uuid.UUID
+	importedHoldTasks     []uuid.UUID
+	importedAttemptNonces []uuid.UUID
+}
+
+const (
+	importOriginTask = iota + 1
+	importOriginHold
+	importOriginAttempt
+)
+
 func NewOrgTransferRepository(database *db.DB) OrgTransferRepository {
 	return &orgTransferRepository{DB: database}
 }
@@ -171,7 +186,45 @@ func quoteIdents(names []string) string {
 }

 func (r *orgTransferRepository) Begin(ctx context.Context) (pgx.Tx, error) {
-	return r.DB.Begin(ctx)
+	tx, err := r.DB.Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	indexed, err := setupImportOrigin(ctx, tx)
+	if err != nil {
+		_ = tx.Rollback(ctx)
+		return nil, err
+	}
+	return &orgImportTx{Tx: tx, indexedOrigin: indexed, importedSendTasks: []uuid.UUID{}, importedHoldTasks: []uuid.UUID{}, importedAttemptNonces: []uuid.UUID{}}, nil
+}
+
+func setupImportOrigin(ctx context.Context, tx pgx.Tx) (bool, error) {
+	if _, err := tx.Exec(ctx, `SAVEPOINT org_import_origin_setup`); err != nil {
+		return false, fmt.Errorf("prepare import provenance: %w", err)
+	}
+	_, createErr := tx.Exec(ctx, `CREATE TEMP TABLE pg_temp.org_import_send_origin (
+		kind smallint NOT NULL, id uuid NOT NULL, PRIMARY KEY (kind, id)) ON COMMIT DROP`)
+	indexed := createErr == nil
+	if createErr != nil {
+		var pgErr *pgconn.PgError
+		if !errors.As(createErr, &pgErr) || pgErr.Code != "42501" {
+			return false, fmt.Errorf("create import provenance: %w", createErr)
+		}
+		if _, err := tx.Exec(ctx, `ROLLBACK TO SAVEPOINT org_import_origin_setup`); err != nil {
+			return false, fmt.Errorf("restore import after missing TEMP privilege: %w", err)
+		}
+		var allowed bool
+		if err := tx.QueryRow(ctx, `SELECT has_database_privilege(current_user,current_database(),'TEMP')`).Scan(&allowed); err != nil {
+			return false, fmt.Errorf("check import TEMP privilege after initialization failure: %w", errors.Join(createErr, err))
+		}
+		if allowed {
+			return false, fmt.Errorf("create import provenance: %w", createErr)
+		}
+	}
+	if _, err := tx.Exec(ctx, `RELEASE SAVEPOINT org_import_origin_setup`); err != nil {
+		return false, fmt.Errorf("finish import provenance setup: %w", err)
+	}
+	return indexed, nil
 }

 // ---------- schema introspection ----------
@@ -474,6 +527,92 @@ func matchClause(pk []string) string {
 	return strings.Join(parts, " AND ")
 }

+func importOriginContains(kind int, id string, indexed bool) string {
+	if indexed {
+		// Keep the lookup scalar to avoid de-correlation into a scan of the growing TEMP ledger.
+		return `COALESCE((SELECT true FROM pg_temp.org_import_send_origin origin WHERE origin.kind=` + strconv.Itoa(kind) + ` AND origin.id=` + id + ` LIMIT 1),false)`
+	}
+	switch kind {
+	case importOriginTask:
+		return id + `=ANY($import_source_send_tasks::uuid[])`
+	case importOriginHold:
+		return id + `=ANY($import_source_hold_tasks::uuid[])`
+	default:
+		return id + `=ANY($import_source_attempt_nonces::uuid[])`
+	}
+}
+
+func localImportSendTask(taskID string, indexed bool) string {
+	return `EXISTS(SELECT 1 FROM tasks local_task WHERE local_task.id=` + taskID + ` AND (` + localImportSendTaskRow("local_task", indexed) + `))`
+}
+
+func localImportSendTaskRow(alias string, indexed bool) string {
+	return strings.ReplaceAll(fmt.Sprintf(`NOT (%s) AND (
+		local_task.send_executor_nonce IS NOT NULL
+		OR local_task.send_reserved_at IS NOT NULL
+		OR local_task.send_executor_started_at IS NOT NULL
+		OR local_task.send_executor_result IS NOT NULL
+		OR local_task.send_result_state IS NOT NULL
+		OR local_task.send_result_evidence IS NOT NULL
+		OR local_task.send_result_applied_at IS NOT NULL
+		OR local_task.send_released_at IS NOT NULL
+		OR (local_task.status='completed' AND local_task.completed_at IS NOT NULL)
+		OR EXISTS(SELECT 1 FROM email_accounts local_mailbox WHERE local_mailbox.id=local_task.email_account_id
+			AND local_mailbox.send_recovery_hold AND local_mailbox.send_recovery_task_id=local_task.id
+			AND NOT (%s))
+		OR EXISTS(SELECT 1 FROM outbound_attempts local_attempt WHERE local_attempt.task_id=local_task.id
+			AND NOT (%s))
+		OR EXISTS(SELECT 1 FROM warmup_tasks local_dispatch WHERE local_dispatch.task_id=local_task.id AND (
+			local_dispatch.dispatch_nonce IS NOT NULL OR local_dispatch.dispatch_started_at IS NOT NULL
+			OR local_dispatch.dispatch_result IS NOT NULL OR local_dispatch.warmup_charged_date IS NOT NULL
+			OR local_dispatch.warmup_refunded_at IS NOT NULL))
+		OR EXISTS(SELECT 1 FROM send_recovery_resolutions local_resolution WHERE local_resolution.recovery_task_id=local_task.id AND local_resolution.recorded_locally_at IS NOT NULL))`,
+		importOriginContains(importOriginTask, "local_task.id", indexed),
+		importOriginContains(importOriginHold, "local_task.id", indexed),
+		importOriginContains(importOriginAttempt, "local_attempt.nonce", indexed)), "local_task.", alias+".")
+}
+
+func importSendOverwriteGuard(table string, indexed bool) string {
+	switch table {
+	case "tasks":
+		return localImportSendTaskRow("dest", indexed)
+	case "email_tasks", "campaign_tasks", "warmup_tasks", "warmup_tokens", "outbound_attempts", "task_failures", "task_dead_letters", "task_execution_keys":
+		return localImportSendTask("dest.task_id", indexed)
+	case "send_recovery_resolutions":
+		return `dest.recorded_locally_at IS NOT NULL`
+	default:
+		return ""
+	}
+}
+
+func importSendColumnValue(table, column string, indexed bool) string {
+	value := "EXCLUDED." + quoteIdent(column)
+	switch table {
+	case "email_accounts":
+		switch column {
+		case "send_recovery_hold", "send_recovery_task_id", "send_recovery_reason":
+			return `CASE WHEN dest.send_recovery_hold THEN dest.` + quoteIdent(column) + ` ELSE ` + value + ` END`
+		case "send_cooldown_until":
+			return `GREATEST(dest.send_cooldown_until,` + value + `)`
+		case "send_cooldown_provider":
+			return `CASE WHEN dest.send_recovery_hold OR (dest.send_cooldown_until>NOW() AND dest.send_cooldown_until>COALESCE(EXCLUDED.send_cooldown_until,'-infinity'::timestamptz)) THEN dest.send_cooldown_provider ELSE ` + value + ` END`
+		}
+	case "warmup_statistics":
+		if column == "emails_sent" || column == "emails_replied" {
+			return `CASE WHEN EXISTS(SELECT 1 FROM warmup_tasks local_charge JOIN tasks local_task ON local_task.id=local_charge.task_id
+				WHERE local_task.email_account_id=dest.email_account_id
+				AND NOT (` + importOriginContains(importOriginTask, "local_task.id", indexed) + `) AND (
+				(local_charge.warmup_charged_date=dest.date AND local_charge.warmup_reply_charged IS NOT NULL)
+				OR (local_charge.warmup_charged_date IS NULL AND local_task.completed_at IS NOT NULL
+				AND (local_task.completed_at AT TIME ZONE 'UTC')::date=dest.date
+				AND (local_task.status='completed' OR local_task.send_result_state IS NOT NULL)))) THEN dest.` + quoteIdent(column) + ` ELSE ` + value + ` END`
+		}
+	}
+	return value
+}
+
+var ErrImportSendRecoveryConflict = errors.New("archive and destination have different restrictive send-recovery holds, so nothing was imported")
+
 func (r *orgTransferRepository) InsertBatch(
 	ctx context.Context,
 	tx pgx.Tx,
@@ -496,10 +635,16 @@ func (r *orgTransferRepository) InsertBatch(
 	}

 	args := []any{payload}
+	importTx, _ := tx.(*orgImportTx)
+	indexed := importTx != nil && importTx.indexedOrigin
 	quoted := quoteIdents(cols)
 	q := `INSERT INTO public.` + quoteIdent(table) + ` AS dest (` + quoted + `)
 	      SELECT ` + quoted + `
-	        FROM jsonb_populate_recordset(NULL::public.` + quoteIdent(table) + `, $1::jsonb)`
+	        FROM jsonb_populate_recordset(NULL::public.` + quoteIdent(table) + `, $1::jsonb) AS incoming`
+	if guard := importSendOverwriteGuard(table, indexed); guard != "" && table != "tasks" && table != "send_recovery_resolutions" {
+		// New child keys cannot inject archive payload or execution evidence into a retained local task.
+		q += ` WHERE NOT (` + strings.ReplaceAll(guard, "dest.", "incoming.") + `)`
+	}

 	switch {
 	case conflict == models.OrgImportConflictOverwrite && len(pk) > 0:
@@ -512,7 +657,7 @@ func (r *orgTransferRepository) InsertBatch(
 			if key[c] {
 				continue
 			}
-			sets = append(sets, quoteIdent(c)+" = EXCLUDED."+quoteIdent(c))
+			sets = append(sets, quoteIdent(c)+" = "+importSendColumnValue(table, c, indexed))
 		}
 		for _, c := range resetOnOverwrite {
 			sets = append(sets, quoteIdent(c)+" = DEFAULT")
@@ -522,12 +667,20 @@ func (r *orgTransferRepository) InsertBatch(
 			q += ` ON CONFLICT DO NOTHING`
 		} else {
 			q += ` ON CONFLICT (` + quoteIdents(pk) + `) DO UPDATE SET ` + strings.Join(sets, ", ")
+			var conditions []string
 			if owner != "" {
 				// The owner sits in its own FROM, since here an unqualified column is ambiguous with EXCLUDED.
-				q += ` WHERE EXISTS (SELECT 1 FROM public.` + quoteIdent(table) + ` AS owned
-				        WHERE ` + pairClause("owned", pk, "dest", pk) + ` AND (` + bindOrgParam(owner, 2) + `))`
+				conditions = append(conditions, `EXISTS (SELECT 1 FROM public.`+quoteIdent(table)+` AS owned
+				        WHERE `+pairClause("owned", pk, "dest", pk)+` AND (`+bindOrgParam(owner, 2)+`))`)
 				args = append(args, orgID)
 			}
+			// Imported history cannot rewind an existing destination-local send or its application marker.
+			if guard := importSendOverwriteGuard(table, indexed); guard != "" {
+				conditions = append(conditions, `NOT (`+guard+`)`)
+			}
+			if len(conditions) > 0 {
+				q += ` WHERE ` + strings.Join(conditions, ` AND `)
+			}
 		}
 	default:
 		// Untargeted, so it covers every unique constraint on the table, not
@@ -535,6 +688,24 @@ func (r *orgTransferRepository) InsertBatch(
 		// unique index is skipped rather than aborting the whole import.
 		q += ` ON CONFLICT DO NOTHING`
 	}
+	if strings.Contains(q, "$import_source_") {
+		if importTx == nil {
+			return 0, errors.New("import send tracking requires an import transaction")
+		}
+		for _, binding := range []struct {
+			name string
+			ids  []uuid.UUID
+		}{
+			{"$import_source_send_tasks", importTx.importedSendTasks},
+			{"$import_source_hold_tasks", importTx.importedHoldTasks},
+			{"$import_source_attempt_nonces", importTx.importedAttemptNonces},
+		} {
+			if strings.Contains(q, binding.name) {
+				args = append(args, binding.ids)
+				q = strings.ReplaceAll(q, binding.name, fmt.Sprintf("$%d", len(args)))
+			}
+		}
+	}

 	if len(invalidations) > 0 {
 		for _, relation := range invalidations {
@@ -548,6 +719,49 @@ func (r *orgTransferRepository) InsertBatch(
 				}
 			}
 		}
+		if table == "email_accounts" {
+			if importTx != nil && indexed {
+				_, err := tx.Exec(ctx, `INSERT INTO pg_temp.org_import_send_origin (kind, id)
+					SELECT DISTINCT $2::smallint, incoming.send_recovery_task_id
+					FROM jsonb_populate_recordset(NULL::public.email_accounts,$1::jsonb) incoming
+					WHERE incoming.send_recovery_hold AND incoming.send_recovery_task_id IS NOT NULL
+					AND NOT EXISTS(SELECT 1 FROM tasks local_task JOIN email_accounts local_mailbox ON local_mailbox.id=local_task.email_account_id
+						WHERE local_task.id=incoming.send_recovery_task_id AND local_mailbox.send_recovery_hold
+						AND local_mailbox.send_recovery_task_id=local_task.id)
+					ON CONFLICT DO NOTHING`, payload, importOriginHold)
+				if err != nil {
+					return 0, fmt.Errorf("classify imported holds: %w", err)
+				}
+			} else if importTx != nil {
+				holdRows, err := tx.Query(ctx, `SELECT DISTINCT incoming.send_recovery_task_id
+					FROM jsonb_populate_recordset(NULL::public.email_accounts,$1::jsonb) incoming
+					WHERE incoming.send_recovery_hold AND incoming.send_recovery_task_id IS NOT NULL
+					AND NOT EXISTS(SELECT 1 FROM tasks local_task JOIN email_accounts local_mailbox ON local_mailbox.id=local_task.email_account_id
+						WHERE local_task.id=incoming.send_recovery_task_id AND local_mailbox.send_recovery_hold
+						AND local_mailbox.send_recovery_task_id=local_task.id)`, payload)
+				if err != nil {
+					return 0, fmt.Errorf("classify imported holds: %w", err)
+				}
+				ids, err := pgx.CollectRows(holdRows, pgx.RowTo[uuid.UUID])
+				if err != nil {
+					return 0, err
+				}
+				importTx.importedHoldTasks = append(importTx.importedHoldTasks, ids...)
+			}
+		}
+		if conflict == models.OrgImportConflictOverwrite && table == "email_accounts" {
+			var incompatible bool
+			err := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM jsonb_populate_recordset(NULL::public.email_accounts,$1::jsonb) incoming
+				JOIN email_accounts current ON current.id=incoming.id WHERE current.organization_id=$2
+				AND current.send_recovery_hold AND incoming.send_recovery_hold
+				AND (current.send_recovery_task_id IS DISTINCT FROM incoming.send_recovery_task_id OR current.send_recovery_reason IS DISTINCT FROM incoming.send_recovery_reason))`, payload, orgID).Scan(&incompatible)
+			if err != nil {
+				return 0, err
+			}
+			if incompatible {
+				return 0, ErrImportSendRecoveryConflict
+			}
+		}
 		keys := make([]string, 0, len(invalidations))
 		seen := map[string]bool{}
 		for _, relation := range invalidations {
@@ -575,6 +789,44 @@ func (r *orgTransferRepository) InsertBatch(
 		}
 		return written, nil
 	}
+	if table == "tasks" || table == "outbound_attempts" {
+		if importTx == nil {
+			return 0, errors.New("import send tracking requires an import transaction")
+		}
+		key := "id"
+		kind := importOriginTask
+		if table == "outbound_attempts" {
+			key = "nonce"
+			kind = importOriginAttempt
+		}
+		if indexed {
+			written := `WITH written AS (` + q + ` RETURNING ` + quoteIdent(key) + `), recorded AS (
+				INSERT INTO pg_temp.org_import_send_origin (kind, id)
+				SELECT ` + strconv.Itoa(kind) + `, ` + quoteIdent(key) + ` FROM written
+				ON CONFLICT DO NOTHING RETURNING 1)
+				SELECT (SELECT COUNT(*) FROM written), (SELECT COUNT(*) FROM recorded)`
+			var count, recorded int64
+			if err := tx.QueryRow(ctx, written, args...).Scan(&count, &recorded); err != nil {
+				return 0, fmt.Errorf("insert into %s and record provenance: %w", table, err)
+			}
+			return count, nil
+		}
+		writtenRows, err := tx.Query(ctx, q+" RETURNING "+quoteIdent(key), args...)
+		if err != nil {
+			return 0, fmt.Errorf("insert into %s: %w", table, err)
+		}
+		ids, err := pgx.CollectRows(writtenRows, pgx.RowTo[uuid.UUID])
+		if err != nil {
+			return 0, err
+		}
+		// Only actual source writes lose preservation provenance; concurrent local completions remain visible.
+		if table == "tasks" {
+			importTx.importedSendTasks = append(importTx.importedSendTasks, ids...)
+		} else {
+			importTx.importedAttemptNonces = append(importTx.importedAttemptNonces, ids...)
+		}
+		return int64(len(ids)), nil
+	}
 	tag, err := tx.Exec(ctx, q, args...)
 	if err != nil {
 		return 0, fmt.Errorf("insert into %s: %w", table, err)