package automation

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

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

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

// stubResolver returns a fixed set of ConnectorFieldMappingInput rows without
// touching the database. Passed as connectorMappingResolverFn on the executor.
func stubResolver(rows []connector.ConnectorFieldMappingInput) func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
	return func(_ context.Context, _ connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return rows, nil
	}
}

func captureResolveQueryResolver(rows []connector.ConnectorFieldMappingInput, captured *connector.ResolveQuery) func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
	return func(_ context.Context, q connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		if captured != nil {
			*captured = q
		}
		return rows, nil
	}
}

func connectorMappingContext(ownerPack string) context.Context {
	return connectorMappingContextWithIdentity(FlowExecutionIdentity{
		APIName:       "Connector_Inbound",
		DefinitionID:  "flow-def-connector",
		VersionID:     "flow-version-connector",
		OwnerPack:     ownerPack,
		VersionSource: flowSourceStandard,
	})
}

func connectorMappingContextWithIdentity(identity FlowExecutionIdentity) context.Context {
	return withFlowRuntimeContext(
		context.Background(),
		flowRuntimeContextFromCtx(context.Background()).WithCurrentFlow(identity),
	)
}

func connectorPolicyPersistence(cap *captureUpdate, objectName, id string, record models.SObject) *stubPersistenceCapture {
	return &stubPersistenceCapture{
		capture: cap,
		rawRecords: map[string]models.SObject{
			objectName + "/" + id: record,
		},
	}
}

func connectorTargetSchema(fields ...models.FieldMetadata) *models.ObjectMetadata {
	return &models.ObjectMetadata{
		APIName: "Contact",
		Fields:  fields,
	}
}

func TestConnectorPushFingerprintsOmitEmptyValueMap(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			Source:       "connector_a",
			SourceObject: "Contact",
			TargetObject: "remoteContact",
			ActionType:   "PushToConnector",
			SourceField:  "LifecycleStage__c",
			TargetField:  "lifecyclestage",
			Direction:    connector.DirectionOutbound,
			SortOrder:    1,
		},
	}

	type legacyMapping struct {
		Source              string `json:"Source"`
		SourceObject        string `json:"SourceObject"`
		TargetObject        string `json:"TargetObject"`
		ActionType          string `json:"ActionType"`
		SourceField         string `json:"SourceField"`
		TargetField         string `json:"TargetField"`
		Direction           string `json:"Direction"`
		TransformExpression string `json:"TransformExpression"`
		ConflictPolicy      string `json:"ConflictPolicy"`
		DefaultValue        string `json:"DefaultValue"`
		SortOrder           int    `json:"SortOrder"`
	}
	legacyMappings := []legacyMapping{{
		Source:       mappings[0].Source,
		SourceObject: mappings[0].SourceObject,
		TargetObject: mappings[0].TargetObject,
		ActionType:   mappings[0].ActionType,
		SourceField:  mappings[0].SourceField,
		TargetField:  mappings[0].TargetField,
		Direction:    mappings[0].Direction,
		SortOrder:    mappings[0].SortOrder,
	}}
	legacyMappingJSON, err := json.Marshal(legacyMappings)
	if err != nil {
		t.Fatalf("marshal legacy mappings: %v", err)
	}
	if got, want := ConnectorPushFieldMappingFingerprint(mappings), connectorPushSHA256Hex(legacyMappingJSON); got != want {
		t.Fatalf("empty ValueMap mapping fingerprint = %s, want legacy %s", got, want)
	}

	withValueMap := append([]connector.ConnectorFieldMappingInput(nil), mappings...)
	withValueMap[0].ValueMap = `{"lead":"subscriber"}`
	if ConnectorPushFieldMappingFingerprint(withValueMap) == ConnectorPushFieldMappingFingerprint(mappings) {
		t.Fatal("non-empty ValueMap must participate in mapping fingerprint")
	}
}

// buildConnectorMappingAction builds a minimal FlowActionCall for
// "applyConnectorFieldMappings" with the given literal string params.
func buildConnectorMappingAction(params map[string]interface{}) *models.FlowActionCall {
	var inputParams []models.FlowActionParameter
	for name, val := range params {
		var wrapped interface{}
		switch v := val.(type) {
		case string:
			wrapped = map[string]interface{}{"stringValue": v}
		case map[string]interface{}:
			wrapped = v // pass through (e.g. elementReference)
		default:
			wrapped = val
		}
		inputParams = append(inputParams, models.FlowActionParameter{
			Name:  name,
			Value: wrapped,
		})
	}
	return &models.FlowActionCall{
		Name:            "ApplyMappings",
		ActionType:      "applyConnectorFieldMappings",
		InputParameters: inputParams,
	}
}

// buildInterviewWithRecord creates a FlowInterview with the given record stored
// at key "$SourceRecord" in ContextData.
func buildInterviewWithSourceRecord(sourceRecord models.SObject, targetRecord models.SObject) *models.FlowInterview {
	return &models.FlowInterview{
		Id: "test-interview",
		ContextData: map[string]interface{}{
			"$SourceRecord": sourceRecord,
			"$TargetRecord": targetRecord,
		},
	}
}

// ---------------------------------------------------------------------------
// Missing resolver → fault
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_NoResolverFaults(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: cap},
		// neither connectorMappingResolverFn nor connectorMappingResolver set
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "a@example.com"},
		models.SObject{},
	)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected error when no resolver is wired, got nil")
	}
}

// ---------------------------------------------------------------------------
// Required parameter validation
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_MissingTrustedOwnerFaults(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(nil),
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		// "source" intentionally omitted
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(context.Background(), interview, action, nil)
	if err == nil {
		t.Fatal("expected error without trusted Flow owner pack, got nil")
	}
}

func TestExecuteApplyConnectorFieldMappings_UsesTrustedOwnerWhenSourceInputOmitted(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}

	var captured connector.ResolveQuery
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: captureResolveQueryResolver(mappings, &captured),
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		// "source" intentionally omitted. Runtime authority comes from FlowDefinition.OwnerPack.
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "a@example.com"},
		models.SObject{},
	)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if captured.Source != "connector_a" {
		t.Fatalf("ResolveQuery.Source = %q, want trusted Flow owner pack connector_a", captured.Source)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_PopulatesPostUpsertRegistry(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	var captured connector.ResolveQuery
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		// txActive=true: the gate (publishPostUpsertRecord) requires IsTransactionActive
		// to be true for an atomic-flow publish — mirrors real inbound-worker execution
		// where the write happens inside the ambient atomic-flow tx.
		persistence:                &stubPersistenceCapture{capture: cap, txActive: true},
		connectorMappingResolverFn: captureResolveQueryResolver(mappings, &captured),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "a@example.com"},
		models.SObject{},
	)
	// withAtomicFlowBoundary mirrors what runPossiblyAtomic stamps; the gate
	// requires both isWithinAtomicFlowBoundary=true AND IsTransactionActive=true.
	ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
	if _, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	// The connector-field-mapping write is the 4th inbound source-record upsert
	// path; it must publish to the post-upsert registry so the register action
	// (issue #2080) resolves the engine-owned identity instead of blank-no-op
	// stranding the record without authorship evidence.
	recs := flowRuntimeContextFromCtx(ctx).PostUpsertRecords("Contact")
	if len(recs) != 1 {
		t.Fatalf("expected 1 post-upsert snapshot for Contact after field-mapping update, got %d", len(recs))
	}
	if !flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
		t.Fatal("expected field-mapping update to publish the engine-owned Contact write outcome")
	}
	if id, _ := recs[0]["Id"].(string); id != "rec-001" {
		t.Fatalf("post-upsert snapshot Id = %q, want rec-001", id)
	}
}

