package automation

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"time"

	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/automation/triggerprojection"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/metadata"
	records "github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	"golang.org/x/sync/singleflight"
)

// PermissionProvider is the subset of PermissionService that the flow executor
// needs to populate the $Permission formula global for all users.
type PermissionProvider interface {
	GetEffectiveSystemPermissions(ctx context.Context, userID, profileID string) (map[string]bool, error)
	GetEffectiveCustomPermissions(ctx context.Context, userID, profileID string) (map[string]bool, error)
}

// FieldReadAuthorizer is the narrow FLS surface consumed by PushToConnector.
// It is split from PermissionProvider so existing $Permission test doubles do
// not need to implement record/field access behavior they do not exercise.
type FieldReadAuthorizer interface {
	CheckFieldVisibilityWithUser(ctx context.Context, objectAPIName, fieldAPIName string, user *models.UserSession) (bool, error)
}

// RecordReadAuthorizer is the narrow record-level READ surface consumed by
// PushToConnector before it serializes the trusted outbound source row.
type RecordReadAuthorizer interface {
	CheckRecordReadAccessWithUser(ctx context.Context, objectAPIName string, record models.SObject, user *models.UserSession) bool
}

// identifierRegex validates API names and IDs to prevent SQL injection
// Allows alphanumeric, underscores, and hyphens (for UUIDs)
var identifierRegex = regexp.MustCompile(`^[a-zA-Z0-9_-]+$`)

// flowRuntimeContextKey is the context.Context key under which a
// *FlowRuntimeContext is stored. It replaces the three private boolean keys
// (flowInterviewPersistModeKey, beforeSaveFlowKey, atomicFlowBoundaryKey) that
// previously threaded individual bool payloads through context.WithValue.
// The five context.WithValue keys are replaced in two phases:
//   - T21b (this commit): PersistMode, BeforeSaveFlow, AtomicBoundary → FlowRuntimeContext
//   - T21c/T21d: CallStack, RecordTriggerStack → FlowRuntimeContext
type flowRuntimeContextKey struct{}

// flowRuntimeContextFromCtx extracts the *FlowRuntimeContext stored on ctx by
// withFlowRuntimeContext, or returns a fresh default context when none is
// present (caller did not go through WithTransientFlowInterviews or
// WithBeforeSaveFlow). The returned value is never nil.
func flowRuntimeContextFromCtx(ctx context.Context) *FlowRuntimeContext {
	if frc, ok := ctx.Value(flowRuntimeContextKey{}).(*FlowRuntimeContext); ok && frc != nil {
		return frc
	}
	return NewFlowRuntimeContext(ctx, nil)
}

// withFlowRuntimeContext stores frc on ctx, returning the updated context.
// The FlowRuntimeContext's embedded ctx field is not updated here; callers
// that need the embedded ctx to reflect the outer context should call
// NewFlowRuntimeContext or update it after extraction.
func withFlowRuntimeContext(ctx context.Context, frc *FlowRuntimeContext) context.Context {
	return context.WithValue(ctx, flowRuntimeContextKey{}, frc)
}

// WithTransientFlowInterviews marks the given context so that any flow
// executed through ExecuteFlow / executeFlowVersion under it uses an
// in-memory transient FlowInterview and skips the two DB round-trips
// that would otherwise insert and update the row. Subflows inherit the
// mode automatically because they propagate the same context.
//
// The record-triggered listener wraps its context with this helper before
// firing RecordBeforeSave / RecordAfterSave / RecordBeforeDelete flows,
// which is where the observed ~160s-per-save latency was originating.
// Scheduled flows and REST-invoked flows intentionally do NOT call this —
// their audit rows are still useful for operator visibility.
//
// Implementation: stores an updated *FlowRuntimeContext on ctx (replacing
// the former flowInterviewPersistModeKey context.WithValue approach).
func WithTransientFlowInterviews(ctx context.Context) context.Context {
	frc := flowRuntimeContextFromCtx(ctx).WithInterviewPersistMode(InterviewPersistModeTransient)
	return withFlowRuntimeContext(ctx, frc)
}

// shouldPersistFlowInterview returns true when the current flow execution
// should write its FlowInterview row to the database (default), and false
// when the caller opted into transient execution via
// WithTransientFlowInterviews.
func shouldPersistFlowInterview(ctx context.Context) bool {
	return flowRuntimeContextFromCtx(ctx).ShouldPersistInterview()
}

// ShouldPersistFlowInterview is the exported form of shouldPersistFlowInterview.
// It lets sibling packages (e.g. the serverruntime connector wiring) assert
// whether a context has opted into transient interviews via
// WithTransientFlowInterviews, without exposing the internal runtime-context key.
func ShouldPersistFlowInterview(ctx context.Context) bool {
	return shouldPersistFlowInterview(ctx)
}

// WithBeforeSaveFlow marks the context as executing under a RecordBeforeSave
// trigger. Only the FlowTriggerListener's before-save branch should call
// this helper. AfterSave, scheduled, and REST-invoked runs must leave the
// flag absent so recordUpdate persists via DB UPDATE as normal.
//
// Implementation: stores an updated *FlowRuntimeContext on ctx (replacing
// the former beforeSaveFlowKey context.WithValue approach).
func WithBeforeSaveFlow(ctx context.Context) context.Context {
	frc := flowRuntimeContextFromCtx(ctx).WithBeforeSaveFlow(true)
	return withFlowRuntimeContext(ctx, frc)
}

// isBeforeSaveFlow reports whether the current execution was marked as
// a before-save record-triggered flow via WithBeforeSaveFlow.
func isBeforeSaveFlow(ctx context.Context) bool {
	return flowRuntimeContextFromCtx(ctx).IsBeforeSaveFlow()
}

// WithPreviewMode marks the given context as a no-DML preview execution.
// PreviewRecordCreateFields wraps its runCtx with this helper so every
// side-effect channel that previewPersistence cannot intercept on its own
// (audit emission, executeRecordUpdate's in-place $Record mutation,
// executeWait pause persistence) sees the marker and self-suppresses.
//
// Production execution paths (REST, scheduled, record-triggered) MUST NOT
// call this. The marker silently disables audit logging and rejects
// recordUpdate / Wait elements.
func WithPreviewMode(ctx context.Context) context.Context {
	frc := flowRuntimeContextFromCtx(ctx).WithPreviewMode(true)
	return withFlowRuntimeContext(ctx, frc)
}

// isPreviewMode reports whether ctx was marked as a preview run via
// WithPreviewMode. Safe to call with a nil/empty ctx (returns false).
func isPreviewMode(ctx context.Context) bool {
	return flowRuntimeContextFromCtx(ctx).IsPreviewMode()
}

// isWithinAtomicFlowBoundary reports whether the current execution is inside
// the Flow engine's RunAtomically wrapper. Reads from the *FlowRuntimeContext
// stored on ctx (replacing the former atomicFlowBoundaryKey approach).
func isWithinAtomicFlowBoundary(ctx context.Context) bool {
	return flowRuntimeContextFromCtx(ctx).IsWithinAtomicBoundary()
}

// withAtomicFlowBoundary returns a context with the atomic boundary marker set
// on the embedded *FlowRuntimeContext. Replaces context.WithValue for
// atomicFlowBoundaryKey — callers (runPossiblyAtomic) now use this helper
// but the information is carried in FlowRuntimeContext, not a bool value.
func withAtomicFlowBoundary(ctx context.Context) context.Context {
	frc := flowRuntimeContextFromCtx(ctx).WithAtomicBoundary(true)
	return withFlowRuntimeContext(ctx, frc)
}

// atomicFlowPessimisticRetryOwnerKey marks the root Flow callback owned by
// RunInPessimisticTransaction. Protected actions that can invert lock order
// with a service-owned transaction require this proof so TiDB 1213 deadlock or
// 1205 lock-wait contention rolls back and replays the entire Flow attempt.
type atomicFlowPessimisticRetryOwnerKey struct{}

func withAtomicFlowPessimisticRetryOwner(ctx context.Context) context.Context {
	return context.WithValue(ctx, atomicFlowPessimisticRetryOwnerKey{}, true)
}

func hasAtomicFlowPessimisticRetryOwner(ctx context.Context) bool {
	owned, _ := ctx.Value(atomicFlowPessimisticRetryOwnerKey{}).(bool)
	return owned
}

func withRecordTriggerRoot(ctx context.Context) context.Context {
	frc := flowRuntimeContextFromCtx(ctx)
	frc = frc.WithRecordTriggerRoot(frc.CallStackDepth())
	return withFlowRuntimeContext(ctx, frc)
}

type recordTriggerRootRecordIDKey struct{}

func withRecordTriggerRootRecordID(ctx context.Context, recordID string) context.Context {
	ctx = withRecordTriggerRoot(ctx)
	if recordID = strings.TrimSpace(recordID); recordID != "" {
		ctx = context.WithValue(ctx, recordTriggerRootRecordIDKey{}, recordID)
	}
	return ctx
}

func recordTriggerRootRecordID(ctx context.Context) string {
	recordID, _ := ctx.Value(recordTriggerRootRecordIDKey{}).(string)
	return strings.TrimSpace(recordID)
}

// isRecordTriggerRootExecution reports whether the current ExecuteFlow call is
// the root flow dispatched directly by FlowTriggerListener. Direct REST /
// invocable execution has no record-trigger dispatch marker. Subflows inherit
// the marker but run at a deeper call-stack depth than the root baseline
// captured by FlowTriggerListener.
func isRecordTriggerRootExecution(ctx context.Context) bool {
	frc := flowRuntimeContextFromCtx(ctx)
	return frc.HasRecordTriggerRoot() && frc.CallStackDepth() == frc.RecordTriggerRootCallDepth()
}

func withCurrentFlowExecution(ctx context.Context, flowAPIName string, flowVersion *models.FlowVersion) context.Context {
	if flowVersion == nil {
		return ctx
	}
	identity := FlowExecutionIdentity{
		APIName:                flowAPIName,
		DefinitionID:           flowVersion.DefinitionId,
		VersionID:              flowVersion.Id,
		OwnerPack:              flowVersion.DefinitionOwnerPack,
		DefinitionIsCustomized: flowVersion.DefinitionIsCustomized,
		VersionSource:          flowVersion.Source,
	}
	frc := flowRuntimeContextFromCtx(ctx).WithCurrentFlow(identity)
	return withFlowRuntimeContext(ctx, frc)
}

const scheduledFlowPageSize = 500
const scheduledFlowMaxMatched = 1000
const scheduledFlowMaxScanned = 50000

type scheduledFlowCandidateScanBudgetError struct {
	ObjectName          string
	ScannedCount        int
	MatchedCount        int
	ScanCap             int
	MatchedCap          int
	SQLPrefilterApplied bool
}

func (e *scheduledFlowCandidateScanBudgetError) Error() string {
	if e == nil {
		return "scheduled flow candidate scan exceeded its stop-loss budget"
	}
	return fmt.Sprintf(
		"scheduled flow candidate scan exceeded %d rows before reaching the matched-record cap",
		e.ScanCap,
	)
}

func (e *FlowExecutor) warnScheduledCandidateScan(outcome, objectName string, scannedCount, matchedCount int, sqlPrefilterApplied bool) {
	if e.logger == nil {
		return
	}
	e.logger.
		WithField("event", "scheduled_flow_candidate_scan").
		WithField("outcome", outcome).
		WithField("object_name", objectName).
		WithField("scanned_count", scannedCount).
		WithField("matched_count", matchedCount).
		WithField("scan_cap", scheduledFlowMaxScanned).
		WithField("matched_cap", scheduledFlowMaxMatched).
		WithField("sql_prefilter_applied", sqlPrefilterApplied).
		Warn("Scheduled flow candidate scan observation")
}

// statusCodeInvalidFlow is returned by executeFlowVersion when flow metadata
// fails pre-execution validation (see validateAtomicFlow).
const statusCodeInvalidFlow = "INVALID_FLOW"

// FlowStatusFault carries a deliberate FlowExecutionResult status code through
// nested event-bus / element-executor wrappers. It is used for flow metadata
// validation failures that are safe to report to callers as INVALID_FLOW rather
// than collapsing into the generic FLOW_ERROR sentinel.
type FlowStatusFault struct {
	StatusCode string
	Message    string
}

func (f *FlowStatusFault) Error() string {
	if f == nil {
		return ""
	}
	if strings.TrimSpace(f.Message) != "" {
		return f.Message
	}
	if strings.TrimSpace(f.StatusCode) != "" {
		return f.StatusCode
	}
	return "flow execution failed"
}

func (e *FlowExecutor) validateAtomicFlowForExecution(ctx context.Context, processType string, meta *models.FlowMetadata) error {
	return validateAtomicFlowForExecutionWithRegistry(ctx, processType, meta, e.flowActionCallRegistry())
}

