package automation

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"testing"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

// ---------------------------------------------------------------------------
// fake ConnectorPushDurableStore for inbound-register tests
// ---------------------------------------------------------------------------

// inboundRegisterDurableStore is a minimal in-memory ConnectorPushDurableStore
// whose only operative methods are the inbound remote-id evidence ones.
// All PushToConnector link/attempt methods panic so a mis-wired test fails loudly.
type inboundRegisterDurableStore struct {
	evidence      map[string]*ConnectorInboundRemoteIDEvidence // keyed by ID
	insertErr     error                                        // inject to simulate conflict/error
	findActiveErr error                                        // inject to simulate DB error
	retired       []string                                     // IDs passed to RetireEvidence
	inserted      []*ConnectorInboundRemoteIDEvidence          // rows given to InsertAuthoredEvidence
	// racingRow models a row committed by a CONCURRENT transaction: it is visible
	// only to the *sql.DB conflict reread (FindActiveInboundRemoteIDEvidence), not
	// to the in-tx idempotency read (FindActiveInboundRemoteIDEvidenceInTx), which
	// is how the production store behaves under the inbound tx snapshot.
	racingRow        *ConnectorInboundRemoteIDEvidence
	alreadyFinalized bool
	admissionErr     error
	historyErr       error
	laneIdentities   []ConnectorPushDurableIdentity
}

func newInboundRegisterDurableStore() *inboundRegisterDurableStore {
	return &inboundRegisterDurableStore{
		evidence: map[string]*ConnectorInboundRemoteIDEvidence{},
	}
}

// seedActive inserts a pre-existing authored evidence row so tests can exercise
// the idempotent and conflict-retire branches.
func (s *inboundRegisterDurableStore) seedActive(ownerKey, ns, sourceObject, sourceRecordId, field, remoteId string) *ConnectorInboundRemoteIDEvidence {
	ev := &ConnectorInboundRemoteIDEvidence{
		ID:                 "seed-" + field,
		ConnectorOwnerKey:  ownerKey,
		ConnectorNamespace: ns,
		SourceObject:       sourceObject,
		SourceRecordId:     sourceRecordId,
		RemoteIDField:      field,
		RemoteId:           remoteId,
		EvidenceStatus:     ConnectorInboundEvidenceStatusAuthored,
		ActiveKey:          ConnectorInboundEvidenceActiveKey(ownerKey, sourceObject, sourceRecordId, field),
	}
	s.evidence[ev.ID] = ev
	return ev
}

func (s *inboundRegisterDurableStore) InsertAuthoredEvidence(_ context.Context, ev ConnectorInboundRemoteIDEvidence) error {
	if s.insertErr != nil {
		return s.insertErr
	}
	ev.EvidenceStatus = ConnectorInboundEvidenceStatusAuthored
	ev.ActiveKey = ConnectorInboundEvidenceActiveKey(ev.ConnectorOwnerKey, ev.SourceObject, ev.SourceRecordId, ev.RemoteIDField)
	if ev.ID == "" {
		ev.ID = "inserted-" + ev.RemoteIDField
	}
	cp := ev
	s.evidence[ev.ID] = &cp
	s.inserted = append(s.inserted, &cp)
	return nil
}

func (s *inboundRegisterDurableStore) lookupActive(ownerKey, sourceObject, sourceRecordId, remoteIDField string) *ConnectorInboundRemoteIDEvidence {
	activeKey := ConnectorInboundEvidenceActiveKey(ownerKey, sourceObject, sourceRecordId, remoteIDField)
	for _, ev := range s.evidence {
		if ev.ActiveKey == activeKey && ev.EvidenceStatus == ConnectorInboundEvidenceStatusAuthored {
			cp := *ev
			return &cp
		}
	}
	return nil
}

// FindActiveInboundRemoteIDEvidence is the *sql.DB-backed conflict reread: it also
// reveals a concurrent committed racing row (racingRow) the in-tx read cannot see.
func (s *inboundRegisterDurableStore) FindActiveInboundRemoteIDEvidence(_ context.Context, ownerKey, sourceObject, sourceRecordId, remoteIDField string) (*ConnectorInboundRemoteIDEvidence, error) {
	if s.findActiveErr != nil {
		return nil, s.findActiveErr
	}
	if s.racingRow != nil {
		cp := *s.racingRow
		return &cp, nil
	}
	return s.lookupActive(ownerKey, sourceObject, sourceRecordId, remoteIDField), nil
}

// FindActiveInboundRemoteIDEvidenceInTx is the in-tx idempotency read: it sees
// only rows already in the store snapshot (not the concurrent racingRow).
func (s *inboundRegisterDurableStore) FindActiveInboundRemoteIDEvidenceInTx(_ context.Context, ownerKey, sourceObject, sourceRecordId, remoteIDField string) (*ConnectorInboundRemoteIDEvidence, error) {
	if s.findActiveErr != nil {
		return nil, s.findActiveErr
	}
	return s.lookupActive(ownerKey, sourceObject, sourceRecordId, remoteIDField), nil
}

func (s *inboundRegisterDurableStore) RetireEvidence(_ context.Context, id string) error {
	s.retired = append(s.retired, id)
	if ev := s.evidence[id]; ev != nil {
		ev.EvidenceStatus = ConnectorInboundEvidenceStatusRetired
		ev.ActiveKey = ""
	}
	return nil
}

func (s *inboundRegisterDurableStore) TransferInboundRemoteIdentity(_ context.Context, _ ConnectorRemoteIdentityTransfer) (string, error) {
	panic("unexpected TransferInboundRemoteIdentity in inbound-register test")
}

func (s *inboundRegisterDurableStore) LockConnectorRemoteIdentityLaneInAmbientTx(_ context.Context, identity ConnectorPushDurableIdentity, _ string) error {
	s.laneIdentities = append(s.laneIdentities, identity)
	return s.admissionErr
}

func (s *inboundRegisterDurableStore) CheckConnectorRemoteIdentityRegistrationInAmbientTx(_ context.Context, _ ConnectorPushDurableIdentity, _ string) (bool, error) {
	return s.alreadyFinalized, s.historyErr
}

type inboundRegisterInstalledResolver struct{}

func (r inboundRegisterInstalledResolver) ResolveConnectorInboundRemoteIdentityInstalledContracts(ctx context.Context, owner string) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return r.ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(ctx, owner)
}

func (inboundRegisterInstalledResolver) ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(_ context.Context, owner string) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return ConnectorInboundRemoteIdentityInstalledContractSet{
		Generation: "generation-1",
		Contracts: []connector.ConnectorPushDurableMetadataContract{{
			ConnectorOwnerKey: owner,
			ActionSlotID:      "contact-push-slot",
			ActionType:        constants.ActionTypePushToConnector,
			SourceObject:      "Contact",
			DestinationSlotID: "remote-contact",
			TargetObject:      "contact",
			ProfileName:       "default",
			RemoteIDField:     "External_Id__c",
		}},
	}, nil
}

func (r inboundRegisterInstalledResolver) ResolveConnectorInboundRemoteIdentityInstalledContractSetsInAmbientTx(ctx context.Context, owners []string) (map[string]ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	sets := make(map[string]ConnectorInboundRemoteIdentityInstalledContractSet, len(owners))
	for _, owner := range owners {
		set, err := r.ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(ctx, owner)
		if err != nil {
			return nil, err
		}
		sets[owner] = set
	}
	return sets, nil
}

type transferRemoteIdentityDurableStore struct {
	*inboundRegisterDurableStore
	transfers          []ConnectorRemoteIdentityTransfer
	remoteID           string
	err                error
	targetHasAuthority bool
}

func newTransferRemoteIdentityDurableStore() *transferRemoteIdentityDurableStore {
	return &transferRemoteIdentityDurableStore{
		inboundRegisterDurableStore: newInboundRegisterDurableStore(),
		remoteID:                    "remote-229009332242",
	}
}

func (s *transferRemoteIdentityDurableStore) TransferInboundRemoteIdentity(_ context.Context, transfer ConnectorRemoteIdentityTransfer) (string, error) {
	if s.err != nil {
		return "", s.err
	}
	s.transfers = append(s.transfers, transfer)
	if s.targetHasAuthority && transfer.TargetRemoteID != "" {
		return "", fmt.Errorf("%w: target record already has active connector remote identity authority", ErrConnectorPushDurableConflict)
	}
	return s.remoteID, nil
}

func (s *transferRemoteIdentityDurableStore) FindInboundRemoteIdentityTransferHintsInAmbientTx(context.Context, string, string, string, string, string) ([]ConnectorRemoteIdentityTransferHint, error) {
	return nil, s.err
}

func (*transferRemoteIdentityDurableStore) LockInboundRemoteIdentityTransferBoundaryInAmbientTx(context.Context, string, string, string, string, string) error {
	return nil
}

// All PushToConnector link/attempt methods are unexpected in these tests — panic loudly.

