// Package runtimeflowexecution owns the app-blind bounded-atomic Flow
// selection used by both Runtime API and Runtime Worker compositions. It
// contains no serving, worker, Integration, or L1 dependency.
package runtimeflowexecution

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/shared/pkg/models"
)

const runtimeFlowMaxElementExecutions uint64 = 1000

var errFlowNotBoundedAtomic = errors.New("runtime flow is not eligible for bounded atomic execution")

type eligibilityProfile uint8

const (
	eligibilityProfileSynchronousAPI eligibilityProfile = iota + 1
	eligibilityProfilePlatformCommand
	eligibilityProfilePlatformCommandDependency
	eligibilityProfileCatalogAction
)

func isPlatformCommandProfile(profile eligibilityProfile) bool {
	return profile == eligibilityProfilePlatformCommand ||
		profile == eligibilityProfilePlatformCommandDependency
}

func isSupportedEligibilityProfile(profile eligibilityProfile) bool {
	return profile == eligibilityProfileSynchronousAPI ||
		isPlatformCommandProfile(profile) ||
		profile == eligibilityProfileCatalogAction
}

type runtimeFlowMetadataView interface {
	ActiveFlowVersion(string) (*models.FlowVersion, bool)
}

// AtomicValidator proves that the exact execution engine which will run a
// selected Flow has every action handler required by that Flow and accepts the
// complete atomic shape. The Runtime API uses the published platform validator;
// role-scoped Workers pass their configured executor's validator.
type AtomicValidator func(
	context.Context,
	string,
	*models.FlowMetadata,
) error

type boundedAtomicFlowSelection struct {
	root                 *models.FlowVersion
	closureDigest        string
	maxElementExecutions uint64
	versions             []boundedAtomicFlowVersion
}

type boundedAtomicFlowVersion struct {
	definitionRef string
	version       *models.FlowVersion
}

type boundedAtomicFlowMember struct {
	DefinitionRef string          `json:"definitionRef"`
	DefinitionID  string          `json:"definitionId"`
	VersionID     string          `json:"versionId"`
	VersionNumber int             `json:"versionNumber"`
	Metadata      json.RawMessage `json:"metadata"`
}

type flowRecordReferenceProducer struct {
	successTarget string
	faultTarget   string
}