// ValidateAtomicFlowForExecution proves atomic eligibility against this exact
// executor's installed action-call registry. Role-scoped Runtime compositions
// use it so admission cannot accept an action that the selected executor does
// not own.
func (e *FlowExecutor) ValidateAtomicFlowForExecution(
	ctx context.Context,
	processType string,
	meta *models.FlowMetadata,
) error {
	if ctx == nil {
		return apperrors.NewValidationError("runAtomically", "validation context is required")
	}
	return e.validateAtomicFlowForExecution(ctx, processType, meta)
}

// validateAtomicFlowForExecution is the ctx-aware validator used at runtime.
// A flow is "effectively atomic" when either:
//   - its own metadata declares RunAtomically=true, or
//   - it is executing under an inherited atomic parent boundary.
//
// The second case closes the child-flow hole where a non-atomic subflow with
// Wait / email I/O could still run inside the parent's open transaction.
func validateAtomicFlowForExecution(ctx context.Context, processType string, meta *models.FlowMetadata) error {
	registry, err := platformFlowActionCallRegistry(DefaultAppFlowActionCallRegistrations()...)
	if err != nil {
		return apperrors.NewValidationError("actionCall", err.Error())
	}
	return validateAtomicFlowForExecutionWithRegistry(ctx, processType, meta, registry)
}

// ValidateAtomicFlowForExecution exposes the existing app-blind atomic safety
// validator to private Runtime admission. Callers may add a stricter
// launch-profile proof, but must not duplicate or weaken the engine's action,
// Wait, trigger, and parent-prelock safety rules.
func ValidateAtomicFlowForExecution(
	ctx context.Context,
	processType string,
	meta *models.FlowMetadata,
) error {
	if ctx == nil {
		return apperrors.NewValidationError("runAtomically", "validation context is required")
	}
	return validateAtomicFlowForExecution(ctx, processType, meta)
}

func validateAtomicFlowForExecutionWithRegistry(ctx context.Context, processType string, meta *models.FlowMetadata, actionRegistry *flowActionCallRegistry) error {
	if meta == nil {
		return nil
	}
	effectiveAtomic := meta.RunAtomically || isWithinAtomicFlowBoundary(ctx)
	if len(meta.AtomicParentPrelocks) > 0 && !effectiveAtomic {
		return apperrors.NewValidationError("atomicParentPrelocks", "parent prelocks require atomic flow execution")
	}
	if len(meta.AtomicHistoryPrelocks) > 0 && !effectiveAtomic {
		return apperrors.NewValidationError("atomicHistoryPrelocks", "history prelocks require atomic flow execution")
	}
	if len(meta.AtomicHistoryPrelocks) > 0 && isBeforeSaveFlow(ctx) {
		return apperrors.NewValidationError("atomicHistoryPrelocks", "history prelocks are not supported for before-save flow execution inside an existing transaction")
	}
	if len(meta.AtomicHistoryPrelocks) > 0 && isWithinAtomicFlowBoundary(ctx) && !isPreparedFlowClosureRootExecution(ctx) {
		return apperrors.NewValidationError("atomicHistoryPrelocks", "history prelocks must be declared by the root atomic flow before its transaction starts")
	}
	if !effectiveAtomic {
		return nil
	}
	if err := validateFlowAtomicParentPrelocks(meta); err != nil {
		return err
	}
	if err := validateFlowAtomicHistoryPrelocks(meta); err != nil {
		return err
	}
	if isRecordTriggerRootExecution(ctx) {
		effectiveMeta := *meta
		effectiveMeta.RunAtomically = true
		if err := triggerprojection.ValidateAtomicTriggerSupport(
			&models.FlowDefinition{ProcessType: processType},
			&models.FlowVersion{ProcessType: processType, Metadata: effectiveMeta},
		); err != nil {
			return apperrors.NewValidationError("runAtomically", err.Error())
		}
	}
	if len(meta.Waits) > 0 {
		return apperrors.NewValidationError("runAtomically",
			"flows executing atomically cannot contain Wait elements; a paused interview would hold the database transaction open")
	}
	for _, action := range meta.ActionCalls {
		registration, registered := actionRegistry.Lookup(action.ActionType)
		if registered {
			switch registration.AtomicPolicy {
			case FlowActionCallAtomicPolicySafe:
				continue
			case FlowActionCallAtomicPolicyReject:
				return apperrors.NewValidationError("runAtomically",
					registration.atomicValidationError(&action))
			default:
				return apperrors.NewValidationError("runAtomically",
					registration.atomicValidationError(&action))
			}
		}
		if actionRegistry.actionCallUsesMetadataBackedAction(action.ActionType) {
			return apperrors.NewValidationError("runAtomically",
				fmt.Sprintf("flows executing atomically cannot contain metadata-backed action calls like %q; action rows can execute external I/O such as callWebhook outside the Flow atomic safety contract", action.ActionName))
		}
	}
	return nil
}

// validateIdentifier checks if a string is a safe identifier (prevents SQL injection)
func validateIdentifier(id string) error {
	if id == "" {
		return apperrors.NewValidationError("identifier", "identifier cannot be empty")
	}
	if len(id) > 255 {
		return apperrors.NewValidationError("identifier", "identifier too long")
	}
	if !identifierRegex.MatchString(id) {
		return apperrors.NewValidationError("identifier", fmt.Sprintf("invalid identifier format: %s", id))
	}
	return nil
}

// FlowExecutor is the core flow execution engine.
// Method implementations are split across domain-specific files:
//   - executor_elements.go    — element processing (Assignment, Decision, Loop, Screen, Wait, etc.)
//   - executor_record_ops.go  — record CRUD operations (Create, Update, Delete, Get, Lookup)
//   - executor_actions.go     — action invocations and subflow execution
//   - executor_helpers.go     — variable resolution, condition evaluation, formula helpers
type FlowExecutor struct {
	metadata      *metadata.MetadataService
	actions       *ActionService
	interviews    *FlowInterviewService
	persistence   PersistenceProvider
	query         ports.QueryService
	fieldMappings FieldMappingApplier
	logger        *logging.Logger

	// actionCallRegistry owns Flow actionCall handler lookup and the
	// engine-visible safety metadata used by atomic validation. Nil means the
	// built-in registry is used.
	actionCallRegistry *flowActionCallRegistry

	// permissions is injected post-construction via SetPermissionService.
	// Used by evaluateFilterFormulaGate to populate the $Permission formula
	// global for all users (system + custom permissions merged). May be nil
	// in tests that do not exercise $Permission formulas; callers must nil-guard.
	permissions PermissionProvider

	// approvalSubmitter is injected post-construction via SetApprovalSubmitter
	// because ApprovalService is constructed after this executor (it takes the
	// executor for notification flows). Nil means the SubmitForApproval
	// actionCall fails closed.
	approvalSubmitter approvalSubmitter

	// fieldReadAuthorizer is the invoking-user FLS gate for provider-neutral
	// outbound-intent payload construction. Nil means the action fails closed.
	fieldReadAuthorizer FieldReadAuthorizer

	// recordReadAuthorizer is the invoking-user object + record/RLS READ gate
	// for provider-neutral outbound-intent source rows. Nil means the action
	// fails closed.
	recordReadAuthorizer RecordReadAuthorizer

	// standardActionDeps exposes the neutral dialect/runtime contracts to
	// product-owned standard-action Flow handlers. Nil fields fail closed in the
	// product handler that requires them.
	standardActionDeps contract.Deps

	// flowVersionFn bypasses the FlowDefinition/Flow query path. Independent
	// Runtime composition binds it to the request-scoped immutable metadata
	// view; legacy monolith composition leaves it nil and uses QueryService.
	// Tests may point it at an in-memory pack.
	flowVersionFn func(ctx context.Context, apiName string, user *models.UserSession) (*models.FlowVersion, error)

	// flowPackOwnedFn is a test seam for the L1 standard-action internal
	// access boundary. Production determines ownership from FlowDefinition +
	// Flow.Source; unit tests can override it without standing up those rows.
	flowPackOwnedFn func(ctx context.Context, apiName string, user *models.UserSession) (bool, error)

	// connectorMappingResolverFn is a test seam for the applyConnectorFieldMappings
	// action. Production code leaves this nil and uses connectorMappingResolver;
	// unit tests override this field to inject effective mappings without a DB.
	connectorMappingResolverFn func(ctx context.Context, q connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error)

	// connectorMappingResolver is the production seam for direction-aware connector
	// field-mapping reads, set via WithConnectorMappingResolver at service-manager
	// wiring time. May be nil in tests that use connectorMappingResolverFn instead.
	connectorMappingResolver ConnectorMappingResolver

	// connectorPackMappingResolverFn is a test seam for trusted pack-owned
	// connector mapping reads. Production uses
	// connectorMappingResolver.ResolvePackConnectorMappings so pack-owned Flow
	// authority never consumes admin/effective mapping rows.
	connectorPackMappingResolverFn func(ctx context.Context, q connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error)

	// connectorDirectoryReader is the scheduled-flow connector-source seam.
	// Nil means buildConnectorDirectoryInputs fails closed.
	connectorDirectoryReader ConnectorDirectoryReader

	// connectorPushCallout and connectorPushProfiles are the synchronous
	// PushToConnector v1a runtime seams. Nil means the action fails closed.
	connectorPushProfiles         ConnectorPushProfileResolver
	connectorPushContracts        ConnectorPushContractResolver
	connectorInboundFinalizer     ConnectorInboundRemoteIdentityFinalizationAuthorityResolver
	connectorInboundInstalled     ConnectorInboundRemoteIdentityInstalledContractResolver
	connectorInboundRecoveryProbe ConnectorInboundRemoteIdentityRecoveryCandidateProbe
	connectorPushDurable          ConnectorPushDurableStore
	runtimeOutboundIntentSink     RuntimeOutboundIntentSink

	// connectorSourceRecordLoaderFn is a test seam for provider-neutral outbound
	// intents' committed current-row reload. Production leaves this nil and uses
	// persistence.GetRaw so outbound authority does not depend on mutable Flow
	// interview state.
	connectorSourceRecordLoaderFn func(ctx context.Context, objectName string, id string) (models.SObject, error)

	// fieldMetaLookup is a test seam used by resolveNestedReferenceCtx to
	// translate foreign-key fields to their referenced object type. Production
	// code leaves this nil and falls back to e.metadata.GetField; unit tests
	// override it to avoid wiring a full MetadataService.
	fieldMetaLookup func(string, string) *models.FieldMetadata

	// objectSchemaLookup is a test seam for lookupObjectSchema. Production code
	// leaves this nil and falls back to e.metadata.GetSchema; unit tests override
	// it to return a canned *models.ObjectMetadata without wiring a full
	// MetadataService (mirrors the fieldMetaLookup pattern).
	objectSchemaLookup func(ctx context.Context, objectAPIName string) *models.ObjectMetadata

	// beforeCreatePreviewSafeFn is a test seam for standard-action preview
	// discovery. Production proves safety from active before-create trigger
	// subscriptions for every object the action can create.
	beforeCreatePreviewSafeFn func(ctx context.Context, objectAPIName string, user *models.UserSession) (bool, error)

	// formulaEngine evaluates FlowFormula resources on demand. It is initialized
	// once per executor so normalized expressions, compiled programs, and
	// function tables are reused across all flow runs.
	formulaEngine *formula.Engine

	// recordLookupQueryGroup coalesces concurrent, identical before-save
	// RecordLookup reads. It does not cache beyond the in-flight DB query; each
	// caller receives a defensive clone. Keep the group behind a pointer because
	// preview executors intentionally share the parent's read machinery; copying
	// a live singleflight.Group would copy its mutex and split in-flight ownership.
	recordLookupQueryGroup *singleflight.Group

	// recordLookupBatcher coalesces concurrent Flow RecordLookup reads that have
	// the same metadata shape but different unique-key values. It is opt-in via
	// env and never owns data: batched reads still go through QueryService.
	recordLookupBatcher *flowRecordLookupBatcher

	// fieldDefaultResolver applies stage 3 (ActionFieldDefault) of the
	// recordCreate default pipeline. Lazy-initialized in NewFlowExecutor
	// against formulaEngine. Tests may override this field directly to
	// inject a fake resolver without standing up an action-field-default
	// row set on ctx.
	fieldDefaultResolver FieldDefaultResolver
}

// lookupFieldMeta returns FieldMetadata for (object, field), honoring the
// test override when set. Centralizing the access here keeps the production
// path unchanged when the override is nil.
func (e *FlowExecutor) lookupFieldMeta(ctx context.Context, objectAPIName, fieldAPIName string) *models.FieldMetadata {
	if e.fieldMetaLookup != nil {
		return e.fieldMetaLookup(objectAPIName, fieldAPIName)
	}
	if e.metadata == nil {
		return nil
	}
	schema := e.metadata.GetSchema(ctx, objectAPIName)
	return models.FindField(schema, fieldAPIName)
}

