package automation

import (
	"context"
	"errors"
	"testing"

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

type inboundFinalizeDurableStore struct {
	*inboundRegisterDurableStore
	finalizations    []ConnectorInboundRemoteIdentityFinalization
	finalizeErr      error
	authority        *ConnectorRemoteIdentityAuthority
	authorityErr     error
	authorityQueries []ConnectorRemoteLocatorQuery
	laneIdentities   []ConnectorPushDurableIdentity
	laneRemoteIDs    []string
	historyErr       error
}

func newInboundFinalizeDurableStore() *inboundFinalizeDurableStore {
	return &inboundFinalizeDurableStore{inboundRegisterDurableStore: newInboundRegisterDurableStore()}
}

func (s *inboundFinalizeDurableStore) FinalizeInboundRemoteIdentityInAmbientTx(_ context.Context, finalization ConnectorInboundRemoteIdentityFinalization) error {
	s.finalizations = append(s.finalizations, finalization)
	return s.finalizeErr
}

func (s *inboundFinalizeDurableStore) LockConnectorRemoteIdentityLaneInAmbientTx(_ context.Context, identity ConnectorPushDurableIdentity, remoteID string) error {
	s.laneIdentities = append(s.laneIdentities, identity)
	s.laneRemoteIDs = append(s.laneRemoteIDs, remoteID)
	return nil
}

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

func (s *inboundFinalizeDurableStore) ResolveFinalizedRemoteLocator(_ context.Context, query ConnectorRemoteLocatorQuery) (*ConnectorRemoteIdentityAuthority, error) {
	s.authorityQueries = append(s.authorityQueries, query)
	if s.authorityErr != nil {
		return nil, s.authorityErr
	}
	if s.authority == nil {
		return nil, nil
	}
	copy := *s.authority
	return &copy, nil
}

func (s *inboundFinalizeDurableStore) RecordCompatibilityMirrorReadback(context.Context, string, string) error {
	panic("unexpected compatibility mirror readback")
}

type captureInboundFinalizeContractResolver struct {
	contract *connector.ConnectorPushDurableMetadataContract
	err      error
	queries  []ConnectorPushContractQuery
}

type captureInboundFinalizeAuthorityResolver struct {
	authority *ConnectorInboundRemoteIdentityFinalizationAuthority
	err       error
	queries   []ConnectorPushContractQuery
}

func (r *captureInboundFinalizeAuthorityResolver) ResolveConnectorInboundRemoteIdentityFinalizationAuthority(
	_ context.Context,
	query ConnectorPushContractQuery,
) (*ConnectorInboundRemoteIdentityFinalizationAuthority, error) {
	r.queries = append(r.queries, query)
	if r.err != nil {
		return nil, r.err
	}
	return r.authority, nil
}

type companionSubflowFinalizePersistence struct {
	*inboundRegisterPersistenceStub
	transactionRuns int
}

func (p *companionSubflowFinalizePersistence) Insert(_ context.Context, objectName string, record models.SObject, _ *models.UserSession) (models.SObject, error) {
	out := cloneRuntimeSObject(record)
	out[constants.FieldID] = "003-created-by-child"
	if p.lockRecords == nil {
		p.lockRecords = map[string]models.SObject{}
	}
	p.lockRecords[objectName+"/003-created-by-child"] = cloneRuntimeSObject(out)
	return out, nil
}

func (p *companionSubflowFinalizePersistence) IsTransactionActive(context.Context) bool {
	return p.txActive
}

func (p *companionSubflowFinalizePersistence) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	p.transactionRuns++
	p.txActive = true
	defer func() { p.txActive = false }()
	return fn(ctx)
}

func (r *captureInboundFinalizeContractResolver) ResolveConnectorPushContract(_ context.Context, query ConnectorPushContractQuery) (*connector.ConnectorPushDurableMetadataContract, error) {
	r.queries = append(r.queries, query)
	if r.err != nil {
		return nil, r.err
	}
	if r.contract == nil {
		return nil, nil
	}
	copy := *r.contract
	return &copy, nil
}

func connectorInboundFinalizeAction() *models.FlowActionCall {
	return &models.FlowActionCall{
		Name:       "FinalizeRemoteIdentity",
		ActionType: flowactions.ActionTypeFinalizeConnectorInboundRemoteIdentity,
		InputParameters: []models.FlowActionParameter{
			{Name: "targetFlowDeveloperName", Value: map[string]interface{}{"stringValue": "Connector_Push"}},
			{Name: "targetFlowActionName", Value: map[string]interface{}{"stringValue": "Push"}},
		},
		Connector: &models.FlowConnector{TargetReference: "Done"},
	}
}

