package automation

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/nexuscrm/backend/pkg/formula"
	"github.com/nexuscrm/backend/pkg/utils"
	"github.com/nexuscrm/shared/pkg/api"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/salesforce"
	"github.com/rs/zerolog/log"
)

const flowSourceStandard = "standard"

func (e *FlowExecutor) getActiveFlowVersion(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error) {
	if e.flowVersionFn != nil {
		return e.flowVersionFn(ctx, apiName, user)
	}
	result, err := e.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlowDefinition,
		Criteria:      []api.QueryCriterion{{Field: "DeveloperName", Op: "=", Val: apiName}},
		Limit:         1,
	}, user)
	if err != nil {
		return nil, err
	}
	if result.TotalSize == 0 || len(result.Records) == 0 {
		return nil, nil
	}

	activeVersionId, _ := result.Records[0]["ActiveVersionId"].(string)
	if activeVersionId == "" {
		return nil, nil
	}
	definitionIsCustomized := flowRecordBool(result.Records[0]["IsCustomized"])

	version, err := e.getFlowVersionById(ctx, activeVersionId, user)
	if version != nil {
		version.DefinitionIsCustomized = definitionIsCustomized
	}
	return version, err
}

func (e *FlowExecutor) getFlowVersionById(ctx context.Context, versionId string, user *models.UserSession) (*models.FlowVersion, error) {
	result, err := e.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlow,
		Criteria:      []api.QueryCriterion{{Field: constants.FieldID, Op: "=", Val: versionId}},
		Limit:         1,
	}, user)
	if err != nil {
		return nil, err
	}
	if result.TotalSize == 0 || len(result.Records) == 0 {
		return nil, nil
	}

	record := result.Records[0]
	version := &models.FlowVersion{
		Id: record[constants.FieldID].(string),
	}
	if status, ok := record["Status"].(string); ok {
		version.Status = status
	}
	if defId, ok := record["DefinitionId"].(string); ok {
		version.DefinitionId = defId
	}
	if source, ok := record["Source"].(string); ok {
		version.Source = source
	}
	if metaJSON, ok := record["Metadata"].(string); ok {
		version.MetadataJSON = metaJSON
	}
	version.HydrateMetadata()
	return version, nil
}

func (e *FlowExecutor) standardActionScopedChildContext(ctx context.Context, apiName string, user *models.UserSession) (context.Context, error) {
	hasInternalScopes := constants.HasStandardActionInternalAccessScopes(ctx)
	hasProtectedFlowMarker := constants.HasStandardActionProtectedFlowExecution(ctx)
	if !hasInternalScopes && !hasProtectedFlowMarker {
		return ctx, nil
	}
	packOwned, err := e.activeFlowIsPackOwned(ctx, apiName, user)
	if err != nil {
		return nil, err
	}
	if packOwned {
		return ctx, nil
	}
	if hasInternalScopes {
		ctx = constants.WithoutStandardActionInternalAccessScopes(ctx)
	}
	if hasProtectedFlowMarker {
		ctx = constants.WithoutStandardActionProtectedFlowExecution(ctx)
	}
	return ctx, nil
}

func (e *FlowExecutor) activeFlowIsPackOwned(ctx context.Context, apiName string, user *models.UserSession) (bool, error) {
	if e.flowPackOwnedFn != nil {
		return e.flowPackOwnedFn(ctx, apiName, user)
	}
	if e.flowVersionFn != nil {
		return true, nil
	}
	if e.query == nil || strings.TrimSpace(apiName) == "" {
		return false, nil
	}

	defResult, err := e.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlowDefinition,
		Criteria:      []api.QueryCriterion{{Field: "DeveloperName", Op: "=", Val: apiName}},
		Limit:         1,
	}, user)
	if err != nil {
		return false, err
	}
	if defResult == nil || defResult.TotalSize == 0 || len(defResult.Records) == 0 {
		return false, nil
	}
	definition := defResult.Records[0]
	if flowRecordBool(definition["IsCustomized"]) {
		return false, nil
	}
	activeVersionID, _ := definition["ActiveVersionId"].(string)
	if strings.TrimSpace(activeVersionID) == "" {
		return false, nil
	}

	flowResult, err := e.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: constants.TableFlow,
		Criteria:      []api.QueryCriterion{{Field: constants.FieldID, Op: "=", Val: activeVersionID}},
		Limit:         1,
	}, user)
	if err != nil {
		return false, err
	}
	if flowResult == nil || flowResult.TotalSize == 0 || len(flowResult.Records) == 0 {
		return false, nil
	}
	source, _ := flowResult.Records[0]["Source"].(string)
	return source == flowSourceStandard, nil
}

func flowRecordBool(value interface{}) bool {
	switch v := value.(type) {
	case bool:
		return v
	case int:
		return v != 0
	case int64:
		return v != 0
	case float64:
		return v != 0
	case string:
		return strings.EqualFold(strings.TrimSpace(v), "true") || strings.TrimSpace(v) == "1"
	default:
		return false
	}
}

func initializeFlowContext(flowMeta *models.FlowMetadata, inputs map[string]interface{}) map[string]interface{} {
	ctx := make(map[string]interface{})

	// Initialize variables with defaults
	for _, variable := range flowMeta.Variables {
		if variable.Value != nil {
			ctx[variable.Name] = variable.Value
		}
	}

	// Apply input values
	for key, value := range inputs {
		ctx[key] = value
	}

	return ctx
}

// evaluateFlowFormulas evaluates every FlowFormula resource declared on the
// flow metadata and writes the result into interview.ContextData keyed by the
// formula's Name. This is called once at interview initialization and again
// on every ResumeInterview so that TODAY()/NOW()-derived formulas refresh
// after a Wait — see design doc GAP 3 / Phase 2.3.
//
// Error policy:
//   - A nil interview, nil ContextData, nil flowMeta, or empty Formulas slice
//     is a no-op (the caller need not guard).
//   - Per-formula evaluation failures (syntax errors, unsupported operations,
//     circular references raised by the Engine) are logged at warn level and
//     skipped. Other formulas continue to be evaluated; we never abort the
//     whole batch because a single bad formula could otherwise block every
//     flow that references any formula.
//   - The logger is optional; a nil logger silently skips the warn.
//
// Context mapping follows the convention already used by action_service and
// approval_service:
//   - interview.ContextData["$Record"]       → formula.Context.Record
//   - interview.ContextData["$Record__Old"]  → formula.Context.Prior
//   - all other top-level entries            → formula.Context.Variables
//
// The "$Record" / "$Record__Old" keys are stripped from Variables because
// they already live under Record/Prior and the engine's convenience
// flattening would otherwise double-inject them.
func (e *FlowExecutor) evaluateFlowFormulas(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, user *models.UserSession) {
	if interview == nil || flowMeta == nil || len(flowMeta.Formulas) == 0 {
		return
	}
	if interview.ContextData == nil {
		interview.ContextData = make(map[string]interface{})
	}
	engine := e.formulaEngine
	if engine == nil {
		// Fallback for tests that construct FlowExecutor by hand. Paying the
		// one-time init cost is cheap and avoids forcing every unit test to
		// wire the engine.
		engine = formula.NewEngine()
	}
	fCtx := buildFlowFormulaContext(interview.ContextData, e.buildFormulaFetcher(ctx, interview, user))
	// Inject $Profile, $User, $UserRole global vars so flow formulas like
	// CONTAINS($Profile.Name, "System Administrator") resolve correctly (#2259).
	fCtx.User = buildFlowUserFormulaScope(user)
	for _, f := range flowMeta.Formulas {
		if f.Name == "" {
			continue
		}
		result, err := engine.Evaluate(f.Expression, fCtx)
		if err != nil {
			if e.logger != nil {
				e.logger.Warn(fmt.Sprintf(
					"flow: formula %q evaluation failed, skipping: %s",
					f.Name, err.Error(),
				))
			}
			continue
		}
		interview.ContextData[f.Name] = result
		// Re-expose on the formula context so later formulas in the list can
		// reference earlier results.
		fCtx.Variables[f.Name] = result
	}
}

// buildFormulaFetcher returns a closure satisfying formula.Context.Fetcher
// that lazy-loads a named relationship against the live Flow runtime
// services. Returns nil when the executor has no query service wired (in
// which case the formula engine falls back to inline-only behaviour, matching
// pre-fix semantics for tests that don't need cross-object resolution).
//
// Routes the actual relation-name → FK column → fetch sequence through the
// shared formula.FetchRelatedRecord helper so this path agrees with
// records.buildRecordFetcher byte-for-byte on the 3-value contract per
// CLAUDE.md R10. The per-interview rCache is layered AROUND the call so
// repeated formula references to the same parent within one interview hit
// the cache (preserving the pre-existing N+1-suppression semantics).
//
// PR #2057 IANTHEREAL blocker 4: the closure now reads the full ObjectMetadata
// via the metadata service so Tier 1 (RelationshipName match) is reachable
// here. Pre-fix, automation passed nil schema to ResolveRelationFK and the
// helper silently fell through to Tier 2/3/4, while records used the schema
// and got Tier 1. An admin who renamed a relationship via RelationshipName
// would see records and automation resolve the same fact differently — the
// canonical R10 split-path defect. Threading schema through cuts the path.
//
// schema may be nil for test executors without a metadata service, or for
// transient cache misses. Tier 2/3/4 (with the polymorphic gate at Tier 4)
// remain the fallback in that case; behavior is identical to pre-fix when
// schema is unavailable.
func (e *FlowExecutor) buildFormulaFetcher(
	ctx context.Context,
	interview *models.FlowInterview,
	user *models.UserSession,
) func(record map[string]interface{}, relationName string) (map[string]interface{}, error) {
	if e == nil || e.query == nil || interview == nil {
		return nil
	}
	return func(record map[string]interface{}, relationName string) (map[string]interface{}, error) {
		objectType := rootObjectType(interview.ContextData, "$Record", record)

		// Read the full schema once per fetch so Tier 1 (RelationshipName)
		// resolution becomes reachable. nil-tolerant: tests without a
		// metadata service drop straight back to Tier 2/3/4 behavior.
		schema := e.lookupObjectSchema(ctx, objectType)

		// Resolve FK upfront so the cache key has both a target and an FK
		// value to discriminate on. Tier 1 fires from RelationshipName;
		// Tier 4 (<segment>Id) keeps automation's polymorphic gate so an
		// FK with multiple ReferenceTo targets is not silently traversed.
		fkColumn := e.resolveRelationshipFKWithSchema(schema, objectType, relationName)
		if fkColumn == "" {
			return nil, nil // unknown relationship — silent skip per existing contract.
		}

		// Cache lookup mirrors loadRelatedRecord's keying:
		// "{relatedType}:{fkID}". Resolve target eagerly via the same
		// adapter so the cache and helper agree on the type. We do this
		// inline (not by routing through FetchRelatedRecord) so cache hits
		// avoid even constructing the deps wrapper.
		fkVal, ok := record[fkColumn]
		if !ok || fkVal == nil {
			return nil, nil // FK absent / cleared — definitive null.
		}
		fkID := fmt.Sprintf("%v", fkVal)
		if fkID == "" {
			return nil, nil
		}

		deps := &automationFormulaFetchDeps{e: e, user: user, currentObjectType: objectType}
		// Resolve target via the SAME helper FetchRelatedRecord uses so the
		// cache key matches the shared helper's notion of target — including
		// the polymorphic <FK>_type discriminator. Picking refTo[0] would
		// cache/negative-cache polymorphic Task.WhoId/WhatId rows under the
		// first declared ReferenceTo entry while the helper resolves to the
		// discriminator-selected target, producing a stale cache hit on a
		// later evaluation (R10 split that codex Stage-2 round-9 flagged).
		refTo := deps.LookupReferenceTo(objectType, fkColumn)
		relatedType, targetResolved := formula.ResolveRelationTarget(record, fkColumn, refTo)
		if !targetResolved {
			// Indeterminate target (no ReferenceTo, polymorphic without
			// discriminator, or discriminator outside ReferenceTo). Skip
			// cache and let the shared helper surface the indeterminate via
			// resolved=false → flow.r10:WRN + sentinel error, matching the
			// no-cache behaviour records/ already has for this case.
			related, target, resolved := formula.FetchRelatedRecord(ctx, deps, schema, objectType, record, relationName)
			if !resolved {
				if e.logger != nil {
					e.logger.Warn(fmt.Sprintf(
						"flow.r10: formula fetcher skipping `%s` — FetchRelatedRecord returned indeterminate (target=%q, metadata-miss or query error); formula result will be nil",
						relationName, target,
					))
				}
				return nil, fmt.Errorf("flow.r10: %s relationship indeterminate (metadata-miss or transient error)", relationName)
			}
			// Helper returned definitive (related, target) — but without a
			// cacheable target we cannot key it; return without caching.
			return related, nil
		}
		cacheKey := relatedType + ":" + fkID
		cache := rCache(interview)
		if cache != nil {
			if cached, ok := cache[cacheKey]; ok {
				if cached == nil {
					// Negative-cache hit — established as definitive null in
					// a prior call within this interview.
					return nil, nil
				}
				return map[string]interface{}(cached), nil
			}
		}

		// Cache miss → call the shared helper. Schema threaded through so
		// Tier 1 RelationshipName matches uniformly with the records path.
		related, target, resolved := formula.FetchRelatedRecord(ctx, deps, schema, objectType, record, relationName)
		if !resolved {
			// R10: indeterminate (metadata-miss / transient error). Return a
			// sentinel error so the formula evaluator surfaces the miss
			// rather than treating it as a definitive null parent.
			if e.logger != nil {
				e.logger.Warn(fmt.Sprintf(
					"flow.r10: formula fetcher skipping `%s` — FetchRelatedRecord returned indeterminate (target=%q, metadata-miss or query error); formula result will be nil",
					relationName, target,
				))
			}
			return nil, fmt.Errorf("flow.r10: %s relationship indeterminate (metadata-miss or transient error)", relationName)
		}
		if related == nil {
			// Definitive null (FK cleared or orphan). Negative-cache so a
			// repeat lookup within the same interview short-circuits.
			if cache != nil {
				cache[cacheKey] = nil
			}
			return nil, nil
		}
		// Stamp attributes.type so downstream reuse can recover the type
		// without another metadata roundtrip — same as loadRelatedRecord.
		relSObj := models.SObject(related)
		stampAttributesType(relSObj, target)
		if cache != nil {
			cache[cacheKey] = relSObj
		}
		return related, nil
	}
}