func (s *inboundRegisterDurableStore) FindLink(_ context.Context, _ string) (*ConnectorPushRemoteLink, error) {
	panic("unexpected FindLink in inbound-register test")
}
func (s *inboundRegisterDurableStore) FindFinalizedLink(_ context.Context, _ ConnectorPushDurableIdentity) (*ConnectorPushRemoteLink, error) {
	panic("unexpected FindFinalizedLink in inbound-register test")
}
func (s *inboundRegisterDurableStore) FindBlockingAttempt(_ context.Context, _ string) (*ConnectorPushAttempt, error) {
	panic("unexpected FindBlockingAttempt in inbound-register test")
}
func (s *inboundRegisterDurableStore) FindStateBySource(_ context.Context, _ ConnectorPushDurableSourceQuery) (*ConnectorPushRemoteLink, *ConnectorPushAttempt, error) {
	panic("unexpected FindStateBySource in inbound-register test")
}
func (s *inboundRegisterDurableStore) FindStateByDestination(_ context.Context, _ ConnectorPushDurableDestinationQuery) (*ConnectorPushRemoteLink, *ConnectorPushAttempt, error) {
	panic("unexpected FindStateByDestination in inbound-register test")
}

func (s *inboundRegisterDurableStore) AdoptFinalizedLink(_ context.Context, _ ConnectorPushLinkAdoption) (*ConnectorPushRemoteLink, error) {
	panic("unexpected AdoptFinalizedLink in inbound-register test")
}

func (s *inboundRegisterDurableStore) FindRecordPushState(_ context.Context, _ ConnectorPushRecordStateQuery) ([]ConnectorPushRecordDestinationState, error) {
	return nil, nil
}

// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------

// inboundRegisterPersistenceStub is a minimal PersistenceProvider for inbound
// register tests. Only GetLock is operative; all other methods panic or no-op.
// lockRecords is keyed by "objectName/id" and maps to the committed SObject.
type inboundRegisterPersistenceStub struct {
	lockRecords map[string]models.SObject
	lockErr     error
	updateCtx   context.Context
	updateErr   error
	txActive    bool
	updates     []models.SObject
	lockCalls   []string
}

func (s *inboundRegisterPersistenceStub) GetLock(_ context.Context, objectName string, id string) (models.SObject, error) {
	s.lockCalls = append(s.lockCalls, objectName+"/"+id)
	if s.lockErr != nil {
		return nil, s.lockErr
	}
	if s.lockRecords != nil {
		if rec, ok := s.lockRecords[objectName+"/"+id]; ok {
			return rec, nil
		}
	}
	return nil, nil
}

func (s *inboundRegisterPersistenceStub) Insert(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	return record, nil
}
func (s *inboundRegisterPersistenceStub) PreviewBeforeInsert(_ context.Context, _ string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	return record, nil
}
func (s *inboundRegisterPersistenceStub) PrelockUpdateParents(_ context.Context, _ string, _ string, _ models.SObject, _ *models.UserSession) error {
	return nil
}
func (s *inboundRegisterPersistenceStub) Update(ctx context.Context, objectName string, id string, updates models.SObject, _ *models.UserSession) error {
	s.updateCtx = ctx
	if s.updateErr != nil {
		return s.updateErr
	}
	s.updates = append(s.updates, updates)
	key := objectName + "/" + id
	if s.lockRecords != nil {
		rec := s.lockRecords[key]
		if rec == nil {
			rec = models.SObject{"Id": id}
		}
		for field, value := range updates {
			rec[field] = value
		}
		s.lockRecords[key] = rec
	}
	return nil
}
func (s *inboundRegisterPersistenceStub) PreviewBeforeUpdate(_ context.Context, _ string, _ string, _ models.SObject, _ models.SObject, _ *models.UserSession) (*models.PreparedRecordUpdate, error) {
	return nil, nil
}
func (s *inboundRegisterPersistenceStub) CommitPreparedUpdate(_ context.Context, _ string, _ string, _ *models.PreparedRecordUpdate, _ models.SObject, _ *models.UserSession) error {
	return nil
}
func (s *inboundRegisterPersistenceStub) Delete(_ context.Context, _ string, _ string, _ *models.UserSession) error {
	return nil
}
func (s *inboundRegisterPersistenceStub) GetRaw(_ context.Context, _ string, _ string) (models.SObject, error) {
	return nil, nil
}
func (s *inboundRegisterPersistenceStub) GetRecordType(_ context.Context, _ string) (string, error) {
	return "", nil
}
func (s *inboundRegisterPersistenceStub) RunInTransaction(_ context.Context, fn func(context.Context) error) error {
	return fn(context.Background())
}
func (s *inboundRegisterPersistenceStub) IsTransactionActive(context.Context) bool { return s.txActive }
func (s *inboundRegisterPersistenceStub) RunInPessimisticTransaction(_ context.Context, fn func(context.Context) error) error {
	return fn(context.Background())
}
func (s *inboundRegisterPersistenceStub) DetachTx(ctx context.Context) context.Context { return ctx }
func (s *inboundRegisterPersistenceStub) RecomputeActiveFormulaProjection(_ context.Context, _ *models.UserSession, _, _ string, _ []string) (models.SObject, error) {
	return nil, nil
}

// newInboundRegisterExec creates a FlowExecutor wired with both the durable store
// and a GetLock persistence stub pre-seeded with the given committed records.
// lockRecords maps "objectName/id" → committed SObject.
func newInboundRegisterExec(store *inboundRegisterDurableStore, lockRecords map[string]models.SObject) *FlowExecutor {
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.SetConnectorInboundRemoteIdentityInstalledContractResolver(inboundRegisterInstalledResolver{})
	exec.persistence = &inboundRegisterPersistenceStub{lockRecords: lockRecords, txActive: true}
	return exec
}

// withEngineSnapshot appends a record to the engine-owned post-upsert registry
// on the FRC carried by ctx, matching what executeRecordCreate /
// applyRecordUpdatePersistence publish after a successful persistence operation.
// This is the replacement for pre-populating interview.ContextData["$Record"]
// for identity in tests.
func withEngineSnapshot(ctx context.Context, object string, record models.SObject) context.Context {
	frc := flowRuntimeContextFromCtx(ctx)
	frc.AddPostUpsertRecord(object, record)
	// AddPostUpsertRecord mutates the shared postUpsert registry in place; the
	// ctx still carries the same *FlowRuntimeContext pointer, so no rebind is
	// needed. But to be safe, rebind so callers get the updated ctx.
	return withFlowRuntimeContext(ctx, frc)
}

// inboundRegisterAction builds a FlowActionCall with bare-string InputParameters.
// The action handler resolves bare strings via fmt.Sprintf("%v", ...) which
// yields the string value directly.
func inboundRegisterAction(params map[string]string) *models.FlowActionCall {
	action := &models.FlowActionCall{ActionType: "registerConnectorInboundRemoteAuthorship"}
	for k, v := range params {
		action.InputParameters = append(action.InputParameters, models.FlowActionParameter{
			Name:  k,
			Value: v,
		})
	}
	return action
}

func inboundRegisterInterview(record models.SObject) *models.FlowInterview {
	return &models.FlowInterview{
		ContextData: map[string]interface{}{
			"$Record": record,
		},
	}
}

// inboundRegisterTestOwnerPack is the pack that owns the standard register Flow
// in these tests; it equals the connectorOwnerKey param the actions carry.
const inboundRegisterTestOwnerPack = "my-connector-pack"

// connectorInboundOriginCtx returns a ctx carrying the connector sync origin
// (the namespace the inbound worker stamps), a pack-owned uncustomized standard
// Flow identity, AND the WithConnectorInboundFlowExecution marker that
// flow_adapter.go stamps on genuine inbound-worker paths — the three facts the
// register action authenticates against (sync-origin gate + trusted-owner
// binding + inbound-execution guard).
func connectorInboundOriginCtx(syncOrigin string) context.Context {
	ctx := connectorInboundCtxWithIdentity(syncOrigin, FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		DefinitionID:  "flow-def-register",
		VersionID:     "flow-version-register",
		OwnerPack:     inboundRegisterTestOwnerPack,
		VersionSource: flowSourceStandard,
	})
	// Mirror flow_adapter.go: stamp WithConnectorInboundFlowExecution using the
	// same syncOrigin the worker stamped on ContextKeyConnectorSyncOrigin.
	return constants.WithConnectorInboundFlowExecution(ctx, "Connector_Inbound_Register", syncOrigin)
}

// connectorInboundCtxWithIdentity carries the sync origin plus an explicit Flow
// execution identity, so tests can exercise the trusted-owner binding (e.g. a
// customized or foreign-owner Flow under a matching sync origin).
func connectorInboundCtxWithIdentity(syncOrigin string, identity FlowExecutionIdentity) context.Context {
	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, syncOrigin)
	ctx = withFlowRuntimeContext(ctx, flowRuntimeContextFromCtx(ctx).WithCurrentFlow(identity))
	// The register action must run inside an atomic Flow boundary (the inbound
	// worker runs the flow with runAtomically); set it so the happy-path tests
	// exercise the real authenticated path.
	return withAtomicFlowBoundary(ctx)
}

func transferConnectorRemoteIdentityCtx() context.Context {
	return withAtomicFlowPessimisticRetryOwner(transferConnectorRemoteIdentityCtxWithoutRetryOwner())
}

func transferConnectorRemoteIdentityCtxWithoutRetryOwner() context.Context {
	ctx := connectorInboundCtxWithIdentity("", FlowExecutionIdentity{
		APIName:       "Std_ConvertLead",
		DefinitionID:  "flow-def-std-convert-lead",
		VersionID:     "flow-version-std-convert-lead",
		OwnerPack:     "salesforce-crm-standard",
		VersionSource: flowSourceStandard,
	})
	ctx = constants.WithStandardActionExecutionRootFlow(ctx, "TransferLeadConnectorRemoteIdentity", "Std_ConvertLead")
	return constants.WithStandardActionProtectedFlowExecution(ctx, "Std_ConvertLead")
}

