package automation

import (
	"context"
	"fmt"
	"log"
	"strings"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/nexuscrm/shared/pkg/stagetiming"
)

// MaxAtomicFilteredRecordOperationRows is the hard pre-effect fan-out bound for
// one filtered RecordUpdate or RecordDelete element. The executor queries
// limit+1 and rejects the complete element before applying any row effect when
// the bound is exceeded. Runtime bounded-command admission relies on this
// engine invariant; changing it requires the corresponding profile proof and
// rollback tests to change in the same release.
const MaxAtomicFilteredRecordOperationRows = 200

const maxFilteredCrossObjectRecords = MaxAtomicFilteredRecordOperationRows

// queryFilteredRecordOperationTargets performs only the candidate-identity
// phase of a filtered RecordUpdate/RecordDelete. The write path below consumes
// no value from the selected row except Id; guarded updates re-evaluate their
// authored filters against the locked current row before mutation. Selecting
// every readable field here turns one bounded identity scan into a wide formula
// view read without adding authority or correctness.
func (e *FlowExecutor) queryFilteredRecordOperationTargets(
	ctx context.Context,
	objectName string,
	filter string,
	user *models.UserSession,
) (*models.QueryResult, error) {
	return e.query.QueryWithSpec(ctx, queryspec.QuerySpec{
		ObjectAPIName: objectName,
		SelectFields:  []string{constants.FieldID},
		Filter:        queryspec.FormulaPredicate(filter),
		Limit:         maxFilteredCrossObjectRecords + 1,
	}, user)
}

func preparedFlowDirectRecordOperationLimitError(
	ctx context.Context,
	elementType string,
	objectName string,
	rows int,
) error {
	limit, bounded := preparedFlowDirectRecordOperationLimit(ctx)
	if !bounded || rows <= limit {
		return nil
	}
	return fmt.Errorf(
		"direct %s on %s exceeded prepared Flow governor limit: %d rows exceeds %d",
		elementType,
		objectName,
		rows,
		limit,
	)
}

const flowFieldMappingMergeFillBlankOnly = "fillBlankOnly"

func flowVariableObjectType(flowMeta *models.FlowMetadata, inputRef string) string {
	if flowMeta == nil || inputRef == "" {
		return ""
	}
	for i := range flowMeta.Variables {
		variable := flowMeta.Variables[i]
		if variable.Name == inputRef {
			return flowVariableObjectTypeFromDeclaration(variable)
		}
	}
	return ""
}

func flowVariableObjectTypeFromDeclaration(variable models.FlowVariable) string {
	if !strings.EqualFold(strings.TrimSpace(variable.DataType), "SObject") {
		return ""
	}
	return variable.ObjectType
}