// automationFormulaFetchDeps adapts FlowExecutor's per-field metadata helper
// and QueryService to formula.RelationFetchDeps. Keeps the executor's RLS-
// aware QueryService scoped to the interview user — the access-control
// boundary records'/automation's adapters split on per CLAUDE.md R10.
type automationFormulaFetchDeps struct {
	e                 *FlowExecutor
	user              *models.UserSession
	currentObjectType string
}

func (d *automationFormulaFetchDeps) LookupReferenceTo(_ string, fkColumn string) []string {
	if d.e != nil && d.currentObjectType != "" {
		if fm := d.e.lookupFieldMeta(d.currentObjectType, fkColumn); fm != nil && len(fm.ReferenceTo) > 0 {
			return fm.ReferenceTo
		}
	}
	// Fallback to the standard FK→target table (SSoT in pkg/formula) so a
	// metadata-cache miss on a standard FK (RecordTypeId, AccountId, …)
	// still resolves a target. Matches loadRelatedRecord's fallback.
	if t, ok := formula.LookupStandardRelationTarget(fkColumn); ok {
		return []string{t}
	}
	return nil
}

func (d *automationFormulaFetchDeps) FetchByID(ctx context.Context, target, id string) (map[string]interface{}, error) {
	// Missing query service is INDETERMINATE, not orphan. Returning (nil, nil)
	// here would let formula.FetchRelatedRecord classify it as definitive null
	// and the formula engine would compute against a null parent — silently
	// wrong vs. loadRelatedRecord which correctly returns resolved=false on
	// e.query == nil. Round-4 codex Pass A HIGH #2.
	if d.e == nil || d.e.query == nil {
		return nil, fmt.Errorf("automation.formulaFetch: query service unavailable for %s/%s", target, id)
	}
	result, err := d.e.query.QueryWithFilter(ctx, target, "Id = '"+escapeSQLValue(id)+"'", d.user, "", "", 1)
	if err != nil {
		return nil, err
	}
	if result == nil || len(result.Records) == 0 {
		return nil, nil
	}
	return map[string]interface{}(result.Records[0]), nil
}


// buildFlowUserFormulaScope constructs the map[string]interface{} that gets
// assigned to formula.Context.User when evaluating FlowFormula resources
// (formula.Engine.Evaluate) and injected into interview.ContextData for
// Decision-element condition checks (resolveNestedReferenceCtx).
//
// Delegates to formula.BuildUserFormulaScope — the single SSoT for the
// $Profile / $User / $UserRole global-variable shapes — so that this path
// and the validation-rule path in records.PersistenceService always produce
// identical shapes from the same session fields. This satisfies CLAUDE.md R10
// Two-Path Consistency.
func buildFlowUserFormulaScope(user *models.UserSession) map[string]interface{} {
	return formula.BuildUserFormulaScope(user)
}

// injectUserGlobalsIntoContext populates $Profile, $User, and $UserRole into
// interview.ContextData so Decision conditions that reference e.g.
// $Profile.Name via resolveNestedReferenceCtx find the value without a DB
// round-trip.
//
// These three keys are transient: they MUST appear in transientContextKeys so
// marshalPersistableContext strips them before every DB persist. They are
// re-injected on every executor entry (executeFlowVersion) and on every resume
// (ResumeInterview / executeFlowMetadata), ensuring the values always reflect
// the live UserSession rather than a stale serialised snapshot.
//
// The map shapes are identical to buildFlowUserFormulaScope's __Profile /
// __User / __UserRole entries — same field names, same source fields — so
// $Profile.Name resolved via resolveNestedReferenceCtx (Decision conditions)
// and $Profile.Name resolved via formula.Engine.Evaluate (FlowFormula
// resources) always agree. This satisfies CLAUDE.md R10 Two-Path Consistency.
//
// When user is nil the function is a no-op (scheduled flows that have no
// session may reach this path during unit tests).
func injectUserGlobalsIntoContext(contextData map[string]interface{}, user *models.UserSession) {
	if contextData == nil || user == nil {
		return
	}
	// Derive all three sub-entity maps from the single SSoT so the shapes are
	// always identical to what BuildUserFormulaScope produces. Using
	// BuildUserEntityMaps here (instead of building the maps inline) satisfies
	// CLAUDE.md R10 Two-Path Consistency: exactly one place decides what fields
	// go into the Profile / User / UserRole entities.
	m := formula.BuildUserEntityMaps(user)
	contextData["$Profile"]  = m.Profile
	contextData["$User"]     = m.User
	contextData["$UserRole"] = m.Role
}

// buildFlowFormulaContext constructs a formula.Context from an interview's
// ContextData. Extracted so both evaluateFlowFormulas and its tests can agree
// on the same mapping rules.
//
// Runtime sentinel keys (__r_cache, _objectType, _loopDepth) are filtered
// out of the Variables scope because (a) they carry non-value Go types
// (map[string]SObject, etc) that formula expressions have no business
// touching, and (b) their underscore-prefixed names could otherwise collide
// with engine-internal identifiers. Keeping them out makes the formula
// evaluation surface strictly what a flow author would see in the
// Salesforce flow canvas.
func buildFlowFormulaContext(contextData map[string]interface{}, fetcher func(record map[string]interface{}, relationName string) (map[string]interface{}, error)) *formula.Context {
	fCtx := &formula.Context{
		Variables: make(map[string]interface{}, len(contextData)),
		Fetcher:   fetcher,
	}
	if record, ok := contextData["$Record"].(models.SObject); ok {
		fCtx.Record = record
	} else if record, ok := contextData["$Record"].(map[string]interface{}); ok {
		fCtx.Record = record
	}
	// Support both $Record__Old and $Record__Prior as aliases for the same data
	if prior, ok := contextData["$Record__Old"].(models.SObject); ok {
		fCtx.Prior = prior
	} else if prior, ok := contextData["$Record__Prior"].(models.SObject); ok {
		fCtx.Prior = prior
	} else if prior, ok := contextData["$Record__Old"].(map[string]interface{}); ok {
		fCtx.Prior = prior
	} else if prior, ok := contextData["$Record__Prior"].(map[string]interface{}); ok {
		fCtx.Prior = prior
	}
	for k, v := range contextData {
		switch k {
		case "$Record", "$Record__Old", "$Record__Prior":
			continue
		case flowContextKeyRCache, flowContextKeyRootObject, flowContextKeyInLoopDepth:
			continue
		}
		fCtx.Variables[k] = v
	}
	// normalizeFormulaExpression rewrites $Record → __Record ($ is invalid in
	// expr-lang identifiers). Inject Record/Prior under both the formula
	// engine's native keys AND the normalized keys so expressions like
	// $Record.AccountId (→ __Record.AccountId) resolve correctly.
	if fCtx.Record != nil {
		fCtx.Variables["__Record"] = fCtx.Record
	}
	if fCtx.Prior != nil {
		fCtx.Variables["__Record__Old"] = fCtx.Prior
		// Also expose under __Record__Prior for bidirectional compatibility
		fCtx.Variables["__Record__Prior"] = fCtx.Prior
	}
	return fCtx
}

// checkEntryConditions evaluates the Start element's filter conditions
// Returns true if the record meets the entry criteria or if there are no filters
func (e *FlowExecutor) checkEntryConditions(flowMeta *models.FlowMetadata, context map[string]interface{}) bool {
	return e.checkEntryConditionsCtx(nil, flowMeta, context)
}

// checkEntryConditionsCtx is the ctx-aware variant of checkEntryConditions.
// Needed so the Start filter evaluator can reach the SF key-prefix resolver
// (sfKeyPrefixMatch), which performs a DB lookup and therefore needs a live
// request context. Existing non-Ctx callers go through checkEntryConditions
// above and forgo prefix resolution — matching pre-fix semantics for any
// caller that does not have a ctx in hand (notably unit-test shims).
func (e *FlowExecutor) checkEntryConditionsCtx(ctx context.Context, flowMeta *models.FlowMetadata, context_ map[string]interface{}) bool {
	if flowMeta.Start == nil || len(flowMeta.Start.Filters) == 0 {
		// No filters - always execute
		return true
	}

	// Get the record from context (for record-triggered flows)
	record, ok := context_["$Record"].(models.SObject)
	if !ok {
		// No record in context - could be an AutoLaunched flow, allow execution
		return true
	}

	// Get the old record for IsChanged comparisons
	oldRecord, _ := context_["$Record__Old"].(models.SObject)

	// Evaluate each filter against the record
	filters := flowMeta.Start.Filters
	logic := strings.ToLower(flowMeta.Start.FilterLogic)

	// Resolve the triggering object once so per-filter evaluators can apply
	// field-type-specific semantics (e.g. multi-picklist IsChanged must
	// compare as a sorted set — #1683 / #1684).
	objectAPIName := ""
	if flowMeta.Start != nil {
		objectAPIName = flowMeta.Start.Object
	}

	results := make([]bool, len(filters))
	for i, filter := range filters {
		var fieldMeta *models.FieldMetadata
		if objectAPIName != "" && filter.Field != "" {
			fieldMeta = e.lookupFieldMeta(objectAPIName, filter.Field)
		}
		results[i] = e.evaluateRecordFilterForFieldCtx(ctx, record, oldRecord, &filter, fieldMeta)
	}

	return evaluateFilterLogic(logic, results)
}

// evaluateFilterLogic evaluates a Salesforce-style filter logic expression
// like "1 AND (2 OR 3)" using 1-indexed references to filter results.
//
// SSoT collapse (R10): the boolean-expression grammar shared with Decision
// rule conditionLogic is owned by evaluateConditionLogic. This wrapper
// preserves the historical "empty filter list = true" base case (filter
// composition runs over zero filters defaults to true) and delegates the
// rest. One parser, one fail-closed policy on malformed / out-of-range
// metadata, one place to change SF parity going forward.
func evaluateFilterLogic(logic string, results []bool) bool {
	if len(results) == 0 {
		return true
	}
	return evaluateConditionLogic(logic, results)
}

// tokenizeFilterLogic splits a filter logic string into tokens
func tokenizeFilterLogic(logic string) []string {
	var tokens []string
	i := 0
	for i < len(logic) {
		ch := logic[i]
		if ch == ' ' || ch == '\t' {
			i++
			continue
		}
		if ch == '(' || ch == ')' {
			tokens = append(tokens, string(ch))
			i++
			continue
		}
		// Read a word or number
		start := i
		for i < len(logic) && logic[i] != ' ' && logic[i] != '(' && logic[i] != ')' && logic[i] != '\t' {
			i++
		}
		tokens = append(tokens, strings.ToUpper(logic[start:i]))
	}
	return tokens
}

// evaluateRecordFilter checks if a record meets a single filter condition.
// Prefer the metadata-aware variant `evaluateRecordFilterForField` in
// production paths where the flow's triggering object is known; this
// signature is retained for existing test call sites.
func (e *FlowExecutor) evaluateRecordFilter(record models.SObject, oldRecord models.SObject, filter *models.FlowRecordFilter) bool {
	return e.evaluateRecordFilterForField(record, oldRecord, filter, nil)
}

// evaluateRecordFilterForField is the metadata-aware single-filter evaluator.
// `fieldMeta` is used to apply field-type-specific comparison semantics — e.g.
// multi-picklist IsChanged must compare as a sorted set to tolerate
// serialization asymmetry (JSON-array in storage vs semicolon-scalar from the
// wire-format join, slice inputs from flow wiring). Pass nil when the field
// is unknown; the evaluator then uses the generic stringified comparison and
// preserves legacy behavior.
func (e *FlowExecutor) evaluateRecordFilterForField(record models.SObject, oldRecord models.SObject, filter *models.FlowRecordFilter, fieldMeta *models.FieldMetadata) bool {
	return e.evaluateRecordFilterForFieldCtx(nil, record, oldRecord, filter, fieldMeta)
}