func transferConnectorRemoteIdentityAction(sourceID, targetID string) *models.FlowActionCall {
	return &models.FlowActionCall{
		Name:       "TransferLeadConnectorRemoteIdentity",
		ActionType: "transferConnectorRemoteIdentity",
		InputParameters: []models.FlowActionParameter{
			{Name: "sourceObject", Value: "Lead"},
			{Name: "sourceId", Value: sourceID},
			{Name: "targetObject", Value: "Contact"},
			{Name: "targetId", Value: targetID},
			{Name: "remoteIDField", Value: "External_Id__c"},
			{Name: "outputReference", Value: "remoteIdentityTransferred"},
		},
		Connector: &models.FlowConnector{TargetReference: "PreviewSourceBeforeSaveUpdate"},
	}
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

func TestTransferConnectorRemoteIdentityActionTransfersMatchingRemoteID(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id":             "00Qsource",
				"External_Id__c": "remote-229009332242",
			},
			"Contact/003target": {
				"Id":             "003target",
				"External_Id__c": "remote-229009332242",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	interview := &models.FlowInterview{}
	next, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		interview,
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err != nil {
		t.Fatalf("executeActionCall(transferConnectorRemoteIdentity): %v", err)
	}
	if next != "PreviewSourceBeforeSaveUpdate" {
		t.Fatalf("next = %q, want PreviewSourceBeforeSaveUpdate", next)
	}
	if len(persistence.updates) != 0 {
		t.Fatalf("target update count = %d, want 0 when target remote id already matches", len(persistence.updates))
	}
	if got, ok := interview.ContextData["remoteIdentityTransferred"].(bool); !ok || !got {
		t.Fatalf("remoteIdentityTransferred = %#v (bool=%t), want true", got, ok)
	}
	if len(store.transfers) != 1 {
		t.Fatalf("transfer count = %d, want 1", len(store.transfers))
	}
	if got, want := strings.Join(persistence.lockCalls, ","), "Contact/003target,Lead/00Qsource"; got != want {
		t.Fatalf("stable record lock order = %q, want %q", got, want)
	}
	got := store.transfers[0]
	if got.SourceObject != "Lead" || got.SourceRecordId != "00Qsource" ||
		got.TargetObject != "Contact" || got.TargetRecordId != "003target" ||
		got.RemoteIDField != "External_Id__c" || got.RemoteId != "remote-229009332242" || got.TargetRemoteID != got.RemoteId {
		t.Fatalf("transfer = %#v, want Lead/00Qsource -> Contact/003target External_Id__c with both locked record mirrors", got)
	}
}

func TestTransferConnectorRemoteIdentityActionAllowsOnlyExactAuthorizedConnectorInboundFlow(t *testing.T) {
	baseIdentity := FlowExecutionIdentity{
		APIName:       "HubSpot_Contact_Lead_Upsert",
		DefinitionID:  "flow-def-hubspot-companion",
		VersionID:     "flow-version-hubspot-companion",
		OwnerPack:     "pingcap-hubspot-companion",
		VersionSource: flowSourceStandard,
	}
	tests := []struct {
		name       string
		identity   FlowExecutionIdentity
		syncOrigin string
		authorize  func(context.Context, FlowExecutionIdentity) context.Context
		wantErr    bool
	}{
		{
			name:       "same-pack name-only inbound root",
			identity:   baseIdentity,
			syncOrigin: baseIdentity.OwnerPack,
			authorize: func(ctx context.Context, identity FlowExecutionIdentity) context.Context {
				return constants.WithConnectorInboundFlowExecution(ctx, identity.APIName, identity.OwnerPack)
			},
			wantErr: true,
		},
		{
			name:       "exact frozen cross-pack companion version",
			identity:   baseIdentity,
			syncOrigin: "hubspot-routing-namespace",
			authorize: func(ctx context.Context, identity FlowExecutionIdentity) context.Context {
				return constants.WithConnectorInboundCompanionFlowExecution(ctx, identity.VersionID)
			},
		},
		{
			name:       "cross-pack inbound root without companion marker",
			identity:   baseIdentity,
			syncOrigin: "hubspot-routing-namespace",
			authorize: func(ctx context.Context, identity FlowExecutionIdentity) context.Context {
				return constants.WithConnectorInboundFlowExecution(ctx, identity.APIName, "hubspot-routing-namespace")
			},
			wantErr: true,
		},
		{
			name:       "different frozen companion version",
			identity:   baseIdentity,
			syncOrigin: "hubspot-routing-namespace",
			authorize: func(ctx context.Context, _ FlowExecutionIdentity) context.Context {
				return constants.WithConnectorInboundCompanionFlowExecution(ctx, "flow-version-other")
			},
			wantErr: true,
		},
		{
			name: "ownerless frozen companion version",
			identity: func() FlowExecutionIdentity {
				identity := baseIdentity
				identity.OwnerPack = ""
				return identity
			}(),
			syncOrigin: "hubspot-routing-namespace",
			authorize: func(ctx context.Context, identity FlowExecutionIdentity) context.Context {
				return constants.WithConnectorInboundCompanionFlowExecution(ctx, identity.VersionID)
			},
			wantErr: true,
		},
		{
			name:       "no inbound authorization",
			identity:   baseIdentity,
			syncOrigin: "hubspot-routing-namespace",
			authorize:  func(ctx context.Context, _ FlowExecutionIdentity) context.Context { return ctx },
			wantErr:    true,
		},
		{
			name: "customized flow",
			identity: func() FlowExecutionIdentity {
				identity := baseIdentity
				identity.DefinitionIsCustomized = true
				return identity
			}(),
			syncOrigin: baseIdentity.OwnerPack,
			authorize: func(ctx context.Context, identity FlowExecutionIdentity) context.Context {
				return constants.WithConnectorInboundFlowExecution(ctx, identity.APIName, identity.OwnerPack)
			},
			wantErr: true,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			store := newTransferRemoteIdentityDurableStore()
			persistence := &inboundRegisterPersistenceStub{
				txActive: true,
				lockRecords: map[string]models.SObject{
					"Lead/00Qsource": {
						"Id":             "00Qsource",
						"External_Id__c": "remote-229009332242",
					},
					"Contact/003target": {
						"Id":             "003target",
						"External_Id__c": "remote-229009332242",
					},
				},
			}
			exec := &FlowExecutor{}
			exec.SetConnectorPushDurableStore(store)
			exec.persistence = persistence
			exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

			ctx := connectorInboundCtxWithIdentity(tc.syncOrigin, tc.identity)
			ctx = tc.authorize(ctx, tc.identity)
			ctx = withAtomicFlowPessimisticRetryOwner(ctx)
			_, err := exec.executeActionCall(
				ctx,
				&models.FlowInterview{},
				&models.FlowMetadata{},
				transferConnectorRemoteIdentityAction("00Qsource", "003target"),
				&models.UserSession{ID: "005-user"},
			)
			if tc.wantErr {
				if err == nil {
					t.Fatal("executeActionCall(transferConnectorRemoteIdentity) error = nil, want authorization failure")
				}
				if len(persistence.lockCalls) != 0 || len(store.transfers) != 0 {
					t.Fatalf("unauthorized flow reached transfer work: lockCalls=%v transfers=%v", persistence.lockCalls, store.transfers)
				}
				return
			}
			if err != nil {
				t.Fatalf("executeActionCall(transferConnectorRemoteIdentity): %v", err)
			}
			if len(store.transfers) != 1 {
				t.Fatalf("authorized flow transfer count = %d, want 1", len(store.transfers))
			}
		})
	}
}

func TestTransferConnectorRemoteIdentityActionAuthorizationFailureUsesFaultConnector(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	persistence := &inboundRegisterPersistenceStub{txActive: true}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	action := transferConnectorRemoteIdentityAction("00Qsource", "003target")
	action.FaultConnector = &models.FlowConnector{TargetReference: "Fault"}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	ctx := connectorInboundCtxWithIdentity("pingcap-hubspot-companion", FlowExecutionIdentity{
		APIName:       "HubSpot_Contact_Lead_Upsert",
		DefinitionID:  "flow-def-hubspot-companion",
		VersionID:     "flow-version-hubspot-companion",
		OwnerPack:     "pingcap-hubspot-companion",
		VersionSource: flowSourceStandard,
	})
	ctx = withAtomicFlowPessimisticRetryOwner(ctx)

	next, err := exec.executeActionCall(
		ctx,
		interview,
		&models.FlowMetadata{},
		action,
		&models.UserSession{ID: "005-user"},
	)
	if err != nil {
		t.Fatalf("executeActionCall(transferConnectorRemoteIdentity) error = %v, want fault-routed nil error", err)
	}
	if next != "Fault" {
		t.Fatalf("next = %q, want Fault", next)
	}
	if message, _ := interview.ContextData["$Flow.FaultMessage"].(string); !strings.Contains(message, "requires an exact authorized") {
		t.Fatalf("$Flow.FaultMessage = %q, want authorization failure", message)
	}
	if len(persistence.lockCalls) != 0 || len(store.transfers) != 0 {
		t.Fatalf("authorization failure reached transfer work: lockCalls=%v transfers=%v", persistence.lockCalls, store.transfers)
	}
}