// executeRecordCreate creates a new record. The field population follows the
// action-scoped recordCreate default pipeline (see field_default_resolver.go
// and the plan doc for the full ordering):
//
//	Stage 1  caller write                — honored at REST layer before reaching here
//	(1.5)    _System_ObjectFieldMapping  — bulk field copy from source (Lead→Contact etc.)
//	Stage 2  flow InputAssignment        — flow-author explicit sets; nil-resolved refs SKIP
//	Stage 3  _System_ActionFieldDefault  — fills absent keys scoped to (StandardActionId, TargetObject)
//	Stage 3.5 connectorFieldMapping       — optional pack-owned connector mapping merged into the create image
//	Stage 4  RecordType resolution       — reserved; separate PR once the assignment table lands
//	Stages 5+6 (schema DefaultValue, required validator) fire inside the selected
//	persistence Insert/UpsertByExternalID path.
func (e *FlowExecutor) executeRecordCreate(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, create *models.FlowRecordCreate, user *models.UserSession) (nextElem string, execErr error) {
	ctx, auditFault := withAuditFaultBox(ctx)
	start := time.Now()
	defer func() {
		status := "success"
		errPayload := ""
		if execErr != nil {
			status = "error"
			errPayload = execErr.Error()
		} else if be := auditFault.Err(); be != nil {
			status = "error"
			errPayload = be.Error()
		}
		emitFlowAudit(ctx, Row{
			FlowExecutionId:  interview.Id,
			FlowDefinitionId: auditFlowDefinitionIdFromCtx(ctx),
			FlowVersionId:    interview.FlowVersionId,
			ElementId:        create.Name,
			ElementType:      "RecordCreate",
			Status:           status,
			DurationMs:       int(time.Since(start).Milliseconds()),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
	}()
	// Preview-mode guard for the triggered-flow path Codex flagged on PR #1008:
	// when previewPersistence.PreviewBeforeUpdate delegates into the live
	// persistence service, the resulting RecordBeforeUpdate publish re-runs
	// matching before-save flows on the trigger listener's shared (live,
	// unwrapped) executor. That executor's persistence is the real one, so
	// without an entry-level isPreviewMode(ctx) check a RecordCreate element
	// in a triggered flow would commit a real row from a no-DML preview path.
	// previewPersistence.Insert handles the top-level capture path; this
	// guard handles the second path where persistence is NOT the wrapper.
	if isPreviewMode(ctx) {
		if _, isPreviewWrapper := e.persistence.(*previewPersistence); !isPreviewWrapper {
			if create.FaultConnector != nil {
				return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, errPreviewRecordCreateAttempt)
			}
			return "", errPreviewRecordCreateAttempt
		}
	}
	record := models.SObject{}

	if isPreviewMode(ctx) && isBeforeSaveFlow(ctx) {
		if create.FaultConnector != nil {
			return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, errPreviewBeforeSaveCreate)
		}
		return "", errPreviewBeforeSaveCreate
	}

	if err := e.applyRecordCreateFieldMapping(ctx, interview, create, record, user); err != nil {
		if create.FaultConnector != nil {
			return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	// Stage 2. Three-way split, driven by presence-aware resource resolution:
	//   - literal null in flow metadata         → exists=true, write nil
	//   - elementReference resolves to nil      → exists=true, write nil
	//     (caller/flow explicitly set null — SF contract: "null this field")
	//   - elementReference absent from context  → exists=false, SKIP so
	//     stage 3 (ActionFieldDefault) / stage 5 (schema DefaultValue) fill
	//
	// The old implementation collapsed the first two cases into the skip
	// path (matching on `value == nil && wasElementReference`), so a
	// caller passing `{"fieldName": null}` lost their explicit-null intent
	// — defaults overwrote it. presence-aware resolution closes that gap
	// without changing literal and resolved-reference behavior elsewhere.
	for _, input := range create.InputAssignments {
		value, exists, resolveErr := e.resolveAssignmentRHSValueWithPresenceE(ctx, interview, user, flowMeta, input.Value)
		if resolveErr != nil {
			if create.FaultConnector != nil {
				return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
		if !exists {
			if e.logger != nil {
				e.logger.Debug(fmt.Sprintf(
					"RecordCreate %q: input %q is an unresolved elementReference; skipping so default layers can fill it",
					create.Name, input.Field))
			}
			continue
		}
		record[input.Field] = value
	}

	// Stage 3. No-op unless the dispatcher stamped _System_ActionFieldDefault
	// rows on ctx (WithActionFieldDefaults). Resolver only writes fields
	// absent after stage 2 — never overwrites a flow-written value.
	// Pass super-user capability so the resolver can inject $Permission keys
	// (Issue #609 — shared formula.InjectSuperUserPermissions pattern).
	if e.fieldDefaultResolver != nil {
		isSuperUser := userHasSuperUserCapability(ctx, user, e.permissions)
		if err := e.fieldDefaultResolver.Apply(ctx, record, create.Object, interview.ContextData, isSuperUser); err != nil {
			if create.FaultConnector != nil {
				return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, err)
			}
			return "", err
		}
	}

	if err := e.applyRecordCreateConnectorFieldMapping(ctx, interview, create, record, user); err != nil {
		if create.FaultConnector != nil {
			return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	// Stage 4 reserved. RecordType resolution via per-profile assignment
	// table is deferred to a separate PR — the assignment table does not
	// exist in this codebase today. When it lands, insert a resolver call
	// here, after stage 3 and before persistence.Insert.

	// Tag the ctx so previewPersistence.Insert recognises this as the
	// user-facing RecordCreate path and captures it into Creates[] (instead
	// of rejecting it as an engine-internal Insert). No-op outside of
	// preview mode — the marker is read only by previewPersistence.
	insertCtx := withRecordCreateInsertMarker(ctx)
	mutationStarted := time.Now()
	var result models.SObject
	var err error
	upsertField := strings.TrimSpace(create.UpsertByExternalIDField)
	guardFields, guardErr := normalizedRecordCreateAdoptionGuardFields(create.UpsertAdoptionGuardFields)
	if guardErr != nil {
		err = fmt.Errorf("RecordCreate %q: %w", create.Name, guardErr)
	} else if create.AdoptionConflictConnector != nil && len(guardFields) == 0 {
		err = fmt.Errorf("RecordCreate %q declares adoptionConflictConnector without adoption guard fields", create.Name)
	} else if len(guardFields) > 0 && upsertField == "" {
		err = fmt.Errorf("RecordCreate %q declares adoption guard fields without upsertByExternalIdField", create.Name)
	}
	if err != nil {
		// Invalid guarded-adoption metadata fails before any persistence call.
	} else if upsertField == "" {
		result, err = e.persistence.Insert(insertCtx, create.Object, record, user)
		err = flowRecordMutationValidation(err)
	} else {
		externalIDValue, externalIDMatches := uniqueRecordFieldValueCaseInsensitive(record, upsertField)
		externalIDString := ""
		externalIDIsString := true
		if externalIDMatches == 1 {
			externalIDString, externalIDIsString = externalIDValue.(string)
		}
		if externalIDMatches > 1 {
			err = fmt.Errorf(
				"RecordCreate %q has ambiguous case-insensitive values for upsert field %q",
				create.Name,
				upsertField,
			)
		} else if !externalIDIsString {
			err = fmt.Errorf(
				"RecordCreate %q requires a string value for upsert field %q when one is supplied",
				create.Name,
				upsertField,
			)
		} else if len(guardFields) > 0 {
			guardedPersistence, supported := e.persistence.(externalIDGuardedUpsertPersistence)
			if !supported {
				err = fmt.Errorf(
					"RecordCreate %q requires guarded atomic external-ID upsert persistence for field %q",
					create.Name,
					upsertField,
				)
			} else {
				result, _, err = guardedPersistence.UpsertByExternalIDWithAdoptionGuards(
					insertCtx,
					create.Object,
					upsertField,
					externalIDString,
					record,
					guardFields,
					user,
				)
				err = flowRecordMutationValidation(err)
			}
		} else if upsertPersistence, ok := e.persistence.(externalIDUpsertPersistence); !ok {
			err = fmt.Errorf(
				"RecordCreate %q requires atomic external-ID upsert persistence for field %q",
				create.Name,
				upsertField,
			)
		} else {
			result, _, err = upsertPersistence.UpsertByUniqueExternalID(
				insertCtx,
				create.Object,
				upsertField,
				externalIDString,
				record,
				user,
			)
			err = flowRecordMutationValidation(err)
		}
	}
	stagetiming.Add(ctx, stagetiming.StageRecordMutation, time.Since(mutationStarted))
	if err != nil {
		if create.AdoptionConflictConnector != nil &&
			!appErrors.IsInternal(err) &&
			ports.IsExternalIDUpsertAdoptionGuardConflict(err) {
			return e.routeFault(ctx, interview, create.AdoptionConflictConnector.TargetReference, err)
		}
		if create.FaultConnector != nil {
			return e.routeFault(ctx, interview, create.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	flowRuntimeContextFromCtx(ctx).MarkRecordMutation()

	// Publish the engine-confirmed inserted record to the shared post-upsert
	// registry so later elements (e.g. registerConnectorInboundRemoteAuthorship)
	// can read the source record id and stamped remote id without trusting the
	// flow-writable interview.ContextData["$Record"].
	e.publishPostUpsertRecord(ctx, create.Object, result)

	// Store the created record ID
	if create.AssignRecordIdToReference != "" {
		interview.ContextData[create.AssignRecordIdToReference] = result[constants.FieldID]
	}
	if create.StoreOutputAutomatically {
		interview.ContextData[create.Name] = result[constants.FieldID]
	}

	if create.Connector != nil {
		return create.Connector.TargetReference, nil
	}
	return "", nil
}

func normalizedRecordCreateAdoptionGuardFields(fields []string) ([]string, error) {
	if len(fields) == 0 {
		return nil, nil
	}
	normalized := make([]string, 0, len(fields))
	for _, field := range fields {
		field = strings.TrimSpace(field)
		if field == "" {
			return nil, fmt.Errorf("adoption guard fields contain a blank name")
		}
		for _, existing := range normalized {
			if strings.EqualFold(existing, field) {
				return nil, fmt.Errorf("duplicate adoption guard field %q", field)
			}
		}
		normalized = append(normalized, field)
	}
	return normalized, nil
}

func uniqueRecordFieldValueCaseInsensitive(record models.SObject, fieldName string) (any, int) {
	fieldName = strings.TrimSpace(fieldName)
	var value any
	matches := 0
	for recordField, recordValue := range record {
		if strings.EqualFold(strings.TrimSpace(recordField), fieldName) {
			matches++
			value = recordValue
		}
	}
	return value, matches
}

// applyRecordCreateFieldMapping applies the registry-driven _System_ObjectFieldMapping
// rows (if the element declares a reference) to record before InputAssignments
// run. InputAssignments always override mapped values.
//
// shouldDetachAfterSaveSelfUpdate distinguishes the legacy record-event
// self-lock escape hatch from a caller-owned bounded Flow transaction.
// ExecutePreparedFlowClosureWithinOwnedTransactionOutcome marks the latter as
// the pessimistic retry owner; detaching there would commit an effect and its
// commit fence before the remaining Flow elements execute.
func shouldDetachAfterSaveSelfUpdate(ctx context.Context) bool {
	return !isBeforeSaveFlow(ctx) && !hasAtomicFlowPessimisticRetryOwner(ctx)
}

func (e *FlowExecutor) executeRecordUpdate(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, update *models.FlowRecordUpdate, user *models.UserSession) (nextElem string, execErr error) {
	ctx, auditFault := withAuditFaultBox(ctx)
	start := time.Now()
	// skipReason is set when the cross-object dotted-ref resolver determines
	// that the target record cannot be reached (null FK, unresolved ref, etc.)
	// and the update should be skipped gracefully without error.
	var skipReason string
	defer func() {
		status := "success"
		errPayload := ""
		if execErr != nil {
			status = "error"
			errPayload = execErr.Error()
		} else if be := auditFault.Err(); be != nil {
			status = "error"
			errPayload = be.Error()
		} else if skipReason != "" {
			status = "skipped"
			errPayload = skipReason
		}
		emitFlowAudit(ctx, Row{
			FlowExecutionId:  interview.Id,
			FlowDefinitionId: auditFlowDefinitionIdFromCtx(ctx),
			FlowVersionId:    interview.FlowVersionId,
			ElementId:        update.Name,
			ElementType:      "RecordUpdate",
			Status:           status,
			DurationMs:       int(time.Since(start).Milliseconds()),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
	}()

	if isBeforeSaveFlow(ctx) {
		if isPreviewMode(ctx) && !constants.HasRecordPreviewPreparation(ctx) {
			if update.FaultConnector != nil {
				return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, errPreviewRecordUpdateAttempt)
			}
			return "", errPreviewRecordUpdateAttempt
		}
		decision := resolveBeforeSaveRecordUpdateEffectForStartObject(flowRuntimeContextFromCtx(ctx).ObjectType(), update)
		if decision.Effect == BeforeSaveSideEffecting {
			err := beforeSaveInvalidFlowError(decision)
			if update.FaultConnector != nil {
				return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
			}
			return "", err
		}
	} else if isPreviewMode(ctx) {
		// Preview remains fail-closed for recordUpdate until the no-DML update
		// path has the same post-before-save validation contract as live update.
		if update.FaultConnector != nil {
			return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, errPreviewRecordUpdateAttempt)
		}
		return "", errPreviewRecordUpdateAttempt
	}
	if update.RecheckFiltersAtWrite && (len(update.Filters) == 0 || strings.TrimSpace(update.InputReference) != "" || strings.TrimSpace(update.Object) == "") {
		err := fmt.Errorf("flow guarded RecordUpdate %q requires explicit object, filters, and no inputReference", update.Name)
		if update.FaultConnector != nil {
			return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	updates := models.SObject{}
	var resolvedUpdateInput interface{}
	var resolvedUpdateInputPresent bool
	if update.InputReference != "" {
		var resolveErr error
		resolvedUpdateInput, resolvedUpdateInputPresent, resolveErr = e.ResolveFlowResource(
			ctx, interview, user,
			map[string]interface{}{"elementReference": update.InputReference},
		)
		if resolveErr != nil {
			if update.FaultConnector != nil {
				return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
	}

	if err := e.applyRecordUpdateFieldMapping(ctx, interview, update, updates, user); err != nil {
		if update.FaultConnector != nil {
			return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	for _, input := range update.InputAssignments {
		// Use presence-aware resolution so a failed __r lazy-load (metadata-miss
		// or transient query error) returns (nil, false) and is skipped, while
		// an explicit null (FK cleared, orphan parent) returns (nil, true) and
		// is written — preserving SF "null this field" semantics.
		//
		// This mirrors the identical guard in executeAssignment (Assign branch)
		// and executeRecordCreate stage 2, eliminating the R10 cross-layer
		// asymmetry: all three InputAssignment callers now behave consistently
		// when resource resolution returns (_, false). (#1817 F4 R10)
		value, resolved, resolveErr := e.resolveAssignmentRHSValueWithPresenceE(ctx, interview, user, flowMeta, input.Value)
		if resolveErr != nil {
			if update.FaultConnector != nil {
				return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
		if !resolved {
			continue
		}
		updates[input.Field] = value
	}

	// Handle Collection Updates
	if update.InputReference != "" {
		// Check for collection of SObjects
		if collection, ok := resolvedUpdateInput.([]models.SObject); ok {
			if err := preparedFlowDirectRecordOperationLimitError(
				ctx, "RecordUpdate", update.Object, len(collection),
			); err != nil {
				if update.FaultConnector != nil {
					return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
				}
				return "", err
			}
			for _, record := range collection {
				id, _ := record[constants.FieldID].(string)
				if id != "" {
					objectName := update.Object
					if objectName == "" {
						objectName = models.RecordObjectType(record)
					}
					if objectName == "" {
						objectName = flowVariableObjectType(flowMeta, update.InputReference)
					}
					if objectName == "" {
						err := fmt.Errorf("recordUpdate %q inputReference %q cannot resolve object type", update.Name, update.InputReference)
						if update.FaultConnector != nil {
							return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
						}
						return "", err
					}
					payload := recordUpdateCollectionPayload(ctx, objectName, id, record, updates)
					err := e.applyRecordUpdatePersistence(ctx, objectName, id, payload, user)
					if err != nil {
						// Fail on first error? or continue? Salesforce usually fails the transaction.
						if update.FaultConnector != nil {
							return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
						}
						return "", err
					}
					flowRuntimeContextFromCtx(ctx).AdvanceRecordLookupBaseline(objectName, id, payload)
				}
			}
			if update.Connector != nil {
				return update.Connector.TargetReference, nil
			}
			return "", nil
		}

		// Check for generic slice (interface{})
		if collection, ok := resolvedUpdateInput.([]interface{}); ok {
			if err := preparedFlowDirectRecordOperationLimitError(
				ctx, "RecordUpdate", update.Object, len(collection),
			); err != nil {
				if update.FaultConnector != nil {
					return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
				}
				return "", err
			}
			for _, item := range collection {
				if record, ok := item.(models.SObject); ok {
					id, _ := record[constants.FieldID].(string)
					if id != "" {
						objectName := update.Object
						if objectName == "" {
							objectName = models.RecordObjectType(record)
						}
						if objectName == "" {
							objectName = flowVariableObjectType(flowMeta, update.InputReference)
						}
						if objectName == "" {
							err := fmt.Errorf("recordUpdate %q inputReference %q cannot resolve object type", update.Name, update.InputReference)
							if update.FaultConnector != nil {
								return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
							}
							return "", err
						}
						payload := recordUpdateCollectionPayload(ctx, objectName, id, record, updates)
						err := e.applyRecordUpdatePersistence(ctx, objectName, id, payload, user)
						if err != nil {
							if update.FaultConnector != nil {
								return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
							}
							return "", err
						}
						flowRuntimeContextFromCtx(ctx).AdvanceRecordLookupBaseline(objectName, id, payload)
					}
				}
			}
			if update.Connector != nil {
				return update.Connector.TargetReference, nil
			}
			return "", nil
		}
	}

	// triggerRecordID is the ID of the record that triggered the current flow
	// (i.e. $Record.Id). Used by the filtered and single-record update paths to
	// detect when a recordUpdate targets the triggering record itself so they can
	// call persistence.Update with a transaction-free context and avoid a
	// TiDB lock-wait timeout (Error 1205) self-deadlock in AfterSave flows:
	// the outbox worker delivers AfterSave events inside its own transaction,
	// and calling persistence.Update (which issues SELECT ... FOR UPDATE) on a
	// row that the enclosing transaction already holds an exclusive lock on
	// causes a deadlock on the same DB connection (#1787).
	var triggerRecordID string
	if trigRec, ok := interview.ContextData["$Record"].(models.SObject); ok {
		triggerRecordID, _ = trigRec[constants.FieldID].(string)
	}

	// Filtered cross-object update: the element specifies Filters (and
	// usually an explicit Object) to find the records to update. Process
	// Builder generates this pattern for "update a related record" actions
	// (e.g. update CampaignMember WHERE LeadId = $Record.Id). When
	// Filters are present AND InputReference is empty, query via filters
	// instead of falling through to the implicit-$Record path, which
	// would misuse the trigger record's ID as the target record ID.
	//
	// When objectName is null (Process Builder cross-object pattern A/C),
	// derive the target object from field metadata on the trigger record's
	// lookup field referenced in the "Id EqualTo $Record.<field>" filter.
	if len(update.Filters) > 0 && update.InputReference == "" {
		targetObject := update.Object
		if targetObject == "" {
			targetObject = e.deriveObjectFromIdFilter(ctx, interview, update.Filters)
		}
		if targetObject != "" {
			// Detect the Salesforce Process Builder "implicit-Id" pattern: a filter
			// on field=Id with processMetadataValues:[{name:"implicit",booleanValue:true}]
			// signals that the element was generated by Process Builder to identify
			// the target record via a relationship field (e.g. $Record.Account__c).
			// When this pattern is present, use Process Builder null semantics for
			// the remaining (non-implicit) criteria — in PB runtime, a NULL field
			// is treated as "not equal to any non-null value" (TRUE for !=), which
			// differs from SQL semantics where NULL != 'x' evaluates to NULL (false).
			// Without this, a null-valued field matching a != criterion silently
			// excludes the target record from the query, causing a silent no-op (#2132).
			implicitFilterIndex, implicitFilter := findImplicitIdFilter(update.Filters)
			preResolvedFilterValues := make(map[int]interface{}, 1)
			implicitTargetBlank := false
			if implicitFilter != nil {
				// Resolve target Id with presence semantics so we can tell
				// "the variable doesn't exist" apart from "the variable
				// exists and its value is null/blank". The legacy
				// A value-only resolver would discard the presence bit and
				// collapse both into "" — conflating a flow-author bug
				// (recordLookup didn't run; assignment skipped) with normal
				// data (a triggering record whose lookup field was never
				// populated, e.g. $Record.AccountId == nil).
				//
				// Fail loud ONLY when the reference is genuinely
				// unresolved (resolved == false). Salesforce treats the
				// resolved-but-blank case as a zero-target no-op — the
				// flow ran, no related record matched, the connector
				// progresses normally. Constructing `Id = ''` here would
				// also be a no-op at the DB layer, but short-circuiting
				// avoids the round-trip and pins the semantic explicitly.
				// (#232 / IANTHEREAL Round-1 P1 on PR #272)
				targetId, resolved, resolveErr := e.ResolveFlowResource(ctx, interview, user, implicitFilter.Value)
				if resolveErr != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, resolveErr)
					}
					return "", resolveErr
				}
				if !resolved {
					ref := implicitFilterReference(implicitFilter.Value)
					err := fmt.Errorf("flow recordUpdate %q (object=%s): implicit-Id filter could not resolve target Id from %s", update.Name, targetObject, ref)
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
				preResolvedFilterValues[implicitFilterIndex] = targetId
				if targetIdStr, _ := targetId.(string); targetIdStr == "" {
					implicitTargetBlank = true
					if update.RecheckFiltersAtWrite {
						// Guarded metadata is validated in full below before this
						// remains a zero-target no-op.
					} else {
						// Resolved-but-blank: zero-target no-op. Skip the
						// query entirely and progress through the normal
						// connector — matches Salesforce semantics.
						if update.Connector != nil {
							return update.Connector.TargetReference, nil
						}
						return "", nil
					}
				}
			}
			var filter string
			var currentWriteFilters []resolvedCurrentRecordFilter
			var currentWriteFilterLogic string
			var err error
			if update.RecheckFiltersAtWrite {
				currentWriteFilterLogic, err = validateGuardedRecordUpdateFilterShape(update.Filters, update.FilterLogic)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
				if implicitTargetBlank {
					if update.Connector != nil {
						return update.Connector.TargetReference, nil
					}
					return "", nil
				}
				currentWriteFilters, err = e.resolveCurrentRecordFilters(ctx, interview, user, targetObject, update.Filters, preResolvedFilterValues)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
				filter, err = e.buildGuardedFilterFromResolvedConditions(currentWriteFilters, currentWriteFilterLogic, implicitFilter != nil)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
			} else if implicitFilter != nil {
				filter, err = e.buildFilterFromConditionsPBCtxE(ctx, interview, user, update.Filters, update.FilterLogic)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
			} else {
				filter, err = e.buildFilterFromConditionsCtxE(ctx, interview, user, update.Filters, update.FilterLogic)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
			}
			if filter != "" {
				// Query with limit+1 to detect overflow without paginating.
				result, err := e.queryFilteredRecordOperationTargets(ctx, targetObject, filter, user)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
				if len(result.Records) > maxFilteredCrossObjectRecords {
					msg := fmt.Sprintf("filtered RecordUpdate on %s exceeded governor limit: matched more than %d rows", targetObject, maxFilteredCrossObjectRecords)
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, fmt.Errorf("%s", msg))
					}
					return "", fmt.Errorf("%s", msg)
				}
				if update.RecheckFiltersAtWrite && triggerRecordID != "" && !isBeforeSaveFlow(ctx) {
					// Reject the complete result set before processing any row. A guarded
					// trigger-record write would detach below and recheck a different
					// transaction snapshot from the selector.
					for _, record := range result.Records {
						id, _ := record[constants.FieldID].(string)
						if id != triggerRecordID {
							continue
						}
						err := fmt.Errorf("flow guarded after-save RecordUpdate %q cannot target trigger record %q: the locked recheck would use a detached transaction snapshot", update.Name, triggerRecordID)
						if update.FaultConnector != nil {
							return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
						}
						return "", err
					}
				}
				for _, record := range result.Records {
					id, _ := record[constants.FieldID].(string)
					if id == "" {
						continue
					}
					// AfterSave self-lock guard (#1787): if the filter matched the
					// triggering record itself, the outbox transaction already holds a
					// lock on that row (via the enclosing transaction context). Calling
					// persistence.Update with the same context would issue
					// SELECT ... FOR UPDATE inside that transaction and block on its own
					// lock. Detach the transaction so Update runs in a fresh, independent
					// transaction instead.
					//
					// DetachTx: In AfterSave context, the triggering record is already
					// locked by the outbox worker's transaction. Executing an UPDATE on
					// the same row within that transaction causes TiDB Error 1205 (lock
					// wait timeout / self-deadlock).
					//
					// ATOMICITY TRADE-OFF: This DetachTx runs the UPDATE in a fresh
					// independent transaction that commits before the outbox transaction
					// finishes. If the outer outbox transaction subsequently rolls back,
					// this flow-driven UPDATE has already been committed and cannot be
					// rolled back with it. This is an accepted semantic: flow side-effects
					// on the triggering record are "best-effort" in AfterSave context.
					updateCtx := ctx
					if id == triggerRecordID && shouldDetachAfterSaveSelfUpdate(ctx) {
						updateCtx = e.persistence.DetachTx(ctx)
					}
					matched := true
					var err error
					if update.RecheckFiltersAtWrite {
						matched, err = e.applyGuardedRecordUpdatePersistence(updateCtx, targetObject, id, updates, user, func(guardCtx context.Context, lockedCurrent models.SObject) (bool, error) {
							return e.lockedRecordMatchesResolvedFilters(guardCtx, lockedCurrent, currentWriteFilters, currentWriteFilterLogic)
						})
					} else {
						err = e.applyRecordUpdatePersistence(updateCtx, targetObject, id, updates, user)
					}
					if err != nil {
						if update.FaultConnector != nil {
							return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
						}
						return "", err
					}
					if !matched {
						continue
					}
					// Context sync for filtered-query self-update: if the matched row is
					// the triggering record, merge written fields back into $Record so
					// later chained nodes' per-node filter gates read committed state
					// (mirrors the single-record persistence path below).
					if id == triggerRecordID {
						if rec, ok := interview.ContextData["$Record"].(models.SObject); ok {
							for k, v := range updates {
								rec[k] = v
							}
							// Re-project active-formula fields so later
							// Decision/Assignment/FlowFormula reads of
							// $Record.<active> see post-update values. Mirrors the
							// single-record $Record path below — both shapes of
							// trigger-record self-update must reproject identically
							// for the chained-reread contract to hold (#1377). Use
							// updateCtx so the read sees the just-committed detached
							// write; honor caller security level (no SystemContext
							// elevation here — see analogous block at line 620+).
							if e.persistence != nil {
								snap, projErr := e.persistence.RecomputeActiveFormulaProjection(updateCtx, user, targetObject, id, nil)
								if projErr != nil {
									return "", projErr
								}
								for k, v := range snap {
									rec[k] = v
								}
							}
							interview.ContextData["$Record"] = rec
						}
					}
				}
				if update.Connector != nil {
					return update.Connector.TargetReference, nil
				}
				return "", nil
			}
		}
	}

	// Single Record Update Logic
	// Salesforce Workflow recordUpdates have no InputReference — they
	// implicitly target $Record (the triggering record). Default to
	// $Record when InputReference is empty and the flow has a $Record
	// in context.
	inputRef := update.InputReference
	if inputRef == "" {
		if _, hasRecord := interview.ContextData["$Record"]; hasRecord {
			inputRef = "$Record"
		}
	}

	// Resolve target object name. Author may omit update.Object on an
	// implicit-$Record recordUpdate; fall back to object-type metadata on the
	// referenced record so AfterSave flows can still persist the change.
	objectName := update.Object

	var recordId string
	if inputRef != "" && strings.Contains(inputRef, ".") {
		// Dotted input reference (e.g. "$Record.Parent__r"): resolve the
		// related record via the cross-object resolver. Non-dotted refs
		// (including "$Record", plain variable names) are handled below
		// by the existing flat-lookup path.
		crossObjRecord, crossObjName, crossObjId, reason, ok, resolveErr := e.resolveDottedRecordReference(ctx, interview, user, inputRef)
		if resolveErr != nil {
			if update.FaultConnector != nil {
				return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
		if !ok {
			// Cannot reach the target record — skip gracefully without error.
			skipReason = reason
			if update.Connector != nil {
				return update.Connector.TargetReference, nil
			}
			return "", nil
		}
		recordId = crossObjId
		if objectName == "" {
			objectName = crossObjName
		}
		_ = crossObjRecord // record is already resolved; proceed to persistence path below
	} else if inputRef != "" {
		resolvedInput := resolvedUpdateInput
		resolvedInputPresent := resolvedUpdateInputPresent
		if inputRef != update.InputReference {
			var resolveErr error
			resolvedInput, resolvedInputPresent, resolveErr = e.ResolveFlowResource(
				ctx, interview, user,
				map[string]interface{}{"elementReference": inputRef},
			)
			if resolveErr != nil {
				if update.FaultConnector != nil {
					return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, resolveErr)
				}
				return "", resolveErr
			}
		}
		if record, ok := resolvedInput.(models.SObject); ok && resolvedInputPresent {
			recordId, _ = record[constants.FieldID].(string)

			if objectName == "" {
				objectName = models.RecordObjectType(record)
			}
			if objectName == "" && flowMeta != nil {
				// A declared SObject working variable with no stamped
				// attributes.type (e.g. an empty seeded variable): resolve the
				// target object from the variable's declared type in flow
				// metadata. Salesforce's Update Records on an SObject variable
				// targets the variable's declared object.
				objectName = flowVariableObjectType(flowMeta, inputRef)
			}

			// Per-node filter gate (issue #1119): when a recordUpdates node
			// carries its own Filters and targets $Record implicitly (no
			// explicit Object or InputReference), evaluate those filters against
			// the current in-flight $Record state before applying any DML.
			// Filter-false = skip this node's assignments; connector is still
			// followed. This matches Salesforce's per-node filter semantics for
			// chained Update Records elements in decision-tree flows.
			//
			// Guard is restricted to the implicit-$Record case: nodes that
			// arrive here via the filtered-cross-object path (explicit Object)
			// already evaluated their filters via a DB query (see the
			// len(update.Filters)>0 block above). Nodes with an explicit
			// InputReference other than "$Record" skip this check as well since
			// they target a specific variable, not the triggering record.
			if len(update.Filters) > 0 && inputRef == "$Record" {
				matches, err := e.recordMatchesNodeFilters(ctx, interview, user, record, update.Filters, update.FilterLogic)
				if err != nil {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
					}
					return "", err
				}
				if !matches {
					if update.Connector != nil {
						return update.Connector.TargetReference, nil
					}
					return "", nil
				}
			}

			// Before-save flows modify $Record in-place — the trigger listener
			// merges changes back into the insert payload. Also use in-place
			// when the record has no ID yet (not yet persisted).
			//
			// AfterSave flows must go through persistence.Update: the record
			// is already committed and the listener does not merge $Record
			// back, so an in-place mutation here would be a silent drop.
			// Previously `update.Object == ""` unconditionally took in-place,
			// which silently lost AfterSave writes whenever the flow author
			// omitted the Object attribute (ywang-official#1122).
			isInPlace := inputRef == "$Record" && (recordId == "" || isBeforeSaveFlow(ctx))
			if isInPlace {
				if isPreviewMode(ctx) && recordId == "" && !isBeforeSaveFlow(ctx) {
					if update.FaultConnector != nil {
						return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, errPreviewRecordUpdateAttempt)
					}
					return "", errPreviewRecordUpdateAttempt
				}
				for k, v := range updates {
					record[k] = v
				}
				interview.ContextData[inputRef] = record
				flowRuntimeContextFromCtx(ctx).MarkRecordMutation()
				// NOTE: the AddPostUpsertRecord publish that was here for the
				// before-save isInPlace path has been removed (carve-down #2080).
				// before-save register is structurally impossible — the engine
				// rejects before-save flows containing ActionCalls at flow entry
				// (before_save_effects.go:94 + flow_executor.go:771-772), so no
				// register action can ever run on an uncommitted record. The
				// snapshot publish here was the only reason before-save isInPlace
				// needed to feed the registry; without before-save register there
				// is no consumer of this snapshot in the register path.
				if update.Connector != nil {
					return update.Connector.TargetReference, nil
				}
				return "", nil
			}
		} else if id, ok := resolvedInput.(string); ok && resolvedInputPresent {
			recordId = id
		}
	}

	if recordId != "" {
		if objectName == "" {
			// AfterSave path with no resolvable object: skip rather than
			// silently succeed. Without this guard we'd call
			// persistence.Update with an empty table name, which either
			// errors deep in the persistence layer or (worse) logs a
			// misleading "Object" not found.
			log.Printf("flow recordUpdate '%s' has empty Object on an AfterSave run and no object-type metadata on inputRef %q — skipping", update.Name, inputRef)
		} else {
			// AfterSave self-lock guard (#1787): when the implicit-$Record
			// path reaches here (inputRef == "$Record", AfterSave context),
			// ctx carries the outbox worker's transaction. persistence.Update
			// would issue SELECT ... FOR UPDATE on the triggering record inside
			// that transaction, which blocks on its own lock — TiDB Error 1205.
			// Detach the transaction so Update starts an independent one.
			// NEW: If InputReference pointed to an SObject (but not $Record), merge its values with InputAssignments
			// (InputAssignments take precedence over SObject variable values)
			finalUpdates := updates
			if inputRef != "$Record" && update.FieldMapping == nil { // Only for flow variable references, not $Record
				if record, ok := resolvedUpdateInput.(models.SObject); ok && resolvedUpdateInputPresent {
					// Create a new map to hold the merged values
					mergedUpdates := make(models.SObject)
					// First, copy all SObject variable values (except Id).
					for k, v := range record {
						// Skip the Id field as its the identifier and should not be part of the update
						if k != constants.FieldID {
							mergedUpdates[k] = v
						}
					}
					// Then, override with InputAssignments (they take precedence)
					for k, v := range updates {
						mergedUpdates[k] = v
					}
					finalUpdates = mergedUpdates
				}
			}
			if update.FieldMapping != nil && len(update.InputAssignments) == 0 && len(finalUpdates) == 0 {
				if update.Connector != nil {
					return update.Connector.TargetReference, nil
				}
				return "", nil
			}

			// AfterSave self-lock guard (#1787): when the implicit-$Record
			// path reaches here (inputRef == "$Record", AfterSave context),
			// ctx carries the outbox worker's transaction. persistence.Update
			// would issue SELECT ... FOR UPDATE on the triggering record inside
			// that transaction, which blocks on its own lock — TiDB Error 1205.
			// Detach the transaction so Update starts an independent one.
			updateCtx := ctx
			if inputRef == "$Record" && shouldDetachAfterSaveSelfUpdate(ctx) {
				updateCtx = e.persistence.DetachTx(ctx)
			}

			err := e.applyRecordUpdatePersistence(updateCtx, objectName, recordId, finalUpdates, user)
			if err != nil {
				if update.FaultConnector != nil {
					return e.routeFault(ctx, interview, update.FaultConnector.TargetReference, err)
				}
				return "", err
			}
			// Post-persistence context sync (issue #1119): merge written fields
			// back into interview.ContextData["$Record"] so subsequent chained
			// recordUpdates nodes see the committed state when evaluating their
			// per-node filters. The in-place (BeforeSave/no-ID) path already
			// syncs at the isInPlace block above.
			if inputRef == "$Record" {
				if rec, ok := interview.ContextData["$Record"].(models.SObject); ok {
					for k, v := range finalUpdates {
						rec[k] = v
					}
					// Re-project active-formula fields so later Decision/Assignment/
					// FlowFormula reads of $Record.<active> see post-update values
					// (issue #1377; closes the chained-reread gap from #1362). The
					// underlying projector uses the same QueryService → view path as
					// the trigger-payload snapshot, so dependency ordering and
					// ProjectionConfig are honored by the view substrate.
					//
					// IMPORTANT — ctx selection: use updateCtx, NOT ctx.
					// applyRecordUpdatePersistence runs through the same updateCtx
					// (line 608) — either the outer ctx for before-save / non-self
					// updates, or a detached tx for AfterSave self-updates (lines
					// 603-606). Reading through the outer outbox-worker ctx after
					// a detached-tx write returns the pre-update snapshot because
					// the outer tx cannot see the detached tx's commit. updateCtx
					// always sees the just-committed state.
					//
					// IMPORTANT — security context: do NOT elevate to SystemContext
					// here. executeRecordUpdate is on the shared FlowExecutor path,
					// not a trigger-only helper, so user-invoked flows also reach
					// this point. Elevation would widen FLS bypass into generic
					// flow execution. Trigger-context callers already arrive with
					// SystemContext (flow_trigger_listener.go:172-175), so honoring
					// the existing ctx security level is correct in both modes.
					if e.persistence != nil {
						snap, projErr := e.persistence.RecomputeActiveFormulaProjection(updateCtx, user, objectName, recordId, nil)
						if projErr != nil {
							return "", projErr
						}
						for k, v := range snap {
							rec[k] = v
						}
					}
					interview.ContextData["$Record"] = rec
				}
			}
		}
	}

	if update.Connector != nil {
		return update.Connector.TargetReference, nil
	}
	return "", nil
}

func (e *FlowExecutor) applyRecordUpdatePersistence(ctx context.Context, objectName string, id string, updates models.SObject, user *models.UserSession) error {
	if isPreviewMode(ctx) {
		return errPreviewRecordUpdateAttempt
	}
	mutationStarted := time.Now()
	if err := e.persistence.Update(ctx, objectName, id, updates, user); err != nil {
		stagetiming.Add(ctx, stagetiming.StageRecordMutation, time.Since(mutationStarted))
		return flowRecordMutationValidation(err)
	}
	stagetiming.Add(ctx, stagetiming.StageRecordMutation, time.Since(mutationStarted))
	flowRuntimeContextFromCtx(ctx).MarkRecordMutation()
	// Publish the engine-confirmed update to the shared post-upsert registry so
	// later elements (e.g. registerConnectorInboundRemoteAuthorship) can read the
	// source record id and stamped remote id without trusting flow-writable
	// interview.ContextData["$Record"]. Build a minimal snapshot with Id + the
	// written fields so remoteIDField is present if it was in the update payload.
	// This is the SINGLE shared persistence boundary for all 5 update paths
	// (direct, collection, filtered query, before-save self, CommitPreparedUpdate
	// via product-owned action handlers — the latter adds via AddPostUpsertRecord
	// separately since it goes through CommitPreparedUpdate, not this function).
	snap := make(models.SObject, len(updates)+1)
	for k, v := range updates {
		snap[k] = v
	}
	snap[constants.FieldID] = id
	e.publishPostUpsertRecord(ctx, objectName, snap)
	return nil
}

func (e *FlowExecutor) applyGuardedRecordUpdatePersistence(
	ctx context.Context,
	objectName string,
	id string,
	updates models.SObject,
	user *models.UserSession,
	guard func(context.Context, models.SObject) (bool, error),
) (bool, error) {
	if isPreviewMode(ctx) {
		return false, errPreviewRecordUpdateAttempt
	}
	guardedPersistence, ok := e.persistence.(guardedRecordUpdatePersistence)
	if !ok {
		return false, fmt.Errorf("flow guarded RecordUpdate requires locked-current persistence support")
	}
	mutationStarted := time.Now()
	matched, err := guardedPersistence.UpdateGuarded(ctx, objectName, id, updates, user, guard)
	stagetiming.Add(ctx, stagetiming.StageRecordMutation, time.Since(mutationStarted))
	if err != nil || !matched {
		return matched, flowRecordMutationValidation(err)
	}
	flowRuntimeContextFromCtx(ctx).MarkRecordMutation()
	snap := make(models.SObject, len(updates)+1)
	for k, v := range updates {
		snap[k] = v
	}
	snap[constants.FieldID] = id
	e.publishPostUpsertRecord(ctx, objectName, snap)
	return true, nil
}

// publishPostUpsertRecord records an engine-confirmed upsert into the per-flow-execution
// post-upsert registry that registerConnectorInboundRemoteAuthorship reads — ONLY when the
// write is in the atomic-flow transaction the register's GetLock can re-read. A DetachTx'd
// best-effort write or a non-atomic write is invisible to that GetLock, so publishing it
// would make register author stale data or strand (#2080 Gate-0 v7, unified invariant).
func (e *FlowExecutor) publishPostUpsertRecord(writeCtx context.Context, object string, record models.SObject) {
	// A successful engine write is independently meaningful even if its snapshot
	// cannot be published. Register actions use this outcome to preserve valid
	// no-write paths while failing closed on a missing snapshot after a write.
	flowRuntimeContextFromCtx(writeCtx).MarkPostUpsertWrite(object)
	if !isWithinAtomicFlowBoundary(writeCtx) {
		return
	}
	if e.persistence == nil || !e.persistence.IsTransactionActive(writeCtx) {
		return
	}
	flowRuntimeContextFromCtx(writeCtx).AddPostUpsertRecord(object, record)
}

// executeRecordDelete deletes records
func (e *FlowExecutor) executeRecordDelete(ctx context.Context, interview *models.FlowInterview, del *models.FlowRecordDelete, user *models.UserSession) (nextElem string, execErr error) {
	ctx, auditFault := withAuditFaultBox(ctx)
	start := time.Now()
	defer func() {
		status := "success"
		errPayload := ""
		if execErr != nil {
			status = "error"
			errPayload = execErr.Error()
		} else if be := auditFault.Err(); be != nil {
			status = "error"
			errPayload = be.Error()
		}
		emitFlowAudit(ctx, Row{
			FlowExecutionId:  interview.Id,
			FlowDefinitionId: auditFlowDefinitionIdFromCtx(ctx),
			FlowVersionId:    interview.FlowVersionId,
			ElementId:        del.Name,
			ElementType:      "RecordDelete",
			Status:           status,
			DurationMs:       int(time.Since(start).Milliseconds()),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
	}()
	// Preview-mode guard for the triggered-flow path Codex flagged on PR #1008:
	// previewPersistence.Delete already rejects on the top-level preview
	// executor, but the live trigger-listener executor (invoked when
	// previewPersistence.PreviewBeforeUpdate delegates into the live
	// persistence service and re-publishes RecordBeforeUpdate) has the real
	// persistence wired. Without this entry-level check a RecordDelete
	// element in a triggered before-save flow would commit a real DELETE
	// from a no-DML preview path.
	if isPreviewMode(ctx) {
		if _, isPreviewWrapper := e.persistence.(*previewPersistence); !isPreviewWrapper {
			if del.FaultConnector != nil {
				return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, errPreviewRecordDeleteAttempt)
			}
			return "", errPreviewRecordDeleteAttempt
		}
	}
	var resolvedDeleteInput interface{}
	var resolvedDeleteInputPresent bool
	if del.InputReference != "" {
		var resolveErr error
		resolvedDeleteInput, resolvedDeleteInputPresent, resolveErr = e.ResolveFlowResource(
			ctx, interview, user,
			map[string]interface{}{"elementReference": del.InputReference},
		)
		if resolveErr != nil {
			if del.FaultConnector != nil {
				return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
	}
	// Handle Collection Deletes
	if del.InputReference != "" {
		// Slice of SObjects
		if collection, ok := resolvedDeleteInput.([]models.SObject); ok {
			if err := preparedFlowDirectRecordOperationLimitError(
				ctx, "RecordDelete", del.Object, len(collection),
			); err != nil {
				if del.FaultConnector != nil {
					return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
				}
				return "", err
			}
			for _, record := range collection {
				id, _ := record[constants.FieldID].(string)
				// Infer object name if missing
				objName := del.Object
				if objName == "" {
					objName = models.RecordObjectType(record)
				}
				if id != "" && objName != "" {
					err := e.applyRecordDeletePersistence(ctx, objName, id, user)
					if err != nil {
						if del.FaultConnector != nil {
							return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
						}
						return "", err
					}
				}
			}
			return del.Connector.TargetReference, nil
		}

		// Generic Slice
		if collection, ok := resolvedDeleteInput.([]interface{}); ok {
			if err := preparedFlowDirectRecordOperationLimitError(
				ctx, "RecordDelete", del.Object, len(collection),
			); err != nil {
				if del.FaultConnector != nil {
					return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
				}
				return "", err
			}
			for _, item := range collection {
				if record, ok := item.(models.SObject); ok {
					id, _ := record[constants.FieldID].(string)
					objName := del.Object
					if objName == "" {
						objName = models.RecordObjectType(record)
					}
					if id != "" && objName != "" {
						err := e.applyRecordDeletePersistence(ctx, objName, id, user)
						if err != nil {
							if del.FaultConnector != nil {
								return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
							}
							return "", err
						}
					}
				}
			}
			return del.Connector.TargetReference, nil
		}
	}

	// Filtered cross-object delete: same pattern as filtered RecordUpdate.
	if len(del.Filters) > 0 && del.InputReference == "" && del.Object != "" {
		filter, err := e.buildFilterFromConditionsCtxE(ctx, interview, user, del.Filters, del.FilterLogic)
		if err != nil {
			if del.FaultConnector != nil {
				return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
			}
			return "", err
		}
		if filter != "" {
			result, err := e.queryFilteredRecordOperationTargets(ctx, del.Object, filter, user)
			if err != nil {
				if del.FaultConnector != nil {
					return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
				}
				return "", err
			}
			if len(result.Records) > maxFilteredCrossObjectRecords {
				msg := fmt.Sprintf("filtered RecordDelete on %s exceeded governor limit: matched more than %d rows", del.Object, maxFilteredCrossObjectRecords)
				if del.FaultConnector != nil {
					return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, fmt.Errorf("%s", msg))
				}
				return "", fmt.Errorf("%s", msg)
			}
			for _, record := range result.Records {
				id, _ := record[constants.FieldID].(string)
				if id == "" {
					continue
				}
				if err := e.applyRecordDeletePersistence(ctx, del.Object, id, user); err != nil {
					if del.FaultConnector != nil {
						return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
					}
					return "", err
				}
			}
			if del.Connector != nil {
				return del.Connector.TargetReference, nil
			}
			return "", nil
		}
	}

	// Single Record Delete Logic
	// Get record ID from InputReference
	var recordId string
	if del.InputReference != "" {
		if record, ok := resolvedDeleteInput.(models.SObject); ok && resolvedDeleteInputPresent {
			recordId, _ = record[constants.FieldID].(string)
		} else if id, ok := resolvedDeleteInput.(string); ok && resolvedDeleteInputPresent {
			recordId = id
		}
	}

	// Determine object name
	objectName := del.Object
	if objectName == "" {
		// Try to infer from the record if it's an SObject
		if record, ok := resolvedDeleteInput.(models.SObject); ok && resolvedDeleteInputPresent {
			objectName = models.RecordObjectType(record)
		}
	}

	if recordId != "" && objectName != "" {
		err := e.applyRecordDeletePersistence(ctx, objectName, recordId, user)
		if err != nil {
			if del.FaultConnector != nil {
				return e.routeFault(ctx, interview, del.FaultConnector.TargetReference, err)
			}
			return "", err
		}
	}

	if del.Connector != nil {
		return del.Connector.TargetReference, nil
	}
	return "", nil
}

func (e *FlowExecutor) applyRecordDeletePersistence(ctx context.Context, objectName string, id string, user *models.UserSession) error {
	mutationStarted := time.Now()
	err := e.persistence.Delete(ctx, objectName, id, user)
	stagetiming.Add(ctx, stagetiming.StageRecordMutation, time.Since(mutationStarted))
	if err == nil {
		flowRuntimeContextFromCtx(ctx).MarkRecordMutation()
	}
	return flowRecordMutationValidation(err)
}