func connectorInboundFinalizeContext(recordID string) context.Context {
	const flowName = "HubSpot_Contact_Lead_Upsert"
	const namespace = "hubspot-routing-namespace"
	ctx := connectorInboundCtxWithIdentity(namespace, FlowExecutionIdentity{
		APIName:       flowName,
		DefinitionID:  "flow-def-hubspot-inbound",
		VersionID:     "flow-version-hubspot-inbound",
		OwnerPack:     "connector_a",
		VersionSource: flowSourceStandard,
	})
	ctx = constants.WithConnectorInboundFlowExecution(ctx, flowName, namespace)
	return withEngineSnapshot(ctx, "Contact", models.SObject{constants.FieldID: recordID})
}

func newConnectorInboundFinalizeExecutor(store *inboundFinalizeDurableStore, contractResolver *captureInboundFinalizeContractResolver, records map[string]models.SObject) *FlowExecutor {
	exec := newInboundRegisterExec(store.inboundRegisterDurableStore, records)
	exec.SetConnectorPushDurableStore(store)
	exec.SetConnectorPushContractResolver(contractResolver)
	exec.connectorPushProfiles = staticPushProfileResolver{profile: &ConnectorPushProfile{
		OwnerPack:            "connector_a",
		ProfileName:          "default",
		CreatedPackInstallID: "install-active",
		CredentialRef:        "nc-main",
		EndpointClass:        "crm.objects.write",
		RetryPolicy:          ConnectorPushRetryPolicy{MaxAttempts: 1},
	}}
	exec.connectorPackMappingResolverFn = func(_ context.Context, query connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return []connector.ConnectorFieldMappingInput{{
			Source:       query.Source,
			SourceObject: query.SourceObject,
			TargetObject: query.TargetObject,
			ActionType:   query.ActionType,
			Direction:    connector.DirectionOutbound,
			SourceField:  "Email",
			TargetField:  "email",
		}}, nil
	}
	return exec
}

func TestFinalizeConnectorInboundRemoteIdentityActionTypeIsReserved(t *testing.T) {
	for _, actionType := range flowactions.PlatformReservedActionTypes() {
		if actionType == flowactions.ActionTypeFinalizeConnectorInboundRemoteIdentity {
			return
		}
	}
	t.Fatal("finalizeConnectorInboundRemoteIdentity must be platform reserved")
}