func TestTransferConnectorRemoteIdentityActionRejectsAtomicTransactionWithoutFullFlowRetryOwner(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	persistence := &inboundRegisterPersistenceStub{txActive: true}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	_, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtxWithoutRetryOwner(),
		&models.FlowInterview{},
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err == nil {
		t.Fatal("executeActionCall(transferConnectorRemoteIdentity) error = nil, want retry-owner failure")
	}
	if !strings.Contains(err.Error(), "owned by full-flow pessimistic retry") {
		t.Fatalf("error = %q, want full-flow pessimistic retry ownership detail", err.Error())
	}
	if len(persistence.lockCalls) != 0 || len(store.transfers) != 0 {
		t.Fatalf("unowned transaction reached transfer work: lockCalls=%v transfers=%v", persistence.lockCalls, store.transfers)
	}
}

func TestTransferConnectorRemoteIdentityActionFailsClosedOnBlankTargetMirrorAfterTransfer(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id": "00Qsource",
			},
			"Contact/003target": {
				"Id": "003target",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	_, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		&models.FlowInterview{},
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err == nil {
		t.Fatal("executeActionCall(transferConnectorRemoteIdentity) error = nil, want blank target mirror failure")
	}
	if !strings.Contains(err.Error(), "is blank after durable remote id transfer") {
		t.Fatalf("error = %q, want blank target mirror detail", err.Error())
	}
	if len(persistence.updates) != 0 {
		t.Fatalf("target update count = %d, want 0; compatibility mirrors must not be stamped in this atomic action", len(persistence.updates))
	}
	if got := persistence.lockRecords["Contact/003target"]["External_Id__c"]; got != nil {
		t.Fatalf("Contact.External_Id__c after transfer = %#v, want nil; durable state is the authority", got)
	}
	if len(store.transfers) != 1 {
		t.Fatalf("transfer count = %d, want 1", len(store.transfers))
	}
}

func TestTransferConnectorRemoteIdentityActionContinuesWhenNoConnectorAuthority(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	store.remoteID = ""
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id": "00Qsource",
			},
			"Contact/003target": {
				"Id": "003target",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	interview := &models.FlowInterview{}
	next, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		interview,
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err != nil {
		t.Fatalf("executeActionCall(transferConnectorRemoteIdentity): %v", err)
	}
	if next != "PreviewSourceBeforeSaveUpdate" {
		t.Fatalf("next = %q, want PreviewSourceBeforeSaveUpdate", next)
	}
	if len(persistence.updates) != 0 {
		t.Fatalf("target update count = %d, want 0 when no connector authority transfers", len(persistence.updates))
	}
	if got, ok := interview.ContextData["remoteIdentityTransferred"].(bool); !ok || got {
		t.Fatalf("remoteIdentityTransferred = %#v (bool=%t), want false", got, ok)
	}
	if len(store.transfers) != 1 {
		t.Fatalf("transfer count = %d, want 1; protected conversion flows may call this action unconditionally", len(store.transfers))
	}
}

func TestTransferConnectorRemoteIdentityActionClearsTargetMirrorAndContinuesWithoutAuthority(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	store.remoteID = ""
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id": "00Qsource",
			},
			"Contact/003target": {
				"Id":             "003target",
				"External_Id__c": "preexisting-mirror-without-authority",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	interview := &models.FlowInterview{}
	next, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		interview,
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err != nil {
		t.Fatalf("executeActionCall(transferConnectorRemoteIdentity): %v", err)
	}
	if next != "PreviewSourceBeforeSaveUpdate" {
		t.Fatalf("next = %q, want PreviewSourceBeforeSaveUpdate", next)
	}
	if len(persistence.updates) != 1 {
		t.Fatalf("target update count = %d, want 1 to clear the untrusted mirror", len(persistence.updates))
	}
	if got, ok := persistence.updates[0]["External_Id__c"]; !ok || got != nil {
		t.Fatalf("target mirror cleanup = %#v (present=%t), want explicit nil", got, ok)
	}
	if got := persistence.lockRecords["Contact/003target"]["External_Id__c"]; got != nil {
		t.Fatalf("Contact.External_Id__c after cleanup = %#v, want nil", got)
	}
	if got := constants.GetSecurityContext(persistence.updateCtx); got < constants.SystemContext {
		t.Fatalf("target mirror cleanup security context = %v, want SystemContext or higher", got)
	}
	if got, _ := persistence.updateCtx.Value(constants.ContextKeySuppressRecordTriggeredAutomation).(string); got != "Contact/003target" {
		t.Fatalf("target mirror cleanup suppressed automation tuple = %q, want Contact/003target", got)
	}
	if got, ok := interview.ContextData["remoteIdentityTransferred"].(bool); !ok || got {
		t.Fatalf("remoteIdentityTransferred = %#v (bool=%t), want false", got, ok)
	}
	if len(store.transfers) != 1 {
		t.Fatalf("transfer count = %d, want 1", len(store.transfers))
	}
}

func TestTransferConnectorRemoteIdentityActionFailsClosedWhenTargetMirrorCleanupFails(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	store.remoteID = ""
	persistence := &inboundRegisterPersistenceStub{
		txActive:  true,
		updateErr: errors.New("cleanup write failed"),
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id": "00Qsource",
			},
			"Contact/003target": {
				"Id":             "003target",
				"External_Id__c": "preexisting-mirror-without-authority",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	_, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		&models.FlowInterview{},
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err == nil {
		t.Fatal("executeActionCall(transferConnectorRemoteIdentity) error = nil, want cleanup failure")
	}
	if !strings.Contains(err.Error(), "clear untrusted target remote-id mirror") {
		t.Fatalf("error = %q, want target mirror cleanup detail", err.Error())
	}
	if got := persistence.lockRecords["Contact/003target"]["External_Id__c"]; got != "preexisting-mirror-without-authority" {
		t.Fatalf("Contact.External_Id__c after failed cleanup = %#v, want original mirror before atomic rollback", got)
	}
}

func TestTransferConnectorRemoteIdentityActionFailsClosedWhenTargetOwnsDurableAuthority(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	store.err = fmt.Errorf("%w: target Contact/003target already owns the mirror", ErrConnectorPushDurableConflict)
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id": "00Qsource",
			},
			"Contact/003target": {
				"Id":             "003target",
				"External_Id__c": "target-owned-authoritative-mirror",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	interview := &models.FlowInterview{}
	_, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		interview,
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("executeActionCall(transferConnectorRemoteIdentity) error = %v, want durable conflict", err)
	}
	if len(persistence.updates) != 0 {
		t.Fatalf("target update count = %d, want 0 when the target owns durable authority", len(persistence.updates))
	}
	if got := persistence.lockRecords["Contact/003target"]["External_Id__c"]; got != "target-owned-authoritative-mirror" {
		t.Fatalf("Contact.External_Id__c after durable conflict = %#v, want preserved authoritative mirror", got)
	}
	if _, ok := interview.ContextData["remoteIdentityTransferred"]; ok {
		t.Fatalf("remoteIdentityTransferred = %#v, want no output on durable conflict", interview.ContextData["remoteIdentityTransferred"])
	}
}

func TestTransferConnectorRemoteIdentityActionContinuesWhenTargetAuthorityHasBlankMirror(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	store.remoteID = ""
	store.targetHasAuthority = true
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id": "00Qsource",
			},
			"Contact/003target": {
				"Id": "003target",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	interview := &models.FlowInterview{}
	next, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		interview,
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err != nil {
		t.Fatalf("executeActionCall(transferConnectorRemoteIdentity): %v", err)
	}
	if next != "PreviewSourceBeforeSaveUpdate" {
		t.Fatalf("next = %q, want PreviewSourceBeforeSaveUpdate", next)
	}
	if len(persistence.updates) != 0 {
		t.Fatalf("target update count = %d, want 0 when target mirror is blank", len(persistence.updates))
	}
	if len(store.transfers) != 1 || store.transfers[0].TargetRemoteID != "" {
		t.Fatalf("transfer target mirror = %#v, want one blank target mirror", store.transfers)
	}
	if got, ok := interview.ContextData["remoteIdentityTransferred"].(bool); !ok || got {
		t.Fatalf("remoteIdentityTransferred = %#v (bool=%t), want false", got, ok)
	}
}

func TestTransferConnectorRemoteIdentityActionFailsClosedOnRemoteIDMismatch(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id":             "00Qsource",
				"External_Id__c": "remote-229009332242",
			},
			"Contact/003target": {
				"Id":             "003target",
				"External_Id__c": "different-remote",
			},
		},
	}
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	_, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		&models.FlowInterview{},
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err == nil {
		t.Fatal("executeActionCall(transferConnectorRemoteIdentity) error = nil, want mismatch failure")
	}
	if !strings.Contains(err.Error(), "target Contact/003target External_Id__c") {
		t.Fatalf("error = %q, want target remote-id mismatch detail", err.Error())
	}
	if len(store.transfers) != 1 {
		t.Fatalf("transfer count = %d, want 1 before atomic rollback on target mirror mismatch", len(store.transfers))
	}
}