// lookupObjectSchema returns the full ObjectMetadata for objectAPIName,
// nil-tolerant for test executors and cold-start cache misses. Centralized
// so all callers that need schema-aware FK resolution (buildFormulaFetcher,
// resolveNestedReferenceCtx, resolveNestedReferenceCtxWithPresence) consult
// the metadata service the same way (PR #2057 round-6 F1).
//
// Returning nil is safe: every downstream caller of the schema-aware
// resolver path falls back to convention-only Tier 2/3/4 when schema is
// nil, preserving pre-#2057 automation behavior for executors without a
// wired metadata service.
func (e *FlowExecutor) lookupObjectSchema(ctx context.Context, objectAPIName string) *models.ObjectMetadata {
	if e == nil || objectAPIName == "" {
		return nil
	}
	if e.objectSchemaLookup != nil {
		return e.objectSchemaLookup(ctx, objectAPIName)
	}
	if e.metadata == nil {
		return nil
	}
	return e.metadata.GetSchema(ctx, objectAPIName)
}

// NewFlowExecutor creates a new FlowExecutor. fieldMappings may be nil;
// when a RecordCreate element declares a FieldMapping ref the executor
// returns a fault without the applier, preventing silent omissions.
//
// Trailing opts are applied in declaration order after the default fields
// are populated; see options.go for the public Option surface used by the
// L0 bridge package backend/pkg/automationtesting. Production callers pass
// no opts and get the same executor as before opts were introduced.
func NewFlowExecutor(
	metadataService *metadata.MetadataService,
	actions *ActionService,
	interviews *FlowInterviewService,
	persistence PersistenceProvider,
	query ports.QueryService,
	fieldMappings FieldMappingApplier,
	logger *logging.Logger,
	opts ...Option,
) *FlowExecutor {
	if logger == nil {
		logger = logging.New(context.Background())
	}
	eng := formula.NewEngine()
	e := &FlowExecutor{
		metadata:               metadataService,
		actions:                actions,
		interviews:             interviews,
		persistence:            persistence,
		query:                  query,
		fieldMappings:          fieldMappings,
		logger:                 logger,
		formulaEngine:          eng,
		recordLookupQueryGroup: &singleflight.Group{},
		recordLookupBatcher:    newFlowRecordLookupBatcherFromEnv(),
		fieldDefaultResolver:   NewDefaultFieldDefaultResolver(eng),
	}
	for _, opt := range opts {
		opt(e)
	}
	return e
}

// SetPermissionService wires a PermissionProvider into the executor
// post-construction. Required for $Permission formula global hydration for
// non-super-user flows (record-triggered and scheduled). When nil, $Permission
// resolves only for super-users via InjectSuperUserPermissions.
func (e *FlowExecutor) SetPermissionService(p PermissionProvider) {
	e.permissions = p
	if authorizer, ok := p.(FieldReadAuthorizer); ok {
		e.fieldReadAuthorizer = authorizer
	}
	if authorizer, ok := p.(RecordReadAuthorizer); ok {
		e.recordReadAuthorizer = authorizer
	}
}

// SetStandardActionContractDeps wires neutral contracts used by product-owned
// standard-action Flow handlers.
func (e *FlowExecutor) SetStandardActionContractDeps(deps contract.Deps) {
	e.standardActionDeps = deps
}

// StandardActionContractDeps returns the neutral contracts available to
// product-owned standard-action Flow handlers.
func (e *FlowExecutor) StandardActionContractDeps() contract.Deps {
	if e == nil {
		return contract.Deps{}
	}
	return e.standardActionDeps
}

// SetApprovalSubmitter wires the approval engine's Submit seam into the
// executor post-construction. Required for the SubmitForApproval actionCall;
// when nil the action faults instead of submitting.
func (e *FlowExecutor) SetApprovalSubmitter(s approvalSubmitter) {
	e.approvalSubmitter = s
}

// SetConnectorDirectoryReader wires the connector directory source reader for
// scheduled flows that pull their input sets from an external list endpoint.
// Called from main server wiring after connector_callout.Engine is available.
func (e *FlowExecutor) SetConnectorDirectoryReader(r ConnectorDirectoryReader) {
	e.connectorDirectoryReader = r
}

// SetConnectorPushProfileResolver supplies installed profile metadata for inbound identity recovery.
func (e *FlowExecutor) SetConnectorPushProfileResolver(profiles ConnectorPushProfileResolver) {
	e.connectorPushProfiles = profiles
}

func (e *FlowExecutor) SetConnectorPushContractResolver(resolver ConnectorPushContractResolver) {
	e.connectorPushContracts = resolver
}

// SetConnectorInboundRemoteIdentityFinalizationAuthorityResolver wires the
// immutable, request-bound authority used only by the no-HTTP finalization
// action. When present, finalization never falls back to the legacy split
// contract/profile/mapping read path.
func (e *FlowExecutor) SetConnectorInboundRemoteIdentityFinalizationAuthorityResolver(
	resolver ConnectorInboundRemoteIdentityFinalizationAuthorityResolver,
) {
	e.connectorInboundFinalizer = resolver
}

func (e *FlowExecutor) SetConnectorInboundRemoteIdentityInstalledContractResolver(resolver ConnectorInboundRemoteIdentityInstalledContractResolver) {
	e.connectorInboundInstalled = resolver
}

func (e *FlowExecutor) SetConnectorInboundRemoteIdentityRecoveryCandidateProbe(probe ConnectorInboundRemoteIdentityRecoveryCandidateProbe) {
	e.connectorInboundRecoveryProbe = probe
}

// SetConnectorPushDurableStore wires inbound identity evidence, recovery and
// historical attempt diagnostics.
func (e *FlowExecutor) SetConnectorPushDurableStore(store ConnectorPushDurableStore) {
	e.connectorPushDurable = store
}

var errAtomicFlowNeedsOwnedRollbackBoundary = apperrors.NewInternalError(
	"atomic flow requires an owned rollback boundary; nested reuse is only allowed inside an existing atomic flow",
	nil,
)

// ErrAtomicFlowDeadlineExceeded marks deadline expiry that occurred while a
// root FlowMetadata.RunAtomically flow was executing inside its owned rollback
// boundary. Standard action adapters use this typed proof for retry-safe
// timeout responses; plain context.DeadlineExceeded is not enough.
var ErrAtomicFlowDeadlineExceeded = errors.New("atomic flow execution deadline exceeded")

// ErrFlowExecutionBudgetExceeded proves that an execution did not reach a
// terminal element within the caller-authorized bound. It must escape as an
// error so the outer transaction owner rolls back every business effect.
var ErrFlowExecutionBudgetExceeded = errors.New("flow execution element budget exceeded")

const defaultFlowExecutionElementBudget uint64 = 1000

type flowExecutionBudgetKey struct{}

type flowExecutionBudget struct {
	remaining uint64
}

type preparedFlowDirectRecordOperationLimitKey struct{}

type preparedFlowRootAuthorizationRequiredKey struct{}

type preparedFlowClosureKey struct{}

type preparedFlowClosure map[string]*models.FlowVersion

// ErrPreparedFlowClosureMissing proves that an immutable Runtime invocation
// attempted to leave the exact Flow closure selected for its Catalog revision.
// It must fail closed instead of reopening the live active-version read path.
var ErrPreparedFlowClosureMissing = errors.New("prepared Flow closure has no exact version")

func withPreparedFlowClosure(
	ctx context.Context,
	versions map[string]*models.FlowVersion,
) (context.Context, error) {
	if ctx == nil || len(versions) == 0 {
		return nil, ErrPreparedFlowClosureMissing
	}
	closure := make(preparedFlowClosure, len(versions))
	for name, version := range versions {
		key := strings.ToLower(strings.TrimSpace(name))
		if key == "" || version == nil || version.Id == "" {
			return nil, ErrPreparedFlowClosureMissing
		}
		if existing, ok := closure[key]; ok && existing.Id != version.Id {
			return nil, ErrPreparedFlowClosureMissing
		}
		closure[key] = version
	}
	return context.WithValue(ctx, preparedFlowClosureKey{}, closure), nil
}

func preparedFlowVersion(
	ctx context.Context,
	flowAPIName string,
) (*models.FlowVersion, bool, bool) {
	closure, present := ctx.Value(preparedFlowClosureKey{}).(preparedFlowClosure)
	if !present || closure == nil {
		return nil, false, false
	}
	version, found := closure[strings.ToLower(strings.TrimSpace(flowAPIName))]
	return version, found, true
}

func hasPreparedFlowClosure(ctx context.Context) bool {
	closure, present := ctx.Value(preparedFlowClosureKey{}).(preparedFlowClosure)
	return present && closure != nil
}

// withoutPreparedFlowClosure starts an independently selected Flow root while
// preserving the caller's transaction, cancellation, and execution budgets.
// It is only valid after the new root has been resolved from the same
// request-bound immutable Runtime view.
func withoutPreparedFlowClosure(ctx context.Context) context.Context {
	return context.WithValue(ctx, preparedFlowClosureKey{}, preparedFlowClosure(nil))
}

func isPreparedFlowClosureRootExecution(ctx context.Context) bool {
	return hasPreparedFlowClosure(ctx) && hasAtomicFlowPessimisticRetryOwner(ctx) && flowRuntimeContextFromCtx(ctx).CallStackDepth() == 0
}

func withFlowExecutionBudget(ctx context.Context, maximum uint64) (context.Context, error) {
	if ctx == nil || maximum == 0 || maximum > defaultFlowExecutionElementBudget {
		return nil, ErrFlowExecutionBudgetExceeded
	}
	if existing, _ := ctx.Value(flowExecutionBudgetKey{}).(*flowExecutionBudget); existing != nil {
		return ctx, nil
	}
	return context.WithValue(ctx, flowExecutionBudgetKey{}, &flowExecutionBudget{remaining: maximum}), nil
}

func consumeFlowExecutionBudget(ctx context.Context) error {
	budget, _ := ctx.Value(flowExecutionBudgetKey{}).(*flowExecutionBudget)
	if budget == nil || budget.remaining == 0 {
		return ErrFlowExecutionBudgetExceeded
	}
	budget.remaining--
	return nil
}

func withPreparedFlowDirectRecordOperationLimit(
	ctx context.Context,
	maximum int,
) (context.Context, error) {
	if ctx == nil || maximum <= 0 || maximum > MaxAtomicFilteredRecordOperationRows {
		return nil, ErrFlowExecutionBudgetExceeded
	}
	return context.WithValue(
		ctx,
		preparedFlowDirectRecordOperationLimitKey{},
		maximum,
	), nil
}

func preparedFlowDirectRecordOperationLimit(ctx context.Context) (int, bool) {
	limit, ok := ctx.Value(preparedFlowDirectRecordOperationLimitKey{}).(int)
	return limit, ok && limit > 0
}

func withPreparedFlowRootAuthorizationRequired(ctx context.Context) context.Context {
	return context.WithValue(
		ctx,
		preparedFlowRootAuthorizationRequiredKey{},
		true,
	)
}

func preparedFlowRootAuthorizationRequired(ctx context.Context) bool {
	required, _ := ctx.Value(
		preparedFlowRootAuthorizationRequiredKey{},
	).(bool)
	return required
}

// runPossiblyAtomic invokes fn directly or wraps it in
// persistence.RunInPessimisticTransaction. Atomic flows can execute generic
// record CRUD that reaches rollup parent locking, so the outer flow wrapper
// must establish a pessimistic transaction rather than a plain retry tx.
//
// Reuse is intentionally narrower than "any active transaction": only a
// caller that is already inside an atomic flow boundary OR a before-save
// trigger root that owns the in-flight record mutation boundary may reuse the
// current transaction. Any other pre-existing transaction is a foreign
// rollback boundary; returning a failed FlowExecutionResult with nil error
// would let that outer transaction commit partial DML if the caller does not
// rethrow. Fail closed instead of silently reusing that transaction.
func (e *FlowExecutor) runPossiblyAtomic(ctx context.Context, atomic bool, fn func(context.Context) error) error {
	_, err := e.runPossiblyAtomicWithCompletion(ctx, atomic, fn)
	return err
}

type flowPessimisticTransactionOutcomeRunner interface {
	RunInPessimisticTransactionWithOutcome(
		context.Context,
		func(context.Context) error,
	) (ports.TransactionCompletionOutcome, error)
}

type atomicFlowExecutionBoundary struct {
	ownsRollbackBoundary bool
	completion           ports.TransactionCompletionOutcome
}

