package automation

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

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	flowSendEmailActionType                     = flowactions.ActionTypeSendEmail
	flowSendEmailTemplateIDParam                = "templateId"
	flowSendEmailFromParam                      = "from"
	flowSendEmailMessageIDOutput                = "messageId"
	flowSendEmailRecipientsOutput               = "recipientResults"
	flowSendEmailSentCountOutput                = "sentCount"
	flowResolveEmailRecipientsActionType        = flowactions.ActionTypeResolveEmailRecipients
	flowResolveEmailRecipientsEmailsOutput      = "emails"
	flowResolveEmailRecipientsDiagnosticsOutput = "diagnostics"
	flowResolveEmailRecipientsResolvedCount     = "resolvedCount"
	flowResolveEmailRecipientsInvalidCount      = "invalidCount"
)

// maxFlowCallDepth caps how deep the parent→child subflow chain can go before
// we assume the author accidentally introduced infinite recursion. Salesforce
// documents a similar depth cap (their runtime uses 2000 elements per interview
// but also guards recursion). 10 is generous for the migrated flows (the known
// Order_Expiration_Notice_90 chain is 1 deep) while remaining safe.
const maxFlowCallDepth = 10

// withFlowCall pushes flowApiName onto the call stack stored in the
// FlowRuntimeContext carried by ctx. Returns the updated context and an error
// if pushing flowApiName would create a cycle or exceed maxFlowCallDepth.
//
// The call stack is now stored on FlowRuntimeContext.callStack (via the
// flowRuntimeContextKey carrier introduced by T21b) rather than via a
// separate flowCallStackKey context.WithValue pair. The function signature is
// preserved so callers — including the conformance harness — compile unchanged.
func withFlowCall(ctx context.Context, flowApiName string) (context.Context, error) {
	frc := flowRuntimeContextFromCtx(ctx)
	existing := frc.GetCallStack()
	for _, name := range existing {
		if strings.EqualFold(name, flowApiName) {
			return ctx, fmt.Errorf(
				"flow subflow cycle detected: %s is already in the call chain [%s]",
				flowApiName, strings.Join(append(existing, flowApiName), " -> "),
			)
		}
	}
	if len(existing) >= maxFlowCallDepth {
		return ctx, fmt.Errorf(
			"flow subflow depth %d exceeds maximum %d (chain: %s)",
			len(existing)+1, maxFlowCallDepth,
			strings.Join(append(existing, flowApiName), " -> "),
		)
	}
	newFrc := frc.PushCallStack(flowApiName)
	return withFlowRuntimeContext(ctx, newFrc), nil
}

// executeActionCall invokes an action
func (e *FlowExecutor) executeActionCall(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, action *models.FlowActionCall, 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:        action.Name,
			ElementType:      "ActionCall",
			Status:           status,
			DurationMs:       int(time.Since(start).Milliseconds()),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
	}()
	registry := e.flowActionCallRegistry()
	registration, registeredActionType := registry.Lookup(action.ActionType)
	isFlowActionCall := registeredActionType && registration.ActionType == normalizeFlowActionType("flow")
	if isFlowActionCall {
		return e.invokeRegisteredActionCall(ctx, interview, flowMeta, action, user, registration)
	}

	// Preview-mode gate: every actionCall except subflow dispatch may carry
	// side effects that the preview path's previewPersistence wrapper does not
	// intercept. Letting those run would mutate real records, send real email,
	// or fire real webhooks under a "no-DML preview" contract. The "flow"
	// subflow registration above is preview-safe because it recurses through
	// ExecuteFlow on the same wrapped executor.
	//
	// The persistence-identity check catches the top-level preview executor;
	// the isPreviewMode(ctx) check catches the second path Codex flagged on
	// PR #1008: previewPersistence.PreviewBeforeUpdate now delegates into
	// the live persistence service, which publishes RecordBeforeUpdate and
	// 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 the ctx-keyed gate a metadata action / approval call
	// from a triggered flow would slip past this guard and fire real side
	// effects from a no-DML preview path. Registry-owned preview policies are
	// the only allowlist for registered actionCall handlers here.
	_, persistenceIsPreview := e.persistence.(interface{ isFlowPreviewPersistence() })
	if persistenceIsPreview || isPreviewMode(ctx) {
		if registeredActionType {
			switch registration.PreviewPolicy {
			case FlowActionCallPreviewPolicyAllow:
				return e.invokeRegisteredActionCall(ctx, interview, flowMeta, action, user, registration)
			case FlowActionCallPreviewPolicyAllowWithPreviewPersistence:
				// Only the top-level preview executor has the no-DML persistence
				// wrapper. On triggered-flow re-entry, ctx may still carry preview
				// mode but e.persistence is live, so allowing the handler would
				// re-enter live publish/update paths.
				if !persistenceIsPreview {
					if action.FaultConnector != nil {
						return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, errPreviewActionCallAttempt)
					}
					return "", errPreviewActionCallAttempt
				}
				return e.invokeRegisteredActionCall(ctx, interview, flowMeta, action, user, registration)
			case FlowActionCallPreviewPolicyRejectEmailIO:
				err := previewEmailIOAttemptError(action.ActionType)
				if action.FaultConnector != nil {
					return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
				}
				return "", err
			case FlowActionCallPreviewPolicyInvokeHandler:
				return e.invokeRegisteredActionCall(ctx, interview, flowMeta, action, user, registration)
			}
		}
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, errPreviewActionCallAttempt)
		}
		return "", errPreviewActionCallAttempt
	}

	if registeredActionType {
		return e.invokeRegisteredActionCall(ctx, interview, flowMeta, action, user, registration)
	}
	if flowactions.IsPlatformReservedActionType(action.ActionType) {
		err := fmt.Errorf("flow actionCall type %q is reserved by the platform runtime but its handler is not installed", strings.TrimSpace(action.ActionType))
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	// Preserve the existing action-call behavior for metadata-backed actions.
	inputRecord := models.SObject{}
	for _, param := range action.InputParameters {
		value, _, resolveErr := e.ResolveFlowResource(ctx, interview, user, param.Value)
		if resolveErr != nil {
			if action.FaultConnector != nil {
				return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
		inputRecord[param.Name] = value
	}

	// Get action metadata through the live lookup (COALESCE(IsDeleted,0)=0):
	// a soft-deleted action must not execute via flow action-call, by id or by
	// name. A read error is an internal fault, not "action missing" — it must
	// not consume the fault connector.
	actionMeta, lookupErr := e.metadata.GetLiveActionByID(ctx, action.ActionName)
	if lookupErr != nil {
		return "", apperrors.NewInternalError(fmt.Sprintf("failed to resolve action '%s'", action.ActionName), lookupErr)
	}
	if actionMeta == nil {
		// Try by name via the metadata service (live-filtered list). A read
		// failure on this fallback is equally an internal fault — it must not
		// collapse into "action not found" (a false not-found would consume
		// the fault connector / report a missing action during a store outage).
		actions, listErr := e.metadata.GetAllActionsAndError(ctx)
		if listErr != nil {
			return "", apperrors.NewInternalError(fmt.Sprintf("failed to resolve action '%s'", action.ActionName), listErr)
		}
		for _, a := range actions {
			if a.Name == action.ActionName || a.ID == action.ActionName {
				actionMeta = a
				break
			}
		}
	}

	if actionMeta == nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, fmt.Errorf("Action '%s' not found", action.ActionName))
		}
		return "", apperrors.NewNotFoundError("action", action.ActionName)
	}

	// Execute using ActionService.ExecuteActionDirect
	templateVariables, err := e.flowActionTemplateVariables(ctx, interview, user, action)
	if err != nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	err = e.actions.ExecuteActionDirectWithVariables(ctx, actionMeta, inputRecord, templateVariables, user)
	if err != nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}

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

func (e *FlowExecutor) invokeRegisteredActionCall(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, action *models.FlowActionCall, user *models.UserSession, registration FlowActionCallRegistration) (string, error) {
	if err := registration.contextValidationError(ctx, action); err != nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	return registration.Handler(ctx, e, interview, flowMeta, action, user)
}

