package automation

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"reflect"
	"sort"
	"strings"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/persistence"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

// connectorPushTxExtractor is a narrow interface for obtaining the ambient
// *sql.Tx from a context (mirrors the TransactionManager pattern used by
// SQLApprovalActionReplayStore and the records package). InsertAuthoredEvidence
// uses it to execute on the caller's transaction when one is present.
type connectorPushTxExtractor interface {
	ExtractTx(ctx context.Context) interface{}
}

type SQLConnectorPushDurableStore struct {
	db          *sql.DB
	txExtractor connectorPushTxExtractor // optional; nil → always use s.db directly
	// adoptAfterInsertTestHook is a narrow test seam (same idiom as the *ForTest
	// seams in scheduler_service.go): nil in production with zero behavior. When
	// set, AdoptFinalizedLink invokes it after the finalized-link INSERT and
	// before the in-transaction evidence consume, so the #3102 §6 T4 co-location
	// artifact can drive the REAL read path through the between-commits window.
	adoptAfterInsertTestHook func()
	// afterLinkKeyLockTestHook is nil in production. Deterministic TiDB tests use
	// it to hold the canonical LinkKey lock before AttemptOccupancyKey is read or
	// written; every create/adopt/transfer authority path keeps that same order.
	afterLinkKeyLockTestHook func(linkKey string)
}

func NewSQLConnectorPushDurableStore(db *sql.DB) *SQLConnectorPushDurableStore {
	return &SQLConnectorPushDurableStore{db: db}
}

// NewSQLConnectorPushDurableStoreWithTxExtractor constructs a store that can
// execute inbound evidence transitions and strict remote-identity finalization
// on an ambient transaction supplied via ctx. Callers that do not need
// ambient-tx support should use
// NewSQLConnectorPushDurableStore.
func NewSQLConnectorPushDurableStoreWithTxExtractor(db *sql.DB, txExtractor connectorPushTxExtractor) *SQLConnectorPushDurableStore {
	return &SQLConnectorPushDurableStore{db: db, txExtractor: txExtractor}
}

// connectorPushAmbientExecer returns the ambient *sql.Tx from ctx (if the
// store was configured with a txExtractor and an active transaction is
// present), or falls back to the store's *sql.DB. Used by
// InsertAuthoredEvidence so that a future inbound-flow caller can write
// evidence atomically with the record.
func (s *SQLConnectorPushDurableStore) connectorPushAmbientExecer(ctx context.Context) connectorPushExecer {
	if s.txExtractor != nil {
		if tx, ok := s.txExtractor.ExtractTx(ctx).(*sql.Tx); ok && tx != nil {
			return tx
		}
	}
	return s.db
}

// connectorPushAmbientQuerier mirrors connectorPushAmbientExecer for the read
// side: it returns the ambient *sql.Tx from ctx (when the store has a
// txExtractor and ctx carries an active transaction) so a read observes the
// caller's own uncommitted writes, or falls back to the store's *sql.DB. Used
// by FindActiveInboundRemoteIDEvidenceInTx for the inbound register action's
// read-after-write idempotency check.
func (s *SQLConnectorPushDurableStore) connectorPushAmbientQuerier(ctx context.Context) connectorPushQuerier {
	if s.txExtractor != nil {
		if tx, ok := s.txExtractor.ExtractTx(ctx).(*sql.Tx); ok && tx != nil {
			return tx
		}
	}
	return s.db
}

const connectorPushLinkSelectPrefix = `
SELECT Id, LinkKey, Status, RemoteId, ConnectorOwnerKey, ActionSlotId, ActionType,
       SourceObject, SourceRecordId, DestinationSlotId, RequesterOwnerPack, ProfileOwnerPack,
       ConnectorNamespace, FlowDeveloperName, FlowActionName, TargetObject, ProfileName,
       CreatedPackInstallId, COALESCE(LastValidatedPackInstallId, ''), CredentialRef, EndpointClass, RetryPolicy, ProfileContractHash, CreatePathTemplate, UpdatePathTemplate,
       ResponseIdPath, COALESCE(RemoteIdField, ''), COALESCE(RemoteIdProvenanceField, ''),
       MappingFingerprint, BodyFingerprint
  FROM _System_ConnectorRemoteLink`

const connectorPushRecoveryHistorySelectPrefix = `
SELECT Id, LinkKey, Status, RemoteId, ConnectorOwnerKey, ActionSlotId, ActionType,
       SourceObject, SourceRecordId, DestinationSlotId, RequesterOwnerPack, ProfileOwnerPack,
       ConnectorNamespace, FlowDeveloperName, FlowActionName, TargetObject, ProfileName,
       CreatedPackInstallId, COALESCE(LastValidatedPackInstallId, ''), CredentialRef, EndpointClass, RetryPolicy, ProfileContractHash, CreatePathTemplate, UpdatePathTemplate,
       ResponseIdPath, COALESCE(RemoteIdField, ''), COALESCE(RemoteIdProvenanceField, ''),
       MappingFingerprint, BodyFingerprint, COALESCE(RemoteLocatorHash, ''),
       COALESCE(RemoteLocatorActiveKey, ''), COALESCE(CompatibilityMirrorStatus, '')
  FROM _System_ConnectorRemoteLink`

const connectorPushAttemptSelectPrefix = `
SELECT Id, LinkKey, Status, DispatchLane, AttemptOrdinal, COALESCE(RemoteId, ''), ConnectorOwnerKey,
       ActionSlotId, ActionType, SourceObject, SourceRecordId, DestinationSlotId,
       RequesterOwnerPack, ProfileOwnerPack, ConnectorNamespace, FlowDeveloperName,
       FlowActionName, TargetObject, ProfileName, CreatedPackInstallId, CredentialRef,
       EndpointClass, RetryPolicy, ProfileContractHash, CreatePathTemplate, UpdatePathTemplate, ResponseIdPath,
       COALESCE(RemoteIdField, ''), COALESCE(RemoteIdProvenanceField, ''),
       COALESCE(RemoteIdMirrorExpectedValue, ''), COALESCE(RemoteIdMirrorCurrentValue, ''),
       COALESCE(RemoteIdProvenanceMirrorExpectedValue, ''), COALESCE(RemoteIdProvenanceMirrorCurrentValue, ''),
	       MappingFingerprint, BodyFingerprint, COALESCE(CreateConflictResolutionFingerprint, ''),
	       COALESCE(CreateConflictPatchIntentJSON, 'null'), COALESCE(PatchDispatchIntentJSON, 'null'),
	       OriginatingUserId, COALESCE(OriginatingUserSessionJSON, '{}'), COALESCE(AllowedSourceFieldsJSON, '[]'),
	       COALESCE(AssociationSourceRecordIdsJSON, '[]'), RequestIdempotencyKey,
       RequestMethod, RequestRelativePath, COALESCE(RequestHeaders, '{}'),
       COALESCE(RequestBodyBase64, ''), COALESCE(RequestBodyHash, ''),
       COALESCE(ResponseStatusCode, 0), COALESCE(ResponseHeaders, '{}'),
       COALESCE(ResponseBodyBase64, ''), COALESCE(ResponseBodyHash, ''),
       RetryCount, COALESCE(ClaimedBy, ''), COALESCE(ClaimToken, ''), ClaimedAt, LeaseExpiresAt, RemoteIOStartedAt
  FROM _System_ConnectorPushAttempt`

func (s *SQLConnectorPushDurableStore) FindLink(ctx context.Context, linkKey string) (*ConnectorPushRemoteLink, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	row := s.db.QueryRowContext(ctx, connectorPushLinkSelectPrefix+`
 WHERE LinkKey = ?
 LIMIT 1`, linkKey)
	link, err := scanConnectorPushLink(row)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return link, err
}

func (s *SQLConnectorPushDurableStore) FindFinalizedLink(ctx context.Context, identity ConnectorPushDurableIdentity) (*ConnectorPushRemoteLink, error) {
	linkKey, err := identity.LinkKey()
	if err != nil {
		return nil, err
	}
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	row := s.db.QueryRowContext(ctx, connectorPushLinkSelectPrefix+`
 WHERE LinkKey = ? AND Status = ?
 LIMIT 1`, linkKey, ConnectorPushLinkStatusFinalized)
	link, err := scanConnectorPushLink(row)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return link, err
}

func (s *SQLConnectorPushDurableStore) ResolveFinalizedRemoteLocator(ctx context.Context, query ConnectorRemoteLocatorQuery) (*ConnectorRemoteIdentityAuthority, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	query.ConnectorOwnerKey = strings.TrimSpace(query.ConnectorOwnerKey)
	query.DestinationSlotID = strings.TrimSpace(query.DestinationSlotID)
	query.RemoteID = strings.TrimSpace(query.RemoteID)
	if query.ConnectorOwnerKey == "" || query.DestinationSlotID == "" || query.RemoteID == "" {
		return nil, fmt.Errorf("connector remote locator query requires connector owner, destination slot, and remote id")
	}
	locatorHash := ConnectorPushRemoteLocatorHash(ConnectorPushDurableIdentity{
		ConnectorOwnerKey: query.ConnectorOwnerKey,
		DestinationSlotID: query.DestinationSlotID,
	}, query.RemoteID)
	rows, err := s.connectorPushAmbientQuerier(ctx).QueryContext(ctx, `
SELECT Id, LinkKey, Status, RemoteId, RemoteLocatorHash,
       ConnectorOwnerKey, ActionSlotId, ActionType, SourceObject, SourceRecordId,
       DestinationSlotId, COALESCE(RemoteIdField, ''), COALESCE(CompatibilityMirrorStatus, '')
  FROM _System_ConnectorRemoteLink
 WHERE RemoteLocatorActiveKey = ?
 LIMIT 2`,
		locatorHash,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var authority *ConnectorRemoteIdentityAuthority
	for rows.Next() {
		if authority != nil {
			return nil, fmt.Errorf("%w: remote locator %s has multiple active owners", ErrConnectorPushDurableConflict, locatorHash)
		}
		candidate := &ConnectorRemoteIdentityAuthority{}
		if err := rows.Scan(
			&candidate.LinkID, &candidate.LinkKey, &candidate.LinkStatus,
			&candidate.RemoteID, &candidate.RemoteLocatorHash,
			&candidate.Identity.ConnectorOwnerKey, &candidate.Identity.ActionSlotID,
			&candidate.Identity.ActionType, &candidate.Identity.SourceObject,
			&candidate.Identity.SourceRecordID, &candidate.Identity.DestinationSlotID,
			&candidate.RemoteIDField, &candidate.CompatibilityMirrorStatus,
		); err != nil {
			return nil, err
		}
		if err := validateConnectorRemoteIdentityAuthority(query, locatorHash, candidate); err != nil {
			return nil, err
		}
		authority = candidate
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return authority, nil
}

func validateConnectorRemoteIdentityAuthority(query ConnectorRemoteLocatorQuery, locatorHash string, authority *ConnectorRemoteIdentityAuthority) error {
	if authority == nil {
		return fmt.Errorf("%w: remote locator authority row is nil", ErrConnectorPushDurableConflict)
	}
	if authority.LinkStatus != ConnectorPushLinkStatusFinalized &&
		authority.LinkStatus != ConnectorPushLinkStatusSuspendedDrift &&
		authority.LinkStatus != ConnectorPushLinkStatusOwnerDeletedPendingRecovery {
		return fmt.Errorf("%w: remote locator has non-authoritative status %q", ErrConnectorPushDurableConflict, authority.LinkStatus)
	}
	if strings.TrimSpace(authority.LinkID) == "" ||
		strings.TrimSpace(authority.Identity.ActionSlotID) == "" ||
		strings.TrimSpace(authority.Identity.ActionType) == "" ||
		strings.TrimSpace(authority.Identity.SourceObject) == "" ||
		strings.TrimSpace(authority.Identity.SourceRecordID) == "" {
		return fmt.Errorf("%w: remote locator authority tuple is incomplete", ErrConnectorPushDurableConflict)
	}
	if authority.Identity.ConnectorOwnerKey != query.ConnectorOwnerKey ||
		authority.Identity.DestinationSlotID != query.DestinationSlotID ||
		strings.TrimSpace(authority.RemoteID) != query.RemoteID ||
		authority.RemoteLocatorHash != locatorHash ||
		ConnectorPushRemoteLocatorHash(authority.Identity, authority.RemoteID) != locatorHash {
		return fmt.Errorf("%w: remote locator authority tuple does not match its active key", ErrConnectorPushDurableConflict)
	}
	linkKey, err := authority.Identity.LinkKey()
	if err != nil || linkKey != authority.LinkKey {
		return fmt.Errorf("%w: remote locator authority link key does not match its identity tuple", ErrConnectorPushDurableConflict)
	}
	return nil
}

// QuarantineFinalizedRemoteIdentitiesForDeletedOwnerInAmbientTx moves every
// finalized identity owned by the deleted record into a non-authoritative
// recovery state. It deliberately preserves RemoteLocatorActiveKey: the same
// remote identity cannot be reclaimed by an attribute fallback while recovery
// is pending. The records engine owns the transaction; this method never opens
// one or performs external I/O.
func (s *SQLConnectorPushDurableStore) QuarantineFinalizedRemoteIdentitiesForDeletedOwnerInAmbientTx(ctx context.Context, objectName, recordID string) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector push durable store is not wired")
	}
	objectName = strings.TrimSpace(objectName)
	recordID = strings.TrimSpace(recordID)
	if objectName == "" || recordID == "" {
		return fmt.Errorf("deleted connector remote identity owner requires object name and record id")
	}
	if s.txExtractor == nil {
		return fmt.Errorf("deleted connector remote identity owner quarantine requires a live ambient transaction")
	}
	tx, ok := s.txExtractor.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return fmt.Errorf("deleted connector remote identity owner quarantine requires a live ambient transaction")
	}
	if err := markConnectorRemoteIdentityOwnerDeletedFenceTx(ctx, tx, objectName, recordID); err != nil {
		return err
	}
	if err := persistence.CheckRuntimeOutboundOwnerQuiescentTx(ctx, tx, objectName, recordID); err != nil {
		return fmt.Errorf("%w: %w", ErrConnectorPushDurableConflict, err)
	}
	if err := s.lockConnectorPushDeletedOwnerStateTx(ctx, tx, objectName, recordID); err != nil {
		return err
	}
	if err := s.rejectConnectorPushRemoteIOInFlightForDeletedOwnerTx(ctx, tx, objectName, recordID); err != nil {
		return err
	}
	if err := s.rejectConnectorIntegrationIOForDeletedOwnerTx(ctx, tx, objectName, recordID); err != nil {
		return err
	}
	_, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET Status = ?, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE SourceObject = ? AND SourceRecordId = ?
   AND Status = ?`,
		ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
		objectName,
		recordID,
		ConnectorPushLinkStatusFinalized,
	)
	if err != nil {
		return err
	}
	_, err = tx.ExecContext(ctx, `
UPDATE _System_ConnectorPushAttempt
   SET Status = ?, ClaimedBy = NULL, ClaimToken = NULL,
       ClaimedAt = NULL, LeaseExpiresAt = NULL,
       LastError = ?, LastErrorAt = CURRENT_TIMESTAMP,
       LastModifiedDate = CURRENT_TIMESTAMP
 WHERE SourceObject = ? AND SourceRecordId = ?
   AND RequestMethod = ?
   AND Status IN (?, ?, ?, ?, ?)`,
		ConnectorPushAttemptStatusAmbiguousRemoteEffect,
		"connector push owner deleted before durable remote identity finalization",
		objectName,
		recordID,
		http.MethodPost,
		ConnectorPushAttemptStatusPending,
		ConnectorPushAttemptStatusProcessing,
		ConnectorPushAttemptStatusPostSent,
		ConnectorPushAttemptStatusWritebackCommitPending,
		ConnectorPushAttemptStatusWritebackPending,
	)
	return err
}

// ClearConnectorRemoteIdentityOwnerDeletedFenceForRestoredOwnerInAmbientTx
// clears only the deleted-owner admission fence after the records engine has
// restored its owner. It deliberately leaves any
// owner_deleted_pending_recovery links quarantined: recovery or retirement
// remains the sole path that changes their durable authority state.
func (s *SQLConnectorPushDurableStore) ClearConnectorRemoteIdentityOwnerDeletedFenceForRestoredOwnerInAmbientTx(ctx context.Context, objectName, recordID string) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector push durable store is not wired")
	}
	objectName = strings.TrimSpace(objectName)
	recordID = strings.TrimSpace(recordID)
	if objectName == "" || recordID == "" {
		return fmt.Errorf("restored connector remote identity owner requires object name and record id")
	}
	if s.txExtractor == nil {
		return fmt.Errorf("restored connector remote identity owner fence clear requires a live ambient transaction")
	}
	tx, ok := s.txExtractor.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return fmt.Errorf("restored connector remote identity owner fence clear requires a live ambient transaction")
	}
	return clearConnectorRemoteIdentityOwnerDeletedFenceTx(ctx, tx, objectName, recordID)
}

// lockConnectorPushDeletedOwnerStateTx serializes owner deletion with every
// outbound POST that could still publish a durable remote identity. The shared
// Lock links before their historical attempt occupancy so owner deletion and
// identity recovery use the same canonical ordering.
func (s *SQLConnectorPushDurableStore) lockConnectorPushDeletedOwnerStateTx(
	ctx context.Context,
	tx *sql.Tx,
	objectName, recordID string,
) error {
	linkKeys := make(map[string]struct{})
	collect := func(query string, args ...any) error {
		rows, err := tx.QueryContext(ctx, query, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var linkKey string
			if err := rows.Scan(&linkKey); err != nil {
				return err
			}
			if linkKey = strings.TrimSpace(linkKey); linkKey != "" {
				linkKeys[linkKey] = struct{}{}
			}
		}
		return rows.Err()
	}
	if err := collect(`
SELECT LinkKey
  FROM _System_ConnectorRemoteLink
 WHERE SourceObject = ? AND SourceRecordId = ?
   AND Status = ?`, objectName, recordID, ConnectorPushLinkStatusFinalized); err != nil {
		return err
	}
	finalizedLinkKeys := make(map[string]struct{}, len(linkKeys))
	for key := range linkKeys {
		finalizedLinkKeys[key] = struct{}{}
	}
	if err := collect(`
SELECT LinkKey
  FROM _System_ConnectorPushAttempt
 WHERE SourceObject = ? AND SourceRecordId = ?
   AND RequestMethod = ?
   AND Status IN (?, ?, ?, ?, ?)`,
		objectName,
		recordID,
		http.MethodPost,
		ConnectorPushAttemptStatusPending,
		ConnectorPushAttemptStatusProcessing,
		ConnectorPushAttemptStatusPostSent,
		ConnectorPushAttemptStatusWritebackCommitPending,
		ConnectorPushAttemptStatusWritebackPending,
	); err != nil {
		return err
	}
	keys := make([]string, 0, len(linkKeys))
	for linkKey := range linkKeys {
		keys = append(keys, linkKey)
	}
	sort.Strings(keys)
	for _, linkKey := range keys {
		state, err := s.lockConnectorPushLinkThenAttemptOccupancyTx(ctx, tx, linkKey)
		if err != nil {
			return err
		}
		if _, wasFinalized := finalizedLinkKeys[linkKey]; wasFinalized &&
			(state.Link.LinkID == "" || state.Link.LinkStatus != ConnectorPushLinkStatusFinalized ||
				!strings.EqualFold(state.Link.Identity.SourceObject, objectName) || state.Link.Identity.SourceRecordID != recordID) {
			return fmt.Errorf("%w: deleted owner link changed before authoritative lock", ErrConnectorPushDurableConflict)
		}
	}
	return nil
}

// rejectConnectorPushRemoteIOInFlightForDeletedOwnerTx is called only after
// the owner fence and every affected LinkKey -> AttemptOccupancyKey have been
// locked. A claim-owned checkpoint is the durable authorization for external
// I/O, so deleting its owner must fail rather than race the callout or erase
// its ability to record the result.
func (s *SQLConnectorPushDurableStore) rejectConnectorPushRemoteIOInFlightForDeletedOwnerTx(
	ctx context.Context,
	tx *sql.Tx,
	objectName, recordID string,
) error {
	var attemptID string
	err := tx.QueryRowContext(ctx, `
SELECT Id
  FROM _System_ConnectorPushAttempt
 WHERE SourceObject = ? AND SourceRecordId = ?
   AND ClaimToken IS NOT NULL AND RemoteIOStartedAt IS NOT NULL
   AND Status IN (?, ?)
 ORDER BY LinkKey, Id
 LIMIT 1
 FOR UPDATE`,
		objectName,
		recordID,
		ConnectorPushAttemptStatusProcessing,
		ConnectorPushAttemptStatusPostSent,
	).Scan(&attemptID)
	if errors.Is(err, sql.ErrNoRows) {
		return nil
	}
	if err != nil {
		return err
	}
	return fmt.Errorf("%w: connector push attempt %s has claim-owned remote I/O in progress", ErrConnectorPushDurableConflict, attemptID)
}

// The legacy link is also the dispatch fence for Integration's V1 bridge. Read
// its complete identity under the existing Link lock before checking Runtime
// receipts with NOWAIT, so a concurrent result commit produces a retryable conflict.
func (s *SQLConnectorPushDurableStore) rejectConnectorIntegrationIOForDeletedOwnerTx(
	ctx context.Context, tx *sql.Tx, objectName, recordID string,
) error {
	rows, err := tx.QueryContext(ctx, `