func (e *FlowExecutor) runPossiblyAtomicWithCompletion(
	ctx context.Context,
	atomic bool,
	fn func(context.Context) error,
) (atomicFlowExecutionBoundary, error) {
	if !atomic {
		return atomicFlowExecutionBoundary{}, fn(ctx)
	}
	if isWithinAtomicFlowBoundary(ctx) {
		return atomicFlowExecutionBoundary{}, fn(withAtomicFlowBoundary(ctx))
	}
	if e.persistence.IsTransactionActive(ctx) {
		if isBeforeSaveFlow(ctx) {
			return atomicFlowExecutionBoundary{}, fn(withAtomicFlowBoundary(ctx))
		}
		return atomicFlowExecutionBoundary{}, errAtomicFlowNeedsOwnedRollbackBoundary
	}
	if scoper, ok := e.connectorInboundInstalled.(ConnectorInboundRemoteIdentityInstalledContractRetryScoper); ok {
		ctx = scoper.WithConnectorInboundRemoteIdentityInstalledContractRetryScope(ctx)
	}
	runAttempt := func(txCtx context.Context) error {
		// This marker is established inside every transaction-manager callback,
		// including retries. Nested atomic flows inherit it; before-save and
		// foreign ambient transactions deliberately do not.
		txCtx = withAtomicFlowPessimisticRetryOwner(txCtx)
		// Reset the post-upsert registry per atomic-transaction ATTEMPT. This
		// callback is re-run by WithPessimisticRetry (transaction_manager.go) after
		// rolling back a lock-contention attempt, and the registry is process-memory
		// on FlowRuntimeContext (not tx-scoped). Without a fresh registry per attempt,
		// a rolled-back attempt's published recordId would survive into the retry and
		// mislead registerConnectorInboundRemoteAuthorship (#2080, IANTHEREAL review):
		// it would see a stale/ambiguous identity instead of only this attempt's.
		attemptFrc := flowRuntimeContextFromCtx(txCtx).CloneWithFreshPostUpsert()
		txCtx = withFlowRuntimeContext(txCtx, attemptFrc)
		return fn(withAtomicFlowBoundary(txCtx))
	}
	if runner, ok := e.persistence.(flowPessimisticTransactionOutcomeRunner); ok && runner != nil {
		completion, err := runner.RunInPessimisticTransactionWithOutcome(ctx, runAttempt)
		return atomicFlowExecutionBoundary{ownsRollbackBoundary: true, completion: completion}, err
	}
	return atomicFlowExecutionBoundary{ownsRollbackBoundary: true}, e.persistence.RunInPessimisticTransaction(ctx, runAttempt)
}

func atomicFlowDeadlineError(rootRunAtomically bool, err error) error {
	if errors.Is(err, context.Canceled) {
		return context.Canceled
	}
	if rootRunAtomically && errors.Is(err, context.DeadlineExceeded) {
		return fmt.Errorf("%w: %w", ErrAtomicFlowDeadlineExceeded, err)
	}
	return nil
}

type flowAttemptRecordReadCacheKey struct{}

// withFlowAttemptRecordReadCache gives one root Flow execution attempt a fresh
// records-layer raw-read cache. Nested subflows inherit that attempt cache so
// child writes invalidate the same cache the parent will read later; retry
// attempts call this from a fresh callback context and therefore get a new
// cache after rollback.
func withFlowAttemptRecordReadCache(ctx context.Context) context.Context {
	if active, _ := ctx.Value(flowAttemptRecordReadCacheKey{}).(bool); active {
		return ctx
	}
	ctx = records.WithRecordReadCache(ctx)
	return context.WithValue(ctx, flowAttemptRecordReadCacheKey{}, true)
}

func atomicFlowDeadlineContextError(atomic bool, ctx context.Context) error {
	if ctx == nil {
		return nil
	}
	return atomicFlowDeadlineError(atomic, ctx.Err())
}

type flowExecutionOutcomeCollectorKey struct{}

type flowExecutionOutcomeCollector struct {
	executionID          string
	preflightExecutionID bool
	cause                *contractmodels.FlowExecutionCause
	dependency           *contractmodels.FlowExecutionDependencyRequest
}

type flowExecutionOutcomeRootKey struct{}

type flowAttemptCauseCollectorKey struct{}

type flowAttemptCauseCollector struct {
	cause       *contractmodels.FlowExecutionCause
	faultTarget string
	terminal    bool
}

type flowAttemptDependencyCollectorKey struct{}

type flowAttemptDependencyCollector struct {
	dependency *contractmodels.FlowExecutionDependencyRequest
}

func withFlowExecutionOutcomeCollector(ctx context.Context) (context.Context, *flowExecutionOutcomeCollector) {
	collector := &flowExecutionOutcomeCollector{}
	return context.WithValue(ctx, flowExecutionOutcomeCollectorKey{}, collector), collector
}

func publishFlowExecutionDependency(ctx context.Context, dependency *contractmodels.FlowExecutionDependencyRequest) {
	collector, _ := ctx.Value(flowExecutionOutcomeCollectorKey{}).(*flowExecutionOutcomeCollector)
	if collector == nil || dependency == nil {
		return
	}
	copy := *dependency
	collector.dependency = &copy
}

func withFlowAttemptDependencyCollector(ctx context.Context) (context.Context, *flowAttemptDependencyCollector) {
	collector := &flowAttemptDependencyCollector{}
	return context.WithValue(ctx, flowAttemptDependencyCollectorKey{}, collector), collector
}

func publishFlowAttemptDependency(ctx context.Context, dependency contractmodels.FlowExecutionDependencyRequest) error {
	if err := dependency.Validate(); err != nil {
		return err
	}
	collector, _ := ctx.Value(flowAttemptDependencyCollectorKey{}).(*flowAttemptDependencyCollector)
	if collector == nil {
		return fmt.Errorf("flow dependency collector is unavailable")
	}
	if collector.dependency != nil {
		if *collector.dependency == dependency {
			return nil
		}
		return fmt.Errorf("flow execution declared conflicting durable dependencies")
	}
	copy := dependency
	collector.dependency = &copy
	return nil
}

func withFlowExecutionOutcomeRoot(ctx context.Context) (context.Context, bool) {
	if ctx.Value(flowExecutionOutcomeCollectorKey{}) == nil {
		return ctx, false
	}
	if active, _ := ctx.Value(flowExecutionOutcomeRootKey{}).(bool); active {
		return ctx, false
	}
	return context.WithValue(ctx, flowExecutionOutcomeRootKey{}, true), true
}

func publishFlowExecutionID(ctx context.Context, executionID string) {
	collector, _ := ctx.Value(flowExecutionOutcomeCollectorKey{}).(*flowExecutionOutcomeCollector)
	executionID = strings.TrimSpace(executionID)
	if collector == nil || executionID == "" || (collector.executionID != "" && !collector.preflightExecutionID) {
		return
	}
	collector.executionID = executionID
	collector.preflightExecutionID = false
}

// reserveFlowExecutionID keeps a terminal connector outcome durable even when
// execution fails before an interview exists (for example missing, invalid,
// or protected roots). When execution reaches interview creation, that real
// interview ID replaces this preflight-only correlation ID.
func reserveFlowExecutionID(ctx context.Context) error {
	collector, _ := ctx.Value(flowExecutionOutcomeCollectorKey{}).(*flowExecutionOutcomeCollector)
	if collector == nil || collector.executionID != "" {
		return nil
	}
	id, err := newFlowExecutionID()
	if err != nil {
		return err
	}
	collector.executionID = id
	collector.preflightExecutionID = true
	return nil
}

func publishFlowExecutionCause(ctx context.Context, cause *contractmodels.FlowExecutionCause) {
	collector, _ := ctx.Value(flowExecutionOutcomeCollectorKey{}).(*flowExecutionOutcomeCollector)
	if collector == nil || cause == nil || collector.cause != nil {
		return
	}
	copy := *cause
	collector.cause = &copy
}

func withFlowAttemptCauseCollector(ctx context.Context) (context.Context, *flowAttemptCauseCollector) {
	if collector, _ := ctx.Value(flowAttemptCauseCollectorKey{}).(*flowAttemptCauseCollector); collector != nil {
		return ctx, collector
	}
	collector := &flowAttemptCauseCollector{}
	return context.WithValue(ctx, flowAttemptCauseCollectorKey{}, collector), collector
}

// publishFlowAttemptFaultCause records the cause currently being presented by
// a fault branch. A terminal child error may have already been classified
// before its parent routes it to a fault connector; retain that richer child
// cause while marking it pending presentation. Otherwise a later fault route
// supersedes an earlier recovered one.
func publishFlowAttemptFaultCause(ctx context.Context, cause *contractmodels.FlowExecutionCause, faultTarget string) {
	collector, _ := ctx.Value(flowAttemptCauseCollectorKey{}).(*flowAttemptCauseCollector)
	if collector == nil || cause == nil {
		return
	}
	if collector.cause == nil || !collector.terminal {
		copy := *cause
		collector.cause = &copy
	}
	collector.faultTarget = strings.TrimSpace(faultTarget)
	collector.terminal = false
}

func publishTerminalFlowAttemptCause(ctx context.Context, cause *contractmodels.FlowExecutionCause) {
	collector, _ := ctx.Value(flowAttemptCauseCollectorKey{}).(*flowAttemptCauseCollector)
	if collector == nil || cause == nil {
		return
	}
	copy := *cause
	collector.cause = &copy
	collector.faultTarget = ""
	collector.terminal = true
}

// publishFailedChildFlowResultCause supplies typed evidence for a child flow
// that returned a legacy failed result before execution. Do not replace an
// already terminal cause: an executed child has richer native error evidence
// than its compatibility FlowExecutionResult can carry.
func publishFailedChildFlowResultCause(ctx context.Context, cause *contractmodels.FlowExecutionCause) {
	collector, _ := ctx.Value(flowAttemptCauseCollectorKey{}).(*flowAttemptCauseCollector)
	if collector != nil && collector.cause != nil && collector.terminal {
		return
	}
	publishTerminalFlowAttemptCause(ctx, cause)
	publishTerminalFlowExecutionCause(ctx, cause)
}

func publishTerminalFlowExecutionCause(ctx context.Context, cause *contractmodels.FlowExecutionCause) {
	collector, _ := ctx.Value(flowExecutionOutcomeCollectorKey{}).(*flowExecutionOutcomeCollector)
	if collector == nil || cause == nil {
		return
	}
	copy := *cause
	collector.cause = &copy
}

func buildFlowExecutionOutcome(collector *flowExecutionOutcomeCollector, result *models.FlowExecutionResult, err error) *contractmodels.FlowExecutionTypedOutcome {
	// A fault connector can deliberately recover from a child-flow failure.
	// The collector is shared across that child and its root, so it can retain
	// the original cause even though the root's business result is successful.
	// Typed failure evidence is terminal-only: never attach it to a success.
	if err == nil && result != nil && result.IsSuccess {
		return &contractmodels.FlowExecutionTypedOutcome{
			Result:      result,
			ExecutionID: collector.executionID,
			Dependency:  collector.dependency,
		}
	}
	if collector.cause == nil {
		if err != nil {
			publishFlowExecutionCauseFromCollector(collector, classifyFlowExecutionCause(err, "", ""))
		} else {
			publishFlowExecutionCauseFromCollector(collector, classifyFailedFlowExecutionResult(result, flowExecutionResultErrorOriginUnknown))
		}
	}
	return &contractmodels.FlowExecutionTypedOutcome{
		Result:      result,
		ExecutionID: collector.executionID,
		Cause:       collector.cause,
		Dependency:  collector.dependency,
	}
}

func publishFlowExecutionCauseFromCollector(collector *flowExecutionOutcomeCollector, cause *contractmodels.FlowExecutionCause) {
	if collector == nil || collector.cause != nil || cause == nil {
		return
	}
	copy := *cause
	collector.cause = &copy
}

// ExecuteFlowOutcome preserves the public FlowExecutionResult shape while
// carrying connector-only typed evidence through a direct-Go envelope.
func (e *FlowExecutor) ExecuteFlowOutcome(ctx context.Context, flowApiName string, inputs map[string]interface{}, user *models.UserSession) (*contractmodels.FlowExecutionTypedOutcome, error) {
	ctx, collector := withFlowExecutionOutcomeCollector(ctx)
	if err := reserveFlowExecutionID(ctx); err != nil {
		return nil, err
	}
	result, err := e.ExecuteFlow(ctx, flowApiName, inputs, user)
	return buildFlowExecutionOutcome(collector, result, err), err
}

