package automation

// executor_helpers_test.go — combined tests for:
//   * T22c (PR #217): generic $User.<f> / $Profile.<f> resolver via FlowRuntimeContext.
//   * LB01 (#10):     custom conditionLogic (1 AND (2 OR 3), nested parens, NOT)
//                     via shared evaluateConditionLogic helper.
//   * #1205:          Flow IsBlank whitespace-only string semantics + formula parity.
//   * #1205 SQL:      flow filter builder IsBlank/IsNotBlank SQL round-trip safety.
//   * #640:           Workflow RecordUpdate InputAssignment resolves myVariable_current.Field.
//   * #641:           IsChanged + multiPicklist normalizes serialization-format differences.

import (
	"context"
	"fmt"
	"go/ast"
	"go/parser"
	"go/token"
	"path/filepath"
	"runtime"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// ===========================================================================
// SECTION A — T22c: generic $User / $Profile / $UserRole resolver tests
// (originally added by PR #217; see commit 3cb7744)
// ===========================================================================

// buildTestUser returns a populated UserSession for resolver tests.
func buildTestUser() *models.UserSession {
	email := "test@example.com"
	roleID := "role-42"
	return &models.UserSession{
		ID:          "user-001",
		Name:        "Test User",
		Email:       &email,
		ProfileID:   "prof-001",
		ProfileName: "System Administrator",
		RoleID:      &roleID,
	}
}

// C3 — $User.Id resolves correctly via the generic resolver.
func TestResolveFlowUserVariable_UserID(t *testing.T) {
	user := buildTestUser()
	got := resolveFlowUserVariable(user, "$User.Id")
	if got != "user-001" {
		t.Errorf("$User.Id = %v; want %q", got, "user-001")
	}
}

// C4 — $User.ProfileId resolves correctly via the generic resolver
// (backward-parity with prior string-match case "$User.ProfileId").
func TestResolveFlowUserVariable_UserProfileId(t *testing.T) {
	user := buildTestUser()
	got := resolveFlowUserVariable(user, "$User.ProfileId")
	if got != "prof-001" {
		t.Errorf("$User.ProfileId = %v; want %q", got, "prof-001")
	}
}

// C5 — $User.<arbitrary other field> resolves correctly, proving the resolver
// is generic and not per-field. Uses $User.Name and $User.Email as examples.
func TestResolveFlowUserVariable_UserArbitraryFields(t *testing.T) {
	user := buildTestUser()

	gotName := resolveFlowUserVariable(user, "$User.Name")
	if gotName != "Test User" {
		t.Errorf("$User.Name = %v; want %q", gotName, "Test User")
	}

	gotEmail := resolveFlowUserVariable(user, "$User.Email")
	if gotEmail != "test@example.com" {
		t.Errorf("$User.Email = %v; want %q", gotEmail, "test@example.com")
	}
}

// C6 — $Profile.Name resolves correctly via the generic resolver.
func TestResolveFlowUserVariable_ProfileName(t *testing.T) {
	user := buildTestUser()
	got := resolveFlowUserVariable(user, "$Profile.Name")
	if got != "System Administrator" {
		t.Errorf("$Profile.Name = %v; want %q", got, "System Administrator")
	}
}

// C7 — $Profile.<arbitrary other field> resolves correctly via the generic resolver.
// Uses $Profile.Id as the example.
func TestResolveFlowUserVariable_ProfileArbitraryField(t *testing.T) {
	user := buildTestUser()
	got := resolveFlowUserVariable(user, "$Profile.Id")
	if got != "prof-001" {
		t.Errorf("$Profile.Id = %v; want %q", got, "prof-001")
	}
}

// $UserRole.<f> — resolves through the shared formula.BuildFormulaUserContext
// helper. Asserts that the resolver routes $UserRole.* through the same
// session-bound surface as $User and $Profile, so direct flow value sites
// (Decision, Assignment, RecordUpdate inputs) and FlowFormula resources
// agree on $UserRole.Name / $UserRole.Id within a single interview.
func TestResolveFlowUserVariable_UserRoleFields(t *testing.T) {
	roleName := "Senior Sales Rep"
	user := buildTestUser()
	user.RoleName = &roleName

	gotID := resolveFlowUserVariable(user, "$UserRole.Id")
	if gotID != "role-42" {
		t.Errorf("$UserRole.Id = %v; want %q", gotID, "role-42")
	}
	gotName := resolveFlowUserVariable(user, "$UserRole.Name")
	if gotName != "Senior Sales Rep" {
		t.Errorf("$UserRole.Name = %v; want %q", gotName, "Senior Sales Rep")
	}
}

// V13 / fail-closed — unknown $User field returns a non-nil error, not silent nil.
func TestResolveFlowUserVariable_UnknownUserField_FailClosed(t *testing.T) {
	user := buildTestUser()
	got := resolveFlowUserVariable(user, "$User.NonExistentField")
	if got == nil {
		t.Error("$User.NonExistentField = nil; want non-nil error (fail-closed)")
	}
	if _, ok := got.(error); !ok {
		t.Errorf("$User.NonExistentField type = %T; want error (fail-closed)", got)
	}
}

// V13 / fail-closed — unknown $Profile field returns a non-nil error, not silent nil.
func TestResolveFlowUserVariable_UnknownProfileField_FailClosed(t *testing.T) {
	user := buildTestUser()
	got := resolveFlowUserVariable(user, "$Profile.NonExistentField")
	if got == nil {
		t.Error("$Profile.NonExistentField = nil; want non-nil error (fail-closed)")
	}
	if _, ok := got.(error); !ok {
		t.Errorf("$Profile.NonExistentField type = %T; want error (fail-closed)", got)
	}
}

// nil user — resolveFlowUserVariable with nil user should be safe (returns nil or error,
// but must not panic).
func TestResolveFlowUserVariable_NilUser_NoPanic(t *testing.T) {
	defer func() {
		if r := recover(); r != nil {
			t.Errorf("resolveFlowUserVariable with nil user panicked: %v", r)
		}
	}()
	got := resolveFlowUserVariable(nil, "$User.Id")
	_ = got // nil or error; either is acceptable — must not panic
}

func TestResolveFlowSystemContextVariable_DerivedRecompute(t *testing.T) {
	if got := resolveFlowSystemContextVariable(context.Background(), "$SystemContext.IsDerivedRecompute"); got != false {
		t.Fatalf("$SystemContext.IsDerivedRecompute without marker = %v, want false", got)
	}

	ctx := constants.WithFormulaDerivedRecompute(context.Background())
	if got := resolveFlowSystemContextVariable(ctx, "$SystemContext.IsDerivedRecompute"); got != true {
		t.Fatalf("$SystemContext.IsDerivedRecompute with marker = %v, want true", got)
	}
	if got := resolveFlowSystemContextVariable(ctx, constants.FormulaContextSystemContext+".IsDerivedRecompute"); got != true {
		t.Fatalf("%s.IsDerivedRecompute with marker = %v, want true", constants.FormulaContextSystemContext, got)
	}
}

func TestResolveFlowSystemContextVariable_ConnectorSyncOrigin(t *testing.T) {
	if got := resolveFlowSystemContextVariable(context.Background(), "$SystemContext.ConnectorSyncOrigin"); got != "" {
		t.Fatalf("$SystemContext.ConnectorSyncOrigin without marker = %q, want blank", got)
	}

	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, " HubSpot-CRM-Connector ")
	if got := resolveFlowSystemContextVariable(ctx, "$SystemContext.ConnectorSyncOrigin"); got != "hubspot-crm-connector" {
		t.Fatalf("$SystemContext.ConnectorSyncOrigin = %q, want canonical connector namespace", got)
	}
	if got := resolveFlowSystemContextVariable(ctx, constants.FormulaContextSystemContext+".ConnectorSyncOrigin"); got != "hubspot-crm-connector" {
		t.Fatalf("%s.ConnectorSyncOrigin = %q, want canonical connector namespace", constants.FormulaContextSystemContext, got)
	}
}

func TestResolveValueCtxWithPresence_SystemContextDerivedRecompute(t *testing.T) {
	executor := &FlowExecutor{}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	ctx := constants.WithFormulaDerivedRecompute(context.Background())

	got, present := executor.resolveValueCtxWithPresence(ctx, interview, nil, map[string]interface{}{"elementReference": "$SystemContext.IsDerivedRecompute"})
	if !present {
		t.Fatal("resolveValueCtxWithPresence present = false, want true")
	}
	if got != true {
		t.Fatalf("resolveValueCtxWithPresence($SystemContext.IsDerivedRecompute) = %v, want true", got)
	}
}

func TestEvaluateConditionCtx_SystemContextDerivedRecompute(t *testing.T) {
	executor := &FlowExecutor{}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	cond := &models.FlowCondition{
		LeftValueReference: "$SystemContext.IsDerivedRecompute",
		Operator:           "EqualTo",
		RightValue:         map[string]interface{}{"booleanValue": true},
	}

	if executor.evaluateConditionCtx(context.Background(), interview, nil, cond) {
		t.Fatal("evaluateConditionCtx without derived marker = true, want false")
	}
	if !executor.evaluateConditionCtx(constants.WithFormulaDerivedRecompute(context.Background()), interview, nil, cond) {
		t.Fatal("evaluateConditionCtx with derived marker = false, want true")
	}
}

func TestEvaluateConditionCtx_SystemContextConnectorSyncOrigin(t *testing.T) {
	executor := &FlowExecutor{}
	interview := &models.FlowInterview{ContextData: map[string]interface{}{}}
	cond := &models.FlowCondition{
		LeftValueReference: "$SystemContext.ConnectorSyncOrigin",
		Operator:           "EqualTo",
		RightValue:         map[string]interface{}{"stringValue": "hubspot-crm-connector"},
	}

	if executor.evaluateConditionCtx(context.Background(), interview, nil, cond) {
		t.Fatal("evaluateConditionCtx without connector origin = true, want false")
	}
	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "HubSpot-CRM-Connector")
	if !executor.evaluateConditionCtx(ctx, interview, nil, cond) {
		t.Fatal("evaluateConditionCtx with connector origin = false, want true")
	}
}