func TestFinalizeConnectorInboundRemoteIdentityUsesOneImmutableAuthorityWithoutLegacyFallback(t *testing.T) {
	contract := *defaultPushConnectorContract()
	authority := &ConnectorInboundRemoteIdentityFinalizationAuthority{
		Contract: contract,
		Profile: ConnectorPushProfile{
			OwnerPack:            "connector_a",
			ProfileName:          "default",
			CreatedPackInstallID: "profile-install-9",
			CredentialRef:        "nc-main",
			EndpointClass:        "crm.objects.write",
			RetryPolicy:          ConnectorPushRetryPolicy{MaxAttempts: 1},
		},
		OutboundMappings: []connector.ConnectorFieldMappingInput{{
			Source: "connector_a", SourceObject: "Contact",
			TargetObject: "remoteContact", ActionType: constants.ActionTypePushToConnector,
			Direction: connector.DirectionOutbound, SourceField: "Email", TargetField: "email",
		}},
	}
	resolver := &captureInboundFinalizeAuthorityResolver{authority: authority}
	legacy := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := &FlowExecutor{connectorPushContracts: legacy}
	exec.SetConnectorInboundRemoteIdentityFinalizationAuthorityResolver(resolver)

	gotContract, gotMappings, gotProfile, err := exec.resolveConnectorInboundFinalizeAuthority(
		context.Background(),
		FlowExecutionIdentity{OwnerPack: "connector_a"},
		connectorInboundFinalizeSelectors{
			targetFlowDeveloperName: "Connector_Push",
			targetFlowActionName:    "Push",
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if gotContract.ActionSlotID != contract.ActionSlotID ||
		gotProfile.CreatedPackInstallID != "profile-install-9" ||
		len(gotMappings) != 1 || gotMappings[0].SourceField != "Email" {
		t.Fatalf("resolved immutable authority = %#v, %#v, %#v", gotContract, gotProfile, gotMappings)
	}
	if len(resolver.queries) != 1 || len(legacy.queries) != 0 {
		t.Fatalf("immutable queries = %#v; legacy queries = %#v", resolver.queries, legacy.queries)
	}

	resolver.authority = nil
	if _, _, _, err := exec.resolveConnectorInboundFinalizeAuthority(
		context.Background(),
		FlowExecutionIdentity{OwnerPack: "connector_a"},
		connectorInboundFinalizeSelectors{
			targetFlowDeveloperName: "Connector_Push",
			targetFlowActionName:    "Push",
		},
	); err == nil {
		t.Fatal("missing immutable authority fell back to the legacy resolver")
	}
	if len(legacy.queries) != 0 {
		t.Fatalf("missing immutable authority used legacy queries: %#v", legacy.queries)
	}
}

func TestFinalizeConnectorInboundRemoteIdentityRequiresLiteralTargetSelectors(t *testing.T) {
	exec := &FlowExecutor{}
	action := &models.FlowActionCall{
		ActionType: flowactions.ActionTypeFinalizeConnectorInboundRemoteIdentity,
		InputParameters: []models.FlowActionParameter{
			{Name: "targetFlowDeveloperName", Value: map[string]interface{}{"elementReference": "$Record.TargetFlow__c"}},
			{Name: "targetFlowActionName", Value: map[string]interface{}{"stringValue": "Push"}},
		},
	}
	if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(context.Background(), nil, action, nil); err == nil {
		t.Fatal("dynamic target selector must fail closed")
	}
}

func TestParseConnectorInboundFinalizeSelectorsRejectsNonLiteralOrUnknownInputs(t *testing.T) {
	valid := func() []models.FlowActionParameter {
		return []models.FlowActionParameter{
			{Name: "targetFlowDeveloperName", Value: map[string]interface{}{"stringValue": "Connector_Push"}},
			{Name: "targetFlowActionName", Value: map[string]interface{}{"stringValue": "Push"}},
		}
	}
	for _, tc := range []struct {
		name   string
		mutate func(*models.FlowActionCall)
	}{
		{name: "raw string", mutate: func(action *models.FlowActionCall) { action.InputParameters[0].Value = "Connector_Push" }},
		{name: "element reference", mutate: func(action *models.FlowActionCall) {
			action.InputParameters[0].Value = map[string]interface{}{"elementReference": "$Record.Target__c"}
		}},
		{name: "flow interpolation", mutate: func(action *models.FlowActionCall) {
			action.InputParameters[0].Value = map[string]interface{}{"stringValue": "{!TargetFlow}"}
		}},
		{name: "unknown input", mutate: func(action *models.FlowActionCall) {
			action.InputParameters = append(action.InputParameters, models.FlowActionParameter{Name: "actionSlotId", Value: map[string]interface{}{"stringValue": "forged"}})
		}},
		{name: "duplicate input", mutate: func(action *models.FlowActionCall) {
			action.InputParameters = append(action.InputParameters, action.InputParameters[0])
		}},
		{name: "missing input", mutate: func(action *models.FlowActionCall) { action.InputParameters = action.InputParameters[:1] }},
		{name: "blank input", mutate: func(action *models.FlowActionCall) {
			action.InputParameters[1].Value = map[string]interface{}{"stringValue": " "}
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			action := &models.FlowActionCall{ActionType: flowactions.ActionTypeFinalizeConnectorInboundRemoteIdentity, InputParameters: valid()}
			tc.mutate(action)
			if _, err := parseConnectorInboundFinalizeSelectors(action); err == nil {
				t.Fatalf("%s must fail closed", tc.name)
			}
		})
	}
}

func TestFinalizeConnectorInboundRemoteIdentityAuthorsAndFinalizesFromInstalledMetadata(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {
			constants.FieldID: "003-inbound",
			"RemoteKey__c":    "remote-123",
			"Email":           "lead@example.com",
		},
	})

	next, err := exec.executeFinalizeConnectorInboundRemoteIdentity(
		connectorInboundFinalizeContext("003-inbound"),
		nil,
		connectorInboundFinalizeAction(),
		nil,
	)
	if err != nil {
		t.Fatalf("execute finalize inbound identity: %v", err)
	}
	if next != "Done" {
		t.Fatalf("next = %q, want Done", next)
	}
	if len(resolver.queries) != 1 || resolver.queries[0] != (ConnectorPushContractQuery{
		ConnectorOwnerKey: "connector_a",
		FlowDeveloperName: "Connector_Push",
		FlowActionName:    "Push",
	}) {
		t.Fatalf("contract queries = %#v, want exact installed target", resolver.queries)
	}
	if len(store.authorityQueries) != 1 {
		t.Fatalf("authority query count = %d, want 1 retry/fan-in check", len(store.authorityQueries))
	}
	if len(store.inserted) != 1 || len(store.finalizations) != 1 || len(store.retired) != 0 {
		t.Fatalf("durable calls inserted=%d finalized=%d retired=%d, want 1/1/0", len(store.inserted), len(store.finalizations), len(store.retired))
	}
	got := store.finalizations[0]
	if got.Evidence.ID == "" || got.Evidence.ID != store.inserted[0].ID {
		t.Fatalf("finalized evidence ID = %q, inserted ID = %q; want one exact caller-named row", got.Evidence.ID, store.inserted[0].ID)
	}
	if got.Evidence.ConnectorOwnerKey != "connector_a" || got.Evidence.ConnectorNamespace != "hubspot-routing-namespace" || got.Evidence.SourceObject != "Contact" || got.Evidence.SourceRecordId != "003-inbound" || got.Evidence.RemoteIDField != "RemoteKey__c" || got.Evidence.RemoteId != "remote-123" {
		t.Fatalf("evidence = %#v, want authenticated inbound tuple", got.Evidence)
	}
	if got.Adoption.Identity != (ConnectorPushDurableIdentity{
		ConnectorOwnerKey: "connector_a",
		ActionSlotID:      "contact-remote-sync",
		ActionType:        constants.ActionTypePushToConnector,
		SourceObject:      "Contact",
		SourceRecordID:    "003-inbound",
		DestinationSlotID: "remote-contact",
	}) {
		t.Fatalf("identity = %#v, want installed contract identity", got.Adoption.Identity)
	}
	if got.Adoption.Snapshot.CreatedPackInstallID != "install-active" || got.Adoption.Snapshot.RemoteIDField != "RemoteKey__c" || got.Adoption.Snapshot.MappingFingerprint == "" || got.Adoption.Snapshot.ProfileContractHash == "" {
		t.Fatalf("snapshot = %#v, want installed profile/mapping lineage", got.Adoption.Snapshot)
	}
}