func TestExecuteApplyConnectorFieldMappings_SourceInputMismatchFaultsBeforeResolver(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
			t.Fatal("resolver must not be called when source assertion mismatches trusted Flow owner pack")
			return nil, nil
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_b",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected source assertion mismatch to fault, got nil")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls after source mismatch, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_CustomizedFlowFaultsBeforeResolver(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
			t.Fatal("resolver must not be called for customized Flow connector mapping authority")
			return nil, nil
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
	})
	ctx := connectorMappingContextWithIdentity(FlowExecutionIdentity{
		APIName:                "Tenant_Customized_Connector_Flow",
		DefinitionID:           "tenant-def",
		VersionID:              "tenant-version",
		OwnerPack:              "connector_a",
		DefinitionIsCustomized: true,
		VersionSource:          flowSourceStandard,
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
	if err == nil {
		t.Fatal("expected customized Flow to fault before resolving connector mappings, got nil")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls for customized Flow, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_NonStandardFlowFaultsBeforeResolver(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
			t.Fatal("resolver must not be called for non-standard FlowVersion connector mapping authority")
			return nil, nil
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
	})
	ctx := connectorMappingContextWithIdentity(FlowExecutionIdentity{
		APIName:       "Tenant_Nonstandard_Connector_Flow",
		DefinitionID:  "tenant-def",
		VersionID:     "tenant-version",
		OwnerPack:     "connector_a",
		VersionSource: "custom",
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
	if err == nil {
		t.Fatal("expected non-standard FlowVersion to fault before resolving connector mappings, got nil")
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls for non-standard FlowVersion, got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// Plain copy (no transform, default conflict policy)
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_PlainCopy(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
		{
			SourceField: "phone",
			TargetField: "Phone",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{
		"email": "user@example.com",
		"phone": "+15550001234",
	}
	// Build action with literal sourceRecord embedded as an elementReference
	// to "$SourceRecord" in the interview context.
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	call := cap.calls[0]
	if call.objectName != "Contact" {
		t.Errorf("Update objectName = %q, want Contact", call.objectName)
	}
	if call.id != "rec-001" {
		t.Errorf("Update id = %q, want rec-001", call.id)
	}
	if got, _ := call.updates["Email"].(string); got != "user@example.com" {
		t.Errorf("updates[Email] = %q, want user@example.com", got)
	}
	if got, _ := call.updates["Phone"].(string); got != "+15550001234" {
		t.Errorf("updates[Phone] = %q, want +15550001234", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_SkipBlankTargetFieldsOmitsOnlyNamedRawBlank(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
		{SourceField: "phone", TargetField: "Phone", Direction: connector.DirectionInbound},
	}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                "connector_a",
		"sourceObject":          "Contact",
		"targetObject":          "Contact",
		"actionType":            "syncContact",
		"direction":             "inbound",
		"targetId":              "rec-001",
		"sourceRecord":          map[string]interface{}{"elementReference": "$SourceRecord"},
		"skipBlankTargetFields": []interface{}{"Email"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "", "phone": ""}, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1 for the non-excluded blank target", len(cap.calls))
	}
	if _, found := cap.calls[0].updates["Email"]; found {
		t.Fatalf("updates must omit excluded raw blank Email: %#v", cap.calls[0].updates)
	}
	if got := cap.calls[0].updates["Phone"]; got != "" {
		t.Fatalf("updates[Phone] = %#v, want legacy raw blank clear", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_MergesExtraUpdatesIntoSingleUpdate(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"extraUpdates": map[string]interface{}{
			"SyncEnabled__c": map[string]interface{}{"booleanValue": true},
		},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "user@example.com"}, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if got, _ := updates["Email"].(string); got != "user@example.com" {
		t.Fatalf("updates[Email] = %q, want user@example.com", got)
	}
	if got, _ := updates["SyncEnabled__c"].(bool); !got {
		t.Fatalf("updates[SyncEnabled__c] = %#v, want true", updates["SyncEnabled__c"])
	}
}

func TestExecuteApplyConnectorFieldMappings_SkipsExtraUpdatesWhenGuardFalse(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":            "connector_a",
		"sourceObject":      "Contact",
		"targetObject":      "Contact",
		"actionType":        "syncContact",
		"direction":         "inbound",
		"targetId":          "rec-001",
		"sourceRecord":      map[string]interface{}{"elementReference": "$SourceRecord"},
		"applyExtraUpdates": map[string]interface{}{"booleanValue": false},
		"extraUpdates": map[string]interface{}{
			"SyncEnabled__c": map[string]interface{}{"booleanValue": true},
		},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "user@example.com"}, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if _, exists := updates["SyncEnabled__c"]; exists {
		t.Fatalf("updates[SyncEnabled__c] exists after applyExtraUpdates=false: %#v", updates["SyncEnabled__c"])
	}
	if got, _ := updates["Email"].(string); got != "user@example.com" {
		t.Fatalf("updates[Email] = %q, want user@example.com", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_ExtraUpdatesConflictWithMappingFieldFaults(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"extraUpdates": map[string]interface{}{
			"Email": map[string]interface{}{"stringValue": "override@example.com"},
		},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "user@example.com"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected conflict error, got nil")
	}
	if !strings.Contains(err.Error(), "conflicts with a connector mapping target field") {
		t.Fatalf("error = %v, want field-conflict diagnostic", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("Update calls = %d, want 0 after conflict", len(cap.calls))
	}
}

func TestRecordCreateConnectorFieldMappingMergesIntoInsert(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{Source: "connector_a", SourceObject: "contacts", TargetObject: "Lead", ActionType: "lead-upsert", Direction: "inbound", SourceField: "email", TargetField: "Email"},
		{Source: "connector_a", SourceObject: "contacts", TargetObject: "Lead", ActionType: "lead-upsert", Direction: "inbound", SourceField: "company", TargetField: "Company"},
		{Source: "connector_a", SourceObject: "contacts", TargetObject: "Lead", ActionType: "lead-upsert", Direction: "inbound", SourceField: "lastname", TargetField: "LastName"},
	}
	var captured connector.ResolveQuery
	persistence := &stubPersistenceCapture{}
	exec := &FlowExecutor{
		persistence: persistence,
		connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
			t.Fatal("RecordCreate connectorFieldMapping must not read effective/admin mappings")
			return nil, nil
		},
		connectorPackMappingResolverFn: captureResolveQueryResolver(mappings, &captured),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return &models.ObjectMetadata{APIName: "Lead", Fields: []models.FieldMetadata{
				{APIName: "Email", Type: constants.FieldTypeEmail, Createable: true, Updateable: true},
				{APIName: "Company", Type: constants.FieldTypeString, Createable: true, Updateable: true},
				{APIName: "LastName", Type: constants.FieldTypeString, Createable: true, Updateable: true},
				{APIName: "HubSpot_Sync_Enabled__c", Type: constants.FieldTypeBoolean, Createable: true, Updateable: false},
			}}
		},
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"leadMappingActionType": "lead-upsert",
		"mappingSourceRecord": models.SObject{
			"email":    "ada@example.test",
			"company":  "Analytical Engines",
			"lastname": "Lovelace",
		},
	}}
	create := &models.FlowRecordCreate{
		Name:   "CreateLeadFromConnector",
		Object: "Lead",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "LastName", Value: map[string]interface{}{"stringValue": "HubSpot Lead"}},
			{Field: "Company", Value: map[string]interface{}{"stringValue": "HubSpot Lead"}},
			{Field: "HubSpot_Sync_Enabled__c", Value: false},
			{Field: "IsConverted", Value: false},
		},
		ConnectorFieldMapping: &models.FlowConnectorFieldMappingRef{
			Source:       map[string]interface{}{"stringValue": "connector_a"},
			SourceObject: map[string]interface{}{"stringValue": "contacts"},
			TargetObject: map[string]interface{}{"stringValue": "Lead"},
			ActionType:   map[string]interface{}{"elementReference": "leadMappingActionType"},
			Direction:    map[string]interface{}{"stringValue": "inbound"},
			SourceRecord: map[string]interface{}{"elementReference": "mappingSourceRecord"},
			ApplyExtraUpdates: map[string]interface{}{
				"booleanValue": true,
			},
			ExtraUpdates: map[string]interface{}{
				"HubSpot_Sync_Enabled__c": map[string]interface{}{"booleanValue": true},
			},
		},
	}

	if _, err := exec.executeRecordCreate(connectorMappingContext("connector_a"), interview, nil, create, nil); err != nil {
		t.Fatalf("executeRecordCreate: %v", err)
	}
	if len(persistence.insertCalls) != 1 {
		t.Fatalf("Insert calls = %d, want 1", len(persistence.insertCalls))
	}
	inserted := persistence.insertCalls[0]
	if got := inserted["Email"]; got != "ada@example.test" {
		t.Fatalf("insert Email = %v, want mapped source email", got)
	}
	if got := inserted["Company"]; got != "Analytical Engines" {
		t.Fatalf("insert Company = %v, want mapped source company", got)
	}
	if got := inserted["LastName"]; got != "Lovelace" {
		t.Fatalf("insert LastName = %v, want mapped source lastname", got)
	}
	if got := inserted["HubSpot_Sync_Enabled__c"]; got != true {
		t.Fatalf("insert HubSpot_Sync_Enabled__c = %v, want true from extraUpdates", got)
	}
	if got := inserted["IsConverted"]; got != false {
		t.Fatalf("insert IsConverted = %v, want original input assignment false", got)
	}
	if captured.Source != "connector_a" || captured.ActionType != "lead-upsert" || captured.Direction != "inbound" {
		t.Fatalf("ResolveQuery = %#v, want connector_a/lead-upsert/inbound", captured)
	}
}

func TestRecordCreateConnectorFieldMapping_SkipBlankTargetFieldsOmitsOnlyNamedRawBlank(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{Source: "connector_a", SourceObject: "contacts", TargetObject: "Lead", ActionType: "lead-upsert", Direction: "inbound", SourceField: "email", TargetField: "Email"},
		{Source: "connector_a", SourceObject: "contacts", TargetObject: "Lead", ActionType: "lead-upsert", Direction: "inbound", SourceField: "phone", TargetField: "Phone"},
	}
	persistence := &stubPersistenceCapture{}
	exec := &FlowExecutor{
		persistence:                    persistence,
		connectorPackMappingResolverFn: stubResolver(mappings),
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"mappingSourceRecord": models.SObject{"email": "", "phone": ""},
	}}
	create := &models.FlowRecordCreate{
		Name:   "CreateLeadFromConnector",
		Object: "Lead",
		ConnectorFieldMapping: &models.FlowConnectorFieldMappingRef{
			Source:                map[string]interface{}{"stringValue": "connector_a"},
			SourceObject:          map[string]interface{}{"stringValue": "contacts"},
			TargetObject:          map[string]interface{}{"stringValue": "Lead"},
			ActionType:            map[string]interface{}{"stringValue": "lead-upsert"},
			Direction:             map[string]interface{}{"stringValue": "inbound"},
			SourceRecord:          map[string]interface{}{"elementReference": "mappingSourceRecord"},
			SkipBlankTargetFields: []interface{}{"Email"},
		},
	}

	if _, err := exec.executeRecordCreate(connectorMappingContext("connector_a"), interview, nil, create, nil); err != nil {
		t.Fatalf("executeRecordCreate: %v", err)
	}
	if len(persistence.insertCalls) != 1 {
		t.Fatalf("Insert calls = %d, want 1", len(persistence.insertCalls))
	}
	if _, found := persistence.insertCalls[0]["Email"]; found {
		t.Fatalf("insert record must omit raw blank Email: %#v", persistence.insertCalls[0])
	}
	if got := persistence.insertCalls[0]["Phone"]; got != "" {
		t.Fatalf("insert Phone = %#v, want legacy raw blank clear", got)
	}
}

func TestRecordCreateConnectorFieldMappingCreateabilityGuard(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{Source: "connector_a", SourceObject: "contacts", TargetObject: "Lead", ActionType: "lead-upsert", Direction: "inbound", SourceField: "email", TargetField: "Email"},
	}
	persistence := &stubPersistenceCapture{}
	exec := &FlowExecutor{
		persistence:                    persistence,
		connectorPackMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return &models.ObjectMetadata{APIName: "Lead", Fields: []models.FieldMetadata{
				{APIName: "Email", Type: constants.FieldTypeEmail, Createable: true, Updateable: false},
			}}
		},
	}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{
		"mappingSourceRecord": models.SObject{"email": "ada@example.test"},
	}}
	create := &models.FlowRecordCreate{
		Name:   "CreateLeadFromConnector",
		Object: "Lead",
		ConnectorFieldMapping: &models.FlowConnectorFieldMappingRef{
			Source:       map[string]interface{}{"stringValue": "connector_a"},
			SourceObject: map[string]interface{}{"stringValue": "contacts"},
			TargetObject: map[string]interface{}{"stringValue": "Lead"},
			ActionType:   map[string]interface{}{"stringValue": "lead-upsert"},
			Direction:    map[string]interface{}{"stringValue": "inbound"},
			SourceRecord: map[string]interface{}{"elementReference": "mappingSourceRecord"},
		},
	}

	if _, err := exec.executeRecordCreate(connectorMappingContext("connector_a"), interview, nil, create, nil); err != nil {
		t.Fatalf("executeRecordCreate createable/updateable split: %v", err)
	}
	if len(persistence.insertCalls) != 1 {
		t.Fatalf("Insert calls = %d, want 1", len(persistence.insertCalls))
	}
	if got := persistence.insertCalls[0]["Email"]; got != "ada@example.test" {
		t.Fatalf("insert Email = %v, want mapped source email", got)
	}
}

// TestExecuteApplyConnectorFieldMappings_SparseFieldNotNulled asserts that when
// a mapped source field is absent with no default and no transform, the action
// SKIPS it rather than writing nil — so a sparse inbound payload never clobbers
// an existing target value with NULL (Codex round 2 B4).
func TestExecuteApplyConnectorFieldMappings_SparseFieldNotNulled(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
		// "phone" has no default and no transform; source omits it.
		{SourceField: "phone", TargetField: "Phone", Direction: connector.DirectionInbound},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	// sourceRecord has email but NOT phone.
	sourceRecord := models.SObject{"email": "user@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if _, present := updates["Phone"]; present {
		t.Errorf("Phone must be skipped (absent source, no default) to avoid nulling the target, but updates has it: %v", updates["Phone"])
	}
	if got, _ := updates["Email"].(string); got != "user@example.com" {
		t.Errorf("updates[Email] = %q, want user@example.com", got)
	}
}