// C8 — buildFlowFormulaContext injects __User and __Profile into Variables so that
// normalizeFormulaExpression's rewriting ($User → __User, $Profile → __Profile) works
// end-to-end for FlowFormula resources (T22c source-doc design intent).
func TestBuildFlowFormulaContext_UserProfileVariablesInjected(t *testing.T) {
	user := buildTestUser()
	fCtx := buildFlowFormulaContext(map[string]interface{}{}, nil, user)

	// __User must be present and have Id + Name fields.
	uRaw, ok := fCtx.Variables["__User"]
	if !ok {
		t.Fatal("buildFlowFormulaContext: __User not in Variables; $User.Id formulas will fail")
	}
	uMap, ok := uRaw.(map[string]interface{})
	if !ok {
		t.Fatalf("__User type = %T; want map[string]interface{}", uRaw)
	}
	if uMap["Id"] != "user-001" {
		t.Errorf("__User.Id = %v; want %q", uMap["Id"], "user-001")
	}
	if uMap["Name"] != "Test User" {
		t.Errorf("__User.Name = %v; want %q", uMap["Name"], "Test User")
	}

	// __Profile must be present with Id and Name from the session's profile reference.
	pRaw, ok := fCtx.Variables["__Profile"]
	if !ok {
		t.Fatal("buildFlowFormulaContext: __Profile not in Variables; $Profile.Name formulas will fail")
	}
	pMap, ok := pRaw.(map[string]interface{})
	if !ok {
		t.Fatalf("__Profile type = %T; want map[string]interface{}", pRaw)
	}
	if pMap["Id"] != "prof-001" {
		t.Errorf("__Profile.Id = %v; want %q", pMap["Id"], "prof-001")
	}
	if pMap["Name"] != "System Administrator" {
		t.Errorf("__Profile.Name = %v; want %q", pMap["Name"], "System Administrator")
	}
}

// C9 — buildFlowFormulaContext with nil user must NOT inject __User.
// (__Profile is also absent, but that's verified by the C8 guard regardless of user state.)
func TestBuildFlowFormulaContext_NilUser_NoUserVars(t *testing.T) {
	fCtx := buildFlowFormulaContext(map[string]interface{}{}, nil, nil)
	if _, ok := fCtx.Variables["__User"]; ok {
		t.Error("__User injected for nil user; want absent")
	}
	if _, ok := fCtx.Variables["__Profile"]; ok {
		t.Error("__Profile injected for nil user; want absent")
	}
}

// SSoT parity regression — REQUIRED by gate-merge codex review:
//
//	"add a regression that proves Flow formulas and ValidationRules see
//	 the same `$Profile.Name`/`$User` facts when the session and Profile
//	 row disagree"
//
// Both paths now consume formula.BuildFormulaUserContext as the single read
// path. This test asserts that for an identical UserSession, the flow-formula
// __User/__Profile/__UserRole projections (built inside buildFlowFormulaContext)
// match formula.BuildFormulaUserContext byte-for-byte. The validation-rule
// path (records/persistence_service.go:buildUserContext) calls the same
// helper directly, so by transitivity validation and flow agree on every
// $User/$Profile/$UserRole field for a given session — including when the
// session ProfileName disagrees with a freshly renamed Profile row in the
// DB, since neither path reads the DB row anymore.
func TestFormulaGlobals_FlowVsValidation_SameUserSession_IdenticalProjection(t *testing.T) {
	user := buildTestUser()

	flowCtx := buildFlowFormulaContext(map[string]interface{}{}, nil, user)
	sharedCtx := formula.BuildFormulaUserContext(user)

	for _, key := range []string{"__User", "__Profile", "__UserRole"} {
		flowMap, ok := flowCtx.Variables[key].(map[string]interface{})
		if !ok {
			t.Fatalf("flow %s missing or wrong type: %T", key, flowCtx.Variables[key])
		}
		sharedMap, ok := sharedCtx[key].(map[string]interface{})
		if !ok {
			t.Fatalf("shared %s missing or wrong type: %T", key, sharedCtx[key])
		}
		if len(flowMap) != len(sharedMap) {
			t.Errorf("%s len mismatch: flow=%d shared=%d", key, len(flowMap), len(sharedMap))
		}
		for k, sharedVal := range sharedMap {
			if flowVal, ok := flowMap[k]; !ok || flowVal != sharedVal {
				t.Errorf("%s.%s diverged: flow=%v shared=%v", key, k, flowVal, sharedVal)
			}
		}
	}
}

func TestResolveValueCtxWithPresence_TypedLiteralsDoNotResolveContextNameCollisions(t *testing.T) {
	executor := &FlowExecutor{}
	interview := &models.FlowInterview{
		ContextData: map[string]interface{}{
			"Candidate":            true,
			"42":                   "not the number",
			"false":                "not the bool",
			"2026-05-14":           "not the date",
			"2026-05-14T00:00:00Z": "not the datetime",
		},
	}

	tests := []struct {
		name  string
		value interface{}
		want  interface{}
	}{
		{
			name:  "stringValue",
			value: map[string]interface{}{"stringValue": "Candidate"},
			want:  "Candidate",
		},
		{
			name:  "numberValue",
			value: map[string]interface{}{"numberValue": "42"},
			want:  "42",
		},
		{
			name:  "booleanValue",
			value: map[string]interface{}{"booleanValue": false},
			want:  false,
		},
		{
			name:  "dateValue",
			value: map[string]interface{}{"dateValue": "2026-05-14"},
			want:  "2026-05-14",
		},
		{
			name:  "dateTimeValue",
			value: map[string]interface{}{"dateTimeValue": "2026-05-14T00:00:00Z"},
			want:  "2026-05-14T00:00:00Z",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got, present := executor.resolveValueCtxWithPresence(context.Background(), interview, nil, tt.value)
			if !present {
				t.Fatal("resolveValueCtxWithPresence present = false, want true")
			}
			if got != tt.want {
				t.Fatalf("resolveValueCtxWithPresence = %#v (%T), want %#v (%T)", got, got, tt.want, tt.want)
			}
		})
	}
}

// ===========================================================================
// SECTION B — LB01: custom conditionLogic tests (issue #10 / line-B cleanup)
// ===========================================================================

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

// minimalExecutor returns a *FlowExecutor with all nil fields.  Safe for
// evaluateConditions / evaluateConditionsCtx because those methods only call
// evaluateCondition / evaluateConditionCtx, which in turn call resolveValue /
// resolveValueCtx — neither of which requires metadata, query, or other
// injected services when operating on plain ContextData maps with literal
// (non-__r) references.
func minimalExecutor() *FlowExecutor {
	return &FlowExecutor{}
}

// ===========================================================================
// SECTION C — issue-602-sub-645: scheduled-start pushdown SSoT tests
// ===========================================================================

func TestScheduledStartPushdown_OperatorSet_MatchesEvaluator(t *testing.T) {
	want := append(append([]string{}, flowBinaryRuntimeOperators...), "IsChanged", "EntersPickTag")
	if strings.Join(scheduledStartValidRuntimeOperators, "\x00") != strings.Join(want, "\x00") {
		t.Fatalf("scheduledStartValidRuntimeOperators = %#v, want %#v", scheduledStartValidRuntimeOperators, want)
	}

	executor := &FlowExecutor{}
	for _, operator := range flowBinaryRuntimeOperators {
		value := interface{}(stringValue("Open"))
		switch operator {
		case "Between", "NotBetween":
			value = []interface{}{"A", "Z"}
		case "In", "NotIn":
			value = []interface{}{"Open", "Closed"}
		case "IsNull", "IsNotNull", "IsBlank", "IsNotBlank":
			value = true
		}
		if _, handled := executor.evaluateFlowBinaryOperator(operator, "Open", value, flowBinaryOperatorOptions{}); !handled {
			t.Fatalf("flowBinaryRuntimeOperators includes %q, but evaluateFlowBinaryOperator did not handle it", operator)
		}
	}

	for _, operator := range scheduledStartValidRuntimeOperators {
		value := interface{}(stringValue("Open"))
		switch operator {
		case "Between", "NotBetween":
			value = []interface{}{"A", "Z"}
		case "In", "NotIn":
			value = []interface{}{"Open", "Closed"}
		case "IsNull", "IsNotNull", "IsBlank", "IsNotBlank":
			value = map[string]interface{}{"booleanValue": true}
		}
		clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
			Field:    "Status",
			Operator: operator,
			Value:    value,
		}, nil)
		if err != nil {
			t.Fatalf("scheduledStartPushdown(%q) error = %v, want nil", operator, err)
		}
		if fallback {
			if clause != "" {
				t.Fatalf("scheduledStartPushdown(%q) clause = %q, want empty clause on fallback", operator, clause)
			}
			continue
		}
		if clause == "" {
			t.Fatalf("scheduledStartPushdown(%q) returned empty clause without fallback", operator)
		}
		if _, _, err := formula.ToSQLWithMapper(clause, func(s string) string { return s }); err != nil {
			t.Fatalf("scheduledStartPushdown(%q) clause %q failed formula.ToSQLWithMapper round-trip: %v", operator, clause, err)
		}
	}
}