SELECT LinkKey, ConnectorOwnerKey, ActionSlotId, ActionType, SourceObject,
       SourceRecordId, DestinationSlotId, ConnectorNamespace, RemoteId,
       COALESCE(RemoteLocatorHash, ''), COALESCE(RemoteLocatorActiveKey, ''),
       COALESCE(FlowDeveloperName, ''), COALESCE(FlowActionName, '')
  FROM _System_ConnectorRemoteLink
 WHERE SourceObject = ? AND SourceRecordId = ? AND Status = ?
 ORDER BY LinkKey
 FOR UPDATE`, objectName, recordID, ConnectorPushLinkStatusFinalized)
	if err != nil {
		return err
	}
	type lockedOwnerLink struct {
		key, namespace, remoteID, locator, activeLocator, flow, action string
		identity                                                       ConnectorPushDurableIdentity
	}
	var links []lockedOwnerLink
	for rows.Next() {
		var link lockedOwnerLink
		if err := rows.Scan(&link.key, &link.identity.ConnectorOwnerKey, &link.identity.ActionSlotID,
			&link.identity.ActionType, &link.identity.SourceObject, &link.identity.SourceRecordID,
			&link.identity.DestinationSlotID, &link.namespace, &link.remoteID, &link.locator,
			&link.activeLocator, &link.flow, &link.action); err != nil {
			_ = rows.Close()
			return err
		}
		links = append(links, link)
	}
	if err := rows.Err(); err != nil {
		_ = rows.Close()
		return err
	}
	if err := rows.Close(); err != nil {
		return err
	}
	for _, link := range links {
		legacyKey, err := link.identity.LinkKey()
		if err != nil || !strings.EqualFold(link.identity.SourceObject, objectName) || link.identity.SourceRecordID != recordID ||
			link.remoteID == "" || link.remoteID != strings.TrimSpace(link.remoteID) {
			return fmt.Errorf("%w: deleted owner link has noncanonical identity", ErrConnectorPushDurableConflict)
		}
		scope, err := resolveConnectorIntegrationOutboundScope(ctx, connectorInboundRegisterInstalledBinding{
			Identity: link.identity, FlowDeveloperName: link.flow, FlowActionName: link.action,
		})
		if err != nil {
			return err
		}
		var locator string
		switch {
		case link.key == legacyKey:
			locator = ConnectorPushRemoteLocatorHash(link.identity, link.remoteID)
			if err := s.validateConnectorIntegrationV1BridgeQuiescent(ctx, link.identity, link.remoteID, scope); err != nil {
				return err
			}
		case strings.HasPrefix(link.key, connectorIntegrationLinkV2Prefix):
			if link.namespace != link.identity.ConnectorOwnerKey {
				return fmt.Errorf("%w: deleted owner integration link has a mismatched namespace", ErrConnectorPushDurableConflict)
			}
			locator, err = s.validateConnectorIntegrationOutboundSourceLink(ctx, link.key, link.identity, link.remoteID, scope)
			if err != nil {
				return err
			}
		default:
			return fmt.Errorf("%w: deleted owner link has a mismatched key", ErrConnectorPushDurableConflict)
		}
		if link.locator != locator || link.activeLocator != locator {
			return fmt.Errorf("%w: deleted owner link has a mismatched active locator", ErrConnectorPushDurableConflict)
		}
	}
	return nil
}

func (s *SQLConnectorPushDurableStore) RecordCompatibilityMirrorReadback(ctx context.Context, linkKey, status string) error {
	const transitionName = "record connector compatibility mirror readback"
	linkKey = strings.TrimSpace(linkKey)
	status = strings.TrimSpace(status)
	if linkKey == "" {
		return fmt.Errorf("connector push durable link key is required for compatibility mirror readback")
	}
	if status != ConnectorPushCompatibilityMirrorSynced && status != ConnectorPushCompatibilityMirrorDrifted {
		return fmt.Errorf("unsupported compatibility mirror readback status %q", status)
	}
	if s == nil || s.db == nil {
		return fmt.Errorf("connector push durable store is not wired")
	}
	result, err := s.connectorPushAmbientExecer(ctx).ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET CompatibilityMirrorStatus = ?, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE LinkKey = ? AND Status IN (?, ?)`,
		status,
		linkKey,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
	)
	if err != nil {
		return err
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("%s rows affected: %w", transitionName, err)
	}
	if rows == 1 {
		return nil
	}
	if rows != 0 {
		return fmt.Errorf("%s updated %d rows, want 1", transitionName, rows)
	}

	// TiDB reports zero affected rows when both the mirror status and the
	// second-resolution timestamp already hold the requested values. Accept
	// that idempotent outcome only after an exact readback in the same ambient
	// transaction; missing, inactive, or differently-valued links still fail.
	var currentStatus string
	err = s.connectorPushAmbientQuerier(ctx).QueryRowContext(ctx, `
SELECT COALESCE(CompatibilityMirrorStatus, '')
  FROM _System_ConnectorRemoteLink
 WHERE LinkKey = ? AND Status IN (?, ?)
 LIMIT 1
 FOR UPDATE`,
		linkKey,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
	).Scan(&currentStatus)
	if errors.Is(err, sql.ErrNoRows) {
		return fmt.Errorf("%s updated 0 rows and authoritative link is missing or inactive", transitionName)
	}
	if err != nil {
		return fmt.Errorf("%s readback: %w", transitionName, err)
	}
	if currentStatus != status {
		return fmt.Errorf("%s updated 0 rows and read back status %q, want %q", transitionName, currentStatus, status)
	}
	return nil
}

func (s *SQLConnectorPushDurableStore) FindBlockingAttempt(ctx context.Context, linkKey string) (*ConnectorPushAttempt, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	row := s.db.QueryRowContext(ctx, connectorPushAttemptSelectPrefix+`
 WHERE AttemptOccupancyKey = ?
 LIMIT 1`, linkKey)
	attempt, err := scanConnectorPushAttempt(row)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return attempt, err
}

func (s *SQLConnectorPushDurableStore) FindStateBySource(ctx context.Context, q ConnectorPushDurableSourceQuery) (*ConnectorPushRemoteLink, *ConnectorPushAttempt, error) {
	if s == nil || s.db == nil {
		return nil, nil, fmt.Errorf("connector push durable store is not wired")
	}
	q = normalizeConnectorPushDurableSourceQuery(q)
	if err := validateConnectorPushDurableSourceQuery(q); err != nil {
		return nil, nil, err
	}
	attempt, err := scanConnectorPushAttempt(s.db.QueryRowContext(ctx, connectorPushAttemptSelectPrefix+`
 WHERE ConnectorOwnerKey = ? AND ActionSlotId = ? AND ActionType = ? AND SourceObject = ? AND SourceRecordId = ?
   AND RequestMethod = ? AND Status NOT IN (?, ?)
 ORDER BY LastModifiedDate DESC, Id DESC
 LIMIT 1`,
		q.ConnectorOwnerKey, q.ActionSlotID, q.ActionType, q.SourceObject, q.SourceRecordID,
		http.MethodPost,
		ConnectorPushAttemptStatusFinalized, ConnectorPushAttemptStatusSuperseded,
	))
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return nil, nil, err
	}
	link, linkErr := scanConnectorPushLink(s.db.QueryRowContext(ctx, connectorPushLinkSelectPrefix+`
 WHERE ConnectorOwnerKey = ? AND ActionSlotId = ? AND ActionType = ? AND SourceObject = ? AND SourceRecordId = ?
   AND Status IN (?, ?, ?)
 ORDER BY LastModifiedDate DESC, Id DESC
 LIMIT 1`,
		q.ConnectorOwnerKey, q.ActionSlotID, q.ActionType, q.SourceObject, q.SourceRecordID,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
		ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
	))
	if linkErr != nil && !errors.Is(linkErr, sql.ErrNoRows) {
		return nil, nil, linkErr
	}
	if errors.Is(err, sql.ErrNoRows) {
		attempt = nil
	}
	if errors.Is(linkErr, sql.ErrNoRows) {
		link = nil
	}
	return link, attempt, nil
}

// FindStateByDestination is the authoritative per-destination point read for
// the runtime create gate. Its two rules are the single source mirrored by the
// diagnostic fan-out FindRecordPushState:
//   - active links are Status IN (ConnectorPushLinkStatusFinalized,
//     ConnectorPushLinkStatusSuspendedDrift,
//     ConnectorPushLinkStatusOwnerDeletedPendingRecovery) — both readers bind
//     the same constants;
//   - a live create attempt is RequestMethod=POST AND Status NOT IN
//     (finalized, superseded) — restated Go-side as
//     connectorPushAttemptIsLiveCreate; change both together or not at all;
//   - the current row among candidates is LastModifiedDate DESC, Id DESC —
//     the same deterministic tiebreaker as linkRowBeats/attemptRowBeats
//     (second-precision DATETIME ties must not be engine-arbitrary).
func (s *SQLConnectorPushDurableStore) FindStateByDestination(ctx context.Context, q ConnectorPushDurableDestinationQuery) (*ConnectorPushRemoteLink, *ConnectorPushAttempt, error) {
	if s == nil || s.db == nil {
		return nil, nil, fmt.Errorf("connector push durable store is not wired")
	}
	q = normalizeConnectorPushDurableDestinationQuery(q)
	if err := validateConnectorPushDurableDestinationQuery(q); err != nil {
		return nil, nil, err
	}
	attempt, err := scanConnectorPushAttempt(s.db.QueryRowContext(ctx, connectorPushAttemptSelectPrefix+`
 WHERE ConnectorOwnerKey = ? AND ActionType = ? AND SourceObject = ? AND SourceRecordId = ? AND DestinationSlotId = ?
   AND RequestMethod = ? AND Status NOT IN (?, ?)
 ORDER BY LastModifiedDate DESC, Id DESC
 LIMIT 1`,
		q.ConnectorOwnerKey, q.ActionType, q.SourceObject, q.SourceRecordID, q.DestinationSlotID,
		http.MethodPost,
		ConnectorPushAttemptStatusFinalized, ConnectorPushAttemptStatusSuperseded,
	))
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return nil, nil, err
	}
	link, linkErr := scanConnectorPushLink(s.db.QueryRowContext(ctx, connectorPushLinkSelectPrefix+`
 WHERE ConnectorOwnerKey = ? AND ActionType = ? AND SourceObject = ? AND SourceRecordId = ? AND DestinationSlotId = ?
   AND Status IN (?, ?, ?)
 ORDER BY LastModifiedDate DESC, Id DESC
 LIMIT 1`,
		q.ConnectorOwnerKey, q.ActionType, q.SourceObject, q.SourceRecordID, q.DestinationSlotID,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
		ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
	))
	if linkErr != nil && !errors.Is(linkErr, sql.ErrNoRows) {
		return nil, nil, linkErr
	}
	if errors.Is(err, sql.ErrNoRows) {
		attempt = nil
	}
	if errors.Is(linkErr, sql.ErrNoRows) {
		link = nil
	}
	return link, attempt, nil
}

// FinalizeInboundRemoteIdentityInAmbientTx consumes exact pre-authored evidence
// into durable inbound authority inside the caller's live transaction. It never
// falls back to s.db and never owns commit or rollback; record, evidence, and
// link therefore share one atomic boundary.
func (s *SQLConnectorPushDurableStore) FinalizeInboundRemoteIdentityInAmbientTx(ctx context.Context, finalization ConnectorInboundRemoteIdentityFinalization) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector push durable store is not wired")
	}
	normalized, activeKey, err := normalizeConnectorInboundRemoteIdentityFinalization(finalization)
	if err != nil {
		return err
	}
	if s.txExtractor == nil {
		return fmt.Errorf("connector inbound remote identity finalization requires a live ambient transaction")
	}
	tx, ok := s.txExtractor.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return fmt.Errorf("connector inbound remote identity finalization requires a live ambient transaction")
	}
	return s.finalizeInboundRemoteIdentityTx(ctx, tx, normalized, activeKey)
}

func (s *SQLConnectorPushDurableStore) finalizeInboundRemoteIdentityTx(ctx context.Context, tx *sql.Tx, finalization ConnectorInboundRemoteIdentityFinalization, activeKey string) error {
	adoption := finalization.Adoption
	// Keep the owner fence before the LinkKey -> AttemptOccupancyKey lock order
	// used by deletion quarantine. Inbound authored evidence is authority
	// admission, so it must not create a new link after deletion commits.
	if err := ensureConnectorRemoteIdentityOwnerCreateAllowedTx(ctx, tx, adoption.Identity.SourceObject, adoption.Identity.SourceRecordID); err != nil {
		return err
	}

	lockState, err := s.lockConnectorPushLinkThenAttemptOccupancyTx(ctx, tx, adoption.LinkKey)
	if err != nil {
		return err
	}
	attemptStatus := strings.TrimSpace(lockState.AttemptStatus)
	if attemptStatus != "" && attemptStatus != ConnectorPushAttemptStatusTerminalFailed {
		return fmt.Errorf("%w: create attempt %s already occupies link key", ErrConnectorPushDurableConflict, attemptStatus)
	}

	existing := &lockState.Link
	existingRemoteLocatorActiveKey := lockState.RemoteLocatorActiveKey
	expectedRemoteLocatorActiveKey := ConnectorPushRemoteLocatorHash(adoption.Identity, adoption.RemoteID)
	if strings.TrimSpace(existing.LinkID) != "" {
		if err := validateConnectorRemoteIdentityAuthority(ConnectorRemoteLocatorQuery{
			ConnectorOwnerKey: adoption.Identity.ConnectorOwnerKey,
			DestinationSlotID: adoption.Identity.DestinationSlotID,
			RemoteID:          adoption.RemoteID,
		}, expectedRemoteLocatorActiveKey, existing); err != nil {
			return err
		}
	}
	identicalFinalizedLink := strings.TrimSpace(existing.LinkID) != "" &&
		strings.TrimSpace(existing.LinkStatus) == ConnectorPushLinkStatusFinalized &&
		strings.TrimSpace(existing.RemoteID) == adoption.RemoteID &&
		strings.TrimSpace(existingRemoteLocatorActiveKey) == expectedRemoteLocatorActiveKey
	pendingRecoveryLink := strings.TrimSpace(existing.LinkID) != "" &&
		strings.TrimSpace(existing.LinkStatus) == ConnectorPushLinkStatusOwnerDeletedPendingRecovery &&
		strings.TrimSpace(existing.RemoteID) == adoption.RemoteID &&
		strings.TrimSpace(existingRemoteLocatorActiveKey) == expectedRemoteLocatorActiveKey
	if strings.TrimSpace(existing.LinkID) != "" && !identicalFinalizedLink && !pendingRecoveryLink {
		return fmt.Errorf("%w: remote link %s already owns link key", ErrConnectorPushDurableConflict, strings.TrimSpace(existing.LinkStatus))
	}
	// Historical-attempt recovery is only an authority-creation path. An
	// attempt coexisting with any materialized link is corrupt/conflicting
	// state, not an idempotent retry, and remains fail-closed.
	if attemptStatus != "" && strings.TrimSpace(existing.LinkID) != "" {
		return fmt.Errorf("%w: create attempt %s already occupies link key", ErrConnectorPushDurableConflict, attemptStatus)
	}
	evidence, err := findConnectorInboundEvidenceForUpdate(ctx, tx, activeKey)
	if err != nil {
		return err
	}
	if evidence != nil && !connectorInboundEvidenceMatchesFinalization(*evidence, finalization) {
		return fmt.Errorf("%w: active inbound authorship evidence does not match finalization", ErrConnectorPushDurableConflict)
	}
	if attemptStatus != "" && evidence != nil {
		recovered, err := supersedeExactHistoricalTerminalAttemptTx(ctx, tx, lockState.AttemptID, adoption.LinkKey)
		if err != nil {
			return err
		}
		if !recovered {
			return fmt.Errorf("%w: create attempt %s already occupies link key", ErrConnectorPushDurableConflict, attemptStatus)
		}
	}

	// A committed identical link with no active evidence is the retry state after
	// the original transaction consumed it. Revalidate the caller-named consumed
	// row so a blank, missing, or unrelated evidence ID cannot authorize the
	// idempotent success. Matching leftover evidence is healed by consuming it
	// below; mismatched active evidence already failed closed.
	if (identicalFinalizedLink || pendingRecoveryLink) && evidence == nil {
		consumedEvidence, err := findConnectorInboundEvidenceByIDForUpdate(ctx, tx, finalization.Evidence.ID)
		if err != nil {
			return err
		}
		if consumedEvidence == nil || !connectorInboundConsumedEvidenceMatchesFinalization(*consumedEvidence, finalization) {
			return fmt.Errorf("%w: exact consumed inbound authorship evidence does not match finalization", ErrConnectorPushDurableConflict)
		}
		if pendingRecoveryLink {
			result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
   SET Status = ?, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND Status = ?`,
				ConnectorPushLinkStatusFinalized,
				existing.LinkID,
				ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
			)
			if err != nil {
				return err
			}
			if err := requireConnectorPushOneRow(result, "recover owner-deleted pending remote identity"); err != nil {
				return fmt.Errorf("%w: %v", ErrConnectorPushDurableConflict, err)
			}
		}
		return nil
	}

	if evidence == nil {
		return fmt.Errorf("%w: exact active inbound authorship evidence is required", ErrConnectorPushDurableConflict)
	}

	if pendingRecoveryLink {
		return fmt.Errorf("%w: pending owner-deleted recovery requires exact consumed evidence", ErrConnectorPushDurableConflict)
	}
	if !identicalFinalizedLink {
		if err := s.insertAdoptedFinalizedLinkTx(ctx, tx, adoption, finalization.CompatibilityMirrorStatus); err != nil {
			return err
		}
	}

	result, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRemoteIDEvidence
   SET EvidenceStatus = ?, ConsumedByLinkKey = ?, ActiveKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND EvidenceStatus = ? AND ActiveKey = ? AND RemoteId = ?`,
		ConnectorInboundEvidenceStatusConsumed,
		adoption.LinkKey,
		evidence.ID,
		ConnectorInboundEvidenceStatusAuthored,
		activeKey,
		adoption.RemoteID,
	)
	if err != nil {
		return err
	}
	if err := requireConnectorPushOneRow(result, "finalize inbound remote identity evidence"); err != nil {
		return fmt.Errorf("%w: %v", ErrConnectorPushDurableConflict, err)
	}
	return nil
}

const connectorInboundSupersedeExactHistoricalTerminalAttemptSQL = `
UPDATE _System_ConnectorPushAttempt
   SET Status = ?, AttemptOccupancyKey = NULL, SupersededAt = CURRENT_TIMESTAMP,
       LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND LinkKey = ? AND Status = ?
   AND AttemptOccupancyKey = ?
   AND (
       (RequestMethod = ? AND ResponseStatusCode = ?)
       OR (RequestMethod = ? AND RemoteIOStartedAt IS NULL AND ResponseStatusCode IS NULL)
   )
   AND COALESCE(RemoteId, '') = ''
   AND COALESCE(CreateConflictResolutionFingerprint, '') = ''
   AND COALESCE(CAST(CreateConflictPatchIntentJSON AS CHAR), 'null') = 'null'
   AND COALESCE(CAST(PatchDispatchIntentJSON AS CHAR), 'null') = 'null'
   AND COALESCE(ClaimedBy, '') = '' AND COALESCE(ClaimToken, '') = ''
   AND ClaimedAt IS NULL AND LeaseExpiresAt IS NULL
   AND SupersededByAttemptId IS NULL AND SupersededAt IS NULL`

// supersedeExactHistoricalTerminalAttemptTx clears only a historical
// terminal POST occupancy that exact active inbound-authorship evidence can
// replace with current installed-snapshot authority: either the established
// POST/409 conflict shape or a create that never began remote I/O and received
// no response. The caller already owns the LinkKey -> AttemptOccupancyKey
// locks and performs supersession, link insertion, and evidence consumption in
// the same ambient transaction.
func supersedeExactHistoricalTerminalAttemptTx(
	ctx context.Context,
	tx *sql.Tx,
	attemptID string,
	linkKey string,
) (bool, error) {
	result, err := tx.ExecContext(ctx, connectorInboundSupersedeExactHistoricalTerminalAttemptSQL,
		ConnectorPushAttemptStatusSuperseded,
		strings.TrimSpace(attemptID),
		strings.TrimSpace(linkKey),
		ConnectorPushAttemptStatusTerminalFailed,
		strings.TrimSpace(linkKey),
		http.MethodPost,
		http.StatusConflict,
		http.MethodPost,
	)
	if err != nil {
		return false, err
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return false, err
	}
	return rows == 1, nil
}

func findConnectorInboundEvidenceForUpdate(ctx context.Context, tx *sql.Tx, activeKey string) (*ConnectorInboundRemoteIDEvidence, error) {
	var evidence ConnectorInboundRemoteIDEvidence
	err := tx.QueryRowContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus, COALESCE(ConsumedByLinkKey, '')
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE ActiveKey = ?
 LIMIT 1
 FOR UPDATE`, activeKey).Scan(
		&evidence.ID,
		&evidence.ConnectorOwnerKey,
		&evidence.ConnectorNamespace,
		&evidence.SourceObject,
		&evidence.SourceRecordId,
		&evidence.RemoteIDField,
		&evidence.RemoteId,
		&evidence.EvidenceStatus,
		&evidence.ConsumedByLinkKey,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	evidence.ActiveKey = activeKey
	return &evidence, nil
}

func findConnectorInboundEvidenceByIDForUpdate(ctx context.Context, tx *sql.Tx, evidenceID string) (*ConnectorInboundRemoteIDEvidence, error) {
	var evidence ConnectorInboundRemoteIDEvidence
	err := tx.QueryRowContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus, COALESCE(ConsumedByLinkKey, ''),
       COALESCE(ActiveKey, '')
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE Id = ?
 LIMIT 1
 FOR UPDATE`, evidenceID).Scan(
		&evidence.ID,
		&evidence.ConnectorOwnerKey,
		&evidence.ConnectorNamespace,
		&evidence.SourceObject,
		&evidence.SourceRecordId,
		&evidence.RemoteIDField,
		&evidence.RemoteId,
		&evidence.EvidenceStatus,
		&evidence.ConsumedByLinkKey,
		&evidence.ActiveKey,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &evidence, nil
}

func connectorInboundEvidenceMatchesFinalization(evidence ConnectorInboundRemoteIDEvidence, finalization ConnectorInboundRemoteIdentityFinalization) bool {
	expected := finalization.Evidence
	return strings.TrimSpace(evidence.ID) == expected.ID &&
		strings.TrimSpace(evidence.EvidenceStatus) == ConnectorInboundEvidenceStatusAuthored &&
		strings.TrimSpace(evidence.ConsumedByLinkKey) == "" &&
		strings.TrimSpace(evidence.ActiveKey) == expected.ActiveKey &&
		strings.TrimSpace(evidence.ConnectorOwnerKey) == expected.ConnectorOwnerKey &&
		strings.EqualFold(strings.TrimSpace(evidence.ConnectorNamespace), expected.ConnectorNamespace) &&
		strings.TrimSpace(evidence.SourceObject) == expected.SourceObject &&
		strings.TrimSpace(evidence.SourceRecordId) == expected.SourceRecordId &&
		strings.TrimSpace(evidence.RemoteIDField) == expected.RemoteIDField &&
		strings.TrimSpace(evidence.RemoteId) == finalization.Adoption.RemoteID
}

func connectorInboundConsumedEvidenceMatchesFinalization(evidence ConnectorInboundRemoteIDEvidence, finalization ConnectorInboundRemoteIdentityFinalization) bool {
	expected := finalization.Evidence
	return strings.TrimSpace(evidence.ID) == expected.ID &&
		strings.TrimSpace(evidence.EvidenceStatus) == ConnectorInboundEvidenceStatusConsumed &&
		strings.TrimSpace(evidence.ConsumedByLinkKey) == finalization.Adoption.LinkKey &&
		strings.TrimSpace(evidence.ActiveKey) == "" &&
		strings.TrimSpace(evidence.ConnectorOwnerKey) == expected.ConnectorOwnerKey &&
		strings.EqualFold(strings.TrimSpace(evidence.ConnectorNamespace), expected.ConnectorNamespace) &&
		strings.TrimSpace(evidence.SourceObject) == expected.SourceObject &&
		strings.TrimSpace(evidence.SourceRecordId) == expected.SourceRecordId &&
		strings.TrimSpace(evidence.RemoteIDField) == expected.RemoteIDField &&
		strings.TrimSpace(evidence.RemoteId) == finalization.Adoption.RemoteID
}

func (s *SQLConnectorPushDurableStore) insertAdoptedFinalizedLinkTx(
	ctx context.Context,
	tx *sql.Tx,
	adoption ConnectorPushLinkAdoption,
	compatibilityMirrorStatus ConnectorPushCompatibilityMirrorStatus,
) error {
	linkID := utils.GenerateID()
	remoteLocatorHash := ConnectorPushRemoteLocatorHash(adoption.Identity, adoption.RemoteID)
	linkArgs := []any{
		linkID, adoption.LinkKey, ConnectorPushLinkStatusFinalized, adoption.RemoteID,
		remoteLocatorHash, remoteLocatorHash,
	}
	linkArgs = appendConnectorPushIdentitySQLArgs(linkArgs, adoption.Identity)
	linkArgs = appendConnectorPushRemoteLinkSnapshotSQLArgs(linkArgs, adoption.Snapshot)
	linkArgs = append(linkArgs, string(compatibilityMirrorStatus))
	_, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorRemoteLink
  (Id, LinkKey, Status, RemoteId, RemoteLocatorHash, RemoteLocatorActiveKey, ConnectorOwnerKey,
   ActionSlotId, ActionType, SourceObject, SourceRecordId, DestinationSlotId, RequesterOwnerPack,
   ProfileOwnerPack, ConnectorNamespace, FlowDeveloperName, FlowActionName, TargetObject, ProfileName,
   CreatedPackInstallId, LastValidatedPackInstallId, CredentialRef, EndpointClass, RetryPolicy, ProfileContractHash, CreatePathTemplate,
   UpdatePathTemplate, ResponseIdPath, RemoteIdField, RemoteIdProvenanceField, MappingFingerprint, BodyFingerprint,
   CompatibilityMirrorStatus)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		linkArgs...)
	if err != nil {
		if conflictErr := connectorPushDuplicateKeyConflict(err); conflictErr != nil {
			return conflictErr
		}
		return err
	}
	return nil
}