func selectBoundedAtomicFlow(
	ctx context.Context,
	view runtimeFlowMetadataView,
	definitionRef string,
	profile eligibilityProfile,
	validateAtomic AtomicValidator,
) (boundedAtomicFlowSelection, error) {
	if ctx == nil || ctx.Err() != nil || view == nil ||
		validateAtomic == nil ||
		strings.TrimSpace(definitionRef) == "" ||
		!isSupportedEligibilityProfile(profile) {
		return boundedAtomicFlowSelection{}, errFlowNotBoundedAtomic
	}
	root, ok := view.ActiveFlowVersion(definitionRef)
	if !ok || root == nil {
		return boundedAtomicFlowSelection{}, errFlowNotBoundedAtomic
	}

	visiting := make(map[string]bool)
	recorded := make(map[string]bool)
	members := make([]boundedAtomicFlowMember, 0, 4)
	versions := make([]boundedAtomicFlowVersion, 0, 4)
	var total uint64
	var walk func(string, *models.FlowVersion, int) error
	walk = func(ref string, version *models.FlowVersion, depth int) error {
		key := strings.ToLower(strings.TrimSpace(ref))
		if key == "" || version == nil || version.Id == "" || version.DefinitionId == "" ||
			version.IsDeleted || !strings.EqualFold(version.Status, models.FlowStatusActive) ||
			(!strings.EqualFold(version.ProcessType, "AutoLaunchedFlow") &&
				!strings.EqualFold(version.ProcessType, "InvocableProcess")) ||
			depth > 10 {
			return errFlowNotBoundedAtomic
		}
		if depth > 0 && isPlatformCommandProfile(profile) &&
			!platformCommandDirectNativeSubflowAuthorized(root, version, depth) {
			return fmt.Errorf(
				"%w: Platform-command profile has no direct child Flow authorization",
				errFlowNotBoundedAtomic,
			)
		}
		if visiting[key] {
			return fmt.Errorf("%w: subflow cycle", errFlowNotBoundedAtomic)
		}
		visiting[key] = true
		defer func() {
			visiting[key] = false
		}()
		metadata := version.Metadata
		if err := validateAtomic(
			ctx,
			version.ProcessType,
			&metadata,
		); err != nil {
			return fmt.Errorf("%w: engine atomic validation", errFlowNotBoundedAtomic)
		}
		count, children, err := validateBoundedAtomicFlowMetadata(
			&metadata,
			profile,
		)
		if err != nil {
			return fmt.Errorf("%s: %w", ref, err)
		}
		if count == 0 || count > runtimeFlowMaxElementExecutions {
			return fmt.Errorf("%w: element bound", errFlowNotBoundedAtomic)
		}
		if profile != eligibilityProfileCatalogAction {
			if total > runtimeFlowMaxElementExecutions-count {
				if !isPlatformCommandProfile(profile) {
					return fmt.Errorf("%w: element bound", errFlowNotBoundedAtomic)
				}
				// Platform-command direct children execute inside the parent's
				// caller-owned transaction and consume the same mutable runtime
				// budget. A loop makes an individual member's static upper bound
				// equal to the hard transaction cap, but does not mint a second
				// budget for its child. Saturate the closure bound and let the
				// shared runtime governor fail and roll back an oversized input.
				total = runtimeFlowMaxElementExecutions
			} else {
				total += count
			}
		}
		raw := version.MetadataRaw
		if len(raw) == 0 {
			raw, err = json.Marshal(metadata)
			if err != nil {
				return errFlowNotBoundedAtomic
			}
		}
		if !recorded[key] {
			members = append(members, boundedAtomicFlowMember{
				DefinitionRef: ref,
				DefinitionID:  version.DefinitionId,
				VersionID:     version.Id,
				VersionNumber: version.VersionNumber,
				Metadata:      append(json.RawMessage(nil), raw...),
			})
			versions = append(versions, boundedAtomicFlowVersion{
				definitionRef: ref,
				version:       version,
			})
			recorded[key] = true
		}
		for _, childRef := range children {
			child, found := view.ActiveFlowVersion(childRef)
			if !found {
				return fmt.Errorf("%w: missing exact subflow", errFlowNotBoundedAtomic)
			}
			if err := walk(childRef, child, depth+1); err != nil {
				return err
			}
		}
		return nil
	}
	if err := walk(definitionRef, root, 0); err != nil {
		return boundedAtomicFlowSelection{}, err
	}
	executionBudget := total
	if profile == eligibilityProfileCatalogAction {
		// Catalog actions can publish record mutations that synchronously dispatch
		// independently selected record-trigger roots from the same immutable
		// request view. Those roots share the transaction execution budget but are
		// intentionally not members of the action's prepared child-Flow closure.
		// The closure's static element count is therefore not a valid transaction
		// budget. Keep the engine's existing hard transaction cap for this profile;
		// the other profiles retain their tighter static closure budget.
		executionBudget = runtimeFlowMaxElementExecutions
	}
	sort.Slice(members, func(i, j int) bool {
		return strings.ToLower(members[i].DefinitionRef) < strings.ToLower(members[j].DefinitionRef)
	})
	payload, err := json.Marshal(members)
	if err != nil {
		return boundedAtomicFlowSelection{}, errFlowNotBoundedAtomic
	}
	sum := sha256.Sum256(payload)
	return boundedAtomicFlowSelection{
		root:                 root,
		closureDigest:        "sha256:" + hex.EncodeToString(sum[:]),
		maxElementExecutions: executionBudget,
		versions:             versions,
	}, nil
}