func TestFlowRegexDecisionOperators(t *testing.T) {
	const emailPattern = `^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$`

	tests := []struct {
		name     string
		operator string
		value    interface{}
		pattern  interface{}
		want     bool
	}{
		{name: "matches_valid_email", operator: "MatchesRegex", value: "valid@example.com", pattern: emailPattern, want: true},
		{name: "rejects_malformed_email", operator: "MatchesRegex", value: "not-an-email", pattern: emailPattern, want: false},
		{name: "negated_matches_malformed_email", operator: "DoesNotMatchRegex", value: "not-an-email", pattern: emailPattern, want: true},
		{name: "negated_rejects_valid_email", operator: "DoesNotMatchRegex", value: "valid@example.com", pattern: emailPattern, want: false},
		{name: "invalid_pattern_fails_closed", operator: "MatchesRegex", value: "valid@example.com", pattern: "[", want: false},
		{name: "negated_invalid_pattern_fails_closed", operator: "DoesNotMatchRegex", value: "valid@example.com", pattern: "[", want: false},
		{name: "nil_value_does_not_match", operator: "MatchesRegex", value: nil, pattern: emailPattern, want: false},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			if _, handled := evaluateFlowBinaryOperatorCore(tc.operator, tc.value, tc.pattern, flowBinaryOperatorOptions{}, nil); handled {
				t.Fatalf("evaluateFlowBinaryOperatorCore(%q) must stay unavailable to record filters", tc.operator)
			}

			interview := &models.FlowInterview{ContextData: map[string]interface{}{
				"record": models.SObject{"email": tc.value},
			}}
			condition := &models.FlowCondition{
				LeftValueReference: "record.email",
				Operator:           tc.operator,
				RightValue:         tc.pattern,
			}
			if got := minimalExecutor().evaluateConditionCtx(context.Background(), interview, nil, condition); got != tc.want {
				t.Fatalf("evaluateConditionCtx(%q, %v, %v) = %v, want %v", tc.operator, tc.value, tc.pattern, got, tc.want)
			}
		})
	}
}

func TestFlowRegexOperatorsFailClosedOutsideDecisionConditions(t *testing.T) {
	executor := &FlowExecutor{}
	for _, operator := range []string{"MatchesRegex", "DoesNotMatchRegex"} {
		t.Run(operator, func(t *testing.T) {
			filter := models.FlowRecordFilter{
				Field:    "Email",
				Operator: operator,
				Value:    stringValue(`^[a-z]+@example\\.test$`),
			}
			if executor.evaluateRecordFilterForFieldCtx(context.Background(), models.SObject{"Email": "valid@example.test"}, nil, &filter, nil) {
				t.Fatalf("%s record filter must fail closed", operator)
			}
			if got := executor.buildFilterFromConditions(nil, []models.FlowRecordFilter{filter}, "and"); got != "1=0" {
				t.Fatalf("buildFilterFromConditions(%s) = %q, want 1=0", operator, got)
			}
			if _, _, err := executor.scheduledStartPushdown(filter, nil); err == nil {
				t.Fatalf("scheduledStartPushdown(%s) error = nil, want unsupported-operator error", operator)
			}
		})
	}
}

func TestScheduledStartPushdown_InvalidField_FailsClosed(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "Bad Field",
		Operator: "EqualTo",
		Value:    stringValue("Open"),
	}, nil)
	if err == nil {
		t.Fatal("scheduledStartPushdown() error = nil, want invalid field error")
	}
	if clause != "" {
		t.Fatalf("scheduledStartPushdown() clause = %q, want empty clause on invalid field", clause)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown() fallback = true, want fail-closed invalid metadata")
	}
}

func TestScheduledStartPushdown_UnknownOperator_FailsClosed(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "Status",
		Operator: "BogusOp",
		Value:    stringValue("Open"),
	}, nil)
	if err == nil {
		t.Fatal("scheduledStartPushdown() error = nil, want unknown operator error")
	}
	if clause != "" {
		t.Fatalf("scheduledStartPushdown() clause = %q, want empty clause on unknown operator", clause)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown() fallback = true, want fail-closed invalid metadata")
	}
}

func TestScheduledStartPushdown_MissingRequiredValue_FailsClosed(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "Status",
		Operator: "EqualTo",
	}, nil)
	if err == nil {
		t.Fatal("scheduledStartPushdown() error = nil, want missing value error")
	}
	if clause != "" {
		t.Fatalf("scheduledStartPushdown() clause = %q, want empty clause on missing value", clause)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown() fallback = true, want fail-closed invalid metadata")
	}
}

func TestBuildScheduledStartFilter_AnyFallbackDropsAggregate(t *testing.T) {
	executor := &FlowExecutor{}
	filters := []models.FlowRecordFilter{
		{Field: "Status", Operator: "EqualTo", Value: stringValue("Open")},
		{Field: "Status", Operator: "In", Value: []interface{}{"Open", "Pending"}},
	}

	for _, logic := range []string{"AND", "OR", "1 AND (2 OR 1)"} {
		t.Run(logic, func(t *testing.T) {
			filterExpr, err := executor.buildScheduledStartFilter(nil, filters, logic)
			if err != nil {
				t.Fatalf("buildScheduledStartFilter() error = %v", err)
			}
			if filterExpr != "" {
				t.Fatalf("buildScheduledStartFilter() = %q, want empty aggregate on fallback", filterExpr)
			}
		})
	}
}

func TestScheduledStartPushdown_NotEqualTo_NullFieldRetained(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "Status",
		Operator: "NotEqualTo",
		Value:    stringValue("Open"),
	}, nil)
	if err != nil {
		t.Fatalf("scheduledStartPushdown() error = %v", err)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown() fallback = true, want state-2 clause")
	}
	if clause != "(Status != 'Open' OR ISBLANK(Status))" {
		t.Fatalf("scheduledStartPushdown() clause = %q, want %q", clause, "(Status != 'Open' OR ISBLANK(Status))")
	}
}

func TestScheduledStartPushdown_LikeOperatorsEscapeWildcards(t *testing.T) {
	executor := &FlowExecutor{}

	tests := []struct {
		operator string
		want     string
	}{
		{operator: "Contains", want: "CONTAINS(Status, 'A\\%\\_B')"},
		{operator: "NotContains", want: "(NOT(CONTAINS(Status, 'A\\%\\_B')) OR ISBLANK(Status))"},
		{operator: "EndsWith", want: "ENDS_WITH(Status, 'A\\%\\_B')"},
		{operator: "DoesNotEndWith", want: "(NOT(ENDS_WITH(Status, 'A\\%\\_B')) OR ISBLANK(Status))"},
	}

	for _, tt := range tests {
		t.Run(tt.operator, func(t *testing.T) {
			clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
				Field:    "Status",
				Operator: tt.operator,
				Value:    stringValue("A%_B"),
			}, nil)
			if err != nil {
				t.Fatalf("scheduledStartPushdown(%q) error = %v", tt.operator, err)
			}
			if fallback {
				t.Fatalf("scheduledStartPushdown(%q) fallback = true, want state-2 clause", tt.operator)
			}
			if clause != tt.want {
				t.Fatalf("scheduledStartPushdown(%q) clause = %q, want %q", tt.operator, clause, tt.want)
			}
		})
	}
}

func TestBuildFilterFromConditions_MultiPicklistOperatorsUseCompatiblePredicate(t *testing.T) {
	executor := &FlowExecutor{}

	tests := []struct {
		name     string
		operator string
		context  map[string]interface{}
		value    interface{}
		want     string
	}{
		{
			name:     "Includes comma-bearing token",
			operator: "Includes",
			value:    "North, East",
			want:     utils.MultiPicklistJSONContainsSQL("Tags__c", "'North, East'", true),
		},
		{
			name:     "Excludes exact token",
			operator: "Excludes",
			value:    "Hot",
			want:     utils.MultiPicklistJSONContainsSQL("Tags__c", "'Hot'", false),
		},
		{
			name:     "Includes typed literal preserves token-wise any match",
			operator: "Includes",
			value:    map[string]interface{}{"stringValue": "North;East"},
			want:     utils.MultiPicklistJSONContainsAnySQL("Tags__c", []string{"'North'", "'East'"}),
		},
		{
			name:     "Excludes typed literal uses token-wise none match",
			operator: "Excludes",
			value:    map[string]interface{}{"stringValue": "Hot;Cold"},
			want:     utils.MultiPicklistJSONContainsNoneSQL("Tags__c", []string{"'Hot'", "'Cold'"}),
		},
		{
			name:     "Includes JSON-looking typed literal as exact option",
			operator: "Includes",
			value:    map[string]interface{}{"stringValue": `["Others"]`},
			want:     utils.MultiPicklistJSONContainsSQL("Tags__c", `'["Others"]'`, true),
		},
		{
			name:     "Excludes JSON-looking typed literal as exact option",
			operator: "Excludes",
			value:    map[string]interface{}{"stringValue": `["Others"]`},
			want:     utils.MultiPicklistJSONContainsSQL("Tags__c", `'["Others"]'`, false),
		},
		{
			name:     "Includes JSON-looking bare string as exact option",
			operator: "Includes",
			value:    `["Others"]`,
			want:     utils.MultiPicklistJSONContainsSQL("Tags__c", `'["Others"]'`, true),
		},
		{
			name:     "Includes legacy bare string keeps token-wise any match",
			operator: "Includes",
			value:    "North;East",
			want:     utils.MultiPicklistJSONContainsAnySQL("Tags__c", []string{"'North'", "'East'"}),
		},
		{
			name:     "Excludes legacy bare string keeps token-wise none match",
			operator: "Excludes",
			value:    "Hot;Cold",
			want:     utils.MultiPicklistJSONContainsNoneSQL("Tags__c", []string{"'Hot'", "'Cold'"}),
		},
		{
			name:     "Includes structured token container keeps token-wise any match",
			operator: "Includes",
			value:    []interface{}{"North", "East"},
			want:     utils.MultiPicklistJSONContainsAnySQL("Tags__c", []string{"'North'", "'East'"}),
		},
		{
			name:     "Excludes structured token container keeps token-wise none match",
			operator: "Excludes",
			value:    []interface{}{"Hot", "Cold"},
			want:     utils.MultiPicklistJSONContainsNoneSQL("Tags__c", []string{"'Hot'", "'Cold'"}),
		},
		{
			name:     "Includes JSON-looking reference string as exact option",
			operator: "Includes",
			context:  map[string]interface{}{"wantedTags": `["Others"]`},
			value:    map[string]interface{}{"elementReference": "wantedTags"},
			want:     utils.MultiPicklistJSONContainsSQL("Tags__c", `'["Others"]'`, true),
		},
		{
			name:     "Excludes empty typed literal matches all rows",
			operator: "Excludes",
			value:    map[string]interface{}{"stringValue": ""},
			want:     "1 = 1",
		},
		{
			name:     "Excludes whitespace typed literal matches all rows",
			operator: "Excludes",
			value:    map[string]interface{}{"stringValue": "   "},
			want:     "1 = 1",
		},
		{
			name:     "Includes reference value preserves token-wise any match",
			operator: "Includes",
			context:  map[string]interface{}{"wantedTags": "North;East"},
			value:    map[string]interface{}{"elementReference": "wantedTags"},
			want:     utils.MultiPicklistJSONContainsAnySQL("Tags__c", []string{"'North'", "'East'"}),
		},
		{
			name:     "Excludes reference value preserves token-wise none match",
			operator: "Excludes",
			context:  map[string]interface{}{"wantedTags": "Hot;Cold"},
			value:    map[string]interface{}{"elementReference": "wantedTags"},
			want:     utils.MultiPicklistJSONContainsNoneSQL("Tags__c", []string{"'Hot'", "'Cold'"}),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := executor.buildFilterFromConditions(tt.context, []models.FlowRecordFilter{{
				Field:    "Tags__c",
				Operator: tt.operator,
				Value:    tt.value,
			}}, "AND")
			if got != tt.want {
				t.Fatalf("buildFilterFromConditions(%s) = %q, want %q", tt.operator, got, tt.want)
			}
		})
	}
}