func (s *SQLConnectorPushDurableStore) AdoptFinalizedLink(ctx context.Context, adoption ConnectorPushLinkAdoption) (*ConnectorPushRemoteLink, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	adoption, err := normalizeConnectorPushLinkAdoption(adoption)
	if err != nil {
		return nil, err
	}
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer func() {
		_ = tx.Rollback()
	}()
	if err := lockConnectorRemoteIdentityLaneTx(ctx, tx, adoption.Identity, adoption.RemoteID); err != nil {
		return nil, err
	}
	// Inbound finalization acquires the remote-locator lane before this owner
	// fence. Keep adoption in the same order so concurrent authority writers
	// cannot form a locator-lane -> owner-fence cycle.
	if err := ensureConnectorRemoteIdentityOwnerCreateAllowedTx(ctx, tx, adoption.Identity.SourceObject, adoption.Identity.SourceRecordID); err != nil {
		return nil, err
	}
	if _, err := checkConnectorRemoteIdentityActiveAuthorityTx(ctx, tx, adoption.Identity, adoption.RemoteID, ConnectorPushLinkStatusFinalized); err != nil {
		return nil, err
	}

	// consumeEvidenceTx revalidates and consumes the authored inbound evidence row
	// identified by adoption.EvidenceActiveKey inside the open transaction.
	// When requireMatch is true, a missing or stale row is a hard conflict (fail
	// closed).  When requireMatch is false (idempotent path), only a real DB error
	// aborts the caller.
	consumeEvidenceTx := func(requireMatch bool) error {
		key := strings.TrimSpace(adoption.EvidenceActiveKey)
		if key == "" {
			return nil
		}
		var evID, evStatus, evRemoteID string
		err := tx.QueryRowContext(ctx, `
SELECT Id, EvidenceStatus, RemoteId
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE ActiveKey = ?
 LIMIT 1
 FOR UPDATE`, key).Scan(&evID, &evStatus, &evRemoteID)
		if errors.Is(err, sql.ErrNoRows) {
			if requireMatch {
				return fmt.Errorf("%w: inbound authorship evidence is no longer active for adopt", ErrConnectorPushDurableConflict)
			}
			return nil
		}
		if err != nil {
			return err
		}
		if strings.TrimSpace(evStatus) != ConnectorInboundEvidenceStatusAuthored || strings.TrimSpace(evRemoteID) != adoption.RemoteID {
			if requireMatch {
				return fmt.Errorf("%w: inbound authorship evidence changed before adopt", ErrConnectorPushDurableConflict)
			}
			return nil
		}
		if _, err := tx.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRemoteIDEvidence
   SET EvidenceStatus = ?, ConsumedByLinkKey = ?, ActiveKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND EvidenceStatus = ?`,
			ConnectorInboundEvidenceStatusConsumed, adoption.LinkKey, evID, ConnectorInboundEvidenceStatusAuthored); err != nil {
			return err
		}
		return nil
	}

	lockState, err := s.lockConnectorPushLinkThenAttemptOccupancyTx(ctx, tx, adoption.LinkKey)
	if err != nil {
		return nil, err
	}
	if strings.TrimSpace(lockState.AttemptStatus) != "" {
		return nil, fmt.Errorf("%w: create attempt %s already occupies link key", ErrConnectorPushDurableConflict, strings.TrimSpace(lockState.AttemptStatus))
	}

	existingID := lockState.Link.LinkID
	existingStatus := lockState.Link.LinkStatus
	existingRemoteID := lockState.Link.RemoteID
	if strings.TrimSpace(existingID) != "" {
		if strings.TrimSpace(existingStatus) == ConnectorPushLinkStatusFinalized && strings.TrimSpace(existingRemoteID) == adoption.RemoteID {
			// Best-effort evidence consume on the idempotent path: the link already
			// exists so a missing/stale evidence row is acceptable.
			if err := consumeEvidenceTx(false); err != nil {
				return nil, err
			}
			if err := tx.Commit(); err != nil {
				return nil, err
			}
			return s.FindLink(ctx, adoption.LinkKey)
		}
		return nil, fmt.Errorf("%w: remote link %s already owns link key", ErrConnectorPushDurableConflict, strings.TrimSpace(existingStatus))
	}

	if err := s.insertAdoptedFinalizedLinkTx(ctx, tx, adoption, ConnectorPushCompatibilityMirrorStatus(ConnectorPushCompatibilityMirrorSynced)); err != nil {
		return nil, err
	}
	if s.adoptAfterInsertTestHook != nil {
		// Test-only seam (nil in production): exercises the between-commits window.
		s.adoptAfterInsertTestHook()
	}
	// Require the evidence still authored + matching; if a concurrent re-author
	// changed the evidence, fail closed and rollback undoes the link insert.
	if err := consumeEvidenceTx(true); err != nil {
		return nil, err
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.FindLink(ctx, adoption.LinkKey)
}

// execTransitionAmbient is identical to execTransition except it sources the
// execer from connectorPushAmbientExecer — using the ambient *sql.Tx from ctx
// when one is available (txExtractor configured + ctx carries a live tx), or
// falling back to s.db. Used by evidence retire/consume transitions so that
// callers inside a save-transaction can retire/consume evidence atomically with
// the record write.
func (s *SQLConnectorPushDurableStore) execTransitionAmbient(ctx context.Context, name string, query string, args ...any) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector push durable store is not wired")
	}
	result, err := s.connectorPushAmbientExecer(ctx).ExecContext(ctx, query, args...)
	if err != nil {
		return err
	}
	return requireConnectorPushOneRow(result, name)
}

type connectorPushExecer interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

type connectorPushQuerier interface {
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

type connectorPushLinkAttemptLockState struct {
	Link                   ConnectorRemoteIdentityAuthority
	RemoteLocatorActiveKey string
	AttemptID              string
	AttemptStatus          string
}

// lockConnectorPushLinkThenAttemptOccupancyTx is the single authority-creation
// lock primitive. A caller may already own its remote-locator lane, but once it
// enters LinkKey exclusion it must always lock LinkKey before
// AttemptOccupancyKey. The exact unique keys make absent rows durable point-lock
// intents on TiDB rather than relying on a broad range scan.
func (s *SQLConnectorPushDurableStore) lockConnectorPushLinkThenAttemptOccupancyTx(
	ctx context.Context,
	tx *sql.Tx,
	linkKey string,
) (connectorPushLinkAttemptLockState, error) {
	linkKey = strings.TrimSpace(linkKey)
	if linkKey == "" {
		return connectorPushLinkAttemptLockState{}, fmt.Errorf("connector push link key is required for authority lock")
	}
	state := connectorPushLinkAttemptLockState{}
	err := tx.QueryRowContext(ctx, `
SELECT Id, LinkKey, Status, RemoteId, COALESCE(RemoteLocatorHash, ''),
       COALESCE(RemoteLocatorActiveKey, ''), ConnectorOwnerKey, ActionSlotId,
       ActionType, SourceObject, SourceRecordId, DestinationSlotId
  FROM _System_ConnectorRemoteLink
 WHERE LinkKey = ?
 LIMIT 1
 FOR UPDATE`, linkKey).Scan(
		&state.Link.LinkID,
		&state.Link.LinkKey,
		&state.Link.LinkStatus,
		&state.Link.RemoteID,
		&state.Link.RemoteLocatorHash,
		&state.RemoteLocatorActiveKey,
		&state.Link.Identity.ConnectorOwnerKey,
		&state.Link.Identity.ActionSlotID,
		&state.Link.Identity.ActionType,
		&state.Link.Identity.SourceObject,
		&state.Link.Identity.SourceRecordID,
		&state.Link.Identity.DestinationSlotID,
	)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return connectorPushLinkAttemptLockState{}, err
	}
	if s.afterLinkKeyLockTestHook != nil {
		s.afterLinkKeyLockTestHook(linkKey)
	}
	err = tx.QueryRowContext(ctx, `
SELECT Id, Status
  FROM _System_ConnectorPushAttempt
 WHERE AttemptOccupancyKey = ?
 LIMIT 1
 FOR UPDATE`, linkKey).Scan(&state.AttemptID, &state.AttemptStatus)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return connectorPushLinkAttemptLockState{}, err
	}
	return state, nil
}

func appendConnectorPushIdentitySQLArgs(args []any, identity ConnectorPushDurableIdentity) []any {
	return append(args,
		identity.ConnectorOwnerKey, identity.ActionSlotID, identity.ActionType,
		identity.SourceObject, identity.SourceRecordID, identity.DestinationSlotID,
	)
}

func appendConnectorPushRemoteLinkSnapshotSQLArgs(args []any, snapshot ConnectorPushDurableSnapshot) []any {
	return append(args,
		snapshot.RequesterOwnerPack, snapshot.ProfileOwnerPack, snapshot.ConnectorNamespace,
		snapshot.FlowDeveloperName, snapshot.FlowActionName, snapshot.TargetObject,
		snapshot.ProfileName, snapshot.CreatedPackInstallID, snapshot.CreatedPackInstallID,
		snapshot.CredentialRef, snapshot.EndpointClass, mustConnectorPushJSON(snapshot.RetryPolicy),
		snapshot.ProfileContractHash, snapshot.CreatePathTemplate, snapshot.UpdatePathTemplate,
		snapshot.ResponseIDPath, nullableConnectorPushString(snapshot.RemoteIDField),
		nullableConnectorPushString(snapshot.RemoteIDProvenanceField),
		snapshot.MappingFingerprint, snapshot.BodyFingerprint,
	)
}

func normalizeConnectorPushLinkAdoption(adoption ConnectorPushLinkAdoption) (ConnectorPushLinkAdoption, error) {
	linkKey, err := adoption.Identity.LinkKey()
	if err != nil {
		return adoption, err
	}
	if strings.TrimSpace(adoption.LinkKey) != "" && adoption.LinkKey != linkKey {
		return adoption, fmt.Errorf("connector push durable link key does not match identity")
	}
	remoteID := strings.TrimSpace(adoption.RemoteID)
	if remoteID == "" {
		return adoption, fmt.Errorf("connector push durable remote id is required")
	}
	if adoption.Snapshot.RetryPolicy.MaxAttempts < 1 {
		return adoption, fmt.Errorf("connector push durable retry policy must set MaxAttempts")
	}
	if strings.TrimSpace(adoption.Snapshot.LastValidatedPackInstallID) == "" {
		adoption.Snapshot.LastValidatedPackInstallID = strings.TrimSpace(adoption.Snapshot.CreatedPackInstallID)
	}
	profileHash := ConnectorPushProfileContractHash(adoption.Snapshot)
	if strings.TrimSpace(adoption.Snapshot.ProfileContractHash) != "" && adoption.Snapshot.ProfileContractHash != profileHash {
		return adoption, fmt.Errorf("connector push durable profile contract hash mismatch")
	}
	adoption.Snapshot.ProfileContractHash = profileHash
	adoption.LinkKey = linkKey
	adoption.RemoteID = remoteID
	return adoption, nil
}

func normalizeConnectorInboundRemoteIdentityFinalization(finalization ConnectorInboundRemoteIdentityFinalization) (ConnectorInboundRemoteIdentityFinalization, string, error) {
	compatibilityMirrorStatus, err := normalizeConnectorInboundRecoveryMirrorStatus(finalization.CompatibilityMirrorStatus, true)
	if err != nil {
		return finalization, "", err
	}
	finalization.CompatibilityMirrorStatus = compatibilityMirrorStatus
	adoption, err := normalizeConnectorPushLinkAdoption(finalization.Adoption)
	if err != nil {
		return finalization, "", err
	}
	evidence := finalization.Evidence
	evidence.ConnectorOwnerKey = strings.TrimSpace(evidence.ConnectorOwnerKey)
	evidence.ConnectorNamespace = strings.TrimSpace(evidence.ConnectorNamespace)
	evidence.SourceObject = strings.TrimSpace(evidence.SourceObject)
	evidence.SourceRecordId = strings.TrimSpace(evidence.SourceRecordId)
	evidence.RemoteIDField = strings.TrimSpace(evidence.RemoteIDField)
	evidence.RemoteId = strings.TrimSpace(evidence.RemoteId)
	if evidence.ConnectorOwnerKey == "" || evidence.ConnectorNamespace == "" || evidence.SourceObject == "" || evidence.SourceRecordId == "" || evidence.RemoteIDField == "" || evidence.RemoteId == "" {
		return finalization, "", fmt.Errorf("connector inbound finalization: ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId, RemoteIDField, and RemoteId are required")
	}
	evidence.ID = strings.TrimSpace(evidence.ID)
	if evidence.ID == "" {
		return finalization, "", fmt.Errorf("connector inbound finalization: exact Evidence.ID is required")
	}
	if evidence.ConnectorOwnerKey != adoption.Identity.ConnectorOwnerKey ||
		evidence.SourceObject != adoption.Identity.SourceObject ||
		evidence.SourceRecordId != adoption.Identity.SourceRecordID ||
		evidence.RemoteId != adoption.RemoteID ||
		evidence.RemoteIDField != strings.TrimSpace(adoption.Snapshot.RemoteIDField) {
		return finalization, "", fmt.Errorf("connector inbound finalization evidence does not match durable identity and snapshot")
	}
	activeKey := ConnectorInboundEvidenceActiveKey(evidence.ConnectorOwnerKey, evidence.SourceObject, evidence.SourceRecordId, evidence.RemoteIDField)
	providedActiveKey := strings.TrimSpace(adoption.EvidenceActiveKey)
	if providedActiveKey != "" && providedActiveKey != activeKey {
		return finalization, "", fmt.Errorf("connector inbound finalization evidence active key does not match evidence identity")
	}
	evidence.EvidenceStatus = ConnectorInboundEvidenceStatusAuthored
	evidence.ActiveKey = activeKey
	adoption.EvidenceActiveKey = activeKey
	finalization.Evidence = evidence
	finalization.Adoption = adoption
	return finalization, activeKey, nil
}

func normalizeConnectorPushDurableSourceQuery(q ConnectorPushDurableSourceQuery) ConnectorPushDurableSourceQuery {
	q.ConnectorOwnerKey = strings.TrimSpace(q.ConnectorOwnerKey)
	q.ActionSlotID = strings.TrimSpace(q.ActionSlotID)
	q.ActionType = strings.TrimSpace(q.ActionType)
	q.SourceObject = strings.TrimSpace(q.SourceObject)
	q.SourceRecordID = strings.TrimSpace(q.SourceRecordID)
	return q
}

func validateConnectorPushDurableSourceQuery(q ConnectorPushDurableSourceQuery) error {
	if q.ConnectorOwnerKey == "" || q.ActionSlotID == "" || q.ActionType == "" || q.SourceObject == "" || q.SourceRecordID == "" {
		return fmt.Errorf("connector push durable source query is incomplete")
	}
	return nil
}

func normalizeConnectorPushDurableDestinationQuery(q ConnectorPushDurableDestinationQuery) ConnectorPushDurableDestinationQuery {
	q.ConnectorOwnerKey = strings.TrimSpace(q.ConnectorOwnerKey)
	q.ActionType = strings.TrimSpace(q.ActionType)
	q.SourceObject = strings.TrimSpace(q.SourceObject)
	q.SourceRecordID = strings.TrimSpace(q.SourceRecordID)
	q.DestinationSlotID = strings.TrimSpace(q.DestinationSlotID)
	return q
}

func validateConnectorPushDurableDestinationQuery(q ConnectorPushDurableDestinationQuery) error {
	if q.ConnectorOwnerKey == "" || q.ActionType == "" || q.SourceObject == "" || q.SourceRecordID == "" || q.DestinationSlotID == "" {
		return fmt.Errorf("connector push durable destination query is incomplete")
	}
	return nil
}

func requireConnectorPushOneRow(result sql.Result, name string) error {
	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("connector push durable transition %s rows affected: %w", name, err)
	}
	if rows != 1 {
		return fmt.Errorf("connector push durable transition %s updated %d rows, want 1", name, rows)
	}
	return nil
}

func connectorPushDuplicateKeyConflict(err error) error {
	if err == nil {
		return nil
	}
	msg := strings.ToLower(err.Error())
	if strings.Contains(msg, "duplicate entry") || strings.Contains(msg, "duplicate key") || strings.Contains(msg, "unique constraint") || strings.Contains(msg, "1062") {
		return fmt.Errorf("%w: duplicate create attempt durable key", ErrConnectorPushDurableConflict)
	}
	return nil
}

type connectorPushScanner interface {
	Scan(dest ...any) error
}

func scanConnectorPushLink(row connectorPushScanner) (*ConnectorPushRemoteLink, error) {
	link := &ConnectorPushRemoteLink{}
	var retryPolicyJSON string
	err := row.Scan(
		&link.ID, &link.LinkKey, &link.Status, &link.RemoteID,
		&link.Identity.ConnectorOwnerKey, &link.Identity.ActionSlotID, &link.Identity.ActionType,
		&link.Identity.SourceObject, &link.Identity.SourceRecordID, &link.Identity.DestinationSlotID,
		&link.Snapshot.RequesterOwnerPack, &link.Snapshot.ProfileOwnerPack,
		&link.Snapshot.ConnectorNamespace, &link.Snapshot.FlowDeveloperName, &link.Snapshot.FlowActionName,
		&link.Snapshot.TargetObject, &link.Snapshot.ProfileName, &link.Snapshot.CreatedPackInstallID, &link.Snapshot.LastValidatedPackInstallID,
		&link.Snapshot.CredentialRef, &link.Snapshot.EndpointClass, &retryPolicyJSON,
		&link.Snapshot.ProfileContractHash, &link.Snapshot.CreatePathTemplate,
		&link.Snapshot.UpdatePathTemplate, &link.Snapshot.ResponseIDPath, &link.Snapshot.RemoteIDField,
		&link.Snapshot.RemoteIDProvenanceField, &link.Snapshot.MappingFingerprint, &link.Snapshot.BodyFingerprint,
	)
	if err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(retryPolicyJSON), &link.Snapshot.RetryPolicy); err != nil {
		return nil, fmt.Errorf("decode connector push link retry policy: %w", err)
	}
	if err := requireConnectorPushProfileContractHash(link.Snapshot); err != nil {
		return nil, err
	}
	return link, nil
}

type connectorPushRecoveryHistoryRow struct {
	link                      ConnectorPushRemoteLink
	remoteLocatorHash         string
	remoteLocatorActiveKey    string
	compatibilityMirrorStatus string
}

func scanConnectorPushRecoveryHistory(row connectorPushScanner) (*connectorPushRecoveryHistoryRow, error) {
	history := &connectorPushRecoveryHistoryRow{}
	link := &history.link
	var retryPolicyJSON string
	err := row.Scan(
		&link.ID, &link.LinkKey, &link.Status, &link.RemoteID,
		&link.Identity.ConnectorOwnerKey, &link.Identity.ActionSlotID, &link.Identity.ActionType,
		&link.Identity.SourceObject, &link.Identity.SourceRecordID, &link.Identity.DestinationSlotID,
		&link.Snapshot.RequesterOwnerPack, &link.Snapshot.ProfileOwnerPack,
		&link.Snapshot.ConnectorNamespace, &link.Snapshot.FlowDeveloperName, &link.Snapshot.FlowActionName,
		&link.Snapshot.TargetObject, &link.Snapshot.ProfileName, &link.Snapshot.CreatedPackInstallID,
		&link.Snapshot.LastValidatedPackInstallID, &link.Snapshot.CredentialRef, &link.Snapshot.EndpointClass,
		&retryPolicyJSON, &link.Snapshot.ProfileContractHash, &link.Snapshot.CreatePathTemplate,
		&link.Snapshot.UpdatePathTemplate, &link.Snapshot.ResponseIDPath, &link.Snapshot.RemoteIDField,
		&link.Snapshot.RemoteIDProvenanceField, &link.Snapshot.MappingFingerprint, &link.Snapshot.BodyFingerprint,
		&history.remoteLocatorHash, &history.remoteLocatorActiveKey, &history.compatibilityMirrorStatus,
	)
	if err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(retryPolicyJSON), &link.Snapshot.RetryPolicy); err != nil {
		return nil, fmt.Errorf("decode connector push recovery history retry policy: %w", err)
	}
	if err := requireConnectorPushProfileContractHash(link.Snapshot); err != nil {
		return nil, err
	}
	return history, nil
}

func connectorPushPersistedRemoteLinkSnapshotEqual(left, right ConnectorPushDurableSnapshot) bool {
	return left.RequesterOwnerPack == right.RequesterOwnerPack &&
		left.ProfileOwnerPack == right.ProfileOwnerPack &&
		left.ConnectorNamespace == right.ConnectorNamespace &&
		left.FlowDeveloperName == right.FlowDeveloperName &&
		left.FlowActionName == right.FlowActionName &&
		left.TargetObject == right.TargetObject &&
		left.ProfileName == right.ProfileName &&
		left.CreatedPackInstallID == right.CreatedPackInstallID &&
		left.LastValidatedPackInstallID == right.LastValidatedPackInstallID &&
		left.CredentialRef == right.CredentialRef &&
		left.EndpointClass == right.EndpointClass &&
		reflect.DeepEqual(left.RetryPolicy, right.RetryPolicy) &&
		left.ProfileContractHash == right.ProfileContractHash &&
		left.CreatePathTemplate == right.CreatePathTemplate &&
		left.UpdatePathTemplate == right.UpdatePathTemplate &&
		left.ResponseIDPath == right.ResponseIDPath &&
		left.RemoteIDField == right.RemoteIDField &&
		left.RemoteIDProvenanceField == right.RemoteIDProvenanceField &&
		left.MappingFingerprint == right.MappingFingerprint &&
		left.BodyFingerprint == right.BodyFingerprint
}

func scanConnectorPushAttempt(row connectorPushScanner) (*ConnectorPushAttempt, error) {
	attempt := &ConnectorPushAttempt{}
	var createConflictPatchIntentJSON, patchDispatchIntentJSON, originatingUserSessionJSON, allowedFieldsJSON, associationSourceRecordIDsJSON, retryPolicyJSON, requestHeadersJSON, requestBodyBase64, responseHeadersJSON, responseBodyBase64 string
	var claimedAt, leaseExpiresAt, remoteIOStartedAt sql.NullTime
	err := row.Scan(
		&attempt.ID, &attempt.LinkKey, &attempt.Status, &attempt.DispatchLane, &attempt.AttemptOrdinal, &attempt.RemoteID,
		&attempt.Identity.ConnectorOwnerKey, &attempt.Identity.ActionSlotID, &attempt.Identity.ActionType,
		&attempt.Identity.SourceObject, &attempt.Identity.SourceRecordID, &attempt.Identity.DestinationSlotID,
		&attempt.Snapshot.RequesterOwnerPack, &attempt.Snapshot.ProfileOwnerPack,
		&attempt.Snapshot.ConnectorNamespace, &attempt.Snapshot.FlowDeveloperName, &attempt.Snapshot.FlowActionName,
		&attempt.Snapshot.TargetObject, &attempt.Snapshot.ProfileName, &attempt.Snapshot.CreatedPackInstallID,
		&attempt.Snapshot.CredentialRef, &attempt.Snapshot.EndpointClass, &retryPolicyJSON,
		&attempt.Snapshot.ProfileContractHash, &attempt.Snapshot.CreatePathTemplate,
		&attempt.Snapshot.UpdatePathTemplate, &attempt.Snapshot.ResponseIDPath, &attempt.Snapshot.RemoteIDField,
		&attempt.Snapshot.RemoteIDProvenanceField, &attempt.Snapshot.RemoteIDMirrorExpectedValue,
		&attempt.Snapshot.RemoteIDMirrorCurrentValue, &attempt.Snapshot.RemoteIDProvenanceMirrorExpectedValue,
		&attempt.Snapshot.RemoteIDProvenanceMirrorCurrentValue, &attempt.Snapshot.MappingFingerprint, &attempt.Snapshot.BodyFingerprint,
		&attempt.Snapshot.CreateConflictResolutionFingerprint,
		&createConflictPatchIntentJSON,
		&patchDispatchIntentJSON,
		&attempt.Snapshot.OriginatingUserID, &originatingUserSessionJSON, &allowedFieldsJSON, &associationSourceRecordIDsJSON, &attempt.Snapshot.RequestIdempotencyKey,
		&attempt.Snapshot.RequestMethod, &attempt.Snapshot.RequestRelativePath, &requestHeadersJSON,
		&requestBodyBase64, &attempt.Snapshot.RequestBodyHash, &attempt.Snapshot.ResponseStatusCode,
		&responseHeadersJSON, &responseBodyBase64, &attempt.Snapshot.ResponseBodyHash,
		&attempt.RetryCount, &attempt.ClaimedBy, &attempt.ClaimToken, &claimedAt, &leaseExpiresAt, &remoteIOStartedAt,
	)
	if err != nil {
		return nil, err
	}
	attempt.DispatchLane = strings.TrimSpace(attempt.DispatchLane)
	if !isConnectorPushAttemptDispatchLane(attempt.DispatchLane) {
		return nil, fmt.Errorf("connector push durable dispatch lane %q is invalid", attempt.DispatchLane)
	}
	if err := decodeConnectorPushOriginatingAuthority(
		originatingUserSessionJSON,
		&attempt.Snapshot,
	); err != nil {
		return nil, err
	}
	if err := json.Unmarshal([]byte(createConflictPatchIntentJSON), &attempt.Snapshot.CreateConflictPatchIntent); err != nil {
		return nil, fmt.Errorf("decode connector push create conflict patch intent: %w", err)
	}
	if intent := attempt.Snapshot.CreateConflictPatchIntent; intent != nil {
		if err := requireConnectorPushBodyHash("create conflict patch intent", intent.RequestBody, intent.RequestBodyHash); err != nil {
			return nil, err
		}
	}
	if err := json.Unmarshal([]byte(patchDispatchIntentJSON), &attempt.Snapshot.PatchDispatchIntent); err != nil {
		return nil, fmt.Errorf("decode connector push PATCH dispatch intent: %w", err)
	}
	err = json.Unmarshal([]byte(allowedFieldsJSON), &attempt.Snapshot.AllowedSourceFields)
	if err != nil {
		return nil, fmt.Errorf("decode connector push allowed source fields: %w", err)
	}
	err = json.Unmarshal([]byte(associationSourceRecordIDsJSON), &attempt.Snapshot.AssociationSourceRecordIDs)
	if err != nil {
		return nil, fmt.Errorf("decode connector push association source record ids: %w", err)
	}
	if err := json.Unmarshal([]byte(retryPolicyJSON), &attempt.Snapshot.RetryPolicy); err != nil {
		return nil, fmt.Errorf("decode connector push retry policy: %w", err)
	}
	if err := requireConnectorPushProfileContractHash(attempt.Snapshot); err != nil {
		return nil, err
	}
	err = json.Unmarshal([]byte(requestHeadersJSON), &attempt.Snapshot.RequestHeaders)
	if err != nil {
		return nil, fmt.Errorf("decode connector push request headers: %w", err)
	}
	if attempt.Snapshot.RequestBody, err = base64.StdEncoding.DecodeString(requestBodyBase64); err != nil {
		return nil, fmt.Errorf("decode connector push request body: %w", err)
	}
	if err := requireConnectorPushBodyHash("request body", attempt.Snapshot.RequestBody, attempt.Snapshot.RequestBodyHash); err != nil {
		return nil, err
	}
	if attempt.Snapshot.PatchDispatchIntent != nil {
		var err error
		switch strings.TrimSpace(attempt.Snapshot.PatchDispatchIntent.State) {
		case connectorPushPatchDispatchStateAssociationDecision:
			err = requireConnectorPushAssociationDecisionIntent(attempt.Snapshot)
		case connectorPushPatchDispatchStateAwaitingCreateLink:
			err = requireConnectorPushAwaitingCreateLinkIntent(attempt.Snapshot)
		default:
			err = requireConnectorPushPatchDispatchIntent(attempt.Snapshot)
		}
		if err != nil {
			return nil, err
		}
	}
	err = json.Unmarshal([]byte(responseHeadersJSON), &attempt.Snapshot.ResponseHeaders)
	if err != nil {
		return nil, fmt.Errorf("decode connector push response headers: %w", err)
	}
	if attempt.Snapshot.ResponseBody, err = base64.StdEncoding.DecodeString(responseBodyBase64); err != nil {
		return nil, fmt.Errorf("decode connector push response body: %w", err)
	}
	if responseBodyBase64 != "" || strings.TrimSpace(attempt.Snapshot.ResponseBodyHash) != "" {
		if err := requireConnectorPushResponseBodyEvidence(attempt.Snapshot.ResponseBody, attempt.Snapshot.ResponseBodyHash); err != nil {
			return nil, err
		}
	}
	if claimedAt.Valid {
		attempt.ClaimedAt = claimedAt.Time
	}
	if leaseExpiresAt.Valid {
		attempt.LeaseExpiresAt = leaseExpiresAt.Time
	}
	if remoteIOStartedAt.Valid {
		attempt.RemoteIOStartedAt = remoteIOStartedAt.Time
	}
	return attempt, nil
}

func isConnectorPushAttemptDispatchLane(dispatchLane string) bool {
	switch dispatchLane {
	case ConnectorPushAttemptDispatchLaneConfigured, ConnectorPushAttemptDispatchLaneDeferredAssociation:
		return true
	default:
		return false
	}
}

func ConnectorPushRemoteLocatorHash(identity ConnectorPushDurableIdentity, remoteID string) string {
	sum := sha256.Sum256([]byte(strings.Join([]string{
		identity.ConnectorOwnerKey,
		identity.DestinationSlotID,
		strings.TrimSpace(remoteID),
	}, "\x1f")))
	return hex.EncodeToString(sum[:])
}

func ConnectorPushProfileContractHash(snapshot ConnectorPushDurableSnapshot) string {
	sum := sha256.Sum256([]byte(strings.Join([]string{
		snapshot.ProfileOwnerPack,
		snapshot.ProfileName,
		connectorPushProfileHashPackInstallID(snapshot),
		snapshot.CredentialRef,
		snapshot.EndpointClass,
		mustConnectorPushJSON(snapshot.RetryPolicy),
	}, "\x1f")))
	return hex.EncodeToString(sum[:])
}

func connectorPushProfileHashPackInstallID(snapshot ConnectorPushDurableSnapshot) string {
	if installID := strings.TrimSpace(snapshot.LastValidatedPackInstallID); installID != "" {
		return installID
	}
	return strings.TrimSpace(snapshot.CreatedPackInstallID)
}

func connectorPushSHA256Hex(data []byte) string {
	sum := sha256.Sum256(data)
	return hex.EncodeToString(sum[:])
}

func requireConnectorPushBodyHash(name string, body []byte, hash string) error {
	if connectorPushSHA256Hex(body) != strings.TrimSpace(hash) {
		return fmt.Errorf("decode connector push %s hash mismatch", name)
	}
	return nil
}

func requireConnectorPushResponseBodyEvidence(body []byte, hash string) error {
	if len(body) == 0 && strings.TrimSpace(hash) != "" {
		return nil
	}
	return requireConnectorPushBodyHash("response body", body, hash)
}

func requireConnectorPushProfileContractHash(snapshot ConnectorPushDurableSnapshot) error {
	if snapshot.RetryPolicy.MaxAttempts < 1 {
		return fmt.Errorf("decode connector push retry policy must set MaxAttempts")
	}
	if ConnectorPushProfileContractHash(snapshot) != strings.TrimSpace(snapshot.ProfileContractHash) {
		return fmt.Errorf("decode connector push profile contract hash mismatch")
	}
	return nil
}

func mustConnectorPushJSON(v any) string {
	data, _ := json.Marshal(v)
	return string(data)
}

func nullableConnectorPushString(s string) any {
	if strings.TrimSpace(s) == "" {
		return nil
	}
	return s
}

// --- Admin sync-status read (two-SSoT, per-record fan-out, #2107) ---

// findRecordPushStateQueryLink is Query A for FindRecordPushState: mirrors
// FindStateByDestination — surfaces finalized, suspended_drift, and
// owner_deleted_pending_recovery,
// EXCLUDING retired (so a retired lineage cannot shadow a live one via the
// LastModifiedDate-DESC collapse, since RetireLink bumps LastModifiedDate).
// The Status placeholders are bound to the SAME ConnectorPushLinkStatus*
// constants FindStateByDestination passes — one rule source, two query shapes
// (a per-destination point read there, a per-record fan-out here).
// Rides idx_connector_remote_link_source (SourceObject, SourceRecordId).
// Package-level const so the no-leak test can scan it without running SQL.
// Id is selected server-side only (deterministic tiebreaker); it never reaches
// the wire DTO.
const findRecordPushStateQueryLink = `
SELECT ConnectorOwnerKey, ActionType, DestinationSlotId, TargetObject,
       Status, LastModifiedDate, Id
FROM _System_ConnectorRemoteLink
WHERE SourceObject = ? AND SourceRecordId = ?
  AND Status IN (?, ?, ?)
`

// findRecordPushStateQueryAttempt is Query B for FindRecordPushState: reads all
// non-superseded attempts (POST or PATCH) for the given (SourceObject, SourceRecordId).
// Intentionally includes both POST (create) and PATCH (update) attempts so that
// a PATCH terminal_failed or ambiguous_remote_effect on an already-linked record
// is surfaced — latest-wins by LastModifiedDate DESC, matching the authoritative
// FindStateByDestination read which also filters by Status only (not RequestMethod).
// KEEP finalized, terminal_failed, ambiguous_remote_effect and all in-flight
// statuses; exclude only superseded (definitionally replaced by a newer attempt).
// Rides idx_connector_push_attempt_source (SourceObject, SourceRecordId).
// Package-level const so the no-leak test can scan it without running SQL.
// Id and RequestMethod are selected server-side only; they never reach the wire DTO.
//
// Dual-rule merge (applied in Go, not SQL):
//   - 5-tuple WITH an active link: use attemptMap (best non-superseded, any method).
//   - 5-tuple WITHOUT an active link: use attemptLiveMap — only rows where
//     RequestMethod='POST' AND Status NOT IN ('finalized','superseded'). This
//     mirrors FindStateByDestination and kills the "retired lineage resurrects as
//     attempt-only row" bug: finalized POST history and orphan PATCH attempts never
//     produce a row when no active link exists.
const findRecordPushStateQueryAttempt = `
SELECT ConnectorOwnerKey, ActionType, DestinationSlotId, TargetObject,
       Status, ResponseStatusCode, LastErrorAt, LastModifiedDate,
       Id, RequestMethod
FROM _System_ConnectorPushAttempt
WHERE SourceObject = ? AND SourceRecordId = ?
  AND Status <> 'superseded'`

// ConnectorPushRecordStateQuery is the input to FindRecordPushState.
type ConnectorPushRecordStateQuery struct {
	SourceObject   string
	SourceRecordID string
}

// ConnectorPushRecordDestinationState is one row of the FindRecordPushState
// result: the combined link+attempt diagnostic status for one destination-lineage
// 5-tuple (ConnectorOwnerKey, ActionType, SourceObject, SourceRecordId,
// DestinationSlotId). ConnectorOwnerKey and ActionType are lineage
// discriminators serialized to the DTO under the ViewConnectorDiagnostics gate.
// LastError is never selected.
//
// Linkage-aware row rule (mirrors FindStateByDestination):
//   - 5-tuple WITH a create-blocking link (Status IN 'finalized',
//     'suspended_drift','owner_deleted_pending_recovery'): always
//     emitted; attempt display uses the best non-superseded attempt (any RequestMethod).
//   - 5-tuple WITHOUT an active link: emitted ONLY when a runtime-live attempt exists
//     (RequestMethod='POST' AND Status NOT IN ('finalized','superseded')). Finalized
//     POST history and orphan PATCH attempts never produce a row on their own — this
//     prevents retired lineage from resurrecting as phantom attempt-only rows.
type ConnectorPushRecordDestinationState struct {
	// Lineage discriminators (serialized to the DTO under ViewConnectorDiagnostics).
	ConnectorOwnerKey string
	ActionType        string

	// Destination label (serialized).
	DestinationSlotID string
	TargetObject      string

	// Link-side fields from _System_ConnectorRemoteLink.
	// HasLink=false means no link row for this 5-tuple.
	HasLink    bool
	LinkStatus string

	// Attempt-side fields from _System_ConnectorPushAttempt.
	// HasAttempt=false means no non-superseded attempt (POST or PATCH) for this 5-tuple.
	HasAttempt         bool
	LastAttemptStatus  string
	ResponseStatusCode sql.NullInt64
	LastErrorAt        sql.NullTime
}

// findRecordPushState5Tuple is the Go-side group key for FindRecordPushState.
type findRecordPushState5Tuple struct {
	ConnectorOwnerKey string
	ActionType        string
	DestinationSlotID string
}

// findRecordPushStateLinkRow is the server-side row type for Query A of
// FindRecordPushState. Id is a deterministic tiebreaker and is never
// serialized to the wire DTO.
type findRecordPushStateLinkRow struct {
	connectorOwnerKey string
	actionType        string
	destinationSlotID string
	targetObject      string
	status            string
	lastModified      sql.NullTime
	id                string // server-side tiebreaker; never serialized
}

// findRecordPushStateAttemptRow is the server-side row type for Query B of
// FindRecordPushState. Id and requestMethod are server-side only and are never
// serialized to the wire DTO.
type findRecordPushStateAttemptRow struct {
	connectorOwnerKey  string
	actionType         string
	destinationSlotID  string
	targetObject       string
	status             string
	responseStatusCode sql.NullInt64
	lastErrorAt        sql.NullTime
	lastModified       sql.NullTime
	id                 string // server-side tiebreaker; never serialized
	requestMethod      string // server-side only; used for attemptLiveMap filter; never serialized
}

// linkRowBeats reports whether candidate beats existing for the link-side
// one-per-5-tuple collapse. Mirrors the authoritative FindStateByDestination
// ordering: LastModifiedDate DESC, then Id DESC as deterministic tiebreaker.
// A valid (non-NULL) LastModifiedDate beats an invalid (NULL) one.
func linkRowBeats(candidate, existing findRecordPushStateLinkRow) bool {
	if candidate.lastModified.Valid && existing.lastModified.Valid {
		if !candidate.lastModified.Time.Equal(existing.lastModified.Time) {
			return candidate.lastModified.Time.After(existing.lastModified.Time)
		}
	} else if candidate.lastModified.Valid && !existing.lastModified.Valid {
		return true
	} else if !candidate.lastModified.Valid && existing.lastModified.Valid {
		return false
	}
	return candidate.id > existing.id
}

// attemptRowBeats reports whether candidate beats existing for the attempt-side
// one-per-5-tuple collapse. Superseded attempts are already excluded by the
// SQL predicate (Status <> 'superseded'), so no supersede signal is needed here.
// Mirrors the authoritative FindStateByDestination ordering: LastModifiedDate DESC,
// then Id DESC as deterministic tiebreaker.
// A valid (non-NULL) LastModifiedDate beats an invalid (NULL) one.
func attemptRowBeats(candidate, existing findRecordPushStateAttemptRow) bool {
	if candidate.lastModified.Valid && existing.lastModified.Valid {
		if !candidate.lastModified.Time.Equal(existing.lastModified.Time) {
			return candidate.lastModified.Time.After(existing.lastModified.Time)
		}
	} else if candidate.lastModified.Valid && !existing.lastModified.Valid {
		return true
	} else if !candidate.lastModified.Valid && existing.lastModified.Valid {
		return false
	}
	return candidate.id > existing.id
}

// connectorPushAttemptIsLiveCreate is the Go-side statement of the runtime-live
// create rule: RequestMethod = POST AND Status NOT IN (finalized, superseded).
// It restates the SQL filter inside the authoritative FindStateByDestination
// attempt query (which binds the same constants); the two are one rule with
// two enforcement points — change both together or not at all.
func connectorPushAttemptIsLiveCreate(requestMethod, status string) bool {
	return requestMethod == http.MethodPost &&
		status != ConnectorPushAttemptStatusFinalized &&
		status != ConnectorPushAttemptStatusSuperseded
}

// FindRecordPushState returns the combined link+attempt diagnostic status for
// every destination-lineage 5-tuple (ConnectorOwnerKey, ActionType,
// SourceObject, SourceRecordId, DestinationSlotId) that has an active link or
// a runtime-live attempt for the given (q.SourceObject, q.SourceRecordID).
//
// Two-pass fan-out + Go merge:
//   - Query A reads finalized, suspended_drift, and owner_deleted_pending_recovery
//     link rows (rides idx_connector_remote_link_source).
//   - Query B reads all non-superseded attempts (POST and PATCH) and populates two maps:
//     attemptMap    — best non-superseded attempt per 5-tuple (any RequestMethod); used
//     for attempt display on linked rows.
//     attemptLiveMap — best runtime-live POST attempt per 5-tuple (POST + NOT IN
//     finalized,superseded); used as the key set for unlinked rows,
//     mirroring FindStateByDestination's linkage-aware rule.
//
// Row-set is: keys(linkMap) ∪ keys(attemptLiveMap)  — NOT all of attemptMap.
// This ensures finalized POST history and orphan PATCH attempts never resurrect
// a destination-lineage row when no active link exists.
//
// Each side is reduced to one-per-5-tuple via LastModifiedDate DESC + Id DESC
// tiebreaker. Output is sorted by (DestinationSlotId, ConnectorOwnerKey).
func (s *SQLConnectorPushDurableStore) FindRecordPushState(ctx context.Context, q ConnectorPushRecordStateQuery) ([]ConnectorPushRecordDestinationState, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	sourceObject := strings.TrimSpace(q.SourceObject)
	sourceRecordID := strings.TrimSpace(q.SourceRecordID)
	if sourceObject == "" || sourceRecordID == "" {
		return nil, fmt.Errorf("FindRecordPushState: SourceObject and SourceRecordID are required")
	}

	// Open a single read transaction so that both Query A and Query B observe the
	// same committed snapshot. Writes update _System_ConnectorRemoteLink and
	// _System_ConnectorPushAttempt atomically inside one BeginTx; without a shared
	// read snapshot the two separate reads can straddle a write commit and fabricate
	// impossible cross-SSoT states. TiDB rejects sql.TxOptions{ReadOnly:true} with
	// Error 1235, so we use a plain transaction — TiDB's default REPEATABLE READ
	// isolation gives both queries one consistent snapshot. The deferred Rollback is
	// a no-op for a read-only path.
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("FindRecordPushState begin tx: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	// --- Query A: link side ---
	// Map from 5-tuple key to best link row.
	// Collapse rule: (1) LastModifiedDate DESC; (2) Id DESC.
	linkMap := make(map[findRecordPushState5Tuple]findRecordPushStateLinkRow)

	rowsA, err := tx.QueryContext(ctx, findRecordPushStateQueryLink, sourceObject, sourceRecordID,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
		ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
	)
	if err != nil {
		return nil, fmt.Errorf("FindRecordPushState query A: %w", err)
	}
	defer rowsA.Close()
	for rowsA.Next() {
		var lr findRecordPushStateLinkRow
		if err := rowsA.Scan(
			&lr.connectorOwnerKey, &lr.actionType,
			&lr.destinationSlotID, &lr.targetObject,
			&lr.status, &lr.lastModified, &lr.id,
		); err != nil {
			return nil, fmt.Errorf("FindRecordPushState scan A: %w", err)
		}
		k := findRecordPushState5Tuple{lr.connectorOwnerKey, lr.actionType, lr.destinationSlotID}
		if existing, ok := linkMap[k]; !ok || linkRowBeats(lr, existing) {
			linkMap[k] = lr
		}
	}
	if err := rowsA.Err(); err != nil {
		return nil, fmt.Errorf("FindRecordPushState rows A: %w", err)
	}

	// --- Query B: attempt side ---
	// attemptMap: best non-superseded attempt per 5-tuple (any RequestMethod).
	//   Used to populate attempt display fields on rows that have an active link.
	// attemptLiveMap: best runtime-live POST attempt per 5-tuple.
	//   RequestMethod='POST' AND Status NOT IN ('finalized','superseded').
	//   Used as the key set for rows without an active link, mirroring
	//   FindStateByDestination: finalized history and orphan PATCH attempts must
	//   not resurrect a destination-lineage row on their own.
	// Collapse rule for both: (1) LastModifiedDate DESC; (2) Id DESC.
	// Superseded attempts are already excluded by the SQL predicate (Status <> 'superseded').
	attemptMap := make(map[findRecordPushState5Tuple]findRecordPushStateAttemptRow)
	attemptLiveMap := make(map[findRecordPushState5Tuple]findRecordPushStateAttemptRow)

	rowsB, err := tx.QueryContext(ctx, findRecordPushStateQueryAttempt, sourceObject, sourceRecordID)
	if err != nil {
		return nil, fmt.Errorf("FindRecordPushState query B: %w", err)
	}
	defer rowsB.Close()
	for rowsB.Next() {
		var ar findRecordPushStateAttemptRow
		if err := rowsB.Scan(
			&ar.connectorOwnerKey, &ar.actionType,
			&ar.destinationSlotID, &ar.targetObject,
			&ar.status, &ar.responseStatusCode,
			&ar.lastErrorAt, &ar.lastModified,
			&ar.id, &ar.requestMethod,
		); err != nil {
			return nil, fmt.Errorf("FindRecordPushState scan B: %w", err)
		}
		k := findRecordPushState5Tuple{ar.connectorOwnerKey, ar.actionType, ar.destinationSlotID}
		if existing, ok := attemptMap[k]; !ok || attemptRowBeats(ar, existing) {
			attemptMap[k] = ar
		}
		// attemptLiveMap: runtime-live POST attempts only (superseded already excluded by SQL).
		if connectorPushAttemptIsLiveCreate(ar.requestMethod, ar.status) {
			if existing, ok := attemptLiveMap[k]; !ok || attemptRowBeats(ar, existing) {
				attemptLiveMap[k] = ar
			}
		}
	}
	if err := rowsB.Err(); err != nil {
		return nil, fmt.Errorf("FindRecordPushState rows B: %w", err)
	}

	// --- Go merge: union key sets, emit one row per 5-tuple ---
	// Row-set = keys(linkMap) ∪ keys(attemptLiveMap).
	// Finalized POST history and orphan PATCH attempts (present in attemptMap but
	// not in attemptLiveMap) are excluded when there is no active link row —
	// this prevents retired lineages from resurrecting as phantom attempt-only rows.
	seen := make(map[findRecordPushState5Tuple]bool)
	for k := range linkMap {
		seen[k] = true
	}
	for k := range attemptLiveMap {
		seen[k] = true
	}

	results := make([]ConnectorPushRecordDestinationState, 0, len(seen))
	for k := range seen {
		row := ConnectorPushRecordDestinationState{
			ConnectorOwnerKey: k.ConnectorOwnerKey,
			ActionType:        k.ActionType,
			DestinationSlotID: k.DestinationSlotID,
		}
		if lr, ok := linkMap[k]; ok {
			row.HasLink = true
			row.LinkStatus = lr.status
			row.TargetObject = lr.targetObject
		}
		// Attempt display: use attemptMap (best non-superseded, any method) for linked rows;
		// use attemptLiveMap (runtime-live POST) for unlinked rows (guaranteed present since
		// the key came from attemptLiveMap).
		var ar findRecordPushStateAttemptRow
		var hasAttempt bool
		if row.HasLink {
			ar, hasAttempt = attemptMap[k]
		} else {
			ar, hasAttempt = attemptLiveMap[k]
		}
		if hasAttempt {
			row.HasAttempt = true
			row.LastAttemptStatus = ar.status
			row.ResponseStatusCode = ar.responseStatusCode
			row.LastErrorAt = ar.lastErrorAt
			if row.TargetObject == "" {
				row.TargetObject = ar.targetObject
			}
		}
		results = append(results, row)
	}

	// Sort deterministically by (DestinationSlotId, ConnectorOwnerKey).
	sort.Slice(results, func(i, j int) bool {
		if results[i].DestinationSlotID != results[j].DestinationSlotID {
			return results[i].DestinationSlotID < results[j].DestinationSlotID
		}
		return results[i].ConnectorOwnerKey < results[j].ConnectorOwnerKey
	})

	return results, nil
}

// --- Inbound remote-id authorship evidence (epic #2028, issue #2040) ---

// InsertAuthoredEvidence inserts one authorship-provenance row into
// _System_ConnectorInboundRemoteIDEvidence with EvidenceStatus='authored' and
// ActiveKey set to ConnectorInboundEvidenceActiveKey(connectorOwnerKey,
// sourceObject, sourceRecordId). When the store carries a txExtractor and an
// ambient *sql.Tx is present on ctx the INSERT executes inside that
// transaction, making the write atomic with any surrounding record mutation.
// On ActiveKey UNIQUE conflict (another authored row for the same connector +
// local record) the method returns ErrConnectorPushDurableConflict.
func (s *SQLConnectorPushDurableStore) InsertAuthoredEvidence(ctx context.Context, evidence ConnectorInboundRemoteIDEvidence) error {
	if s == nil || s.db == nil {
		return fmt.Errorf("connector push durable store is not wired")
	}
	if strings.TrimSpace(evidence.ConnectorOwnerKey) == "" || strings.TrimSpace(evidence.SourceObject) == "" || strings.TrimSpace(evidence.SourceRecordId) == "" || strings.TrimSpace(evidence.RemoteIDField) == "" || strings.TrimSpace(evidence.ConnectorNamespace) == "" || strings.TrimSpace(evidence.RemoteId) == "" {
		return fmt.Errorf("connector inbound evidence: ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId, RemoteIDField, and RemoteId are required")
	}
	if strings.TrimSpace(evidence.ID) == "" {
		evidence.ID = utils.GenerateID()
	}
	// Normalize identity/value fields so the stored row, the ActiveKey, and a later
	// FindActiveInboundRemoteIDEvidence lookup all agree regardless of caller
	// whitespace (validation above trims, so these must too — otherwise " owner "
	// and "owner" would key to different active rows for the same logical evidence).
	evidence.ConnectorOwnerKey = strings.TrimSpace(evidence.ConnectorOwnerKey)
	evidence.ConnectorNamespace = strings.TrimSpace(evidence.ConnectorNamespace)
	evidence.SourceObject = strings.TrimSpace(evidence.SourceObject)
	evidence.SourceRecordId = strings.TrimSpace(evidence.SourceRecordId)
	evidence.RemoteIDField = strings.TrimSpace(evidence.RemoteIDField)
	evidence.RemoteId = strings.TrimSpace(evidence.RemoteId)
	activeKey := ConnectorInboundEvidenceActiveKey(evidence.ConnectorOwnerKey, evidence.SourceObject, evidence.SourceRecordId, evidence.RemoteIDField)
	execer := s.connectorPushAmbientExecer(ctx)
	_, err := execer.ExecContext(ctx, `
INSERT INTO _System_ConnectorInboundRemoteIDEvidence
  (Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
   RemoteIDField, RemoteId, EvidenceStatus, ActiveKey)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		evidence.ID,
		evidence.ConnectorOwnerKey,
		evidence.ConnectorNamespace,
		evidence.SourceObject,
		evidence.SourceRecordId,
		evidence.RemoteIDField,
		evidence.RemoteId,
		ConnectorInboundEvidenceStatusAuthored,
		activeKey,
	)
	if err != nil {
		if conflictErr := connectorPushDuplicateKeyConflict(err); conflictErr != nil {
			return fmt.Errorf("%w: active authored evidence already exists for this connector and record", ErrConnectorPushDurableConflict)
		}
		return err
	}
	return nil
}