func TestTransferConnectorRemoteIdentityActionDoesNotStampTargetWhenTransferFails(t *testing.T) {
	store := newTransferRemoteIdentityDurableStore()
	store.err = errors.New("no transferable authority")
	persistence := &inboundRegisterPersistenceStub{
		txActive: true,
		lockRecords: map[string]models.SObject{
			"Lead/00Qsource": {
				"Id":             "00Qsource",
				"External_Id__c": "remote-229009332242",
			},
			"Contact/003target": {
				"Id": "003target",
			},
		},
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = persistence
	exec.RegisterFlowActionCalls(ConnectorPlatformFlowActionCallRegistrations()...)

	_, err := exec.executeActionCall(
		transferConnectorRemoteIdentityCtx(),
		&models.FlowInterview{},
		&models.FlowMetadata{},
		transferConnectorRemoteIdentityAction("00Qsource", "003target"),
		&models.UserSession{ID: "005-user"},
	)
	if err == nil {
		t.Fatal("executeActionCall(transferConnectorRemoteIdentity) error = nil, want transfer failure")
	}
	if len(persistence.updates) != 0 {
		t.Fatalf("target update count = %d, want 0 when durable transfer fails", len(persistence.updates))
	}
	if got := persistence.lockRecords["Contact/003target"]["External_Id__c"]; got != nil {
		t.Fatalf("Contact.External_Id__c after failed transfer = %#v, want nil", got)
	}
}

// TestRegisterConnectorInboundNoSyncOriginMintsNothing verifies that a save
// without a connector sync origin (user/admin/import) produces no evidence row.
func TestRegisterConnectorInboundNoSyncOriginMintsNothing(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// No sync-origin in context — must not mint.
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(context.Background(), interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts for non-connector-origin save, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInboundOwnerKeyOriginMintsNothing verifies the ctx gate is
// keyed on the connector NAMESPACE (the value the inbound worker actually stamps,
// connector_inbound/worker.go), NOT the pack owner key. A sync origin equal to
// connectorOwnerKey but not the namespace must mint nothing — this guards against
// regressing the gate to connectorOwnerKey, which would never match in production
// where the namespace differs from the pack owner key and would silently skip
// authorship registration.
func TestRegisterConnectorInboundOwnerKeyOriginMintsNothing(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Origin = the owner key, NOT the namespace the worker stamps → gate must reject.
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(connectorInboundOriginCtx("my-connector-pack"), interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts when sync origin matches owner key but not namespace, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInboundConnectorOriginMintsEvidence verifies that a
// connector-origin in-flow save mints one authored evidence row with the
// engine-confirmed local record id and the stamped remote id.
func TestRegisterConnectorInboundConnectorOriginMintsEvidence(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// v3: GetLock is called for the single-record registry path; pre-seed the
	// committed record so the persistence stub returns the full snapshot.
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-remote-1"},
	})

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	// Set the engine-owned snapshot (what executeRecordCreate publishes after a
	// successful persistence.Insert). Identity (recordId) comes from here, NOT $Record.
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 1 {
		t.Fatalf("expected 1 evidence insert, got %d", len(store.inserted))
	}
	ev := store.inserted[0]
	if ev.SourceRecordId != "003local" {
		t.Errorf("SourceRecordId = %q, want 003local", ev.SourceRecordId)
	}
	if ev.RemoteId != "ext-remote-1" {
		t.Errorf("RemoteId = %q, want ext-remote-1", ev.RemoteId)
	}
	if ev.SourceObject != "Contact" {
		t.Errorf("SourceObject = %q, want Contact", ev.SourceObject)
	}
	if ev.RemoteIDField != "External_Id__c" {
		t.Errorf("RemoteIDField = %q, want External_Id__c", ev.RemoteIDField)
	}
	if strings.TrimSpace(ev.ConnectorOwnerKey) == "" {
		t.Errorf("ConnectorOwnerKey must not be blank")
	}
	if len(store.laneIdentities) != 1 || store.laneIdentities[0].DestinationSlotID != "remote-contact" || store.laneIdentities[0].SourceRecordID != "003local" {
		t.Fatalf("engine-derived lane identity = %#v", store.laneIdentities)
	}
}

func TestRegisterConnectorInboundExactFinalizedHistoryDoesNotAuthorEvidence(t *testing.T) {
	store := newInboundRegisterDurableStore()
	store.alreadyFinalized = true
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-remote-1"},
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey": "my-connector-pack", "connectorNamespace": "my_connector",
		"sourceObject": "Contact", "remoteIDField": "External_Id__c",
	})
	ctx := withEngineSnapshot(connectorInboundOriginCtx("my_connector"), "Contact", models.SObject{"Id": "003local"})
	if _, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, inboundRegisterInterview(nil), action); err != nil {
		t.Fatalf("exact finalized retry: %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("exact finalized retry inserted evidence %#v", store.inserted)
	}
}

func TestRegisterConnectorInboundNonfinalizedHistoryFailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	store.historyErr = ErrConnectorPushDurableConflict
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-remote-1"},
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey": "my-connector-pack", "connectorNamespace": "my_connector",
		"sourceObject": "Contact", "remoteIDField": "External_Id__c",
	})
	ctx := withEngineSnapshot(connectorInboundOriginCtx("my_connector"), "Contact", models.SObject{"Id": "003local"})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, inboundRegisterInterview(nil), action)
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("nonfinalized history error = %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("nonfinalized history inserted evidence %#v", store.inserted)
	}
}

// TestRegisterConnectorInboundIdempotentSameRemoteId verifies that calling the
// action with an already-active evidence row carrying the same remote id is a
// no-op (no new insert, no retire).
func TestRegisterConnectorInboundIdempotentSameRemoteId(t *testing.T) {
	store := newInboundRegisterDurableStore()
	store.seedActive("my-connector-pack", "my_connector", "Contact", "003local", "External_Id__c", "ext-remote-1")
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-remote-1"},
	})

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 new inserts for same remote id (idempotent), got %d", len(store.inserted))
	}
	if len(store.retired) != 0 {
		t.Fatalf("expected 0 retirements for same remote id, got %d", len(store.retired))
	}
}

// TestRegisterConnectorInboundConflictingRemoteIdRetires verifies that when an
// active evidence row exists with a DIFFERENT remote id, the action retires that
// existing row and inserts nothing, leaving no active evidence (outbound fails closed).
func TestRegisterConnectorInboundConflictingRemoteIdRetires(t *testing.T) {
	store := newInboundRegisterDurableStore()
	seeded := store.seedActive("my-connector-pack", "my_connector", "Contact", "003local", "External_Id__c", "old-remote-id")
	// GetLock returns the committed record with the NEW remote id (the inbound write
	// updated the field before the register action runs).
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "new-remote-id"},
	})

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "new-remote-id", // different from seeded
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
		"Id":             "003local",
		"External_Id__c": "new-remote-id",
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 new inserts for conflict case (retire only), got %d", len(store.inserted))
	}
	if len(store.retired) != 1 || store.retired[0] != seeded.ID {
		t.Fatalf("expected RetireEvidence(%q), got %v", seeded.ID, store.retired)
	}
	if store.evidence[seeded.ID].EvidenceStatus != ConnectorInboundEvidenceStatusRetired {
		t.Errorf("seeded evidence status = %q, want %q", store.evidence[seeded.ID].EvidenceStatus, ConnectorInboundEvidenceStatusRetired)
	}
}

// TestRegisterConnectorInboundInsertErrorIsFatal verifies a non-conflict durable
// store failure is returned as an error (not swallowed), so the inbound atomic
// flow rolls back and the record never commits without authorship evidence.
func TestRegisterConnectorInboundInsertErrorIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	store.insertErr = errors.New("db unavailable")
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-1"},
	})

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected a fatal error when InsertAuthoredEvidence fails; the record must roll back, not commit without evidence")
	}
}

// TestRegisterConnectorInboundFindErrorIsFatal verifies an evidence lookup
// failure is fatal (not swallowed as success-without-minting).
func TestRegisterConnectorInboundFindErrorIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	store.findActiveErr = errors.New("db unavailable")
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-1"},
	})

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected a fatal error when FindActiveInboundRemoteIDEvidence fails")
	}
}