func TestMultiPicklistIncludesSemicolonLiteralPreservesFlowCompatibility(t *testing.T) {
	executor := minimalExecutor()

	tests := []struct {
		name  string
		value interface{}
		want  bool
	}{
		{name: "both tokens selected", value: "C;B;A", want: true},
		{name: "one token selected", value: "C;A", want: true},
		{name: "no tokens selected", value: "C", want: false},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			flowCondition := &models.FlowCondition{
				LeftValueReference: "Tags__c",
				Operator:           "Includes",
				RightValue:         map[string]interface{}{"stringValue": "A;B"},
			}
			gotAutomation := executor.evaluateCondition(map[string]interface{}{"Tags__c": tt.value}, flowCondition)
			if gotAutomation != tt.want {
				t.Fatalf("automation Includes = %v, want %v", gotAutomation, tt.want)
			}
		})
	}

	gotSQL := executor.buildFilterFromConditions(map[string]interface{}{}, []models.FlowRecordFilter{{
		Field:    "Tags__c",
		Operator: "Includes",
		Value:    map[string]interface{}{"stringValue": "A;B"},
	}}, "AND")
	wantSQL := utils.MultiPicklistJSONContainsAnySQL("Tags__c", []string{"'A'", "'B'"})
	if gotSQL != wantSQL {
		t.Fatalf("automation SQL Includes = %q, want %q", gotSQL, wantSQL)
	}

	gotAutomation := executor.evaluateCondition(map[string]interface{}{
		"Tags__c":      `["A","B"]`,
		"WantedTag__c": "A;B",
	}, &models.FlowCondition{
		LeftValueReference: "Tags__c",
		Operator:           "Includes",
		RightValue:         map[string]interface{}{"elementReference": "WantedTag__c"},
	})
	if !gotAutomation {
		t.Fatalf("automation dynamic Includes = false, want true tokenized RHS compatibility")
	}

	gotDynamicSQL := executor.buildFilterFromConditions(map[string]interface{}{"wantedTags": "A;B"}, []models.FlowRecordFilter{{
		Field:    "Tags__c",
		Operator: "Includes",
		Value:    map[string]interface{}{"elementReference": "wantedTags"},
	}}, "AND")
	wantDynamicSQL := utils.MultiPicklistJSONContainsAnySQL("Tags__c", []string{"'A'", "'B'"})
	if gotDynamicSQL != wantDynamicSQL {
		t.Fatalf("automation SQL dynamic Includes = %q, want %q", gotDynamicSQL, wantDynamicSQL)
	}
}

func TestScheduledStartPushdown_EqualTo_UsesTypedLiteral(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "IsActive",
		Operator: "EqualTo",
		Value:    map[string]interface{}{"booleanValue": true},
	}, nil)
	if err != nil {
		t.Fatalf("scheduledStartPushdown() error = %v", err)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown() fallback = true, want state-2 clause")
	}
	if clause != "IsActive = true" {
		t.Fatalf("scheduledStartPushdown() clause = %q, want %q", clause, "IsActive = true")
	}
}

func TestScheduledStartPushdown_IsChanged_StateTwoBroadPass(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "Status",
		Operator: "IsChanged",
	}, nil)
	if err != nil {
		t.Fatalf("scheduledStartPushdown() error = %v", err)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown() fallback = true, want locked state-2 tuple")
	}
	if clause != "true" {
		t.Fatalf("scheduledStartPushdown() clause = %q, want %q", clause, "true")
	}
}

func TestScheduledStartPushdown_KeyPrefixSensitiveStartsWithOperatorsFallback(t *testing.T) {
	executor := &FlowExecutor{}

	for _, operator := range []string{"StartsWith", "DoesNotStartWith"} {
		t.Run(operator, func(t *testing.T) {
			clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
				Field:    "WhatId",
				Operator: operator,
				Value:    stringValue("006"),
			}, nil)
			if err != nil {
				t.Fatalf("scheduledStartPushdown(%q) error = %v", operator, err)
			}
			if !fallback {
				t.Fatalf("scheduledStartPushdown(%q) fallback = false, want state-3 fallback", operator)
			}
			if clause != "" {
				t.Fatalf("scheduledStartPushdown(%q) clause = %q, want empty clause on fallback", operator, clause)
			}
		})
	}
}

// trueCondition builds a FlowCondition that evaluates to true against a
// context map containing key=>"__true__": left==right via EqualTo.
// The key must be present in the context map passed to evaluateConditions.
func trueCondition(key string) models.FlowCondition {
	return models.FlowCondition{
		LeftValueReference: key,
		Operator:           "EqualTo",
		RightValue:         map[string]interface{}{"stringValue": "__true__"},
	}
}

// falseCondition builds a FlowCondition that evaluates to false: left != right.
func falseCondition(key string) models.FlowCondition {
	return models.FlowCondition{
		LeftValueReference: key,
		Operator:           "EqualTo",
		RightValue:         map[string]interface{}{"stringValue": "__false__"},
	}
}

// makeContext returns a context map where each key maps to "__true__".
// Use trueCondition(key) / falseCondition(key) to make conditions resolve
// to true / false against this map.
func makeContext(trueKeys ...string) map[string]interface{} {
	m := make(map[string]interface{}, len(trueKeys))
	for _, k := range trueKeys {
		m[k] = "__true__"
	}
	return m
}

// makeInterview wraps a context map in a minimal *models.FlowInterview so
// that evaluateConditionsCtx can resolve variables via ContextData.
func makeInterview(ctx map[string]interface{}) *models.FlowInterview {
	return &models.FlowInterview{ContextData: ctx}
}

// ---------------------------------------------------------------------------
// AC.1–AC.3: evaluateConditionLogic — shared helper exists and handles
// the three legacy dispatch paths ("", "and", "or").
// Covered here and also by AC.9.
// ---------------------------------------------------------------------------

// TestEvaluateConditionLogic_AC9_LegacyDispatch covers AC.9: the legacy "",
// "and", and "or" logic strings produce the same behaviour as before.
func TestEvaluateConditionLogic_AC9_LegacyDispatch(t *testing.T) {
	allTrue := []bool{true, true, true}
	mixed := []bool{true, false, true}
	allFalse := []bool{false, false, false}

	// --- empty string → AND semantics ---

	// all true → true
	if got := evaluateConditionLogic(allTrue, ""); !got {
		t.Error("AC.9 empty: [true,true,true] → want true, got false")
	}
	// mixed → false (AND)
	if got := evaluateConditionLogic(mixed, ""); got {
		t.Error("AC.9 empty: [true,false,true] → want false, got true")
	}
	// all false → false
	if got := evaluateConditionLogic(allFalse, ""); got {
		t.Error("AC.9 empty: [false,false,false] → want false, got true")
	}

	// --- "and" (lowercase) → AND semantics ---

	if got := evaluateConditionLogic(allTrue, "and"); !got {
		t.Error("AC.9 'and': [true,true,true] → want true, got false")
	}
	if got := evaluateConditionLogic(mixed, "and"); got {
		t.Error("AC.9 'and': [true,false,true] → want false, got true")
	}

	// --- "AND" (uppercase) → AND semantics ---

	if got := evaluateConditionLogic(allTrue, "AND"); !got {
		t.Error("AC.9 'AND': [true,true,true] → want true, got false")
	}
	if got := evaluateConditionLogic(mixed, "AND"); got {
		t.Error("AC.9 'AND': [true,false,true] → want false, got true")
	}

	// --- "or" (lowercase) → OR semantics ---

	if got := evaluateConditionLogic(mixed, "or"); !got {
		t.Error("AC.9 'or': [true,false,true] → want true, got false")
	}
	if got := evaluateConditionLogic(allFalse, "or"); got {
		t.Error("AC.9 'or': [false,false,false] → want false, got true")
	}

	// --- "OR" (uppercase) → OR semantics ---

	if got := evaluateConditionLogic(mixed, "OR"); !got {
		t.Error("AC.9 'OR': [true,false,true] → want true, got false")
	}
}

// ---------------------------------------------------------------------------
// AC.4: 1 AND (2 OR 3)
// ---------------------------------------------------------------------------

func TestEvaluateConditionLogic_AC4_AndOrNested(t *testing.T) {
	expr := "1 AND (2 OR 3)"

	// [true, false, true] → 1=true, (2 OR 3)=(false OR true)=true → true
	if got := evaluateConditionLogic([]bool{true, false, true}, expr); !got {
		t.Errorf("AC.4: %q [true,false,true] → want true, got false", expr)
	}

	// [false, false, true] → 1=false → whole expression false
	if got := evaluateConditionLogic([]bool{false, false, true}, expr); got {
		t.Errorf("AC.4: %q [false,false,true] → want false, got true", expr)
	}
}