// FindActiveInboundRemoteIDEvidence returns the authored (active) evidence row
// for the given connector owner key, source object, source record id, and
// remote-id field (the per-destination dimension), or nil if no such row exists.
// FindActiveInboundRemoteIDEvidence reads through the store's *sql.DB, so it
// observes rows COMMITTED by other transactions. The inbound register action
// uses it ONLY for the post-insert conflict reread, where the conflicting row
// was committed by a CONCURRENT inbound transaction and would not be visible in
// the caller's own tx snapshot. For the in-tx idempotency check use
// FindActiveInboundRemoteIDEvidenceInTx instead.
func (s *SQLConnectorPushDurableStore) FindActiveInboundRemoteIDEvidence(ctx context.Context, connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField string) (*ConnectorInboundRemoteIDEvidence, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	return s.findActiveInboundRemoteIDEvidence(ctx, s.db, connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField)
}

// FindActiveInboundRemoteIDEvidenceInTx reads through the ambient transaction
// (when one is present) so the inbound register action's idempotency check
// observes evidence the SAME inbound tx wrote earlier in the flow
// (read-after-write within one atomic upsert). It falls back to *sql.DB when the
// store has no tx-extractor or ctx carries no transaction (e.g. the outbound
// plain store and unit tests).
func (s *SQLConnectorPushDurableStore) FindActiveInboundRemoteIDEvidenceInTx(ctx context.Context, connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField string) (*ConnectorInboundRemoteIDEvidence, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	return s.findActiveInboundRemoteIDEvidence(ctx, s.connectorPushAmbientQuerier(ctx), connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField)
}