func validateBoundedAtomicFlowMetadata(
	metadata *models.FlowMetadata,
	profile eligibilityProfile,
) (uint64, []string, error) {
	if metadata == nil || !metadata.RunAtomically ||
		len(metadata.Waits) != 0 || len(metadata.Screens) != 0 ||
		(len(metadata.AtomicParentPrelocks) != 0 &&
			profile != eligibilityProfileCatalogAction) {
		return 0, nil, fmt.Errorf(
			"%w: atomic declaration, Wait, Screen, or prelock shape",
			errFlowNotBoundedAtomic,
		)
	}
	// Catalog actions execute through the resource-bounded prepared-closure
	// entrypoint. That engine seam caps direct record fan-out and parent-prelock
	// discovery before business effects, so this profile does not need the
	// single-lookup provenance restriction used by the other profiles.
	runtimeBoundsRecordReferences := profile == eligibilityProfileCatalogAction
	if profile == eligibilityProfileSynchronousAPI && len(metadata.Loops) != 0 {
		return 0, nil, fmt.Errorf(
			"%w: synchronous profile does not accept loops",
			errFlowNotBoundedAtomic,
		)
	}
	if metadata.Start != nil &&
		(metadata.Start.Object != "" || metadata.Start.TriggerType != "" ||
			metadata.Start.RecordTriggerType != "" || metadata.Start.Schedule != nil ||
			metadata.Start.ConnectorSource != nil || len(metadata.Start.ScheduledPaths) != 0) {
		return 0, nil, fmt.Errorf(
			"%w: trigger or schedule root",
			errFlowNotBoundedAtomic,
		)
	}
	graph, start, err := boundedAtomicElementGraph(metadata)
	if err != nil {
		return 0, nil, err
	}
	inputReferences := make(map[string]struct{}, len(metadata.Variables))
	collectionReferences := make(map[string]struct{}, len(metadata.Variables)+len(metadata.RecordLookups)*2)
	for _, variable := range metadata.Variables {
		if profile == eligibilityProfileSynchronousAPI {
			if variable.IsCollection {
				return 0, nil, fmt.Errorf(
					"%w: synchronous profile does not accept collection variables",
					errFlowNotBoundedAtomic,
				)
			}
			continue
		}
		name := foldedFlowReference(variable.Name)
		if name == "" {
			return 0, nil, fmt.Errorf(
				"%w: Flow variable has no name",
				errFlowNotBoundedAtomic,
			)
		}
		if variable.IsInput {
			inputReferences[name] = struct{}{}
		}
		if variable.IsCollection {
			collectionReferences[name] = struct{}{}
		}
	}
	singleRecordReferences := make(map[string]struct{}, len(metadata.RecordLookups)*2)
	singleRecordReferenceProducers := make(
		map[string]map[string]flowRecordReferenceProducer,
		len(metadata.RecordLookups)*2,
	)
	for _, lookup := range metadata.RecordLookups {
		if profile == eligibilityProfileSynchronousAPI &&
			!lookup.GetFirstRecordOnly {
			return 0, nil, fmt.Errorf(
				"%w: synchronous profile does not accept multi-row lookups",
				errFlowNotBoundedAtomic,
			)
		}
		targets := []string{lookup.Name, lookup.OutputReference}
		if lookup.GetFirstRecordOnly {
			producer := flowRecordReferenceProducer{}
			if lookup.Connector != nil {
				producer.successTarget = strings.TrimSpace(lookup.Connector.TargetReference)
			}
			if lookup.FaultConnector != nil {
				producer.faultTarget = strings.TrimSpace(lookup.FaultConnector.TargetReference)
			}
			for _, target := range targets {
				reference := foldedFlowReferenceForProfile(target, profile)
				if reference != "" {
					singleRecordReferences[reference] = struct{}{}
					if singleRecordReferenceProducers[reference] == nil {
						singleRecordReferenceProducers[reference] = make(
							map[string]flowRecordReferenceProducer,
						)
					}
					singleRecordReferenceProducers[reference][strings.TrimSpace(lookup.Name)] = producer
				}
			}
		} else {
			for _, target := range targets {
				if reference := foldedFlowReference(target); reference != "" {
					collectionReferences[reference] = struct{}{}
				}
			}
		}
	}
	competingRecordReferenceWriters := flowReferenceWritersOutsideRecordLookups(metadata)
	for _, loop := range metadata.Loops {
		reference := foldedFlowReference(loop.CollectionReference)
		if reference == "" {
			return 0, nil, fmt.Errorf(
				"%w: loop has no collection reference",
				errFlowNotBoundedAtomic,
			)
		}
		if _, found := collectionReferences[reference]; found {
			continue
		}
		root := reference
		if index := strings.IndexByte(root, '.'); index >= 0 {
			root = root[:index]
		}
		if _, found := inputReferences[root]; !found {
			return 0, nil, fmt.Errorf(
				"%w: loop collection is not lookup output, collection variable, or input",
				errFlowNotBoundedAtomic,
			)
		}
	}
	for _, update := range metadata.RecordUpdates {
		if profile == eligibilityProfileSynchronousAPI {
			if update.InputReference == "" || len(update.Filters) != 0 {
				return 0, nil, fmt.Errorf(
					"%w: synchronous update is not a single-record reference",
					errFlowNotBoundedAtomic,
				)
			}
			if _, provenSingle := singleRecordReferences[foldedFlowReferenceForProfile(
				update.InputReference,
				profile,
			)]; !provenSingle {
				return 0, nil, fmt.Errorf(
					"%w: synchronous update reference is not a single lookup",
					errFlowNotBoundedAtomic,
				)
			}
			continue
		}
		if update.InputReference != "" {
			if len(update.Filters) != 0 {
				return 0, nil, fmt.Errorf(
					"%w: record-reference update also declares filters",
					errFlowNotBoundedAtomic,
				)
			}
			reference := foldedFlowReference(update.InputReference)
			if reference == "" {
				return 0, nil, fmt.Errorf(
					"%w: update has an empty record reference",
					errFlowNotBoundedAtomic,
				)
			}
			_, provenSingle := singleRecordReferences[reference]
			_, alsoCollection := collectionReferences[reference]
			_, hasCompetingWriter := competingRecordReferenceWriters[reference]
			if !runtimeBoundsRecordReferences &&
				(!provenSingle || alsoCollection || hasCompetingWriter ||
					!flowReferenceHasAuthoritativeProvenance(
						graph,
						start,
						singleRecordReferenceProducers[reference],
						strings.TrimSpace(update.Name),
					)) {
				return 0, nil, fmt.Errorf(
					"%w: update reference does not have authoritative single-record cardinality or provenance",
					errFlowNotBoundedAtomic,
				)
			}
			continue
		}
		if len(update.Filters) == 0 ||
			automation.MaxAtomicFilteredRecordOperationRows <= 0 ||
			models.ValidateFlowFilterLogic(
				update.FilterLogic,
				len(update.Filters),
			) != nil {
			return 0, nil, fmt.Errorf(
				"%w: update has no bounded filter expression",
				errFlowNotBoundedAtomic,
			)
		}
	}
	for _, deleteElement := range metadata.RecordDeletes {
		if profile == eligibilityProfileSynchronousAPI {
			if deleteElement.InputReference == "" ||
				len(deleteElement.Filters) != 0 {
				return 0, nil, fmt.Errorf(
					"%w: synchronous delete is not a single-record reference",
					errFlowNotBoundedAtomic,
				)
			}
			if _, provenSingle := singleRecordReferences[foldedFlowReferenceForProfile(
				deleteElement.InputReference,
				profile,
			)]; !provenSingle {
				return 0, nil, fmt.Errorf(
					"%w: synchronous delete reference is not a single lookup",
					errFlowNotBoundedAtomic,
				)
			}
			continue
		}
		if deleteElement.InputReference != "" {
			if len(deleteElement.Filters) != 0 {
				return 0, nil, fmt.Errorf(
					"%w: record-reference delete also declares filters",
					errFlowNotBoundedAtomic,
				)
			}
			reference := foldedFlowReference(deleteElement.InputReference)
			if reference == "" {
				return 0, nil, fmt.Errorf(
					"%w: delete has an empty record reference",
					errFlowNotBoundedAtomic,
				)
			}
			_, provenSingle := singleRecordReferences[reference]
			_, alsoCollection := collectionReferences[reference]
			_, hasCompetingWriter := competingRecordReferenceWriters[reference]
			if !runtimeBoundsRecordReferences &&
				(!provenSingle || alsoCollection || hasCompetingWriter ||
					!flowReferenceHasAuthoritativeProvenance(
						graph,
						start,
						singleRecordReferenceProducers[reference],
						strings.TrimSpace(deleteElement.Name),
					)) {
				return 0, nil, fmt.Errorf(
					"%w: delete reference does not have authoritative single-record cardinality or provenance",
					errFlowNotBoundedAtomic,
				)
			}
			continue
		}
		if len(deleteElement.Filters) == 0 ||
			automation.MaxAtomicFilteredRecordOperationRows <= 0 ||
			models.ValidateFlowFilterLogic(
				deleteElement.FilterLogic,
				len(deleteElement.Filters),
			) != nil {
			return 0, nil, fmt.Errorf(
				"%w: delete has no bounded filter expression",
				errFlowNotBoundedAtomic,
			)
		}
	}

	children := make([]string, 0, len(metadata.Subflows)+len(metadata.ActionCalls))
	for _, action := range metadata.ActionCalls {
		if (profile == eligibilityProfilePlatformCommand ||
			profile == eligibilityProfileCatalogAction) &&
			strings.EqualFold(
				strings.TrimSpace(action.ActionType),
				flowactions.ActionTypeSetIdentityConvergenceDependency,
			) {
			return 0, nil, fmt.Errorf(
				"%w: Platform-command profile has no deferred dependency lifecycle",
				errFlowNotBoundedAtomic,
			)
		}
		if profile == eligibilityProfileSynchronousAPI &&
			(!strings.EqualFold(strings.TrimSpace(action.ActionType), "flow") ||
				strings.TrimSpace(action.ActionName) == "") {
			return 0, nil, fmt.Errorf(
				"%w: synchronous profile does not accept platform actions",
				errFlowNotBoundedAtomic,
			)
		}
		if strings.EqualFold(strings.TrimSpace(action.ActionType), "flow") {
			if isPlatformCommandProfile(profile) {
				return 0, nil, fmt.Errorf(
					"%w: Platform-command profile has no direct child Flow authorization",
					errFlowNotBoundedAtomic,
				)
			}
			if strings.TrimSpace(action.ActionName) == "" {
				return 0, nil, fmt.Errorf(
					"%w: Flow action has no child definition",
					errFlowNotBoundedAtomic,
				)
			}
			children = append(children, action.ActionName)
		}
	}
	for _, subflow := range metadata.Subflows {
		if strings.TrimSpace(subflow.FlowName) == "" {
			return 0, nil, fmt.Errorf(
				"%w: subflow has no child definition",
				errFlowNotBoundedAtomic,
			)
		}
		children = append(children, subflow.FlowName)
	}

	loopElements := make(map[string]struct{}, len(metadata.Loops))
	for _, loop := range metadata.Loops {
		loopElements[strings.TrimSpace(loop.Name)] = struct{}{}
	}
	reachable, err := validateBoundedAtomicElementCycles(
		graph,
		start,
		loopElements,
	)
	if err != nil {
		return 0, nil, err
	}
	if reachable == 0 {
		return 0, nil, fmt.Errorf(
			"%w: no reachable Flow elements",
			errFlowNotBoundedAtomic,
		)
	}
	if len(metadata.Loops) != 0 {
		reachable = runtimeFlowMaxElementExecutions
	}
	sort.Strings(children)
	return reachable, children, nil
}