// evaluateRecordFilterForFieldCtx is the ctx-aware variant. Callers that have
// a request context (currently checkEntryConditionsCtx from executeFlowVersion
// / buildScheduledRunInputs) route StartsWith / DoesNotStartWith through
// sfKeyPrefixMatch so Salesforce-idiomatic key-prefix checks on polymorphic
// reference fields (e.g. Start filter `WhatId StartsWith "006"` on
// Task_Hellowin_Feedback_Is_Resuired) resolve against UUID-style IDs. Passing
// a nil ctx disables prefix resolution and preserves legacy semantics — that
// is how the non-Ctx evaluateRecordFilter[ForField] keeps behaving the same
// for any caller that still goes through the legacy entry points.
func (e *FlowExecutor) evaluateRecordFilterForFieldCtx(ctx context.Context, record models.SObject, oldRecord models.SObject, filter *models.FlowRecordFilter, fieldMeta *models.FieldMetadata) bool {
	fieldValue := record[filter.Field]
	filterValue, _ := extractTypedValue(filter.Value)

	switch normalizeFlowOperator(filter.Operator) {
	case "EqualTo":
		return fmt.Sprintf("%v", fieldValue) == fmt.Sprintf("%v", filterValue)
	case "NotEqualTo":
		return fmt.Sprintf("%v", fieldValue) != fmt.Sprintf("%v", filterValue)
	case "GreaterThan":
		return isGreater(fieldValue, filterValue)
	case "GreaterThanOrEqualTo":
		return isGreaterOrEqual(fieldValue, filterValue)
	case "LessThan":
		return isLess(fieldValue, filterValue)
	case "LessThanOrEqualTo":
		return isLessOrEqual(fieldValue, filterValue)
	case "Contains":
		return containsString(fieldValue, filterValue)
	case "NotContains":
		return !containsString(fieldValue, filterValue)
	case "IsNull":
		isNull := fieldValue == nil || fieldValue == ""
		if bv, ok := filterValue.(bool); ok && !bv {
			return !isNull
		}
		return isNull
	case "IsNotNull":
		isNull := fieldValue == nil || fieldValue == ""
		if bv, ok := filterValue.(bool); ok && bv {
			return !isNull
		}
		return !isNull
	case "IsBlank":
		// Salesforce ISBLANK: nil, empty/whitespace string, empty collection.
		// Missing map keys read as nil, so this also handles "field not set".
		// filterValue=false means "IS NOT blank" (same pattern as IsNull).
		blank := isBlankValue(fieldValue)
		if bv, ok := filterValue.(bool); ok && !bv {
			return !blank
		}
		return blank
	case "IsNotBlank":
		return !isBlankValue(fieldValue)
	case "IsChanged":
		if oldRecord == nil {
			return true // No old record (new record), consider field as changed
		}
		oldValue := oldRecord[filter.Field]
		// Multi-picklist storage (TiDB JSON) and the wire-format read path
		// (semicolon-joined scalar via FromStorageRecord) produce different
		// byte representations for the same selected-set. Gate on metadata
		// so a strict stringified comparison is still used for JSON fields
		// (where array order matters) and plain String fields (where a `;`
		// may legitimately be part of the value). Closes the flow-layer
		// residual of #1683; mirror of the records-layer PR #1684 fix.
		if fieldMeta != nil && fieldMeta.Type == constants.FieldTypeMultiPicklist {
			return !multiPicklistSetsEqual(fieldValue, oldValue)
		}
		return fmt.Sprintf("%v", fieldValue) != fmt.Sprintf("%v", oldValue)
	case "StartsWith":
		strFilter := fmt.Sprintf("%v", filterValue)
		if ctx != nil {
			if matched, handled := e.sfKeyPrefixMatch(ctx, fieldValue, strFilter); handled {
				return matched
			}
		}
		strVal := fmt.Sprintf("%v", fieldValue)
		return strings.HasPrefix(strVal, strFilter)
	case "DoesNotStartWith":
		strFilter := fmt.Sprintf("%v", filterValue)
		if ctx != nil {
			if matched, handled := e.sfKeyPrefixMatch(ctx, fieldValue, strFilter); handled {
				return !matched
			}
		}
		strVal := fmt.Sprintf("%v", fieldValue)
		return !strings.HasPrefix(strVal, strFilter)
	case "EndsWith":
		strVal := fmt.Sprintf("%v", fieldValue)
		strFilter := fmt.Sprintf("%v", filterValue)
		return strings.HasSuffix(strVal, strFilter)
	case "DoesNotEndWith":
		strVal := fmt.Sprintf("%v", fieldValue)
		strFilter := fmt.Sprintf("%v", filterValue)
		return !strings.HasSuffix(strVal, strFilter)
	case "Between":
		return isBetween(fieldValue, filterValue)
	case "NotBetween":
		return !isBetween(fieldValue, filterValue)
	case "In":
		return isIn(fieldValue, filterValue)
	case "NotIn":
		return !isIn(fieldValue, filterValue)
	case "Includes":
		return includes(fieldValue, filterValue)
	case "Excludes":
		return excludes(fieldValue, filterValue)
	case "EmptyString":
		return fmt.Sprintf("%v", fieldValue) == ""
	case "EntersPickTag":
		// True iff the field value just transitioned INTO the named picklist
		// tag category — i.e. the current value carries the tag AND the prior
		// value did not. The tag string itself is opaque to the engine; it
		// is matched against PicklistOption.Tags on the field's metadata.
		//
		// Create-time semantics: oldRecord is nil, so oldValue reads as nil,
		// optionHasTag(..., nil, ...) returns false, and the transition
		// condition reduces to "new value has the tag" — the right behavior
		// when a record is created directly at a tagged value.
		//
		// Stale or unknown values (value present on record but not in
		// Options) fail closed (false) — safer than guessing a tag set.
		tag, ok := filterValue.(string)
		if !ok || tag == "" {
			return false
		}
		var oldValue interface{}
		if oldRecord != nil {
			oldValue = oldRecord[filter.Field]
		}
		return optionHasTag(fieldMeta, fieldValue, tag) && !optionHasTag(fieldMeta, oldValue, tag)
	default:
		return true // Unknown operator, allow execution
	}
}

// PicklistOptionHasTag reports whether the PicklistOption identified by value (on
// fieldMeta's Options list) carries tag. Returns false for any of: nil
// fieldMeta, nil/empty value, value whose string form matches no option,
// or option with no matching tag. The stringified comparison on Option.Value
// mirrors the rest of the operator switch (which also stringifies) so
// numeric or typed-wrapper values behave consistently.
func PicklistOptionHasTag(fieldMeta *models.FieldMetadata, value interface{}, tag string) bool {
	if fieldMeta == nil || tag == "" {
		return false
	}
	if value == nil {
		return false
	}
	strVal := fmt.Sprintf("%v", value)
	if strVal == "" {
		return false
	}
	for _, opt := range fieldMeta.Options {
		if opt.Value != strVal {
			continue
		}
		for _, t := range opt.Tags {
			if t == tag {
				return true
			}
		}
		return false
	}
	return false
}

func optionHasTag(fieldMeta *models.FieldMetadata, value interface{}, tag string) bool {
	return PicklistOptionHasTag(fieldMeta, value, tag)
}

func (e *FlowExecutor) evaluateConditions(context map[string]interface{}, conditions []models.FlowCondition, logic string) bool {
	if len(conditions) == 0 {
		return true
	}

	results := make([]bool, len(conditions))
	for i, cond := range conditions {
		results[i] = e.evaluateCondition(context, &cond)
	}

	return evaluateConditionLogic(logic, results)
}

func (e *FlowExecutor) evaluateCondition(context map[string]interface{}, cond *models.FlowCondition) bool {
	leftValue := e.resolveValue(context, cond.LeftValueReference)
	rawRight := e.resolveValue(context, cond.RightValue)
	rightValue, _ := extractTypedValue(rawRight)

	if isAllNilRightValue(rawRight) {
		switch normalizeFlowOperator(cond.Operator) {
		case "EqualTo":
			return isTruthy(leftValue)
		case "NotEqualTo":
			return !isTruthy(leftValue)
		}
	}

	switch normalizeFlowOperator(cond.Operator) {
	case "EqualTo":
		return fmt.Sprintf("%v", leftValue) == fmt.Sprintf("%v", rightValue)
	case "NotEqualTo":
		return fmt.Sprintf("%v", leftValue) != fmt.Sprintf("%v", rightValue)
	case "GreaterThan":
		return isGreater(leftValue, rightValue)
	case "GreaterThanOrEqualTo":
		return isGreaterOrEqual(leftValue, rightValue)
	case "LessThan":
		return isLess(leftValue, rightValue)
	case "LessThanOrEqualTo":
		return isLessOrEqual(leftValue, rightValue)
	case "Contains":
		return containsString(leftValue, rightValue)
	case "NotContains":
		return !containsString(leftValue, rightValue)
	case "IsNull":
		isNull := leftValue == nil || leftValue == ""
		if bv, ok := rightValue.(bool); ok && !bv {
			return !isNull
		}
		return isNull
	case "IsNotNull":
		isNull := leftValue == nil || leftValue == ""
		if bv, ok := rightValue.(bool); ok && bv {
			return !isNull
		}
		return !isNull
	case "IsBlank":
		// Salesforce ISBLANK: nil, empty/whitespace string, empty collection.
		// Zero numerics are NOT blank — critical difference vs IsNull for
		// numeric fields. See isBlankValue().
		// rightValue=false means "IS NOT blank" (same pattern as IsNull).
		blank := isBlankValue(leftValue)
		if bv, ok := rightValue.(bool); ok && !bv {
			return !blank
		}
		return blank
	case "IsNotBlank":
		return !isBlankValue(leftValue)
	case "StartsWith":
		strVal := fmt.Sprintf("%v", leftValue)
		strFilter := fmt.Sprintf("%v", rightValue)
		return strings.HasPrefix(strVal, strFilter)
	case "DoesNotStartWith":
		strVal := fmt.Sprintf("%v", leftValue)
		strFilter := fmt.Sprintf("%v", rightValue)
		return !strings.HasPrefix(strVal, strFilter)
	case "EndsWith":
		strVal := fmt.Sprintf("%v", leftValue)
		strFilter := fmt.Sprintf("%v", rightValue)
		return strings.HasSuffix(strVal, strFilter)
	case "DoesNotEndWith":
		strVal := fmt.Sprintf("%v", leftValue)
		strFilter := fmt.Sprintf("%v", rightValue)
		return !strings.HasSuffix(strVal, strFilter)
	case "Between":
		return isBetween(leftValue, rightValue)
	case "NotBetween":
		return !isBetween(leftValue, rightValue)
	case "In":
		return isIn(leftValue, rightValue)
	case "NotIn":
		return !isIn(leftValue, rightValue)
	case "Includes":
		return includes(leftValue, rightValue)
	case "Excludes":
		return excludes(leftValue, rightValue)
	case "EmptyString":
		return fmt.Sprintf("%v", leftValue) == ""
	case "IsChanged":
		// Mirrors start-filter IsChanged semantics (evaluateRecordFilterForField):
		// compare the referenced field on $Record against the same field on
		// $Record__Old. When no prior record exists (CREATE) or the field is
		// missing on the prior, the field is treated as changed. References
		// outside the "$Record." root (e.g. "$Flow.*") return false, matching
		// the prior behavior of the unknown-operator default.
		priorRef, ok := deriveRecordOldRef(cond.LeftValueReference)
		if !ok {
			return false
		}
		priorValue := e.resolveValue(context, priorRef)
		return fmt.Sprintf("%v", leftValue) != fmt.Sprintf("%v", priorValue)
	default:
		return false
	}
}

func (e *FlowExecutor) resolveValue(context map[string]interface{}, value interface{}) interface{} {
	if value == nil {
		return nil
	}

	// Unwrap Salesforce-style typed value objects (e.g. {"elementReference": "..."})
	// before checking for string references. Flow metadata stores values as typed
	// wrappers when loaded from JSON with interface{} fields.
	var wasLiteral bool
	value, wasLiteral = extractTypedValue(value)

	// If it's a string reference, resolve it
	ref, ok := value.(string)
	if !ok {
		return value
	}

	// Handle $Flow system variables
	if strings.HasPrefix(ref, "$Flow.") {
		return e.resolveFlowSystemVariable(ref)
	}

	// Handle $GlobalConstant
	if strings.HasPrefix(ref, "$GlobalConstant.") {
		return e.resolveGlobalConstant(ref)
	}

	// Handle nested references like $Record.Account.Name or varName.FieldName.
	// Only route through resolveNestedReference when the string looks like an
	// identifier path: all characters must be letters, digits, underscores, $,
	// or dots. A literal string like "Acme Corp. Ltd." contains a space and
	// therefore cannot be a variable reference path — skip nested-reference
	// resolution and let it fall through to the literal passthrough below.
	// See issue #1817 Bug A and the same guard in resolveValueCtxWithPresence.
	//
	// Additionally, skip the heuristic entirely when wasLiteral=true: the value
	// arrived via a typed-value wrapper (stringValue/numberValue/…) which means
	// the flow author explicitly authored a literal — it is never a variable
	// reference regardless of its character content. This fixes issue #2046
	// where dotted identifier-shape literals like "v1.2.3" or "name.tld" were
	// misrouted through resolveNestedReference and silently dropped.
	if !wasLiteral && strings.Contains(ref, ".") && looksLikeVariableReference(ref) {
		return e.resolveNestedReference(context, ref)
	}

	// Simple variable lookup
	if resolved, exists := context[ref]; exists {
		return resolved
	}

	return value
}

// looksLikeVariableReference returns true when s could be a Salesforce Flow
// variable reference path (e.g. "$Record.FieldName", "myVar.SubField").
//
// Salesforce Flow variable and field names are identifiers: they consist only
// of letters, digits, and underscores — no spaces, no hyphens, no parens.
// A reference path like "$Record.Region__c" satisfies this invariant.
// A literal string value like "Acme Corp. Ltd." or "Acme Co. (US), Inc."
// contains spaces and/or parens, so it cannot be a variable reference path
// even though it contains a dot. This check lets resolveValue skip the
// nested-reference lookup for such literals and return them unchanged.
//
// Fallback for bare-string callers: wrapper-aware callers (resolveValue,
// resolveValueCtxWithPresence) use the wasLiteral flag from extractTypedValue
// and skip this heuristic entirely when the value arrived via a typed-value
// wrapper — so dotted identifier-shape literals like "v1.2.3" or "name.tld"
// are never misrouted regardless of their character content (issue #2046).
func looksLikeVariableReference(s string) bool {
	for _, r := range s {
		if r == '.' {
			continue
		}
		if r == '$' || r == '_' {
			continue
		}
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
			continue
		}
		return false
	}
	return true
}

// resolveFlowSystemVariable handles $Flow.* system variables
func (e *FlowExecutor) resolveFlowSystemVariable(ref string) interface{} {
	now := time.Now()

	switch ref {
	case "$Flow.CurrentDate":
		return now.Format("2006-01-02")
	case "$Flow.CurrentDateTime":
		return now.Format(time.RFC3339)
	case "$Flow.CurrentStage":
		return "" // Stage flows not yet supported
	case "$Flow.InterviewStartTime":
		return now.Format(time.RFC3339)
	case "$Flow.FaultMessage":
		return "" // Set during fault handling
	case "$Flow.ActiveStages":
		return []interface{}{}
	default:
		return nil
	}
}

// resolveGlobalConstant handles $GlobalConstant.* references
func (e *FlowExecutor) resolveGlobalConstant(ref string) interface{} {
	// Extract constant name
	parts := strings.SplitN(ref, ".", 2)
	if len(parts) < 2 {
		return nil
	}

	// Common global constants
	switch parts[1] {
	case "True":
		return true
	case "False":
		return false
	case "EmptyString":
		return ""
	default:
		return nil
	}
}

// resolveNestedReference handles dot-notation references like $Record.Account.Name.
//
// This is the pure in-memory variant: it walks map-like structures only. Segments
// ending in `__r` are NOT lazy-loaded here — this variant is used by scheduled
// filter builders and tests that have no ctx/user/interview. Runtime callers that
// need cross-object `__r` traversal must use resolveNestedReferenceCtx instead.
func (e *FlowExecutor) resolveNestedReference(context map[string]interface{}, ref string) interface{} {
	parts := strings.Split(ref, ".")
	if len(parts) == 0 {
		return nil
	}

	// Get the root object
	current := context[parts[0]]
	if current == nil {
		return nil
	}

	// Traverse the path
	for i := 1; i < len(parts); i++ {
		fieldName := parts[i]

		switch obj := current.(type) {
		case models.SObject:
			current = obj[fieldName]
		case map[string]interface{}:
			current = obj[fieldName]
		default:
			return nil
		}

		if current == nil {
			return nil
		}
	}

	return current
}

// Private ContextData keys used by the __r lazy-load mechanism. They are
// stripped before persisting the interview (see flow_interview_service.go)
// so they never leak into the DB row.
const (
	flowContextKeyRCache      = "__r_cache"   // map[string]models.SObject keyed by "{objectType}:{recordId}"
	flowContextKeyRootObject  = "_objectType" // string — root trigger object type for $Record
	flowContextKeyInLoopDepth = "_loopDepth"  // float64 — current Loop nesting depth for warn gating
)