func previewEmailIOAttemptError(actionType string) error {
	actionType = strings.TrimSpace(actionType)
	if actionType == "" {
		actionType = "email"
	}
	return fmt.Errorf("flow preview: actionCall is not allowed in a no-DML preview run (%s action calls send email; email provider I/O is an external side effect)", actionType)
}

func connectorSyncOriginPresent(ctx context.Context) bool {
	origin, ok := ctx.Value(constants.ContextKeyConnectorSyncOrigin).(string)
	return ok && strings.TrimSpace(origin) != ""
}

func connectorSyncOriginMatchesOwner(ctx context.Context, ownerPack string) bool {
	origin, ok := ctx.Value(constants.ContextKeyConnectorSyncOrigin).(string)
	if !ok {
		return false
	}
	origin = strings.TrimSpace(origin)
	ownerPack = strings.TrimSpace(ownerPack)
	return origin != "" && ownerPack != "" && strings.EqualFold(ownerPack, origin)
}

func connectorSyncOriginMatchesCurrentFlow(ctx context.Context) bool {
	ownerPack := flowRuntimeContextFromCtx(ctx).CurrentFlowOwnerPack()
	return connectorSyncOriginMatchesOwner(ctx, ownerPack)
}

// normalizeEmailSimpleAction returns a shallow copy of the emailSimple action
// with its InputParameter keys remapped to the canonical sendEmail names
// (emailTemplate->templateId, recipient->to). Values are forwarded VERBATIM —
// the {stringValue:...} / {elementReference:...} typed-value wrapper is
// preserved so the downstream canonical builder (extractTypedValue /
// unwrapCanonicalSendEmailTypedValue) unwraps it identically to the sendEmail
// path. This is the ONLY place the legacy->canonical key mapping lives, which
// keeps buildCanonicalSendEmailRequest pure (it only ever sees the canonical
// keys templateId/to/cc/variables/from). Already-canonical keys and
// variables/cc pass through unchanged.
func normalizeEmailSimpleAction(action *models.FlowActionCall) *models.FlowActionCall {
	if action == nil {
		return action
	}
	normalized := *action
	normalized.InputParameters = make([]models.FlowActionParameter, len(action.InputParameters))
	for i, param := range action.InputParameters {
		name := strings.TrimSpace(param.Name)
		switch {
		case strings.EqualFold(name, constants.ConfigEmailTemplate):
			name = flowSendEmailTemplateIDParam
		case strings.EqualFold(name, constants.ConfigRecipient):
			name = constants.ConfigTo
		}
		normalized.InputParameters[i] = models.FlowActionParameter{Name: name, Value: param.Value}
	}
	return &normalized
}

func (e *FlowExecutor) flowActionTemplateVariables(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, action *models.FlowActionCall) (map[string]interface{}, error) {
	if e == nil || action == nil {
		return nil, nil
	}
	for _, param := range action.InputParameters {
		if !strings.EqualFold(strings.TrimSpace(param.Name), constants.ConfigVariables) {
			continue
		}
		return e.resolveFlowActionTemplateVariables(ctx, interview, user, param.Value)
	}
	return nil, nil
}

func (e *FlowExecutor) resolveFlowActionTemplateVariables(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, value interface{}) (map[string]interface{}, error) {
	switch variables := value.(type) {
	case map[string]interface{}:
		if _, isElementReference := variables["elementReference"]; isElementReference {
			return nil, nil
		}
		return e.resolveFlowActionTemplateVariableMap(ctx, interview, user, variables)
	default:
		return nil, nil
	}
}

func (e *FlowExecutor) resolveFlowActionTemplateVariableMap(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, source map[string]interface{}) (map[string]interface{}, error) {
	variables := make(map[string]interface{}, len(source))
	for key, value := range source {
		key = strings.TrimSpace(key)
		if key == "" || strings.HasPrefix(key, "$") {
			continue
		}
		resolved, _, err := e.ResolveFlowResource(ctx, interview, user, value)
		if err != nil {
			return nil, err
		}
		variables[key] = resolved
	}
	if len(variables) == 0 {
		return nil, nil
	}
	return variables, nil
}