func platformCommandDirectNativeSubflowAuthorized(
	root *models.FlowVersion,
	child *models.FlowVersion,
	depth int,
) bool {
	if depth != 1 || root == nil || child == nil ||
		!root.DefinitionRuntimeLoaded || !child.DefinitionRuntimeLoaded ||
		root.DefinitionIsCustomized || child.DefinitionIsCustomized ||
		!strings.EqualFold(strings.TrimSpace(root.Source), "standard") ||
		!strings.EqualFold(strings.TrimSpace(child.Source), "standard") {
		return false
	}
	rootOwner := strings.TrimSpace(root.DefinitionOwnerPack)
	childOwner := strings.TrimSpace(child.DefinitionOwnerPack)
	return rootOwner != "" && childOwner != "" &&
		strings.EqualFold(rootOwner, childOwner)
}

func flowReferenceHasAuthoritativeProvenance(
	graph map[string][]string,
	start string,
	producers map[string]flowRecordReferenceProducer,
	target string,
) bool {
	if start == "" || target == "" || len(producers) == 0 {
		return false
	}
	type provenanceState struct {
		element string
		proven  bool
	}
	pending := []provenanceState{{element: start}}
	visited := make(map[provenanceState]struct{}, len(graph)*2)
	reachedTarget := false
	for len(pending) != 0 {
		last := len(pending) - 1
		state := pending[last]
		pending = pending[:last]
		if _, seen := visited[state]; seen {
			continue
		}
		visited[state] = struct{}{}
		if state.element == target {
			reachedTarget = true
			if !state.proven {
				return false
			}
		}
		for _, next := range graph[state.element] {
			nextState := provenanceState{element: next, proven: state.proven}
			if producer, found := producers[state.element]; found {
				switch {
				case next == producer.faultTarget:
					nextState.proven = false
				case producer.successTarget != "" && next == producer.successTarget:
					nextState.proven = true
				default:
					nextState.proven = false
				}
			}
			pending = append(pending, nextState)
		}
	}
	return reachedTarget
}