// loopDepth reads the current Loop nesting depth from ContextData.
// Returns 0 when not inside any Loop.
func loopDepth(contextData map[string]interface{}) int {
	if contextData == nil {
		return 0
	}
	if depth, ok := contextData[flowContextKeyInLoopDepth].(float64); ok {
		return int(depth)
	}
	return 0
}

// setLoopDepth writes the loop depth back into ContextData, deleting the key
// when depth reaches 0 to keep serialized snapshots clean.
func setLoopDepth(contextData map[string]interface{}, depth int) {
	if depth <= 0 {
		delete(contextData, flowContextKeyInLoopDepth)
		return
	}
	contextData[flowContextKeyInLoopDepth] = float64(depth)
}

// rCache returns (or creates) the per-interview `__r` resolution cache.
// Cache keys are of the form "{objectType}:{recordId}".
func rCache(interview *models.FlowInterview) map[string]models.SObject {
	if interview == nil || interview.ContextData == nil {
		return nil
	}
	if existing, ok := interview.ContextData[flowContextKeyRCache].(map[string]models.SObject); ok {
		return existing
	}
	cache := make(map[string]models.SObject)
	interview.ContextData[flowContextKeyRCache] = cache
	return cache
}

// relationshipFKField translates a relationship segment to the matching
// foreign-key column name using Salesforce naming conventions:
//   - Custom: "Order__r" -> "Order__c" (the "__r → __c" rule)
//   - Standard: "Account" -> "AccountId" (via the SSoT standard-relationship
//     table in formula.LookupStandardRelationFK)
//
// Returns an empty string when the segment is neither a __r suffix nor a
// recognized standard relationship name. This is the metadata-free fast path
// retained for tests and for the managed-filter path that has no executor to
// consult. Runtime resolvers with access to ObjectMetadata should use
// FlowExecutor.resolveRelationshipFK instead, which falls back to the
// Salesforce "<Relation>Id" convention when the hardcoded table misses.
//
// The standard-relationship table itself lives in pkg/formula so records/
// and automation/ share a single source per CLAUDE.md R10.
func relationshipFKField(segment string) string {
	if segment == "" {
		return ""
	}
	// Custom relationship: Foo__r → Foo__c
	if strings.HasSuffix(segment, "__r") {
		return strings.TrimSuffix(segment, "__r") + "__c"
	}
	// Standard relationship table — single source via pkg/formula.
	if fk, ok := formula.LookupStandardRelationFK(segment); ok {
		return fk
	}
	return ""
}

// resolveRelationshipFK maps a relationship segment to its FK column, layering
// a metadata-driven lookup on top of the metadata-free relationshipFKField.
//
// Resolution order:
//  1. relationshipFKField's hardcoded rules (custom "__r" + standard map).
//  2. Salesforce "<Relation>Id" convention against the current object's
//     metadata: if `currentObjectType` has a Reference field named
//     `segment + "Id"`, return that field. This covers standard relationships
//     outside the hardcoded map (e.g. User.UserRoleId, User.ManagerId,
//     User.ProfileId) and keeps the engine admin-configurable — adding a new
//     reference field via metadata immediately makes it traversable without
//     a code change.
//
// Returns "" when neither path resolves.
func (e *FlowExecutor) resolveRelationshipFK(currentObjectType, segment string) string {
	if segment == "" {
		return ""
	}
	if fk := relationshipFKField(segment); fk != "" {
		return fk
	}
	if currentObjectType == "" {
		return ""
	}
	candidateFK := segment + "Id"
	// Gate on non-polymorphic targets only. loadRelatedRecord picks
	// ReferenceTo[0] unconditionally — widening the reachable set of
	// relationships to polymorphic ones (e.g. Task.WhoId → [Contact,Lead])
	// would let the resolver silently query the wrong object. Polymorphic
	// traversal is tracked separately; this fix keeps to single-target refs.
	if fm := e.lookupFieldMeta(currentObjectType, candidateFK); fm != nil && len(fm.ReferenceTo) == 1 {
		return candidateFK
	}
	return ""
}

// resolveRelationshipFKWithSchema augments resolveRelationshipFK with Tier 1
// (RelationshipName lookup on the full schema) so admin-renamed relationships
// resolve identically across records and automation formula evaluation.
//
// Background (PR #2057 IANTHEREAL blocker 4): records calls
// formula.ResolveRelationFK(schema, segment) which checks RelationshipName on
// every field. Automation historically skipped Tier 1 because the executor
// only had per-field metadata accessors, so an admin who renamed a
// relationship by setting RelationshipName='MyParent' on field 'CustomFK__c'
// would see records resolve "MyParent" → "CustomFK__c" while automation
// returned "" (definitive null). Same fact, two read paths — R10 violation.
//
// Fix: read the full schema via the metadata service once when the formula
// fetcher is constructed and pass it through here. Tier 1 fires when schema
// is available; Tier 2/3/4 (with the polymorphic gate) remain the fallback.
//
// schema may be nil (test executor without a metadata service, or a cache
// miss). When nil, behavior reduces to the original resolveRelationshipFK.
func (e *FlowExecutor) resolveRelationshipFKWithSchema(schema *models.ObjectMetadata, currentObjectType, segment string) string {
	if segment == "" {
		return ""
	}
	// Tier 1: RelationshipName match on the full schema (metadata-driven).
	if schema != nil {
		for i := range schema.Fields {
			f := &schema.Fields[i]
			if f.RelationshipName != nil && *f.RelationshipName == segment {
				return f.APIName
			}
		}
	}
	// Tier 2/3/4: existing convention-first resolver with polymorphism gate.
	return e.resolveRelationshipFK(currentObjectType, segment)
}

// resolveNestedReferenceCtx resolves a dot-notation reference with lazy-load
// support for `__r` cross-object segments (design GAP 6).
//
// Behavior:
//   - Non-`__r` segments behave exactly like resolveNestedReference.
//   - A `__r` segment triggers a one-shot DB lookup of the related record.
//   - Related records are cached per-interview under flowContextKeyRCache,
//     keyed by "{objectType}:{recordId}", to prevent repeated DB hits within
//     a single interview.
//   - When the FK is absent or the related record is missing, this returns nil
//     (no panic) and logs a warning.
//   - When `__r` is resolved while inside a Loop (loopDepth > 0), a warning is
//     emitted to flag the N+1 risk called out in the design doc.
//
// Known limitation (design doc): the cache is populated once per reference per
// interview. If a flow action writes to the related record, subsequent reads
// in the same interview return stale data. Acceptable for current flows.
func (e *FlowExecutor) resolveNestedReferenceCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, ref string) interface{} {
	if interview == nil {
		return e.resolveNestedReference(nil, ref)
	}
	parts := strings.Split(ref, ".")
	if len(parts) == 0 {
		return nil
	}

	contextData := interview.ContextData
	rootName := parts[0]
	current := contextData[rootName]
	if current == nil {
		return nil
	}

	// Track the object type of the current record as we walk. The type of the
	// ROOT variable is determined per-variable: $Record / $Record__Old map to
	// the flow's trigger object (stashed in _objectType); other variables (e.g.
	// results of RecordLookup) carry their own type via the Salesforce
	// attributes.type convention. Without that, we fall back to the trigger
	// object type — good enough for the common case, and safer than silently
	// using the wrong type.
	currentObjectType := rootObjectType(contextData, rootName, current)

	for i := 1; i < len(parts); i++ {
		fieldName := parts[i]

		// Handle `__r` / standard cross-object traversal. Schema-aware
		// resolution per CLAUDE.md R10 (PR #2057 round-6 F1): same
		// admin-authored `RelationshipName` must resolve identically across
		// formula evaluation, nested-reference traversal, and validation
		// formulas — all four readers share the same SSoT FK derivation.
		schema := e.lookupObjectSchema(ctx, currentObjectType)
		if fkField := e.resolveRelationshipFKWithSchema(schema, currentObjectType, fieldName); fkField != "" {
			if inline := lookupField(current, fieldName); inline != nil {
				current = inline
				currentObjectType = refineObjectType(inline, currentObjectType)
				continue
			}
			related, nextType, resolved := e.loadRelatedRecord(ctx, interview, user, schema, current, currentObjectType, fieldName, fkField)
			if !resolved {
				// R10 (issue #2047): indeterminate result (metadata-miss or
				// transient error). Emit a WRN so the miss is observable,
				// then return nil — same value as before, but now the log
				// distinguishes this from a definitive null FK.
				// Chosen interpretation: treat as nil (preserve existing
				// return-nil semantics) so downstream conditions see nil
				// rather than stale data. This matches the write-path
				// callers' behavior (skip-write on unresolved).
				if e.logger != nil {
					e.logger.Warn(fmt.Sprintf(
						"flow.r10: resolveNestedReferenceCtx skipping `%s` — loadRelatedRecord returned indeterminate (metadata-miss or query error)",
						fieldName,
					))
				}
				return nil
			}
			if related == nil {
				return nil
			}
			current = related
			currentObjectType = nextType
			continue
		}

		switch obj := current.(type) {
		case models.SObject:
			current = obj[fieldName]
		case map[string]interface{}:
			current = obj[fieldName]
		default:
			return nil
		}

		if current == nil {
			return nil
		}

		// After stepping into a nested map/sObject, refresh currentObjectType
		// from the nested record's attributes.type when available. Otherwise a
		// downstream `__r` hop would look up metadata on the PARENT object
		// rather than the current record's object.
		if newType := extractAttributesType(current); newType != "" {
			currentObjectType = newType
		}
	}

	return current
}

// resolveNestedReferenceCtxWithPresence is the leaf-existence-tracking
// sibling of resolveNestedReferenceCtx, used by resolveValueCtxWithPresence
// so RecordCreate stage 2 can distinguish a deep explicit-null
// (`$lead.SomeNullableField` where the leaf KEY exists with value nil —
// author/caller intent: "null this field") from a deep absent key
// (`$lead.MissingField` — let defaults fill).
//
// Presence is tracked at every hop via the two-value map read `v, ok := m[k]`:
//   - root variable missing from ContextData              → (nil, false)
//   - any intermediate key absent from its parent map     → (nil, false)
//   - any intermediate key present-but-nil (can't walk)   → (nil, false)
//   - __r relationship fails to load or resolves to nil   → (nil, false)
//     (preserves the existing resolveNestedReferenceCtx semantics for
//     cross-object hops; explicit-null on inline __r refs is an edge
//     case not exercised by shipped flows)
//   - leaf key present, value may be anything incl. nil   → (value, true)
//
// The function is otherwise identical in shape to resolveNestedReferenceCtx
// so future maintenance updates apply consistently to both.
func (e *FlowExecutor) resolveNestedReferenceCtxWithPresence(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, ref string) (interface{}, bool) {
	if interview == nil {
		// No interview context — presence is undefined. Preserve the
		// existing nil-return behavior; mark absent so the caller skips
		// rather than writing a spurious nil.
		return e.resolveNestedReference(nil, ref), false
	}
	parts := strings.Split(ref, ".")
	if len(parts) == 0 {
		return nil, false
	}

	contextData := interview.ContextData
	rootName := parts[0]
	current, rootPresent := contextData[rootName]
	if !rootPresent {
		return nil, false
	}
	// Single-part "nested" ref is really just a root lookup — return
	// present-with-value (value may be nil). The caller checking exists=true
	// then honors nil as the author's intent.
	if len(parts) == 1 {
		return current, true
	}
	// Multi-part ref but root is nil — can't traverse; treat as absent.
	if current == nil {
		return nil, false
	}

	currentObjectType := rootObjectType(contextData, rootName, current)

	for i := 1; i < len(parts); i++ {
		fieldName := parts[i]
		isLeaf := i == len(parts)-1

		// Schema-aware resolution per CLAUDE.md R10 (PR #2057 round-6 F1):
		// see resolveNestedReferenceCtx for the full rationale. Both nested
		// reference paths must consult the same RelationshipName-aware
		// resolver as buildFormulaFetcher.
		schema := e.lookupObjectSchema(ctx, currentObjectType)
		if fkField := e.resolveRelationshipFKWithSchema(schema, currentObjectType, fieldName); fkField != "" {
			if inline := lookupField(current, fieldName); inline != nil {
				current = inline
				currentObjectType = refineObjectType(inline, currentObjectType)
				continue
			}
			related, nextType, resolved := e.loadRelatedRecord(ctx, interview, user, schema, current, currentObjectType, fieldName, fkField)
			if !resolved {
				// R10 (issue #2047): indeterminate result — presence=false.
				// Chosen interpretation: return (nil, false) so the
				// write-path caller (RecordCreate stage 2, executeAssignment)
				// skips the assignment rather than writing nil. This is
				// identical in value to the prior definitive-null path but
				// now correctly represents "we don't know" semantics.
				// Matches main #2017's intent (skip write to avoid target
				// corruption) but adds the observability log.
				if e.logger != nil {
					e.logger.Warn(fmt.Sprintf(
						"flow.r10: resolveNestedReferenceCtxWithPresence skipping `%s` — loadRelatedRecord returned indeterminate (metadata-miss or query error)",
						fieldName,
					))
				}
				return nil, false
			}
			if related == nil {
				// Definitive null: FK was cleared or parent was deleted.
				// Return (nil, true) so the caller writes nil per SF semantics
				// (main #2017 / #1817 F4 fix). The earlier !resolved branch
				// above handles indeterminate (metadata-miss / query error).
				return nil, true
			}
			current = related
			currentObjectType = nextType
			continue
		}

		// Leaf / intermediate map access, tracking key presence
		// explicitly. This is the whole point of this function: ok=true
		// even when value is nil is the "author set explicit null" signal
		// that must reach RecordCreate stage 2.
		var val interface{}
		var present bool
		switch obj := current.(type) {
		case models.SObject:
			val, present = obj[fieldName]
		case map[string]interface{}:
			val, present = obj[fieldName]
		default:
			return nil, false
		}

		if !present {
			return nil, false
		}

		if isLeaf {
			return val, true
		}

		// Intermediate hop must have a non-nil value we can descend into.
		if val == nil {
			return nil, false
		}
		current = val
		if newType := extractAttributesType(current); newType != "" {
			currentObjectType = newType
		}
	}

	return current, true
}