// ---------------------------------------------------------------------------
// Default value used when source field is absent
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_DefaultValueUsedWhenSourceFieldMissing(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:  "status",
			TargetField:  "LeadSource",
			Direction:    connector.DirectionInbound,
			DefaultValue: "Web",
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	// sourceRecord does NOT have "status".
	sourceRecord := models.SObject{"email": "a@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["LeadSource"].(string); got != "Web" {
		t.Errorf("updates[LeadSource] = %q, want Web (default)", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_MissingReferencedTransformSourceSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "company",
			TargetField:         "Company",
			Direction:           connector.DirectionInbound,
			TransformExpression: `LEFT($Record.company, 255)`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}

	sourceRecord := models.SObject{"email": "user@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"Company": "Existing Company"})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls for missing $Record.company in sparse payload, got %d: %#v", len(cap.calls), cap.calls[0].updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_PresentAlternativeTransformSourceWrites(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "company",
			TargetField:         "Company",
			Direction:           connector.DirectionInbound,
			TransformExpression: `LEFT($Record.alternate_company, 255)`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}

	sourceRecord := models.SObject{"alternate_company": "Acme"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, present := cap.calls[0].updates["Company"]; !present || got != "Acme" {
		t.Fatalf("updates[Company] = %#v, present=%v; want Acme", got, present)
	}
}

func TestExecuteApplyConnectorFieldMappings_MultiFieldTransformRequiresAllSourcesPresent(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "first_name",
			TargetField:         "FullName",
			Direction:           connector.DirectionInbound,
			TransformExpression: `$Record.first_name & " " & $Record.last_name`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}

	sourceRecord := models.SObject{"first_name": "Ada"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"FullName": "Ada Lovelace"})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls when one referenced transform source is missing, got %d: %#v", len(cap.calls), cap.calls[0].updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_DefaultBackedReferencedTransformWrites(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "company",
			TargetField:         "Company",
			Direction:           connector.DirectionInbound,
			DefaultValue:        "Unknown Company",
			TransformExpression: `LEFT($Record.company, 255)`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}

	sourceRecord := models.SObject{"email": "user@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"Company": "Existing Company"})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call for default-backed transform, got %d", len(cap.calls))
	}
	if got, present := cap.calls[0].updates["Company"]; !present || got != "Unknown Company" {
		t.Fatalf("updates[Company] = %#v, present=%v; want Unknown Company", got, present)
	}
}

func TestExecuteApplyConnectorFieldMappings_FalseAndZeroReferencedTransformSourcesWrite(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "active",
			TargetField:         "IsActive",
			Direction:           connector.DirectionInbound,
			TransformExpression: `$Record.active`,
		},
		{
			SourceField:         "quantity",
			TargetField:         "Quantity",
			Direction:           connector.DirectionInbound,
			TransformExpression: `$Record.quantity`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}

	sourceRecord := models.SObject{"active": false, "quantity": 0}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if got, present := updates["IsActive"]; !present || got != false {
		t.Fatalf("updates[IsActive] = %#v, present=%v; want false", got, present)
	}
	if got, present := updates["Quantity"]; !present || got != 0 {
		t.Fatalf("updates[Quantity] = %#v, present=%v; want 0", got, present)
	}
}

func TestExecuteApplyConnectorFieldMappings_ValueMapTranslated(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "stage",
			TargetField: "LifecycleStage__c",
			Direction:   connector.DirectionInbound,
			ValueMap:    `{"customer":"Won","lead":"Open"}`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"stage": "customer"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["LifecycleStage__c"].(string); got != "Won" {
		t.Errorf("updates[LifecycleStage__c] = %q, want Won", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_OverwriteBlankClearUsesTargetSchemaTypes(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "firstname", TargetField: "FirstName", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
		{SourceField: "annualrevenue", TargetField: "AnnualRevenue", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
		{SourceField: "opted_out", TargetField: "HasOptedOutOfEmail", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
		{SourceField: "trial_end", TargetField: "TrialEndDate__c", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
		{SourceField: "owner", TargetField: "OwnerId", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(
				models.FieldMetadata{APIName: "FirstName", Type: constants.FieldTypeString, Updateable: true},
				models.FieldMetadata{APIName: "AnnualRevenue", Type: constants.FieldTypeCurrency, Updateable: true},
				models.FieldMetadata{APIName: "HasOptedOutOfEmail", Type: constants.FieldTypeBoolean, Updateable: true},
				models.FieldMetadata{APIName: "TrialEndDate__c", Type: constants.FieldTypeDate, Updateable: true},
				models.FieldMetadata{APIName: "OwnerId", Type: constants.FieldTypeReference, Updateable: true},
			)
		},
	}

	sourceRecord := models.SObject{
		"firstname":     "",
		"annualrevenue": "",
		"opted_out":     "",
		"trial_end":     "",
		"owner":         "",
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "hubspotContactLeadInbound",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call for schema-aware blank clears, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if got, present := updates["FirstName"]; !present || got != "" {
		t.Fatalf("updates[FirstName] = %#v, present=%v; want string blank clear", got, present)
	}
	for _, field := range []string{"AnnualRevenue", "HasOptedOutOfEmail", "TrialEndDate__c", "OwnerId"} {
		if got, present := updates[field]; !present || got != nil {
			t.Fatalf("updates[%s] = %#v, present=%v; want nil typed clear", field, got, present)
		}
	}
}

func TestExecuteApplyConnectorFieldMappings_RequiredTargetBlankOverwriteSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "company", TargetField: "Company", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
		{SourceField: "firstname", TargetField: "FirstName", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(
				models.FieldMetadata{APIName: "Company", Type: constants.FieldTypeString, Required: true, Updateable: true},
				models.FieldMetadata{APIName: "FirstName", Type: constants.FieldTypeString, Updateable: true},
			)
		},
	}

	sourceRecord := models.SObject{"company": "", "firstname": ""}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "hubspotContactLeadInbound",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call for optional blank clear only, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if got, present := updates["Company"]; present {
		t.Fatalf("updates[Company] = %#v; required blank overwrite must be skipped", got)
	}
	if got, present := updates["FirstName"]; !present || got != "" {
		t.Fatalf("updates[FirstName] = %#v, present=%v; optional blank overwrite should still clear", got, present)
	}
}

func TestExecuteApplyConnectorFieldMappings_RequiredTargetTransformBlankOverwriteSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "company",
			TargetField:         "Company",
			Direction:           connector.DirectionInbound,
			ConflictPolicy:      connector.ConflictPolicyOverwrite,
			TransformExpression: `LEFT($Record.company, 255)`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(models.FieldMetadata{APIName: "Company", Type: constants.FieldTypeString, Required: true, Updateable: true})
		},
	}

	sourceRecord := models.SObject{"company": ""}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "hubspotContactLeadInbound",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"Company": "Fallback Company"})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls for required target blank transform overwrite, got %d: %#v", len(cap.calls), cap.calls[0].updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_RequiredTargetValueMapPassthroughBlankSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "stage",
			TargetField:     "LifecycleStage__c",
			Direction:       connector.DirectionInbound,
			ConflictPolicy:  connector.ConflictPolicyOverwrite,
			ValueMap:        `{"active":"Active"}`,
			OnUnmappedValue: connector.OnUnmappedValuePassthrough,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(models.FieldMetadata{APIName: "LifecycleStage__c", Type: constants.FieldTypePicklist, Required: true, Updateable: true})
		},
	}

	sourceRecord := models.SObject{"stage": ""}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "hubspotContactLeadInbound",
		"direction":    "inbound",
		"targetId":     "lead-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"LifecycleStage__c": "Existing"})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls for required target ValueMap passthrough blank, got %d: %#v", len(cap.calls), cap.calls[0].updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_ValueMapPassthroughBlankOutboundSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "LifecycleStage__c",
			TargetField:     "lifecyclestage",
			Direction:       connector.DirectionOutbound,
			ValueMap:        `{"active":"active"}`,
			OnUnmappedValue: connector.OnUnmappedValuePassthrough,
		},
	}

	sourceRecord := models.SObject{"LifecycleStage__c": ""}
	updates, err := (&FlowExecutor{}).buildConnectorFieldMappingUpdates(sourceRecord, models.SObject{}, mappings, nil, false, nil)
	if err != nil {
		t.Fatalf("buildConnectorFieldMappingUpdates: %v", err)
	}
	if len(updates) != 0 {
		t.Fatalf("expected outbound ValueMap passthrough blank to be omitted, got %#v", updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_ValueMapAuthoredBlankOutboundPreserved(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "LifecycleStage__c",
			TargetField: "lifecyclestage",
			Direction:   connector.DirectionOutbound,
			ValueMap:    `{"retired":""}`,
		},
	}

	sourceRecord := models.SObject{"LifecycleStage__c": "retired"}
	updates, err := (&FlowExecutor{}).buildConnectorFieldMappingUpdates(sourceRecord, models.SObject{}, mappings, nil, false, nil)
	if err != nil {
		t.Fatalf("buildConnectorFieldMappingUpdates: %v", err)
	}
	if got, present := updates["lifecyclestage"]; !present || got != "" {
		t.Fatalf("updates[lifecyclestage] = %#v, present=%v; want authored blank clear", got, present)
	}
}

func TestExecuteApplyConnectorFieldMappings_TypeAuthoredMasterDetailBlankOverwriteSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "account_id",
			TargetField:    "Account__c",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyOverwrite,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(models.FieldMetadata{
				APIName:    "Account__c",
				Type:       constants.FieldTypeMasterDetail,
				Updateable: true,
			})
		},
	}

	sourceRecord := models.SObject{"account_id": ""}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Child__c",
		"actionType":   "hubspotContactChildInbound",
		"direction":    "inbound",
		"targetId":     "child-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"Account__c": "001-parent"})

	if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls for type-authored master-detail blank overwrite, got %d: %#v", len(cap.calls), cap.calls[0].updates)
	}
}

// TestExecuteApplyConnectorFieldMappings_ValueMapMissingTranslationFailsLoud
// verifies that a source value with no ValueMap translation fails loud and
// performs no update. An empty OnUnmappedValue policy defaults to this error
// behavior.
func TestExecuteApplyConnectorFieldMappings_ValueMapMissingTranslationFailsLoud(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "stage",
			TargetField: "LifecycleStage__c",
			Direction:   connector.DirectionInbound,
			ValueMap:    `{"lead":"Open"}`,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"stage": "customer"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected missing ValueMap translation to fail (empty OnUnmappedValue defaults to error), got nil")
	}
	if !strings.Contains(err.Error(), "has no translation") {
		t.Fatalf("error = %v, want missing translation", err)
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after missing ValueMap translation, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_OnUnmappedValueSkip(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "stage",
			TargetField:     "LifecycleStage__c",
			Direction:       connector.DirectionInbound,
			ValueMap:        `{"lead":"Open"}`,
			OnUnmappedValue: connector.OnUnmappedValueSkip,
		},
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"stage": "customer", "email": "test@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("OnUnmappedValue=skip should not error, got: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if _, ok := updates["LifecycleStage__c"]; ok {
		t.Errorf("LifecycleStage__c should be skipped, but was present: %v", updates["LifecycleStage__c"])
	}
	if updates["Email"] != "test@example.com" {
		t.Errorf("Email = %v, want test@example.com", updates["Email"])
	}
}