// ---------------------------------------------------------------------------
// AC.5: (1 OR 2) AND 3
// ---------------------------------------------------------------------------

func TestEvaluateConditionLogic_AC5_OrAndNested(t *testing.T) {
	expr := "(1 OR 2) AND 3"

	// [false, true, true] → (false OR true)=true, 3=true → true
	if got := evaluateConditionLogic([]bool{false, true, true}, expr); !got {
		t.Errorf("AC.5: %q [false,true,true] → want true, got false", expr)
	}

	// [true, true, false] → (true OR true)=true, 3=false → false
	if got := evaluateConditionLogic([]bool{true, true, false}, expr); got {
		t.Errorf("AC.5: %q [true,true,false] → want false, got true", expr)
	}
}

// ---------------------------------------------------------------------------
// AC.6: 1 OR (2 AND NOT 3)  — tests NOT support
// ---------------------------------------------------------------------------

func TestEvaluateConditionLogic_AC6_NotSupport(t *testing.T) {
	expr := "1 OR (2 AND NOT 3)"

	// [false, true, false] → 1=false, (2 AND NOT 3)=(true AND NOT false)=(true AND true)=true → true
	if got := evaluateConditionLogic([]bool{false, true, false}, expr); !got {
		t.Errorf("AC.6: %q [false,true,false] → want true, got false", expr)
	}

	// [false, true, true] → 1=false, (2 AND NOT 3)=(true AND NOT true)=(true AND false)=false → false
	if got := evaluateConditionLogic([]bool{false, true, true}, expr); got {
		t.Errorf("AC.6: %q [false,true,true] → want false, got true", expr)
	}

	// [true, false, true] → 1=true → true (short-circuits OR)
	if got := evaluateConditionLogic([]bool{true, false, true}, expr); !got {
		t.Errorf("AC.6: %q [true,false,true] → want true, got false", expr)
	}
}

// ---------------------------------------------------------------------------
// AC.7: out-of-range reference → fallback AND, no panic
// ---------------------------------------------------------------------------

func TestEvaluateConditionLogic_AC7_OutOfRange(t *testing.T) {
	// Only 3 conditions; condition 99 is out of range.
	expr := "1 AND 99"

	// Must not panic regardless of input values.
	got := evaluateConditionLogic([]bool{true, true, true}, expr)

	// Fallback to AND over all results: all true → true
	if !got {
		t.Errorf("AC.7: %q [true,true,true] fallback-AND → want true, got false", expr)
	}

	// Fallback AND over [true, false, true] → false
	got2 := evaluateConditionLogic([]bool{true, false, true}, expr)
	if got2 {
		t.Errorf("AC.7: %q [true,false,true] fallback-AND → want false, got true", expr)
	}
}

// ---------------------------------------------------------------------------
// AC.8: unparseable expression → fallback AND, no panic
// ---------------------------------------------------------------------------

func TestEvaluateConditionLogic_AC8_Unparseable(t *testing.T) {
	// "1 AND AND" is syntactically invalid.
	expr := "1 AND AND"

	// Must not panic.
	got := evaluateConditionLogic([]bool{true, true, true}, expr)

	// Fallback AND over all true → true
	if !got {
		t.Errorf("AC.8: %q [true,true,true] fallback-AND → want true, got false", expr)
	}

	// Fallback AND over [true, false, true] → false
	got2 := evaluateConditionLogic([]bool{true, false, true}, expr)
	if got2 {
		t.Errorf("AC.8: %q [true,false,true] fallback-AND → want false, got true", expr)
	}
}

// ---------------------------------------------------------------------------
// AC.10: evaluateConditionsCtx wired to same shared helper as evaluateConditions
// ---------------------------------------------------------------------------

// TestEvaluateConditions_AC10_SharedHelper exercises both evaluateConditions
// and evaluateConditionsCtx with a custom logic string and the same inputs,
// asserting identical results.  This proves evaluateConditionLogic is wired
// into both code paths.
func TestEvaluateConditions_AC10_SharedHelper(t *testing.T) {
	exec := minimalExecutor()

	// Three conditions against a context map / interview:
	//   cond1 ("k1") → true  (k1=="__true__")
	//   cond2 ("k2") → false (k2 is absent from context)
	//   cond3 ("k3") → true  (k3=="__true__")
	contextMap := makeContext("k1", "k3")
	conds := []models.FlowCondition{
		trueCondition("k1"),
		falseCondition("k2"), // k2 absent → left resolves to nil ≠ "__false__"
		trueCondition("k3"),
	}

	// For cond2 to consistently evaluate to false across both paths we need
	// k2 absent from the map so left is nil; right is "__false__" → nil != "__false__"
	// via EqualTo → false.  That is what falseCondition("k2") achieves when k2
	// is not in the context.

	logic := "1 AND (2 OR 3)" // [true, false, true] → true AND (false OR true) → true

	// --- evaluateConditions ---
	got1 := exec.evaluateConditions(contextMap, conds, logic)

	// --- evaluateConditionsCtx ---
	interview := makeInterview(contextMap)
	got2 := exec.evaluateConditionsCtx(context.Background(), interview, nil, conds, logic)

	if got1 != got2 {
		t.Errorf("AC.10: evaluateConditions=%v, evaluateConditionsCtx=%v; want identical results", got1, got2)
	}

	// Additionally assert the expected value (true) so a regression is obvious.
	if !got1 {
		t.Errorf("AC.10: expected true for %q with [true,false,true], got false", logic)
	}

	// Second case: logic that should return false.
	logic2 := "1 AND 2" // [true, false, true] → true AND false → false
	got3 := exec.evaluateConditions(contextMap, conds, logic2)
	got4 := exec.evaluateConditionsCtx(context.Background(), interview, nil, conds, logic2)

	if got3 != got4 {
		t.Errorf("AC.10 (false case): evaluateConditions=%v, evaluateConditionsCtx=%v; want identical results", got3, got4)
	}
	if got3 {
		t.Errorf("AC.10 (false case): expected false for %q with [true,false,true], got true", logic2)
	}
}

// ---------------------------------------------------------------------------
// AC.1–AC.3 explicit: shared helper exists (function-level)
// ---------------------------------------------------------------------------

// TestEvaluateConditionLogic_AC1_HelperExists verifies AC.1: the shared
// helper is a real callable function (compilation proves existence; this
// test adds an explicit runtime call to satisfy AND-coverage).
func TestEvaluateConditionLogic_AC1_HelperExists(t *testing.T) {
	// Call with both dispatch paths to confirm the function is reachable.
	if !evaluateConditionLogic([]bool{true}, "") {
		t.Error("AC.1: evaluateConditionLogic([]bool{true}, '') should return true")
	}
	if evaluateConditionLogic([]bool{false}, "and") {
		t.Error("AC.1: evaluateConditionLogic([]bool{false}, 'and') should return false")
	}
	if !evaluateConditionLogic([]bool{false, true}, "or") {
		t.Error("AC.1: evaluateConditionLogic([]bool{false,true}, 'or') should return true")
	}
}

// TestEvaluateConditions_AC2_NoOwnSwitch verifies AC.2 at runtime: calling
// evaluateConditions with "or" dispatches to OR semantics via the shared
// helper (if evaluateConditions had its own switch it would still behave
// correctly, but this test at least exercises the delegation path so a
// coverage-based check can confirm no duplicated switch block is reachable).
func TestEvaluateConditions_AC2_NoOwnSwitch(t *testing.T) {
	exec := minimalExecutor()
	ctx := makeContext("k1", "k3")
	conds := []models.FlowCondition{
		trueCondition("k1"),
		falseCondition("k2"),
		trueCondition("k3"),
	}
	// "or": first true condition is k1 → result must be true
	if !exec.evaluateConditions(ctx, conds, "or") {
		t.Error("AC.2: evaluateConditions with 'or' and first cond=true should return true")
	}
	// "and": k2 evaluates to false → result must be false
	if exec.evaluateConditions(ctx, conds, "and") {
		t.Error("AC.2: evaluateConditions with 'and' and k2=false should return false")
	}
}

// TestEvaluateConditionsCtx_AC3_NoOwnSwitch verifies AC.3: evaluateConditionsCtx
// delegates to the shared helper for "or" and "and".
func TestEvaluateConditionsCtx_AC3_NoOwnSwitch(t *testing.T) {
	exec := minimalExecutor()
	ctxMap := makeContext("k1", "k3")
	interview := makeInterview(ctxMap)
	conds := []models.FlowCondition{
		trueCondition("k1"),
		falseCondition("k2"),
		trueCondition("k3"),
	}
	// "or": first condition true → result must be true
	if !exec.evaluateConditionsCtx(context.Background(), interview, nil, conds, "or") {
		t.Error("AC.3: evaluateConditionsCtx with 'or' and first cond=true should return true")
	}
	// "and": k2 false → result must be false
	if exec.evaluateConditionsCtx(context.Background(), interview, nil, conds, "and") {
		t.Error("AC.3: evaluateConditionsCtx with 'and' and k2=false should return false")
	}
}

// ===========================================================================
// SECTION C — #1205: isBlankValue whitespace semantics + formula parity
// ===========================================================================

// TestIsBlankValueWhitespace verifies that isBlankValue follows the current
// Salesforce-compatible Flow semantics for issue #1205: whitespace-only strings
// are blank.
func TestIsBlankValueWhitespace(t *testing.T) {
	tests := []struct {
		name  string
		input interface{}
		want  bool
	}{
		// Issue #1205 parity: whitespace-only strings are blank.
		{name: "spaces_only", input: "   ", want: true},
		{name: "tab_only", input: "\t", want: true},
		{name: "newline_only", input: "\n", want: true},
		// Already-correct cases pinned for regression.
		{name: "nil", input: nil, want: true},
		{name: "empty_string", input: "", want: true},
		{name: "non_blank_text", input: "hello", want: false},
		{name: "text_with_spaces", input: " hello ", want: false},
		// ISBLANK(0) is false — zero is a value, not blank.
		{name: "zero_int", input: 0, want: false},
		{name: "zero_float64", input: float64(0), want: false},
		// Booleans are never blank.
		{name: "bool_false", input: false, want: false},
		{name: "bool_true", input: true, want: false},
		// Empty collections are blank.
		{name: "empty_slice", input: []interface{}{}, want: true},
		{name: "non_empty_slice", input: []interface{}{"x"}, want: false},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got := isBlankValue(tc.input)
			if got != tc.want {
				t.Errorf("isBlankValue(%#v) = %v, want %v", tc.input, got, tc.want)
			}
		})
	}
}