// FindInboundRemoteIDEvidenceHintsInAmbientTx performs the deliberately
// non-locking first read in the shared admission order. Hints only narrow work;
// installed metadata and a later FOR UPDATE reread remain authority.
func (s *SQLConnectorPushDurableStore) FindInboundRemoteIDEvidenceHintsInAmbientTx(
	ctx context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
) ([]ConnectorInboundRemoteIDEvidence, error) {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return nil, err
	}
	selector, err = normalizeConnectorInboundRemoteIdentityRecoverySelector(selector)
	if err != nil {
		return nil, err
	}
	rows, err := tx.QueryContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus,
       COALESCE(ConsumedByLinkKey, ''), COALESCE(ActiveKey, ''),
       CreatedDate, LastModifiedDate
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE ConnectorOwnerKey = ?
   AND ConnectorNamespace = ?
   AND RemoteId = ?
   AND EvidenceStatus IN (?, ?)
 ORDER BY SourceObject, RemoteIDField, Id`,
		selector.ConnectorOwnerKey,
		selector.ConnectorNamespace,
		selector.RemoteID,
		ConnectorInboundEvidenceStatusAuthored,
		ConnectorInboundEvidenceStatusConsumed,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var hints []ConnectorInboundRemoteIDEvidence
	for rows.Next() {
		evidence, err := scanConnectorInboundEvidence(rows)
		if err != nil {
			return nil, err
		}
		hints = append(hints, *evidence)
	}
	return hints, rows.Err()
}

// ReadConnectorInboundRemoteIdentityRecoveryEvidence performs the fresh,
// post-transaction evidence readback used to resolve ambiguous commit or
// rollback acknowledgements. It deliberately reads through *sql.DB rather
// than an ambient transaction carried by ctx.
func (s *SQLConnectorPushDurableStore) ReadConnectorInboundRemoteIdentityRecoveryEvidence(
	ctx context.Context,
	evidenceID string,
) (*ConnectorInboundRemoteIDEvidence, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	evidenceID = strings.TrimSpace(evidenceID)
	if evidenceID == "" {
		return nil, fmt.Errorf("connector inbound remote identity recovery evidence ID is required")
	}
	evidence, err := scanConnectorInboundEvidence(s.db.QueryRowContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus,
       COALESCE(ConsumedByLinkKey, ''), COALESCE(ActiveKey, ''),
       CreatedDate, LastModifiedDate
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE Id = ?
 LIMIT 1`, evidenceID))
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return evidence, err
}

// FindUniqueInboundRemoteIDEvidenceForRecoveryInAmbientTx is the authoritative
// evidence reread after the source row and lane are locked. It locks candidates
// inside the locked installed binding allowlist and rejects fan-in.
func (s *SQLConnectorPushDurableStore) FindUniqueInboundRemoteIDEvidenceForRecoveryInAmbientTx(
	ctx context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	bindings []ConnectorInboundRemoteIDEvidenceBinding,
) (*ConnectorInboundRemoteIDEvidence, error) {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return nil, err
	}
	return s.findUniqueInboundRemoteIDEvidenceForRecovery(ctx, tx, selector, bindings, true)
}

// LockConnectorRemoteIdentityLaneInAmbientTx acquires the persistent PK lane
// for one engine-derived remote locator. The lane is only a mutex; callers must
// reread RemoteLink/evidence authority after this point.
func (s *SQLConnectorPushDurableStore) LockConnectorRemoteIdentityLaneInAmbientTx(
	ctx context.Context,
	identity ConnectorPushDurableIdentity,
	remoteID string,
) error {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return err
	}
	return lockConnectorRemoteIdentityLaneTx(ctx, tx, identity, remoteID)
}

func lockConnectorRemoteIdentityLaneTx(ctx context.Context, tx *sql.Tx, identity ConnectorPushDurableIdentity, remoteID string) error {
	remoteID = strings.TrimSpace(remoteID)
	if strings.TrimSpace(identity.ConnectorOwnerKey) == "" || strings.TrimSpace(identity.DestinationSlotID) == "" || remoteID == "" {
		return fmt.Errorf("connector remote identity admission lane requires connector owner, destination slot, and remote id")
	}
	return lockConnectorRemoteIdentityLaneHashTx(ctx, tx, ConnectorPushRemoteLocatorHash(identity, remoteID))
}

func lockConnectorRemoteIdentityLaneHashTx(ctx context.Context, tx *sql.Tx, laneHash string) error {
	laneHash = strings.TrimSpace(laneHash)
	if laneHash == "" {
		return fmt.Errorf("connector remote identity lane hash is required")
	}
	if _, err := tx.ExecContext(ctx, `
INSERT INTO _System_ConnectorRemoteIdentityLane
       (RemoteLocatorHash, CreatedDate, LastModifiedDate)
VALUES (?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON DUPLICATE KEY UPDATE RemoteLocatorHash = VALUES(RemoteLocatorHash)`, laneHash); err != nil {
		return err
	}
	var lockedHash string
	if err := tx.QueryRowContext(ctx, `
SELECT RemoteLocatorHash
  FROM _System_ConnectorRemoteIdentityLane
 WHERE RemoteLocatorHash = ?
 FOR UPDATE`, laneHash).Scan(&lockedHash); err != nil {
		return err
	}
	if lockedHash != laneHash {
		return fmt.Errorf("%w: connector remote identity admission lane readback mismatch", ErrConnectorPushDurableConflict)
	}
	return nil
}

// ensureConnectorRemoteIdentityOwnerCreateAllowedTx serializes new outbound
// creates with a deleted-owner transaction. The fence is runtime coordination
// state only; RemoteLink remains the sole remote-identity authority.
func ensureConnectorRemoteIdentityOwnerCreateAllowedTx(ctx context.Context, tx *sql.Tx, sourceObject, sourceRecordID string) error {
	ownerDeleted, err := lockConnectorRemoteIdentityOwnerFenceTx(ctx, tx, sourceObject, sourceRecordID)
	if err != nil {
		return err
	}
	if ownerDeleted {
		return fmt.Errorf("%w: connector remote identity owner is deleted", ErrConnectorPushDurableConflict)
	}
	return nil
}

func markConnectorRemoteIdentityOwnerDeletedFenceTx(ctx context.Context, tx *sql.Tx, sourceObject, sourceRecordID string) error {
	return setConnectorRemoteIdentityOwnerDeletedFenceTx(ctx, tx, sourceObject, sourceRecordID, true)
}

func clearConnectorRemoteIdentityOwnerDeletedFenceTx(ctx context.Context, tx *sql.Tx, sourceObject, sourceRecordID string) error {
	return setConnectorRemoteIdentityOwnerDeletedFenceTx(ctx, tx, sourceObject, sourceRecordID, false)
}

func setConnectorRemoteIdentityOwnerDeletedFenceTx(ctx context.Context, tx *sql.Tx, sourceObject, sourceRecordID string, ownerDeleted bool) error {
	return persistence.SetConnectorRemoteIdentityOwnerDeletedTx(ctx, tx, sourceObject, sourceRecordID, ownerDeleted)
}

func lockConnectorRemoteIdentityOwnerFenceTx(ctx context.Context, tx *sql.Tx, sourceObject, sourceRecordID string) (bool, error) {
	return persistence.LockConnectorRemoteIdentityOwnerFenceTx(ctx, tx, sourceObject, sourceRecordID)
}

func connectorRemoteIdentityTransferBoundaryLaneHash(sourceObject, sourceRecordID, remoteIDField string) (string, error) {
	sourceObject = strings.TrimSpace(sourceObject)
	sourceRecordID = strings.TrimSpace(sourceRecordID)
	remoteIDField = strings.TrimSpace(remoteIDField)
	if sourceObject == "" || sourceRecordID == "" || remoteIDField == "" {
		return "", fmt.Errorf("connector remote identity transfer boundary requires object, record, and remote id field")
	}
	sum := sha256.Sum256([]byte(strings.Join([]string{
		"connector-remote-identity-transfer-boundary-v1",
		sourceObject,
		sourceRecordID,
		remoteIDField,
	}, "\x00")))
	return hex.EncodeToString(sum[:]), nil
}

func (s *SQLConnectorPushDurableStore) LockInboundRemoteIdentityTransferBoundaryInAmbientTx(
	ctx context.Context,
	sourceObject, sourceRecordID, targetObject, targetRecordID, remoteIDField string,
) error {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return err
	}
	laneHashes := make([]string, 0, 2)
	for _, record := range [][2]string{{sourceObject, sourceRecordID}, {targetObject, targetRecordID}} {
		laneHash, err := connectorRemoteIdentityTransferBoundaryLaneHash(record[0], record[1], remoteIDField)
		if err != nil {
			return err
		}
		laneHashes = append(laneHashes, laneHash)
	}
	sort.Strings(laneHashes)
	for i, laneHash := range laneHashes {
		if i > 0 && laneHash == laneHashes[i-1] {
			continue
		}
		if err := lockConnectorRemoteIdentityLaneHashTx(ctx, tx, laneHash); err != nil {
			return err
		}
	}
	return nil
}

// CheckConnectorRemoteIdentityRegistrationInAmbientTx runs only after the
// caller owns the locator lane. No history permits a new evidence row. The
// sole accepted retry is one exact, active finalized link for the same durable
// identity. Retired history is legitimate transfer lineage and is deliberately
// ignored here; strict recovery applies its stronger all-history policy through
// CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx.
func (s *SQLConnectorPushDurableStore) CheckConnectorRemoteIdentityRegistrationInAmbientTx(
	ctx context.Context,
	identity ConnectorPushDurableIdentity,
	remoteID string,
) (bool, error) {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return false, err
	}
	return checkConnectorRemoteIdentityActiveAuthorityTx(ctx, tx, identity, remoteID, ConnectorPushLinkStatusFinalized)
}