// TestRegisterConnectorInboundAdvancesFlowToNextElement verifies the action
// returns the next-element reference (not an empty string) so the flow loop does
// not treat it as completion and silently skip later flow elements — on both the
// connector-origin (mints) path and the no-origin (no-op) path.
func TestRegisterConnectorInboundAdvancesFlowToNextElement(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-1"},
	})

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})
	action.Connector = &models.FlowConnector{TargetReference: "NextElement"}

	originCtx := connectorInboundOriginCtx("my_connector")
	originCtx = withEngineSnapshot(originCtx, "Contact", models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	next, err := exec.executeRegisterConnectorInboundRemoteAuthorship(originCtx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if next != "NextElement" {
		t.Fatalf("connector-origin next = %q, want NextElement (must advance the flow)", next)
	}

	nextNoOrigin, err := exec.executeRegisterConnectorInboundRemoteAuthorship(context.Background(), interview, action)
	if err != nil {
		t.Fatalf("unexpected error (no origin): %v", err)
	}
	if nextNoOrigin != "NextElement" {
		t.Fatalf("no-origin next = %q, want NextElement (no-op must still advance the flow)", nextNoOrigin)
	}
}

// TestRegisterConnectorInboundConflict verifies the conflict-reread behavior when
// a CONCURRENT transaction committed a racing evidence row (Insert returns a
// unique conflict and the *sql.DB reread reveals the racing row): a same-value
// row is idempotent and must NOT be retired (Gate 0 P1), while a different-value
// row is retired so outbound fails closed.
func TestRegisterConnectorInboundConflict(t *testing.T) {
	tests := []struct {
		name           string
		racingID       string
		racingRemoteID string
		wantRetired    []string
		failMessage    string
	}{
		{
			name:           "SameValueIsIdempotent",
			racingID:       "racing-same",
			racingRemoteID: "ext-remote-1",
			failMessage:    "same-value conflict must be idempotent (no retire)",
		},
		{
			name:           "DifferentValueRetires",
			racingID:       "racing-diff",
			racingRemoteID: "other-remote",
			wantRetired:    []string{"racing-diff"},
			failMessage:    "different-value conflict must retire the racing row",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			store := newInboundRegisterDurableStore()
			store.insertErr = ErrConnectorPushDurableConflict
			store.racingRow = &ConnectorInboundRemoteIDEvidence{
				ID:                tt.racingID,
				ConnectorOwnerKey: inboundRegisterTestOwnerPack,
				SourceObject:      "Contact",
				SourceRecordId:    "003local",
				RemoteIDField:     "External_Id__c",
				RemoteId:          tt.racingRemoteID,
				EvidenceStatus:    ConnectorInboundEvidenceStatusAuthored,
				ActiveKey:         ConnectorInboundEvidenceActiveKey(inboundRegisterTestOwnerPack, "Contact", "003local", "External_Id__c"),
			}
			exec := newInboundRegisterExec(store, map[string]models.SObject{
				"Contact/003local": {"Id": "003local", "External_Id__c": "ext-remote-1"},
			})

			interview := inboundRegisterInterview(models.SObject{
				"Id":             "003local",
				"External_Id__c": "ext-remote-1",
			})
			action := inboundRegisterAction(map[string]string{
				"connectorOwnerKey":  "my-connector-pack",
				"connectorNamespace": "my_connector",
				"sourceObject":       "Contact",
				"remoteIDField":      "External_Id__c",
			})

			ctx := connectorInboundOriginCtx("my_connector")
			ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
				"Id":             "003local",
				"External_Id__c": "ext-remote-1",
			})
			_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
			if err != nil {
				t.Fatalf("unexpected error: %v", err)
			}
			if len(store.retired) != len(tt.wantRetired) {
				t.Fatalf("%s, got retired=%v", tt.failMessage, store.retired)
			}
			for i, want := range tt.wantRetired {
				if store.retired[i] != want {
					t.Fatalf("%s, got retired=%v", tt.failMessage, store.retired)
				}
			}
		})
	}
}

// TestRegisterConnectorInboundCustomizedFlowIsFatal verifies that a flow running
// under a matching sync origin but that is NOT a pack-owned uncustomized standard
// FlowVersion cannot author evidence — the trusted-owner binding fails closed.
func TestRegisterConnectorInboundCustomizedFlowIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundCtxWithIdentity("my_connector", FlowExecutionIdentity{
		APIName:                "Connector_Inbound_Register",
		OwnerPack:              "my-connector-pack",
		DefinitionIsCustomized: true,
		VersionSource:          flowSourceStandard,
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected a fatal error when the executing Flow is customized; evidence must not be authored")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("customized flow must mint nothing, got %d inserts", len(store.inserted))
	}
}

// TestRegisterConnectorInboundForeignOwnerFlowIsFatal verifies that a flow whose
// trusted OwnerPack differs from the supplied connectorOwnerKey cannot author
// evidence under another connector's owner — the binding fails closed (Gate 0 P1).
func TestRegisterConnectorInboundForeignOwnerFlowIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundCtxWithIdentity("my_connector", FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		OwnerPack:     "other-connector-pack",
		VersionSource: flowSourceStandard,
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected a fatal error when connectorOwnerKey does not match the executing Flow owner pack")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("foreign-owner flow must mint nothing, got %d inserts", len(store.inserted))
	}
}

// TestRegisterConnectorInboundOutsideAtomicBoundaryIsFatal verifies the action
// fails closed when it runs outside an atomic Flow boundary — otherwise the
// store's s.db fallback would write evidence decoupled from the inbound record's
// rollback (Gate 0 P0).
func TestRegisterConnectorInboundOutsideAtomicBoundaryIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Matching sync origin + pack-owned standard identity + inbound-execution
	// marker, but NO atomic boundary — must fail closed at Step 1c.
	base := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "my_connector")
	base = constants.WithConnectorInboundFlowExecution(base, "Connector_Inbound_Register", "my_connector")
	ctx := withFlowRuntimeContext(base, flowRuntimeContextFromCtx(base).WithCurrentFlow(FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		OwnerPack:     "my-connector-pack",
		VersionSource: flowSourceStandard,
	}))
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected a fatal error when executing outside an atomic Flow boundary")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("non-atomic execution must mint nothing, got %d inserts", len(store.inserted))
	}
}

// TestRegisterConnectorInboundBlankSourceObjectIsFatal verifies a blank
// pack-declared sourceObject fails closed (mirrors applyConnectorFieldMappings).
func TestRegisterConnectorInboundBlankSourceObjectIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "",
		"remoteIDField":      "External_Id__c",
	})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(connectorInboundOriginCtx("my_connector"), interview, action)
	if err == nil {
		t.Fatal("expected a fatal error for blank sourceObject")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("blank sourceObject must mint nothing, got %d inserts", len(store.inserted))
	}
}

// TestRegisterConnectorInboundBlankRemoteIDFieldIsFatal verifies a blank
// pack-declared remoteIDField fails closed before any $Record read or store op.
func TestRegisterConnectorInboundBlankRemoteIDFieldIsFatal(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "",
	})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(connectorInboundOriginCtx("my_connector"), interview, action)
	if err == nil {
		t.Fatal("expected a fatal error for blank remoteIDField")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("blank remoteIDField must mint nothing, got %d inserts", len(store.inserted))
	}
}

// ---------------------------------------------------------------------------
// Inbound-execution guard (Step 1b-ii) tests
// ---------------------------------------------------------------------------

// TestRegisterConnectorInbound_InboundExecutionAuthorizes verifies that a ctx
// carrying both the connector sync-origin AND the WithConnectorInboundFlowExecution
// marker for the executing flow + namespace (as flow_adapter.go stamps on the
// genuine inbound-worker path) successfully authors one evidence row.
func TestRegisterConnectorInbound_InboundExecutionAuthorizes(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-remote-1"},
	})

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// connectorInboundOriginCtx already stamps WithConnectorInboundFlowExecution
	// for "Connector_Inbound_Register" + "my_connector" (mirrors flow_adapter.go).
	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 1 {
		t.Fatalf("expected 1 evidence insert for genuine inbound-execution ctx, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInbound_NoInboundExecutionMarker_FailsClosed verifies that
// a ctx with the correct sync-origin but WITHOUT WithConnectorInboundFlowExecution
// (e.g. a non-flow_adapter caller) FAILS CLOSED — the new defense-in-depth guard
// (Step 1b-ii) errors rather than letting the stamped-remote-id record strand.
func TestRegisterConnectorInbound_NoInboundExecutionMarker_FailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Build ctx with sync-origin + trusted identity + atomic boundary, but
	// deliberately omit WithConnectorInboundFlowExecution (the new guard's key).
	ctx := connectorInboundCtxWithIdentity("my_connector", FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		DefinitionID:  "flow-def-register",
		VersionID:     "flow-version-register",
		OwnerPack:     inboundRegisterTestOwnerPack,
		VersionSource: flowSourceStandard,
	})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatalf("expected fail-closed error when inbound-execution marker is absent, got nil")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts when inbound-execution marker is absent, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInbound_CrossFlowInboundMarker_FailsClosed verifies that a
// ctx whose WithConnectorInboundFlowExecution was stamped for a DIFFERENT root
// flow name (rootFlow mismatch) FAILS CLOSED — the guard is root-flow-scoped.
func TestRegisterConnectorInbound_CrossFlowInboundMarker_FailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Stamp the inbound marker for a DIFFERENT flow name — rootFlow mismatch.
	base := connectorInboundCtxWithIdentity("my_connector", FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		DefinitionID:  "flow-def-register",
		VersionID:     "flow-version-register",
		OwnerPack:     inboundRegisterTestOwnerPack,
		VersionSource: flowSourceStandard,
	})
	ctx := constants.WithConnectorInboundFlowExecution(base, "Connector_Inbound_Other_Flow", "my_connector")
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatalf("expected fail-closed error when inbound marker has mismatched flow name, got nil")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts when inbound marker has mismatched flow name, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInbound_CrossPackInboundMarker_FailsClosed verifies that a
// ctx whose WithConnectorInboundFlowExecution was stamped with a DIFFERENT
// namespace (ns mismatch) FAILS CLOSED — the guard enforces same-pack isolation.
func TestRegisterConnectorInbound_CrossPackInboundMarker_FailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Stamp the inbound marker with a DIFFERENT namespace — ns mismatch.
	base := connectorInboundCtxWithIdentity("my_connector", FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		DefinitionID:  "flow-def-register",
		VersionID:     "flow-version-register",
		OwnerPack:     inboundRegisterTestOwnerPack,
		VersionSource: flowSourceStandard,
	})
	ctx := constants.WithConnectorInboundFlowExecution(base, "Connector_Inbound_Register", "other_connector")
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatalf("expected fail-closed error when inbound marker has mismatched namespace, got nil")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts when inbound marker has mismatched namespace, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInboundNoSnapshotIsNoOp verifies that a trusted inbound
// flow with no source-object write preserves the established no-op behavior.
func TestRegisterConnectorInboundNoSnapshotIsNoOp(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := newInboundRegisterExec(store, nil)

	// $Record is present and has valid-looking data, but the engine channel
	// carries no snapshot — the action must NOT read from $Record.
	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// connectorInboundOriginCtx passes all trusted-inbound gates but records no
	// source-object write and sets no engine snapshot. The action must not fall
	// back to $Record.
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(connectorInboundOriginCtx("my_connector"), interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts when no source-object write occurred, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInboundWrittenWithoutSnapshotFailsClosed verifies that
// a source-object write cannot be silently accepted when its engine snapshot is
// absent. The write outcome is engine-owned and cannot be forged by flow data.
func TestRegisterConnectorInboundWrittenWithoutSnapshotFailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := newInboundRegisterExec(store, nil)
	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})
	ctx := connectorInboundOriginCtx("my_connector")
	flowRuntimeContextFromCtx(ctx).MarkPostUpsertWrite("Contact")

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected missing engine snapshot after a source-object write to fail closed, got nil")
	}
	if !strings.Contains(err.Error(), "engine wrote") || !strings.Contains(err.Error(), "no engine-confirmed post-upsert record") {
		t.Errorf("error = %q, want missing engine snapshot after write error", err.Error())
	}
	if len(store.inserted) != 0 {
		t.Fatalf("expected 0 evidence inserts when engine snapshot is absent, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInboundEngineSnapshotWinsOverContextData is the key
// security test: when the engine snapshot carries one record identity and
// interview.ContextData["$Record"] carries a DIFFERENT (forged) identity,
// the authored evidence must use the ENGINE snapshot — not ContextData.
func TestRegisterConnectorInboundEngineSnapshotWinsOverContextData(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// v3: GetLock is called with the engine-confirmed recordId (003real-local).
	// The committed record carries the real remoteId — not the forged one.
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003real-local": {"Id": "003real-local", "External_Id__c": "real-remote-id"},
	})

	// Forged $Record: a flow author set a different Id and remoteID trying to
	// misbind authorship evidence to a different local row.
	forgedRecord := models.SObject{
		"Id":             "FORGED-local-id",
		"External_Id__c": "FORGED-remote-id",
	}
	interview := inboundRegisterInterview(forgedRecord)

	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  "my-connector-pack",
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	// Engine snapshot carries the REAL identity (from the actual persistence.Insert).
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
		"Id":             "003real-local",
		"External_Id__c": "real-remote-id",
	})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if len(store.inserted) != 1 {
		t.Fatalf("expected 1 evidence insert, got %d", len(store.inserted))
	}
	ev := store.inserted[0]
	if ev.SourceRecordId != "003real-local" {
		t.Errorf("SourceRecordId = %q (from $Record FORGED-local-id was used), want 003real-local (engine snapshot)", ev.SourceRecordId)
	}
	if ev.RemoteId != "real-remote-id" {
		t.Errorf("RemoteId = %q (from $Record FORGED-remote-id was used), want real-remote-id (engine snapshot)", ev.RemoteId)
	}
}