// ExecuteFlow executes a flow by API name (Actions API entry point)
func (e *FlowExecutor) ExecuteFlow(ctx context.Context, flowApiName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	// Design §3.5: stamp Entrypoint=FLOW onto ctx so any record mutations
	// performed downstream (RecordCreate/Update/Delete elements) can recover
	// the correct AuditEvent.Entrypoint via audit.EntrypointFromContext(ctx).
	// The emitter still requires callers to set Entrypoint on the AuditEvent
	// struct explicitly (§3.1 CH-4) — this is transport only.
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)

	// Private Runtime invocations carry the complete exact Flow closure selected
	// from one immutable Catalog revision. Descendants must stay inside that
	// closure and must never reopen the live active-version read path.
	flowVersion, foundPrepared, hasPreparedClosure := preparedFlowVersion(ctx, flowApiName)
	if hasPreparedClosure && !foundPrepared {
		return nil, fmt.Errorf("%w: %q", ErrPreparedFlowClosureMissing, flowApiName)
	}
	if !hasPreparedClosure {
		var err error
		flowVersion, err = e.getActiveFlowVersion(ctx, flowApiName, user)
		if err != nil {
			return nil, err
		}
	}
	if flowVersion == nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("Flow '%s' not found or not active", flowApiName),
			}},
		}, nil
	}
	if !hasPreparedClosure || preparedFlowRootAuthorizationRequired(ctx) {
		var result *models.FlowExecutionResult
		ctx, result = e.authorizeProtectedStandardFlow(ctx, flowApiName, flowVersion)
		if result != nil {
			return result, nil
		}
	}

	return e.executeFlowVersion(ctx, flowVersion, flowApiName, inputs, user)
}

func (e *FlowExecutor) executeApprovalNotificationFlowVersion(ctx context.Context, flowApiName string, flowVersion *models.FlowVersion, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)
	if flowVersion == nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("Flow '%s' not found or not active", flowApiName),
			}},
		}, nil
	}
	// ApprovalService validates the pinned Flow version as notification-safe
	// before it reaches this entrypoint. That constrained bridge may execute
	// pack-owned standard notification Flows without the standard-action or
	// connector authorization markers that authorizeProtectedStandardFlow
	// requires for generic ExecuteFlow calls.
	return e.executeFlowVersion(ctx, flowVersion, flowApiName, inputs, user)
}

func (e *FlowExecutor) authorizeProtectedStandardFlow(ctx context.Context, flowApiName string, flowVersion *models.FlowVersion) (context.Context, *models.FlowExecutionResult) {
	if !flowVersionIsPackOwnedStandard(flowVersion) {
		return ctx, nil
	}
	if constants.ConnectorInboundExecutionAuthorizesFlow(ctx, flowApiName, flowVersion.DefinitionOwnerPack) {
		// Connector-inbound worker may execute exactly the routed root flow as a
		// protected pack-owned standard flow, but ONLY when the flow is owned by
		// the same pack that declared the connector route (same-pack check via
		// DefinitionOwnerPack). Do NOT stamp standard-action markers
		// (previewRecordUpdate / stampConvertedAnchors stay locked), and do NOT
		// authorize protected descendants: a subflow has a different flowApiName,
		// so this match fails for it. Cross-pack routes (pack A routes to pack
		// B's protected standard flow) are rejected because DefinitionOwnerPack
		// will not match the routing namespace stamped on ctx.
		return ctx, nil
	}
	// Cross-pack companion execution: the worker stamped the frozen Flow.Id after
	// re-verifying the companion incarnation (§5.2 of design). Authorization is
	// keyed on the immutable flowVersion.Id — a subflow has a DIFFERENT Id than
	// the frozen root so this check fails for descendants (subflow protection
	// preserved). Consent was checked at enqueue; this drain path is enqueue-only.
	if constants.ConnectorInboundCompanionExecutionAuthorizesVersion(ctx, flowVersion.Id) {
		return ctx, nil
	}
	if connectorInboundCompanionSubflowExecutionAuthorized(ctx, FlowExecutionIdentity{
		APIName:                flowApiName,
		DefinitionID:           flowVersion.DefinitionId,
		VersionID:              flowVersion.Id,
		OwnerPack:              flowVersion.DefinitionOwnerPack,
		DefinitionIsCustomized: flowVersion.DefinitionIsCustomized,
		VersionSource:          flowVersion.Source,
	}) {
		return ctx, nil
	}
	// A strict durable AsyncAfterCommit handoff reopens the exact immutable
	// record-trigger root version in the independent worker. The binder mints
	// this version-scoped marker only after validating that scheduled-plan
	// identity. Accept the frozen root without stamping the broader standard
	// action marker: a protected child has a different version ID and remains
	// denied, and standard-action-only actions remain locked.
	if AfterCommitScheduledFlowExecutionAuthorizesVersion(ctx, flowVersion.Id) {
		return ctx, nil
	}
	if constants.HasStandardActionProtectedFlowExecution(ctx) {
		return ctx, nil
	}
	if constants.StandardActionExecutionOwnsRootFlow(ctx, flowApiName) {
		return constants.WithStandardActionProtectedFlowExecution(ctx, flowApiName), nil
	}
	return ctx, &models.FlowExecutionResult{
		ActionName: flowApiName,
		IsSuccess:  false,
		Errors: []models.FlowExecutionError{{
			StatusCode: "FLOW_EXECUTION_ERROR",
			Message:    fmt.Sprintf("Flow %q is protected and can only run through its standard action", flowApiName),
		}},
	}
}

// ExecuteFlowVersionById executes a companion app-pack flow by its exact frozen
// Flow.Id (companionFlowVersionID). This is the cross-pack drain path:
//
//  1. Load the exact version by id via getFlowVersionById (never by name).
//  2. Hydrate definition runtime fields (OwnerPack, IsCustomized) via
//     ensureFlowVersionDefinitionRuntimeFields.
//  3. Authorize via authorizeProtectedStandardFlow (ctx must carry the companion
//     marker stamped by the worker BEFORE calling this method).
//  4. Execute via executeFlowVersion.
//
// Returns a NOT_FOUND terminal result (not nil/error) when the version is gone.
func (e *FlowExecutor) ExecuteFlowVersionById(ctx context.Context, flowDeveloperName, companionFlowVersionID string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)

	v, err := e.getFlowVersionById(ctx, companionFlowVersionID, user)
	if err != nil {
		return nil, err
	}
	if v == nil {
		return &models.FlowExecutionResult{
			ActionName: flowDeveloperName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("companion flow version %q gone (flow %q)", companionFlowVersionID, flowDeveloperName),
			}},
		}, nil
	}

	// Hydrate DefinitionOwnerPack, DefinitionIsCustomized, DefinitionRuntimeLoaded
	// — getFlowVersionById does NOT set these (it only queries the Flow table).
	apiName, err := e.ensureFlowVersionDefinitionRuntimeFields(ctx, v, flowDeveloperName, user)
	if err != nil {
		return nil, err
	}
	if apiName == "" {
		apiName = flowDeveloperName
	}

	var authResult *models.FlowExecutionResult
	ctx, authResult = e.authorizeProtectedStandardFlow(ctx, apiName, v)
	if authResult != nil {
		return authResult, nil
	}
	return e.executePreparedFlowVersion(ctx, apiName, v, inputs, user)
}

// executePreparedFlowVersion executes one exact FlowVersion selected by an
// immutable metadata authority. It never reopens Flow or FlowDefinition rows
// to choose a different version.
func (e *FlowExecutor) executePreparedFlowVersion(
	ctx context.Context,
	flowDeveloperName string,
	version *models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
) (*models.FlowExecutionResult, error) {
	if version == nil {
		return &models.FlowExecutionResult{
			ActionName: flowDeveloperName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("prepared flow version missing (flow %q)", flowDeveloperName),
			}},
		}, nil
	}
	return e.executeFlowVersion(ctx, version, flowDeveloperName, inputs, user)
}

// ExecuteFlowVersionByIdOutcome is the exact-version connector companion path
// with the same direct-Go outcome envelope as ExecuteFlowOutcome.
func (e *FlowExecutor) ExecuteFlowVersionByIdOutcome(ctx context.Context, flowDeveloperName, companionFlowVersionID string, inputs map[string]interface{}, user *models.UserSession) (*contractmodels.FlowExecutionTypedOutcome, error) {
	ctx, collector := withFlowExecutionOutcomeCollector(ctx)
	if err := reserveFlowExecutionID(ctx); err != nil {
		return nil, err
	}
	result, err := e.ExecuteFlowVersionById(ctx, flowDeveloperName, companionFlowVersionID, inputs, user)
	return buildFlowExecutionOutcome(collector, result, err), err
}

// ExecutePreparedFlowClosureWithinOwnedTransactionOutcome is the private
// Runtime adapter seam for the complete exact Flow closure selected from one
// request-bound immutable metadata view. The caller owns the pessimistic
// transaction and retry callback. Transient interviews keep the durable
// invocation receipt as the sole execution/result authority.
func (e *FlowExecutor) ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	flowDeveloperName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
	maxElementExecutions uint64,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)
	ctx = WithTransientFlowInterviews(ctx)
	ctx = withAtomicFlowBoundary(withAtomicFlowPessimisticRetryOwner(ctx))
	ctx, collector := withFlowExecutionOutcomeCollector(ctx)
	if err := reserveFlowExecutionID(ctx); err != nil {
		return nil, err
	}
	var err error
	ctx, err = withPreparedFlowClosure(ctx, versions)
	if err != nil {
		return nil, err
	}
	ctx, err = withFlowExecutionBudget(ctx, maxElementExecutions)
	if err != nil {
		return nil, err
	}
	version, found, _ := preparedFlowVersion(ctx, flowDeveloperName)
	if !found {
		return nil, fmt.Errorf("%w: root %q", ErrPreparedFlowClosureMissing, flowDeveloperName)
	}
	if preparedFlowRootAuthorizationRequired(ctx) {
		var authorizationFailure *models.FlowExecutionResult
		ctx, authorizationFailure = e.authorizeProtectedStandardFlow(
			ctx,
			flowDeveloperName,
			version,
		)
		if authorizationFailure != nil {
			return buildFlowExecutionOutcome(
				collector,
				authorizationFailure,
				nil,
			), nil
		}
	}
	result, err := e.executePreparedFlowVersion(ctx, flowDeveloperName, version, inputs, user)
	return buildFlowExecutionOutcome(collector, result, err), err
}

// ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome adds the
// Platform-command execution profile. Catalog has already authorized the exact
// root/version. The profile accepts only native, directly declared, same-owner
// standard child Flows. The existing connector-companion runtime delegation is
// replaced at every native Subflow boundary, so a child can consume the root's
// exact-version authority but cannot pass it to a grandchild. The complete
// closure executes under one caller-owned transaction and one shared element
// budget.
//
// The protected-Flow gate is consumed for the root and each direct child. A
// cross-pack companion root is authorized by its exact version marker while an
// unrelated protected root or child fails before any Flow element executes.
// The ordinary prepared entrypoint remains unchanged for the synchronous
// Runtime API and scheduled Flow roles.
func (e *FlowExecutor) ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	flowDeveloperName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
	maxElementExecutions uint64,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	if err := validateBoundedPlatformCommandFlowClosure(
		flowDeveloperName,
		versions,
	); err != nil {
		return nil, err
	}
	return e.ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		flowDeveloperName,
		versions,
		inputs,
		user,
		maxElementExecutions,
	)
}