func TestFinalizeConnectorInboundRemoteIdentityRejectsRetiredLocatorHistory(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	store.historyErr = ErrConnectorPushDurableConflict
	exec := newConnectorInboundFinalizeExecutor(store, &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
	})
	_, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil)
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("retired history error = %v", err)
	}
	if len(store.inserted) != 0 || len(store.finalizations) != 0 {
		t.Fatalf("retired history mutated inserted=%d finalized=%d", len(store.inserted), len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityReusesExactActiveEvidence(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	store.seedActive("connector_a", "hubspot-routing-namespace", "Contact", "003-inbound", "RemoteKey__c", "remote-123")
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
	})

	if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil); err != nil {
		t.Fatalf("reuse exact evidence: %v", err)
	}
	if len(store.inserted) != 0 || len(store.authorityQueries) != 0 || len(store.finalizations) != 1 {
		t.Fatalf("calls inserted=%d authority=%d finalized=%d, want 0/0/1", len(store.inserted), len(store.authorityQueries), len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityMismatchedEvidenceFailsWithoutRetire(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	store.seedActive("connector_a", "hubspot-routing-namespace", "Contact", "003-inbound", "RemoteKey__c", "remote-old")
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-new"},
	})

	_, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil)
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("mismatched evidence error = %v, want durable conflict", err)
	}
	if len(store.retired) != 0 || len(store.inserted) != 0 || len(store.finalizations) != 0 {
		t.Fatalf("mismatch mutated durable state: retired=%d inserted=%d finalized=%d", len(store.retired), len(store.inserted), len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityRejectsActiveEvidenceWithoutExactID(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	evidence := store.seedActive("connector_a", "hubspot-routing-namespace", "Contact", "003-inbound", "RemoteKey__c", "remote-123")
	delete(store.evidence, evidence.ID)
	evidence.ID = ""
	store.evidence["corrupt-active-row"] = evidence
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
	})

	_, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil)
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("blank exact evidence ID error = %v, want durable conflict", err)
	}
	if len(store.finalizations) != 0 {
		t.Fatalf("blank exact evidence ID reached finalizer %d times", len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityIdenticalAuthorityRetryIsNoOp(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	contract := defaultPushConnectorContract()
	identity := connectorPushDurableIdentityFromContract(contract, "003-inbound")
	linkKey, err := identity.LinkKey()
	if err != nil {
		t.Fatalf("LinkKey: %v", err)
	}
	store.authority = &ConnectorRemoteIdentityAuthority{
		LinkID:     "link-existing",
		LinkKey:    linkKey,
		LinkStatus: ConnectorPushLinkStatusFinalized,
		RemoteID:   "remote-123",
		Identity:   identity,
	}
	resolver := &captureInboundFinalizeContractResolver{contract: contract}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
	})

	next, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil)
	if err != nil || next != "Done" {
		t.Fatalf("retry result next=%q err=%v, want Done/nil", next, err)
	}
	if len(store.inserted) != 0 || len(store.finalizations) != 0 {
		t.Fatalf("retry created new provenance: inserted=%d finalized=%d", len(store.inserted), len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityForeignAuthorityFailsBeforeEvidence(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	contract := defaultPushConnectorContract()
	identity := connectorPushDurableIdentityFromContract(contract, "003-other")
	store.authority = &ConnectorRemoteIdentityAuthority{
		LinkID:     "link-foreign",
		LinkKey:    mustConnectorPushTestLinkKey(identity),
		LinkStatus: ConnectorPushLinkStatusFinalized,
		RemoteID:   "remote-123",
		Identity:   identity,
	}
	resolver := &captureInboundFinalizeContractResolver{contract: contract}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
	})

	_, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil)
	if !errors.Is(err, ErrConnectorPushDurableConflict) {
		t.Fatalf("foreign authority error = %v, want durable conflict", err)
	}
	if len(store.inserted) != 0 || len(store.finalizations) != 0 {
		t.Fatalf("foreign authority mutated evidence/link: inserted=%d finalized=%d", len(store.inserted), len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityContractDriftFailsBeforeEvidence(t *testing.T) {
	for _, tc := range []struct {
		name   string
		mutate func(*connector.ConnectorPushDurableMetadataContract)
	}{
		{name: "foreign owner", mutate: func(contract *connector.ConnectorPushDurableMetadataContract) {
			contract.ConnectorOwnerKey = "foreign-pack"
		}},
		{name: "wrong target flow", mutate: func(contract *connector.ConnectorPushDurableMetadataContract) {
			contract.FlowDeveloperName = "Other_Flow"
		}},
		{name: "wrong action type", mutate: func(contract *connector.ConnectorPushDurableMetadataContract) { contract.ActionType = "OtherAction" }},
		{name: "wrong source object", mutate: func(contract *connector.ConnectorPushDurableMetadataContract) { contract.SourceObject = "Account" }},
		{name: "blank remote field", mutate: func(contract *connector.ConnectorPushDurableMetadataContract) { contract.RemoteIDField = "" }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newInboundFinalizeDurableStore()
			contract := defaultPushConnectorContract()
			tc.mutate(contract)
			resolver := &captureInboundFinalizeContractResolver{contract: contract}
			exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
				"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
			})
			if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil); err == nil {
				t.Fatalf("%s must fail closed", tc.name)
			}
			if len(store.inserted) != 0 || len(store.finalizations) != 0 {
				t.Fatalf("%s mutated durable state: inserted=%d finalized=%d", tc.name, len(store.inserted), len(store.finalizations))
			}
		})
	}
}