// flowReferenceWritersOutsideRecordLookups returns context keys whose runtime
// value is not owned exclusively by RecordLookup. Platform-command direct
// Update/Delete may iterate a collection-valued InputReference, so a GetFirst
// declaration proves single-record cardinality only when no other element can
// replace that same key. Field writes such as "record.Name" are intentionally
// excluded: they mutate a field on the already-proven record rather than
// replacing the record reference itself.
func flowReferenceWritersOutsideRecordLookups(
	metadata *models.FlowMetadata,
) map[string]struct{} {
	writers := make(map[string]struct{})
	if metadata == nil {
		return writers
	}
	addExact := func(reference string) {
		if reference := foldedFlowReference(reference); reference != "" {
			writers[reference] = struct{}{}
		}
	}
	addRootReplacement := func(reference string) {
		reference = strings.TrimPrefix(strings.TrimSpace(reference), "$")
		if reference == "" || strings.Contains(reference, ".") {
			return
		}
		writers[strings.ToLower(reference)] = struct{}{}
	}

	for _, variable := range metadata.Variables {
		addExact(variable.Name)
	}
	for _, lookup := range metadata.RecordLookups {
		for _, output := range lookup.OutputAssignments {
			addExact(output.AssignToReference)
		}
	}
	for _, assignment := range metadata.Assignments {
		for _, item := range assignment.Items {
			addRootReplacement(item.AssignToReference)
		}
	}
	for _, create := range metadata.RecordCreates {
		if create.StoreOutputAutomatically {
			addExact(create.Name)
		}
		addExact(create.AssignRecordIdToReference)
	}
	for _, loop := range metadata.Loops {
		if loop.AssignNextValueToReference != "" {
			addExact(loop.AssignNextValueToReference)
		} else {
			addExact(loop.Name)
		}
	}
	for _, subflow := range metadata.Subflows {
		for _, output := range subflow.OutputAssignments {
			addExact(output.AssignToReference)
		}
	}
	for _, action := range metadata.ActionCalls {
		for _, output := range action.OutputParameters {
			addRootReplacement(flowOutputTargetReference(output.Value))
		}
	}
	for _, formula := range metadata.Formulas {
		addExact(formula.Name)
	}
	for _, decision := range metadata.Decisions {
		for _, rule := range decision.Rules {
			addExact(rule.Name)
		}
	}
	return writers
}