func TestExecuteApplyConnectorFieldMappings_OnUnmappedValuePassthrough(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "stage",
			TargetField:     "LifecycleStage__c",
			Direction:       connector.DirectionInbound,
			ValueMap:        `{"lead":"Open"}`,
			OnUnmappedValue: connector.OnUnmappedValuePassthrough,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"stage": "customer"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("OnUnmappedValue=passthrough should not error, got: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if updates["LifecycleStage__c"] != "customer" {
		t.Errorf("LifecycleStage__c = %v, want customer (passthrough)", updates["LifecycleStage__c"])
	}
}

func TestExecuteApplyConnectorFieldMappings_OnUnmappedValueErrorDefault(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "stage",
			TargetField:     "LifecycleStage__c",
			Direction:       connector.DirectionInbound,
			ValueMap:        `{"lead":"Open"}`,
			OnUnmappedValue: connector.OnUnmappedValueError,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"stage": "customer"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("OnUnmappedValue=error should fail on unmapped value, got nil")
	}
	if !strings.Contains(err.Error(), "has no translation") {
		t.Fatalf("error = %v, want missing translation", err)
	}
}

func TestExecuteApplyConnectorFieldMappings_OnUnmappedValueSkipEmptyString(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "stage",
			TargetField:     "LifecycleStage__c",
			Direction:       connector.DirectionInbound,
			ValueMap:        `{"lead":"Open"}`,
			OnUnmappedValue: connector.OnUnmappedValueSkip,
		},
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"stage": "", "email": "test@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("OnUnmappedValue=skip with empty string should not error, got: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if _, ok := updates["LifecycleStage__c"]; ok {
		t.Errorf("LifecycleStage__c should be skipped for empty string, but was present: %v", updates["LifecycleStage__c"])
	}
	if updates["Email"] != "test@example.com" {
		t.Errorf("Email = %v, want test@example.com", updates["Email"])
	}
}

func TestExecuteApplyConnectorFieldMappings_OnUnmappedValueNonScalarStillErrors(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:     "payload",
			TargetField:     "Data",
			Direction:       connector.DirectionInbound,
			ValueMap:        `{"lead":"Open"}`,
			OnUnmappedValue: connector.OnUnmappedValueSkip,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"payload": map[string]interface{}{"nested": "object"}}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("non-scalar source value should still error even with OnUnmappedValue=skip, got nil")
	}
	if !strings.Contains(err.Error(), "non-scalar") {
		t.Fatalf("error = %v, want non-scalar error", err)
	}
}

// ---------------------------------------------------------------------------
// TransformExpression applied
// ---------------------------------------------------------------------------

// A transform that resolves to a non-scalar (a bare $Record, or a nested
// object/array field from the source payload) must fail closed rather than write
// a whole structure into one target field. Regression for the scalar-result guard.
func TestExecuteApplyConnectorFieldMappings_NonScalarTransformFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "payload",
			TargetField:         "Data",
			Direction:           connector.DirectionInbound,
			TransformExpression: `payload`, // resolves to the nested object → non-scalar
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}

	sourceRecord := models.SObject{"payload": map[string]interface{}{"a": "1"}}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected error for non-scalar transform result, got nil (writing a map into a field is a fail-open)")
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls when transform yields a non-scalar, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_TransformExpressionApplied(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "first_name",
			TargetField:         "FirstName",
			Direction:           connector.DirectionInbound,
			TransformExpression: `UPPER(first_name)`,
		},
	}

	cap := &captureUpdate{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              eng,
	}

	sourceRecord := models.SObject{"first_name": "alice"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["FirstName"].(string); got != "ALICE" {
		t.Errorf("updates[FirstName] = %q, want ALICE (UPPER transform)", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_DefaultValueUsedByTransform(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "email",
			TargetField:         "Email",
			Direction:           connector.DirectionInbound,
			DefaultValue:        "fallback@example.com",
			TransformExpression: `UPPER(email)`,
		},
	}

	cap := &captureUpdate{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              eng,
	}

	// Source record does NOT contain "email" — default should be injected
	// into the formula context so UPPER(email) operates on the default.
	sourceRecord := models.SObject{}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["Email"].(string); got != "FALLBACK@EXAMPLE.COM" {
		t.Errorf("updates[Email] = %q, want FALLBACK@EXAMPLE.COM (UPPER of default)", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_DefaultValueUsedByTransformWhenSourceFieldExplicitNil(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "email",
			TargetField:         "Email",
			Direction:           connector.DirectionInbound,
			DefaultValue:        "fallback@example.com",
			TransformExpression: `UPPER(email)`,
		},
	}

	cap := &captureUpdate{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              eng,
	}

	sourceRecord := models.SObject{"email": nil}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["Email"].(string); got != "FALLBACK@EXAMPLE.COM" {
		t.Errorf("updates[Email] = %q, want FALLBACK@EXAMPLE.COM (UPPER of default when source field is explicit nil)", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_TransformSparsePayloadBareFieldFaults(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "email",
			TargetField:         "Email",
			Direction:           connector.DirectionInbound,
			TransformExpression: `UPPER(email)`,
		},
	}

	cap := &captureUpdate{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              eng,
	}

	sourceRecord := models.SObject{}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	action.FaultConnector = &models.FlowConnector{TargetReference: "fault_node"}
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"Email": "existing@example.com"})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls (bare field in sparse source faults), got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_TransformNilResultSparsePayloadSkipped(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "email",
			TargetField:         "Email",
			Direction:           connector.DirectionInbound,
			TransformExpression: `IF(false, "x", NULL)`,
		},
	}

	cap := &captureUpdate{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              eng,
	}

	sourceRecord := models.SObject{}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{"Email": "existing@example.com"})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls (nil transform on sparse source should skip, not clobber), got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// ConflictPolicy drift: a live row carrying an unrecognised policy value