// TestFlowIsBlankWhitespaceOperatorSemantics verifies the actual Flow runtime
// evaluators, not just the helper. It covers the positive IsBlank path and the
// negated metadata shape (`operator=IsBlank`, `booleanValue=false`).
func TestFlowIsBlankWhitespaceOperatorSemantics(t *testing.T) {
	exec := minimalExecutor()

	cases := []struct {
		name      string
		value     interface{}
		wantBlank bool
	}{
		{name: "nil", value: nil, wantBlank: true},
		{name: "empty", value: "", wantBlank: true},
		{name: "spaces", value: "   ", wantBlank: true},
		{name: "non_blank", value: "hello", wantBlank: false},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			record := models.SObject{"Name": tc.value}
			interview := &models.FlowInterview{
				ContextData: map[string]interface{}{
					"$Record": record,
				},
			}

			recordFilterBlank := &models.FlowRecordFilter{
				Field:    "Name",
				Operator: "IsBlank",
				Value:    map[string]interface{}{"booleanValue": true},
			}
			if got := exec.evaluateRecordFilterForFieldCtx(context.Background(), record, nil, recordFilterBlank, nil); got != tc.wantBlank {
				t.Fatalf("evaluateRecordFilterForFieldCtx(IsBlank=true, %#v) = %v, want %v", tc.value, got, tc.wantBlank)
			}

			recordFilterNegated := &models.FlowRecordFilter{
				Field:    "Name",
				Operator: "IsBlank",
				Value:    map[string]interface{}{"booleanValue": false},
			}
			if got := exec.evaluateRecordFilterForFieldCtx(context.Background(), record, nil, recordFilterNegated, nil); got != !tc.wantBlank {
				t.Fatalf("evaluateRecordFilterForFieldCtx(IsBlank=false, %#v) = %v, want %v", tc.value, got, !tc.wantBlank)
			}

			decisionBlank := &models.FlowCondition{
				LeftValueReference: "$Record.Name",
				Operator:           "IsBlank",
				RightValue:         map[string]interface{}{"booleanValue": true},
			}
			if got := exec.evaluateConditionCtx(context.Background(), interview, nil, decisionBlank); got != tc.wantBlank {
				t.Fatalf("evaluateConditionCtx(IsBlank=true, %#v) = %v, want %v", tc.value, got, tc.wantBlank)
			}

			decisionNegated := &models.FlowCondition{
				LeftValueReference: "$Record.Name",
				Operator:           "IsBlank",
				RightValue:         map[string]interface{}{"booleanValue": false},
			}
			if got := exec.evaluateConditionCtx(context.Background(), interview, nil, decisionNegated); got != !tc.wantBlank {
				t.Fatalf("evaluateConditionCtx(IsBlank=false, %#v) = %v, want %v", tc.value, got, !tc.wantBlank)
			}
		})
	}
}

// TestFlowIsBlankWhitespaceDistinctFromFormula verifies the intentional
// contract split for issue #1205: Flow IsBlank treats whitespace-only strings
// as blank, while formula ISBLANK keeps the Salesforce formula nil-or-empty rule.
func TestFlowIsBlankWhitespaceDistinctFromFormula(t *testing.T) {
	engine := formula.NewEngine()

	inputs := []struct {
		name        string
		value       interface{}
		expr        string
		wantFlow    bool
		wantFormula bool
	}{
		{name: "spaces", value: "   ", expr: `ISBLANK("   ")`, wantFlow: true, wantFormula: false},
		{name: "empty", value: "", expr: `ISBLANK("")`, wantFlow: true, wantFormula: true},
		{name: "non_blank", value: "hello", expr: `ISBLANK("hello")`, wantFlow: false, wantFormula: false},
	}

	for _, tc := range inputs {
		t.Run(tc.name, func(t *testing.T) {
			flowResult := isBlankValue(tc.value)
			if flowResult != tc.wantFlow {
				t.Fatalf("isBlankValue(%#v) = %v, want %v", tc.value, flowResult, tc.wantFlow)
			}

			got, err := engine.Evaluate(tc.expr, &formula.Context{})
			if err != nil {
				t.Fatalf("formula.Evaluate(%q) error = %v", tc.expr, err)
			}
			formulaResult, ok := got.(bool)
			if !ok {
				t.Fatalf("formula.Evaluate(%q) returned %T, want bool", tc.expr, got)
			}
			if formulaResult != tc.wantFormula {
				t.Fatalf("formula.Evaluate(%q) = %v, want %v", tc.expr, formulaResult, tc.wantFormula)
			}
			if tc.name == "spaces" && flowResult == formulaResult {
				t.Fatalf("whitespace-only strings must stay contract-distinct: flow=%v formula=%v", flowResult, formulaResult)
			}
		})
	}
}

// ===========================================================================
// SECTION D — #1205 SQL: flow filter builder IsBlank/IsNotBlank round-trip safety
// ===========================================================================

// identityResolver is a no-op value resolver for filter builder tests.
func identityResolver(v interface{}) interface{} { return v }

// TestFlowFilterBuilder_ISBLANK_RoundTripsThroughFormulaToSQL verifies that
// the SQL fragment emitted by buildFilterFromConditionsImpl for IsBlank and
// IsNotBlank operators round-trips cleanly through formula.ToSQLWithMapper.
//
// Flow filter strings are passed as FilterExpr to QueryWithFilter, which calls
// formula.ToSQLWithMapper (analytics/query_service.go → persistence/query_repository.go:129).
// The expression SQL walker only accepts functions it knows by name; emitting
// CHAR_LENGTH(...) directly (as round-2 commit ba3d76a did) caused
// "unsupported function: CHAR_LENGTH" at query time for every RecordLookup/
// RecordUpdate/RecordDelete using IsBlank or IsNotBlank filters.
func TestFlowFilterBuilder_ISBLANK_RoundTripsThroughFormulaToSQL(t *testing.T) {
	e := minimalExecutor()
	cases := []struct {
		name     string
		operator string
		value    interface{}
		wantFrag string
	}{
		{"IsBlank_true", "IsBlank", true, "FLOWISBLANK(Name)"},
		{"IsNotBlank", "IsNotBlank", nil, "NOT(FLOWISBLANK(Name))"},
		{"IsBlank_false", "IsBlank", false, "NOT(FLOWISBLANK(Name))"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			filters := []models.FlowRecordFilter{
				{Field: "Name", Operator: tc.operator, Value: tc.value},
			}
			sqlFrag := e.buildFilterFromConditionsImpl(filters, "AND", false, identityResolver)
			if sqlFrag == "" {
				t.Fatalf("buildFilterFromConditionsImpl returned empty string for %s", tc.name)
			}
			if sqlFrag != tc.wantFrag {
				t.Fatalf("buildFilterFromConditionsImpl(%s) = %q, want %q", tc.name, sqlFrag, tc.wantFrag)
			}
			_, _, err := formula.ToSQLWithMapper(sqlFrag, func(s string) string { return s })
			if err != nil {
				t.Errorf("flow filter fragment %q failed formula.ToSQLWithMapper round-trip: %v", sqlFrag, err)
			}
		})
	}
}

func TestFlowFilterBuilder_OrdinaryNotInKeepsLegacySQL(t *testing.T) {
	e := minimalExecutor()
	filters := []models.FlowRecordFilter{{
		Field:    "Status",
		Operator: "NotIn",
		Value:    []interface{}{"Closed", "Canceled"},
	}}

	const want = "Status NOT IN '[Closed Canceled]'"
	if got := e.buildFilterFromConditionsImpl(filters, "AND", false, identityResolver); got != want {
		t.Fatalf("ordinary buildFilterFromConditionsImpl(NotIn) = %q, want %q", got, want)
	}
	if got := e.buildFilterFromConditionsImpl(filters, "AND", true, identityResolver); got != want {
		t.Fatalf("ordinary PB buildFilterFromConditionsImpl(NotIn) = %q, want %q", got, want)
	}
}

func TestScheduledStartPushdown_IsBlankBooleanValueFalse_UsesNegatedBlankPredicate(t *testing.T) {
	executor := &FlowExecutor{}

	clause, fallback, err := executor.scheduledStartPushdown(models.FlowRecordFilter{
		Field:    "Status",
		Operator: "IsBlank",
		Value:    map[string]interface{}{"booleanValue": false},
	}, nil)
	if err != nil {
		t.Fatalf("scheduledStartPushdown(IsBlank=false) error = %v", err)
	}
	if fallback {
		t.Fatal("scheduledStartPushdown(IsBlank=false) fallback = true, want state-2 clause")
	}
	if clause != "NOT(FLOWISBLANK(Status))" {
		t.Fatalf("scheduledStartPushdown(IsBlank=false) clause = %q, want %q", clause, "NOT(FLOWISBLANK(Status))")
	}
}