func validateBoundedPlatformCommandFlowClosure(
	flowDeveloperName string,
	versions map[string]*models.FlowVersion,
) error {
	rootName := strings.ToLower(strings.TrimSpace(flowDeveloperName))
	if rootName == "" || len(versions) == 0 {
		return errors.New("bounded Platform-command Flow has no exact authorized root")
	}
	normalized := make(map[string]*models.FlowVersion, len(versions))
	for definitionRef, version := range versions {
		name := strings.ToLower(strings.TrimSpace(definitionRef))
		if name == "" || version == nil {
			return errors.New("bounded Platform-command Flow closure is invalid")
		}
		if existing, found := normalized[name]; found && existing.Id != version.Id {
			return errors.New("bounded Platform-command Flow closure is ambiguous")
		}
		normalized[name] = version
	}
	root, rootPresent := normalized[rootName]
	if !rootPresent {
		return fmt.Errorf(
			"bounded Platform-command Flow has no exact authorized root %q",
			flowDeveloperName,
		)
	}

	directChildren := make(map[string]struct{}, len(root.Metadata.Subflows))
	for _, action := range root.Metadata.ActionCalls {
		if strings.EqualFold(strings.TrimSpace(action.ActionType), "flow") {
			return errors.New(
				"bounded Platform-command Flow has no one-hop same-owner native child authority for Flow actions",
			)
		}
	}
	for _, subflow := range root.Metadata.Subflows {
		childName := strings.ToLower(strings.TrimSpace(subflow.FlowName))
		if childName == "" {
			return errors.New(
				"bounded Platform-command Flow has an invalid native child declaration",
			)
		}
		directChildren[childName] = struct{}{}
	}
	if len(directChildren) == 0 {
		if len(normalized) != 1 {
			return errors.New(
				"bounded Platform-command Flow closure exceeds one-hop same-owner native child authority",
			)
		}
		return nil
	}
	if !boundedPlatformCommandPackOwnedStandardVersion(root) {
		return errors.New(
			"bounded Platform-command Flow root cannot delegate one-hop same-owner native child authority",
		)
	}
	rootOwner := strings.TrimSpace(root.DefinitionOwnerPack)
	for childName := range directChildren {
		child, found := normalized[childName]
		if !found || !boundedPlatformCommandPackOwnedStandardVersion(child) ||
			!strings.EqualFold(rootOwner, strings.TrimSpace(child.DefinitionOwnerPack)) {
			return errors.New(
				"bounded Platform-command Flow closure exceeds one-hop same-owner native child authority",
			)
		}
		if len(child.Metadata.Subflows) != 0 {
			return errors.New(
				"bounded Platform-command Flow child cannot delegate transitive one-hop same-owner authority",
			)
		}
		for _, action := range child.Metadata.ActionCalls {
			if strings.EqualFold(strings.TrimSpace(action.ActionType), "flow") {
				return errors.New(
					"bounded Platform-command Flow child cannot delegate transitive one-hop same-owner authority",
				)
			}
		}
	}
	for name := range normalized {
		if name == rootName {
			continue
		}
		if _, declared := directChildren[name]; !declared {
			return errors.New(
				"bounded Platform-command Flow closure exceeds one-hop same-owner native child authority",
			)
		}
	}
	return nil
}

func boundedPlatformCommandPackOwnedStandardVersion(
	version *models.FlowVersion,
) bool {
	return version != nil && version.DefinitionRuntimeLoaded &&
		flowVersionIsPackOwnedStandard(version) &&
		strings.TrimSpace(version.DefinitionOwnerPack) != ""
}

// ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome adds
// the existing direct-record governor to an exact, already-authorized prepared
// closure. Unlike the Platform-command profile, this entrypoint permits
// Catalog-authorized subflows; every version still comes from the immutable
// closure selected before the transaction and no live Flow lookup is allowed.
func (e *FlowExecutor) ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	flowDeveloperName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
	maxElementExecutions uint64,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	ctx, err := withPreparedFlowDirectRecordOperationLimit(
		ctx,
		MaxAtomicFilteredRecordOperationRows,
	)
	if err != nil {
		return nil, err
	}
	ctx = withPreparedFlowRootAuthorizationRequired(ctx)
	return e.ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx,
		flowDeveloperName,
		versions,
		inputs,
		user,
		maxElementExecutions,
	)
}

func (e *FlowExecutor) executeFlowVersion(ctx context.Context, flowVersion *models.FlowVersion, flowApiName string, inputs map[string]interface{}, user *models.UserSession) (*models.FlowExecutionResult, error) {
	var outcomeRoot bool
	ctx, outcomeRoot = withFlowExecutionOutcomeRoot(ctx)
	flowStarted := time.Now()
	if flowVersion == nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: "NOT_FOUND",
				Message:    fmt.Sprintf("Flow '%s' not found or not active", flowApiName),
			}},
		}, nil
	}
	var identityErr error
	flowApiName, identityErr = e.ensureFlowVersionDefinitionRuntimeFields(ctx, flowVersion, flowApiName, user)
	if identityErr != nil {
		return nil, identityErr
	}
	ctx = withCurrentFlowExecution(ctx, flowApiName, flowVersion)
	defer func() {
		logFlowTiming(ctx, flowApiName, flowVersion.Id, "", "Flow", "", "finished", "", flowStarted, nil)
	}()

	// Use hydrated metadata (HydrateMetadata is called by the data layer)
	flowMeta := flowVersion.Metadata

	if err := e.validateAtomicFlowForExecution(ctx, flowVersion.ProcessType, &flowMeta); err != nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: statusCodeInvalidFlow,
				Message:    err.Error(),
			}},
		}, nil
	}
	initialHistoryPrelocks, err := e.resolveFlowAtomicHistoryPrelocks(ctx, &flowMeta)
	if err != nil {
		return &models.FlowExecutionResult{
			ActionName: flowApiName,
			IsSuccess:  false,
			Errors: []models.FlowExecutionError{{
				StatusCode: statusCodeInvalidFlow,
				Message:    err.Error(),
			}},
		}, nil
	}
	if isBeforeSaveFlow(ctx) {
		if err := ValidateBeforeSaveFlowEffects(&flowMeta); err != nil {
			return &models.FlowExecutionResult{
				ActionName: flowApiName,
				IsSuccess:  false,
				Errors: []models.FlowExecutionError{{
					StatusCode: statusCodeInvalidFlow,
					Message:    err.Error(),
				}},
			}, nil
		}
	}
	// Thread audit FlowDefinitionId onto ctx so every element executor's
	// defer-emit can populate Row.FlowDefinitionId without a per-element
	// metadata lookup. Subflows inherit via ctx propagation. Audit sample
	// rate uses the global DefaultAuditSampleRate (every element completion);
	// per-flow override is not yet wired up to a public-core authoring path.
	ctx = withAuditFlowDefinitionId(ctx, flowVersion.DefinitionId)

	// Decide persistence mode once per run. The default (true) preserves
	// prior behaviour for REST-invoked, scheduled, and subflow-invoked
	// flows — all of which benefit from an audit trail. Record-triggered
	// flows opt out via WithTransientFlowInterviews in the listener,
	// avoiding 2 * N DB round-trips on the hot save path.
	persist := shouldPersistFlowInterview(ctx)

	// Create flow interview (persistent or transient per the mode above).
	if e.interviews == nil {
		return nil, apperrors.NewInternalError("flow executor has no FlowInterviewService wired", nil)
	}
	var interview *models.FlowInterview
	var createErr error
	if persist {
		interview, createErr = e.interviews.CreateInterview(ctx, flowVersion.Id, flowApiName, user)
	} else {
		interview, createErr = e.interviews.CreateTransientInterview(flowVersion.Id, flowApiName, user)
	}
	if createErr != nil {
		return nil, apperrors.NewInternalError("failed to create flow interview", createErr)
	}
	if outcomeRoot {
		publishFlowExecutionID(ctx, interview.Id)
	}
	// Connector workers stamp only immutable lineage plus one-way hashes. Emit a
	// dedicated row for every root/subflow interview so FlowExecutionId can be
	// joined back to all represented WorkItems and Events without inspecting a
	// webhook payload or assuming one execution per WorkItem.
	emitConnectorInboundExecutionAudit(ctx, interview)

	// Initialize context with input variables. Pass processType so Workflow flows
	// receive myVariable_current/myVariable_old aliases bound to $Record/$Record__Old.
	initialContextData := cloneFlowContextData(initializeFlowContext(&flowMeta, inputs, flowVersion.ProcessType))
	bindWorkflowRecordAliases(&flowMeta, initialContextData, flowVersion.ProcessType)
	interview.ContextData = cloneFlowContextData(initialContextData)
	bindWorkflowRecordAliases(&flowMeta, interview.ContextData, flowVersion.ProcessType)

	// Check entry conditions (for record-triggered flows). Use the Ctx
	// variant so Start filters like `WhatId StartsWith "006"` on
	// Salesforce-imported flows resolve against UUID IDs via sfKeyPrefixMatch.
	if !e.checkEntryConditionsCtx(ctx, &flowMeta, interview.ContextData, user) {
		// Record doesn't meet entry conditions - skip execution.
		if persist {
			// Internal fresh-execution bookkeeping: interview lifecycle writes run elevated
			// (caller already passed the action permission gate). The user-facing
			// ResumeInterview path deliberately does NOT elevate — issue #1619 / PR review.
			if completeErr := e.interviews.CompleteInterview(processTableWriteContext(ctx, "flow interview bookkeeping write"), interview.Id, models.FlowInterviewStatusFinished, user); completeErr != nil {
				e.logger.Warn("Failed to mark skipped interview as finished: " + completeErr.Error())
			}
		}
		return &models.FlowExecutionResult{
			ActionName:   flowApiName,
			IsSuccess:    true,
			OutputValues: map[string]interface{}{"_skipped": true},
		}, nil
	}

	// The interview row is created above, outside any transaction, so the
	// audit trail survives rollback of an atomic run.
	var outputs map[string]interface{}
	var retriable bool
	var cause *contractmodels.FlowExecutionCause
	var attemptCause *flowAttemptCauseCollector
	var dependency *contractmodels.FlowExecutionDependencyRequest
	executionCtx := ports.WithInitialHistorySequenceLocks(ctx, initialHistoryPrelocks)
	executionCtx = ports.WithInitialHistoryBusinessRowObjects(executionCtx, flowAtomicHistoryBusinessRowObjects(&flowMeta))
	if isPreparedFlowClosureRootExecution(executionCtx) {
		firstAttemptPrelocks := ports.OrderedHistorySequenceFields(append(
			ports.PlannedHistorySequenceLocks(executionCtx),
			initialHistoryPrelocks...,
		))
		executionCtx = ports.WithPlannedHistorySequenceLocks(executionCtx, firstAttemptPrelocks)
	}
	atomicBoundary, err := e.runPossiblyAtomicWithCompletion(executionCtx, flowMeta.RunAtomically, func(runCtx context.Context) error {
		runCtx, attemptCause = withFlowAttemptCauseCollector(runCtx)
		var attemptDependency *flowAttemptDependencyCollector
		runCtx, attemptDependency = withFlowAttemptDependencyCollector(runCtx)
		defer func() { cause = attemptCause.cause }()
		// Scope records-layer raw-read reuse to this execution attempt. A
		// pessimistic retry rolls back the previous attempt, so its cached raw
		// rows and owner proofs must not be replayed by the next attempt.
		runCtx = withFlowAttemptRecordReadCache(runCtx)
		// Each pessimistic retry attempt must start from the original input-bound
		// interview context. Element execution mutates ContextData in-process
		// ($Record writeback, lookup outputs, assignment variables); if a lock-
		// contention attempt rolls back, those mutations must not leak into the
		// next attempt's control flow.
		interview.ContextData = cloneFlowContextData(initialContextData)
		bindWorkflowRecordAliases(&flowMeta, interview.ContextData, flowVersion.ProcessType)
		interview.CurrentElement = ""
		discoveredPrelocks, prelockErr := e.discoverFlowAtomicParentPrelocks(runCtx, interview, &flowMeta, user)
		if prelockErr != nil {
			return prelockErr
		}
		runCtx = withFlowAtomicParentPrelocks(runCtx, discoveredPrelocks)
		businessPrelocks := recordlocks.ForcedParentPrelocksFromContext(runCtx)
		var elementErr error
		runErr := e.runWithFlowAtomicHistoryBusinessRows(runCtx, &flowMeta, businessPrelocks, func(prelockedCtx context.Context) error {
			outputs, retriable, elementErr = e.executeFlowMetadata(prelockedCtx, interview, &flowMeta, user)
			return elementErr
		})
		dependency = attemptDependency.dependency
		return runErr
	})
	if isFlowDependencySuspension(err) {
		if dependency == nil {
			err = fmt.Errorf("flow dependency suspension did not publish a dependency: %w", err)
		} else if isWithinAtomicFlowBoundary(ctx) {
			publishFlowExecutionDependency(ctx, dependency)
			return &models.FlowExecutionResult{
				ActionName:  flowApiName,
				IsSuccess:   false,
				IsRetriable: true,
			}, err
		} else if atomicBoundary.ownsRollbackBoundary &&
			atomicBoundary.completion == ports.TransactionCompletionRolledBack {
			publishFlowExecutionDependency(ctx, dependency)
			if persist {
				if completeErr := e.interviews.CompleteInterview(
					processTableWriteContext(ctx, "flow interview bookkeeping write"),
					interview.Id,
					models.FlowInterviewStatusFinished,
					user,
				); completeErr != nil {
					e.logger.Error("Failed to mark suspended interview as finished", completeErr)
				}
			}
			return &models.FlowExecutionResult{
				ActionName:  flowApiName,
				IsSuccess:   true,
				IsRetriable: true,
			}, nil
		} else {
			err = fmt.Errorf(
				"atomic flow dependency suspension has no confirmed rollback outcome %q: %w",
				atomicBoundary.completion,
				err,
			)
		}
	} else if dependency != nil {
		err = fmt.Errorf("flow published a dependency without suspending the attempt")
	}
	if err != nil {
		// Log the error with enough context to debug without the DB row.
		// For transient runs this is the only record of the failure.
		e.logger.Error(fmt.Sprintf("flow '%s' (version %s) execution failed: %s", flowApiName, flowVersion.Id, err.Error()), err)
		if persist {
			if completeErr := e.interviews.CompleteInterview(processTableWriteContext(ctx, "flow interview bookkeeping write"), interview.Id, models.FlowInterviewStatusError, user); completeErr != nil {
				e.logger.Error("Failed to mark interview as error", completeErr)
			}
		}

		// SF-native statusCode propagates through typed flow faults even when
		// the lower-level failure crossed event-bus / element-executor wrappers.
		// Absent a deliberate flow-facing error, fall back to "FLOW_ERROR".
		statusCode := "FLOW_ERROR"
		message := err.Error()
		isCustomError := false
		isRetriable := sharedErrors.IsRetriable(err)
		customFault, hasCustomFault := flowCustomErrorFaultForPresentation(err)
		if hasCustomFault {
			isCustomError = true
			if customFault.StatusCode != "" {
				statusCode = customFault.StatusCode
			}
			if customFault.Message != "" {
				message = customFault.Message
			}
		}
		var retriableFault *FlowRetriableFault
		if errors.As(err, &retriableFault) {
			isRetriable = true
			if retriableFault.StatusCode != "" {
				statusCode = retriableFault.StatusCode
			}
			if retriableFault.Message != "" {
				message = retriableFault.Message
			}
		}
		var statusFault *FlowStatusFault
		if errors.As(err, &statusFault) {
			if statusFault.StatusCode != "" {
				statusCode = statusFault.StatusCode
			}
			if statusFault.Message != "" {
				message = statusFault.Message
			}
		}
		authoredErrors, hasAuthoredCustomErrors := authoredCustomFlowExecutionErrors(err)
		var validationErrors []models.FlowExecutionError
		hasClientValidationErrors := false
		if !hasAuthoredCustomErrors {
			validationErrors, hasClientValidationErrors = flowExecutionErrorsForPresentation(err)
		}
		if hasClientValidationErrors {
			isCustomError = true
			statusCode = validationErrors[0].StatusCode
			if validationErrors[0].Message != "" {
				message = validationErrors[0].Message
			}
		}
		resultErrorOrigin := flowExecutionResultErrorOriginUnknown
		if hasCustomFault || hasAuthoredCustomErrors {
			resultErrorOrigin = flowExecutionResultErrorOriginAuthored
		} else if hasClientValidationErrors {
			resultErrorOrigin = flowExecutionResultErrorOriginNeutralEngine
		}
		publishFlowExecutionResultErrorOrigin(ctx, resultErrorOrigin)
		elementType, _ := flowElementTimingDescriptor(&flowMeta, interview.CurrentElement)
		cause = selectTerminalFlowExecutionCause(attemptCause, err, elementType, interview.CurrentElement)
		isRetriable = isRetriable || isRetriableFlowExecutionFailure(err, cause)
		// Nested Flow action calls still execute inside the root connector
		// outcome context. Publish their terminal classification to both shared
		// collectors before the legacy child result is wrapped by the parent, or
		// the parent can reduce a retriable TiDB failure to generic failure.
		publishTerminalFlowAttemptCause(ctx, cause)
		publishTerminalFlowExecutionCause(ctx, cause)
		emitConnectorInboundTerminalCauseAudit(ctx, interview, cause, statusCode)

		flowErrors := []models.FlowExecutionError{{
			StatusCode: statusCode,
			Message:    message,
		}}
		if hasAuthoredCustomErrors {
			flowErrors = authoredErrors
		} else if hasClientValidationErrors {
			flowErrors = validationErrors
		}
		failureResult := &models.FlowExecutionResult{
			ActionName:    flowApiName,
			IsSuccess:     false,
			IsCustomError: isCustomError,
			IsRetriable:   isRetriable,
			Errors:        flowErrors,
		}
		// Prelock replay plans are internal transaction-control flow, not
		// child-flow business results. Nested flows share the parent's atomic
		// transaction, so preserve the typed error chain for the root retry owner
		// instead of forcing the caller to reconstruct it from Errors[].
		if _, retryRequired := ports.HistorySequencePrelockFields(err); retryRequired {
			return failureResult, err
		}
		if ports.IsUniqueFieldUpsertRace(err) {
			return failureResult, err
		}
		var parentPrelockReplay recordlocks.RetryablePrelockError
		if errors.As(err, &parentPrelockReplay) &&
			len(parentPrelockReplay.ParentPrelockKeys()) > 0 {
			return failureResult, err
		}
		if errors.Is(err, ErrFlowExecutionBudgetExceeded) {
			return failureResult, err
		}
		// This signal is transaction control owned by the Platform-command
		// profile, not a Flow business failure. Preserve its typed chain only
		// after the atomic attempt has rolled back so the Worker can run the
		// separate recovery transition and replay the exact command. The action
		// is installed only in that production composition.
		var recoveryRequested *ConnectorInboundRemoteIdentityRecoveryRequestedError
		if errors.As(err, &recoveryRequested) {
			return failureResult, err
		}
		return failureResult, atomicFlowDeadlineError(flowMeta.RunAtomically || isWithinAtomicFlowBoundary(ctx), err)
	}

	// If the flow paused on a Wait element, leave the interview in Paused
	// status for the scheduler to resume — do NOT overwrite with Finished.
	// Note: a transient interview that hit a Wait was promoted to a
	// persistent Paused row inside pauseForWait (IsTransient cleared).
	paused := interview.InterviewStatus == models.FlowInterviewStatusPaused

	// Mark interview as complete (persistent path only; transient runs
	// have no row to update; paused runs must be left for resume).
	if persist && !paused {
		if completeErr := e.interviews.CompleteInterview(processTableWriteContext(ctx, "flow interview bookkeeping write"), interview.Id, models.FlowInterviewStatusFinished, user); completeErr != nil {
			e.logger.Error("Failed to mark interview as finished", completeErr)
		}
	}

	return &models.FlowExecutionResult{
		ActionName:   flowApiName,
		IsSuccess:    true,
		IsRetriable:  retriable,
		OutputValues: outputs,
	}, nil
}