func TestFinalizeConnectorInboundRemoteIdentityRequiresAuthenticatedAtomicRoot(t *testing.T) {
	const flowName = "HubSpot_Contact_Lead_Upsert"
	identity := FlowExecutionIdentity{
		APIName:       flowName,
		DefinitionID:  "flow-def-hubspot-inbound",
		VersionID:     "flow-version-hubspot-inbound",
		OwnerPack:     "connector_a",
		VersionSource: flowSourceStandard,
	}
	for _, tc := range []struct {
		name      string
		contextFn func() context.Context
		mutate    func(*FlowExecutor)
	}{
		{
			name: "missing origin",
			contextFn: func() context.Context {
				return connectorInboundCtxWithIdentity("", identity)
			},
		},
		{
			name: "missing root marker",
			contextFn: func() context.Context {
				return connectorInboundCtxWithIdentity("hubspot-routing-namespace", identity)
			},
		},
		{
			name: "customized flow",
			contextFn: func() context.Context {
				custom := identity
				custom.DefinitionIsCustomized = true
				ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", custom)
				return constants.WithConnectorInboundFlowExecution(ctx, flowName, "hubspot-routing-namespace")
			},
		},
		{
			name:      "no live transaction",
			contextFn: func() context.Context { return connectorInboundFinalizeContext("003-inbound") },
			mutate: func(exec *FlowExecutor) {
				exec.persistence.(*inboundRegisterPersistenceStub).txActive = false
			},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newInboundFinalizeDurableStore()
			resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
			exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
				"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
			})
			if tc.mutate != nil {
				tc.mutate(exec)
			}
			if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(tc.contextFn(), nil, connectorInboundFinalizeAction(), nil); err == nil {
				t.Fatalf("%s must fail closed", tc.name)
			}
			if len(store.inserted) != 0 || len(store.finalizations) != 0 {
				t.Fatalf("%s mutated durable state", tc.name)
			}
		})
	}
}