// refineObjectType picks the best known object type after stepping through an
// inline relationship record. It prefers the nested record's own
// attributes.type (Salesforce convention) and falls back to the previous type
// when attributes are missing (better than silently emptying the string).
func refineObjectType(record interface{}, previous string) string {
	if typeName := extractAttributesType(record); typeName != "" {
		return typeName
	}
	return previous
}

// rootObjectType resolves the object type of a root context variable so __r
// traversal issues DB lookups against the correct object.
//
// Rules (in order):
//  1. $Record / $Record__Old → use `_objectType` (the flow trigger object)
//  2. attributes.type on the record itself → canonical Salesforce convention
//  3. Fallback to `_objectType` — safer than empty, but noted as imperfect for
//     non-$Record variables that lack attributes.type.
func rootObjectType(contextData map[string]interface{}, rootName string, record interface{}) string {
	rootType, _ := contextData[flowContextKeyRootObject].(string)
	if rootName == "$Record" || rootName == "$Record__Old" {
		return rootType
	}
	// Try attributes.type per Salesforce SObject convention.
	if typeName := extractAttributesType(record); typeName != "" {
		return typeName
	}
	return rootType
}

// inferRootObjectType recovers the trigger object type from a $Record /
// $Record__Old carried in ContextData, using the Salesforce attributes.type
// convention. Used by executeFlowMetadata for AutoLaunchedFlow subflows that
// have no declared Start.Object but receive typed $Record inputs from the
// parent. Returns "" when neither input is present or carries attributes.
func inferRootObjectType(contextData map[string]interface{}) string {
	for _, key := range []string{"$Record", "$Record__Old"} {
		if rec, ok := contextData[key]; ok {
			if typeName := extractAttributesType(rec); typeName != "" {
				return typeName
			}
		}
	}
	return ""
}

// extractAttributesType pulls the Salesforce attributes.type string off an
// SObject or plain map. Returns "" when unavailable.
func extractAttributesType(record interface{}) string {
	var attrs interface{}
	switch obj := record.(type) {
	case models.SObject:
		attrs = obj["attributes"]
	case map[string]interface{}:
		attrs = obj["attributes"]
	default:
		return ""
	}
	attrMap, ok := attrs.(map[string]interface{})
	if !ok {
		return ""
	}
	typeName, _ := attrMap["type"].(string)
	return typeName
}

// loadRelatedRecord performs the cross-object lookup for a single `__r` segment.
// Returns (relatedRecord, relatedObjectType, resolved) where resolved distinguishes:
//
//   - resolved=true, record=nil: definitive null per SF semantics (FK cleared or
//     orphan parent — the field is genuinely absent; downstream may treat as null).
//
//   - resolved=false, record=nil: indeterminate (metadata-miss or transient query
//     error — we don't know; downstream should SKIP rather than write nil, to
//     preserve the target's prior value).
//
//   - resolved=true, record!=nil: successful lookup.
//
// This distinction fixes the Codex P1 on PR #2017: the `if !resolved { break }`
// gate in executeAssignment was previously too broad — it skipped writes even for
// FK-null and orphan lookups, leaving ghost data. Now only metadata-miss and
// query errors are indeterminate; FK-null and orphan are definitive nulls.
//
// Read-path callers (resolveNestedReferenceCtx, resolveNestedReferenceCtxWithPresence,
// buildFormulaFetcher) apply the same distinction per R10 (issue #2047): no path
// silently conflates metadata-miss with explicit null.
func (e *FlowExecutor) loadRelatedRecord(
	ctx context.Context,
	interview *models.FlowInterview,
	user *models.UserSession,
	schema *models.ObjectMetadata,
	currentRecord interface{},
	currentObjectType, relationshipName, fkField string,
) (models.SObject, string, bool) {
	// CLAUDE.md R10: this wrapper layers per-interview concerns (rCache,
	// N+1 warn, attributes.type stamping, automation-specific WRN policy)
	// AROUND a single shared helper — formula.FetchRelatedRecord — so the
	// nested-reference read path and the formula read path agree byte-for-
	// byte on FK extraction, target resolution, and DB fetch. The polymorphic
	// single-ReferenceTo Tier-4 gate already lives upstream in
	// resolveRelationshipFK; by the time we get here, fkField has been
	// vetted, so we keep this wrapper focused on caching + logging.

	// 1. Extract the FK value from the current record.
	fkValue := lookupField(currentRecord, fkField)
	if fkValue == nil || fkValue == "" {
		// FK is null/empty — definitive null per SF semantics: the FK field is
		// genuinely absent or cleared. resolved=true: we KNOW there is no
		// parent. Mirror FetchRelatedRecord's definitive-null semantics so
		// downstream callers (read + write paths) treat both helpers the same.
		return nil, "", true
	}
	fkID := fmt.Sprintf("%v", fkValue)

	// 2. Resolve the related object type ONCE so the cache key and the
	// indeterminate-gate decision agree, and so the WRN messages can name the
	// resolved type. Routes through formula.ResolveRelationTarget so
	// polymorphic references (Task.WhoId → [Contact, Lead], Task.WhatId →
	// [Account, Opportunity, …]) honour the row's `<FK>_type` discriminator
	// — pre-#2057-round-8 this picked field.ReferenceTo[0] for any length
	// and the cache key + attributes.type stamp drifted away from the
	// shared helper's resolved target. The shared helper also gates
	// indeterminate cases (missing / empty / non-matching discriminator)
	// the same way as FetchRelatedRecord, so both reads agree byte-for-byte.
	recordMap := recordAsMap(currentRecord)
	relatedType := ""
	if currentObjectType != "" {
		if field := e.lookupFieldMeta(currentObjectType, fkField); field != nil {
			if t, ok := formula.ResolveRelationTarget(recordMap, fkField, field.ReferenceTo); ok {
				relatedType = t
			}
		}
	}
	if relatedType == "" {
		// SSoT fallback: standard FK→target table in pkg/formula. Records/
		// does not need this fallback because its schema is always fully
		// loaded before computeFormulaFields runs.
		if t, ok := formula.LookupStandardRelationTarget(fkField); ok {
			relatedType = t
		}
	}

	// 3. Cache lookup keyed by {relatedType}:{fkID}. When relatedType is
	// empty (no metadata) we fall back to a type-less key so a previously
	// seeded entry can still hit. Done BEFORE the N+1 Loop warn so we don't
	// flood logs when the cache absorbs repeat lookups within a Loop body.
	cacheKey := relatedType + ":" + fkID
	cache := rCache(interview)
	if cache != nil {
		if cached, ok := cache[cacheKey]; ok {
			// Cache hit — resolved=true regardless of whether the cached value
			// is nil. A nil cache entry is the negative cache: a definitive
			// null (orphan or FK-cleared) established in a prior call within
			// the same interview.
			return cached, relatedType, true
		}
	}

	// 4. Warn when we would issue a DB query inside a Loop (cache MISS path).
	// Emitted BEFORE FetchRelatedRecord so the warning is visible even when
	// the helper subsequently classifies the result as indeterminate.
	if loopDepth(interview.ContextData) > 0 {
		if e.logger != nil {
			e.logger.Warn(fmt.Sprintf(
				"flow: `%s` resolved inside Loop (cache miss) — potential N+1 query (relationship=%s, fkField=%s, fkId=%s)",
				relationshipName, relationshipName, fkField, fkID,
			))
		}
	}

	// 5. Indeterminate gate: without a concrete type, or without a query
	// service, we cannot fetch. Bail out before constructing the deps adapter
	// so the WRN message preserves the pre-refactor wording. Same outcome as
	// FetchRelatedRecord would produce (LookupReferenceTo would return empty
	// → resolved=false), but the WRN is more specific because we know which
	// gate fired.
	if relatedType == "" || e.query == nil {
		if e.logger != nil {
			e.logger.Warn(fmt.Sprintf(
				"flow: cannot resolve `%s` — missing related object type (fkField=%s, fkId=%s)",
				relationshipName, fkField, fkID,
			))
		}
		return nil, "", false // indeterminate: metadata-miss
	}

	// 6. Delegate FK→fetch to the SHARED helper. The deps adapter is the
	// same shape buildFormulaFetcher uses, so both read paths agree on
	// LookupReferenceTo + FetchByID semantics (R10). recordMap was already
	// derived in step 2 for ResolveRelationTarget; reuse it.
	deps := &automationFormulaFetchDeps{e: e, user: user, currentObjectType: currentObjectType}
	// Schema is threaded through so FetchRelatedRecord's internal
	// ResolveRelationFK call hits Tier 1 (RelationshipName) consistently
	// with the pre-resolved fkField above. Pre-#2057-R6, this passed nil
	// and the helper silently fell through to T2/T3/T4 — which would
	// disagree with the caller for any admin-renamed relationship.
	related, fetchedType, resolved := formula.FetchRelatedRecord(ctx, deps, schema, currentObjectType, recordMap, relationshipName)
	// Pre-#2057-round-8 we discarded fetchedType and re-stamped attributes
	// with the locally pre-computed relatedType. For polymorphic FKs the two
	// could disagree (pre-compute picked refTo[0]; helper consulted the
	// discriminator), so the cache + stamp drifted from the actual fetched
	// row's SObject. Now both sides go through ResolveRelationTarget; if a
	// future divergence ever sneaks back in, prefer the helper's value as
	// the authoritative target since it owns the FetchByID call.
	if fetchedType != "" && fetchedType != relatedType {
		relatedType = fetchedType
		cacheKey = relatedType + ":" + fkID
	}

	if !resolved {
		// Helper hit a transient error during FetchByID. Emit the same WRN
		// shape as before so log greps stay stable.
		if e.logger != nil {
			e.logger.Warn(fmt.Sprintf(
				"flow: `%s` lookup failed (relatedType=%s, fkId=%s)",
				relationshipName, relatedType, fkID,
			))
		}
		return nil, "", false // indeterminate: transient query error
	}
	if related == nil {
		// Definitive null: orphan FK (FK present, parent row absent). Mirror
		// pre-refactor WRN + negative-cache so a repeat lookup short-circuits.
		if e.logger != nil {
			e.logger.Warn(fmt.Sprintf(
				"flow: `%s` related record not found (relatedType=%s, fkId=%s)",
				relationshipName, relatedType, fkID,
			))
		}
		if cache != nil {
			cache[cacheKey] = nil
		}
		return nil, "", true // definitive null: parent deleted / orphan
	}

	// 7. Stamp attributes.type so later reuse (assignment, subflow input,
	// further __r traversal from a nested position) can recover the type
	// without another metadata roundtrip. Cache the annotated record.
	record := models.SObject(related)
	stampAttributesType(record, relatedType)
	if cache != nil {
		cache[cacheKey] = record
	}
	return record, relatedType, true
}

// lookupField reads a field value from either an SObject or a plain map, used
// when we need a specific column (the FK) from a record whose concrete type
// may vary at the call site.
func lookupField(record interface{}, field string) interface{} {
	switch obj := record.(type) {
	case models.SObject:
		return obj[field]
	case map[string]interface{}:
		return obj[field]
	default:
		return nil
	}
}

// recordAsMap exposes a record as a plain map[string]interface{} for callers
// that need to pass it into shared helpers (e.g. formula.ResolveRelationTarget,
// formula.FetchRelatedRecord). Returns nil when the concrete type is neither
// SObject nor map[string]interface{} — callers must treat nil as "no record"
// and not dereference.
func recordAsMap(record interface{}) map[string]interface{} {
	switch obj := record.(type) {
	case models.SObject:
		return map[string]interface{}(obj)
	case map[string]interface{}:
		return obj
	default:
		return nil
	}
}

// resolveValueCtx is the __r-aware counterpart of resolveValue. It is a thin
// wrapper that routes dotted references through resolveNestedReferenceCtx so
// cross-object lookups actually happen during runtime evaluation. Callers that
// have access to ctx + interview + user (record ops, actions, assignments)
// must prefer this method; callers without a live interview (scheduled filter
// builders, condition-only evaluators) keep using the pure resolveValue.
// deepResolveValue is the map/slice-aware counterpart of resolveValueCtx.
// It recurses into literal map/slice values declared directly in flow
// metadata so an inline structure like
//
//	{"Account": {elementReference: "accountId"}, "Contact": ...}
//
// resolves every nested elementReference wrapper. Critically, it stops
// recursing as soon as it has unwrapped an elementReference: whatever
// lives inside a resolved variable is returned verbatim, so strings in a
// resolved SObject are never re-run through the variable resolver (which
// would echo unknown names back and risk accidental rebinding).
func (e *FlowExecutor) deepResolveValue(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, value interface{}) interface{} {
	if m, ok := value.(map[string]interface{}); ok {
		if _, isRef := m["elementReference"]; isRef {
			return e.resolveValueCtx(ctx, interview, user, value)
		}
		out := make(map[string]interface{}, len(m))
		for k, v := range m {
			out[k] = e.deepResolveValue(ctx, interview, user, v)
		}
		return out
	}
	if s, ok := value.([]interface{}); ok {
		out := make([]interface{}, len(s))
		for i, v := range s {
			out[i] = e.deepResolveValue(ctx, interview, user, v)
		}
		return out
	}
	return e.resolveValueCtx(ctx, interview, user, value)
}

func (e *FlowExecutor) resolveValueCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, value interface{}) interface{} {
	// Preserve the original single-return signature for all existing
	// callers (evaluateCondition, executeAssignment, …). The new
	// resolveValueCtxWithPresence carries the extra "was the referenced
	// key actually found?" bit used only by executeRecordCreate stage 2
	// to distinguish an explicit-null write from an unresolved reference.
	// See that function for the full semantics.
	v, _ := e.resolveValueCtxWithPresence(ctx, interview, user, value)
	return v
}