// (Removed Gate-0 v6: the case-0 trigger-snapshot fallback in
// registerConnectorInboundRemoteAuthorship was unreachable dead code — the #2089
// inbound guard and the trigger-snapshot precondition are mutually exclusive — so
// TestRegisterConnectorInboundTriggerSnapshotFallback and
// ...ObjectMismatchIsNoOp were deleted along with the fallback.)

// TestRegisterConnectorInboundRegistrySpansSubflow verifies that a record added
// to the post-upsert registry on a parent FlowRuntimeContext is visible on a
// derived child context produced by PushCallStack (P1-2: registry must survive
// the per-subflow executionState reset).
func TestRegisterConnectorInboundRegistrySpansSubflow(t *testing.T) {
	baseCtx := context.Background()
	parent := NewFlowRuntimeContext(baseCtx, nil)
	parent.AddPostUpsertRecord("Contact", models.SObject{
		"Id":             "003-parent",
		"External_Id__c": "ext-parent",
	})

	// Simulate PushCallStack — child gets a fresh executionState but shares postUpsert.
	child := parent.PushCallStack("child_flow_api")

	recs := child.PostUpsertRecords("Contact")
	if len(recs) != 1 {
		t.Fatalf("PostUpsertRecords(Contact) on child = %d entries, want 1 (registry must span subflow reset)", len(recs))
	}
	if id, _ := recs[0]["Id"].(string); id != "003-parent" {
		t.Errorf("PostUpsertRecords[0].Id = %q, want 003-parent", id)
	}
}

// TestRegisterConnectorInboundAmbiguousTwoRecordsFails verifies that when two
// DISTINCT records of the same object are added to the registry (e.g. two
// executeRecordCreate calls for "Contact"), the register action fails closed
// instead of arbitrarily picking one (P1-3: fail-closed ambiguity).
// v3: ambiguity is based on DISTINCT recordIds; the >1 case fails before GetLock.
func TestRegisterConnectorInboundAmbiguousTwoRecordsFails(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// newInboundRegisterExec sets txActive:true; the ambiguity check fires before
	// GetLock, so lockRecords can be empty.
	exec := newInboundRegisterExec(store, nil)

	interview := inboundRegisterInterview(models.SObject{})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// connectorInboundOriginCtx sets up sync origin + trusted flow identity + atomic boundary.
	ctx := connectorInboundOriginCtx("my_connector")

	// Add two DISTINCT Contact records — simulates two executeRecordCreate calls in the same execution.
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003-first", "External_Id__c": "ext-1"})
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003-second", "External_Id__c": "ext-2"})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected ambiguity error when two Contact records are in the registry, got nil")
	}
	if !strings.Contains(err.Error(), "ambiguous source record") {
		t.Errorf("error = %q, want substring 'ambiguous source record'", err.Error())
	}
	if len(store.inserted) != 0 {
		t.Errorf("expected 0 evidence inserts on ambiguity, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInboundSameRecordMultiUpdateNotAmbiguous verifies that
// two registry snapshots with the SAME recordId (chained $Record updates — a
// supported flow shape) are treated as a single record, not as ambiguous.
// This is the P1-B regression test: v2 counted raw snapshots, so same-record
// multi-update triggered the ambiguity error; v3 deduplicates by recordId first.
func TestRegisterConnectorInboundSameRecordMultiUpdateNotAmbiguous(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// GetLock returns the committed record (with remoteIDField) for the deduped recordId.
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003-same": {"Id": "003-same", "External_Id__c": "ext-same"},
	})

	interview := inboundRegisterInterview(models.SObject{})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	// Two snapshots with the SAME Id — simulates applyRecordUpdatePersistence
	// being called twice for the same record (chained updates).
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003-same"})                        // partial: no remoteIDField
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003-same", "Status__c": "Active"}) // another partial update

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error for same-record multi-update: %v", err)
	}
	if len(store.inserted) != 1 {
		t.Fatalf("expected exactly 1 evidence insert for same-record multi-update, got %d", len(store.inserted))
	}
	ev := store.inserted[0]
	if ev.SourceRecordId != "003-same" {
		t.Errorf("SourceRecordId = %q, want 003-same", ev.SourceRecordId)
	}
	if ev.RemoteId != "ext-same" {
		t.Errorf("RemoteId = %q, want ext-same (from GetLock committed record)", ev.RemoteId)
	}
}

// TestRegisterConnectorInboundUpdateWithoutRemoteIDFieldStillAuthors verifies
// that when the registry snapshot for the single matched record does NOT contain
// remoteIDField (a partial update snapshot that only wrote other fields), the
// action still authors evidence by reading remoteIDField from the GetLock
// committed record.
// This is the P1-A regression test: v2 read remoteIDField from the snapshot
// and silently no-op'd when the snapshot lacked it; v3 reads from GetLock.
func TestRegisterConnectorInboundUpdateWithoutRemoteIDFieldStillAuthors(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// The committed record has remoteIDField; the registry snapshot does not.
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003-partial": {"Id": "003-partial", "External_Id__c": "ext-committed"},
	})

	interview := inboundRegisterInterview(models.SObject{})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	// Registry snapshot has Id but NO External_Id__c — simulates an update that
	// only wrote other fields (e.g. Status__c) without re-writing the remote id.
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{
		"Id":        "003-partial",
		"Status__c": "Active",
		// External_Id__c intentionally absent
	})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err != nil {
		t.Fatalf("unexpected error when registry snapshot lacks remoteIDField: %v", err)
	}
	if len(store.inserted) != 1 {
		t.Fatalf("expected 1 evidence insert (from GetLock committed record), got %d", len(store.inserted))
	}
	ev := store.inserted[0]
	if ev.SourceRecordId != "003-partial" {
		t.Errorf("SourceRecordId = %q, want 003-partial", ev.SourceRecordId)
	}
	if ev.RemoteId != "ext-committed" {
		t.Errorf("RemoteId = %q, want ext-committed (read from GetLock, not from partial snapshot)", ev.RemoteId)
	}
}

// ---------------------------------------------------------------------------
// P1-A: per-execution isolation tests
// ---------------------------------------------------------------------------