func TestFinalizeConnectorInboundRemoteIdentityAcceptsExactAuthorizedCompanionVersion(t *testing.T) {
	const flowVersionID = "flow-version-hubspot-companion"
	identity := FlowExecutionIdentity{
		APIName:       "HubSpot_Contact_Lead_Upsert",
		DefinitionID:  "flow-def-hubspot-companion",
		VersionID:     flowVersionID,
		OwnerPack:     "connector_a",
		VersionSource: flowSourceStandard,
	}
	for _, tc := range []struct {
		name                    string
		authorizedFlowVersionID string
		wantErr                 bool
	}{
		{name: "exact companion version", authorizedFlowVersionID: flowVersionID},
		{name: "different companion version", authorizedFlowVersionID: "flow-version-other", wantErr: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newInboundFinalizeDurableStore()
			resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
			exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
				"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123", "Email": "lead@example.com"},
			})
			ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", identity)
			ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, tc.authorizedFlowVersionID)
			ctx = withEngineSnapshot(ctx, "Contact", models.SObject{constants.FieldID: "003-inbound"})

			_, err := exec.executeFinalizeConnectorInboundRemoteIdentity(ctx, nil, connectorInboundFinalizeAction(), nil)
			if tc.wantErr {
				if err == nil {
					t.Fatal("execute finalize inbound identity error = nil, want companion version authorization failure")
				}
				if len(store.inserted) != 0 || len(store.finalizations) != 0 {
					t.Fatal("unauthorized companion version mutated durable state")
				}
				return
			}
			if err != nil {
				t.Fatalf("execute finalize inbound identity: %v", err)
			}
			if len(store.inserted) != 1 || len(store.finalizations) != 1 {
				t.Fatalf("durable calls inserted=%d finalized=%d, want 1/1", len(store.inserted), len(store.finalizations))
			}
		})
	}
}

func TestFinalizeConnectorInboundRemoteIdentityAcceptsExactAuthorizedCompanionSubflow(t *testing.T) {
	const (
		ownerPack       = "connector_a"
		parentVersionID = "flow-version-hubspot-companion"
		childFlowName   = "HubSpot_Workflow_Action_Lead_Create_Finalize"
	)
	parentIdentity := FlowExecutionIdentity{
		APIName:       "HubSpot_Workflow_Action_CampaignMember_Upsert",
		DefinitionID:  "flow-def-hubspot-companion",
		VersionID:     parentVersionID,
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	}
	childIdentity := FlowExecutionIdentity{
		APIName:       childFlowName,
		DefinitionID:  "flow-def-hubspot-create-finalize",
		VersionID:     "flow-version-hubspot-create-finalize",
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	}

	store := newInboundFinalizeDurableStore()
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123", "Email": "lead@example.com"},
	})
	ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", parentIdentity)
	ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
	ctx = withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)
	ctx = withFlowRuntimeContext(ctx, flowRuntimeContextFromCtx(ctx).WithCurrentFlow(childIdentity))
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{constants.FieldID: "003-inbound"})

	if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(ctx, nil, connectorInboundFinalizeAction(), nil); err != nil {
		t.Fatalf("execute finalize inbound identity from exact companion subflow: %v", err)
	}
	if len(store.inserted) != 1 || len(store.finalizations) != 1 {
		t.Fatalf("durable calls inserted=%d finalized=%d, want 1/1", len(store.inserted), len(store.finalizations))
	}
}