func (e *FlowExecutor) executeSendEmailAction(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, action *models.FlowActionCall, user *models.UserSession) (string, error) {
	fault := func(err error) (string, error) {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	if e.actions == nil {
		return fault(apperrors.NewInternalError("sendEmail action unavailable: ActionService not wired into FlowExecutor", nil))
	}

	req, err := e.buildCanonicalSendEmailRequest(ctx, interview, flowMeta, action, user)
	if err != nil {
		return fault(err)
	}
	result, err := e.actions.executeCanonicalSendEmail(ctx, req)
	if err != nil {
		return fault(err)
	}
	assignActionCallOutputs(interview, action, emailSendResultOutputValues(result))

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

func (e *FlowExecutor) buildCanonicalSendEmailRequest(ctx context.Context, interview *models.FlowInterview, flowMeta *models.FlowMetadata, action *models.FlowActionCall, user *models.UserSession) (canonicalSendEmailRequest, error) {
	req := canonicalSendEmailRequest{
		Record: models.SObject{},
		User:   user,
	}
	if interview != nil {
		req.Record = coerceFlowRecord(interview.ContextData["$Record"])
	}
	if req.Record == nil {
		req.Record = models.SObject{}
	}
	if flowMeta != nil && flowMeta.Start != nil {
		req.ObjectAPIName = strings.TrimSpace(flowMeta.Start.Object)
	}
	if req.ObjectAPIName == "" {
		req.ObjectAPIName = strings.TrimSpace(flowRuntimeContextFromCtx(ctx).ObjectType())
	}
	if req.ObjectAPIName == "" {
		req.ObjectAPIName = strings.TrimSpace(models.RecordObjectType(req.Record))
	}

	for _, param := range action.InputParameters {
		name := strings.TrimSpace(param.Name)
		switch {
		case strings.EqualFold(name, flowSendEmailTemplateIDParam):
			resolved, err := e.deepResolveFlowResource(ctx, interview, user, param.Value)
			if err != nil {
				return req, err
			}
			templateID, err := canonicalSendEmailStringParam(flowSendEmailTemplateIDParam, unwrapCanonicalSendEmailTypedValue(resolved))
			if err != nil {
				return req, err
			}
			req.TemplateID = templateID
		case strings.EqualFold(name, constants.ConfigTo):
			to, err := e.resolveCanonicalSendEmailRecipients(ctx, interview, user, constants.ConfigTo, param.Value)
			if err != nil {
				return req, err
			}
			req.To = to
		case strings.EqualFold(name, constants.ConfigCc):
			cc, err := e.resolveCanonicalSendEmailRecipients(ctx, interview, user, constants.ConfigCc, param.Value)
			if err != nil {
				return req, err
			}
			req.Cc = cc
		case strings.EqualFold(name, constants.ConfigVariables):
			variables, err := e.resolveCanonicalSendEmailVariables(ctx, interview, user, param.Value)
			if err != nil {
				return req, err
			}
			req.Variables = variables
		case strings.EqualFold(name, flowSendEmailFromParam):
			resolved, err := e.deepResolveFlowResource(ctx, interview, user, param.Value)
			if err != nil {
				return req, err
			}
			senderPolicy, err := canonicalSendEmailSenderPolicyParam(flowSendEmailFromParam, unwrapCanonicalSendEmailTypedValue(resolved))
			if err != nil {
				return req, err
			}
			req.SenderPolicy = senderPolicy
		case strings.EqualFold(name, constants.ConfigBcc):
			return req, apperrors.NewValidationError(constants.ConfigBcc, "bcc is not supported by sendEmail")
		case strings.EqualFold(name, "sender"), strings.EqualFold(name, "senderAddress"), strings.EqualFold(name, "senderType"):
			return req, apperrors.NewValidationError(name, "WorkflowAlert sender fields are not supported by sendEmail; use from with a generic sender policy")
		case strings.EqualFold(name, "templateDeveloperName"):
			return req, apperrors.NewValidationError(name, "templateDeveloperName lookup is not supported by sendEmail; use templateId")
		}
	}
	return req, nil
}

func canonicalSendEmailStringParam(fieldName string, value interface{}) (string, error) {
	text, ok := value.(string)
	if !ok {
		return "", apperrors.NewValidationError(fieldName, fmt.Sprintf("%s must be a string", fieldName))
	}
	text = strings.TrimSpace(text)
	if text == "" {
		return "", apperrors.NewValidationError(fieldName, fmt.Sprintf("%s is required", fieldName))
	}
	return text, nil
}

func canonicalSendEmailSenderPolicyParam(fieldName string, value interface{}) (*canonicalSendEmailSenderPolicy, error) {
	switch v := value.(type) {
	case nil:
		return nil, apperrors.NewValidationError(fieldName, "from sender policy is required when from is supplied")
	case string:
		return canonicalSendEmailSenderPolicyFromType(fieldName, v)
	case map[string]interface{}:
		if _, isElementReference := v["elementReference"]; isElementReference {
			return nil, apperrors.NewValidationError(fieldName, "from sender policy reference is unresolved")
		}
		for _, explicitAddressKey := range []string{"address", "email", "senderAddress", "fromAddress"} {
			if _, ok := v[explicitAddressKey]; ok {
				return nil, apperrors.NewValidationError(fieldName, "explicit sender addresses are not supported by sendEmail; use a validated sender policy")
			}
		}
		rawType, ok := v["type"]
		if !ok {
			rawType, ok = v["policy"]
		}
		if !ok {
			return nil, apperrors.NewValidationError(fieldName, "from sender policy must declare type")
		}
		policyType, ok := rawType.(string)
		if !ok {
			return nil, apperrors.NewValidationError(fieldName, "from sender policy type must be a string")
		}
		return canonicalSendEmailSenderPolicyFromType(fieldName, policyType)
	default:
		return nil, apperrors.NewValidationError(fieldName, "from must be a sender policy string or object")
	}
}

func canonicalSendEmailSenderPolicyFromType(fieldName string, policyType string) (*canonicalSendEmailSenderPolicy, error) {
	policyType = strings.TrimSpace(policyType)
	if policyType == "" {
		return nil, apperrors.NewValidationError(fieldName, "from sender policy type is required")
	}
	if strings.Contains(policyType, "@") {
		return nil, apperrors.NewValidationError(fieldName, "explicit sender addresses are not supported by sendEmail; use a validated sender policy")
	}
	normalized := normalizeCanonicalSendEmailSenderPolicyType(policyType)
	switch normalized {
	case canonicalSendEmailSenderDefault:
		return &canonicalSendEmailSenderPolicy{Type: normalized}, nil
	default:
		return nil, apperrors.NewValidationError(fieldName, fmt.Sprintf("unsupported sendEmail sender policy %q; only default is supported until a generic sender-identity source of truth exists", policyType))
	}
}

func unwrapCanonicalSendEmailTypedValue(value interface{}) interface{} {
	if literal, wasLiteral := extractTypedValue(value); wasLiteral {
		return unwrapCanonicalSendEmailTypedValue(literal)
	}
	switch v := value.(type) {
	case map[string]interface{}:
		out := make(map[string]interface{}, len(v))
		for key, item := range v {
			out[key] = unwrapCanonicalSendEmailTypedValue(item)
		}
		return out
	case []interface{}:
		out := make([]interface{}, len(v))
		for i, item := range v {
			out[i] = unwrapCanonicalSendEmailTypedValue(item)
		}
		return out
	default:
		return value
	}
}

func (e *FlowExecutor) resolveCanonicalSendEmailRecipients(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, fieldName string, value interface{}) ([]string, error) {
	recipients, err := e.collectCanonicalSendEmailRecipients(ctx, interview, user, fieldName, nil, value, true)
	if err != nil {
		return nil, err
	}
	if len(recipients) == 0 && strings.EqualFold(fieldName, constants.ConfigTo) {
		return nil, apperrors.NewValidationError(fieldName, "sendEmail requires at least one to recipient")
	}
	return recipients, nil
}

func (e *FlowExecutor) collectCanonicalSendEmailRecipients(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, fieldName string, recipients []string, value interface{}, resolveReferences bool) ([]string, error) {
	switch v := value.(type) {
	case nil:
		return nil, apperrors.NewValidationError(fieldName, fmt.Sprintf("%s recipient is empty", fieldName))
	case map[string]interface{}:
		if _, isElementReference := v["elementReference"]; isElementReference && resolveReferences {
			resolved, present, err := e.ResolveFlowResource(ctx, interview, user, v)
			if err != nil {
				return nil, err
			}
			if !present || resolved == nil {
				return nil, unresolvedCanonicalSendEmailRecipientReferenceError(fieldName, v)
			}
			return e.collectCanonicalSendEmailRecipients(ctx, interview, user, fieldName, recipients, resolved, false)
		}
		if literal, wasLiteral := extractTypedValue(v); wasLiteral {
			return e.collectCanonicalSendEmailRecipients(ctx, interview, user, fieldName, recipients, literal, false)
		}
		return nil, apperrors.NewValidationError(fieldName, fmt.Sprintf("%s must be an email address string or collection of email address strings", fieldName))
	case string:
		if resolveReferences {
			resolved, present, err := e.ResolveFlowResource(ctx, interview, user, v)
			if err != nil {
				return nil, err
			}
			if !present || resolved == nil {
				return nil, unresolvedCanonicalSendEmailRecipientReferenceError(fieldName, v)
			}
			resolvedString, ok := resolved.(string)
			if !ok || resolvedString != v {
				return e.collectCanonicalSendEmailRecipients(ctx, interview, user, fieldName, recipients, resolved, false)
			}
		}
		address, err := validateCanonicalSendEmailAddress(fieldName, v)
		if err != nil {
			return nil, err
		}
		return append(recipients, address), nil
	case []string:
		for _, item := range v {
			var err error
			recipients, err = e.collectCanonicalSendEmailRecipients(ctx, interview, user, fieldName, recipients, item, resolveReferences)
			if err != nil {
				return nil, err
			}
		}
		return recipients, nil
	case []interface{}:
		for _, item := range v {
			var err error
			recipients, err = e.collectCanonicalSendEmailRecipients(ctx, interview, user, fieldName, recipients, item, resolveReferences)
			if err != nil {
				return nil, err
			}
		}
		return recipients, nil
	default:
		return nil, apperrors.NewValidationError(fieldName, fmt.Sprintf("%s must be an email address string or collection of email address strings", fieldName))
	}
}

func unresolvedCanonicalSendEmailRecipientReferenceError(fieldName string, value interface{}) error {
	if ref := canonicalSendEmailRecipientReferenceName(value); ref != "" {
		return apperrors.NewValidationError(fieldName, fmt.Sprintf("%s recipient reference %s is unresolved", fieldName, ref))
	}
	return apperrors.NewValidationError(fieldName, fmt.Sprintf("%s recipient reference is unresolved", fieldName))
}

func canonicalSendEmailRecipientReferenceName(value interface{}) string {
	if m, ok := value.(map[string]interface{}); ok {
		if ref, ok := m["elementReference"].(string); ok {
			return strings.TrimSpace(ref)
		}
	}
	if ref, ok := value.(string); ok && strings.Contains(ref, ".") && looksLikeVariableReference(ref) {
		return strings.TrimSpace(ref)
	}
	return ""
}

func (e *FlowExecutor) resolveCanonicalSendEmailVariables(ctx context.Context, interview *models.FlowInterview, user *models.UserSession, value interface{}) (map[string]interface{}, error) {
	if m, ok := value.(map[string]interface{}); ok {
		if _, isElementReference := m["elementReference"]; isElementReference {
			return nil, apperrors.NewValidationError(constants.ConfigVariables, "variables must be an object literal")
		}
	}
	source, ok := value.(map[string]interface{})
	if !ok {
		return nil, apperrors.NewValidationError(constants.ConfigVariables, "variables must be an object literal")
	}
	variables := make(map[string]interface{}, len(source))
	for key, rawValue := range source {
		key = strings.TrimSpace(key)
		if key == "" || strings.HasPrefix(key, "$") {
			return nil, apperrors.NewValidationError(constants.ConfigVariables, "variables keys must be non-empty names without a $ prefix")
		}
		resolved, err := e.deepResolveFlowResource(ctx, interview, user, rawValue)
		if err != nil {
			return nil, err
		}
		variables[key] = unwrapCanonicalSendEmailTypedValue(resolved)
	}
	if len(variables) == 0 {
		return nil, nil
	}
	return variables, nil
}

func emailSendResultOutputValues(result *ports.EmailSendResult) map[string]interface{} {
	outputs := map[string]interface{}{
		flowSendEmailSentCountOutput:  0,
		flowSendEmailRecipientsOutput: []interface{}{},
	}
	if result == nil {
		return outputs
	}
	outputs[flowSendEmailSentCountOutput] = result.SentCount
	recipients := make([]interface{}, 0, len(result.RecipientResults))
	for _, recipient := range result.RecipientResults {
		row := map[string]interface{}{
			"address": recipient.Address,
			"role":    recipient.Role,
			"status":  recipient.Status,
		}
		if recipient.Error != "" {
			row["error"] = recipient.Error
		}
		recipients = append(recipients, row)
	}
	outputs[flowSendEmailRecipientsOutput] = recipients
	if result.MessageID != "" {
		outputs[flowSendEmailMessageIDOutput] = result.MessageID
	}
	return outputs
}

func assignActionCallOutputs(interview *models.FlowInterview, action *models.FlowActionCall, outputs map[string]interface{}) {
	if interview == nil || action == nil || len(outputs) == 0 {
		return
	}
	if interview.ContextData == nil {
		interview.ContextData = map[string]interface{}{}
	}
	for _, param := range action.OutputParameters {
		if param.Name == "" {
			continue
		}
		target := extractFlowOutputTarget(param.Value)
		if target == "" {
			continue
		}
		if val, ok := outputs[param.Name]; ok {
			setNestedContextValue(interview.ContextData, target, val)
		}
	}
}

func coerceFlowRecord(value interface{}) models.SObject {
	switch record := value.(type) {
	case models.SObject:
		return record
	case map[string]interface{}:
		return models.SObject(record)
	default:
		return nil
	}
}

// executeFlowAction dispatches an actionCall whose actionType is "flow" —
// Salesforce's alternative to the native <subflows> element. Input parameters
// map to the child flow's input variables, output parameters assign the
// child's returned output values back into the parent's ContextData.
//
// The call is synchronous (parent blocks until child completes) and is guarded
// against direct recursion and cycles via the call stack on FlowRuntimeContext
// (stored via flowRuntimeContextKey on ctx). The stack also caps overall depth
// at maxFlowCallDepth to bound runaway chains.
//
// Fault handling mirrors the existing executeSubflow behaviour: any error or
// non-success result from the child is routed through action.FaultConnector
// when present, otherwise the error bubbles up to abort the parent flow.
func (e *FlowExecutor) executeFlowAction(ctx context.Context, interview *models.FlowInterview, action *models.FlowActionCall, user *models.UserSession) (string, error) {
	if action.ActionName == "" {
		err := apperrors.NewValidationError("actionName", "flow action call requires an actionName (child flow API name)")
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	childCtx, cycleErr := withFlowCall(ctx, action.ActionName)
	if cycleErr != nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, cycleErr)
		}
		return "", cycleErr
	}
	childCtx, scopeErr := e.standardActionScopedChildContext(childCtx, action.ActionName, user)
	if scopeErr != nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, scopeErr)
		}
		return "", scopeErr
	}

	// Build child flow inputs: parameter name -> resolved value from parent
	// context. Values may be literals or elementReference strings; resolveValue
	// already unwraps the Salesforce typed-value wrappers.
	inputs := make(map[string]interface{}, len(action.InputParameters))
	for _, param := range action.InputParameters {
		if param.Name == "" {
			continue
		}
		value, _, resolveErr := e.ResolveFlowResource(ctx, interview, user, param.Value)
		if resolveErr != nil {
			if action.FaultConnector != nil {
				return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
		inputs[param.Name] = value
	}

	childCtx, resultErrorOrigin := withFlowExecutionResultErrorOrigin(childCtx)
	result, err := e.ExecuteFlow(childCtx, action.ActionName, inputs, user)
	if err != nil {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", fmt.Errorf("flow action '%s' failed: %w", action.ActionName, err)
	}
	if result == nil || !result.IsSuccess {
		// ExecuteFlow intentionally preserves its legacy result shape. A child
		// can therefore fail before entering executeFlowVersion (for example a
		// missing, invalid, or protected flow) without returning an error. Save
		// that result-level taxonomy before wrapping it below, otherwise the
		// parent can only classify the generic formatted error as FLOW_ERROR.
		cause := classifyFailedFlowExecutionResultAtElement(result, resultErrorOrigin.origin, flowCauseElementType(ctx), interviewElementName(interview))
		publishFailedChildFlowResultCause(ctx, cause)
		failure := errorFromFailedFlowExecutionResult(
			result,
			fmt.Sprintf("flow action '%s' returned failure", action.ActionName),
			action.ActionName,
			resultErrorOrigin.origin,
		)
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, failure)
		}
		return "", failure
	}

	assignActionCallOutputs(interview, action, result.OutputValues)

	if result.IsRetriable {
		if rt := flowRuntimeContextFromCtx(ctx); rt != nil {
			rt.SetRetriable()
		}
	}

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