func checkConnectorRemoteIdentityActiveAuthorityTx(
	ctx context.Context,
	tx *sql.Tx,
	identity ConnectorPushDurableIdentity,
	remoteID string,
	allowedStatus string,
) (bool, error) {
	linkKey, err := identity.LinkKey()
	if err != nil {
		return false, err
	}
	remoteID = strings.TrimSpace(remoteID)
	if remoteID == "" {
		return false, fmt.Errorf("connector remote identity registration remote id is required")
	}
	locatorHash := ConnectorPushRemoteLocatorHash(identity, remoteID)
	rows, err := tx.QueryContext(ctx, `
SELECT Id, LinkKey, Status, RemoteId, COALESCE(RemoteLocatorHash, ''),
       COALESCE(RemoteLocatorActiveKey, ''), ConnectorOwnerKey, ActionSlotId,
       ActionType, SourceObject, SourceRecordId, DestinationSlotId
  FROM _System_ConnectorRemoteLink
 WHERE RemoteLocatorActiveKey = ?
 LIMIT 1
 FOR UPDATE`, locatorHash)
	if err != nil {
		return false, err
	}
	defer rows.Close()
	count := 0
	var authority ConnectorRemoteIdentityAuthority
	var activeKey string
	for rows.Next() {
		count++
		if err := rows.Scan(
			&authority.LinkID, &authority.LinkKey, &authority.LinkStatus,
			&authority.RemoteID, &authority.RemoteLocatorHash, &activeKey,
			&authority.Identity.ConnectorOwnerKey, &authority.Identity.ActionSlotID,
			&authority.Identity.ActionType, &authority.Identity.SourceObject,
			&authority.Identity.SourceRecordID, &authority.Identity.DestinationSlotID,
		); err != nil {
			return false, err
		}
	}
	if err := rows.Err(); err != nil {
		return false, err
	}
	if count == 0 {
		return false, nil
	}
	if authority.LinkID == "" ||
		authority.LinkStatus != allowedStatus ||
		authority.LinkKey != linkKey ||
		authority.RemoteID != remoteID ||
		authority.RemoteLocatorHash != locatorHash ||
		activeKey != locatorHash ||
		authority.Identity != identity {
		return false, fmt.Errorf("%w: active remote locator is owned by another or incompatible identity", ErrConnectorPushDurableConflict)
	}
	return true, nil
}

// CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx is the strict
// pre-finalizer policy. Zero locator history permits a create only from active
// authored evidence. The sole retry state is one exact finalized authority plus
// the exact consumed evidence row. An exact owner-deleted pending link is the
// only recovery transition: the finalizer restores it to finalized after it
// revalidates that same consumed evidence. Retired, suspended, foreign,
// duplicate, or metadata-drifted history blocks before mutation.
func (s *SQLConnectorPushDurableStore) CheckConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
	ctx context.Context,
	finalization ConnectorInboundRemoteIdentityFinalization,
) (bool, error) {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return false, err
	}
	originalEvidence := finalization.Evidence
	normalized, _, err := normalizeConnectorInboundRemoteIdentityFinalization(finalization)
	if err != nil {
		return false, err
	}
	// Strict recovery subsequently locks its exact RemoteLink history. Take the
	// owner fence first, matching deletion quarantine's owner-fence -> LinkKey
	// order, so recovery cannot invert that order before finalization rechecks
	// the same fence.
	if err := ensureConnectorRemoteIdentityOwnerCreateAllowedTx(
		ctx,
		tx,
		normalized.Adoption.Identity.SourceObject,
		normalized.Adoption.Identity.SourceRecordID,
	); err != nil {
		return false, err
	}
	history, err := readConnectorRemoteIdentityRecoveryHistoryTx(ctx, tx, normalized)
	if err != nil {
		return false, err
	}
	if history == nil {
		if strings.TrimSpace(originalEvidence.EvidenceStatus) != ConnectorInboundEvidenceStatusAuthored ||
			strings.TrimSpace(originalEvidence.ConsumedByLinkKey) != "" ||
			strings.TrimSpace(originalEvidence.ActiveKey) == "" {
			return false, fmt.Errorf("%w: zero-history recovery requires exact active authored evidence", ErrConnectorPushDurableConflict)
		}
		return false, nil
	}
	if err := validateExactConnectorRemoteIdentityRecoveryHistory(history, normalized, true); err != nil {
		return false, err
	}
	if strings.TrimSpace(originalEvidence.EvidenceStatus) != ConnectorInboundEvidenceStatusConsumed ||
		strings.TrimSpace(originalEvidence.ConsumedByLinkKey) != normalized.Adoption.LinkKey ||
		strings.TrimSpace(originalEvidence.ActiveKey) != "" {
		return false, fmt.Errorf("%w: exact finalized recovery retry requires exact consumed evidence", ErrConnectorPushDurableConflict)
	}
	return true, nil
}

// AssertExactInboundRemoteIDEvidenceRecoveryInAmbientTx is the post-finalizer
// fan-in guard. It runs in the same transaction after the canonical finalizer
// has consumed the exact row and requires that row to be the sole authored or
// consumed candidate inside the installed binding allowlist.
func (s *SQLConnectorPushDurableStore) AssertExactInboundRemoteIDEvidenceRecoveryInAmbientTx(
	ctx context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	bindings []ConnectorInboundRemoteIDEvidenceBinding,
	evidenceID string,
	consumedByLinkKey string,
) error {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return err
	}
	evidence, err := s.findUniqueInboundRemoteIDEvidenceForRecovery(ctx, tx, selector, bindings, true)
	if err != nil {
		return err
	}
	if evidence == nil ||
		strings.TrimSpace(evidence.ID) != strings.TrimSpace(evidenceID) ||
		strings.TrimSpace(evidence.EvidenceStatus) != ConnectorInboundEvidenceStatusConsumed ||
		strings.TrimSpace(evidence.ConsumedByLinkKey) != strings.TrimSpace(consumedByLinkKey) {
		return fmt.Errorf("%w: exact consumed recovery evidence readback failed", ErrConnectorPushDurableConflict)
	}
	return nil
}

// AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx proves that the
// locator has exactly one history row and that it is the current exact
// finalized link. Any retired, suspended, foreign, duplicate, or metadata-
// drifted history blocks recovery and rolls the transaction back.
func (s *SQLConnectorPushDurableStore) AssertExactConnectorRemoteIdentityRecoveryHistoryInAmbientTx(
	ctx context.Context,
	finalization ConnectorInboundRemoteIdentityFinalization,
) error {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return err
	}
	normalized, _, err := normalizeConnectorInboundRemoteIdentityFinalization(finalization)
	if err != nil {
		return err
	}
	history, err := readConnectorRemoteIdentityRecoveryHistoryTx(ctx, tx, normalized)
	if err != nil {
		return err
	}
	if history == nil {
		return fmt.Errorf("%w: remote locator history is missing after recovery", ErrConnectorPushDurableConflict)
	}
	return validateExactConnectorRemoteIdentityRecoveryHistory(history, normalized, false)
}

func readConnectorRemoteIdentityRecoveryHistoryTx(
	ctx context.Context,
	tx *sql.Tx,
	finalization ConnectorInboundRemoteIdentityFinalization,
) (*connectorPushRecoveryHistoryRow, error) {
	// Search the union of the expected materialized hash and its unhashed logical
	// locator. Either side can be corrupt independently: a NULL/wrong hash must
	// still be found by the tuple, while an expected hash attached to a foreign
	// tuple must still be found by the hash. The persistent locator lane excludes
	// normal new writers; this bounded scan is only a corruption detector, not
	// phantom-based admission authority.
	locatorHash := ConnectorPushRemoteLocatorHash(finalization.Adoption.Identity, finalization.Adoption.RemoteID)
	rows, err := tx.QueryContext(ctx, connectorPushRecoveryHistorySelectPrefix+`
 WHERE RemoteLocatorHash = ?
    OR (ConnectorOwnerKey = ?
        AND DestinationSlotId = ?
        AND RemoteId = ?)
 ORDER BY Id
 LIMIT 2
 FOR UPDATE`,
		locatorHash,
		finalization.Adoption.Identity.ConnectorOwnerKey,
		finalization.Adoption.Identity.DestinationSlotID,
		finalization.Adoption.RemoteID,
	)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var history *connectorPushRecoveryHistoryRow
	for rows.Next() {
		if history != nil {
			return nil, fmt.Errorf("%w: remote locator has multiple history rows", ErrConnectorPushDurableConflict)
		}
		candidate, err := scanConnectorPushRecoveryHistory(rows)
		if err != nil {
			return nil, err
		}
		history = candidate
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return history, nil
}

func validateExactConnectorRemoteIdentityRecoveryHistory(
	history *connectorPushRecoveryHistoryRow,
	finalization ConnectorInboundRemoteIdentityFinalization,
	allowOwnerDeletedPendingRecovery bool,
) error {
	locatorHash := ConnectorPushRemoteLocatorHash(finalization.Adoption.Identity, finalization.Adoption.RemoteID)
	if history == nil ||
		(history.link.Status != ConnectorPushLinkStatusFinalized &&
			(!allowOwnerDeletedPendingRecovery || history.link.Status != ConnectorPushLinkStatusOwnerDeletedPendingRecovery)) ||
		history.link.LinkKey != finalization.Adoption.LinkKey ||
		history.link.RemoteID != finalization.Adoption.RemoteID ||
		history.link.Identity != finalization.Adoption.Identity ||
		history.remoteLocatorHash != locatorHash ||
		history.remoteLocatorActiveKey != locatorHash ||
		!connectorPushPersistedRemoteLinkSnapshotEqual(history.link.Snapshot, finalization.Adoption.Snapshot) ||
		history.compatibilityMirrorStatus != string(finalization.CompatibilityMirrorStatus) {
		return fmt.Errorf("%w: remote locator history is not the exact finalized recovery authority", ErrConnectorPushDurableConflict)
	}
	return nil
}

func (s *SQLConnectorPushDurableStore) findUniqueInboundRemoteIDEvidenceForRecovery(
	ctx context.Context,
	tx *sql.Tx,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
	bindings []ConnectorInboundRemoteIDEvidenceBinding,
	forUpdate bool,
) (*ConnectorInboundRemoteIDEvidence, error) {
	var err error
	selector, err = normalizeConnectorInboundRemoteIdentityRecoverySelector(selector)
	if err != nil {
		return nil, err
	}
	if len(bindings) == 0 {
		return nil, fmt.Errorf("connector inbound remote identity recovery requires at least one installed evidence binding")
	}
	normalized := make([]ConnectorInboundRemoteIDEvidenceBinding, 0, len(bindings))
	seen := make(map[string]struct{}, len(bindings))
	for _, binding := range bindings {
		binding, err = normalizeConnectorInboundRemoteIDEvidenceBinding(binding)
		if err != nil {
			return nil, err
		}
		key := connectorInboundRemoteIDEvidenceBindingKey(binding)
		if _, exists := seen[key]; exists {
			return nil, fmt.Errorf("connector inbound remote identity recovery has duplicate evidence binding %s/%s", binding.SourceObject, binding.RemoteIDField)
		}
		seen[key] = struct{}{}
		normalized = append(normalized, binding)
	}
	sort.Slice(normalized, func(i, j int) bool {
		return connectorInboundRemoteIDEvidenceBindingKey(normalized[i]) < connectorInboundRemoteIDEvidenceBindingKey(normalized[j])
	})

	var candidate *ConnectorInboundRemoteIDEvidence
	for _, binding := range normalized {
		query := `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus,
       COALESCE(ConsumedByLinkKey, ''), COALESCE(ActiveKey, ''),
       CreatedDate, LastModifiedDate
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE ConnectorOwnerKey = ?
   AND ConnectorNamespace = ?
   AND SourceObject = ?
   AND RemoteIDField = ?
   AND RemoteId = ?
   AND EvidenceStatus IN (?, ?)
 LIMIT 2`
		if forUpdate {
			query += " FOR UPDATE"
		}
		rows, err := tx.QueryContext(ctx, query,
			selector.ConnectorOwnerKey,
			selector.ConnectorNamespace,
			binding.SourceObject,
			binding.RemoteIDField,
			selector.RemoteID,
			ConnectorInboundEvidenceStatusAuthored,
			ConnectorInboundEvidenceStatusConsumed,
		)
		if err != nil {
			return nil, err
		}
		for rows.Next() {
			evidence, err := scanConnectorInboundEvidence(rows)
			if err != nil {
				_ = rows.Close()
				return nil, err
			}
			if candidate != nil {
				_ = rows.Close()
				return nil, fmt.Errorf("%w: multiple authored or consumed evidence rows match installed recovery bindings", ErrConnectorPushDurableConflict)
			}
			candidate = evidence
		}
		if err := rows.Err(); err != nil {
			_ = rows.Close()
			return nil, err
		}
		if err := rows.Close(); err != nil {
			return nil, err
		}
	}
	if candidate == nil {
		return nil, nil
	}
	if err := validateConnectorInboundRemoteIdentityRecoveryEvidence(selector, *candidate); err != nil {
		return nil, err
	}
	return candidate, nil
}

func (s *SQLConnectorPushDurableStore) connectorInboundRecoveryAmbientTx(ctx context.Context) (*sql.Tx, error) {
	if s == nil || s.db == nil {
		return nil, fmt.Errorf("connector push durable store is not wired")
	}
	if s.txExtractor == nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery requires a live ambient transaction")
	}
	tx, ok := s.txExtractor.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return nil, fmt.Errorf("connector inbound remote identity recovery requires a live ambient transaction")
	}
	return tx, nil
}