func TestExecuteSubflowCarriesExactCompanionDelegationIntoAtomicFinalizer(t *testing.T) {
	const (
		ownerPack       = "connector_a"
		parentVersionID = "flow-version-hubspot-companion"
		childFlowName   = "HubSpot_Workflow_Action_Lead_Create_Finalize"
	)
	persistence := &companionSubflowFinalizePersistence{
		inboundRegisterPersistenceStub: &inboundRegisterPersistenceStub{lockRecords: map[string]models.SObject{}},
	}
	store := newInboundFinalizeDurableStore()
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := NewFlowExecutor(nil, nil, NewFlowInterviewService(persistence, nil), persistence, nil, nil, nil)
	exec.SetConnectorPushDurableStore(store)
	exec.SetConnectorPushContractResolver(resolver)
	exec.connectorPushProfiles = staticPushProfileResolver{profile: &ConnectorPushProfile{
		OwnerPack:            ownerPack,
		ProfileName:          "default",
		CreatedPackInstallID: "install-active",
		CredentialRef:        "nc-main",
		EndpointClass:        "crm.objects.write",
		RetryPolicy:          ConnectorPushRetryPolicy{MaxAttempts: 1},
	}}
	exec.connectorPackMappingResolverFn = func(_ context.Context, query connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return []connector.ConnectorFieldMappingInput{{
			Source:       query.Source,
			SourceObject: query.SourceObject,
			TargetObject: query.TargetObject,
			ActionType:   query.ActionType,
			Direction:    connector.DirectionOutbound,
			SourceField:  "Email",
			TargetField:  "email",
		}}, nil
	}
	exec.RegisterFlowActionCalls(testRuntimeFlowActionCallRegistrations()...)
	finalizeAction := *connectorInboundFinalizeAction()
	finalizeAction.Connector = nil
	exec.flowVersionFn = func(_ context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
		if apiName != childFlowName {
			return nil, nil
		}
		return &models.FlowVersion{
			Id:                  "flow-version-hubspot-create-finalize",
			DefinitionId:        "flow-def-hubspot-create-finalize",
			DefinitionOwnerPack: ownerPack,
			Source:              flowSourceStandard,
			Status:              models.FlowStatusActive,
			ProcessType:         "AutoLaunchedFlow",
			Metadata: models.FlowMetadata{
				RunAtomically: true,
				Start:         &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CreateContact"}},
				Variables:     []models.FlowVariable{{Name: "createdRecordId", DataType: "String", IsOutput: true}},
				RecordCreates: []models.FlowRecordCreate{{
					Name:   "CreateContact",
					Object: "Contact",
					InputAssignments: []models.FlowInputAssignment{
						{Field: "RemoteKey__c", Value: map[string]interface{}{"stringValue": "remote-123"}},
						{Field: "Email", Value: map[string]interface{}{"stringValue": "lead@example.com"}},
					},
					AssignRecordIdToReference: "createdRecordId",
					Connector:                 &models.FlowConnector{TargetReference: "FinalizeRemoteIdentity"},
				}},
				ActionCalls: []models.FlowActionCall{finalizeAction},
			},
		}, nil
	}

	parentIdentity := FlowExecutionIdentity{
		APIName:       "HubSpot_Workflow_Action_CampaignMember_Upsert",
		DefinitionID:  "flow-def-hubspot-companion",
		VersionID:     parentVersionID,
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	}
	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "hubspot-routing-namespace")
	ctx = withFlowRuntimeContext(ctx, flowRuntimeContextFromCtx(ctx).WithCurrentFlow(parentIdentity))
	ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
	ctx = WithTransientFlowInterviews(ctx)
	parentInterview := &models.FlowInterview{Id: "parent-interview", FlowVersionId: parentVersionID, ContextData: map[string]interface{}{}}
	subflow := &models.FlowSubflow{
		Name:     "CreateAndFinalizeWorkflowActionLead",
		FlowName: childFlowName,
		OutputAssignments: []models.FlowSubflowOutput{{
			Name:              "createdRecordId",
			AssignToReference: "matchedRecordId",
		}},
	}

	if _, err := exec.executeSubflow(ctx, parentInterview, subflow, nil); err != nil {
		t.Fatalf("execute exact companion child subflow: %v", err)
	}
	if persistence.transactionRuns != 1 {
		t.Fatalf("atomic child transaction runs = %d, want 1", persistence.transactionRuns)
	}
	if got := parentInterview.ContextData["matchedRecordId"]; got != "003-created-by-child" {
		t.Fatalf("parent child output = %#v, want created record id", got)
	}
	if len(store.inserted) != 1 || len(store.finalizations) != 1 {
		t.Fatalf("durable calls inserted=%d finalized=%d, want 1/1", len(store.inserted), len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityRejectsInvalidCompanionSubflowDelegation(t *testing.T) {
	const (
		ownerPack       = "connector_a"
		parentVersionID = "flow-version-hubspot-companion"
		childFlowName   = "HubSpot_Workflow_Action_Lead_Create_Finalize"
	)
	parentIdentity := FlowExecutionIdentity{
		APIName:       "HubSpot_Workflow_Action_CampaignMember_Upsert",
		DefinitionID:  "flow-def-hubspot-companion",
		VersionID:     parentVersionID,
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	}
	childIdentity := FlowExecutionIdentity{
		APIName:       childFlowName,
		DefinitionID:  "flow-def-hubspot-create-finalize",
		VersionID:     "flow-version-hubspot-create-finalize",
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	}

	for _, tc := range []struct {
		name    string
		context func() context.Context
		child   FlowExecutionIdentity
	}{
		{
			name: "different child name",
			context: func() context.Context {
				ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", parentIdentity)
				ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
				return withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)
			},
			child: func() FlowExecutionIdentity { v := childIdentity; v.APIName = "Different_Child"; return v }(),
		},
		{
			name: "different owner pack",
			context: func() context.Context {
				ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", parentIdentity)
				ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
				return withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)
			},
			child: func() FlowExecutionIdentity { v := childIdentity; v.OwnerPack = "connector_b"; return v }(),
		},
		{
			name: "customized child",
			context: func() context.Context {
				ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", parentIdentity)
				ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
				return withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)
			},
			child: func() FlowExecutionIdentity { v := childIdentity; v.DefinitionIsCustomized = true; return v }(),
		},
		{
			name: "unauthorized parent version",
			context: func() context.Context {
				ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", parentIdentity)
				ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, "different-parent-version")
				return withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)
			},
			child: childIdentity,
		},
		{
			name: "transitive grandchild",
			context: func() context.Context {
				ctx := connectorInboundCtxWithIdentity("hubspot-routing-namespace", parentIdentity)
				ctx = constants.WithConnectorInboundCompanionFlowExecution(ctx, parentVersionID)
				ctx = withConnectorInboundCompanionSubflowDelegation(ctx, childFlowName)
				ctx = withFlowRuntimeContext(ctx, flowRuntimeContextFromCtx(ctx).WithCurrentFlow(childIdentity))
				return withConnectorInboundCompanionSubflowDelegation(ctx, "Grandchild_Finalize")
			},
			child: func() FlowExecutionIdentity {
				v := childIdentity
				v.APIName = "Grandchild_Finalize"
				v.VersionID = "grandchild-version"
				return v
			}(),
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newInboundFinalizeDurableStore()
			resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
			exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
				"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
			})
			ctx := tc.context()
			ctx = withFlowRuntimeContext(ctx, flowRuntimeContextFromCtx(ctx).WithCurrentFlow(tc.child))
			ctx = withEngineSnapshot(ctx, "Contact", models.SObject{constants.FieldID: "003-inbound"})

			if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(ctx, nil, connectorInboundFinalizeAction(), nil); err == nil {
				t.Fatal("execute finalize inbound identity error = nil, want delegation rejection")
			}
			if len(store.inserted) != 0 || len(store.finalizations) != 0 {
				t.Fatalf("rejected delegation mutated durable state: inserted=%d finalized=%d", len(store.inserted), len(store.finalizations))
			}
		})
	}
}