func (e *FlowExecutor) buildScheduledRunInputs(ctx context.Context, flowVersion *models.FlowVersion, user *models.UserSession) ([]map[string]interface{}, error) {
	// Connector-sourced scheduled flow: pull input sets from an external list
	// endpoint instead of a local SObject query. Must come FIRST so a flow
	// without Start.Object does not collapse to the single-nil-input path.
	if flowVersion != nil && flowVersion.Metadata.Start != nil && flowVersion.Metadata.Start.ConnectorSource != nil {
		return e.buildConnectorDirectoryInputs(ctx, flowVersion, user)
	}

	if flowVersion == nil || flowVersion.Metadata.Start == nil || flowVersion.Metadata.Start.Object == "" {
		return []map[string]interface{}{nil}, nil
	}
	if e.query == nil {
		return nil, apperrors.NewInternalError("scheduled flow query service unavailable", nil)
	}

	flowMeta := flowVersion.Metadata
	filterExpr, err := e.buildScheduledStartFilter(nil, flowMeta.Start.Filters, flowMeta.Start.FilterLogic)
	if err != nil {
		return nil, err
	}

	// Scheduled filterFormula: correctness guaranteed by in-memory per-row eval
	// (checkEntryConditionsCtx) + the existing scheduledFlowMaxScanned stop-loss.
	// There is no SQL prefilter for filterFormula — the structured filters[] SQL
	// prefilter path (buildScheduledStartFilter above) is unaffected.

	inputs := make([]map[string]interface{}, 0, scheduledFlowPageSize)
	objectName := flowMeta.Start.Object
	sqlPrefilterApplied := filterExpr != ""
	scanned := 0
	lastID := ""

	for {
		pageFilterExpr := filterExpr
		if lastID != "" {
			cursorExpr := fmt.Sprintf("Id > '%s'", escapeSQLValue(lastID))
			if pageFilterExpr != "" {
				pageFilterExpr = fmt.Sprintf("(%s) AND %s", pageFilterExpr, cursorExpr)
			} else {
				pageFilterExpr = cursorExpr
			}
		}

		result, err := e.query.QueryWithFilter(ctx, flowMeta.Start.Object, pageFilterExpr, user, "Id", "ASC", scheduledFlowPageSize)
		if err != nil {
			return nil, err
		}

		for _, record := range result.Records {
			scanned++

			// Stamp neutral object-type metadata so downstream RecordUpdate elements with
			// object=null can resolve the target object name.
			models.StampObjectType(record, objectName)

			runInputs := map[string]interface{}{
				"$Record": record,
			}
			if e.checkEntryConditionsCtx(ctx, &flowMeta, runInputs, user) {
				inputs = append(inputs, runInputs)
				if len(inputs) >= scheduledFlowMaxMatched {
					e.warnScheduledCandidateScan("matched_cap_reached", objectName, scanned, len(inputs), sqlPrefilterApplied)
					return inputs, nil
				}
			}
		}

		if scanned >= scheduledFlowMaxScanned {
			scanErr := &scheduledFlowCandidateScanBudgetError{
				ObjectName:          objectName,
				ScannedCount:        scanned,
				MatchedCount:        len(inputs),
				ScanCap:             scheduledFlowMaxScanned,
				MatchedCap:          scheduledFlowMaxMatched,
				SQLPrefilterApplied: sqlPrefilterApplied,
			}
			e.warnScheduledCandidateScan("scan_budget_exceeded", objectName, scanned, len(inputs), sqlPrefilterApplied)
			return nil, apperrors.NewInternalError("scheduled flow candidate scan exhausted stop-loss budget", scanErr)
		}

		if len(result.Records) == 0 || len(result.Records) < scheduledFlowPageSize {
			return inputs, nil
		}

		lastID = result.Records[len(result.Records)-1].GetString("Id")
		if lastID == "" {
			return nil, apperrors.NewInternalError("scheduled flow pagination encountered a record without Id", nil)
		}
	}
}

// ResumeInterview resumes a paused flow interview. On entry the interview
// is in Paused state with CurrentElement pointing at the Wait (or other
// pauseable) element. This method:
//
//  1. Loads the interview + FlowVersion (re-hydrates Metadata).
//  2. Re-evaluates any time-sensitive context so expressions that depend
//     on $Flow.CurrentDate / $Flow.CurrentDateTime (TODAY / NOW) pick up
//     the resume-time value, not the pause-time value.
//  3. Clears PausedUntil and flips InterviewStatus back to Running atomically.
//  4. Re-enters the executor loop at CurrentElement. On that re-entry the
//     Wait element's conditions are re-evaluated — so a catch-up resume
//     (PausedUntil already in the past at pause time) completes without
//     further delay.
//  5. If the resumed execution pauses again (e.g. a second Wait element),
//     leave the interview Paused; otherwise mark Finished.
func (e *FlowExecutor) ResumeInterview(ctx context.Context, interviewId string, user *models.UserSession) error {
	// Scheduled/paused flow resumptions run outside the original REST ctx;
	// stamp Entrypoint=FLOW so downstream record mutations produce correct
	// AuditEvent rows. See ExecuteFlow comment for the transport contract.
	ctx = audit.WithEntrypoint(ctx, audit.EntrypointFlow)

	interview, err := e.interviews.GetInterview(ctx, interviewId, user)
	if err != nil {
		return err
	}
	if interview == nil {
		return apperrors.NewNotFoundError("flow interview", interviewId)
	}
	if interview.InterviewStatus != models.FlowInterviewStatusPaused {
		// Already claimed by another scheduler instance or manually resumed.
		return nil
	}

	// Atomic claim: use ClaimPausedInterview which does a conditional
	// UPDATE ... WHERE InterviewStatus = 'Paused'. Returns false if
	// another scheduler already claimed this interview.
	claimed, err := e.interviews.ClaimPausedInterview(ctx, interview.Id, user)
	if err != nil {
		return err
	}
	if !claimed {
		return nil // another scheduler won the race
	}

	// Get flow version to get metadata
	flowVersion, err := e.getFlowVersionById(ctx, interview.FlowVersionId, user)
	if err != nil {
		return err
	}
	if flowVersion == nil {
		return apperrors.NewNotFoundError("flow version", interview.FlowVersionId)
	}
	resumeFlowAPIName, err := e.hydrateFlowVersionDefinitionRuntimeFields(ctx, flowVersion, user)
	if err != nil {
		return err
	}

	// Use hydrated metadata (HydrateMetadata is called by the data layer)
	flowMeta := flowVersion.Metadata

	// Resume execution must thread the same audit scope as a fresh run so
	// post-resume element emissions carry FlowDefinitionId. Audit sample
	// rate falls back to DefaultAuditSampleRate (every element completion)
	// via auditSampleRateFromCtx; per-flow override is not yet wired up
	// to a public-core authoring path. Mirrors executeFlowVersion above.
	ctx = withAuditFlowDefinitionId(ctx, flowVersion.DefinitionId)
	ctx = withCurrentFlowExecution(ctx, resumeFlowAPIName, flowVersion)

	// Refresh time-of-day context so any downstream Wait / condition check
	// that reads $Flow.CurrentDateTime or $Flow.CurrentDate sees resume
	// time rather than pause time.
	if interview.ContextData == nil {
		interview.ContextData = make(map[string]interface{})
	}
	now := time.Now()
	interview.ContextData["$Flow.CurrentDateTime"] = now.Format(time.RFC3339)
	interview.ContextData["$Flow.CurrentDate"] = now.Format("2006-01-02")

	// ClaimPausedInterview already flipped status to Running.
	interview.InterviewStatus = models.FlowInterviewStatusRunning
	interview.PausedUntil = nil

	// Resume execution from current element.
	_, _, err = e.executeFlowMetadata(ctx, interview, &flowMeta, user)
	if err != nil {
		if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusError, user); completeErr != nil {
			e.logger.Error("Failed to mark interview as error during resume", completeErr)
		}
		return err
	}

	// If the resumed run paused again (another Wait element), leave it Paused.
	if interview.InterviewStatus == models.FlowInterviewStatusPaused {
		return nil
	}

	if completeErr := e.interviews.CompleteInterview(ctx, interview.Id, models.FlowInterviewStatusFinished, user); completeErr != nil {
		e.logger.Error("Failed to mark interview as finished during resume", completeErr)
	}
	return nil
}