// (manual edit / migration / future writer bug) must fail closed, not fall
// through to an implicit overwrite. Regression for the no-default switch.
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_DriftedConflictPolicyFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: "merge", // not a recognised policy — simulates DB drift
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	targetRecord := models.SObject{"Email": "existing@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected error for drifted ConflictPolicy, got nil (silent overwrite is a fail-open)")
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls when policy is unrecognised, got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// ConflictPolicy=skip_if_set leaves existing target value
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_SkipIfSet(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
		{
			SourceField:    "phone",
			TargetField:    "Phone",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{
		"email": "new@example.com",
		"phone": "+15559999",
	}
	// Caller snapshot is stale/empty; ConflictPolicy must use the authoritative
	// persistence row above, where Email is already set and Phone is not.
	targetRecord := models.SObject{
		"Email": "",
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	// Email must NOT be overwritten (skip_if_set, target already has value).
	if _, present := updates["Email"]; present {
		t.Errorf("updates[Email] should not be set (skip_if_set with existing value), got %v", updates["Email"])
	}
	// Phone MUST be written (target was empty).
	if got, _ := updates["Phone"].(string); got != "+15559999" {
		t.Errorf("updates[Phone] = %q, want +15559999", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_SkipIfSetShortCircuitsBeforeTransform(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:         "email",
			TargetField:         "Email",
			Direction:           connector.DirectionInbound,
			ConflictPolicy:      connector.ConflictPolicySkipIfSet,
			TransformExpression: `UPPER(email)`,
		},
	}

	cap := &captureUpdate{}
	eng := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              eng,
	}

	sourceRecord := models.SObject{}
	targetRecord := models.SObject{"Email": "existing@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("skip_if_set should short-circuit before transform (not fault on missing source field), got: %v", err)
	}

	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls (skip_if_set with existing target), got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_SkipIfSetTreatsBlankTargetAsEmpty(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "   "}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	targetRecord := models.SObject{"Email": "   "}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call (whitespace-only target is blank), got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["Email"].(string); got != "new@example.com" {
		t.Errorf("updates[Email] = %q, want new@example.com (blank target should be overwritten)", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_ConflictPolicyWriteUsesPessimisticLock(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
	}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	// Caller snapshot is stale in the opposite direction: it claims Email is
	// already set, but the locked authoritative row is blank and can be updated.
	targetRecord := models.SObject{"Email": "stale@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if persistence.pessimisticTxCalls != 1 {
		t.Fatalf("RunInPessimisticTransaction calls = %d, want 1", persistence.pessimisticTxCalls)
	}
	if len(persistence.lockCalls) != 1 {
		t.Fatalf("GetLock calls = %d, want 1", len(persistence.lockCalls))
	}
	if lock := persistence.lockCalls[0]; lock.objectName != "Contact" || lock.id != "rec-001" || !lock.pessimisticTx {
		t.Fatalf("GetLock call = %#v, want Contact/rec-001 inside pessimistic tx", lock)
	}
	if len(persistence.rawCalls) != 0 {
		t.Fatalf("GetRaw calls = %d, want 0 on write path", len(persistence.rawCalls))
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(cap.calls))
	}
	if !cap.calls[0].pessimisticTx {
		t.Fatal("Update was not executed inside the pessimistic transaction")
	}
	if got, _ := cap.calls[0].updates["Email"].(string); got != "new@example.com" {
		t.Errorf("updates[Email] = %q, want new@example.com", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_DuplicateTargetFieldFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
		{
			SourceField: "alternateEmail",
			TargetField: "Email",
			Direction:   connector.DirectionBidirectional,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{
		"email":          "primary@example.com",
		"alternateEmail": "alternate@example.com",
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected duplicate target field error, got nil")
	}
	if !strings.Contains(err.Error(), "duplicate target field") {
		t.Errorf("error should mention duplicate target field, got: %v", err)
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after duplicate target field fault, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_TargetSchemaUnavailableFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return nil
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "a@example.com"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected missing target schema error, got nil")
	}
	if !strings.Contains(err.Error(), "schema is unavailable") {
		t.Fatalf("error = %v, want schema unavailable", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("Update calls = %d, want 0", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_TargetFieldMissingFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "MissingEmail", Direction: connector.DirectionInbound},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(models.FieldMetadata{APIName: "Email", Type: constants.FieldTypeEmail, Updateable: true})
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected missing target field error, got nil")
	}
	if !strings.Contains(err.Error(), "does not exist") {
		t.Fatalf("error = %v, want missing field", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("Update calls = %d, want 0", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_AutoNumberTargetFieldFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "external_id", TargetField: "AutoNumber__c", Direction: connector.DirectionInbound},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(models.FieldMetadata{APIName: "AutoNumber__c", Type: constants.FieldTypeAutonumber, Updateable: true})
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"external_id": "EXT-1"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected autonumber target field error, got nil")
	}
	if !strings.Contains(err.Error(), "is autonumber") {
		t.Fatalf("error = %v, want autonumber", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("Update calls = %d, want 0", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_NonUpdateableTargetFieldFailsClosed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		objectSchemaLookup: func(context.Context, string) *models.ObjectMetadata {
			return connectorTargetSchema(models.FieldMetadata{APIName: "Email", Type: constants.FieldTypeEmail, Updateable: false})
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "a@example.com"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected non-updateable target field error, got nil")
	}
	if !strings.Contains(err.Error(), "is not updateable") {
		t.Fatalf("error = %v, want not updateable", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("Update calls = %d, want 0", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_DuplicateTargetFieldFailsBeforeSkipIfSet(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
		{
			SourceField: "alternateEmail",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{
		"email":          "primary@example.com",
		"alternateEmail": "alternate@example.com",
	}
	targetRecord := models.SObject{"Email": "existing@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected duplicate target field error before skip_if_set short-circuit, got nil")
	}
	if !strings.Contains(err.Error(), "duplicate target field") {
		t.Errorf("error should mention duplicate target field, got: %v", err)
	}
	if persistence.pessimisticTxCalls != 0 || len(persistence.lockCalls) != 0 {
		t.Fatalf("duplicate target field should fault before loading authoritative target, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after duplicate target field fault, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_DuplicateTargetFieldFailsBeforeSparseNoop(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
		{
			SourceField: "alternateEmail",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "primary@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected duplicate target field error before sparse no-op, got nil")
	}
	if !strings.Contains(err.Error(), "duplicate target field") {
		t.Errorf("error should mention duplicate target field, got: %v", err)
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after duplicate target field fault, got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// ConflictPolicy=error faults when target field is already set
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_ErrorPolicyFaultsWhenTargetSet(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyError,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	// Target already has Email — must fault.
	targetRecord := models.SObject{"Email": "existing@example.com"}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected error for ConflictPolicy=error with existing target value, got nil")
	}
	// No Update must have been called.
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after fault, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_ErrorPolicySparseSourceNoopsWhenTargetSet(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyError,
		},
	}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{}
	// Caller snapshot is irrelevant; the locked row has Email set, but there is
	// no incoming email value, so ConflictPolicy=error has no write to block.
	targetRecord := models.SObject{"Email": ""}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("sparse source should no-op before ConflictPolicy=error, got: %v", err)
	}
	if persistence.pessimisticTxCalls != 1 {
		t.Fatalf("RunInPessimisticTransaction calls = %d, want 1", persistence.pessimisticTxCalls)
	}
	if len(persistence.lockCalls) != 1 {
		t.Fatalf("GetLock calls = %d, want 1", len(persistence.lockCalls))
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls for sparse source no-op, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_ErrorPolicyBlankSourceFaultsWhenTargetSet(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyError,
		},
	}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": ""}
	targetRecord := models.SObject{"Email": ""}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected ConflictPolicy=error to fault on explicit blank source with existing target value")
	}
	if !strings.Contains(err.Error(), "ConflictPolicy=error") {
		t.Fatalf("error = %v, want ConflictPolicy=error fault", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("expected 0 Update calls after blank conflict fault, got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// ConflictPolicy=error succeeds when target field is empty
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_ErrorPolicySucceedsWhenTargetEmpty(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyError,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	targetRecord := models.SObject{} // empty target
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("expected success for ConflictPolicy=error with empty target, got: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	if got, _ := cap.calls[0].updates["Email"].(string); got != "new@example.com" {
		t.Errorf("updates[Email] = %q, want new@example.com", got)
	}
}

// ---------------------------------------------------------------------------
// Empty mapping set → no Update called (no-op success)
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_EmptyMappingsNoUpdate(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(nil), // no rows
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})
	ctx := connectorMappingContext("connector_a")

	_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
	if err != nil {
		t.Fatalf("expected no error for empty mappings, got: %v", err)
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls for empty mappings, got %d", len(cap.calls))
	}
	if flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
		t.Fatal("empty mappings must not report an engine-owned Contact write")
	}
}

// ---------------------------------------------------------------------------
// Resolver error propagates as fault
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_ResolverErrorFaults(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence: &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: func(_ context.Context, _ connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
			return nil, errors.New("db unavailable")
		},
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected error when resolver fails, got nil")
	}
}

// ---------------------------------------------------------------------------
// Connector.TargetReference forwarded on success
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_ForwardsConnector(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	action.Connector = &models.FlowConnector{TargetReference: "NextElement"}
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "a@example.com"},
		models.SObject{},
	)

	next, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if next != "NextElement" {
		t.Errorf("next = %q, want NextElement", next)
	}
}

// ---------------------------------------------------------------------------
// ConflictPolicy=skip_if_set uses authoritative target state when targetRecord is absent
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_SkipIfSetUsesAuthoritativeTargetWhenNoTargetRecord(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	// targetRecord intentionally NOT provided in params. The action must lock
	// the authoritative target row instead of trusting caller-provided snapshots.
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		// "targetRecord" deliberately omitted
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	// No Update must have been called because the authoritative row has Email set.
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after authoritative skip_if_set, got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// ConflictPolicy=error uses authoritative target state when targetRecord is absent
// ---------------------------------------------------------------------------

func TestExecuteApplyConnectorFieldMappings_ErrorPolicyUsesAuthoritativeTargetWhenNoTargetRecord(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyError,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	// targetRecord intentionally NOT provided in params. The action must lock
	// the authoritative target row and fault because Email is already set.
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		// "targetRecord" deliberately omitted
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected ConflictPolicy=error to fault from authoritative target row, got nil")
	}
	// No Update must have been called.
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after authoritative ConflictPolicy=error fault, got %d", len(cap.calls))
	}
}

// ---------------------------------------------------------------------------
// JSON round-trip sanity (verifies FlowActionCall unmarshal is compatible)
// ---------------------------------------------------------------------------

// ---------------------------------------------------------------------------
// Preview mode — no DML (FIX 4)
// ---------------------------------------------------------------------------

// TestExecuteApplyConnectorFieldMappings_PreviewModeDoesNotWrite asserts that
// when the ctx is in preview mode the action does NOT call persistence.Update
// (zero Update calls) and returns errPreviewRecordUpdateAttempt.
// This mirrors executeRecordUpdate's preview guard (executor_record_ops.go):
//
//	} else if isPreviewMode(ctx) {
//	    return fault(errPreviewRecordUpdateAttempt)
//	}
func TestExecuteApplyConnectorFieldMappings_PreviewModeDoesNotWrite(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField: "email",
			TargetField: "Email",
			Direction:   connector.DirectionInbound,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}

	// Mark ctx as preview mode — same helper used in flow_preview.go.
	ctx := WithPreviewMode(connectorMappingContext("connector_a"))

	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "a@example.com"},
		models.SObject{},
	)

	_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
	if err == nil {
		t.Fatal("expected errPreviewRecordUpdateAttempt in preview mode, got nil")
	}
	if err != errPreviewRecordUpdateAttempt {
		t.Errorf("error = %v, want errPreviewRecordUpdateAttempt", err)
	}
	// persistence.Update must NOT have been called.
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls in preview mode, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_SkipIfSetCasingDrift(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
		{
			SourceField:    "phone",
			TargetField:    "Phone",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicySkipIfSet,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{
		"email": "new@example.com",
		"phone": "+15559999",
	}
	targetRecord := models.SObject{}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}

	if len(cap.calls) != 1 {
		t.Fatalf("expected 1 Update call, got %d", len(cap.calls))
	}
	updates := cap.calls[0].updates
	if _, present := updates["Email"]; present {
		t.Errorf("updates[Email] should not be set (skip_if_set with existing value via case-insensitive match), got %v", updates["Email"])
	}
	if got, _ := updates["Phone"].(string); got != "+15559999" {
		t.Errorf("updates[Phone] = %q, want +15559999", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_ErrorPolicyCasingDrift(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{
			SourceField:    "email",
			TargetField:    "Email",
			Direction:      connector.DirectionInbound,
			ConflictPolicy: connector.ConflictPolicyError,
		},
	}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"email": "existing@example.com"}),
		connectorMappingResolverFn: stubResolver(mappings),
	}

	sourceRecord := models.SObject{"email": "new@example.com"}
	targetRecord := models.SObject{}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "connector_a",
		"sourceObject": "Contact",
		"targetObject": "Contact",
		"actionType":   "syncContact",
		"direction":    "inbound",
		"targetId":     "rec-001",
		"sourceRecord": map[string]interface{}{"elementReference": "$SourceRecord"},
		"targetRecord": map[string]interface{}{"elementReference": "$TargetRecord"},
	})
	interview := buildInterviewWithSourceRecord(sourceRecord, targetRecord)

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil {
		t.Fatal("expected error for ConflictPolicy=error with existing target value (case-insensitive match), got nil")
	}
	if len(cap.calls) != 0 {
		t.Errorf("expected 0 Update calls after fault, got %d", len(cap.calls))
	}
}

func TestExecuteApplyConnectorFieldMappings_ActionCallJSONRoundTrip(t *testing.T) {
	raw := `{
		"name": "ApplyMappings",
		"actionType": "applyConnectorFieldMappings",
		"actionName": "applyConnectorFieldMappings",
		"inputParameters": [
			{"name": "source",       "value": {"stringValue": "connector_a"}},
			{"name": "sourceObject", "value": {"stringValue": "Contact"}},
			{"name": "targetObject", "value": {"stringValue": "Contact"}},
			{"name": "actionType",   "value": {"stringValue": "syncContact"}},
			{"name": "direction",    "value": {"stringValue": "inbound"}},
			{"name": "targetId",     "value": {"stringValue": "rec-001"}},
			{"name": "sourceRecord", "value": {"elementReference": "$SourceRecord"}}
		]
	}`

	var action models.FlowActionCall
	if err := json.Unmarshal([]byte(raw), &action); err != nil {
		t.Fatalf("unmarshal FlowActionCall: %v", err)
	}
	if action.ActionType != "applyConnectorFieldMappings" {
		t.Errorf("ActionType = %q, want applyConnectorFieldMappings", action.ActionType)
	}
	if len(action.InputParameters) != 7 {
		t.Errorf("InputParameters len = %d, want 7", len(action.InputParameters))
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceDisabledPreservesLegacyDefaults(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:    "email",
		TargetField:    "Email",
		Direction:      connector.DirectionInbound,
		DefaultValue:   "fallback@example.com",
		ConflictPolicy: connector.ConflictPolicySkipIfSet,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"filterMappingsBySourcePresence": false,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 1 || len(persistence.lockCalls) != 1 {
		t.Fatalf("legacy default mapping should lock once, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 1 || cap.calls[0].updates["Email"] != "fallback@example.com" {
		t.Fatalf("legacy default update = %#v, want Email fallback", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSkipsAbsentMappingsWithoutLock(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:    "email",
		TargetField:    "Email",
		Direction:      connector.DirectionInbound,
		ConflictPolicy: connector.ConflictPolicySkipIfSet,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 0 || len(persistence.lockCalls) != 0 {
		t.Fatalf("absent filtered mapping should not lock, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 0 {
		t.Fatalf("absent filtered mapping should not update, calls=%#v", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSelectedConflictMappingLocksOnce(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:    "email",
		TargetField:    "Email",
		Direction:      connector.DirectionInbound,
		ConflictPolicy: connector.ConflictPolicySkipIfSet,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": ""}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 1 || len(persistence.lockCalls) != 1 {
		t.Fatalf("selected conflict mapping should lock once, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(cap.calls))
	}
	if got, ok := cap.calls[0].updates["Email"]; !ok || got != "" {
		t.Fatalf("explicit blank source update = %#v, want Email blank", cap.calls[0].updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSelectsTransformDependencies(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "company",
		TargetField:         "Company",
		Direction:           connector.DirectionInbound,
		ConflictPolicy:      connector.ConflictPolicySkipIfSet,
		TransformExpression: `LEFT($Record.alternate_company, 255)`,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Company": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"alternate_company": "Acme"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 1 || len(persistence.lockCalls) != 1 {
		t.Fatalf("selected transform mapping should lock once, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 1 || cap.calls[0].updates["Company"] != "Acme" {
		t.Fatalf("transform update = %#v, want Company Acme", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSelectsCaseInsensitiveTransformDependencies(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "company",
		TargetField:         "Company",
		Direction:           connector.DirectionInbound,
		ConflictPolicy:      connector.ConflictPolicySkipIfSet,
		TransformExpression: `LEFT($Record.Alternate_Company, 255)`,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Company": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"alternate_company": "Acme"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 1 || len(persistence.lockCalls) != 1 {
		t.Fatalf("case-insensitive transform dependency should lock once, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 1 || cap.calls[0].updates["Company"] != "Acme" {
		t.Fatalf("case-insensitive transform update = %#v, want Company Acme", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceDisabledPreservesLegacyTransformRecord(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "company",
		TargetField:         "Company",
		Direction:           connector.DirectionInbound,
		TransformExpression: `LEFT($Record.Alternate_Company, 255)`,
	}}

	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": false,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"alternate_company": "Acme"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want 1", len(cap.calls))
	}
	if got, ok := cap.calls[0].updates["Company"]; !ok || got != "" {
		t.Fatalf("legacy transform update = %#v, want Company blank", cap.calls[0].updates)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSkipsTransformWithoutSourceDependencies(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "ignored",
		TargetField:         "Status__c",
		Direction:           connector.DirectionInbound,
		ConflictPolicy:      connector.ConflictPolicySkipIfSet,
		TransformExpression: `IF(true, "new", "old")`,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Status__c": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"ignored": "present"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 0 || len(persistence.lockCalls) != 0 {
		t.Fatalf("transform without source dependencies should not lock, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 0 {
		t.Fatalf("transform without source dependencies should not update, calls=%#v", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSkipsTransformWithMissingDependency(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "first_name",
		TargetField:         "FullName",
		Direction:           connector.DirectionInbound,
		ConflictPolicy:      connector.ConflictPolicySkipIfSet,
		TransformExpression: `$Record.first_name & " " & $Record.last_name`,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"FullName": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"first_name": "Ada"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 0 || len(persistence.lockCalls) != 0 {
		t.Fatalf("transform with a missing dependency should not lock, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 0 {
		t.Fatalf("transform with a missing dependency should not update, calls=%#v", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceSkipsTransformWithOnlyBlankDependencies(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "company",
		TargetField:         "Company",
		Direction:           connector.DirectionInbound,
		ConflictPolicy:      connector.ConflictPolicySkipIfSet,
		TransformExpression: `LEFT($Record.alternate_company, 255)`,
	}}

	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Company": ""})
	exec := &FlowExecutor{
		persistence:                persistence,
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              false,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"alternate_company": ""}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 0 || len(persistence.lockCalls) != 0 {
		t.Fatalf("transform with only blank dependencies should not lock, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 0 {
		t.Fatalf("transform with only blank dependencies should not update, calls=%#v", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_FilterMappingsBySourcePresenceRejectsExtraUpdates(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(nil),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                         "connector_a",
		"sourceObject":                   "Contact",
		"targetObject":                   "Contact",
		"actionType":                     "syncContact",
		"direction":                      "inbound",
		"targetId":                       "rec-001",
		"applyExtraUpdates":              true,
		"filterMappingsBySourcePresence": true,
		"sourceRecord":                   map[string]interface{}{"elementReference": "$SourceRecord"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{"email": "new@example.com"}, models.SObject{})

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil || !strings.Contains(err.Error(), "filterMappingsBySourcePresence cannot be used with applyExtraUpdates") {
		t.Fatalf("error = %v, want incompatible option error", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("incompatible options should not update, calls=%#v", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesReduceByOccurredAt(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "first_name", TargetField: "FirstName", Direction: connector.DirectionInbound},
		{SourceField: "mobilephone", TargetField: "MobilePhone", Direction: connector.DirectionInbound},
	}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"sourceRecord":             map[string]interface{}{"elementReference": "$SourceRecord"},
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
	})
	interview := buildInterviewWithSourceRecord(models.SObject{
		"first_name": "stale full-record value",
	}, models.SObject{})
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "first_name", "propertyValue": "newest", "occurredAt": int64(300)}},
		models.SObject{"payload": models.SObject{"propertyName": "mobilephone", "propertyValue": "+86 13800000000", "occurredAt": int64(250)}},
		// The older event arrives later in the settled collection. occurredAt,
		// not collection order, chooses the source value.
		models.SObject{"payload": models.SObject{"propertyName": "first_name", "propertyValue": "older", "occurredAt": int64(100)}},
	}

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want one coalesced write", len(cap.calls))
	}
	if got := cap.calls[0].updates["FirstName"]; got != "newest" {
		t.Fatalf("updates[FirstName] = %#v, want newest occurredAt value", got)
	}
	if got := cap.calls[0].updates["MobilePhone"]; got != "+86 13800000000" {
		t.Fatalf("updates[MobilePhone] = %#v, want settled mobilephone value", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationUsesSelectedValue(t *testing.T) {
	const emailPattern = `^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap, txActive: true},
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"sourceRecord":             map[string]interface{}{"elementReference": "$SourceRecord"},
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
		"coalescedSourceFieldRegexValidations": []interface{}{
			map[string]interface{}{
				"sourceField": "email",
				"pattern":     emailPattern,
				"allowBlank":  true,
			},
		},
	})
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "valid-hydrated@example.test"},
		models.SObject{},
	)
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-older", "occurredAt": int64(100)}},
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "selected@example.test", "occurredAt": int64(200)}},
	}

	ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
	if _, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil); err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want one selected-value write", len(cap.calls))
	}
	if got := cap.calls[0].updates["Email"]; got != "selected@example.test" {
		t.Fatalf("updates[Email] = %#v, want newest valid value", got)
	}
	if !flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
		t.Fatal("successful selected-value mapping must publish post-upsert evidence")
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationRejectsSelectedMalformedValue(t *testing.T) {
	const emailPattern = `^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap, txActive: true},
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
		"coalescedSourceFieldRegexValidations": []interface{}{
			map[string]interface{}{
				"sourceField": "email",
				"pattern":     emailPattern,
				"allowBlank":  true,
			},
		},
	})
	interview := buildInterviewWithSourceRecord(
		models.SObject{"email": "valid-hydrated@example.test"},
		models.SObject{},
	)
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "older@example.test", "occurredAt": int64(100)}},
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-selected", "occurredAt": int64(200)}},
	}

	ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
	_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
	if err == nil || !strings.Contains(err.Error(), `selected source field "email" does not match its required pattern`) {
		t.Fatalf("error = %v, want selected source validation failure", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("malformed selected value must not update, calls=%#v", cap.calls)
	}
	if flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
		t.Fatal("malformed selected value must not publish post-upsert evidence")
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationRejectsBlankWhenNotAllowed(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	for _, tt := range []struct {
		name       string
		validation map[string]interface{}
		value      string
	}{
		{
			name: "allow blank omitted",
			validation: map[string]interface{}{
				"sourceField": "email",
				"pattern":     "^$",
			},
		},
		{
			name: "allow blank false",
			validation: map[string]interface{}{
				"sourceField": "email",
				"pattern":     `^\s*$`,
				"allowBlank":  false,
			},
			value: " \t",
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence:                &stubPersistenceCapture{capture: cap, txActive: true},
				connectorMappingResolverFn: stubResolver(mappings),
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                   "connector_a",
				"sourceObject":             "Contact",
				"targetObject":             "Contact",
				"actionType":               "syncContact",
				"direction":                "inbound",
				"targetId":                 "rec-001",
				"applyExtraUpdates":        false,
				"coalescedPropertyChanges": true,
				"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
				"coalescedSourceFieldRegexValidations": []interface{}{
					tt.validation,
				},
			})
			interview := buildInterviewWithSourceRecord(nil, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": tt.value, "occurredAt": int64(100)}},
			}

			ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
			_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
			if err == nil || !strings.Contains(err.Error(), `selected source field "email" is blank but allowBlank is false`) {
				t.Fatalf("error = %v, want blank selected source validation failure", err)
			}
			if len(cap.calls) != 0 {
				t.Fatalf("disallowed blank must not update, calls=%#v", cap.calls)
			}
			if flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
				t.Fatal("disallowed blank must not publish post-upsert evidence")
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationPreservesInlineMetadataLiterals(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap, txActive: true},
		connectorMappingResolverFn: stubResolver(mappings),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
		"coalescedSourceFieldRegexValidations": []interface{}{
			map[string]interface{}{
				"sourceField": "email",
				"pattern":     map[string]interface{}{"stringValue": `^[^@\s]+@[^@\s]+\.[^@\s]+$`},
				"allowBlank":  map[string]interface{}{"booleanValue": true},
			},
		},
	})
	interview := buildInterviewWithSourceRecord(nil, models.SObject{})
	interview.ContextData["email"] = "unrelatedRuntimeVariable"
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-selected", "occurredAt": int64(100)}},
	}

	ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
	_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
	if err == nil || !strings.Contains(err.Error(), `selected source field "email" does not match its required pattern`) {
		t.Fatalf("error = %v, want literal email validation failure", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("literal validation mismatch must not update, calls=%#v", cap.calls)
	}
	if flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
		t.Fatal("literal validation mismatch must not publish post-upsert evidence")
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationSharesCanonicalFieldIdentityWithMapping(t *testing.T) {
	const emailPattern = `^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
	}
	for _, tt := range []struct {
		name        string
		payloads    []interface{}
		wantErr     bool
		wantUpdated string
	}{
		{
			name: "newer malformed case variant fails",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "Email", "propertyValue": "older@example.test", "occurredAt": int64(100)}},
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-selected", "occurredAt": int64(200)}},
			},
			wantErr: true,
		},
		{
			name: "newer valid case variant maps",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "Email", "propertyValue": "malformed-older", "occurredAt": int64(100)}},
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "selected@example.test", "occurredAt": int64(200)}},
			},
			wantUpdated: "selected@example.test",
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence:                &stubPersistenceCapture{capture: cap, txActive: true},
				connectorMappingResolverFn: stubResolver(mappings),
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                   "connector_a",
				"sourceObject":             "Contact",
				"targetObject":             "Contact",
				"actionType":               "syncContact",
				"direction":                "inbound",
				"targetId":                 "rec-001",
				"applyExtraUpdates":        false,
				"coalescedPropertyChanges": true,
				"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
				"coalescedSourceFieldRegexValidations": []interface{}{
					map[string]interface{}{
						"sourceField": "Email",
						"pattern":     emailPattern,
						"allowBlank":  true,
					},
				},
			})
			interview := buildInterviewWithSourceRecord(nil, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = tt.payloads

			ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
			_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
			if tt.wantErr {
				if err == nil || !strings.Contains(err.Error(), `selected source field "Email" does not match its required pattern`) {
					t.Fatalf("error = %v, want canonical selected source validation failure", err)
				}
				if len(cap.calls) != 0 || flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
					t.Fatalf("canonical selected validation failure wrote state: calls=%#v postUpsert=%v", cap.calls, flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact"))
				}
				return
			}
			if err != nil {
				t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
			}
			if len(cap.calls) != 1 || cap.calls[0].updates["Email"] != tt.wantUpdated {
				t.Fatalf("updates = %#v, want canonical selected Email %q", cap.calls, tt.wantUpdated)
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationAllowsBlankAbsentAndOtherFields(t *testing.T) {
	const emailPattern = `^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`
	tests := []struct {
		name     string
		payloads []interface{}
		mappings []connector.ConnectorFieldMappingInput
		wantCall bool
	}{
		{
			name: "selected blank is allowed and skipped by mapping policy",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "", "occurredAt": int64(100)}},
			},
			mappings: []connector.ConnectorFieldMappingInput{{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound}},
		},
		{
			name: "absent validated field",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "first_name", "propertyValue": "Ada", "occurredAt": int64(100)}},
			},
			mappings: []connector.ConnectorFieldMappingInput{{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound}},
		},
		{
			name: "non validated field still maps",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "first_name", "propertyValue": "Ada", "occurredAt": int64(100)}},
			},
			mappings: []connector.ConnectorFieldMappingInput{{SourceField: "first_name", TargetField: "FirstName", Direction: connector.DirectionInbound}},
			wantCall: true,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence:                &stubPersistenceCapture{capture: cap},
				connectorMappingResolverFn: stubResolver(tt.mappings),
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                   "connector_a",
				"sourceObject":             "Contact",
				"targetObject":             "Contact",
				"actionType":               "syncContact",
				"direction":                "inbound",
				"targetId":                 "rec-001",
				"applyExtraUpdates":        false,
				"skipBlankTargetFields":    []interface{}{"Email"},
				"coalescedPropertyChanges": true,
				"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
				"coalescedSourceFieldRegexValidations": []interface{}{
					map[string]interface{}{
						"sourceField": "email",
						"pattern":     emailPattern,
						"allowBlank":  true,
					},
				},
			})
			interview := buildInterviewWithSourceRecord(nil, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = tt.payloads

			if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil); err != nil {
				t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
			}
			if got := len(cap.calls) > 0; got != tt.wantCall {
				t.Fatalf("Update called = %v, want %v; calls=%#v", got, tt.wantCall, cap.calls)
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationRejectsInvalidConfigurationBeforeResolver(t *testing.T) {
	tests := []struct {
		name        string
		coalesced   bool
		validations interface{}
	}{
		{name: "empty collection still requires coalesced mode", validations: []interface{}{}},
		{name: "requires coalesced mode", validations: []interface{}{map[string]interface{}{"sourceField": "email", "pattern": "^x$"}}},
		{name: "requires collection", coalesced: true, validations: map[string]interface{}{"sourceField": "email", "pattern": "^x$"}},
		{name: "requires source field", coalesced: true, validations: []interface{}{map[string]interface{}{"pattern": "^x$"}}},
		{name: "requires pattern", coalesced: true, validations: []interface{}{map[string]interface{}{"sourceField": "email"}}},
		{name: "requires valid pattern", coalesced: true, validations: []interface{}{map[string]interface{}{"sourceField": "email", "pattern": "["}}},
		{name: "requires boolean allow blank", coalesced: true, validations: []interface{}{map[string]interface{}{"sourceField": "email", "pattern": "^x$", "allowBlank": "yes"}}},
		{name: "rejects duplicate field", coalesced: true, validations: []interface{}{
			map[string]interface{}{"sourceField": "email", "pattern": "^x$"},
			map[string]interface{}{"sourceField": "Email", "pattern": "^y$"},
		}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			exec := &FlowExecutor{
				connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
					t.Fatal("invalid validation configuration must fail before mapping resolution")
					return nil, nil
				},
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                               "connector_a",
				"sourceObject":                         "Contact",
				"targetObject":                         "Contact",
				"actionType":                           "syncContact",
				"direction":                            "inbound",
				"targetId":                             "rec-001",
				"coalescedPropertyChanges":             tt.coalesced,
				"coalescedSourceFieldRegexValidations": tt.validations,
			})
			if _, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), buildInterviewWithSourceRecord(nil, nil), action, nil); err == nil {
				t.Fatal("invalid validation configuration unexpectedly succeeded")
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationRejectsNilOrUnresolvedAuthoredInputBeforeResolver(t *testing.T) {
	for _, tt := range []struct {
		name          string
		coalesced     bool
		explicitNil   bool
		directNil     bool
		validationRef string
	}{
		{name: "coalesced missing reference", coalesced: true, validationRef: "$MissingValidations"},
		{name: "coalesced explicit nil reference", coalesced: true, explicitNil: true, validationRef: "$NilValidations"},
		{name: "coalesced direct nil", coalesced: true, directNil: true},
		{name: "noncoalesced missing reference", validationRef: "$MissingValidations"},
		{name: "noncoalesced explicit nil reference", explicitNil: true, validationRef: "$NilValidations"},
		{name: "noncoalesced direct nil", directNil: true},
	} {
		t.Run(tt.name, func(t *testing.T) {
			resolverCalls := 0
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence: &stubPersistenceCapture{capture: cap, txActive: true},
				connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
					resolverCalls++
					return []connector.ConnectorFieldMappingInput{
						{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
					}, nil
				},
			}
			validationInput := interface{}(map[string]interface{}{"elementReference": tt.validationRef})
			if tt.directNil {
				validationInput = nil
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                               "connector_a",
				"sourceObject":                         "Contact",
				"targetObject":                         "Contact",
				"actionType":                           "syncContact",
				"direction":                            "inbound",
				"targetId":                             "rec-001",
				"sourceRecord":                         map[string]interface{}{"elementReference": "$SourceRecord"},
				"applyExtraUpdates":                    false,
				"coalescedPropertyChanges":             tt.coalesced,
				"coalescedPayloads":                    map[string]interface{}{"elementReference": "$CoalescedPayloads"},
				"coalescedSourceFieldRegexValidations": validationInput,
			})
			interview := buildInterviewWithSourceRecord(models.SObject{"email": "malformed-selected"}, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-selected", "occurredAt": int64(100)}},
			}
			if tt.explicitNil {
				interview.ContextData[tt.validationRef] = nil
			}

			ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
			_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
			if err == nil || !strings.Contains(err.Error(), "coalescedSourceFieldRegexValidations") {
				t.Fatalf("error = %v, want unresolved authored validation failure", err)
			}
			if resolverCalls != 0 {
				t.Fatalf("mapping resolver calls = %d, want zero", resolverCalls)
			}
			if len(cap.calls) != 0 || flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
				t.Fatalf("unresolved authored validation wrote state: calls=%#v postUpsert=%v", cap.calls, flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact"))
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationRejectsInvalidAuthoredShapeBeforeResolver(t *testing.T) {
	for _, tt := range []struct {
		name  string
		input interface{}
	}{
		{name: "scalar typed wrapper", input: map[string]interface{}{"stringValue": []interface{}{}}},
		{name: "mixed wrapper arms", input: map[string]interface{}{"elementReference": "$MissingValidations", "stringValue": []interface{}{}}},
		{name: "non string element reference", input: map[string]interface{}{"elementReference": []interface{}{}}},
		{name: "blank element reference", input: map[string]interface{}{"elementReference": " \t"}},
		{name: "nested multi arm literal", input: []interface{}{map[string]interface{}{
			"sourceField": map[string]interface{}{"stringValue": "email", "booleanValue": true},
			"pattern":     `.*`,
		}}},
		{name: "nested wrong type literal", input: []interface{}{map[string]interface{}{
			"sourceField": map[string]interface{}{"stringValue": []interface{}{"email"}},
			"pattern":     `.*`,
		}}},
		{name: "nested nil literal", input: []interface{}{map[string]interface{}{
			"sourceField": map[string]interface{}{"stringValue": nil},
			"pattern":     `.*`,
		}}},
	} {
		t.Run(tt.name, func(t *testing.T) {
			resolverCalls := 0
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence: &stubPersistenceCapture{capture: cap, txActive: true},
				connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
					resolverCalls++
					return []connector.ConnectorFieldMappingInput{
						{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
					}, nil
				},
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                               "connector_a",
				"sourceObject":                         "Contact",
				"targetObject":                         "Contact",
				"actionType":                           "syncContact",
				"direction":                            "inbound",
				"targetId":                             "rec-001",
				"applyExtraUpdates":                    false,
				"coalescedPropertyChanges":             true,
				"coalescedPayloads":                    map[string]interface{}{"elementReference": "$CoalescedPayloads"},
				"coalescedSourceFieldRegexValidations": tt.input,
			})
			interview := buildInterviewWithSourceRecord(nil, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-selected", "occurredAt": int64(100)}},
			}

			ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
			_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
			if err == nil || !strings.Contains(err.Error(), "coalescedSourceFieldRegexValidations") {
				t.Fatalf("error = %v, want invalid authored validation input failure", err)
			}
			if resolverCalls != 0 {
				t.Fatalf("mapping resolver calls = %d, want zero", resolverCalls)
			}
			if len(cap.calls) != 0 || flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
				t.Fatalf("invalid top-level validation input wrote state: calls=%#v postUpsert=%v", cap.calls, flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact"))
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedSourceRegexValidationRejectsDuplicateParametersBeforeResolver(t *testing.T) {
	strictValidation := []interface{}{
		map[string]interface{}{
			"sourceField": "email",
			"pattern":     `^[^@\s]+@[^@\s]+\.[^@\s]+$`,
			"allowBlank":  true,
		},
	}
	permissiveValidation := []interface{}{
		map[string]interface{}{
			"sourceField": "email",
			"pattern":     `.*`,
			"allowBlank":  true,
		},
	}
	for _, tt := range []struct {
		name   string
		first  interface{}
		second interface{}
	}{
		{name: "strict then empty", first: strictValidation, second: []interface{}{}},
		{name: "invalid then valid", first: []interface{}{map[string]interface{}{"sourceField": "email", "pattern": "["}}, second: permissiveValidation},
	} {
		t.Run(tt.name, func(t *testing.T) {
			resolverCalls := 0
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence: &stubPersistenceCapture{capture: cap, txActive: true},
				connectorMappingResolverFn: func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
					resolverCalls++
					return []connector.ConnectorFieldMappingInput{
						{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound},
					}, nil
				},
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                   "connector_a",
				"sourceObject":             "Contact",
				"targetObject":             "Contact",
				"actionType":               "syncContact",
				"direction":                "inbound",
				"targetId":                 "rec-001",
				"applyExtraUpdates":        false,
				"coalescedPropertyChanges": true,
				"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
			})
			action.InputParameters = append(action.InputParameters,
				models.FlowActionParameter{Name: "coalescedSourceFieldRegexValidations", Value: tt.first},
				models.FlowActionParameter{Name: "coalescedSourceFieldRegexValidations", Value: tt.second},
			)
			interview := buildInterviewWithSourceRecord(nil, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "malformed-selected", "occurredAt": int64(100)}},
			}

			ctx := withAtomicFlowBoundary(connectorMappingContext("connector_a"))
			_, err := exec.executeApplyConnectorFieldMappings(ctx, interview, action, nil)
			if err == nil || !strings.Contains(err.Error(), "duplicate coalescedSourceFieldRegexValidations input") {
				t.Fatalf("error = %v, want duplicate validation input failure", err)
			}
			if resolverCalls != 0 {
				t.Fatalf("mapping resolver calls = %d, want zero", resolverCalls)
			}
			if len(cap.calls) != 0 || flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact") {
				t.Fatalf("duplicate validation input wrote state: calls=%#v postUpsert=%v", cap.calls, flowRuntimeContextFromCtx(ctx).HasPostUpsertWrite("Contact"))
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesPreserveBlankWithoutUnselectedLock(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "first_name", TargetField: "FirstName", Direction: connector.DirectionInbound},
		{SourceField: "email", TargetField: "Email", Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicySkipIfSet},
	}
	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": "existing@example.com"})
	exec := &FlowExecutor{persistence: persistence, connectorMappingResolverFn: stubResolver(mappings)}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
	})
	interview := buildInterviewWithSourceRecord(nil, models.SObject{})
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "first_name", "propertyValue": "", "occurredAt": int64(300)}},
	}

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 0 || len(persistence.lockCalls) != 0 {
		t.Fatalf("unselected ConflictPolicy mapping should not lock, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want one blank-clear write", len(cap.calls))
	}
	if got, present := cap.calls[0].updates["FirstName"]; !present || got != "" {
		t.Fatalf("updates[FirstName] = %#v, want explicit blank", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesPreserveTransformedBlank(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:         "company",
		TargetField:         "Company",
		Direction:           connector.DirectionInbound,
		TransformExpression: `LEFT($Record.company, 255)`,
	}}
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(mappings),
		formulaEngine:              formula.NewEngine(),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
	})
	interview := buildInterviewWithSourceRecord(nil, models.SObject{})
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "company", "propertyValue": "", "occurredAt": int64(300)}},
	}

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if len(cap.calls) != 1 {
		t.Fatalf("Update calls = %d, want one transformed blank-clear write", len(cap.calls))
	}
	if got, present := cap.calls[0].updates["Company"]; !present || got != "" {
		t.Fatalf("updates[Company] = %#v, want transformed explicit blank", got)
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesDeriveFromSelectedLifecycleEvent(t *testing.T) {
	const (
		enteredDisqualifiedField = "hs_v2_date_entered_68053680"
		disqualifiedStage        = "68053680"
	)
	mappings := []connector.ConnectorFieldMappingInput{
		{SourceField: "lifecyclestage", TargetField: "LifecycleStage__c", Direction: connector.DirectionInbound},
		{
			SourceField:         enteredDisqualifiedField,
			TargetField:         "Disqualified_At__c",
			Direction:           connector.DirectionInbound,
			TransformExpression: "DATETIMEFROMEPOCHMILLIS($Record." + enteredDisqualifiedField + ")",
		},
	}

	for _, tc := range []struct {
		name          string
		payloads      []interface{}
		wantStage     string
		wantEnteredAt *time.Time
	}{
		{
			name: "latest selected stage materializes its timestamp",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "lifecyclestage", "propertyValue": disqualifiedStage, "occurredAt": int64(1700000000600)}},
				models.SObject{"payload": models.SObject{"propertyName": "lifecyclestage", "propertyValue": "subscriber", "occurredAt": int64(1700000000500)}},
			},
			wantStage:     disqualifiedStage,
			wantEnteredAt: pointerToTime(time.UnixMilli(1700000000600).UTC()),
		},
		{
			name: "newer non-matching stage does not use stale entry timestamp",
			payloads: []interface{}{
				models.SObject{"payload": models.SObject{"propertyName": "lifecyclestage", "propertyValue": disqualifiedStage, "occurredAt": int64(1700000000500)}},
				models.SObject{"payload": models.SObject{"propertyName": "lifecyclestage", "propertyValue": "subscriber", "occurredAt": int64(1700000000600)}},
			},
			wantStage: "subscriber",
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			cap := &captureUpdate{}
			exec := &FlowExecutor{
				persistence:                &stubPersistenceCapture{capture: cap},
				connectorMappingResolverFn: stubResolver(mappings),
				formulaEngine:              formula.NewEngine(),
			}
			action := buildConnectorMappingAction(map[string]interface{}{
				"source":                   "connector_a",
				"sourceObject":             "Contact",
				"targetObject":             "Contact",
				"actionType":               "syncContact",
				"direction":                "inbound",
				"targetId":                 "rec-001",
				"applyExtraUpdates":        false,
				"coalescedPropertyChanges": true,
				"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
				"coalescedSourceDerivations": []interface{}{
					map[string]interface{}{
						"sourceField": enteredDisqualifiedField,
						"fromField":   "lifecyclestage",
						"whenEquals":  disqualifiedStage,
						"valueFrom":   "occurredAt",
					},
				},
			})
			interview := buildInterviewWithSourceRecord(models.SObject{"lifecyclestage": "stale"}, models.SObject{})
			interview.ContextData["$CoalescedPayloads"] = tc.payloads

			_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
			if err != nil {
				t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
			}
			if len(cap.calls) != 1 {
				t.Fatalf("Update calls = %d, want one coalesced write", len(cap.calls))
			}
			if got := cap.calls[0].updates["LifecycleStage__c"]; got != tc.wantStage {
				t.Fatalf("updates[LifecycleStage__c] = %#v, want %q", got, tc.wantStage)
			}
			gotEnteredAt, present := cap.calls[0].updates["Disqualified_At__c"]
			if tc.wantEnteredAt == nil {
				if present {
					t.Fatalf("updates[Disqualified_At__c] = %#v, want absent", gotEnteredAt)
				}
				return
			}
			gotTime, ok := gotEnteredAt.(time.Time)
			if !ok || !gotTime.Equal(*tc.wantEnteredAt) {
				t.Fatalf("updates[Disqualified_At__c] = %#v, want %s", gotEnteredAt, tc.wantEnteredAt.Format(time.RFC3339Nano))
			}
		})
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesSelectedConflictMappingLocksOnce(t *testing.T) {
	mappings := []connector.ConnectorFieldMappingInput{{
		SourceField:    "email",
		TargetField:    "Email",
		Direction:      connector.DirectionInbound,
		ConflictPolicy: connector.ConflictPolicySkipIfSet,
	}}
	cap := &captureUpdate{}
	persistence := connectorPolicyPersistence(cap, "Contact", "rec-001", models.SObject{"Email": ""})
	exec := &FlowExecutor{persistence: persistence, connectorMappingResolverFn: stubResolver(mappings)}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
	})
	interview := buildInterviewWithSourceRecord(nil, models.SObject{})
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "new@example.com", "occurredAt": int64(300)}},
	}

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err != nil {
		t.Fatalf("executeApplyConnectorFieldMappings: %v", err)
	}
	if persistence.pessimisticTxCalls != 1 || len(persistence.lockCalls) != 1 {
		t.Fatalf("selected ConflictPolicy mapping should lock exactly once, tx=%d locks=%d", persistence.pessimisticTxCalls, len(persistence.lockCalls))
	}
	if len(cap.calls) != 1 || cap.calls[0].updates["Email"] != "new@example.com" {
		t.Fatalf("coalesced conflict update = %#v, want one Email update", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesRejectDerivationSourceConflict(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(nil),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
		"coalescedSourceDerivations": []interface{}{
			map[string]interface{}{
				"sourceField": "hs_v2_date_entered_68053680",
				"fromField":   "lifecyclestage",
				"whenEquals":  "68053680",
				"valueFrom":   "occurredAt",
			},
		},
	})
	interview := buildInterviewWithSourceRecord(nil, models.SObject{})
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "lifecyclestage", "propertyValue": "68053680", "occurredAt": int64(300)}},
		models.SObject{"payload": models.SObject{"propertyName": "hs_v2_date_entered_68053680", "propertyValue": "connector-authored", "occurredAt": int64(301)}},
	}

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil || !strings.Contains(err.Error(), "conflicts with a coalesced payload property") {
		t.Fatalf("error = %v, want derivation source conflict", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("conflicting derivation must not update, calls=%#v", cap.calls)
	}
}

func TestExecuteApplyConnectorFieldMappings_CoalescedPropertyChangesRejectMissingOccurredAt(t *testing.T) {
	cap := &captureUpdate{}
	exec := &FlowExecutor{
		persistence:                &stubPersistenceCapture{capture: cap},
		connectorMappingResolverFn: stubResolver(nil),
	}
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":                   "connector_a",
		"sourceObject":             "Contact",
		"targetObject":             "Contact",
		"actionType":               "syncContact",
		"direction":                "inbound",
		"targetId":                 "rec-001",
		"applyExtraUpdates":        false,
		"coalescedPropertyChanges": true,
		"coalescedPayloads":        map[string]interface{}{"elementReference": "$CoalescedPayloads"},
	})
	interview := buildInterviewWithSourceRecord(nil, models.SObject{})
	interview.ContextData["$CoalescedPayloads"] = []interface{}{
		models.SObject{"payload": models.SObject{"propertyName": "email", "propertyValue": "new@example.com"}},
	}

	_, err := exec.executeApplyConnectorFieldMappings(connectorMappingContext("connector_a"), interview, action, nil)
	if err == nil || !strings.Contains(err.Error(), "occurredAt is required") {
		t.Fatalf("error = %v, want missing occurredAt fault", err)
	}
	if len(cap.calls) != 0 {
		t.Fatalf("malformed coalesced payload should not update, calls=%#v", cap.calls)
	}
}

func pointerToTime(value time.Time) *time.Time {
	return &value
}