func TestFinalizeConnectorInboundRemoteIdentityAmbiguousPostUpsertFailsClosed(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-one": {constants.FieldID: "003-one", "RemoteKey__c": "remote-1"},
		"Contact/003-two": {constants.FieldID: "003-two", "RemoteKey__c": "remote-2"},
	})
	ctx := connectorInboundFinalizeContext("003-one")
	ctx = withEngineSnapshot(ctx, "Contact", models.SObject{constants.FieldID: "003-two"})
	if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(ctx, nil, connectorInboundFinalizeAction(), nil); err == nil {
		t.Fatal("two engine-confirmed records must fail closed")
	}
	if len(store.inserted) != 0 || len(store.finalizations) != 0 {
		t.Fatal("ambiguous record resolution mutated durable state")
	}
}

func TestFinalizeConnectorInboundRemoteIdentityMetadataResolutionFailsBeforeEvidence(t *testing.T) {
	for _, tc := range []struct {
		name   string
		mutate func(*FlowExecutor)
	}{
		{
			name: "mapping error",
			mutate: func(exec *FlowExecutor) {
				exec.connectorPackMappingResolverFn = func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
					return nil, errors.New("mapping unavailable")
				}
			},
		},
		{
			name: "profile drift",
			mutate: func(exec *FlowExecutor) {
				exec.connectorPushProfiles = staticPushProfileResolver{profile: &ConnectorPushProfile{OwnerPack: "other-pack", ProfileName: "default", CreatedPackInstallID: "install-active", CredentialRef: "nc", EndpointClass: "scope", RetryPolicy: ConnectorPushRetryPolicy{MaxAttempts: 1}}}
			},
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			store := newInboundFinalizeDurableStore()
			resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
			exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
				"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
			})
			tc.mutate(exec)
			if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil); err == nil {
				t.Fatalf("%s must fail closed", tc.name)
			}
			if len(store.inserted) != 0 || len(store.finalizations) != 0 {
				t.Fatalf("%s mutated durable state", tc.name)
			}
		})
	}
}

func TestFinalizeConnectorInboundRemoteIdentityPropagatesAmbientFinalizerFailure(t *testing.T) {
	store := newInboundFinalizeDurableStore()
	store.finalizeErr = errors.New("ambient adoption failed")
	resolver := &captureInboundFinalizeContractResolver{contract: defaultPushConnectorContract()}
	exec := newConnectorInboundFinalizeExecutor(store, resolver, map[string]models.SObject{
		"Contact/003-inbound": {constants.FieldID: "003-inbound", "RemoteKey__c": "remote-123"},
	})
	if _, err := exec.executeFinalizeConnectorInboundRemoteIdentity(connectorInboundFinalizeContext("003-inbound"), nil, connectorInboundFinalizeAction(), nil); err == nil {
		t.Fatal("ambient finalizer failure must fault the atomic Flow")
	}
	if len(store.finalizations) != 1 {
		t.Fatalf("finalizer calls = %d, want 1", len(store.finalizations))
	}
}

func TestFinalizeConnectorInboundRemoteIdentityRegistrationIsAtomicSafe(t *testing.T) {
	for _, registration := range ConnectorPlatformFlowActionCallRegistrations() {
		if registration.ActionType != flowactions.ActionTypeFinalizeConnectorInboundRemoteIdentity {
			continue
		}
		if registration.AtomicPolicy != FlowActionCallAtomicPolicySafe || registration.HandlerRef != "platform:connector/finalize-inbound-remote-identity" {
			t.Fatalf("registration = %#v, want atomic-safe platform handler", registration)
		}
		return
	}
	t.Fatal("finalize inbound remote identity registration not found")
}