// resolveValueCtxWithPresence is the typed-wrapper counterpart of
// resolveValueCtx, used by executeRecordCreate stage 2 to distinguish:
//
//   - "elementReference wrapper whose target EXISTS in ContextData with
//     value nil" — the caller/flow explicitly set null; the create MUST
//     write nil (e.g. to clear a nullable field, suppress a schema
//     DefaultValue via layer 1/2 semantics). exists=true.
//
//   - "elementReference wrapper whose target is ABSENT" — nothing was
//     provided; stage 3 (ActionFieldDefault) / stage 5 (schema
//     DefaultValue) should fill. exists=false → caller skips the
//     assignment.
//
// Return-value semantics mirror resolveValueCtx exactly; the only new
// information is the boolean. Literals always report exists=true (a
// literal nil is a deliberate write, not an unresolved reference).
// Raw strings preserve the legacy echo-back passthrough — only
// elementReference wrappers whose reference doesn't resolve report
// exists=false. Nested references (`$lead.Company`) delegate to
// resolveNestedReferenceCtxWithPresence, which does `(val, ok)` tracking
// at every hop so a leaf key that exists with value nil returns
// (nil, true) — RecordCreate stage 2 honors the author's explicit-null
// intent instead of letting stage 3/5 defaults overwrite it.
func (e *FlowExecutor) resolveValueCtxWithPresence(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, value interface{}) (interface{}, bool) {
	if value == nil {
		// Literal nil: the flow metadata explicitly authored null (or a
		// REST caller passed a JSON null). Either way it's "present".
		return nil, true
	}

	// Detect an elementReference typed-wrapper BEFORE unwrapping so we can
	// tell "unresolvable element reference" apart from "literal string".
	// Flow metadata encodes element/formula references as wrappers
	// (e.g. {"elementReference": "formula_2_..."}), never as bare strings.
	// When the referenced name has no entry in ContextData — the common case
	// being a FlowFormula whose Engine.Evaluate returned an error and was
	// WARN-logged + skipped by evaluateFlowFormulas (see the WARN at
	// executor_helpers.go:~132) — the legacy code path echoed the reference
	// string back, so downstream executeRecordUpdate / executeRecordCreate
	// wrote the element's NAME into the DB column (see issue #1123 where
	// an Order picklist column was persisted as the literal string
	// "formula_2_myRule_1_A1_6505267583"). Returning nil instead surfaces
	// "no value" to the persistence layer, which matches the WARN-and-
	// continue contract.
	wasElementReference := false
	if m, ok := value.(map[string]interface{}); ok {
		if ref, exists := m["elementReference"]; exists {
			if _, isStr := ref.(string); isStr {
				wasElementReference = true
			}
		}
	}

	var wasLiteral bool
	value, wasLiteral = extractTypedValue(value)

	ref, ok := value.(string)
	if !ok {
		return value, true
	}

	if strings.HasPrefix(ref, "$Flow.") {
		return e.resolveFlowSystemVariable(ref), true
	}
	if strings.HasPrefix(ref, "$GlobalConstant.") {
		return e.resolveGlobalConstant(ref), true
	}
	// Route through nested-reference resolution when the string looks like a
	// variable path. Three signals qualify:
	//   1. wasElementReference: the original value was an elementReference
	//      wrapper (e.g. {"elementReference": "$Record.Id"}) — always a reference.
	//   2. looksLikeVariableReference: a bare string whose characters are all
	//      identifier-safe (letters, digits, underscores, $, dots) — e.g. a bare
	//      "$Record.Order__r.Status__c" passed directly as filter.Value. Salesforce
	//      Flow variable names cannot contain spaces, parens, or hyphens.
	//
	// wasLiteral=true (value came from a typed-value wrapper like stringValue)
	// overrides both: the flow author explicitly authored a concrete value, so it
	// is never a variable reference regardless of its character content. This
	// fixes issue #2046 where dotted identifier-shape literals like "v1.2.3" or
	// "name.tld" were misrouted through resolveNestedReferenceCtxWithPresence and
	// silently dropped (first segment lookup miss → (nil, false) → Assign skips).
	//
	// A literal stringValue that happens to contain dots but also contains spaces
	// or other non-identifier chars (e.g. "Acme Corp. Ltd.", "v1.2.3 (beta)")
	// fails looksLikeVariableReference and passes through unchanged — treating it
	// as a variable path would split it at the first dot, return nil from ContextData
	// lookup, and silently drop the write (issue #1817 Bug A: literals containing
	// spaces+dots silently lost their value because they were misrouted through
	// the nested-reference resolver).
	if !wasLiteral && strings.Contains(ref, ".") && (wasElementReference || looksLikeVariableReference(ref)) {
		// Nested ref: delegate to the presence-tracking sibling of
		// resolveNestedReferenceCtx. Leaf-level map lookup uses (val, ok)
		// form at every hop so a deep explicit-null like
		// `$lead.SomeNullableField` where the leaf KEY exists with value
		// nil returns (nil, true) — allowing RecordCreate stage 2 to
		// honor the author's null-claim instead of letting stage 3 / 5
		// defaults overwrite it. Same semantic class as the simple-ref
		// fix above; no "future work" asterisk.
		return e.resolveNestedReferenceCtxWithPresence(ctx, interview, user, ref)
	}

	if interview != nil && interview.ContextData != nil {
		if resolved, exists := interview.ContextData[ref]; exists {
			// Key IS in ContextData. Value may be nil (caller set explicit
			// null) or any other value. Either way the semantic is "the
			// author/caller chose this; write it through".
			return resolved, true
		}
	}
	// Unresolved. For elementReference wrappers signal presence=false so
	// RecordCreate stage 2 can skip the assignment and let stages 3/5
	// fill. Raw literal strings keep their legacy passthrough — callers
	// such as evaluateCondition compare via fmt.Sprintf and tolerate the
	// echo; changing that would broaden the blast radius of this fix.
	if wasElementReference {
		return nil, false
	}
	return value, true
}

// currentAssignmentValueCtx is the __r-aware counterpart of currentAssignmentValue.
// Used by executeAssignment where both the read (of the current value) and the
// subsequent write need to agree on whether cross-object lookups happen.
func (e *FlowExecutor) currentAssignmentValueCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, ref string) interface{} {
	if !strings.Contains(ref, ".") {
		if interview != nil && interview.ContextData != nil {
			return interview.ContextData[ref]
		}
		return nil
	}
	return e.resolveNestedReferenceCtx(ctx, interview, user, ref)
}

// evaluateConditionsCtx is the __r-aware counterpart of evaluateConditions. It
// routes each condition through evaluateConditionCtx so decisions and loop
// filters can reason about `$Record.Order__r.Status__c` without changing the
// non-__r code paths.
func (e *FlowExecutor) evaluateConditionsCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, conditions []models.FlowCondition, logic string) bool {
	if len(conditions) == 0 {
		return true
	}

	results := make([]bool, len(conditions))
	for i, cond := range conditions {
		results[i] = e.evaluateConditionCtx(ctx, interview, user, &cond)
	}

	return evaluateConditionLogic(logic, results)
}

// sfKeyPrefixMatch evaluates an SF-style `StartsWith "<3-char-prefix>"` check
// against a reference-type value. Salesforce's 15/18-char IDs encode object
// type in the prefix; UUIDs do not. When a flow condition compares an ID
// against a known standard SObject key prefix, translate it to a polymorphic
// type check by looking up the ID in the prefix's canonical table.
//
// The standard prefix → object table is sourced from shared/pkg/salesforce
// (single SSoT for SF standard prefix facts; no parallel copy lives here).
// Returns (matched, handled):
//   - handled=true, matched=bool    → definitive result, use matched
//   - handled=false                  → prefix does not look like an SF key
//     prefix, or resolution unavailable; caller falls back to literal
//     string matching so non-ID StartsWith comparisons stay unchanged.
func (e *FlowExecutor) sfKeyPrefixMatch(ctx context.Context, leftValue interface{}, prefix string) (matched bool, handled bool) {
	if len(prefix) != 3 {
		return false, false
	}
	expectedObject, ok := salesforce.StandardObjectForKeyPrefix(prefix)
	if !ok {
		return false, false
	}
	idStr, ok := leftValue.(string)
	if !ok || idStr == "" {
		return false, false
	}
	// Backward-compat: if the raw string literally starts with the prefix
	// (a non-UUID ID that happens to carry the SF-style prefix), defer to
	// the literal comparison so existing behaviour is preserved.
	if strings.HasPrefix(idStr, prefix) {
		return false, false
	}
	if e.query == nil {
		return false, false
	}
	result, err := e.query.Query(ctx, api.QueryRequest{
		ObjectAPIName: expectedObject,
		Criteria:      []api.QueryCriterion{{Field: constants.FieldID, Op: "=", Val: idStr}},
		Limit:         1,
	}, newSystemAutomationUser())
	if err != nil || result == nil {
		return false, false
	}
	return result.TotalSize > 0 && len(result.Records) > 0, true
}

// evaluateConditionCtx mirrors evaluateCondition but resolves both sides of
// the comparison through the __r-aware value resolver. The operator-switch
// body is intentionally identical to evaluateCondition — this keeps semantics
// guaranteed-equivalent for all existing operators and only changes HOW
// the left/right values are fetched.
func (e *FlowExecutor) evaluateConditionCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, cond *models.FlowCondition) bool {
	// Decision conditions always treat LeftValueReference as a variable
	// path (Salesforce Flow contract). Wrap it as an elementReference so
	// resolveValueCtx returns nil for unresolved names instead of echoing
	// the literal string — otherwise an unset variable looks non-blank to
	// IsBlank/IsNotBlank and a missing input silently takes the wrong
	// decision branch.
	leftValue := e.resolveValueCtx(ctx, interview, user, map[string]interface{}{"elementReference": cond.LeftValueReference})
	rawRight := e.resolveValueCtx(ctx, interview, user, cond.RightValue)
	rightValue, _ := extractTypedValue(rawRight)

	// Salesforce Process Builder emits conditions like
	//   leftValueReference: "isChangedRule_2_..."  operator: "EqualTo"  rightValue: {all-null}
	// where an all-null typed wrapper means "check if leftValue is truthy."
	// Detect this pattern and short-circuit to a truthiness check.
	if isAllNilRightValue(rawRight) {
		switch normalizeFlowOperator(cond.Operator) {
		case "EqualTo":
			return isTruthy(leftValue)
		case "NotEqualTo":
			return !isTruthy(leftValue)
		}
	}

	switch normalizeFlowOperator(cond.Operator) {
	case "EqualTo":
		return fmt.Sprintf("%v", leftValue) == fmt.Sprintf("%v", rightValue)
	case "NotEqualTo":
		return fmt.Sprintf("%v", leftValue) != fmt.Sprintf("%v", rightValue)
	case "GreaterThan":
		return isGreater(leftValue, rightValue)
	case "GreaterThanOrEqualTo":
		return isGreaterOrEqual(leftValue, rightValue)
	case "LessThan":
		return isLess(leftValue, rightValue)
	case "LessThanOrEqualTo":
		return isLessOrEqual(leftValue, rightValue)
	case "Contains":
		return containsString(leftValue, rightValue)
	case "NotContains":
		return !containsString(leftValue, rightValue)
	case "IsNull":
		isNull := leftValue == nil || leftValue == ""
		if bv, ok := rightValue.(bool); ok && !bv {
			return !isNull
		}
		return isNull
	case "IsNotNull":
		isNull := leftValue == nil || leftValue == ""
		if bv, ok := rightValue.(bool); ok && bv {
			return !isNull
		}
		return !isNull
	case "IsBlank":
		blank := isBlankValue(leftValue)
		if bv, ok := rightValue.(bool); ok && !bv {
			return !blank
		}
		return blank
	case "IsNotBlank":
		return !isBlankValue(leftValue)
	case "StartsWith":
		strFilter := fmt.Sprintf("%v", rightValue)
		if matched, handled := e.sfKeyPrefixMatch(ctx, leftValue, strFilter); handled {
			return matched
		}
		strVal := fmt.Sprintf("%v", leftValue)
		return strings.HasPrefix(strVal, strFilter)
	case "DoesNotStartWith":
		strFilter := fmt.Sprintf("%v", rightValue)
		if matched, handled := e.sfKeyPrefixMatch(ctx, leftValue, strFilter); handled {
			return !matched
		}
		strVal := fmt.Sprintf("%v", leftValue)
		return !strings.HasPrefix(strVal, strFilter)
	case "EndsWith":
		strVal := fmt.Sprintf("%v", leftValue)
		strFilter := fmt.Sprintf("%v", rightValue)
		return strings.HasSuffix(strVal, strFilter)
	case "DoesNotEndWith":
		strVal := fmt.Sprintf("%v", leftValue)
		strFilter := fmt.Sprintf("%v", rightValue)
		return !strings.HasSuffix(strVal, strFilter)
	case "Between":
		return isBetween(leftValue, rightValue)
	case "NotBetween":
		return !isBetween(leftValue, rightValue)
	case "In":
		return isIn(leftValue, rightValue)
	case "NotIn":
		return !isIn(leftValue, rightValue)
	case "Includes":
		return includes(leftValue, rightValue)
	case "Excludes":
		return excludes(leftValue, rightValue)
	case "EmptyString":
		return fmt.Sprintf("%v", leftValue) == ""
	case "IsChanged":
		// See evaluateCondition for semantics. The Ctx variant routes the
		// prior-field lookup through resolveValueCtx so __r-relationship
		// traversal stays consistent with the rest of the evaluator.
		priorRef, ok := deriveRecordOldRef(cond.LeftValueReference)
		if !ok {
			return false
		}
		priorValue := e.resolveValueCtx(ctx, interview, user, priorRef)
		return fmt.Sprintf("%v", leftValue) != fmt.Sprintf("%v", priorValue)
	default:
		return false
	}
}

// deriveRecordOldRef translates a "$Record.<path>" or "$Record__Prior.<path>" reference to its
// "$Record__Old.<path>" counterpart for IsChanged comparisons. Returns
// (priorRef, true) when the input is rooted at $Record or $Record__Prior, and ("", false)
// otherwise so callers can fall through to their unknown-ref handling.
//
// A bare "$Record" or "$Record__Prior" (no trailing dot) is intentionally not translated:
// Salesforce always emits IsChanged on a specific field, and a bare
// record reference has no field identity to compare.
//
// Also supports translating "$Record__Prior.<path>" back to "$Record__Old.<path>"
// and "$Record__Old.<path>" to itself (for consistency).
func deriveRecordOldRef(ref string) (string, bool) {
	const recordPrefix = "$Record."
	const priorPrefix = "$Record__Prior."
	const oldPrefix = "$Record__Old."

	if strings.HasPrefix(ref, recordPrefix) {
		return "$Record__Old." + ref[len(recordPrefix):], true
	} else if strings.HasPrefix(ref, priorPrefix) {
		return "$Record__Old." + ref[len(priorPrefix):], true
	} else if strings.HasPrefix(ref, oldPrefix) {
		// Already a $Record__Old reference, keep as is
		return ref, true
	}
	return "", false
}