// executeFlowMetadata executes flow from parsed metadata
func (e *FlowExecutor) executeFlowMetadata(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, user *models.UserSession) (map[string]interface{}, bool, error) {
	if _, ok := ctx.Value(flowExecutionBudgetKey{}).(*flowExecutionBudget); !ok {
		var err error
		ctx, err = withFlowExecutionBudget(ctx, defaultFlowExecutionElementBudget)
		if err != nil {
			return nil, false, err
		}
	}
	// Callable-only flows (InvocableProcess callees) have no start element
	// but do have startElementReference pointing to the first element.
	// Fall back to it so the executor can enter the flow.
	var firstElement string
	if flowMeta.Start != nil && flowMeta.Start.Connector != nil {
		firstElement = flowMeta.Start.Connector.TargetReference
	} else if flowMeta.StartElementReference != "" {
		firstElement = flowMeta.StartElementReference
	} else if flowMeta.Start != nil {
		// AsyncAfterCommit and other scheduled-path flows have no direct Start.Connector.
		// Instead, the entry element is carried by Start.ScheduledPaths[].Connector.
		// Use the first non-empty scheduled path connector as the async entry point.
		for _, sp := range flowMeta.Start.ScheduledPaths {
			if sp.Connector != nil && sp.Connector.TargetReference != "" {
				firstElement = sp.Connector.TargetReference
				break
			}
		}
		if firstElement == "" {
			return nil, false, apperrors.NewValidationError("start", "flow has no start element or startElementReference")
		}
	} else {
		return nil, false, apperrors.NewValidationError("start", "flow has no start element or startElementReference")
	}

	// Ensure each top-level flow execution has its own per-interview executionState.
	// Record-triggered callers (WithTransientFlowInterviews / WithBeforeSaveFlow) attach
	// a single FlowRuntimeContext to ctx and then iterate over N matching interviews
	// reusing the same ctx — so the first flow's objectType/rCache/loopDepth would leak
	// into subsequent flows without isolation here.  REST/scheduled/resume callers attach
	// no FRC at all; flowRuntimeContextFromCtx returns a zero-value sentinel in that case,
	// and clone() on it produces a correctly-initialised child.  Either way: always clone
	// and reset executionState so this call tree owns its own per-interview state.
	{
		n := flowRuntimeContextFromCtx(ctx).clone()
		n.executionState = &flowExecutionState{}
		ctx = withFlowRuntimeContext(ctx, n)
	}

	// Set the trigger object type on FlowRuntimeContext so resolveNestedReferenceCtx
	// can translate __r segments to concrete object types via the FK field's
	// ReferenceTo. Previously stored as a transient key in interview.ContextData;
	// now held on the typed FlowRuntimeContext field (T22a).
	frc := flowRuntimeContextFromCtx(ctx)
	if frc.ObjectType() == "" {
		// Primary source: the flow's own trigger object.
		if flowMeta.Start != nil && flowMeta.Start.Object != "" {
			frc.SetObjectType(flowMeta.Start.Object)
		} else if interview.ContextData != nil {
			// Fallback for AutoLaunchedFlow subflows invoked with a
			// $Record input: recover the type from record metadata stamped by
			// the parent executeSubflow or by a RecordLookup that produced the record.
			if inferred := inferRootObjectType(interview.ContextData); inferred != "" {
				frc.SetObjectType(inferred)
			}
		}
	}

	// Formula resources are immutable metadata indexed at the execution
	// boundary and evaluated only when an elementReference consumes them.
	// Installing the scope also discards eager results serialized by older
	// runtimes, so resume cannot resurrect stale derived values.
	ctx = withFlowFormulaResourceScope(ctx, interview, flowMeta.Formulas)

	// Start from the first element
	currentElement := firstElement
	if interview.CurrentElement != "" {
		// Resume from paused element
		currentElement = interview.CurrentElement
	}

	completedSteps := 0

	for currentElement != "" {
		if deadlineErr := atomicFlowDeadlineContextError(
			flowMeta.RunAtomically || isWithinAtomicFlowBoundary(ctx),
			ctx,
		); deadlineErr != nil {
			return nil, false, deadlineErr
		}
		if err := consumeFlowExecutionBudget(ctx); err != nil {
			return nil, false, err
		}
		interview.CurrentElement = currentElement

		nextElement, err := e.executeElement(ctx, interview, flowMeta, currentElement, user)
		if err != nil {
			if deadlineErr := atomicFlowDeadlineError(flowMeta.RunAtomically || isWithinAtomicFlowBoundary(ctx), err); deadlineErr != nil {
				return nil, false, deadlineErr
			}
			if _, ok := flowCustomErrorFaultForPresentation(err); ok {
				return nil, false, err
			}
			// Only validation marked at a record-mutation/child-Flow boundary and
			// legacy rule-only failures cross this generic element boundary.
			// Flow configuration and action-input ValidationError values remain
			// internal even though they share the same concrete error type.
			if clientErr, ok := flowClientValidationForPresentation(err); ok {
				return nil, false, clientErr
			}
			return nil, false, apperrors.NewInternalError(fmt.Sprintf("error executing element '%s'", currentElement), err)
		}
		if deadlineErr := atomicFlowDeadlineContextError(flowMeta.RunAtomically || isWithinAtomicFlowBoundary(ctx), ctx); deadlineErr != nil {
			return nil, false, deadlineErr
		}

		completedSteps++
		interview.ContextData["_completedSteps"] = float64(completedSteps)

		// Move to next element
		currentElement = nextElement
	}

	// Collect output variables
	outputs := make(map[string]interface{})
	for _, variable := range flowMeta.Variables {
		if variable.IsOutput {
			if val, ok := interview.ContextData[variable.Name]; ok {
				outputs[variable.Name] = val
			}
		}
	}
	if record, ok := interview.ContextData["$Record"]; ok {
		outputs["$Record"] = record
	}
	if oldRecord, ok := interview.ContextData["$Record__Old"]; ok {
		outputs["$Record__Old"] = oldRecord
	}

	retriable := false
	if frc := flowRuntimeContextFromCtx(ctx); frc != nil {
		retriable = frc.IsRetriable()
	}

	return outputs, retriable, nil
}

// executeElement executes a single flow element and returns the next element name.
// Individual element executors are in:
//   - executor_elements.go    (Decision, Assignment, Loop)
//   - executor_record_ops.go  (RecordCreate, RecordUpdate, RecordDelete, RecordLookup)
//   - executor_actions.go     (ActionCall, Subflow)
func (e *FlowExecutor) executeElement(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, elementName string, user *models.UserSession) (nextElement string, err error) {
	// Find and execute the element by type
	elementStarted := time.Now()
	elementType, objectName := flowElementTimingDescriptor(flowMeta, elementName)
	ctx = withFlowCauseElementType(ctx, elementType)
	defer func() {
		if v, _ := ctx.Value(constants.ContextKeyPerfTiming).(bool); !v {
			return
		}
		status := "success"
		if err != nil {
			status = "error"
		}
		flowName := ""
		if current, ok := flowRuntimeContextFromCtx(ctx).CurrentFlow(); ok && current.APIName != "" {
			flowName = current.APIName
		}
		versionID := ""
		if interview != nil {
			versionID = interview.FlowVersionId
		}
		logFlowTiming(ctx, flowName, versionID, elementName, elementType, objectName, status, nextElement, elementStarted, err)
	}()

	// Check Decisions
	for _, decision := range flowMeta.Decisions {
		if decision.Name == elementName {
			return e.executeDecision(ctx, interview, &decision, user)
		}
	}

	// Check Assignments
	for _, assignment := range flowMeta.Assignments {
		if assignment.Name == elementName {
			return e.executeAssignment(ctx, interview, flowMeta, &assignment, user)
		}
	}

	// Check Record Creates
	for _, create := range flowMeta.RecordCreates {
		if create.Name == elementName {
			return e.executeRecordCreate(ctx, interview, flowMeta, &create, user)
		}
	}

	// Check Record Updates
	for _, update := range flowMeta.RecordUpdates {
		if update.Name == elementName {
			return e.executeRecordUpdate(ctx, interview, flowMeta, &update, user)
		}
	}

	// Check Record Deletes
	for _, del := range flowMeta.RecordDeletes {
		if del.Name == elementName {
			return e.executeRecordDelete(ctx, interview, &del, user)
		}
	}

	// Check Record Lookups
	for _, lookup := range flowMeta.RecordLookups {
		if lookup.Name == elementName {
			return e.executeRecordLookup(ctx, interview, &lookup, user)
		}
	}

	// Check Action Calls
	for _, action := range flowMeta.ActionCalls {
		if action.Name == elementName {
			return e.executeActionCall(ctx, interview, flowMeta, &action, user)
		}
	}

	// Check Subflows
	for _, subflow := range flowMeta.Subflows {
		if subflow.Name == elementName {
			return e.executeSubflow(ctx, interview, &subflow, user)
		}
	}

	// Check Loops
	for _, loop := range flowMeta.Loops {
		if loop.Name == elementName {
			return e.executeLoop(ctx, interview, &loop, user)
		}
	}

	// Check Custom Errors
	for _, customErr := range flowMeta.CustomErrors {
		if customErr.Name == elementName {
			return e.executeCustomError(ctx, interview, &customErr)
		}
	}

	// Check Waits
	for _, wait := range flowMeta.Waits {
		if wait.Name == elementName {
			return e.executeWait(ctx, interview, &wait, user)
		}
	}

	return "", apperrors.NewNotFoundError("flow element", elementName)
}

// FlowCustomErrorFault is returned by executeCustomError when a flow's
// customError element fires. It also carries an opaque authored result through
// record-mutation and child-Flow boundaries without interpreting its dialect
// status codes.
type FlowCustomErrorFault struct {
	Message    string
	StatusCode string

	resultErrors []models.FlowExecutionError
	cause        error
}

func (f *FlowCustomErrorFault) Error() string {
	if f == nil {
		return ""
	}
	if f.cause != nil {
		return f.cause.Error()
	}
	return f.Message
}

// executeCustomError handles a custom error element, returning an error to
// block execution. The returned error is a *FlowCustomErrorFault when the
// author declared a statusCode, so executeFlowVersion can propagate the
// SF-native code into the dispatcher response.
func (e *FlowExecutor) executeCustomError(ctx context.Context, interview *models.FlowInterview, customErr *models.FlowCustomError) (nextElem string, execErr error) {
	start := time.Now()
	defer func() {
		errPayload := ""
		if execErr != nil {
			errPayload = execErr.Error()
		}
		emitFlowAudit(ctx, Row{
			FlowExecutionId:  interview.Id,
			FlowDefinitionId: auditFlowDefinitionIdFromCtx(ctx),
			FlowVersionId:    interview.FlowVersionId,
			ElementId:        customErr.Name,
			ElementType:      "CustomError",
			Status:           "error", // CustomError always terminates with error
			DurationMs:       int(time.Since(start).Milliseconds()),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
	}()
	msg := customErr.Label
	statusCode := ""
	if len(customErr.CustomErrorMessages) > 0 {
		msg = customErr.CustomErrorMessages[0].ErrorMessage
		statusCode = customErr.CustomErrorMessages[0].StatusCode
	}
	return "", &FlowCustomErrorFault{Message: msg, StatusCode: statusCode}
}