func flowOutputTargetReference(value interface{}) string {
	switch typed := value.(type) {
	case string:
		return typed
	case map[string]interface{}:
		reference, _ := typed["elementReference"].(string)
		return reference
	case map[string]string:
		return typed["elementReference"]
	default:
		return ""
	}
}

func validateBoundedAtomicElementCycles(
	graph map[string][]string,
	start string,
	loopElements map[string]struct{},
) (uint64, error) {
	reachable := make(map[string]bool, len(graph))
	pending := []string{start}
	for len(pending) != 0 {
		last := len(pending) - 1
		name := pending[last]
		pending = pending[:last]
		if name == "" || reachable[name] {
			continue
		}
		next, ok := graph[name]
		if !ok {
			return 0, fmt.Errorf(
				"%w: missing connector target %q",
				errFlowNotBoundedAtomic,
				name,
			)
		}
		reachable[name] = true
		pending = append(pending, next...)
	}

	const (
		elementUnvisited uint8 = iota
		elementVisiting
		elementVisited
	)
	state := make(map[string]uint8, len(reachable))
	var visit func(string) error
	visit = func(name string) error {
		if _, declaredLoop := loopElements[name]; declaredLoop {
			return nil
		}
		switch state[name] {
		case elementVisiting:
			return fmt.Errorf(
				"%w: element cycle does not pass through a declared Loop at %q",
				errFlowNotBoundedAtomic,
				name,
			)
		case elementVisited:
			return nil
		}
		state[name] = elementVisiting
		for _, target := range graph[name] {
			if !reachable[target] {
				continue
			}
			if err := visit(target); err != nil {
				return err
			}
		}
		state[name] = elementVisited
		return nil
	}
	for name := range reachable {
		if err := visit(name); err != nil {
			return 0, err
		}
	}
	return uint64(len(reachable)), nil
}