// TestRegisterConnectorInboundPerExecutionIsolation verifies that two root
// flows dispatched from a shared base ctx each get an independent
// postUpsertRegistry via CloneWithFreshPostUpsert — root A's snapshot must NOT
// be visible to root B's register action.
func TestRegisterConnectorInboundPerExecutionIsolation(t *testing.T) {
	baseCtx := context.Background()
	baseCtx = withFlowRuntimeContext(baseCtx, NewFlowRuntimeContext(baseCtx, nil))

	// Simulate the dispatch loop in flow_trigger_listener.go:
	// each root flow gets a CloneWithFreshPostUpsert.
	rootACtx := withFlowRuntimeContext(baseCtx, flowRuntimeContextFromCtx(baseCtx).CloneWithFreshPostUpsert())
	rootBCtx := withFlowRuntimeContext(baseCtx, flowRuntimeContextFromCtx(baseCtx).CloneWithFreshPostUpsert())

	// Root A publishes a record.
	flowRuntimeContextFromCtx(rootACtx).AddPostUpsertRecord("Contact", models.SObject{
		"Id":             "003-root-a",
		"External_Id__c": "ext-a",
	})

	// Root B must not see root A's record.
	recsB := flowRuntimeContextFromCtx(rootBCtx).PostUpsertRecords("Contact")
	if len(recsB) != 0 {
		t.Fatalf("root B PostUpsertRecords(Contact) = %d entries, want 0 (CloneWithFreshPostUpsert must isolate)", len(recsB))
	}

	// Root A's own registry is unaffected.
	recsA := flowRuntimeContextFromCtx(rootACtx).PostUpsertRecords("Contact")
	if len(recsA) != 1 {
		t.Fatalf("root A PostUpsertRecords(Contact) = %d entries, want 1", len(recsA))
	}
}

// TestRegisterConnectorInboundSubflowSharingWithinRoot verifies that within one
// root flow, a child context produced by PushCallStack shares the registry —
// so a record added in a subflow IS visible to the root's register action.
func TestRegisterConnectorInboundSubflowSharingWithinRoot(t *testing.T) {
	baseCtx := context.Background()
	rootFRC := NewFlowRuntimeContext(baseCtx, nil)

	// Simulate one root flow dispatch (CloneWithFreshPostUpsert).
	rootFRC = rootFRC.CloneWithFreshPostUpsert()
	ctx := withFlowRuntimeContext(baseCtx, rootFRC)

	// Simulate a subflow pushing onto the call stack (PushCallStack returns a
	// *FlowRuntimeContext that shares postUpsert via clone()).
	subflowFRC := rootFRC.PushCallStack("SubFlow_API")

	// Subflow publishes a record to its registry.
	subflowFRC.AddPostUpsertRecord("Contact", models.SObject{
		"Id":             "003-sub",
		"External_Id__c": "ext-sub",
	})

	// The root FRC's register action (which uses flowRuntimeContextFromCtx(ctx),
	// i.e. the root pointer) must see the subflow's record because they share
	// the same *postUpsertRegistry pointer.
	recsFromRoot := flowRuntimeContextFromCtx(ctx).PostUpsertRecords("Contact")
	if len(recsFromRoot) != 1 {
		t.Fatalf("root PostUpsertRecords(Contact) after subflow add = %d entries, want 1 (subflow must share registry)", len(recsFromRoot))
	}
	if id, _ := recsFromRoot[0]["Id"].(string); id != "003-sub" {
		t.Errorf("root PostUpsertRecords[0].Id = %q, want 003-sub", id)
	}
}

// TestRegisterConnectorInbound_BlankConnectorNamespace_FailsClosed verifies
// that a blank connectorNamespace (pack misconfiguration) fails closed — Fix C.
// Without the fix, connectorSyncOriginMatchesOwner returns false for any blank
// namespace (ownerPack != "" guard), so the function silently no-ops before
// reaching the Step 1d blank-key checks. Fix C adds an early fail-closed check
// before the ctx gate so blank namespace always errors regardless of origin.
func TestRegisterConnectorInbound_BlankConnectorNamespace_FailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)

	interview := inboundRegisterInterview(models.SObject{
		"Id":             "003local",
		"External_Id__c": "ext-remote-1",
	})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "", // blank — pack misconfiguration
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// The error fires before the ctx gate, so context.Background() is sufficient.
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(context.Background(), interview, action)
	if err == nil {
		t.Fatal("expected a fatal error for blank connectorNamespace (pack misconfiguration must fail closed)")
	}
	if !strings.Contains(err.Error(), "connectorNamespace") {
		t.Errorf("error = %q, want substring 'connectorNamespace'", err.Error())
	}
	if len(store.inserted) != 0 {
		t.Fatalf("blank connectorNamespace must mint nothing, got %d inserts", len(store.inserted))
	}
}

// ---------------------------------------------------------------------------
// GetLock committed-truth path tests (carve-down #2080)
// ---------------------------------------------------------------------------

// TestRegisterConnectorInbound_GetLockError_FailsClosed verifies that when
// GetLock returns an error (db unavailable, lock timeout, etc.) the register
// action returns a fatal error — the inbound record rolls back rather than
// committing without authorship evidence.
func TestRegisterConnectorInbound_GetLockError_FailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	stub := &inboundRegisterPersistenceStub{
		lockErr:  errors.New("db lock timeout"),
		txActive: true,
	}
	exec := &FlowExecutor{}
	exec.SetConnectorPushDurableStore(store)
	exec.persistence = stub

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003local"})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected fatal error when GetLock returns an error; inbound record must roll back")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("GetLock error must prevent any evidence insert, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInbound_GetLockNilRecord_FailsClosed verifies that when
// GetLock returns (nil, nil) — the committed record is not found — the register
// action returns a fatal error rather than silently no-oping. A committed record
// that GetLocks as nil is a DB inconsistency; failing closed forces the inbound
// transaction to roll back.
func TestRegisterConnectorInbound_GetLockNilRecord_FailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// lockRecords is empty — GetLock returns (nil, nil) for any key.
	exec := newInboundRegisterExec(store, map[string]models.SObject{})

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003local"})
	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected fatal error when GetLock returns nil record; inbound record must roll back")
	}
	if len(store.inserted) != 0 {
		t.Fatalf("nil GetLock result must prevent any evidence insert, got %d", len(store.inserted))
	}
}

// TestRegisterConnectorInbound_GetLockCalledWithinAtomicBoundary verifies
// that register always runs within an atomic flow boundary (isWithinAtomicFlowBoundary
// guard at Step 1c) before GetLock is called. A call outside the atomic boundary
// fails closed at Step 1c before reaching GetLock — this is the structural guarantee
// that the GetLock + evidence write are atomic with the inbound record.
func TestRegisterConnectorInbound_GetLockCalledWithinAtomicBoundary(t *testing.T) {
	store := newInboundRegisterDurableStore()
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-1"},
	})

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-1"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Build ctx with matching sync-origin + trusted identity + inbound-execution
	// marker, but deliberately omit withAtomicFlowBoundary.
	base := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "my_connector")
	base = constants.WithConnectorInboundFlowExecution(base, "Connector_Inbound_Register", "my_connector")
	ctx := withFlowRuntimeContext(base, flowRuntimeContextFromCtx(base).WithCurrentFlow(FlowExecutionIdentity{
		APIName:       "Connector_Inbound_Register",
		OwnerPack:     inboundRegisterTestOwnerPack,
		VersionSource: flowSourceStandard,
	}))
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003local"})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected fatal error when register runs outside atomic boundary; GetLock must not be reached")
	}
	// Evidence must not be inserted — atomic boundary guard fires before GetLock.
	if len(store.inserted) != 0 {
		t.Fatalf("non-atomic execution must mint nothing; got %d inserts", len(store.inserted))
	}
}

// TestRegisterConnectorInbound_NoLiveTxFailsClosed verifies that the read-side
// gate added in #2080 rejects a call where all other guards pass (atomic
// boundary + inbound-execution marker + trusted owner) but the underlying
// PersistenceProvider has no live DB transaction (IsTransactionActive == false).
// This is the symmetric gate to publishPostUpsertRecord: the durable store falls
// back to s.db when no ambient tx is present, so atomicity is only guaranteed
// when IsTransactionActive is true.
func TestRegisterConnectorInbound_NoLiveTxFailsClosed(t *testing.T) {
	store := newInboundRegisterDurableStore()
	// newInboundRegisterExec sets txActive:true; override to false after
	// construction so all other happy-path conditions are met.
	exec := newInboundRegisterExec(store, map[string]models.SObject{
		"Contact/003local": {"Id": "003local", "External_Id__c": "ext-notx"},
	})
	exec.persistence.(*inboundRegisterPersistenceStub).txActive = false

	interview := inboundRegisterInterview(models.SObject{"Id": "003local", "External_Id__c": "ext-notx"})
	action := inboundRegisterAction(map[string]string{
		"connectorOwnerKey":  inboundRegisterTestOwnerPack,
		"connectorNamespace": "my_connector",
		"sourceObject":       "Contact",
		"remoteIDField":      "External_Id__c",
	})

	// Build ctx with atomic boundary + inbound-execution marker so the #2089
	// boundary guard and inbound guard both pass — only the IsTransactionActive
	// gate should fire.
	ctx := connectorInboundOriginCtx("my_connector")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{"Id": "003local"})

	_, err := exec.executeRegisterConnectorInboundRemoteAuthorship(ctx, interview, action)
	if err == nil {
		t.Fatal("expected fatal error when IsTransactionActive is false; register must fail closed without a live DB transaction")
	}
	if !strings.Contains(err.Error(), "live DB transaction") {
		t.Errorf("error = %q, want substring 'live DB transaction'", err.Error())
	}
	// Durable store must record zero inserts — gate fires before any write.
	if len(store.inserted) != 0 {
		t.Fatalf("no-live-tx path must mint nothing; got %d inserts", len(store.inserted))
	}
}