// extractFlowOutputTarget pulls the parent-side variable reference out of an
// output parameter's Value. Salesforce serialises it either as a bare string
// ("parentVar") or as an elementReference-typed wrapper
// ({"elementReference": "parentVar"}). Anything else yields "" — the caller
// should skip the assignment rather than fault, matching Salesforce's lenient
// behaviour when an outputParameter lacks a target.
func extractFlowOutputTarget(value interface{}) string {
	unwrapped, _ := extractTypedValue(value)
	if s, ok := unwrapped.(string); ok {
		return s
	}
	return ""
}

// executeSubflow executes another flow
func (e *FlowExecutor) executeSubflow(ctx context.Context, interview *models.FlowInterview, subflow *models.FlowSubflow, 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:        subflow.Name,
			ElementType:      "Subflow",
			Status:           status,
			DurationMs:       int(time.Since(start).Milliseconds()),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
	}()
	// Guard against cycles and runaway depth for native <subflows> calls too —
	// the same stack protects both actionType="flow" action calls and
	// <subflows> elements so a chain that mixes the two cannot loop forever.
	childCtx, cycleErr := withFlowCall(ctx, subflow.FlowName)
	if cycleErr != nil {
		if subflow.FaultConnector != nil {
			return e.routeFault(ctx, interview, subflow.FaultConnector.TargetReference, cycleErr)
		}
		return "", cycleErr
	}
	childCtx = withConnectorInboundCompanionSubflowDelegation(childCtx, subflow.FlowName)
	childCtx, scopeErr := e.standardActionScopedChildContext(childCtx, subflow.FlowName, user)
	if scopeErr != nil {
		if subflow.FaultConnector != nil {
			return e.routeFault(ctx, interview, subflow.FaultConnector.TargetReference, scopeErr)
		}
		return "", scopeErr
	}

	// Build inputs. For sObject inputs passed to the child flow, stamp neutral
	// object-type metadata so the child can resolve __r traversal.
	parentObjectType := flowRuntimeContextFromCtx(ctx).ObjectType()
	inputs := make(map[string]interface{})
	for _, input := range subflow.InputAssignments {
		val, _, resolveErr := e.ResolveFlowResource(ctx, interview, user, input.Value)
		if resolveErr != nil {
			if subflow.FaultConnector != nil {
				return e.routeFault(ctx, interview, subflow.FaultConnector.TargetReference, resolveErr)
			}
			return "", resolveErr
		}
		if ref, ok := flowElementReferenceName(input.Value); ok && parentObjectType != "" {
			if ref == "$Record" || ref == "$Record__Old" || ref == "$Record__Prior" {
				if rec, ok := val.(models.SObject); ok {
					models.StampObjectType(rec, parentObjectType)
				} else if rec, ok := val.(map[string]interface{}); ok {
					models.StampObjectType(models.SObject(rec), parentObjectType)
				}
			}
		}
		inputs[input.Name] = val
	}

	// Execute the subflow
	childCtx, resultErrorOrigin := withFlowExecutionResultErrorOrigin(childCtx)
	result, err := e.ExecuteFlow(childCtx, subflow.FlowName, inputs, user)
	if err != nil {
		if subflow.FaultConnector != nil {
			return e.routeFault(ctx, interview, subflow.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	if result == nil || !result.IsSuccess {
		cause := classifyFailedFlowExecutionResultAtElement(result, resultErrorOrigin.origin, flowCauseElementType(ctx), interviewElementName(interview))
		publishFailedChildFlowResultCause(ctx, cause)
		failure := errorFromFailedFlowExecutionResult(
			result,
			fmt.Sprintf("subflow %q failed", subflow.FlowName),
			subflow.FlowName,
			resultErrorOrigin.origin,
		)
		if subflow.FaultConnector != nil {
			return e.routeFault(ctx, interview, subflow.FaultConnector.TargetReference, failure)
		}
		// No fault connector declared → subflow failure MUST propagate. A
		// parent running under RunAtomically has its tx rolled back by the
		// returned error; a non-atomic parent surfaces the child error
		// cleanly. Silently falling through to the normal Connector here
		// previously let required bookkeeping subflows fail while still
		// committing the parent action, defeating the atomicity contract.
		// If a flow author wants the "child failure is optional" semantics,
		// they must declare a faultConnector explicitly.
		return "", fmt.Errorf("subflow %q failed without fault connector: %w", subflow.FlowName, failure)
	}

	// Store outputs
	for _, output := range subflow.OutputAssignments {
		if val, ok := result.OutputValues[output.Name]; ok {
			interview.ContextData[output.AssignToReference] = val
		}
	}

	if result.IsRetriable {
		if rt := flowRuntimeContextFromCtx(ctx); rt != nil {
			rt.SetRetriable()
		}
	}

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

func (e *FlowExecutor) executePreviewRecordUpdate(ctx context.Context, interview *models.FlowInterview, action *models.FlowActionCall, user *models.UserSession) (string, error) {
	fault := func(err error) (string, error) {
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}

	params := map[string]interface{}{}
	for _, p := range action.InputParameters {
		resolved, err := e.deepResolveFlowResource(ctx, interview, user, p.Value)
		if err != nil {
			return fault(err)
		}
		params[p.Name] = resolved
	}

	objectName, _ := params["object"].(string)
	recordID, _ := params["recordId"].(string)
	outputReference, _ := params["outputReference"].(string)
	updates, _ := params["updates"].(map[string]interface{})
	rawInternalUpdates, _ := params["internalUpdates"].(map[string]interface{})
	if outputReference == "" {
		for _, p := range action.InputParameters {
			if p.Name != "outputReference" {
				continue
			}
			if raw, _ := extractTypedValue(p.Value); raw != nil {
				if s, ok := raw.(string); ok {
					outputReference = s
				}
			}
			break
		}
	}
	if outputReference == "" {
		outputReference = "$Record"
	}
	if strings.TrimSpace(objectName) == "" || strings.TrimSpace(recordID) == "" {
		return fault(fmt.Errorf("previewRecordUpdate requires object and recordId"))
	}
	if len(updates) == 0 && len(rawInternalUpdates) == 0 {
		return fault(fmt.Errorf("previewRecordUpdate requires updates or internalUpdates"))
	}

	// Filter out empty-string values from internalUpdates so optional internal
	// stamp fields do not fail schema validation as non-empty reference fields.
	// Preserve explicit null: protected standard actions use it to clear
	// system-managed references that are intentionally non-updateable through
	// public CRUD. Missing optional references resolve to "" and are skipped;
	// an authored JSON null is a real write value, not a missing input.
	var internalStamps models.SObject
	if len(rawInternalUpdates) > 0 {
		internalStamps = make(models.SObject, len(rawInternalUpdates))
		for k, v := range rawInternalUpdates {
			if s, ok := v.(string); ok && s == "" {
				continue
			}
			internalStamps[k] = v
		}
		if len(internalStamps) == 0 {
			internalStamps = nil
		}
	}

	prepared, err := e.persistence.PreviewBeforeUpdate(ctx, objectName, recordID, models.SObject(updates), internalStamps, user)
	if err != nil {
		return fault(flowRecordMutationValidation(err))
	}
	if prepared != nil {
		flowRuntimeContextFromCtx(ctx).SetPreparedRecordUpdate(objectName, recordID, prepared)
		if prepared.Record != nil {
			preview := cloneActionSObject(prepared.Record)
			models.StampObjectType(preview, objectName)
			interview.ContextData[outputReference] = preview
		}
	}

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

func cloneActionSObject(in models.SObject) models.SObject {
	if in == nil {
		return models.SObject{}
	}
	out := make(models.SObject, len(in))
	for k, v := range in {
		out[k] = v
	}
	return out
}

// executeRegisterConnectorInboundRemoteAuthorship is the implementation of the
// "registerConnectorInboundRemoteAuthorship" generic invocable action (PR-2 of
// epic #2028). It is called from an inbound connector flow that runs atomically
// (runAtomically:true / RunInPessimisticTransaction), so the ambient transaction
// is present when the store was constructed with a txExtractor.
//
// Input parameters (all pack-declared literals — no vendor-specific strings):
//   - connectorOwnerKey  string — pack owner key that minted this inbound write
//   - connectorNamespace string — connector namespace (pack-level discriminator)
//   - sourceObject       string — SObject API name whose local record was upserted
//   - remoteIDField      string — API name of the field that holds the remote id
//
// Behavior (no-op on mismatch, idempotent on same-value, retire on conflict):
//  1. Guard: if the ambient connector sync origin does not match connectorOwnerKey,
//     return nil — this is not a connector-origin write. Never fault the flow.
//  2. Read sourceRecordId from $Record.Id (engine-confirmed local id).
//     Read remoteId from $Record[remoteIDField] (the stamped value). If either
//     is blank, return nil — nothing to prove.
//  3. Look up the existing active evidence row for this (owner, object, record, field).
//     - Not found → InsertAuthoredEvidence; on ErrConnectorPushDurableConflict
//     (race) call RetireEvidence on the racing row if found, then return.
//     - Found, same remoteId → no-op (already proved, idempotent).
//     - Found, different remoteId → RetireEvidence (ambiguous; outbound fails closed).
func (e *FlowExecutor) executeRegisterConnectorInboundRemoteAuthorship(
	ctx context.Context,
	interview *models.FlowInterview,
	action *models.FlowActionCall,
) (string, error) {
	// Resolve input parameters.
	params := make(map[string]string, len(action.InputParameters))
	for _, p := range action.InputParameters {
		v, _ := p.Value.(map[string]interface{})
		if sv, ok := v["stringValue"].(string); ok {
			params[p.Name] = strings.TrimSpace(sv)
		} else if sv, ok := p.Value.(string); ok {
			params[p.Name] = strings.TrimSpace(sv)
		} else {
			// Try resolving as an element reference via the existing helper.
			value, _, err := e.ResolveFlowResource(ctx, interview, nil, p.Value)
			if err != nil {
				return "", err
			}
			resolved := fmt.Sprintf("%v", value)
			params[p.Name] = strings.TrimSpace(resolved)
		}
	}
	connectorOwnerKey := params["connectorOwnerKey"]
	connectorNamespace := params["connectorNamespace"]
	sourceObject := params["sourceObject"]
	remoteIDField := params["remoteIDField"]

	// Pack-declared evidence-key components must be present before the ctx gate.
	// A blank connectorNamespace is a pack misconfiguration: connectorSyncOriginMatchesOwner
	// returns false for any blank namespace (ownerPack != "" guard), silently
	// no-oping instead of reporting the misconfiguration. Fail closed here,
	// consistent with the sourceObject / remoteIDField blank-key checks below.
	if connectorNamespace == "" {
		return "", fmt.Errorf("register inbound authorship: connectorNamespace is a required pack-declared evidence-key component and must not be blank")
	}

	// The flow loop treats an empty next-element reference as flow completion, so
	// every non-error exit must return the action's next reference to keep the
	// flow advancing (matches the other in-flow connector actions).
	next := ""
	if action.Connector != nil {
		next = action.Connector.TargetReference
	}

	// Step 1: ctx gate — unforgeable: ContextKeyConnectorSyncOrigin is set ONLY by
	// the inbound worker, to the connector NAMESPACE (connector_inbound/worker.go
	// stamps w.cfg.Namespace; flow_adapter.go reuses it). Gate on connectorNamespace
	// — the value the worker actually stamps — NOT connectorOwnerKey, which is the
	// pack name and may differ from the namespace, so an ownerKey gate would never
	// match in production and silently skip authorship.
	if !connectorSyncOriginMatchesOwner(ctx, connectorNamespace) {
		return next, nil
	}
	authenticatedNamespace := constants.ConnectorSyncOriginFromContext(ctx)
	if authenticatedNamespace == "" {
		return "", fmt.Errorf("register inbound authorship: authenticated connector namespace is required")
	}
	connectorNamespace = authenticatedNamespace

	// Step 1b: bind the evidence owner to the TRUSTED executing flow identity
	// (mirrors applyConnectorFieldMappings in executor_connector_field_mapping.go).
	// The evidence owner is the pack that owns the running uncustomized standard
	// Flow — NOT the free-text connectorOwnerKey param — so a flow cannot author or
	// retire durable authorship evidence under another connector's owner. A failed
	// binding is fatal: the inbound record rolls back rather than committing with
	// unauthenticated authorship.
	flowIdentity, ok := flowRuntimeContextFromCtx(ctx).CurrentFlow()
	trustedOwner := strings.TrimSpace(flowIdentity.OwnerPack)
	if !ok || trustedOwner == "" {
		return "", fmt.Errorf("register inbound authorship: executing Flow owner pack is required")
	}
	if !flowIdentityIsPackOwnedStandard(flowIdentity) {
		return "", fmt.Errorf("register inbound authorship: Flow %q must be an uncustomized standard FlowVersion to author connector evidence", flowIdentity.APIName)
	}
	if connectorOwnerKey != "" && connectorOwnerKey != trustedOwner {
		return "", fmt.Errorf("register inbound authorship: connectorOwnerKey %q does not match executing Flow owner pack %q", connectorOwnerKey, trustedOwner)
	}
	// The trusted flow identity is the SSoT for the evidence owner.
	connectorOwnerKey = trustedOwner

	// Step 1b-ii: defense-in-depth — only a genuine inbound-worker execution
	// (stamped by constants.WithConnectorInboundFlowExecution in flow_adapter.go)
	// may author. By this point sync-origin matched and the flow is a trusted
	// pack-owned standard flow, so the source record already carries a stamped
	// remote-id. A missing/mismatched inbound marker therefore FAILS CLOSED
	// (errors, rolling back the atomic flow) rather than silently skipping — a
	// silent skip would let the record commit with a remote-id but no evidence
	// row, a state the outbound adopt path treats as a hard failure (strand).
	if !constants.ConnectorInboundExecutionAuthorizesFlow(ctx, flowIdentity.APIName, connectorNamespace) {
		return "", fmt.Errorf("register inbound authorship: Flow %q is not executing within a genuine inbound-worker connector flow (namespace %q); refusing to skip authoring because the record already carries a stamped remote-id and would strand without evidence", flowIdentity.APIName, connectorNamespace)
	}

	// Step 1c: the evidence write MUST be atomic with the inbound record. The
	// store's ambient execer/querier fall back to s.db when no ambient tx is
	// present, so OUTSIDE an atomic Flow boundary the Insert/Retire would commit
	// immediately, decoupled from the record's rollback (orphan evidence, or a
	// conflict-retire that destroys prior authorship with no rollback). Require
	// the atomic boundary — the inverse of the outbound PushToConnector guard —
	// and fail closed otherwise.
	if !isWithinAtomicFlowBoundary(ctx) || e.persistence == nil || !e.persistence.IsTransactionActive(ctx) {
		return "", fmt.Errorf("register inbound authorship: must execute within an atomic Flow transaction (runAtomically) WITH a live DB transaction so the evidence write is atomic with the inbound record — the durable store falls back to s.db when no ambient tx is present, and the atomic-boundary marker survives DetachTx, so IsTransactionActive is required for true atomicity (symmetric to the publishPostUpsertRecord gate)")
	}
	// Step 1d: the pack-declared evidence-key components must be present (these are
	// flow metadata config, distinct from the per-record values read below; a blank
	// one is a pack misconfiguration and fails closed, mirroring applyConnectorFieldMappings).
	if sourceObject == "" {
		return "", fmt.Errorf("register inbound authorship: sourceObject must not be blank")
	}
	if remoteIDField == "" {
		return "", fmt.Errorf("register inbound authorship: remoteIDField must not be blank")
	}
	installedBinding, err := e.resolveConnectorInboundRegisterInstalledBinding(ctx, connectorOwnerKey, sourceObject, remoteIDField)
	if err != nil {
		return "", fmt.Errorf("register inbound authorship: resolve locked installed destination: %w", err)
	}

	// Step 2: read engine-confirmed ids from the engine-owned snapshot channel.
	// Identity is NOT read from interview.ContextData["$Record"] — that is
	// flow-writable and could be forged by a flow author to misbind authorship
	// evidence to the wrong local row (the security fix for issue #2080).
	//
	// Resolution (registry-only, carve-down):
	//   PostUpsertRecords: dedup by recordId (same-record multi-update is legal —
	//   chained $Record writes each append a snapshot with the same Id). Count
	//   DISTINCT ids, not raw snapshots.
	//       0 distinct ids → no-op only when the engine reports no source-object
	//                         write; otherwise fail closed because a confirmed
	//                         write lost its required snapshot.
	//       1 distinct id  → GetLock the committed record for remoteIDField.
	//       >1 distinct ids → fail closed (genuinely ambiguous).
	//   (The case-0 RecordTriggerNewSnapshot fallback was removed as unreachable
	//   dead code — Gate-0 v6; see the case 0 branch below.)
	frc := flowRuntimeContextFromCtx(ctx)
	recs := frc.PostUpsertRecords(sourceObject)

	// Dedup by recordId (trim whitespace, skip blank).
	seen := make(map[string]struct{})
	var distinctIDs []string
	for _, rec := range recs {
		id, _ := rec["Id"].(string)
		id = strings.TrimSpace(id)
		if id == "" {
			continue
		}
		if _, dup := seen[id]; !dup {
			seen[id] = struct{}{}
			distinctIDs = append(distinctIDs, id)
		}
	}

	sourceRecordId := ""
	remoteId := ""

	switch len(distinctIDs) {
	case 0:
		if frc.HasPostUpsertWrite(sourceObject) {
			return "", fmt.Errorf("register inbound authorship: engine wrote %q but no engine-confirmed post-upsert record is available; refusing to commit without durable authorship evidence", sourceObject)
		}
		return next, nil
	case 1:
		// Engine-confirmed recordId from the post-upsert registry (deduped to
		// exactly one distinct id). Read remoteIDField ONLY from the committed
		// record via GetLock — the snapshot-scan path that used registry snapshot
		// values has been removed (carve-down for #2080). before-save register
		// is structurally impossible (the engine rejects before-save flows
		// containing ActionCalls at flow entry), so GetLock always has a
		// committed record to read from.
		recordId := distinctIDs[0]
		sourceRecordId = recordId
		if e.persistence == nil {
			return "", fmt.Errorf("register inbound authorship: persistence provider is not wired")
		}
		committedRec, err := e.persistence.GetLock(ctx, sourceObject, recordId)
		if err != nil {
			return "", fmt.Errorf("register inbound authorship: GetLock %s/%s: %w", sourceObject, recordId, err)
		}
		if committedRec == nil {
			return "", fmt.Errorf("register inbound authorship: committed record %s/%s not found", sourceObject, recordId)
		}
		if rv, ok := committedRec[remoteIDField]; ok && rv != nil {
			remoteId = strings.TrimSpace(fmt.Sprintf("%v", rv))
		}
	default:
		// More than one DISTINCT record of sourceObject was upserted: cannot
		// determine which is the authoritative inbound target → fail closed.
		return "", ErrAmbiguousPostUpsertf(len(distinctIDs), sourceObject)
	}

	if remoteId == "" {
		// The engine-confirmed record exists, but there is no stamped remote id to
		// author. This is a normal no-op: no remote-id mirror can be stranded.
		return next, nil
	}

	if e.connectorPushDurable == nil {
		// After the trusted gates have positively identified a genuine inbound
		// connector sync that must author evidence, a missing durable store is a
		// wiring fault, not a no-op: committing the inbound record without
		// authorship evidence would silently break the evidence SSoT. Fail closed.
		return "", fmt.Errorf("register inbound authorship: durable store is not wired")
	}
	registrationStore, ok := e.connectorPushDurable.(ConnectorRemoteIdentityRegistrationStore)
	if !ok {
		return "", fmt.Errorf("register inbound authorship: durable store does not support remote identity admission")
	}
	identity := installedBinding.Identity
	identity.SourceRecordID = sourceRecordId
	if err := registrationStore.LockConnectorRemoteIdentityLaneInAmbientTx(ctx, identity, remoteId); err != nil {
		return "", fmt.Errorf("register inbound authorship: lock remote identity admission: %w", err)
	}
	alreadyFinalized, err := registrationStore.CheckConnectorRemoteIdentityRegistrationInAmbientTx(ctx, identity, remoteId)
	if err != nil {
		return "", fmt.Errorf("register inbound authorship: check remote identity history: %w", err)
	}
	if alreadyFinalized {
		return next, nil
	}

	// Step 3: idempotency + conflict resolution. Every durable store error is
	// FATAL: the evidence write must be atomic with the inbound record, so a
	// failure returns an error (rolling back the record) rather than committing
	// the record without matching authorship evidence.
	// In-tx read: use the ambient-tx querier so this idempotency check observes
	// evidence the SAME inbound flow tx wrote earlier (a flow that registers the
	// same (owner,object,record,field) more than once must see its own pending
	// write, or it would suppress the conflict path and commit stale authorship).
	existing, err := e.connectorPushDurable.FindActiveInboundRemoteIDEvidenceInTx(
		ctx, connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField,
	)
	if err != nil {
		return "", fmt.Errorf("register inbound authorship: lookup active evidence: %w", err)
	}

	if existing != nil {
		if existing.RemoteId == remoteId {
			// Idempotent: same value already authored.
			return next, nil
		}
		// Different value: ambiguous re-author — retire so outbound fails closed.
		if err := e.connectorPushDurable.RetireEvidence(ctx, existing.ID); err != nil {
			return "", fmt.Errorf("register inbound authorship: retire ambiguous evidence: %w", err)
		}
		return next, nil
	}

	// Not found: insert.
	insertErr := e.connectorPushDurable.InsertAuthoredEvidence(ctx, ConnectorInboundRemoteIDEvidence{
		ConnectorOwnerKey:  connectorOwnerKey,
		ConnectorNamespace: connectorNamespace,
		SourceObject:       sourceObject,
		SourceRecordId:     sourceRecordId,
		RemoteIDField:      remoteIDField,
		RemoteId:           remoteId,
	})
	if insertErr != nil {
		// ErrConnectorPushDurableConflict: a concurrent write raced us. Retire the
		// row that won the race so outbound fails closed (ambiguous); the record
		// itself may continue. A failed find/retire in this path is FATAL.
		if errors.Is(insertErr, ErrConnectorPushDurableConflict) {
			// Cross-tx reread: the in-tx idempotency check above already saw any
			// row this tx wrote, so a unique conflict here means a CONCURRENT
			// transaction committed the racing row. Use the *sql.DB read (not the
			// ambient tx) so we can observe that committed row — it is outside this
			// tx's snapshot — and retire it.
			racing, findErr := e.connectorPushDurable.FindActiveInboundRemoteIDEvidence(
				ctx, connectorOwnerKey, sourceObject, sourceRecordId, remoteIDField,
			)
			if findErr != nil {
				return "", fmt.Errorf("register inbound authorship: lookup racing evidence after conflict: %w", findErr)
			}
			// Same-value race is idempotent: a concurrent tx authored the SAME
			// remote id for this key, so the active evidence is already correct —
			// do NOT retire it (that would erase valid authorship). Only a
			// different-value race is ambiguous and must be retired so outbound
			// fails closed (mirrors the pre-insert idempotency branch above).
			if racing != nil && racing.RemoteId != remoteId {
				if err := e.connectorPushDurable.RetireEvidence(ctx, racing.ID); err != nil {
					return "", fmt.Errorf("register inbound authorship: retire racing evidence after conflict: %w", err)
				}
			}
			return next, nil
		}
		// FATAL: a non-conflict insert failure rolls back the inbound record.
		return "", fmt.Errorf("register inbound authorship: insert authored evidence: %w", insertErr)
	}
	return next, nil
}

func (e *FlowExecutor) executeTransferConnectorRemoteIdentity(
	ctx context.Context,
	interview *models.FlowInterview,
	action *models.FlowActionCall,
	user *models.UserSession,
) (string, error) {
	next := ""
	if action.Connector != nil {
		next = action.Connector.TargetReference
	}
	flowIdentity, ok := flowRuntimeContextFromCtx(ctx).CurrentFlow()
	if !ok || !connectorRemoteIdentityTransferExecutionAuthorized(ctx, flowIdentity) {
		err := fmt.Errorf("transfer connector remote identity: requires an exact authorized uncustomized pack-owned standard action or frozen connector-inbound companion FlowVersion")
		if action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	if !isWithinAtomicFlowBoundary(ctx) || e.persistence == nil || !e.persistence.IsTransactionActive(ctx) || !hasAtomicFlowPessimisticRetryOwner(ctx) {
		return "", fmt.Errorf("transfer connector remote identity: must execute within an atomic Flow transaction owned by full-flow pessimistic retry")
	}
	if e.connectorPushDurable == nil {
		return "", fmt.Errorf("transfer connector remote identity: durable store is not wired")
	}

	params := make(map[string]string, len(action.InputParameters))
	for _, p := range action.InputParameters {
		if v, ok := p.Value.(map[string]interface{}); ok {
			if sv, ok := v["stringValue"].(string); ok {
				params[p.Name] = strings.TrimSpace(sv)
				continue
			}
		}
		if sv, ok := p.Value.(string); ok {
			params[p.Name] = strings.TrimSpace(sv)
			continue
		}
		resolved, _, err := e.ResolveFlowResource(ctx, interview, nil, p.Value)
		if err != nil {
			return "", err
		}
		if resolved != nil {
			params[p.Name] = strings.TrimSpace(fmt.Sprintf("%v", resolved))
		}
	}
	sourceObject := params["sourceObject"]
	sourceID := params["sourceId"]
	targetObject := params["targetObject"]
	targetID := params["targetId"]
	remoteIDField := params["remoteIDField"]
	outputReference := params["outputReference"]
	if sourceObject == "" || sourceID == "" || targetObject == "" || targetID == "" || remoteIDField == "" {
		return "", fmt.Errorf("transfer connector remote identity: sourceObject, sourceId, targetObject, targetId, and remoteIDField are required")
	}
	setTransferResult := func(transferred bool) {
		if outputReference == "" {
			return
		}
		if interview.ContextData == nil {
			interview.ContextData = make(map[string]interface{})
		}
		interview.ContextData[outputReference] = transferred
	}

	hintStore, ok := e.connectorPushDurable.(ConnectorRemoteIdentityTransferHintStore)
	if !ok {
		return "", fmt.Errorf("transfer connector remote identity: durable store does not support installed transfer admission")
	}
	initialHints, err := hintStore.FindInboundRemoteIdentityTransferHintsInAmbientTx(
		ctx,
		sourceObject,
		sourceID,
		targetObject,
		targetID,
		remoteIDField,
	)
	if err != nil {
		return "", fmt.Errorf("transfer connector remote identity: discover initial non-authoritative transfer hints: %w", err)
	}
	transfer := ConnectorRemoteIdentityTransfer{
		SourceObject:   sourceObject,
		SourceRecordId: sourceID,
		TargetObject:   targetObject,
		TargetRecordId: targetID,
		RemoteIDField:  remoteIDField,
	}
	transfer.admissions, err = e.resolveConnectorRemoteIdentityTransferAdmissions(ctx, transfer, initialHints)
	if err != nil {
		return "", fmt.Errorf("transfer connector remote identity: derive installed transfer admissions: %w", err)
	}
	if sourceObject == targetObject && sourceID == targetID {
		return "", fmt.Errorf("transfer connector remote identity: source and target records must be distinct")
	}
	type transferRecordLock struct {
		object string
		id     string
		source bool
	}
	recordLocks := []transferRecordLock{
		{object: sourceObject, id: sourceID, source: true},
		{object: targetObject, id: targetID},
	}
	sort.Slice(recordLocks, func(i, j int) bool {
		return recordLocks[i].object+"\x00"+recordLocks[i].id < recordLocks[j].object+"\x00"+recordLocks[j].id
	})
	var sourceRec, targetRec models.SObject
	for _, lock := range recordLocks {
		record, lockErr := e.persistence.GetLock(ctx, lock.object, lock.id)
		if lockErr != nil {
			return "", fmt.Errorf("transfer connector remote identity: GetLock %s/%s: %w", lock.object, lock.id, lockErr)
		}
		if record == nil {
			return "", fmt.Errorf("transfer connector remote identity: record %s/%s not found", lock.object, lock.id)
		}
		if lock.source {
			sourceRec = record
		} else {
			targetRec = record
		}
	}
	if sourceRec == nil || targetRec == nil {
		return "", fmt.Errorf("transfer connector remote identity: stable record lock assembly is incomplete")
	}
	if err := hintStore.LockInboundRemoteIdentityTransferBoundaryInAmbientTx(
		ctx,
		sourceObject,
		sourceID,
		targetObject,
		targetID,
		remoteIDField,
	); err != nil {
		return "", fmt.Errorf("transfer connector remote identity: lock transfer boundary: %w", err)
	}
	lockedHints, err := hintStore.FindInboundRemoteIdentityTransferHintsInAmbientTx(
		ctx,
		sourceObject,
		sourceID,
		targetObject,
		targetID,
		remoteIDField,
	)
	if err != nil {
		return "", fmt.Errorf("transfer connector remote identity: revalidate transfer hints under boundary lock: %w", err)
	}
	if !connectorRemoteIdentityTransferHintsEqual(initialHints, lockedHints) {
		return "", apperrors.NewLockContentionError(fmt.Errorf(
			"%w: connector remote identity transfer hints changed while acquiring record and transfer boundary locks",
			ErrConnectorPushDurableConflict,
		))
	}
	targetRemoteID := actionRecordString(targetRec, remoteIDField)
	transfer.RemoteId = actionRecordString(sourceRec, remoteIDField)
	transfer.TargetRemoteID = targetRemoteID
	transferredRemoteID, err := e.connectorPushDurable.TransferInboundRemoteIdentity(ctx, transfer)
	if err != nil {
		return "", fmt.Errorf("transfer connector remote identity: %w", err)
	}
	if transferredRemoteID == "" {
		if targetRemoteID != "" {
			// The target mirror came from conversion field mapping, not durable
			// connector authority. Remove it atomically without launching a second
			// record-automation chain, then let the business conversion continue.
			cleanupCtx := constants.WithSecurityContext(
				ctx,
				constants.SystemContext,
				"clear untrusted target remote-id mirror during protected identity transfer",
			)
			cleanupCtx = context.WithValue(
				cleanupCtx,
				constants.ContextKeySuppressRecordTriggeredAutomation,
				targetObject+"/"+targetID,
			)
			if err := e.persistence.Update(
				cleanupCtx,
				targetObject,
				targetID,
				models.SObject{remoteIDField: nil},
				user,
			); err != nil {
				return "", fmt.Errorf(
					"transfer connector remote identity: clear untrusted target remote-id mirror %s/%s %s: %w",
					targetObject,
					targetID,
					remoteIDField,
					flowRecordMutationValidation(err),
				)
			}
			if e.logger != nil {
				e.logger.Warn(fmt.Sprintf(
					"transfer connector remote identity: cleared untrusted target mirror and continued without durable authority (target=%s/%s field=%s)",
					targetObject,
					targetID,
					remoteIDField,
				))
			}
		}
		setTransferResult(false)
		return next, nil
	}
	if targetRemoteID == "" {
		return "", fmt.Errorf("transfer connector remote identity: target %s/%s %s is blank after durable remote id transfer", targetObject, targetID, remoteIDField)
	}
	if targetRemoteID != transferredRemoteID {
		return "", fmt.Errorf("transfer connector remote identity: target %s/%s %s does not match durable remote id", targetObject, targetID, remoteIDField)
	}
	setTransferResult(true)
	return next, nil
}

func connectorRemoteIdentityTransferExecutionAuthorized(ctx context.Context, identity FlowExecutionIdentity) bool {
	if !flowIdentityIsPackOwnedStandard(identity) || strings.TrimSpace(identity.OwnerPack) == "" {
		return false
	}
	if constants.HasStandardActionExecutionContext(ctx) &&
		constants.HasStandardActionProtectedFlowExecution(ctx) {
		return true
	}
	// A name/namespace-bound inbound root marker is intentionally insufficient
	// for this durable authority mutation: it can resolve a different active
	// FlowVersion after enqueue. Companion execution is frozen and rechecked by
	// immutable VersionID before the worker invokes the Flow.
	return constants.ConnectorInboundCompanionExecutionAuthorizesVersion(ctx, identity.VersionID) ||
		AfterCommitScheduledFlowExecutionAuthorizesVersion(ctx, identity.VersionID)
}

func actionRecordString(rec models.SObject, field string) string {
	if rec == nil {
		return ""
	}
	v, ok := rec[field]
	if !ok || v == nil {
		return ""
	}
	return strings.TrimSpace(fmt.Sprintf("%v", v))
}

// executeSetRetriable is a built-in invocable action that marks the current
// flow execution as retriable. When a flow calls this action, the inbound
// worker will call MarkFailed instead of MarkProcessed on successful
// completion, leaving the event in the queue for retry after prerequisites
// land.
//
// This is the leaf authoring path for IsRetriable=true: flows can call this
// action when they detect a transient condition (e.g. missing prerequisite
// record) that should cause the event to be retried.
//
// Input parameters:
//   - reason (optional): a human-readable reason for the retriable condition
func (e *FlowExecutor) executeSetRetriable(ctx context.Context, interview *models.FlowInterview, action *models.FlowActionCall) (string, error) {
	if rt := flowRuntimeContextFromCtx(ctx); rt != nil {
		rt.SetRetriable()
	}
	if action.Connector != nil {
		return action.Connector.TargetReference, nil
	}
	return "", nil
}