// setNestedReference writes a value into a dot-notation target such as
// $Record.Account_Status__c. Missing intermediate maps are created on demand.
func (e *FlowExecutor) setNestedReference(context map[string]interface{}, ref string, value interface{}) bool {
	parts := strings.Split(ref, ".")
	if len(parts) == 0 {
		return false
	}
	if len(parts) == 1 {
		context[parts[0]] = value
		return true
	}

	current, exists := context[parts[0]]
	if !exists || current == nil {
		next := make(map[string]interface{})
		context[parts[0]] = next
		current = next
	}

	for i := 1; i < len(parts)-1; i++ {
		fieldName := parts[i]

		switch obj := current.(type) {
		case models.SObject:
			next := obj[fieldName]
			if next == nil {
				newMap := make(map[string]interface{})
				obj[fieldName] = newMap
				current = newMap
				continue
			}
			current = next
		case map[string]interface{}:
			next := obj[fieldName]
			if next == nil {
				newMap := make(map[string]interface{})
				obj[fieldName] = newMap
				current = newMap
				continue
			}
			current = next
		default:
			return false
		}
	}

	lastField := parts[len(parts)-1]
	switch obj := current.(type) {
	case models.SObject:
		obj[lastField] = value
		return true
	case map[string]interface{}:
		obj[lastField] = value
		return true
	default:
		return false
	}
}

// extractTypedValue unwraps Salesforce-style typed value objects like
// {"stringValue": "Technical Validation"} or {"booleanValue": true}.
//
// The second return value, wasLiteral, is true when the input was a typed-value
// wrapper carrying a concrete literal (stringValue / numberValue / booleanValue /
// dateValue / dateTimeValue). It is false for elementReference wrappers and for
// any non-map input. Wrapper-aware callers (resolveValue,
// resolveValueCtxWithPresence) use wasLiteral=true to bypass the
// looksLikeVariableReference heuristic so that dotted identifier-shape literals
// like "v1.2.3" or "name.tld" are never misrouted through the nested-reference
// resolver (issue #2046).
func extractTypedValue(v interface{}) (interface{}, bool) {
	m, ok := v.(map[string]interface{})
	if !ok {
		return v, false
	}
	// Literal-typed keys: wasLiteral=true.
	for _, key := range []string{"stringValue", "numberValue", "booleanValue", "dateValue", "dateTimeValue"} {
		if val, exists := m[key]; exists && val != nil {
			return val, true
		}
	}
	// elementReference is a variable reference, not a literal: wasLiteral=false.
	if val, exists := m["elementReference"]; exists && val != nil {
		return val, false
	}
	return v, false
}

func normalizeFlowOperator(operator string) string {
	if operator == "LessOrEqualTo" {
		return "LessThanOrEqualTo"
	}
	return operator
}

// isBlankValue implements Salesforce ISBLANK semantics.
//
// Returns true when the value is:
//   - nil
//   - an empty string "" (whitespace-only strings are NOT blank — this matches
//     the repo's expression engine, see pkg/expression/walker.go:402 and its
//     unicode_test.go "spaces are NOT blank" case)
//   - an empty slice or empty map / SObject
//
// Critically, ISBLANK(0) is false — zero is a value, not blank. This is the
// key differentiator from generic JS-style falsy checks. Booleans (true/false)
// are likewise never blank.
//
// Typed-value wrappers from Salesforce JSON (e.g. `{"stringValue": ""}`) are
// unwrapped first so a flow variable whose default is `{"stringValue": ""}`
// reads as blank rather than as a non-empty map.
//
// Reference: Salesforce Formulas docs — ISBLANK / ISNOTBLANK.
func isBlankValue(v interface{}) bool {
	if v == nil {
		return true
	}
	// Unwrap Salesforce-style typed value wrappers before checking so flow
	// variables whose stored form is a typed wrapper (e.g. loaded from JSON
	// metadata) are handled correctly on the left-hand side of a condition.
	v, _ = extractTypedValue(v)
	if v == nil {
		return true
	}
	// If the value is a Salesforce typed-value envelope with every payload
	// field null (extractTypedValue above returns the map unchanged in that
	// case, because it only unwraps non-nil arms), treat it as blank. This
	// matches how SF metadata serializes "no value provided" — a wide
	// wrapper with all arms nulled out.
	if m, ok := v.(map[string]interface{}); ok && isAllNilTypedWrapper(m) {
		return true
	}
	switch val := v.(type) {
	case string:
		return val == ""
	case []interface{}:
		return len(val) == 0
	case []string:
		return len(val) == 0
	case []models.SObject:
		// RecordLookup with GetFirstRecordOnly=false stores []SObject in
		// ContextData (executor_record_ops.go:282). A lookup that returns
		// zero rows must count as blank, not a non-blank empty collection.
		return len(val) == 0
	case map[string]interface{}:
		return len(val) == 0
	case models.SObject:
		return len(val) == 0
	}
	return false
}

// isAllNilTypedWrapper reports whether m is a Salesforce typed-value envelope
// (stringValue/numberValue/booleanValue/dateValue/dateTimeValue/elementReference)
// whose every payload field is nil. SF metadata serializes "no value" as the
// full wide map with all arms null, so from a blank-check perspective this is
// indistinguishable from a true nil payload.
//
// Requires at least one known typed-value key to be present so we never
// misclassify a genuine map with one nil entry as blank.
func isAllNilTypedWrapper(m map[string]interface{}) bool {
	typedKeys := []string{
		"stringValue", "numberValue", "booleanValue",
		"dateValue", "dateTimeValue", "elementReference",
	}
	seenTypedKey := false
	for _, k := range typedKeys {
		if v, ok := m[k]; ok {
			seenTypedKey = true
			if v != nil {
				return false
			}
		}
	}
	return seenTypedKey
}

// isAllNilRightValue returns true when a raw resolved rightValue is a
// Salesforce typed-value wrapper with every payload field nil. This signals
// "no right operand" — the condition is a truthiness check, not a comparison.
//
// Plain nil (author wrote `<ref> EqualTo null` or the referenced key is
// absent) is NOT the same signal — it must fall through to the literal
// comparison path so `"foo" EqualTo nil` stays false.
func isAllNilRightValue(raw interface{}) bool {
	m, ok := raw.(map[string]interface{})
	if !ok {
		return false
	}
	return isAllNilTypedWrapper(m)
}

// isTruthy applies SF-flow-style truthiness: non-nil, non-empty, non-zero.
// Covers the numeric shapes the TiDB driver (int64, []byte for strings) and
// JSON decoders (float64, json.Number) emit. Unrecognized types fall through
// to `return true` so map/struct payloads coming from upstream resolvers
// preserve prior non-nil-is-truthy behavior.
func isTruthy(v interface{}) bool {
	if v == nil {
		return false
	}
	switch val := v.(type) {
	case bool:
		return val
	case string:
		return val != ""
	case []byte:
		return len(val) != 0
	case float64:
		return val != 0
	case float32:
		return val != 0
	case int:
		return val != 0
	case int8:
		return val != 0
	case int16:
		return val != 0
	case int32:
		return val != 0
	case int64:
		return val != 0
	case uint:
		return val != 0
	case uint8:
		return val != 0
	case uint16:
		return val != 0
	case uint32:
		return val != 0
	case uint64:
		return val != 0
	case json.Number:
		if f, err := val.Float64(); err == nil {
			return f != 0
		}
		return val != ""
	}
	return true
}

func (e *FlowExecutor) buildFilterFromConditions(context map[string]interface{}, filters []models.FlowRecordFilter, logic string) string {
	return e.buildFilterFromConditionsImpl(filters, logic, false, func(v interface{}) interface{} {
		return e.resolveValue(context, v)
	})
}

// buildFilterFromConditionsCtx is the __r-aware counterpart. Used by
// executeRecordLookup so filter right-hand-side references like
// "$Record.Order__r.Id" resolve correctly via the interview-scoped cache.
func (e *FlowExecutor) buildFilterFromConditionsCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, filters []models.FlowRecordFilter, logic string) string {
	return e.buildFilterFromConditionsImpl(filters, logic, false, func(v interface{}) interface{} {
		return e.resolveValueCtx(ctx, interview, user, v)
	})
}

// buildFilterFromConditionsPBCtx builds a filter string using Salesforce
// Process Builder criteria semantics: for NotEqualTo comparisons against a
// non-null value, NULL field values are treated as "not equal" (TRUE), so the
// generated SQL adds "OR field IS NULL" to include NULL rows. This matches the
// runtime behaviour of Process Builder "Update Records" criteria — distinct
// from SOQL / standard-Flow semantics where NULL is excluded by !=.
func (e *FlowExecutor) buildFilterFromConditionsPBCtx(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, filters []models.FlowRecordFilter, logic string) string {
	return e.buildFilterFromConditionsImpl(filters, logic, true, func(v interface{}) interface{} {
		return e.resolveValueCtx(ctx, interview, user, v)
	})
}

// buildFilterFromConditionsImpl centralizes the SQL-generation loop so the
// callers only differ in how they resolve a filter's right-hand value and
// whether Process Builder null semantics apply.
//
// pbNullSemantics: when true, NotEqualTo against a non-null value generates
// "(field != 'v' OR field IS NULL)" so that NULL rows are included — matching
// Salesforce Process Builder criteria evaluation where a NULL field is treated
// as "not equal to any specific non-null value". Standard SOQL / Flow semantics
// (pbNullSemantics=false) leave the condition as "field != 'v'", which SQL
// evaluates to NULL for NULL fields, effectively excluding them.
func (e *FlowExecutor) buildFilterFromConditionsImpl(filters []models.FlowRecordFilter, logic string, pbNullSemantics bool, resolver func(interface{}) interface{}) string {
	if len(filters) == 0 {
		return ""
	}

	// Map Salesforce flow operator names to SQL. Flow metadata uses SF names
	// (EqualTo, Contains, IsBlank); raw SQL operators are kept for backward
	// compat. The previous SQL-only whitelist silently dropped every SF-named
	// operator — a blocking bug for all RecordLookup filters.
	sfToSQL := map[string]string{
		"EQUALTO": "=", "NOTEQUALTO": "!=",
		"GREATERTHAN": ">", "GREATERTHANOREQUALTO": ">=",
		"LESSTHAN": "<", "LESSTHANOREQUALTO": "<=", "LESSOREQUALTO": "<=",
		"CONTAINS": "LIKE_CONTAINS", "NOTCONTAINS": "NOT_LIKE_CONTAINS",
		"STARTSWITH": "LIKE_STARTS", "ENDSWITH": "LIKE_ENDS",
		"DOESNOTSTARTWITH": "NOT_LIKE_STARTS", "DOESNOTENDWITH": "NOT_LIKE_ENDS",
		"ISNULL": "IS_NULL", "ISNOTNULL": "IS_NOT_NULL",
		"ISBLANK": "IS_NULL", "ISNOTBLANK": "IS_NOT_NULL",
		"IN": "IN", "NOTIN": "NOT IN",
		"BETWEEN": "BETWEEN", "NOTBETWEEN": "NOT BETWEEN",
		"INCLUDES": "INCLUDES", "EXCLUDES": "EXCLUDES",
		"EMPTYSTRING": "EMPTY_STRING",
		// Raw SQL operators
		"=": "=", "!=": "!=", "<>": "<>",
		">": ">", ">=": ">=", "<": "<", "<=": "<=",
		"LIKE": "LIKE_CONTAINS", "NOT IN": "NOT IN",
	}

	conditions := make([]string, 0, len(filters))
	for _, filter := range filters {
		if err := validateIdentifier(filter.Field); err != nil {
			e.logger.Error("Invalid field name in flow filter, skipping", err)
			continue
		}

		opKey := strings.ToUpper(normalizeFlowOperator(filter.Operator))
		sqlOp, ok := sfToSQL[opKey]
		if !ok {
			e.logger.Warn("Unknown operator in flow filter, skipping: " + filter.Operator)
			continue
		}

		value := resolver(filter.Value)
		escapedValue := escapeSQLValue(value)
		literalValue := formatFilterLiteral(value)
		likeValue := escapeLikeValue(value) // also escapes % and _ for LIKE patterns

		switch sqlOp {
		case "IS_NULL":
			if boolValue, ok := value.(bool); ok && !boolValue {
				conditions = append(conditions, fmt.Sprintf("(%s IS NOT NULL AND %s != '')", filter.Field, filter.Field))
			} else {
				conditions = append(conditions, fmt.Sprintf("(%s IS NULL OR %s = '')", filter.Field, filter.Field))
			}
		case "IS_NOT_NULL":
			conditions = append(conditions, fmt.Sprintf("(%s IS NOT NULL AND %s != '')", filter.Field, filter.Field))
		case "LIKE_CONTAINS":
			conditions = append(conditions, fmt.Sprintf("CONTAINS(%s, '%s')", filter.Field, likeValue))
		case "NOT_LIKE_CONTAINS":
			conditions = append(conditions, fmt.Sprintf("NOT(CONTAINS(%s, '%s'))", filter.Field, likeValue))
		case "LIKE_STARTS":
			conditions = append(conditions, fmt.Sprintf("STARTS_WITH(%s, '%s')", filter.Field, likeValue))
		case "LIKE_ENDS":
			conditions = append(conditions, fmt.Sprintf("ENDS_WITH(%s, '%s')", filter.Field, likeValue))
		case "NOT_LIKE_STARTS":
			conditions = append(conditions, fmt.Sprintf("NOT(STARTS_WITH(%s, '%s'))", filter.Field, likeValue))
		case "NOT_LIKE_ENDS":
			conditions = append(conditions, fmt.Sprintf("NOT(ENDS_WITH(%s, '%s'))", filter.Field, likeValue))
		case "BETWEEN":
			parts := parseBetweenValue(value)
			if len(parts) == 2 {
				conditions = append(conditions, fmt.Sprintf("%s BETWEEN '%s' AND '%s'", filter.Field, escapeSQLValue(parts[0]), escapeSQLValue(parts[1])))
			} else {
				e.logger.Warn("Between operator requires 2-element array, skipping: " + filter.Field)
			}
		case "NOT BETWEEN":
			parts := parseBetweenValue(value)
			if len(parts) == 2 {
				conditions = append(conditions, fmt.Sprintf("%s NOT BETWEEN '%s' AND '%s'", filter.Field, escapeSQLValue(parts[0]), escapeSQLValue(parts[1])))
			} else {
				e.logger.Warn("NotBetween operator requires 2-element array, skipping: " + filter.Field)
			}
		case "INCLUDES":
			conditions = append(conditions, fmt.Sprintf("FIND_IN_SET('%s', %s) > 0", escapedValue, filter.Field))
		case "EXCLUDES":
			conditions = append(conditions, fmt.Sprintf("FIND_IN_SET('%s', %s) = 0", escapedValue, filter.Field))
		case "EMPTY_STRING":
			conditions = append(conditions, fmt.Sprintf("%s = ''", filter.Field))
		default:
			// For != / <> with Process Builder null semantics, include NULL rows:
			// in PB criteria evaluation NULL is "not equal to any non-null value",
			// so "(field != v OR field IS NULL)" matches what PB runtime does.
			// `literalValue` carries proper type formatting (quoted strings, bare
			// numbers/booleans) per `formatFilterLiteral`; the empty-value gate
			// stays on `escapedValue` since that's the resolved-string signal.
			if pbNullSemantics && (sqlOp == "!=" || sqlOp == "<>") && escapedValue != "" {
				conditions = append(conditions, fmt.Sprintf("(%s %s %s OR %s IS NULL)", filter.Field, sqlOp, literalValue, filter.Field))
			} else {
				conditions = append(conditions, fmt.Sprintf("%s %s %s", filter.Field, sqlOp, literalValue))
			}
		}
	}

	if len(conditions) == 0 {
		return "1=1" // No valid conditions, return always-true
	}
	return buildFilterLogicExpression(logic, conditions)
}