func foldedFlowReference(value string) string {
	return strings.ToLower(strings.TrimPrefix(strings.TrimSpace(value), "$"))
}

func foldedFlowReferenceForProfile(
	value string,
	profile eligibilityProfile,
) string {
	if profile == eligibilityProfileSynchronousAPI {
		return strings.ToLower(strings.TrimSpace(value))
	}
	return foldedFlowReference(value)
}

func boundedAtomicElementGraph(metadata *models.FlowMetadata) (map[string][]string, string, error) {
	graph := make(map[string][]string)
	add := func(name string, connectors ...*models.FlowConnector) error {
		name = strings.TrimSpace(name)
		if name == "" {
			return errFlowNotBoundedAtomic
		}
		if _, exists := graph[name]; exists {
			return fmt.Errorf("%w: duplicate element name", errFlowNotBoundedAtomic)
		}
		next := make([]string, 0, len(connectors))
		for _, connector := range connectors {
			if connector != nil && strings.TrimSpace(connector.TargetReference) != "" {
				next = append(next, connector.TargetReference)
			}
		}
		graph[name] = next
		return nil
	}
	for i := range metadata.Decisions {
		value := &metadata.Decisions[i]
		connectors := []*models.FlowConnector{value.DefaultConnector}
		for j := range value.Rules {
			connectors = append(connectors, value.Rules[j].Connector)
		}
		if err := add(value.Name, connectors...); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.Assignments {
		if err := add(metadata.Assignments[i].Name, metadata.Assignments[i].Connector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.RecordCreates {
		value := &metadata.RecordCreates[i]
		if err := add(value.Name, value.Connector, value.FaultConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.RecordUpdates {
		value := &metadata.RecordUpdates[i]
		if err := add(value.Name, value.Connector, value.FaultConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.RecordDeletes {
		value := &metadata.RecordDeletes[i]
		if err := add(value.Name, value.Connector, value.FaultConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.RecordLookups {
		value := &metadata.RecordLookups[i]
		if err := add(value.Name, value.Connector, value.FaultConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.ActionCalls {
		value := &metadata.ActionCalls[i]
		if err := add(value.Name, value.Connector, value.FaultConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.Subflows {
		value := &metadata.Subflows[i]
		if err := add(value.Name, value.Connector, value.FaultConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.Loops {
		value := &metadata.Loops[i]
		if err := add(value.Name, value.NextValueConnector, value.NoMoreValuesConnector); err != nil {
			return nil, "", err
		}
	}
	for i := range metadata.CustomErrors {
		value := &metadata.CustomErrors[i]
		if err := add(value.Name, value.Connector); err != nil {
			return nil, "", err
		}
	}
	start := strings.TrimSpace(metadata.StartElementReference)
	if metadata.Start != nil && metadata.Start.Connector != nil {
		start = strings.TrimSpace(metadata.Start.Connector.TargetReference)
	}
	if start == "" {
		return nil, "", errFlowNotBoundedAtomic
	}
	return graph, start, nil
}

// MetadataView is the immutable Flow lookup surface required for selection.
type MetadataView interface {
	ActiveFlowVersion(string) (*models.FlowVersion, bool)
}

// SelectedVersion is one exact member of a selected Flow closure.
type SelectedVersion struct {
	DefinitionRef string
	Version       *models.FlowVersion
}

// Selection is the complete immutable bounded-atomic Flow closure.
type Selection struct {
	Root                 *models.FlowVersion
	ClosureDigest        string
	MaxElementExecutions uint64
	Versions             []SelectedVersion
}

// MaxElementExecutions is the hard per-transaction execution budget. A Flow
// that reaches it fails and rolls back the complete caller-owned transaction.
const MaxElementExecutions = runtimeFlowMaxElementExecutions

// ErrNotBoundedAtomic classifies an ineligible or incompletely proven Flow.
var ErrNotBoundedAtomic = errFlowNotBoundedAtomic

func selectForProfile(
	ctx context.Context,
	view MetadataView,
	definitionRef string,
	profile eligibilityProfile,
	validateAtomic AtomicValidator,
) (Selection, error) {
	selected, err := selectBoundedAtomicFlow(
		ctx,
		view,
		definitionRef,
		profile,
		validateAtomic,
	)
	if err != nil {
		return Selection{}, err
	}
	versions := make([]SelectedVersion, 0, len(selected.versions))
	for _, version := range selected.versions {
		versions = append(versions, SelectedVersion{
			DefinitionRef: version.definitionRef,
			Version:       version.version,
		})
	}
	return Selection{
		Root: selected.root, ClosureDigest: selected.closureDigest,
		MaxElementExecutions: selected.maxElementExecutions,
		Versions:             versions,
	}, nil
}

// SelectSynchronousAPI validates and binds the exact bounded-atomic Flow
// closure accepted by the published synchronous ExecuteFlow API. This closed
// profile intentionally preserves the M1 accepted set: no loops, collections,
// multi-row lookups, filtered writes, or non-Flow action calls.
func SelectSynchronousAPI(
	ctx context.Context,
	view MetadataView,
	definitionRef string,
) (Selection, error) {
	return SelectSynchronousAPIWithValidator(
		ctx,
		view,
		definitionRef,
		automation.ValidateAtomicFlowForExecution,
	)
}

// SelectSynchronousAPIWithValidator preserves the synchronous accepted set
// while proving action handlers against the exact executor that will run the
// request. It exists so serving composition cannot validate against a generic
// registry and execute against a different role-specific registry.
func SelectSynchronousAPIWithValidator(
	ctx context.Context,
	view MetadataView,
	definitionRef string,
	validateAtomic AtomicValidator,
) (Selection, error) {
	return selectForProfile(
		ctx,
		view,
		definitionRef,
		eligibilityProfileSynchronousAPI,
		validateAtomic,
	)
}

// SelectCatalogAction validates the complete exact closure of one
// Catalog-declared bounded action. This explicit profile combines the bounded
// collection/loop rules used by Platform-command execution with the exact
// child-Flow closure used by the synchronous API. Non-Flow action calls remain
// subject to the production engine's atomic validator. The caller must
// separately prove the immutable Catalog action binding before using this
// selection; ordinary ExecuteFlow requests never select this profile.
func SelectCatalogAction(
	ctx context.Context,
	view MetadataView,
	definitionRef string,
	validateAtomic AtomicValidator,
) (Selection, error) {
	return selectForProfile(
		ctx,
		view,
		definitionRef,
		eligibilityProfileCatalogAction,
		validateAtomic,
	)
}

// SelectPlatformCommand validates and binds the exact bounded-atomic Flow
// closure accepted by the durable Platform-command Worker. Its additional
// shapes remain subject to the same atomic engine validation and hard
// transaction execution budget. Flows that can hand off a deferred dependency
// are excluded until a separate durable dependency lifecycle owns them.
func SelectPlatformCommand(
	ctx context.Context,
	view MetadataView,
	definitionRef string,
	validateAtomic AtomicValidator,
) (Selection, error) {
	return selectForProfile(
		ctx,
		view,
		definitionRef,
		eligibilityProfilePlatformCommand,
		validateAtomic,
	)
}

// SelectPlatformCommandWithDependency preserves every bounded Platform-command
// proof while allowing the one terminal dependency action whose V2 caller
// scope can be durably bound after the owned Flow transaction rolls back.
// V1 and synchronous API callers deliberately cannot select this profile.
func SelectPlatformCommandWithDependency(
	ctx context.Context,
	view MetadataView,
	definitionRef string,
	validateAtomic AtomicValidator,
) (Selection, error) {
	return selectForProfile(
		ctx,
		view,
		definitionRef,
		eligibilityProfilePlatformCommandDependency,
		validateAtomic,
	)
}