func TestScheduledStartPushdown_IsNullPredicatesStayCastSafe(t *testing.T) {
	executor := &FlowExecutor{}

	tests := []struct {
		name       string
		filter     models.FlowRecordFilter
		wantClause string
	}{
		{
			name: "IsNull_true",
			filter: models.FlowRecordFilter{
				Field:    "Amount",
				Operator: "IsNull",
				Value:    map[string]interface{}{"booleanValue": true},
			},
			wantClause: "ISBLANK(Amount)",
		},
		{
			name: "IsNull_false",
			filter: models.FlowRecordFilter{
				Field:    "Amount",
				Operator: "IsNull",
				Value:    map[string]interface{}{"booleanValue": false},
			},
			wantClause: "NOT(ISBLANK(Amount))",
		},
		{
			name: "IsNotNull",
			filter: models.FlowRecordFilter{
				Field:    "Amount",
				Operator: "IsNotNull",
				Value:    map[string]interface{}{"booleanValue": true},
			},
			wantClause: "NOT(ISBLANK(Amount))",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			clause, fallback, err := executor.scheduledStartPushdown(tc.filter, nil)
			if err != nil {
				t.Fatalf("scheduledStartPushdown(%s) error = %v", tc.name, err)
			}
			if fallback {
				t.Fatalf("scheduledStartPushdown(%s) fallback = true, want state-2 clause", tc.name)
			}
			if clause != tc.wantClause {
				t.Fatalf("scheduledStartPushdown(%s) clause = %q, want %q", tc.name, clause, tc.wantClause)
			}
			sql, _, err := formula.ToSQLWithMapper(clause, func(s string) string { return s })
			if err != nil {
				t.Fatalf("formula.ToSQLWithMapper(%q) error = %v", clause, err)
			}
			if !strings.Contains(sql, "CAST(") {
				t.Fatalf("scheduledStartPushdown(%s) SQL = %q, want CAST-safe blank predicate", tc.name, sql)
			}
		})
	}
}

// ===========================================================================
// SECTION E — #598: Workflow myVariable_current/myVariable_old alias injection
// Parity tests: verifier (initializeFlowContext) must inject aliases iff
// processType=="Workflow". Non-Workflow flows must not receive aliases.
// ===========================================================================

// TestWorkflow_MyVariableCurrentAliased_DirectField verifies that for a
// processType=Workflow flow, initializeFlowContext injects myVariable_current
// as an alias for $Record and myVariable_old as an alias for $Record__Old.
func TestWorkflow_MyVariableCurrentAliased_DirectField(t *testing.T) {
	record := models.SObject{"Id": "001abc", "Name": "Test Account", "Amount__c": float64(1000)}
	oldRecord := models.SObject{"Id": "001abc", "Name": "Old Account", "Amount__c": float64(500)}

	flowMeta := &models.FlowMetadata{
		Variables: []models.FlowVariable{
			{Name: "myVariable_current", DataType: "SObject", ObjectType: "Account", IsInput: true},
			{Name: "myVariable_old", DataType: "SObject", ObjectType: "Account", IsInput: true},
		},
	}
	inputs := map[string]interface{}{
		"$Record":      models.SObject(record),
		"$Record__Old": models.SObject(oldRecord),
		"trigger":      "AfterSave",
	}

	ctx := initializeFlowContext(flowMeta, inputs, "Workflow")

	gotCurrent, ok := ctx["myVariable_current"]
	if !ok {
		t.Fatal("myVariable_current not in ctx for Workflow flow")
	}
	if fmt.Sprintf("%v", gotCurrent) != fmt.Sprintf("%v", record) {
		t.Errorf("myVariable_current = %v; want %v", gotCurrent, record)
	}

	gotOld, ok := ctx["myVariable_old"]
	if !ok {
		t.Fatal("myVariable_old not in ctx for Workflow flow")
	}
	if fmt.Sprintf("%v", gotOld) != fmt.Sprintf("%v", oldRecord) {
		t.Errorf("myVariable_old = %v; want %v", gotOld, oldRecord)
	}

	// Standard inputs must still be present.
	if ctx["$Record"] == nil {
		t.Error("$Record missing from ctx")
	}
	if ctx["trigger"] != "AfterSave" {
		t.Errorf("trigger = %v; want AfterSave", ctx["trigger"])
	}
}

// TestWorkflow_MyVariableCurrentAliased_CrossObjectViaR_3Hop verifies that the
// alias preserves the full nested record structure, enabling 3-hop cross-object
// field traversal (myVariable_current.Order__r.Account__r.Owner__r.Name).
func TestWorkflow_MyVariableCurrentAliased_CrossObjectViaR_3Hop(t *testing.T) {
	// Simulate an Opportunity record with 3-hop nested __r relations.
	nestedRecord := models.SObject{
		"Id": "006opp",
		"Order__r": map[string]interface{}{
			"Id":        "801ord",
			"AccountId": "001acc",
			"Account__r": map[string]interface{}{
				"Id": "001acc",
				"Owner__r": map[string]interface{}{
					"Id":   "005usr",
					"Name": "Partner Name",
				},
			},
		},
	}

	flowMeta := &models.FlowMetadata{
		Variables: []models.FlowVariable{
			{Name: "myVariable_current", DataType: "SObject", ObjectType: "Opportunity", IsInput: true},
		},
	}
	inputs := map[string]interface{}{
		"$Record": models.SObject(nestedRecord),
	}

	ctx := initializeFlowContext(flowMeta, inputs, "Workflow")

	aliased, ok := ctx["myVariable_current"]
	if !ok {
		t.Fatal("myVariable_current not in ctx for Workflow flow with 3-hop record")
	}
	aliasedMap, ok := aliased.(models.SObject)
	if !ok {
		t.Fatalf("myVariable_current type = %T; want models.SObject", aliased)
	}
	if aliasedMap["Id"] != "006opp" {
		t.Errorf("myVariable_current[Id] = %v; want 006opp", aliasedMap["Id"])
	}

	// Verify all 3 hops of the nested structure are intact.
	orderR, ok := aliasedMap["Order__r"].(map[string]interface{})
	if !ok {
		t.Fatal("myVariable_current.Order__r missing or wrong type after alias")
	}
	accountR, ok := orderR["Account__r"].(map[string]interface{})
	if !ok {
		t.Fatal("myVariable_current.Order__r.Account__r missing or wrong type after alias")
	}
	ownerR, ok := accountR["Owner__r"].(map[string]interface{})
	if !ok {
		t.Fatal("myVariable_current.Order__r.Account__r.Owner__r missing or wrong type after alias")
	}
	if ownerR["Name"] != "Partner Name" {
		t.Errorf("3-hop Owner__r.Name = %v; want Partner Name", ownerR["Name"])
	}
}

// TestNonWorkflow_NoAliasInjection_NoRegression verifies that non-Workflow
// processTypes do NOT alias myVariable_current to $Record, even when such a
// variable is declared in the flow.
func TestNonWorkflow_NoAliasInjection_NoRegression(t *testing.T) {
	record := models.SObject{"Id": "001abc", "Name": "Test"}

	flowMeta := &models.FlowMetadata{
		Variables: []models.FlowVariable{
			// Variable named myVariable_current declared but in a non-Workflow flow.
			{Name: "myVariable_current", DataType: "SObject", ObjectType: "Account"},
		},
	}
	inputs := map[string]interface{}{
		"$Record": models.SObject(record),
	}

	nonWorkflowTypes := []string{"AutoLaunchedFlow", "ScheduledFlow", "Flow", ""}
	for _, processType := range nonWorkflowTypes {
		t.Run("processType="+processType, func(t *testing.T) {
			ctx := initializeFlowContext(flowMeta, inputs, processType)

			// myVariable_current has no default Value, so it must NOT be present
			// in ctx for non-Workflow flows (no alias injection).
			if val, present := ctx["myVariable_current"]; present {
				recordStr := fmt.Sprintf("%v", record)
				if fmt.Sprintf("%v", val) == recordStr {
					t.Errorf("processType=%q: myVariable_current aliased to $Record; must not be for non-Workflow flows", processType)
				}
			}
		})
	}
}

// ===========================================================================
// SECTION D — issue #902: executeWait must use the ctx-aware evaluator so
// wait-event Includes/Excludes get field-aware multi-picklist comparison.
// ===========================================================================

// TestExecuteWait_UsesEvaluateConditionsCtx is a source-level structural
// assertion: executeWait's body MUST call e.evaluateConditionsCtxE(...),
// not the legacy non-ctx e.evaluateConditions(...). The legacy path is
// option-blind on Includes/Excludes (the non-field-aware includes()/excludes())
// and would silently fail the
// same multi-picklist class that #902 fixes on Decision/RecordFilter.
//
// Going through evaluateConditionsCtxE routes each wait-event condition
// through evaluateConditionCtx, which applies the lookupTriggerFieldMeta
// gate and dispatches to includesForField/excludesForField when the LHS
// is a multi-picklist on the trigger object.
//
// AST parse rather than a full executeWait integration test because the
// element's pause path persists a FlowInterview row via
// pauseForWait → e.persistence.Insert, requiring extensive fixture
// scaffolding for a binary structural assertion.
func TestExecuteWait_UsesEvaluateConditionsCtx(t *testing.T) {
	_, thisFile, _, ok := runtime.Caller(0)
	if !ok {
		t.Fatal("runtime.Caller(0) failed")
	}

	fset := token.NewFileSet()
	file, err := parser.ParseFile(
		fset,
		filepath.Join(filepath.Dir(thisFile), "executor_elements.go"),
		nil,
		0,
	)
	if err != nil {
		t.Fatalf("ParseFile(executor_elements.go): %v", err)
	}

	var executeWait *ast.FuncDecl
	for _, decl := range file.Decls {
		fn, ok := decl.(*ast.FuncDecl)
		if !ok || fn.Name == nil || fn.Name.Name != "executeWait" || fn.Body == nil {
			continue
		}
		executeWait = fn
		break
	}
	if executeWait == nil {
		t.Fatal("executeWait function not found in executor_elements.go")
	}

	var (
		foundCtx    bool
		foundLegacy bool
	)
	ast.Inspect(executeWait.Body, func(n ast.Node) bool {
		call, ok := n.(*ast.CallExpr)
		if !ok {
			return true
		}
		sel, ok := call.Fun.(*ast.SelectorExpr)
		if !ok {
			return true
		}
		switch sel.Sel.Name {
		case "evaluateConditionsCtxE":
			foundCtx = true
		case "evaluateConditions":
			foundLegacy = true
		}
		return true
	})

	if !foundCtx {
		t.Error("executeWait body must call e.evaluateConditionsCtxE(...) — wait-event conditions need field-aware multi-picklist routing and Formula Resource error propagation (#902)")
	}
	if foundLegacy {
		t.Error("executeWait body must NOT call e.evaluateConditions(...) — legacy non-ctx evaluator bypasses includesForField/excludesForField and silently fails declared-literal multi-picklist (#902 R2/R3 BLOCKER from crossfire)")
	}
}

