fix: rebind send now claims on session handoff (#3845)

Co-authored-by: ayattara <alassane.yattara@savoirfairelinux.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>

apps/backend/internal/orchestrator/messagequeue/repository_send_now_fallback_test.go

@@ -0,0 +1,185 @@
+package messagequeue
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+)
+
+func TestSQLiteLegacyTransferRejectsIdentityAwareSendNowClaim(t *testing.T) {
+	ctx := context.Background()
+	repo := newTestSQLiteRepo(t).(*sqliteRepository)
+	sourceIdentity := QueueSessionIdentity{
+		TaskID:               "task-1",
+		SessionID:            "session-old",
+		SessionIncarnationID: "incarnation-old",
+	}
+	destinationIdentity := QueueSessionIdentity{
+		TaskID:               sourceIdentity.TaskID,
+		SessionID:            "session-new",
+		SessionIncarnationID: "incarnation-new",
+	}
+	seedQueueSessionIdentity(t, repo, sourceIdentity)
+	seedQueueSessionIdentity(t, repo, destinationIdentity)
+	source := &QueuedMessage{
+		ID:        "source",
+		TaskID:    sourceIdentity.TaskID,
+		SessionID: sourceIdentity.SessionID,
+		Content:   "identity-aware source",
+		QueuedBy:  QueuedByUser,
+	}
+	if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source}); err != nil {
+		t.Fatal(err)
+	}
+
+	err := repo.TransferSession(ctx, sourceIdentity.SessionID, destinationIdentity.SessionID)
+	if !errors.Is(err, ErrSessionIdentityMismatch) {
+		t.Fatalf("legacy transfer error = %v, want ErrSessionIdentityMismatch", err)
+	}
+	pending, err := repo.ListPendingSendNowClaims(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pending) != 1 || pending[0].Claim.Identity != sourceIdentity {
+		t.Fatalf("pending claims after rejected transfer = %#v, want source identity", pending)
+	}
+}
+
+func TestDurableTransferRejectsIdentityAwareClaimWhenDestinationIdentityIsUnavailable(t *testing.T) {
+	ctx := context.Background()
+	repo := newTestSQLiteRepo(t).(*sqliteRepository)
+	service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession)
+	sourceIdentity := QueueSessionIdentity{
+		TaskID:               "task-1",
+		SessionID:            "session-old",
+		SessionIncarnationID: "incarnation-old",
+	}
+	destinationIdentity := QueueSessionIdentity{TaskID: sourceIdentity.TaskID, SessionID: "session-new"}
+	seedQueueSessionIdentity(t, repo, sourceIdentity)
+	seedQueueSessionIdentity(t, repo, destinationIdentity)
+	source := &QueuedMessage{
+		ID:        "source",
+		TaskID:    sourceIdentity.TaskID,
+		SessionID: sourceIdentity.SessionID,
+		Content:   "identity-aware source",
+		QueuedBy:  QueuedByUser,
+	}
+	if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source}); err != nil {
+		t.Fatal(err)
+	}
+
+	err := service.TransferSessionWithDurableAttachmentPreparation(
+		ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil,
+	)
+	if !errors.Is(err, ErrSessionIdentityMismatch) {
+		t.Fatalf("durable transfer error = %v, want ErrSessionIdentityMismatch", err)
+	}
+	pending, err := repo.ListPendingSendNowClaims(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(pending) != 1 || pending[0].Claim.Identity != sourceIdentity {
+		t.Fatalf("pending claims after rejected durable transfer = %#v, want source identity", pending)
+	}
+}
+
+func TestTransfersRejectReincarnatedIdentityAwareSendNowClaim(t *testing.T) {
+	transfers := []struct {
+		name string
+		run  func(context.Context, *sqliteRepository, QueueSessionIdentity, QueueSessionIdentity) error
+	}{
+		{
+			name: "direct",
+			run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error {
+				return repo.TransferSessionIdentities(ctx, source, destination)
+			},
+		},
+		{
+			name: "durable",
+			run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error {
+				service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession)
+				return service.TransferSessionWithDurableAttachmentPreparation(
+					ctx, source.TaskID, source.SessionID, destination.SessionID, nil, nil,
+				)
+			},
+		},
+	}
+
+	for _, transfer := range transfers {
+		for _, accepted := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/accepted=%t", transfer.name, accepted), func(t *testing.T) {
+				ctx := context.Background()
+				repo := newTestSQLiteRepo(t).(*sqliteRepository)
+				originalSource := QueueSessionIdentity{
+					TaskID: "task-1", SessionID: "session-old", SessionIncarnationID: "incarnation-old",
+				}
+				liveSource := originalSource
+				liveSource.SessionIncarnationID = "incarnation-recreated"
+				destination := QueueSessionIdentity{
+					TaskID: originalSource.TaskID, SessionID: "session-new", SessionIncarnationID: "incarnation-new",
+				}
+				seedQueueSessionIdentity(t, repo, originalSource)
+				seedQueueSessionIdentity(t, repo, destination)
+				first := &QueuedMessage{
+					ID: "first", TaskID: originalSource.TaskID, SessionID: originalSource.SessionID,
+					Content: "first", QueuedBy: QueuedByUser,
+				}
+				second := &QueuedMessage{
+					ID: "second", TaskID: originalSource.TaskID, SessionID: originalSource.SessionID,
+					Content: "second", QueuedBy: QueuedByUser,
+				}
+				if err := repo.InsertForSession(ctx, originalSource, first, DefaultMaxPerSession); err != nil {
+					t.Fatal(err)
+				}
+				if err := repo.InsertForSession(ctx, originalSource, second, DefaultMaxPerSession); err != nil {
+					t.Fatal(err)
+				}
+				claim, err := repo.ClaimSendNowForSession(ctx, originalSource, []QueuedMessage{*first, *second})
+				if err != nil {
+					t.Fatal(err)
+				}
+				if accepted {
+					if err := repo.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil {
+						t.Fatal(err)
+					}
+				}
+				seedQueueSessionIdentity(t, repo, liveSource)
+
+				err = transfer.run(ctx, repo, liveSource, destination)
+				if !errors.Is(err, ErrSessionIdentityMismatch) {
+					t.Fatalf("transfer error = %v, want ErrSessionIdentityMismatch", err)
+				}
+				pending, err := repo.ListPendingSendNowClaims(ctx)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if len(pending) != 1 {
+					t.Fatalf("pending claims = %#v, want one preserved claim", pending)
+				}
+				preserved := pending[0]
+				if preserved.Claim.ClaimID != claim.ClaimID || preserved.Claim.Identity != originalSource ||
+					preserved.Accepted != accepted || preserved.Claim.Dispatch.Content != "first\n\nsecond" {
+					t.Fatalf("preserved claim = %#v", preserved)
+				}
+				contents := []string{preserved.Claim.Sources[0].Content, preserved.Claim.Sources[1].Content}
+				if contents[0] != "first" || contents[1] != "second" {
+					t.Fatalf("preserved claim sources = %#v", preserved.Claim.Sources)
+				}
+				destinationEntries, err := repo.ListBySession(ctx, destination.SessionID)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if len(destinationEntries) != 0 {
+					t.Fatalf("destination entries after rejected transfer = %#v", destinationEntries)
+				}
+			})
+		}
+	}
+}

apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery.go

@@ -140,6 +140,9 @@ func (r *sqliteRepository) transferPendingSendNowClaimTx(
 	ctx context.Context,
 	tx *sqlx.Tx,
 	oldSessionID, newSessionID string,
+	source, destination *QueueSessionIdentity,
+	sourceGeneration int64,
+	queuePositionOffset int64,
 ) error {
 	var claimID, claimJSON string
 	err := tx.QueryRowxContext(ctx, r.db.Rebind(`
@@ -174,14 +177,30 @@ func (r *sqliteRepository) transferPendingSendNowClaimTx(
 	if err != nil || sessionID != oldSessionID || claim.Dispatch.SessionID != oldSessionID {
 		return ErrSendNowClaimChanged
 	}
+	if claim.Identity.SessionIncarnationID != "" {
+		if source == nil || claim.Identity != *source {
+			return ErrSessionIdentityMismatch
+		}
+		if claim.OperationGeneration != sourceGeneration || claim.SessionGeneration != sourceGeneration {
+			return ErrSendNowClaimChanged
+		}
+	}
+	if destination == nil && claim.Identity.SessionIncarnationID != "" {
+		return ErrSessionIdentityMismatch
+	}
 	for sourceIndex := range claim.Sources {
 		claim.Sources[sourceIndex].SessionID = newSessionID
+		claim.Sources[sourceIndex].Position += queuePositionOffset
 	}
 	claim.Dispatch.SessionID = newSessionID
+	if destination != nil {
+		claim.Identity = *destination
+	}
 	claim.SessionGeneration, err = r.getSendNowGenerationTx(ctx, tx, newSessionID)
 	if err != nil {
 		return err
 	}
+	claim.OperationGeneration = claim.SessionGeneration
 	claimJSONBytes, err := json.Marshal(&claim)
 	if err != nil {
 		return fmt.Errorf("marshal transferred Send Now claim: %w", err)

apps/backend/internal/orchestrator/messagequeue/repository_send_now_recovery_test.go

@@ -113,6 +113,201 @@ func TestSQLiteTransferSessionMovesPendingSendNowClaim(t *testing.T) {
 	}
 }
 
+func TestSQLiteTransferSessionIdentitiesRebindsPendingSendNowClaim(t *testing.T) {
+	for _, accepted := range []bool{false, true} {
+		t.Run(fmt.Sprintf("accepted=%t", accepted), func(t *testing.T) {
+			ctx := context.Background()
+			repo := newTestSQLiteRepo(t).(*sqliteRepository)
+			sourceIdentity := QueueSessionIdentity{
+				TaskID:               "task-1",
+				SessionID:            "session-old",
+				SessionIncarnationID: "incarnation-old",
+			}
+			destinationIdentity := QueueSessionIdentity{
+				TaskID:               "task-1",
+				SessionID:            "session-new",
+				SessionIncarnationID: "incarnation-new",
+			}
+			seedQueueSessionIdentity(t, repo, sourceIdentity)
+			seedQueueSessionIdentity(t, repo, destinationIdentity)
+			destinationPrime := &QueuedMessage{
+				ID: "destination-prime", TaskID: destinationIdentity.TaskID, SessionID: destinationIdentity.SessionID,
+				Content: "already settled", QueuedBy: QueuedByUser,
+			}
+			if err := repo.InsertForSession(ctx, destinationIdentity, destinationPrime, DefaultMaxPerSession); err != nil {
+				t.Fatal(err)
+			}
+			primeClaim, err := repo.ClaimSendNowForSession(ctx, destinationIdentity, []QueuedMessage{*destinationPrime})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := repo.AcknowledgeSendNowClaim(ctx, primeClaim); err != nil {
+				t.Fatal(err)
+			}
+			source := &QueuedMessage{
+				ID: "source", TaskID: sourceIdentity.TaskID, SessionID: sourceIdentity.SessionID,
+				Content: "durable source", QueuedBy: QueuedByWorkflow,
+				Metadata: map[string]interface{}{MetadataLifecycleDurable: true},
+			}
+			if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil {
+				t.Fatal(err)
+			}
+			claim, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source})
+			if err != nil {
+				t.Fatal(err)
+			}
+			sourceGeneration := claim.OperationGeneration
+			if accepted {
+				if err := repo.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			if err := repo.TransferSessionIdentities(ctx, sourceIdentity, destinationIdentity); err != nil {
+				t.Fatal(err)
+			}
+
+			pending, err := repo.ListPendingSendNowClaims(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(pending) != 1 {
+				t.Fatalf("pending claims = %#v, want one transferred claim", pending)
+			}
+			transferred := pending[0]
+			if transferred.Claim.Identity != destinationIdentity {
+				t.Fatalf("transferred claim identity = %#v, want %#v", transferred.Claim.Identity, destinationIdentity)
+			}
+			if transferred.Claim.OperationGeneration != transferred.Claim.SessionGeneration {
+				t.Fatalf("transferred claim generations = operation %d, session %d", transferred.Claim.OperationGeneration, transferred.Claim.SessionGeneration)
+			}
+			if transferred.Claim.OperationGeneration <= sourceGeneration {
+				t.Fatalf("transferred operation generation = %d, want greater than source %d", transferred.Claim.OperationGeneration, sourceGeneration)
+			}
+			if accepted {
+				err = repo.AcknowledgeSendNowClaim(ctx, &transferred.Claim)
+			} else {
+				err = repo.RestoreSendNowClaim(ctx, &transferred.Claim)
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestTransferredSendNowClaimFencesOriginalSourceActions(t *testing.T) {
+	transfers := []struct {
+		name string
+		run  func(context.Context, *sqliteRepository, QueueSessionIdentity, QueueSessionIdentity) error
+	}{
+		{
+			name: "direct",
+			run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error {
+				return repo.TransferSessionIdentities(ctx, source, destination)
+			},
+		},
+		{
+			name: "durable",
+			run: func(ctx context.Context, repo *sqliteRepository, source, destination QueueSessionIdentity) error {
+				service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession)
+				return service.TransferSessionWithDurableAttachmentPreparation(
+					ctx, source.TaskID, source.SessionID, destination.SessionID, nil, nil,
+				)
+			},
+		},
+	}
+	actions := []struct {
+		name string
+		run  func(context.Context, *sqliteRepository, *SendNowClaim) error
+	}{
+		{name: "restore", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error {
+			return repo.RestoreSendNowClaim(ctx, claim)
+		}},
+		{name: "acknowledge", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error {
+			return repo.AcknowledgeSendNowClaim(ctx, claim)
+		}},
+		{name: "accept", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error {
+			return repo.MarkPendingSendNowClaimAccepted(ctx, claim)
+		}},
+		{name: "startup discard", run: func(ctx context.Context, repo *sqliteRepository, claim *SendNowClaim) error {
+			return repo.DeletePendingSendNowClaim(ctx, claim)
+		}},
+	}
+
+	for _, transfer := range transfers {
+		for _, accepted := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/accepted=%t", transfer.name, accepted), func(t *testing.T) {
+				ctx := context.Background()
+				repo := newTestSQLiteRepo(t).(*sqliteRepository)
+				sourceIdentity := QueueSessionIdentity{
+					TaskID:               "task-1",
+					SessionID:            "session-old",
+					SessionIncarnationID: "incarnation-old",
+				}
+				destinationIdentity := QueueSessionIdentity{
+					TaskID:               sourceIdentity.TaskID,
+					SessionID:            "session-new",
+					SessionIncarnationID: "incarnation-new",
+				}
+				seedQueueSessionIdentity(t, repo, sourceIdentity)
+				seedQueueSessionIdentity(t, repo, destinationIdentity)
+				for _, source := range []*QueuedMessage{
+					{ID: "first", TaskID: sourceIdentity.TaskID, SessionID: sourceIdentity.SessionID, Content: "first", QueuedBy: QueuedByUser},
+					{ID: "second", TaskID: sourceIdentity.TaskID, SessionID: sourceIdentity.SessionID, Content: "second", QueuedBy: QueuedByUser},
+				} {
+					if err := repo.InsertForSession(ctx, sourceIdentity, source, DefaultMaxPerSession); err != nil {
+						t.Fatal(err)
+					}
+				}
+				sources, err := repo.ListBySession(ctx, sourceIdentity.SessionID)
+				if err != nil {
+					t.Fatal(err)
+				}
+				claim, err := repo.ClaimSendNowForSession(ctx, sourceIdentity, sources)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if accepted {
+					if err := repo.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if err := transfer.run(ctx, repo, sourceIdentity, destinationIdentity); err != nil {
+					t.Fatal(err)
+				}
+
+				for _, action := range actions {
+					t.Run(action.name, func(t *testing.T) {
+						err := action.run(ctx, repo, claim)
+						if !errors.Is(err, ErrSendNowClaimChanged) && !errors.Is(err, ErrSessionIdentityMismatch) {
+							t.Fatalf("old source %s = %v, want claim fencing error", action.name, err)
+						}
+						pending, err := repo.ListPendingSendNowClaims(ctx)
+						if err != nil {
+							t.Fatal(err)
+						}
+						if len(pending) != 1 {
+							t.Fatalf("pending claims after stale %s = %#v, want successor claim", action.name, pending)
+						}
+						transferred := pending[0]
+						if transferred.Claim.ClaimID != claim.ClaimID || transferred.Claim.Identity != destinationIdentity || transferred.Accepted != accepted {
+							t.Fatalf("transferred claim after stale %s = %#v", action.name, transferred)
+						}
+						if len(transferred.Claim.Sources) != 2 ||
+							transferred.Claim.Sources[0].Content != "first" ||
+							transferred.Claim.Sources[1].Content != "second" ||
+							transferred.Claim.Sources[0].SessionID != destinationIdentity.SessionID ||
+							transferred.Claim.Sources[1].SessionID != destinationIdentity.SessionID {
+							t.Fatalf("transferred FIFO sources after stale %s = %#v", action.name, transferred.Claim.Sources)
+						}
+					})
+				}
+			})
+		}
+	}
+}
+
 func TestSQLitePurgeRemovesDurableSendNowClaims(t *testing.T) {
 	for _, purge := range []struct {
 		name string

apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go

@@ -5529,8 +5529,9 @@ func lifecycleReservationFromMetadataJSON(metadataJSON string) (string, bool, er
 func (r *sqliteRepository) transferSessionOwned(
 	ctx context.Context,
 	oldSessionID, newSessionID, operationID string,
+	source, destination *QueueSessionIdentity,
 ) error {
-	return r.transferSessionOwnedTx(ctx, oldSessionID, newSessionID, operationID)
+	return r.transferSessionOwnedTx(ctx, oldSessionID, newSessionID, operationID, source, destination)
 }
 
 func (r *sqliteRepository) beginAuthorizedSessionTransferTx(
@@ -5643,30 +5644,32 @@ func (r *sqliteRepository) transferSessionQueueRowsTx(
 	ctx context.Context,
 	tx *sqlx.Tx,
 	oldSessionID, newSessionID string,
-) error {
+) (int64, error) {
 	var destinationMax sql.NullInt64
 	if err := tx.GetContext(ctx, &destinationMax, r.db.Rebind(`
 		SELECT MAX(position) FROM queued_messages WHERE session_id = ?
 	`), newSessionID); err != nil {
-		return fmt.Errorf("transfer max: %w", err)
+		return 0, fmt.Errorf("transfer max: %w", err)
 	}
 	if _, err := tx.ExecContext(ctx, r.db.Rebind(`
 		UPDATE queued_messages
 		SET session_id = ?, position = position + ?
 		WHERE session_id = ?
 	`), newSessionID, destinationMax.Int64, oldSessionID); err != nil {
-		return fmt.Errorf("transfer queued: %w", err)
+		return 0, fmt.Errorf("transfer queued: %w", err)
 	}
 	var transferredMax sql.NullInt64
 	if err := tx.GetContext(ctx, &transferredMax, r.db.Rebind(`
 		SELECT MAX(position) FROM queued_messages WHERE session_id = ?
 	`), newSessionID); err != nil {
-		return fmt.Errorf("transfer destination max after move: %w", err)
+		return 0, fmt.Errorf("transfer destination max after move: %w", err)
 	}
 	if transferredMax.Valid {
-		return r.bumpQueuePositionTx(ctx, tx, newSessionID, transferredMax.Int64)
+		if err := r.bumpQueuePositionTx(ctx, tx, newSessionID, transferredMax.Int64); err != nil {
+			return 0, err
+		}
 	}
-	return nil
+	return destinationMax.Int64, nil
 }
 
 func (r *sqliteRepository) transferSessionStateTx(
@@ -5745,6 +5748,13 @@ func (r *sqliteRepository) transferSession(
 	if err := r.lockAndValidateTransferSessionsTx(ctx, tx, source, destination, first, second); err != nil {
 		return err
 	}
+	sourceGeneration := int64(0)
+	if source != nil {
+		sourceGeneration, err = r.getSendNowGenerationTx(ctx, tx, oldSessionID)
+		if err != nil {
+			return err
+		}
+	}
 	if oldSessionID == newSessionID {
 		return tx.Commit()
 	}
@@ -5770,7 +5780,8 @@ func (r *sqliteRepository) transferSession(
 	if err := r.deleteEditLeasesForTransferTx(ctx, tx, oldSessionID, newSessionID); err != nil {
 		return err
 	}
-	if err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID); err != nil {
+	queuePositionOffset, err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID)
+	if err != nil {
 		return err
 	}
 	if _, err := tx.ExecContext(ctx, r.db.Rebind(`DELETE FROM pending_moves WHERE session_id = ?`), newSessionID); err != nil {
@@ -5791,7 +5802,9 @@ func (r *sqliteRepository) transferSession(
 	if err := r.bumpSendNowGenerationTx(ctx, tx, newSessionID); err != nil {
 		return err
 	}
-	if err := r.transferPendingSendNowClaimTx(ctx, tx, oldSessionID, newSessionID); err != nil {
+	if err := r.transferPendingSendNowClaimTx(
+		ctx, tx, oldSessionID, newSessionID, source, destination, sourceGeneration, queuePositionOffset,
+	); err != nil {
 		return err
 	}
 	return tx.Commit()
@@ -5800,6 +5813,7 @@ func (r *sqliteRepository) transferSession(
 func (r *sqliteRepository) transferSessionOwnedTx(
 	ctx context.Context,
 	oldSessionID, newSessionID, operationID string,
+	source, destination *QueueSessionIdentity,
 ) error {
 	if err := r.ensureQueueDispatchRecoverySchema(ctx); err != nil {
 		return err
@@ -5834,6 +5848,23 @@ func (r *sqliteRepository) transferSessionOwnedTx(
 		return err
 	}
 	defer func() { _ = tx.Rollback() }()
+	if source != nil {
+		if err := r.validateSessionIdentityTx(ctx, tx, *source); err != nil {
+			return err
+		}
+	}
+	if destination != nil {
+		if err := r.validateSessionIdentityTx(ctx, tx, *destination); err != nil {
+			return err
+		}
+	}
+	sourceGeneration := int64(0)
+	if source != nil {
+		sourceGeneration, err = r.getSendNowGenerationTx(ctx, tx, oldSessionID)
+		if err != nil {
+			return err
+		}
+	}
 	if oldSessionID == newSessionID {
 		return commitAuthorizedSessionTransferTx(
 			ctx, tx, r.db, oldSessionID, newSessionID, operationID,
@@ -5845,13 +5876,16 @@ func (r *sqliteRepository) transferSessionOwnedTx(
 	if err := r.deleteEditLeasesForTransferTx(ctx, tx, oldSessionID, newSessionID); err != nil {
 		return err
 	}
-	if err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID); err != nil {
+	queuePositionOffset, err := r.transferSessionQueueRowsTx(ctx, tx, oldSessionID, newSessionID)
+	if err != nil {
 		return err
 	}
 	if err := r.transferSessionStateTx(ctx, tx, oldSessionID, newSessionID); err != nil {
 		return err
 	}
-	if err := r.transferPendingSendNowClaimTx(ctx, tx, oldSessionID, newSessionID); err != nil {
+	if err := r.transferPendingSendNowClaimTx(
+		ctx, tx, oldSessionID, newSessionID, source, destination, sourceGeneration, queuePositionOffset,
+	); err != nil {
 		return err
 	}
 	return commitAuthorizedSessionTransferTx(

apps/backend/internal/orchestrator/messagequeue/service.go

@@ -3373,12 +3373,13 @@ func (s *Service) SnapshotSession(ctx context.Context, sessionID string) ([]Queu
 }
 
 type ownedSessionTransferRepository interface {
-	transferSessionOwned(context.Context, string, string, string) error
+	transferSessionOwned(context.Context, string, string, string, *QueueSessionIdentity, *QueueSessionIdentity) error
 }
 
 func (s *Service) transferRepositorySession(
 	ctx context.Context,
 	oldSessionID, newSessionID, operationID string,
+	source, destination *QueueSessionIdentity,
 ) error {
 	if operationID == "" {
 		return s.repo.TransferSession(ctx, oldSessionID, newSessionID)
@@ -3387,7 +3388,7 @@ func (s *Service) transferRepositorySession(
 	if !ok {
 		return errors.New("owned session transfer unavailable")
 	}
-	return repo.transferSessionOwned(ctx, oldSessionID, newSessionID, operationID)
+	return repo.transferSessionOwned(ctx, oldSessionID, newSessionID, operationID, source, destination)
 }
 
 // SnapshotSessionForIdentity captures queue and deferred-move state for one exact incarnation.
@@ -3473,25 +3474,45 @@ func (s *Service) transferRepositorySessionForTask(
 	ctx context.Context,
 	taskID, oldSessionID, newSessionID, operationID string,
 ) error {
+	transferred, err := s.transferRepositorySessionWithLiveIdentities(
+		ctx, taskID, oldSessionID, newSessionID, operationID,
+	)
+	if err != nil {
+		return err
+	}
+	if transferred {
+		return nil
+	}
+	return s.transferRepositorySession(ctx, oldSessionID, newSessionID, operationID, nil, nil)
+}
+
+func (s *Service) transferRepositorySessionWithLiveIdentities(
+	ctx context.Context,
+	taskID, oldSessionID, newSessionID, operationID string,
+) (bool, error) {
 	// Transfers that do not have a durable compensation record can use the
 	// immutable session identities. This fences a workflow handoff to one task
 	// and lets repositories reject a stale or cross-task destination. Durable
 	// transfers keep the owned-operation path because it also carries the
 	// compensation lease and operation token.
-	if taskID != "" && operationID == "" {
-		source, sourceErr := s.repo.ResolveSessionIdentity(ctx, taskID, oldSessionID)
-		if sourceErr != nil && !errors.Is(sourceErr, ErrSessionIdentityMismatch) {
-			return sourceErr
-		}
-		destination, destinationErr := s.repo.ResolveSessionIdentity(ctx, taskID, newSessionID)
-		if destinationErr != nil && !errors.Is(destinationErr, ErrSessionIdentityMismatch) {
-			return destinationErr
-		}
-		if sourceErr == nil && destinationErr == nil {
-			return s.repo.TransferSessionIdentities(ctx, source, destination)
-		}
+	if taskID == "" {
+		return false, nil
 	}
-	return s.transferRepositorySession(ctx, oldSessionID, newSessionID, operationID)
+	source, sourceErr := s.repo.ResolveSessionIdentity(ctx, taskID, oldSessionID)
+	if sourceErr != nil && !errors.Is(sourceErr, ErrSessionIdentityMismatch) {
+		return false, sourceErr
+	}
+	destination, destinationErr := s.repo.ResolveSessionIdentity(ctx, taskID, newSessionID)
+	if destinationErr != nil && !errors.Is(destinationErr, ErrSessionIdentityMismatch) {
+		return false, destinationErr
+	}
+	if sourceErr != nil || destinationErr != nil {
+		return false, nil
+	}
+	if operationID == "" {
+		return true, s.repo.TransferSessionIdentities(ctx, source, destination)
+	}
+	return true, s.transferRepositorySession(ctx, oldSessionID, newSessionID, operationID, &source, &destination)
 }
 
 type attachmentCleanupRepository interface {

apps/backend/internal/orchestrator/messagequeue/service_session_transfer.go

@@ -170,6 +170,14 @@ func (s *Service) prepareDurableSessionTransfer(
 	}
 	entryIDs = append(entryIDs, dispatchEntryIDs...)
 	attachmentIDs = append(attachmentIDs, dispatchAttachments...)
+	claimEntryIDs, claimAttachments, err := s.snapshotPendingSendNowClaimTransfer(
+		ctx, taskID, oldSessionID, seenEntryIDs, seenAttachmentIDs,
+	)
+	if err != nil {
+		return err
+	}
+	entryIDs = append(entryIDs, claimEntryIDs...)
+	attachmentIDs = append(attachmentIDs, claimAttachments...)
 	cleanupEntryIDs, cleanupAttachments, cleanupLocators, err := s.snapshotAttachmentCleanupTransfer(
 		ctx, taskID, oldSessionID, seenEntryIDs, seenAttachmentIDs,
 	)
@@ -208,6 +216,43 @@ func (s *Service) prepareDurableSessionTransfer(
 	return nil
 }
 
+func (s *Service) snapshotPendingSendNowClaimTransfer(
+	ctx context.Context,
+	taskID, sessionID string,
+	seenEntryIDs, seenAttachmentIDs map[string]struct{},
+) ([]string, []string, error) {
+	if !s.PendingSendNowClaimPersistenceAvailable() {
+		return nil, nil, nil
+	}
+	claims, err := s.ListPendingSendNowClaims(ctx)
+	if err != nil {
+		return nil, nil, fmt.Errorf("snapshot pending Send Now claims for transfer: %w", err)
+	}
+	entryIDs := make([]string, 0)
+	attachmentIDs := make([]string, 0)
+	for _, pending := range claims {
+		claim := pending.Claim
+		if claim.Dispatch.SessionID != sessionID ||
+			(taskID != "" && claim.Dispatch.TaskID != "" && claim.Dispatch.TaskID != taskID) {
+			continue
+		}
+		for _, source := range claim.Sources {
+			if source.SessionID != sessionID || (taskID != "" && source.TaskID != taskID) {
+				continue
+			}
+			if source.ID != "" {
+				if _, seen := seenEntryIDs[source.ID]; !seen {
+					entryIDs = append(entryIDs, source.ID)
+					seenEntryIDs[source.ID] = struct{}{}
+				}
+			}
+			attachmentIDs = appendTransferAttachmentIDs(attachmentIDs, seenAttachmentIDs, source.Attachments)
+		}
+		attachmentIDs = appendTransferAttachmentIDs(attachmentIDs, seenAttachmentIDs, claim.Dispatch.Attachments)
+	}
+	return entryIDs, attachmentIDs, nil
+}
+
 func (s *Service) snapshotPendingDispatchTransfer(
 	ctx context.Context,
 	sessionID string,

apps/backend/internal/orchestrator/messagequeue/service_transfer_compensation_test.go

@@ -3,9 +3,11 @@ package messagequeue
 import (
 	"context"
 	"errors"
+	"fmt"
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"testing"
 )
 
 func TestTransferSessionWithPreparationRollsBackPreparationError(t *testing.T) {
@@ -126,6 +128,100 @@ func TestDurableSessionTransferIncludesInFlightOrdinaryAttachment(t *testing.T)
 	assert.Equal(t, queued.ID, pending[0].Message.ID)
 }
 
+func TestDurableSessionTransferRebindsPendingSendNowClaimIdentity(t *testing.T) {
+	for _, accepted := range []bool{false, true} {
+		t.Run(fmt.Sprintf("accepted=%t", accepted), func(t *testing.T) {
+			ctx := context.Background()
+			repo := newTestSQLiteRepo(t)
+			sourceIdentity := QueueSessionIdentity{
+				TaskID:               "task",
+				SessionID:            "session-old",
+				SessionIncarnationID: "incarnation-old",
+			}
+			destinationIdentity := QueueSessionIdentity{
+				TaskID:               sourceIdentity.TaskID,
+				SessionID:            "session-new",
+				SessionIncarnationID: "incarnation-new",
+			}
+			seedQueueSessionIdentity(t, repo, sourceIdentity)
+			seedQueueSessionIdentity(t, repo, destinationIdentity)
+			service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession)
+			source, err := service.QueueMessageWithMetadataForSession(
+				ctx, sourceIdentity, "handoff", "", "user", false, nil, nil,
+			)
+			require.NoError(t, err)
+			claim, err := service.ClaimSendNowForSession(ctx, sourceIdentity, []QueuedMessage{*source})
+			require.NoError(t, err)
+			if accepted {
+				require.NoError(t, service.MarkPendingSendNowClaimAccepted(ctx, claim))
+			}
+
+			require.NoError(t, service.TransferSessionWithDurableAttachmentPreparation(
+				ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil,
+			))
+			pending, err := service.ListPendingSendNowClaims(ctx)
+			require.NoError(t, err)
+			require.Len(t, pending, 1)
+			assert.Equal(t, destinationIdentity, pending[0].Claim.Identity)
+			assert.Equal(t, pending[0].Claim.SessionGeneration, pending[0].Claim.OperationGeneration)
+			assert.Equal(t, accepted, pending[0].Accepted)
+		})
+	}
+}
+
+func TestDurableSessionTransferRestoresPendingSendNowClaimAfterDestinationQueue(t *testing.T) {
+	ctx := context.Background()
+	repo := newTestSQLiteRepo(t)
+	service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession)
+	service.SetAutoMergeEnabled(false)
+	sourceIdentity := QueueSessionIdentity{
+		TaskID:               "task",
+		SessionID:            "session-old",
+		SessionIncarnationID: "incarnation-old",
+	}
+	destinationIdentity := QueueSessionIdentity{
+		TaskID:               sourceIdentity.TaskID,
+		SessionID:            "session-new",
+		SessionIncarnationID: "incarnation-new",
+	}
+	seedQueueSessionIdentity(t, repo, sourceIdentity)
+	seedQueueSessionIdentity(t, repo, destinationIdentity)
+	_, err := service.QueueMessageWithMetadataForSession(
+		ctx, destinationIdentity, "destination", "", QueuedByUser, false, nil, nil,
+	)
+	require.NoError(t, err)
+	first, err := service.QueueMessageWithMetadataForSession(
+		ctx, sourceIdentity, "first", "", QueuedByUser, false, nil, nil,
+	)
+	require.NoError(t, err)
+	second, err := service.QueueMessageWithMetadataForSession(
+		ctx, sourceIdentity, "second", "", QueuedByUser, false, nil, nil,
+	)
+	require.NoError(t, err)
+	sources, err := repo.ListBySession(ctx, sourceIdentity.SessionID)
+	require.NoError(t, err)
+	require.Equal(t, []string{first.ID, second.ID}, []string{sources[0].ID, sources[1].ID})
+	claim, err := service.ClaimSendNowForSession(ctx, sourceIdentity, sources)
+	require.NoError(t, err)
+
+	require.NoError(t, service.TransferSessionWithDurableAttachmentPreparation(
+		ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil,
+	))
+	pending, err := service.ListPendingSendNowClaims(ctx)
+	require.NoError(t, err)
+	require.Len(t, pending, 1)
+	require.Equal(t, claim.ClaimID, pending[0].Claim.ClaimID)
+	require.NoError(t, service.RestoreSendNowClaim(ctx, &pending[0].Claim))
+
+	entries, err := repo.ListBySession(ctx, destinationIdentity.SessionID)
+	require.NoError(t, err)
+	require.Equal(t, []string{"destination", "first", "second"}, []string{
+		entries[0].Content, entries[1].Content, entries[2].Content,
+	})
+	require.Less(t, entries[0].Position, entries[1].Position)
+	require.Less(t, entries[1].Position, entries[2].Position)
+}
+
 func TestDurableSessionTransferIncludesCleanupOnlyAttachmentClaim(t *testing.T) {
 	ctx := context.Background()
 	repo := newTestSQLiteRepo(t)

apps/backend/internal/orchestrator/messagequeue/service_transfer_send_now_attachments_test.go

@@ -0,0 +1,56 @@
+package messagequeue
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDurableSessionTransferIncludesPendingSendNowClaimAttachments(t *testing.T) {
+	ctx := context.Background()
+	repo := newTestSQLiteRepo(t)
+	service := newAutoMergeTestServiceWithRepository(t, repo, DefaultMaxPerSession)
+	source, err := service.QueueMessage(
+		ctx,
+		"session-old",
+		"task",
+		"handoff",
+		"",
+		QueuedByUser,
+		false,
+		[]MessageAttachment{{AttachmentID: "claim-attachment"}},
+	)
+	require.NoError(t, err)
+	_, err = service.ClaimSendNow(ctx, source.SessionID, []QueuedMessage{*source})
+	require.NoError(t, err)
+
+	var preparedAttachments []string
+	err = service.TransferSessionWithDurableAttachmentPreparation(
+		ctx,
+		"task",
+		"session-old",
+		"session-new",
+		func(_ context.Context, attachmentIDs []string) error {
+			preparedAttachments = append(preparedAttachments, attachmentIDs...)
+			compensations, listErr := service.ListSessionTransferCompensations(ctx)
+			require.NoError(t, listErr)
+			require.Len(t, compensations, 1)
+			require.Equal(t, []string{"claim-attachment"}, compensations[0].AttachmentIDs)
+			return nil
+		},
+		nil,
+	)
+	require.NoError(t, err)
+	require.Equal(t, []string{"claim-attachment"}, preparedAttachments)
+
+	pending, err := service.ListPendingSendNowClaims(ctx)
+	require.NoError(t, err)
+	require.Len(t, pending, 1)
+	require.Equal(t, "session-new", pending[0].Claim.Dispatch.SessionID)
+	require.NoError(t, service.RestoreSendNowClaim(ctx, &pending[0].Claim))
+	entries, err := repo.ListBySession(ctx, "session-new")
+	require.NoError(t, err)
+	require.Len(t, entries, 1)
+	require.Equal(t, "claim-attachment", entries[0].Attachments[0].AttachmentID)
+}

apps/backend/internal/orchestrator/queue_send_now_recovery_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"sync/atomic"
 
@@ -111,6 +112,103 @@ func TestSendNowAcceptedClaimIsAcknowledgedAfterProcessRestart(t *testing.T) {
 	}
 }
 
+func TestTransferredIdentityAwareSendNowClaimReconcilesAfterProcessRestart(t *testing.T) {
+	for _, accepted := range []bool{false, true} {
+		t.Run(fmt.Sprintf("accepted=%t", accepted), func(t *testing.T) {
+			ctx := context.Background()
+			dbPath := filepath.Join(t.TempDir(), "queue.db")
+			queue, db, sourceIdentity, destinationIdentity := newIdentityWorkflowTransferQueue(t, dbPath, true)
+			source, err := queue.QueueMessageWithMetadataForSession(
+				ctx, sourceIdentity, "transferred prompt", "", "user", false, nil, nil,
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+			claim, err := queue.ClaimSendNowForSession(ctx, sourceIdentity, []messagequeue.QueuedMessage{*source})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if accepted {
+				if err := queue.MarkPendingSendNowClaimAccepted(ctx, claim); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if err := queue.TransferSessionWithDurableAttachmentPreparation(
+				ctx, sourceIdentity.TaskID, sourceIdentity.SessionID, destinationIdentity.SessionID, nil, nil,
+			); err != nil {
+				t.Fatal(err)
+			}
+			if _, err := db.Exec(`DELETE FROM task_sessions WHERE id = ?`, sourceIdentity.SessionID); err != nil {
+				t.Fatal(err)
+			}
+			if err := db.Close(); err != nil {
+				t.Fatal(err)
+			}
+
+			restartedQueue, restartedDB, _, _ := newIdentityWorkflowTransferQueue(t, dbPath, false)
+			t.Cleanup(func() { _ = restartedDB.Close() })
+			restarted := &Service{logger: testLogger(), messageQueue: restartedQueue}
+			if err := restarted.reconcilePendingSendNowClaimsOnStartup(ctx); err != nil {
+				t.Fatal(err)
+			}
+			status := restartedQueue.GetStatus(ctx, destinationIdentity.SessionID)
+			if accepted && status.Count != 0 {
+				t.Fatalf("accepted transferred claim restored after restart: %#v", status.Entries)
+			}
+			if !accepted && (len(status.Entries) != 1 || status.Entries[0].ID != source.ID) {
+				t.Fatalf("unaccepted transferred claim after restart = %#v, want source %s", status.Entries, source.ID)
+			}
+		})
+	}
+}
+
+func newIdentityWorkflowTransferQueue(
+	t *testing.T,
+	dbPath string,
+	seed bool,
+) (*messagequeue.Service, *sqlx.DB, messagequeue.QueueSessionIdentity, messagequeue.QueueSessionIdentity) {
+	t.Helper()
+	raw, err := sql.Open("sqlite3", dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw.SetMaxOpenConns(1)
+	db := sqlx.NewDb(raw, "sqlite3")
+	statements := []string{
+		`CREATE TABLE IF NOT EXISTS tasks (id TEXT PRIMARY KEY, archived_at TIMESTAMP, updated_at TIMESTAMP)`,
+		`CREATE TABLE IF NOT EXISTS task_sessions (id TEXT PRIMARY KEY, task_id TEXT NOT NULL, queue_incarnation_id TEXT NOT NULL)`,
+	}
+	for _, statement := range statements {
+		if _, err := db.Exec(statement); err != nil {
+			_ = db.Close()
+			t.Fatal(err)
+		}
+	}
+	source := messagequeue.QueueSessionIdentity{TaskID: "task-1", SessionID: "session-old", SessionIncarnationID: "incarnation-old"}
+	destination := messagequeue.QueueSessionIdentity{TaskID: source.TaskID, SessionID: "session-new", SessionIncarnationID: "incarnation-new"}
+	if seed {
+		if _, err := db.Exec(`INSERT OR IGNORE INTO tasks (id, updated_at) VALUES (?, CURRENT_TIMESTAMP)`, source.TaskID); err != nil {
+			_ = db.Close()
+			t.Fatal(err)
+		}
+		for _, identity := range []messagequeue.QueueSessionIdentity{source, destination} {
+			if _, err := db.Exec(`
+				INSERT INTO task_sessions (id, task_id, queue_incarnation_id) VALUES (?, ?, ?)
+				ON CONFLICT(id) DO UPDATE SET task_id = excluded.task_id, queue_incarnation_id = excluded.queue_incarnation_id
+			`, identity.SessionID, identity.TaskID, identity.SessionIncarnationID); err != nil {
+				_ = db.Close()
+				t.Fatal(err)
+			}
+		}
+	}
+	repo, err := messagequeue.NewSQLiteRepository(db, db)
+	if err != nil {
+		_ = db.Close()
+		t.Fatal(err)
+	}
+	return messagequeue.NewService(repo, messagequeue.DefaultMaxPerSession, testLogger()), db, source, destination
+}
+
 type transientAcceptedMarkerRepository struct {
 	messagequeue.Repository
 	pending interface {

docs/plans/send-now-claim-identity/plan.md

@@ -0,0 +1,34 @@
+---
+created: 2026-09-21
+status: complete
+requirements:
+  - REQ-UI-MESSAGE-QUEUE-SEND-NOW-001
+system_design:
+  - ../../specs/ui/system-design/message-queue-send-now.md
+legacy_specs: []
+---
+
+# Implementation Plan: Preserve Send Now claim identity across session handoff
+
+Session handoff must carry pending Send Now claims to the successor session
+without carrying the retired session's incarnation or operation generation.
+This corrective package records the narrow persistence repair and its restart
+reconciliation proof. The existing Send Now requirement and system design
+remain authoritative for queue ownership, FIFO ordering, and fencing.
+
+## Scope
+
+- Rebind accepted and unaccepted pending claims to the live destination
+  session identity during both direct and durable owned transfers.
+- Preserve claim IDs, accepted state, source ordering, dispatch metadata,
+  transaction rollback, and stale source fencing.
+- Prove recovery after the source session is removed and the process restarts.
+
+## Out of scope
+
+Queue feature behavior, UI changes, provider routing, schema changes, incident
+database repair, deployment, and backup cleanup.
+
+## Delivery
+
+See [Task 01: Rebind transferred Send Now claims](task-01-rebind-claims.md).

docs/plans/send-now-claim-identity/task-01-rebind-claims.md

@@ -0,0 +1,56 @@
+---
+id: "01-rebind-claims"
+title: "Rebind transferred Send Now claims"
+status: complete
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-UI-MESSAGE-QUEUE-SEND-NOW-001
+acceptance_criteria:
+  - AC-UI-MESSAGE-QUEUE-SEND-NOW-001.8
+  - AC-UI-MESSAGE-QUEUE-SEND-NOW-001.10
+system_design:
+  - ../../specs/ui/system-design/message-queue-send-now.md
+---
+
+# Task 01: Rebind transferred Send Now claims
+
+## Summary
+
+When a session handoff moves a pending Send Now claim, persist the successor's
+full queue session identity and operation generation. This keeps restart
+reconciliation usable while retaining exact claim fencing.
+
+## Acceptance
+
+- Accepted and unaccepted claims transferred through direct and durable owned
+  paths use the destination session incarnation and generation.
+- Claim IDs, accepted state, source rows, dispatch metadata, FIFO order, and
+  rollback behavior remain unchanged.
+- Requests and settlements using the retired source identity remain fenced.
+- Legacy transfer fallback fails closed when it encounters an identity-aware
+  claim instead of persisting the retired identity.
+- Reconciliation after source-session removal and process restart can settle
+  both claim states under the successor identity.
+
+## Verification
+
+```bash
+(cd apps/backend && go test -v -run '^(TestSQLiteTransferSessionIdentitiesRebindsPendingSendNowClaim|TestTransferredIdentityAwareSendNowClaimReconcilesAfterProcessRestart)$' ./internal/orchestrator/messagequeue ./internal/orchestrator)
+(cd apps/backend && go test -race ./internal/orchestrator/messagequeue)
+(cd apps/backend && go test ./internal/orchestrator/messagequeue ./internal/orchestrator)
+```
+
+## Results
+
+The direct and durable transfer paths now pass the live destination identity
+into the atomic transfer transaction, which validates it before rebinding the
+claim and derives the claim generation from the destination session generation.
+Durable transfer preparation also includes attachments held only by a pending
+Send Now claim after the source queue row is removed. The legacy transfer
+fallback rejects an identity-aware claim rather than retaining the retired
+identity, while identity-free legacy claims keep their existing behavior.
+Accepted and unaccepted transfer, restart reconciliation, source fencing,
+fallback rejection, claim attachment preparation, and race coverage pass. No
+UI or public documentation change is required.