func (e *FlowExecutor) buildScheduledStartFilter(context map[string]interface{}, filters []models.FlowRecordFilter, logic string) string {
	if len(filters) == 0 {
		return ""
	}

	conditions := make([]string, 0, len(filters))
	for _, filter := range filters {
		clause := e.buildScheduledStartFilterClause(context, filter)
		if clause == "" {
			continue
		}
		conditions = append(conditions, clause)
	}

	if len(conditions) == 0 {
		return ""
	}

	return buildFilterLogicExpression(logic, conditions)
}

func (e *FlowExecutor) buildScheduledStartFilterClause(context map[string]interface{}, filter models.FlowRecordFilter) string {
	if err := validateIdentifier(filter.Field); err != nil {
		if e.logger != nil {
			e.logger.Error("Invalid field name in scheduled start filter, skipping", err)
		}
		return ""
	}

	operator := normalizeFlowOperator(filter.Operator)
	resolvedFilterValue, _ := extractTypedValue(e.resolveValue(context, filter.Value))
	escapedValue := escapeSQLValue(resolvedFilterValue)

	switch operator {
	case "EqualTo":
		return fmt.Sprintf("%s = '%s'", filter.Field, escapedValue)
	case "NotEqualTo":
		return fmt.Sprintf("%s != '%s'", filter.Field, escapedValue)
	case "GreaterThan":
		return fmt.Sprintf("%s > '%s'", filter.Field, escapedValue)
	case "GreaterThanOrEqualTo":
		return fmt.Sprintf("%s >= '%s'", filter.Field, escapedValue)
	case "LessThan":
		return fmt.Sprintf("%s < '%s'", filter.Field, escapedValue)
	case "LessThanOrEqualTo":
		return fmt.Sprintf("%s <= '%s'", filter.Field, escapedValue)
	case "Contains":
		return fmt.Sprintf("CONTAINS(%s, '%s')", filter.Field, escapedValue)
	case "StartsWith":
		return fmt.Sprintf("STARTS_WITH(%s, '%s')", filter.Field, escapedValue)
	case "EndsWith":
		return fmt.Sprintf("ENDS_WITH(%s, '%s')", filter.Field, escapedValue)
	case "IsNull":
		if boolValue, ok := resolvedFilterValue.(bool); ok && !boolValue {
			return fmt.Sprintf("NOT(ISBLANK(%s))", filter.Field)
		}
		return fmt.Sprintf("ISBLANK(%s)", filter.Field)
	case "IsBlank":
		return fmt.Sprintf("ISBLANK(%s)", filter.Field)
	case "IsNotBlank":
		return fmt.Sprintf("NOT(ISBLANK(%s))", filter.Field)
	case "IsChanged":
		// Scheduled flows have no prior-record state; keep the query broad and let
		// the final in-memory gate preserve the existing semantics.
		return "true"
	default:
		return ""
	}
}

// buildFilterLogicExpression renders a filterLogic expression as a SQL
// boolean expression over the per-condition predicates in `conditions`.
//
// SSoT collapse (R10): walk the SAME AST that evaluateConditionLogic
// uses, so the SQL prefilter and the in-memory runtime evaluator agree
// on the truth value for every record. Without this, the SQL prefilter
// would silently drop NOT / unknown tokens and load a different record
// set than the runtime evaluator accepts — the failure mode that
// motivated the round-2 R10 collapse.
//
// Failure semantics match the runtime evaluator: malformed expressions
// or out-of-range refs render as "1=0" (fail closed) so a scheduled
// flow with broken metadata never loads any records, instead of
// silently broadening the query.
func buildFilterLogicExpression(logic string, conditions []string) string {
	if len(conditions) == 0 {
		return ""
	}

	trimmedLogic := strings.TrimSpace(logic)
	switch strings.ToUpper(trimmedLogic) {
	case "", "AND":
		return strings.Join(conditions, " AND ")
	case "OR":
		return "(" + strings.Join(conditions, " OR ") + ")"
	}

	expr, err := parseConditionLogic(trimmedLogic)
	if err != nil || !expr.refsInRange(len(conditions)) {
		return "1=0"
	}
	return renderConditionLogicSQL(expr, conditions)
}

// renderConditionLogicSQL converts a conditionLogicNode AST into a SQL
// boolean expression by substituting each ref with the corresponding
// per-condition predicate string in `conditions` (1-indexed).
func renderConditionLogicSQL(n *conditionLogicNode, conditions []string) string {
	if n == nil {
		return "1=0"
	}
	switch n.kind {
	case "ref":
		idx := n.index - 1
		if idx < 0 || idx >= len(conditions) {
			return "1=0"
		}
		return "(" + conditions[idx] + ")"
	case "not":
		return "NOT (" + renderConditionLogicSQL(n.child, conditions) + ")"
	case "and", "or":
		op := " AND "
		if n.kind == "or" {
			op = " OR "
		}
		parts := make([]string, 0, len(n.children))
		for _, c := range n.children {
			parts = append(parts, renderConditionLogicSQL(c, conditions))
		}
		return "(" + strings.Join(parts, op) + ")"
	}
	return "1=0"
}

// escapeSQLValue escapes text for formula-like filter expressions to keep
// parser string-literal handling safe (expr parser expects backslash-style
// escaping for embedded quotes).
func escapeSQLValue(value interface{}) string {
	if value == nil {
		return ""
	}
	str := fmt.Sprintf("%v", value)
	// Escape backslash first so escaped characters stay stable in parser input.
	str = strings.ReplaceAll(str, "\\", "\\\\")
	// Use backslash-quote escaping (not doubled-quote), which this parser
	// recognizes for embedded apostrophes.
	str = strings.ReplaceAll(str, "'", "\\'")
	return str
}

func formatFilterLiteral(value interface{}) string {
	value, _ = extractTypedValue(value)
	switch v := value.(type) {
	case bool:
		if v {
			return "true"
		}
		return "false"
	default:
		return fmt.Sprintf("'%s'", escapeSQLValue(value))
	}
}

// escapeLikeValue escapes SQL LIKE wildcard characters (% and _) in addition
// to single quotes. Must be used for values interpolated into LIKE patterns
// (Contains, StartsWith, EndsWith) to prevent wildcard injection.
func escapeLikeValue(value interface{}) string {
	s := escapeSQLValue(value)
	s = strings.ReplaceAll(s, "%", "\\%")
	s = strings.ReplaceAll(s, "_", "\\_")
	return s
}

// defaultCompareLogger is the zerolog.Logger used by compareValues for WRN
// messages. It points at zerolog's global logger by default and can be
// overridden in tests to capture output.
var defaultCompareLogger = &log.Logger

// sfDateFormats lists the Salesforce-supported date/datetime formats in parse
// priority order. time.Parse tries each in sequence. RFC3339Nano covers
// fractional-second variants (e.g. "2025-06-01T15:04:05.000Z" — used by SF
// SOAP/REST datetime serialisation and by repo audit history persistence).
var sfDateFormats = []string{
	"2006-01-02",
	"2006-01-02T15:04:05Z07:00",
	"2006-01-02T15:04:05Z",
	"2006-01-02T15:04:05-07:00",
	time.RFC3339Nano,
}

// parseSFDateTime attempts to parse s as a Salesforce date or datetime value.
// Returns (t, true) on success, zero Time and false on failure.
func parseSFDateTime(s string) (time.Time, bool) {
	for _, layout := range sfDateFormats {
		if t, err := time.Parse(layout, s); err == nil {
			return t, true
		}
	}
	return time.Time{}, false
}

// compareValues compares two values for ordering and returns (cmp, ok).
//
// cmp is -1, 0, or 1 when ok=true. When ok=false the comparison is undefined
// (operands are nil, mixed-type, or neither numeric nor parseable as
// Salesforce date/datetime) and callers MUST treat all four ordering
// predicates (<, <=, >, >=) as false — see the isLess / isLessOrEqual /
// isGreater / isGreaterOrEqual helpers below.
//
// Resolution order:
//  1. Either operand nil → ok=false (no SF ordering against null).
//  2. Both numeric → float comparison.
//  3. Both parse as Salesforce date / datetime strings → time comparison.
//  4. Otherwise → ok=false; emit a WRN log with prefix "flow.compareValues:"
//     so callers can grep.
//
// All call sites that previously used compareNumbers for ordering operators
// route through the four helpers below; compareNumbers is kept unchanged for
// any other use.
func compareValues(a, b interface{}) (int, bool) {
	// 1. Nil operand → undefined (SF semantics: ordering against null is false).
	if a == nil || b == nil {
		return 0, false
	}

	// 2. Try numeric path first (preserves existing behavior).
	aFloat, aOk := utils.ToFloat64(a)
	bFloat, bOk := utils.ToFloat64(b)
	if aOk && bOk {
		switch {
		case aFloat < bFloat:
			return -1, true
		case aFloat > bFloat:
			return 1, true
		default:
			return 0, true
		}
	}

	// 3. Try date / datetime path when both operands are strings.
	aStr, aIsStr := a.(string)
	bStr, bIsStr := b.(string)
	if aIsStr && bIsStr {
		aTime, aDateOk := parseSFDateTime(aStr)
		bTime, bDateOk := parseSFDateTime(bStr)
		if aDateOk && bDateOk {
			switch {
			case aTime.Before(bTime):
				return -1, true
			case aTime.After(bTime):
				return 1, true
			default:
				return 0, true
			}
		}
	}

	// 4. Undefined: emit WRN and signal not-comparable.
	defaultCompareLogger.Warn().
		Str("a", fmt.Sprintf("%v", a)).
		Str("b", fmt.Sprintf("%v", b)).
		Msg("flow.compareValues: operands are neither numeric nor parseable as SF date/datetime; ordering predicates return false")

	return 0, false
}

// isLess / isLessOrEqual / isGreater / isGreaterOrEqual fail closed when
// compareValues reports the operands are uncomparable (ok=false). This keeps
// the four ordering operators consistent with SF semantics: predicates
// against null/uncomparable values evaluate to false, never true.
func isLess(a, b interface{}) bool {
	cmp, ok := compareValues(a, b)
	return ok && cmp < 0
}

func isLessOrEqual(a, b interface{}) bool {
	cmp, ok := compareValues(a, b)
	return ok && cmp <= 0
}

func isGreater(a, b interface{}) bool {
	cmp, ok := compareValues(a, b)
	return ok && cmp > 0
}

func isGreaterOrEqual(a, b interface{}) bool {
	cmp, ok := compareValues(a, b)
	return ok && cmp >= 0
}

// Utility functions
func compareNumbers(a, b interface{}) int {
	aFloat, aOk := utils.ToFloat64(a)
	bFloat, bOk := utils.ToFloat64(b)
	if !aOk || !bOk {
		return 0
	}
	if aFloat < bFloat {
		return -1
	}
	if aFloat > bFloat {
		return 1
	}
	return 0
}

func containsString(a, b interface{}) bool {
	aStr := fmt.Sprintf("%v", a)
	bStr := fmt.Sprintf("%v", b)
	return strings.Contains(aStr, bStr)
}

func isBetween(value, rangeValue interface{}) bool {
	// Assuming rangeValue is an array of two elements [lower, upper]
	rangeArr, ok := rangeValue.([]interface{})
	if !ok || len(rangeArr) != 2 {
		return false
	}

	valueFloat, valueOk := utils.ToFloat64(value)
	lowerFloat, lowerOk := utils.ToFloat64(rangeArr[0])
	upperFloat, upperOk := utils.ToFloat64(rangeArr[1])

	if !valueOk || !lowerOk || !upperOk {
		// If not numeric, try string comparison
		valueStr := fmt.Sprintf("%v", value)
		lowerStr := fmt.Sprintf("%v", rangeArr[0])
		upperStr := fmt.Sprintf("%v", rangeArr[1])
		return strings.Compare(lowerStr, valueStr) <= 0 && strings.Compare(valueStr, upperStr) <= 0
	}

	return lowerFloat <= valueFloat && valueFloat <= upperFloat
}

func isIn(value, listValue interface{}) bool {
	list, ok := listValue.([]interface{})
	if !ok {
		return false
	}

	valueStr := fmt.Sprintf("%v", value)
	for _, item := range list {
		if fmt.Sprintf("%v", item) == valueStr {
			return true
		}
	}

	return false
}

func includes(fieldValue, valueToCheck interface{}) bool {
	fieldStr := fmt.Sprintf("%v", fieldValue)
	valueToCheckStr := fmt.Sprintf("%v", valueToCheck)

	// For multi-picklist fields, values are stored as semicolon-separated
	items := strings.Split(fieldStr, ";")
	valueToFind := strings.TrimSpace(valueToCheckStr)

	for _, item := range items {
		if strings.TrimSpace(item) == valueToFind {
			return true
		}
	}

	return false
}

func excludes(fieldValue, valueToCheck interface{}) bool {
	return !includes(fieldValue, valueToCheck)
}

// parseBetweenValue extracts a 2-element slice from the filter value for
// BETWEEN/NOT BETWEEN SQL operators. Accepts []interface{} (from JSON
// unmarshal) or returns nil if the shape is unexpected.
func parseBetweenValue(value interface{}) []interface{} {
	switch v := value.(type) {
	case []interface{}:
		if len(v) == 2 {
			return v
		}
	}
	return nil
}