func (s *SQLConnectorPushDurableStore) findActiveInboundRemoteIDEvidence(ctx context.Context, q connectorPushQuerier, connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField string) (*ConnectorInboundRemoteIDEvidence, error) {
	activeKey := ConnectorInboundEvidenceActiveKey(connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField)
	row := q.QueryRowContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus,
       COALESCE(ConsumedByLinkKey, ''), COALESCE(ActiveKey, ''),
       CreatedDate, LastModifiedDate
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE ActiveKey = ?
 LIMIT 1`, activeKey)
	evidence, err := scanConnectorInboundEvidence(row)
	if errors.Is(err, sql.ErrNoRows) {
		return nil, nil
	}
	return evidence, err
}

func (s *SQLConnectorPushDurableStore) FindInboundRemoteIdentityTransferHintsInAmbientTx(
	ctx context.Context,
	sourceObject, sourceRecordID, targetObject, targetRecordID, remoteIDField string,
) ([]ConnectorRemoteIdentityTransferHint, error) {
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return nil, err
	}
	sourceObject = strings.TrimSpace(sourceObject)
	sourceRecordID = strings.TrimSpace(sourceRecordID)
	targetObject = strings.TrimSpace(targetObject)
	targetRecordID = strings.TrimSpace(targetRecordID)
	remoteIDField = strings.TrimSpace(remoteIDField)
	if sourceObject == "" || sourceRecordID == "" || targetObject == "" || targetRecordID == "" || remoteIDField == "" {
		return nil, fmt.Errorf("connector remote identity transfer hints require source, target, and remote id field")
	}

	type hintKey struct{ owner, remote, blockingLinkKey string }
	type mergedHint struct {
		value                 ConnectorRemoteIdentityTransferHint
		namespaceFromEvidence bool
	}
	hints := map[hintKey]mergedHint{}
	merge := func(candidate ConnectorRemoteIdentityTransferHint, fromEvidence bool) error {
		candidate.ConnectorOwnerKey = strings.TrimSpace(candidate.ConnectorOwnerKey)
		candidate.ConnectorNamespace = strings.TrimSpace(candidate.ConnectorNamespace)
		candidate.RemoteID = strings.TrimSpace(candidate.RemoteID)
		candidate.DestinationSlotID = strings.TrimSpace(candidate.DestinationSlotID)
		candidate.BlockingAttemptLinkKey = strings.TrimSpace(candidate.BlockingAttemptLinkKey)
		candidate.BlockingAttemptOccupancyKey = strings.TrimSpace(candidate.BlockingAttemptOccupancyKey)
		blockingLinkKey := ""
		if candidate.BlockingAttemptIdentity != (ConnectorPushDurableIdentity{}) {
			candidate.BlockingAttemptIdentity.ConnectorOwnerKey = strings.TrimSpace(candidate.BlockingAttemptIdentity.ConnectorOwnerKey)
			candidate.BlockingAttemptIdentity.ActionSlotID = strings.TrimSpace(candidate.BlockingAttemptIdentity.ActionSlotID)
			candidate.BlockingAttemptIdentity.ActionType = strings.TrimSpace(candidate.BlockingAttemptIdentity.ActionType)
			candidate.BlockingAttemptIdentity.SourceObject = strings.TrimSpace(candidate.BlockingAttemptIdentity.SourceObject)
			candidate.BlockingAttemptIdentity.SourceRecordID = strings.TrimSpace(candidate.BlockingAttemptIdentity.SourceRecordID)
			candidate.BlockingAttemptIdentity.DestinationSlotID = strings.TrimSpace(candidate.BlockingAttemptIdentity.DestinationSlotID)
			var err error
			blockingLinkKey, err = candidate.BlockingAttemptIdentity.LinkKey()
			if err != nil {
				return fmt.Errorf("%w: unresolved create attempt hint identity: %v", ErrConnectorPushDurableConflict, err)
			}
			if candidate.BlockingAttemptIdentity.ConnectorOwnerKey != candidate.ConnectorOwnerKey ||
				candidate.BlockingAttemptIdentity.DestinationSlotID != candidate.DestinationSlotID {
				return fmt.Errorf("%w: unresolved create attempt hint dimensions disagree", ErrConnectorPushDurableConflict)
			}
			if candidate.BlockingAttemptLinkKey != blockingLinkKey || candidate.BlockingAttemptOccupancyKey != blockingLinkKey {
				return fmt.Errorf("%w: unresolved create attempt key materialization disagrees with its durable identity", ErrConnectorPushDurableConflict)
			}
		}
		if candidate.ConnectorOwnerKey == "" || candidate.ConnectorNamespace == "" ||
			(candidate.RemoteID == "" && blockingLinkKey == "") {
			return fmt.Errorf("%w: connector remote identity transfer hint is incomplete", ErrConnectorPushDurableConflict)
		}
		key := hintKey{owner: candidate.ConnectorOwnerKey, remote: candidate.RemoteID, blockingLinkKey: blockingLinkKey}
		if existing, ok := hints[key]; ok {
			if existing.namespaceFromEvidence && fromEvidence &&
				!strings.EqualFold(existing.value.ConnectorNamespace, candidate.ConnectorNamespace) {
				return fmt.Errorf("%w: connector remote identity transfer hints disagree", ErrConnectorPushDurableConflict)
			}
			if !existing.namespaceFromEvidence && !fromEvidence &&
				!strings.EqualFold(existing.value.ConnectorNamespace, candidate.ConnectorNamespace) {
				return fmt.Errorf("%w: connector remote identity transfer link hints disagree", ErrConnectorPushDurableConflict)
			}
			if existing.value.DestinationSlotID != "" && candidate.DestinationSlotID != "" &&
				existing.value.DestinationSlotID != candidate.DestinationSlotID {
				return fmt.Errorf("%w: connector remote identity transfer hints disagree", ErrConnectorPushDurableConflict)
			}
			if fromEvidence {
				existing.value.ConnectorNamespace = candidate.ConnectorNamespace
				existing.namespaceFromEvidence = true
			}
			if existing.value.DestinationSlotID == "" {
				existing.value.DestinationSlotID = candidate.DestinationSlotID
			}
			hints[key] = existing
			return nil
		}
		hints[key] = mergedHint{value: candidate, namespaceFromEvidence: fromEvidence}
		return nil
	}

	readEvidenceHints := func(q connectorPushQuerier) error {
		evidenceRows, err := q.QueryContext(ctx, `
SELECT ConnectorOwnerKey, ConnectorNamespace, RemoteId
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE (
        (SourceObject = ? AND SourceRecordId = ?)
        OR
        (SourceObject = ? AND SourceRecordId = ?)
       )
   AND RemoteIDField = ?
   AND EvidenceStatus = ?`,
			sourceObject,
			sourceRecordID,
			targetObject,
			targetRecordID,
			remoteIDField,
			ConnectorInboundEvidenceStatusAuthored,
		)
		if err != nil {
			return err
		}
		for evidenceRows.Next() {
			var hint ConnectorRemoteIdentityTransferHint
			if err := evidenceRows.Scan(&hint.ConnectorOwnerKey, &hint.ConnectorNamespace, &hint.RemoteID); err != nil {
				_ = evidenceRows.Close()
				return err
			}
			if err := merge(hint, true); err != nil {
				_ = evidenceRows.Close()
				return err
			}
		}
		if err := evidenceRows.Err(); err != nil {
			_ = evidenceRows.Close()
			return err
		}
		return evidenceRows.Close()
	}
	if err := readEvidenceHints(tx); err != nil {
		return nil, err
	}
	// Preserve ambient read-your-writes while also closing the repeatable-read
	// gap for evidence committed by a record writer that won before the transfer
	// acquired the corresponding record/boundary lock.
	if err := readEvidenceHints(s.db); err != nil {
		return nil, err
	}

	readLinkHints := func(q connectorPushQuerier, object, recordID string) error {
		linkRows, err := q.QueryContext(ctx, `
SELECT ConnectorOwnerKey, ConnectorNamespace, RemoteId, DestinationSlotId
  FROM _System_ConnectorRemoteLink
 WHERE SourceObject = ?
   AND SourceRecordId = ?
   AND COALESCE(RemoteIdField, '') = ?
   AND Status IN (?, ?, ?)`,
			object,
			recordID,
			remoteIDField,
			ConnectorPushLinkStatusFinalized,
			ConnectorPushLinkStatusSuspendedDrift,
			ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
		)
		if err != nil {
			return err
		}
		for linkRows.Next() {
			var hint ConnectorRemoteIdentityTransferHint
			if err := linkRows.Scan(&hint.ConnectorOwnerKey, &hint.ConnectorNamespace, &hint.RemoteID, &hint.DestinationSlotID); err != nil {
				_ = linkRows.Close()
				return err
			}
			if err := merge(hint, false); err != nil {
				_ = linkRows.Close()
				return err
			}
		}
		if err := linkRows.Err(); err != nil {
			_ = linkRows.Close()
			return err
		}
		return linkRows.Close()
	}
	for _, boundary := range [][2]string{
		{sourceObject, sourceRecordID},
		{targetObject, targetRecordID},
	} {
		if err := readLinkHints(tx, boundary[0], boundary[1]); err != nil {
			return nil, err
		}
	}
	// The ambient transaction supplies read-your-writes, while this latest
	// committed pass closes the repeatable-read gap when an RPR transaction won
	// the boundary lane before Transfer. The shared boundary prevents another
	// attempt-to-link publication between this pass and the attempt read below.
	for _, boundary := range [][2]string{
		{sourceObject, sourceRecordID},
		{targetObject, targetRecordID},
	} {
		if err := readLinkHints(s.db, boundary[0], boundary[1]); err != nil {
			return nil, err
		}
	}

	// AttemptOccupancyKey is the writer-owned unresolved-create sentinel. These
	// rows are hints only: Transfer revalidates them under the canonical
	// LinkKey -> AttemptOccupancyKey lock order before they can affect authority.
	// Create reservations commit in their own durable-store transaction. Read
	// them from the latest committed state after the ambient transfer boundary
	// lanes are held; using the ambient transaction's earlier repeatable-read
	// snapshot could miss a reservation that committed while record locks were
	// being acquired.
	attemptRows, err := s.db.QueryContext(ctx, `
SELECT ConnectorOwnerKey, ConnectorNamespace, COALESCE(RemoteId, ''),
       DestinationSlotId, ActionSlotId, ActionType, SourceObject, SourceRecordId,
       LinkKey, AttemptOccupancyKey
  FROM _System_ConnectorPushAttempt
 WHERE (
        (SourceObject = ? AND SourceRecordId = ?)
        OR
        (SourceObject = ? AND SourceRecordId = ?)
       )
   AND COALESCE(RemoteIdField, '') = ?
   AND AttemptOccupancyKey IS NOT NULL`,
		sourceObject,
		sourceRecordID,
		targetObject,
		targetRecordID,
		remoteIDField,
	)
	if err != nil {
		return nil, err
	}
	for attemptRows.Next() {
		var hint ConnectorRemoteIdentityTransferHint
		if err := attemptRows.Scan(
			&hint.ConnectorOwnerKey,
			&hint.ConnectorNamespace,
			&hint.RemoteID,
			&hint.DestinationSlotID,
			&hint.BlockingAttemptIdentity.ActionSlotID,
			&hint.BlockingAttemptIdentity.ActionType,
			&hint.BlockingAttemptIdentity.SourceObject,
			&hint.BlockingAttemptIdentity.SourceRecordID,
			&hint.BlockingAttemptLinkKey,
			&hint.BlockingAttemptOccupancyKey,
		); err != nil {
			_ = attemptRows.Close()
			return nil, err
		}
		hint.BlockingAttemptIdentity.ConnectorOwnerKey = hint.ConnectorOwnerKey
		hint.BlockingAttemptIdentity.DestinationSlotID = hint.DestinationSlotID
		if err := merge(hint, false); err != nil {
			_ = attemptRows.Close()
			return nil, err
		}
	}
	if err := attemptRows.Err(); err != nil {
		_ = attemptRows.Close()
		return nil, err
	}
	if err := attemptRows.Close(); err != nil {
		return nil, err
	}
	result := make([]ConnectorRemoteIdentityTransferHint, 0, len(hints))
	for _, hint := range hints {
		result = append(result, hint.value)
	}
	sort.Slice(result, func(i, j int) bool {
		leftBlockingLinkKey, _ := result[i].BlockingAttemptIdentity.LinkKey()
		rightBlockingLinkKey, _ := result[j].BlockingAttemptIdentity.LinkKey()
		left := result[i].ConnectorOwnerKey + "\x00" + result[i].RemoteID + "\x00" + leftBlockingLinkKey
		right := result[j].ConnectorOwnerKey + "\x00" + result[j].RemoteID + "\x00" + rightBlockingLinkKey
		return left < right
	})
	return result, nil
}

func normalizeConnectorRemoteIdentityTransferAdmissions(
	transfer ConnectorRemoteIdentityTransfer,
) ([]connectorRemoteIdentityTransferAdmission, map[string]connectorRemoteIdentityTransferAdmission, map[string]struct{}, error) {
	admissions := make([]connectorRemoteIdentityTransferAdmission, 0, len(transfer.admissions))
	byOwner := make(map[string]connectorRemoteIdentityTransferAdmission, len(transfer.admissions))
	locatorHashes := make(map[string]struct{}, len(transfer.admissions))
	for _, candidate := range transfer.admissions {
		candidate.ConnectorNamespace = strings.TrimSpace(candidate.ConnectorNamespace)
		candidate.RemoteID = strings.TrimSpace(candidate.RemoteID)
		candidate.SourceIdentity.ConnectorOwnerKey = strings.TrimSpace(candidate.SourceIdentity.ConnectorOwnerKey)
		candidate.SourceIdentity.ActionSlotID = strings.TrimSpace(candidate.SourceIdentity.ActionSlotID)
		candidate.SourceIdentity.ActionType = strings.TrimSpace(candidate.SourceIdentity.ActionType)
		candidate.SourceIdentity.SourceObject = strings.TrimSpace(candidate.SourceIdentity.SourceObject)
		candidate.SourceIdentity.SourceRecordID = strings.TrimSpace(candidate.SourceIdentity.SourceRecordID)
		candidate.SourceIdentity.DestinationSlotID = strings.TrimSpace(candidate.SourceIdentity.DestinationSlotID)
		candidate.TargetIdentity.ConnectorOwnerKey = strings.TrimSpace(candidate.TargetIdentity.ConnectorOwnerKey)
		candidate.TargetIdentity.ActionSlotID = strings.TrimSpace(candidate.TargetIdentity.ActionSlotID)
		candidate.TargetIdentity.ActionType = strings.TrimSpace(candidate.TargetIdentity.ActionType)
		candidate.TargetIdentity.SourceObject = strings.TrimSpace(candidate.TargetIdentity.SourceObject)
		candidate.TargetIdentity.SourceRecordID = strings.TrimSpace(candidate.TargetIdentity.SourceRecordID)
		candidate.TargetIdentity.DestinationSlotID = strings.TrimSpace(candidate.TargetIdentity.DestinationSlotID)
		if candidate.ConnectorNamespace == "" || (candidate.RemoteID == "" && !candidate.UnresolvedCreateBlock) {
			return nil, nil, nil, fmt.Errorf("connector remote identity transfer admission is incomplete")
		}
		if _, err := candidate.SourceIdentity.LinkKey(); err != nil {
			return nil, nil, nil, fmt.Errorf("connector remote identity transfer source admission: %w", err)
		}
		if _, err := candidate.TargetIdentity.LinkKey(); err != nil {
			return nil, nil, nil, fmt.Errorf("connector remote identity transfer target admission: %w", err)
		}
		if candidate.SourceIdentity.ConnectorOwnerKey != candidate.TargetIdentity.ConnectorOwnerKey ||
			candidate.SourceIdentity.DestinationSlotID != candidate.TargetIdentity.DestinationSlotID ||
			candidate.SourceIdentity.SourceObject != transfer.SourceObject ||
			candidate.SourceIdentity.SourceRecordID != transfer.SourceRecordId ||
			candidate.TargetIdentity.SourceObject != transfer.TargetObject ||
			candidate.TargetIdentity.SourceRecordID != transfer.TargetRecordId {
			return nil, nil, nil, fmt.Errorf("%w: connector remote identity transfer admission does not match the protected record boundary", ErrConnectorPushDurableConflict)
		}
		if candidate.RemoteID != "" && transfer.RemoteId != "" && transfer.RemoteId != candidate.RemoteID {
			return nil, nil, nil, fmt.Errorf("%w: connector remote identity transfer admissions disagree on remote id", ErrConnectorPushDurableConflict)
		}
		if candidate.RemoteID != "" && transfer.TargetRemoteID != "" && transfer.TargetRemoteID != candidate.RemoteID {
			return nil, nil, nil, fmt.Errorf("%w: target mirror does not match installed transfer admission", ErrConnectorPushDurableConflict)
		}
		if candidate.TargetSnapshot.RequesterOwnerPack != candidate.TargetIdentity.ConnectorOwnerKey ||
			candidate.TargetSnapshot.RemoteIDField != transfer.RemoteIDField ||
			strings.TrimSpace(candidate.TargetSnapshot.ConnectorNamespace) == "" {
			return nil, nil, nil, fmt.Errorf("%w: target snapshot does not match the installed transfer admission", ErrConnectorPushDurableConflict)
		}
		if candidate.RemoteID != "" {
			adoption, err := normalizeConnectorPushLinkAdoption(ConnectorPushLinkAdoption{
				Identity: candidate.TargetIdentity, RemoteID: candidate.RemoteID, Snapshot: candidate.TargetSnapshot,
			})
			if err != nil {
				return nil, nil, nil, fmt.Errorf("%w: target transfer snapshot: %v", ErrConnectorPushDurableConflict, err)
			}
			candidate.TargetSnapshot = adoption.Snapshot
		}
		owner := candidate.SourceIdentity.ConnectorOwnerKey
		if existing, ok := byOwner[owner]; ok {
			if existing.ConnectorNamespace != candidate.ConnectorNamespace || existing.RemoteID != candidate.RemoteID ||
				existing.SourceIdentity != candidate.SourceIdentity || existing.TargetIdentity != candidate.TargetIdentity ||
				existing.SourceIntegrationScope != candidate.SourceIntegrationScope || existing.TargetIntegrationScope != candidate.TargetIntegrationScope ||
				existing.UnresolvedCreateBlock != candidate.UnresolvedCreateBlock ||
				!connectorPushPersistedRemoteLinkSnapshotEqual(existing.TargetSnapshot, candidate.TargetSnapshot) {
				return nil, nil, nil, fmt.Errorf("%w: connector remote identity transfer has ambiguous installed admissions for owner %s", ErrConnectorPushDurableConflict, owner)
			}
			continue
		}
		byOwner[owner] = candidate
		if candidate.RemoteID != "" {
			locatorHashes[ConnectorPushRemoteLocatorHash(candidate.SourceIdentity, candidate.RemoteID)] = struct{}{}
		}
		admissions = append(admissions, candidate)
	}
	sort.Slice(admissions, func(i, j int) bool {
		leftLinkKey, _ := admissions[i].SourceIdentity.LinkKey()
		rightLinkKey, _ := admissions[j].SourceIdentity.LinkKey()
		left := admissions[i].RemoteID + "\x00" + leftLinkKey
		right := admissions[j].RemoteID + "\x00" + rightLinkKey
		return left < right
	})
	return admissions, byOwner, locatorHashes, nil
}

func (s *SQLConnectorPushDurableStore) lockConnectorRemoteIdentityTransferLinkAttemptsTx(
	ctx context.Context,
	tx *sql.Tx,
	admissions []connectorRemoteIdentityTransferAdmission,
) error {
	keys := make([]string, 0, len(admissions)*2)
	seen := make(map[string]struct{}, len(admissions)*2)
	for _, admission := range admissions {
		for _, identity := range []ConnectorPushDurableIdentity{admission.SourceIdentity, admission.TargetIdentity} {
			linkKey, err := identity.LinkKey()
			if err != nil {
				return err
			}
			if _, ok := seen[linkKey]; ok {
				continue
			}
			seen[linkKey] = struct{}{}
			keys = append(keys, linkKey)
		}
	}
	sort.Strings(keys)
	for _, linkKey := range keys {
		lockState, err := s.lockConnectorPushLinkThenAttemptOccupancyTx(ctx, tx, linkKey)
		if err != nil {
			return err
		}
		if strings.TrimSpace(lockState.AttemptID) != "" {
			return fmt.Errorf("%w: connector remote identity transfer link has an unresolved create attempt", ErrConnectorPushDurableConflict)
		}
	}
	return nil
}

const (
	connectorIntegrationLinkV2Prefix = "integration_link_v2:"
)

type connectorIntegrationOutboundScope struct {
	orgID, appInstallationID, authorizedWorkloadID string
}

func connectorIntegrationContractScope(scope connectorIntegrationOutboundScope) outboundintentv1.LogicalEffectScope {
	return outboundintentv1.LogicalEffectScope{OrgID: scope.orgID, AppInstallationID: scope.appInstallationID, AuthorizedWorkloadID: scope.authorizedWorkloadID}
}

func connectorIntegrationOutboundLinkKey(scope connectorIntegrationOutboundScope, identity ConnectorPushDurableIdentity) string {
	return outboundintentv1.IntegrationLinkKeyV2(connectorIntegrationContractScope(scope), identity.ConnectorOwnerKey, identity.ActionSlotID, identity.SourceObject, identity.SourceRecordID, identity.DestinationSlotID)
}

func connectorIntegrationOutboundRemoteLocatorHash(scope connectorIntegrationOutboundScope, identity ConnectorPushDurableIdentity, remoteID string) string {
	return outboundintentv1.IntegrationRemoteLocatorHash(connectorIntegrationContractScope(scope), identity.ConnectorOwnerKey, identity.DestinationSlotID, remoteID)
}

func connectorRuntimeOutboundIdentity(identity ConnectorPushDurableIdentity) persistence.RuntimeOutboundIdentity {
	return persistence.RuntimeOutboundIdentity{SourceObject: identity.SourceObject, SourceRecordID: identity.SourceRecordID,
		ConnectorOwnerKey: identity.ConnectorOwnerKey, ActionSlotID: identity.ActionSlotID, DestinationSlotID: identity.DestinationSlotID}
}

// V2 identity proof is the Runtime-owned completed result receipt. Integration
// publishes its provider result before this receipt commits, and a receipt is
// retained with identity history independently of the private delivery ledger.
func (s *SQLConnectorPushDurableStore) validateConnectorIntegrationOutboundSourceLink(
	ctx context.Context, linkKey string, identity ConnectorPushDurableIdentity, remoteID string,
	expectedScope connectorIntegrationOutboundScope,
) (string, error) {
	if expectedScope == (connectorIntegrationOutboundScope{}) || connectorIntegrationOutboundLinkKey(expectedScope, identity) != linkKey {
		return "", fmt.Errorf("%w: scoped integration link is outside its installed scope", ErrConnectorPushDurableConflict)
	}
	return s.validateConnectorRuntimeOutboundReceipts(ctx, identity, remoteID, expectedScope, true)
}

// A finalized legacy link supplies its own remote identity. The Runtime intent
// owner still has to prove no accepted or pending effect can outlive mutation.
func (s *SQLConnectorPushDurableStore) validateConnectorIntegrationV1BridgeQuiescent(
	ctx context.Context, identity ConnectorPushDurableIdentity, remoteID string, scope connectorIntegrationOutboundScope,
) error {
	_, err := s.validateConnectorRuntimeOutboundReceipts(ctx, identity, remoteID, scope, false)
	return err
}

func (s *SQLConnectorPushDurableStore) validateConnectorRuntimeOutboundReceipts(
	ctx context.Context, identity ConnectorPushDurableIdentity, remoteID string,
	expectedScope connectorIntegrationOutboundScope, requireTerminalProof bool,
) (string, error) {
	if identity.ActionType != "PushToConnector" {
		return "", fmt.Errorf("%w: unsupported identity action", ErrConnectorPushDurableConflict)
	}
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return "", err
	}
	rows, err := persistence.RuntimeOutboundIdentityReceiptsTx(ctx, tx, connectorRuntimeOutboundIdentity(identity))
	if err != nil {
		return "", fmt.Errorf("%w: %w", ErrConnectorPushDurableConflict, err)
	}
	var provedLocator string
	for _, row := range rows {
		scope := connectorIntegrationOutboundScope{row.Scope.OrgID, row.Scope.AppInstallationID, row.Scope.AuthorizedWorkloadID}
		if scope != expectedScope {
			return "", fmt.Errorf("%w: source intent is outside its independently installed scope", ErrConnectorPushDurableConflict)
		}
		switch row.AdmissionState {
		case persistence.RuntimeOutboundIntentPending, persistence.RuntimeOutboundIntentAccepted:
			return "", apperrors.NewLockContentionError(fmt.Errorf("%w: source intent has no terminal receipt", ErrConnectorPushDurableConflict))
		case persistence.RuntimeOutboundIntentRejected:
			continue
		case persistence.RuntimeOutboundIntentCompleted:
		default:
			return "", fmt.Errorf("%w: source intent has an unknown admission state", ErrConnectorPushDurableConflict)
		}
		result := outboundintentv1.CommitResultRequest{IntentID: row.Intent.IntentID, PayloadDigest: row.Intent.PayloadDigest,
			CatalogRevision: row.Intent.CatalogRevision, IntegrationCommandID: row.IntegrationCommandID,
			RemoteID: row.ResultRemoteID, ResultFormatVersion: row.ResultFormatVersion,
			ResultDigest: row.ResultDigest, CanonicalResult: row.CanonicalResult}
		if err := outboundintentv1.ValidateCommitResultRequestForIntent(row.Intent, result); err != nil {
			return "", fmt.Errorf("%w: invalid source completion receipt", ErrConnectorPushDurableConflict)
		}
		if row.ResultFormatVersion == outboundintentv1.ResourceNoEffectResultFormatV1 {
			continue
		}
		if row.Intent.Operation != outboundintentv1.OperationUpsert || row.ResultRemoteID == "" || row.ResultRemoteID != remoteID {
			return "", fmt.Errorf("%w: source result does not match remote identity", ErrConnectorPushDurableConflict)
		}
		provedLocator = connectorIntegrationOutboundRemoteLocatorHash(expectedScope, identity, remoteID)
	}
	if requireTerminalProof && provedLocator == "" {
		return "", fmt.Errorf("%w: scoped integration link has no Runtime completion proof", ErrConnectorPushDurableConflict)
	}
	return provedLocator, nil
}

// TransferInboundRemoteIdentity retires the source record's proven evidence and
// links, then publishes the installed target identity as a finalized link in the
// same ambient transaction. Conversion is not an inbound write and must not
// manufacture target authorship evidence or infer its connector namespace from
// a source link. The caller owns the protected atomic conversion and its mirror
// validation. No transferable authority returns an empty remote id.
func (s *SQLConnectorPushDurableStore) TransferInboundRemoteIdentity(ctx context.Context, transfer ConnectorRemoteIdentityTransfer) (string, error) {
	if s == nil || s.db == nil {
		return "", fmt.Errorf("connector push durable store is not wired")
	}
	transfer.SourceObject = strings.TrimSpace(transfer.SourceObject)
	transfer.SourceRecordId = strings.TrimSpace(transfer.SourceRecordId)
	transfer.TargetObject = strings.TrimSpace(transfer.TargetObject)
	transfer.TargetRecordId = strings.TrimSpace(transfer.TargetRecordId)
	transfer.RemoteIDField = strings.TrimSpace(transfer.RemoteIDField)
	transfer.RemoteId = strings.TrimSpace(transfer.RemoteId)
	transfer.TargetRemoteID = strings.TrimSpace(transfer.TargetRemoteID)
	if transfer.SourceObject == "" || transfer.SourceRecordId == "" || transfer.TargetObject == "" || transfer.TargetRecordId == "" || transfer.RemoteIDField == "" {
		return "", fmt.Errorf("connector remote identity transfer: SourceObject, SourceRecordId, TargetObject, TargetRecordId, and RemoteIDField are required")
	}

	querier := s.connectorPushAmbientQuerier(ctx)
	execer := s.connectorPushAmbientExecer(ctx)
	tx, err := s.connectorInboundRecoveryAmbientTx(ctx)
	if err != nil {
		return "", fmt.Errorf("connector remote identity transfer admission: %w", err)
	}
	admissions, admissionByOwner, lockedLocatorHashes, err := normalizeConnectorRemoteIdentityTransferAdmissions(transfer)
	if err != nil {
		return "", err
	}
	for _, admission := range admissions {
		if admission.RemoteID == "" {
			continue
		}
		if err := lockConnectorRemoteIdentityLaneTx(ctx, tx, admission.SourceIdentity, admission.RemoteID); err != nil {
			return "", err
		}
	}
	// Hold both owner fences in canonical order. Current NOWAIT intent reads
	// reject concurrent result commits rather than reversing their record lock.
	owners := [][2]string{{transfer.SourceObject, transfer.SourceRecordId}, {transfer.TargetObject, transfer.TargetRecordId}}
	sort.Slice(owners, func(i, j int) bool {
		return strings.ToLower(owners[i][0])+"\x00"+owners[i][1] < strings.ToLower(owners[j][0])+"\x00"+owners[j][1]
	})
	for _, owner := range owners {
		if err := persistence.CheckRuntimeOutboundOwnerQuiescentTx(ctx, tx, owner[0], owner[1]); err != nil {
			return "", fmt.Errorf("%w: %w", ErrConnectorPushDurableConflict, err)
		}
	}
	// Match adoption's locator -> owner fence -> LinkKey -> Attempt order.
	// A delayed conversion cannot recreate authority for a deleted target owner.
	if len(admissions) != 0 {
		if err := ensureConnectorRemoteIdentityOwnerCreateAllowedTx(ctx, tx, transfer.TargetObject, transfer.TargetRecordId); err != nil {
			return "", err
		}
	}
	if err := s.lockConnectorRemoteIdentityTransferLinkAttemptsTx(ctx, tx, admissions); err != nil {
		return "", err
	}

	rows, err := querier.QueryContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, SourceObject, SourceRecordId,
       RemoteIDField, RemoteId, EvidenceStatus,
       COALESCE(ConsumedByLinkKey, ''), COALESCE(ActiveKey, ''),
       CreatedDate, LastModifiedDate
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE SourceObject = ?
   AND SourceRecordId = ?
   AND RemoteIDField = ?
   AND EvidenceStatus = ?
 FOR UPDATE`,
		transfer.SourceObject,
		transfer.SourceRecordId,
		transfer.RemoteIDField,
		ConnectorInboundEvidenceStatusAuthored,
	)
	if err != nil {
		return "", err
	}
	defer rows.Close()

	type transferAuthority struct {
		connectorOwnerKey     string
		connectorNamespace    string
		namespaceFromEvidence bool
		evidenceID            string
		sourceLinkKeys        map[string]struct{}
		remoteID              string
	}

	authorities := map[string]transferAuthority{}
	remoteID := transfer.RemoteId
	addAuthority := func(authority transferAuthority) error {
		authority.connectorOwnerKey = strings.TrimSpace(authority.connectorOwnerKey)
		authority.connectorNamespace = strings.TrimSpace(authority.connectorNamespace)
		authority.remoteID = strings.TrimSpace(authority.remoteID)
		if authority.connectorOwnerKey == "" {
			return fmt.Errorf("connector remote identity transfer: source authority has blank connector owner")
		}
		if authority.connectorNamespace == "" {
			return fmt.Errorf("connector remote identity transfer: source authority has blank connector namespace")
		}
		if authority.remoteID == "" {
			return fmt.Errorf("connector remote identity transfer: source authority has blank remote id")
		}
		admission, admitted := admissionByOwner[authority.connectorOwnerKey]
		if !admitted || strings.TrimSpace(admission.RemoteID) != authority.remoteID {
			return fmt.Errorf("%w: source authority is outside the engine-derived installed transfer admissions", ErrConnectorPushDurableConflict)
		}
		existing, hasExisting := authorities[authority.connectorOwnerKey]
		if authority.namespaceFromEvidence &&
			!strings.EqualFold(strings.TrimSpace(admission.ConnectorNamespace), authority.connectorNamespace) {
			return fmt.Errorf("%w: authored evidence namespace is outside the engine-derived installed transfer admission", ErrConnectorPushDurableConflict)
		}
		if !authority.namespaceFromEvidence && (!hasExisting || !existing.namespaceFromEvidence) &&
			!strings.EqualFold(strings.TrimSpace(admission.ConnectorNamespace), authority.connectorNamespace) {
			return fmt.Errorf("%w: active source link namespace is outside the engine-derived installed transfer admission", ErrConnectorPushDurableConflict)
		}
		if remoteID == "" {
			remoteID = authority.remoteID
		}
		if remoteID != authority.remoteID {
			return fmt.Errorf("%w: source remote identity has conflicting remote ids for transfer", ErrConnectorPushDurableConflict)
		}
		if hasExisting {
			if existing.remoteID != authority.remoteID {
				return fmt.Errorf("%w: source remote identity has conflicting remote ids for owner %s", ErrConnectorPushDurableConflict, authority.connectorOwnerKey)
			}
			if existing.evidenceID != "" && authority.evidenceID != "" {
				return fmt.Errorf("%w: source authority has multiple active authored evidence rows for owner %s", ErrConnectorPushDurableConflict, authority.connectorOwnerKey)
			}
			if existing.namespaceFromEvidence && authority.namespaceFromEvidence &&
				!strings.EqualFold(existing.connectorNamespace, authority.connectorNamespace) {
				return fmt.Errorf("%w: source authority has conflicting authored evidence namespaces", ErrConnectorPushDurableConflict)
			}
			if authority.namespaceFromEvidence {
				existing.connectorNamespace = authority.connectorNamespace
				existing.namespaceFromEvidence = true
			}
			if existing.evidenceID == "" && authority.evidenceID != "" {
				existing.evidenceID = authority.evidenceID
			}
			if existing.sourceLinkKeys == nil {
				existing.sourceLinkKeys = map[string]struct{}{}
			}
			for linkKey := range authority.sourceLinkKeys {
				existing.sourceLinkKeys[linkKey] = struct{}{}
			}
			authorities[authority.connectorOwnerKey] = existing
			return nil
		}
		authorities[authority.connectorOwnerKey] = authority
		return nil
	}
	for rows.Next() {
		ev, err := scanConnectorInboundEvidence(rows)
		if err != nil {
			return "", err
		}
		if ev != nil {
			expectedActiveKey := ConnectorInboundEvidenceActiveKey(
				ev.ConnectorOwnerKey,
				ev.SourceObject,
				ev.SourceRecordId,
				ev.RemoteIDField,
			)
			if ev.ActiveKey != expectedActiveKey {
				return "", fmt.Errorf("%w: authored evidence active key materialization changed before authoritative reread", ErrConnectorPushDurableConflict)
			}
			if err := addAuthority(transferAuthority{
				connectorOwnerKey:     ev.ConnectorOwnerKey,
				connectorNamespace:    ev.ConnectorNamespace,
				namespaceFromEvidence: true,
				evidenceID:            ev.ID,
				remoteID:              ev.RemoteId,
			}); err != nil {
				return "", err
			}
		}
	}
	if err := rows.Err(); err != nil {
		return "", err
	}

	linkRows, err := querier.QueryContext(ctx, `
SELECT LinkKey, Status, ConnectorOwnerKey, ConnectorNamespace, ActionSlotId, ActionType,
	       SourceObject, SourceRecordId, DestinationSlotId, RemoteId,
	       COALESCE(RemoteLocatorHash, ''), COALESCE(RemoteLocatorActiveKey, '')
  FROM _System_ConnectorRemoteLink
 WHERE SourceObject = ?
   AND SourceRecordId = ?
   AND COALESCE(RemoteIdField, '') = ?
   AND Status IN (?, ?, ?)
 FOR UPDATE`,
		transfer.SourceObject,
		transfer.SourceRecordId,
		transfer.RemoteIDField,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
		ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
	)
	if err != nil {
		return "", err
	}
	type transferSourceLink struct {
		linkKey, status, namespace, remoteID, locatorHash, locatorActiveKey string
		identity                                                            ConnectorPushDurableIdentity
	}
	sourceLinks := make([]transferSourceLink, 0)
	for linkRows.Next() {
		var link transferSourceLink
		if err := linkRows.Scan(
			&link.linkKey,
			&link.status,
			&link.identity.ConnectorOwnerKey,
			&link.namespace,
			&link.identity.ActionSlotID,
			&link.identity.ActionType,
			&link.identity.SourceObject,
			&link.identity.SourceRecordID,
			&link.identity.DestinationSlotID,
			&link.remoteID,
			&link.locatorHash,
			&link.locatorActiveKey,
		); err != nil {
			_ = linkRows.Close()
			return "", err
		}
		sourceLinks = append(sourceLinks, link)
	}
	if err := linkRows.Err(); err != nil {
		_ = linkRows.Close()
		return "", err
	}
	if err := linkRows.Close(); err != nil {
		return "", err
	}
	for _, link := range sourceLinks {
		admission, admitted := admissionByOwner[strings.TrimSpace(link.identity.ConnectorOwnerKey)]
		if !admitted || admission.SourceIdentity != link.identity {
			return "", fmt.Errorf("%w: active source link does not match the engine-derived installed transfer identity", ErrConnectorPushDurableConflict)
		}
		legacyLinkKey, err := link.identity.LinkKey()
		if err != nil {
			return "", fmt.Errorf("%w: active source link durable identity is invalid", ErrConnectorPushDurableConflict)
		}
		var expectedHash string
		switch {
		case link.linkKey == legacyLinkKey:
			expectedHash = ConnectorPushRemoteLocatorHash(link.identity, link.remoteID)
			if _, locked := lockedLocatorHashes[expectedHash]; !locked {
				return "", fmt.Errorf("%w: connector remote identity transfer discovered an unadmitted locator", ErrConnectorPushDurableConflict)
			}
			if err := s.validateConnectorIntegrationV1BridgeQuiescent(ctx, link.identity, link.remoteID, admission.SourceIntegrationScope); err != nil {
				return "", err
			}
		case strings.HasPrefix(link.linkKey, connectorIntegrationLinkV2Prefix):
			if link.namespace != link.identity.ConnectorOwnerKey {
				return "", fmt.Errorf("%w: scoped integration source-link namespace does not match its durable owner", ErrConnectorPushDurableConflict)
			}
			expectedHash, err = s.validateConnectorIntegrationOutboundSourceLink(
				ctx,
				link.linkKey,
				link.identity,
				link.remoteID,
				admission.SourceIntegrationScope,
			)
			if err != nil {
				return "", err
			}
		default:
			return "", fmt.Errorf("%w: active source link key materialization changed before authoritative reread", ErrConnectorPushDurableConflict)
		}
		if link.locatorHash != expectedHash {
			return "", fmt.Errorf("%w: connector remote identity transfer locator history changed before authoritative reread", ErrConnectorPushDurableConflict)
		}
		switch link.status {
		case ConnectorPushLinkStatusFinalized, ConnectorPushLinkStatusSuspendedDrift:
			if link.locatorActiveKey != expectedHash {
				return "", fmt.Errorf("%w: active source link locator key materialization changed before authoritative reread", ErrConnectorPushDurableConflict)
			}
		case ConnectorPushLinkStatusOwnerDeletedPendingRecovery:
			return "", fmt.Errorf("%w: source link remains quarantined after owner deletion", ErrConnectorPushDurableConflict)
		default:
			return "", fmt.Errorf("%w: source link status changed before authoritative reread", ErrConnectorPushDurableConflict)
		}
		if err := addAuthority(transferAuthority{
			connectorOwnerKey:  link.identity.ConnectorOwnerKey,
			connectorNamespace: link.namespace,
			sourceLinkKeys:     map[string]struct{}{link.linkKey: {}},
			remoteID:           link.remoteID,
		}); err != nil {
			return "", err
		}
	}
	targetRemoteID, targetHasAuthority, err := s.findExactCompletedTargetRemoteIdentityForUpdate(
		ctx,
		querier,
		transfer,
		admissionByOwner,
	)
	if err != nil {
		return "", err
	}
	if targetHasAuthority {
		if len(authorities) != 0 {
			return "", fmt.Errorf("%w: source and target records both have active connector remote identity authority", ErrConnectorPushDurableConflict)
		}
		return targetRemoteID, nil
	}
	if len(authorities) == 0 {
		// Neither compatibility mirror is authority. Only the exact completed
		// target replay above or a proven source authority may return a remote ID.
		return "", nil
	}

	for _, authority := range authorities {
		if authority.evidenceID != "" {
			result, err := execer.ExecContext(ctx, `
UPDATE _System_ConnectorInboundRemoteIDEvidence
   SET EvidenceStatus = ?, ActiveKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND EvidenceStatus = ?`,
				ConnectorInboundEvidenceStatusRetired,
				authority.evidenceID,
				ConnectorInboundEvidenceStatusAuthored,
			)
			if err != nil {
				return "", err
			}
			if err := requireConnectorPushOneRow(result, "retire transferred inbound evidence"); err != nil {
				return "", fmt.Errorf("%w: %v", ErrConnectorPushDurableConflict, err)
			}
		}
		linkKeys := make([]string, 0, len(authority.sourceLinkKeys))
		for linkKey := range authority.sourceLinkKeys {
			linkKeys = append(linkKeys, linkKey)
		}
		sort.Strings(linkKeys)
		for _, linkKey := range linkKeys {
			result, err := execer.ExecContext(ctx, `
UPDATE _System_ConnectorRemoteLink
	   SET Status = ?, RemoteLocatorActiveKey = NULL, LastModifiedDate = CURRENT_TIMESTAMP
		 WHERE LinkKey = ?
		   AND Status IN (?, ?)`,
				ConnectorPushLinkStatusRetired,
				linkKey,
				ConnectorPushLinkStatusFinalized,
				ConnectorPushLinkStatusSuspendedDrift,
			)
			if err != nil {
				return "", err
			}
			if affected, err := result.RowsAffected(); err != nil || affected != 1 {
				return "", fmt.Errorf("%w: source link changed before atomic retirement", ErrConnectorPushDurableConflict)
			}
		}
		// Source and target share one remote-locator unique key. Retirement and
		// publication must commit together; insert failure leaves rollback to the
		// full-flow transaction owner rather than exposing an ownership gap.
		admission := admissionByOwner[authority.connectorOwnerKey]
		adoption, err := normalizeConnectorPushLinkAdoption(ConnectorPushLinkAdoption{
			Identity: admission.TargetIdentity, RemoteID: remoteID, Snapshot: admission.TargetSnapshot,
		})
		if err != nil {
			return "", err
		}
		mirrorStatus := ConnectorPushCompatibilityMirrorStatus(ConnectorPushCompatibilityMirrorPending)
		if transfer.TargetRemoteID == remoteID {
			mirrorStatus = ConnectorPushCompatibilityMirrorStatus(ConnectorPushCompatibilityMirrorSynced)
		}
		if err := s.insertAdoptedFinalizedLinkTx(ctx, tx, adoption, mirrorStatus); err != nil {
			return "", err
		}
	}
	return remoteID, nil
}