// ===========================================================================
// SECTION F — #640: Workflow RecordUpdate InputAssignment reads from
// myVariable_current dotted reference.
//
// Regression: in Workflow-processType flows, InputAssignment.Value of the
// form "myVariable_current.SomeField" must resolve against $Record (the
// aliased context entry), not return nil/absent.
// ===========================================================================

// TestWorkflow_RecordUpdate_InputAssignment_ReadsFromMyVariableCurrent verifies
// that a Workflow-processType flow can use "myVariable_current.SourceField" as
// the Value in a RecordUpdate InputAssignment and write the correct value.
//
// End-to-end chain:
//  1. initializeFlowContext aliases myVariable_current → $Record (Section E, #598).
//  2. resolveValueCtxWithPresence routes "myVariable_current.SourceField" through
//     resolveNestedReferenceCtxWithPresence (dotted + looksLikeVariableReference).
//  3. The nested resolver walks interview.ContextData["myVariable_current"]["SourceField"].
//  4. The resolved value ends up in the updates map passed to persistence.Update.
func TestWorkflow_RecordUpdate_InputAssignment_ReadsFromMyVariableCurrent(t *testing.T) {
	// Build a triggering record that has a SourceField value.
	triggerRecord := models.SObject{
		constants.FieldID: "obj-001",
		"SourceField":     "expected_value",
		"attributes":      map[string]interface{}{"type": "GenericObject"},
	}

	// Build the FlowMetadata with a Workflow-style myVariable_current variable.
	flowMeta := &models.FlowMetadata{
		Variables: []models.FlowVariable{
			{Name: "myVariable_current", DataType: "SObject", ObjectType: "GenericObject", IsInput: true},
		},
	}

	// Simulate the alias injection performed by initializeFlowContext for
	// processType=Workflow before a flow element executes.
	contextData := initializeFlowContext(flowMeta, map[string]interface{}{
		"$Record": models.SObject(triggerRecord),
	}, "Workflow")

	interview := &models.FlowInterview{
		ContextData: contextData,
	}

	// RecordUpdate element: write the trigger record's own field using a
	// value read from myVariable_current.SourceField.
	update := &models.FlowRecordUpdate{
		Name:           "Update_Target",
		InputReference: "$Record",
		Object:         "GenericObject",
		InputAssignments: []models.FlowInputAssignment{
			// Value is a bare reference string (not a typed wrapper) — this is the
			// form Workflow flows use for "read from the current record variable".
			{Field: "TargetField", Value: "myVariable_current.SourceField"},
		},
		Connector: &models.FlowConnector{TargetReference: "Done"},
	}

	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, nil, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Done" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Done")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
	}
	if got := cap.calls[0].updates["TargetField"]; got != "expected_value" {
		t.Fatalf("TargetField update = %#v (%T), want %q", got, got, "expected_value")
	}
}

// TestWorkflow_RecordUpdate_InputAssignment_ReadsCrossObjectPathFromMyVariableCurrent
// verifies the exact #1774 shape: a Workflow recordUpdate writes the current
// record's field from a related-record path rooted at myVariable_current.
func TestWorkflow_RecordUpdate_InputAssignment_ReadsCrossObjectPathFromMyVariableCurrent(t *testing.T) {
	triggerRecord := models.SObject{
		constants.FieldID: "rev-001",
		"Order__c":        "ord-001",
		"Partner_Name__c": nil,
		"Order__r": map[string]interface{}{
			constants.FieldID: "ord-001",
			"Partner_Name__c": "partner-001",
			"attributes":      map[string]interface{}{"type": "Order__c"},
		},
		"attributes": map[string]interface{}{"type": "Revenue__c"},
	}

	flowMeta := &models.FlowMetadata{
		Variables: []models.FlowVariable{
			{Name: "myVariable_current", DataType: "SObject", ObjectType: "Revenue__c", IsInput: true},
		},
	}

	contextData := initializeFlowContext(flowMeta, map[string]interface{}{
		"$Record": models.SObject(triggerRecord),
	}, "Workflow")

	interview := &models.FlowInterview{
		ContextData: contextData,
	}

	update := &models.FlowRecordUpdate{
		Name:           "Sync_Partner_Name",
		InputReference: "$Record",
		Object:         "Revenue__c",
		InputAssignments: []models.FlowInputAssignment{
			{Field: "Partner_Name__c", Value: "myVariable_current.Order__r.Partner_Name__c"},
		},
		Connector: &models.FlowConnector{TargetReference: "Done"},
	}

	cap := &captureUpdate{}
	exec := minimalPersistenceWithCapture(cap)

	nextElem, err := exec.executeRecordUpdate(context.Background(), interview, flowMeta, update, nil)
	if err != nil {
		t.Fatalf("executeRecordUpdate error = %v", err)
	}
	if nextElem != "Done" {
		t.Fatalf("nextElem = %q, want %q", nextElem, "Done")
	}
	if len(cap.calls) != 1 {
		t.Fatalf("persistence.Update called %d times, want 1", len(cap.calls))
	}
	if got := cap.calls[0].updates["Partner_Name__c"]; got != "partner-001" {
		t.Fatalf("Partner_Name__c update = %#v (%T), want %q", got, got, "partner-001")
	}
}

// ===========================================================================
// SECTION G — #641: IsChanged + multiPicklist set-equality normalization.
//
// evaluateRecordFilterForFieldCtx must compare multiPicklist fields by token
// set, not by byte representation, so "A;B" == ["A","B"] and "B;A" == "A;B".
// ===========================================================================

// multiPicklistFieldMeta returns a minimal FieldMetadata whose Type is
// FieldTypeMultiPicklist. Used by the #641 IsChanged regression tests.
func multiPicklistFieldMeta(apiName string) *models.FieldMetadata {
	return &models.FieldMetadata{
		APIName: apiName,
		Type:    constants.FieldTypeMultiPicklist,
	}
}

// TestIsChanged_MultiPicklist_TokenSetChanged verifies that IsChanged returns
// true when the token SET differs (A,B → A,B,C).
func TestIsChanged_MultiPicklist_TokenSetChanged(t *testing.T) {
	exec := &FlowExecutor{}
	record := models.SObject{"Tags": `["A","B","C"]`}
	oldRecord := models.SObject{"Tags": `["A","B"]`}
	filter := &models.FlowRecordFilter{
		Field:    "Tags",
		Operator: "IsChanged",
	}
	fieldMeta := multiPicklistFieldMeta("Tags")
	got := exec.evaluateRecordFilterForFieldCtx(context.Background(), record, oldRecord, filter, fieldMeta)
	if !got {
		t.Error("IsChanged multiPicklist: A,B → A,B,C should be IsChanged=true, got false")
	}
}

// TestIsChanged_MultiPicklist_SameTokensDifferentFormat verifies that IsChanged
// returns false when the token SET is identical despite format differences
// (JSON array vs semicolon-joined scalar).
func TestIsChanged_MultiPicklist_SameTokensDifferentFormat(t *testing.T) {
	exec := &FlowExecutor{}
	// New value: JSON-array form (storage shape from TiDB)
	record := models.SObject{"Tags": `["A","B"]`}
	// Old value: semicolon-joined scalar (wire-format from FromStorageRecord)
	oldRecord := models.SObject{"Tags": "A;B"}
	filter := &models.FlowRecordFilter{
		Field:    "Tags",
		Operator: "IsChanged",
	}
	fieldMeta := multiPicklistFieldMeta("Tags")
	got := exec.evaluateRecordFilterForFieldCtx(context.Background(), record, oldRecord, filter, fieldMeta)
	if got {
		t.Error("IsChanged multiPicklist: [\"A\",\"B\"] vs \"A;B\" should be IsChanged=false (same token set), got true")
	}
}

// TestIsChanged_MultiPicklist_OrderOnlyDifference verifies that IsChanged
// returns false when the token SET is identical but the order in the
// serialization differs ("B;A" vs "A;B").
func TestIsChanged_MultiPicklist_OrderOnlyDifference(t *testing.T) {
	exec := &FlowExecutor{}
	record := models.SObject{"Tags": "A;B"}
	oldRecord := models.SObject{"Tags": "B;A"}
	filter := &models.FlowRecordFilter{
		Field:    "Tags",
		Operator: "IsChanged",
	}
	fieldMeta := multiPicklistFieldMeta("Tags")
	got := exec.evaluateRecordFilterForFieldCtx(context.Background(), record, oldRecord, filter, fieldMeta)
	if got {
		t.Error("IsChanged multiPicklist: \"A;B\" vs \"B;A\" should be IsChanged=false (order-only difference), got true")
	}
}

func TestEscapeSQLValue_LargeFloatRendersWithoutScientificNotation(t *testing.T) {
	// A provider object id delivered as a JSON number decodes to float64.
	// Before the fix, fmt.Sprintf("%v", ...) rendered it in scientific
	// notation, so a RecordLookup filter `id_field = payload.objectId` never
	// matched the stored integer-string id.
	cases := []struct {
		name string
		in   float64
		want string
	}{
		{"large integer id", 55489476744, "55489476744"},
		{"another large id", 60971395360, "60971395360"},
		{"small whole number", 100, "100"},
		{"decimal preserved", 1.5, "1.5"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			if got := escapeSQLValue(tc.in); got != tc.want {
				t.Errorf("escapeSQLValue(%v) = %q, want %q", tc.in, got, tc.want)
			}
			// formatFilterLiteral wraps the escaped value in single quotes for
			// the expr-parser filter expression.
			if got := formatFilterLiteral(tc.in); got != "'"+tc.want+"'" {
				t.Errorf("formatFilterLiteral(%v) = %q, want %q", tc.in, got, "'"+tc.want+"'")
			}
		})
	}
}