// findExactCompletedTargetRemoteIdentityForUpdate recognizes the sole
// idempotent replay state for a completed transfer: no source authority remains
// and each active target owner matches its installed admission and both locked
// record mirrors. An owner's exact V1 and scoped V2 physical links may coexist,
// as they can on the source.
// Foreign, ambiguous, drifted, suspended, or owner-deleted target state remains
// a durable conflict. The caller separately rejects simultaneous source authority.
func (s *SQLConnectorPushDurableStore) findExactCompletedTargetRemoteIdentityForUpdate(
	ctx context.Context,
	querier connectorPushQuerier,
	transfer ConnectorRemoteIdentityTransfer,
	admissionByOwner map[string]connectorRemoteIdentityTransferAdmission,
) (string, bool, error) {
	type targetEvidenceAuthority struct {
		id, owner, namespace, remoteID, activeKey string
	}
	evidenceRows, err := querier.QueryContext(ctx, `
SELECT Id, ConnectorOwnerKey, ConnectorNamespace, RemoteId, COALESCE(ActiveKey, '')
  FROM _System_ConnectorInboundRemoteIDEvidence
 WHERE SourceObject = ?
   AND SourceRecordId = ?
   AND RemoteIDField = ?
   AND EvidenceStatus = ?
 FOR UPDATE`,
		transfer.TargetObject,
		transfer.TargetRecordId,
		transfer.RemoteIDField,
		ConnectorInboundEvidenceStatusAuthored,
	)
	if err != nil {
		return "", false, err
	}
	evidenceAuthorities := make([]targetEvidenceAuthority, 0, 1)
	for evidenceRows.Next() {
		var authority targetEvidenceAuthority
		if err := evidenceRows.Scan(
			&authority.id,
			&authority.owner,
			&authority.namespace,
			&authority.remoteID,
			&authority.activeKey,
		); err != nil {
			_ = evidenceRows.Close()
			return "", false, err
		}
		evidenceAuthorities = append(evidenceAuthorities, authority)
	}
	if err := evidenceRows.Err(); err != nil {
		_ = evidenceRows.Close()
		return "", false, err
	}
	if err := evidenceRows.Close(); err != nil {
		return "", false, err
	}

	type targetLinkAuthority struct {
		id, linkKey, status, namespace, remoteID, locatorHash, locatorActiveKey string
		identity                                                                ConnectorPushDurableIdentity
	}
	linkRows, err := querier.QueryContext(ctx, `
SELECT Id, LinkKey, Status, ConnectorOwnerKey, ConnectorNamespace, ActionSlotId,
       ActionType, SourceObject, SourceRecordId, DestinationSlotId, RemoteId,
       COALESCE(RemoteLocatorHash, ''), COALESCE(RemoteLocatorActiveKey, '')
  FROM _System_ConnectorRemoteLink
 WHERE SourceObject = ?
   AND SourceRecordId = ?
   AND COALESCE(RemoteIdField, '') = ?
   AND Status IN (?, ?, ?)
 FOR UPDATE`,
		transfer.TargetObject,
		transfer.TargetRecordId,
		transfer.RemoteIDField,
		ConnectorPushLinkStatusFinalized,
		ConnectorPushLinkStatusSuspendedDrift,
		ConnectorPushLinkStatusOwnerDeletedPendingRecovery,
	)
	if err != nil {
		return "", false, err
	}
	linkAuthorities := make([]targetLinkAuthority, 0, 1)
	for linkRows.Next() {
		var authority targetLinkAuthority
		if err := linkRows.Scan(
			&authority.id,
			&authority.linkKey,
			&authority.status,
			&authority.identity.ConnectorOwnerKey,
			&authority.namespace,
			&authority.identity.ActionSlotID,
			&authority.identity.ActionType,
			&authority.identity.SourceObject,
			&authority.identity.SourceRecordID,
			&authority.identity.DestinationSlotID,
			&authority.remoteID,
			&authority.locatorHash,
			&authority.locatorActiveKey,
		); err != nil {
			_ = linkRows.Close()
			return "", false, err
		}
		linkAuthorities = append(linkAuthorities, authority)
	}
	if err := linkRows.Err(); err != nil {
		_ = linkRows.Close()
		return "", false, err
	}
	if err := linkRows.Close(); err != nil {
		return "", false, err
	}

	exactMirrors := func(remoteID string) error {
		remoteID = strings.TrimSpace(remoteID)
		if remoteID == "" || transfer.RemoteId == "" || transfer.TargetRemoteID == "" ||
			remoteID != transfer.RemoteId || remoteID != transfer.TargetRemoteID {
			return fmt.Errorf("%w: target connector remote identity authority does not match both locked record mirrors", ErrConnectorPushDurableConflict)
		}
		return nil
	}
	// Distinct owners may legitimately share the record's remote-ID mirror.
	// Physical V1/V2 coexistence is allowed only within one exact admitted link
	// identity; authored evidence must remain unambiguous for each owner.
	provedOwners := make(map[string]bool, len(admissionByOwner))
	provedRemoteID := transfer.RemoteId
	for _, authority := range evidenceAuthorities {
		authority.owner = strings.TrimSpace(authority.owner)
		authority.namespace = strings.TrimSpace(authority.namespace)
		authority.remoteID = strings.TrimSpace(authority.remoteID)
		admission, admitted := admissionByOwner[authority.owner]
		if !admitted || strings.TrimSpace(admission.RemoteID) != authority.remoteID ||
			!strings.EqualFold(strings.TrimSpace(admission.ConnectorNamespace), authority.namespace) ||
			authority.activeKey != ConnectorInboundEvidenceActiveKey(
				authority.owner,
				transfer.TargetObject,
				transfer.TargetRecordId,
				transfer.RemoteIDField,
			) {
			return "", false, fmt.Errorf("%w: target authored evidence does not match the engine-derived installed transfer admission", ErrConnectorPushDurableConflict)
		}
		if err := exactMirrors(authority.remoteID); err != nil {
			return "", false, err
		}
		if _, exists := provedOwners[authority.owner]; exists {
			return "", false, fmt.Errorf("%w: target owner has multiple authored remote identity authorities", ErrConnectorPushDurableConflict)
		}
		provedOwners[authority.owner] = true
		provedRemoteID = authority.remoteID
	}

	for _, authority := range linkAuthorities {
		admission, admitted := admissionByOwner[authority.identity.ConnectorOwnerKey]
		if !admitted || authority.status != ConnectorPushLinkStatusFinalized ||
			admission.TargetIdentity != authority.identity ||
			strings.TrimSpace(admission.RemoteID) != authority.remoteID {
			return "", false, fmt.Errorf("%w: target link does not match one active finalized installed transfer admission", ErrConnectorPushDurableConflict)
		}
		if provedOwners[authority.identity.ConnectorOwnerKey] {
			return "", false, fmt.Errorf("%w: target owner has both authored evidence and active link authority", ErrConnectorPushDurableConflict)
		}
		legacyLinkKey, err := authority.identity.LinkKey()
		if err != nil {
			return "", false, fmt.Errorf("%w: target link durable identity is invalid", ErrConnectorPushDurableConflict)
		}
		var expectedHash string
		switch {
		case authority.linkKey == legacyLinkKey:
			if !strings.EqualFold(strings.TrimSpace(admission.TargetSnapshot.ConnectorNamespace), authority.namespace) {
				return "", false, fmt.Errorf("%w: target link namespace does not match its installed snapshot", ErrConnectorPushDurableConflict)
			}
			expectedHash = ConnectorPushRemoteLocatorHash(authority.identity, authority.remoteID)
		case strings.HasPrefix(authority.linkKey, connectorIntegrationLinkV2Prefix):
			if authority.namespace != authority.identity.ConnectorOwnerKey {
				return "", false, fmt.Errorf("%w: scoped integration target-link namespace does not match its durable owner", ErrConnectorPushDurableConflict)
			}
			expectedHash, err = s.validateConnectorIntegrationOutboundSourceLink(
				ctx,
				authority.linkKey,
				authority.identity,
				authority.remoteID,
				admission.TargetIntegrationScope,
			)
			if err != nil {
				return "", false, err
			}
		default:
			return "", false, fmt.Errorf("%w: target link key does not match the installed transfer identity", ErrConnectorPushDurableConflict)
		}
		if authority.locatorHash != expectedHash || authority.locatorActiveKey != expectedHash {
			return "", false, fmt.Errorf("%w: target link locator materialization does not match the installed transfer identity", ErrConnectorPushDurableConflict)
		}
		if err := exactMirrors(authority.remoteID); err != nil {
			return "", false, err
		}
		provedOwners[authority.identity.ConnectorOwnerKey] = false
		provedRemoteID = authority.remoteID
	}
	if len(provedOwners) == 0 {
		return "", false, nil
	}
	return provedRemoteID, true, nil
}

// RetireEvidence transitions the evidence row to status='retired' and clears
// ActiveKey so the UNIQUE index no longer blocks a future authored row.
func (s *SQLConnectorPushDurableStore) RetireEvidence(ctx context.Context, id string) error {
	return s.execTransitionAmbient(ctx, "retire evidence", `
UPDATE _System_ConnectorInboundRemoteIDEvidence
   SET EvidenceStatus = ?, ActiveKey = NULL,
       LastModifiedDate = CURRENT_TIMESTAMP
 WHERE Id = ? AND EvidenceStatus = ?`,
		ConnectorInboundEvidenceStatusRetired,
		id,
		ConnectorInboundEvidenceStatusAuthored,
	)
}

func scanConnectorInboundEvidence(row connectorPushScanner) (*ConnectorInboundRemoteIDEvidence, error) {
	e := &ConnectorInboundRemoteIDEvidence{}
	err := row.Scan(
		&e.ID, &e.ConnectorOwnerKey, &e.ConnectorNamespace,
		&e.SourceObject, &e.SourceRecordId,
		&e.RemoteIDField, &e.RemoteId, &e.EvidenceStatus,
		&e.ConsumedByLinkKey, &e.ActiveKey,
		&e.CreatedDate, &e.LastModifiedDate,
	)
	if err != nil {
		return nil, err
	}
	return e, nil
}