package runtimeflowexecution

import (
	"context"
	"strings"
	"testing"

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

type profileTestView map[string]*models.FlowVersion

func (v profileTestView) ActiveFlowVersion(
	name string,
) (*models.FlowVersion, bool) {
	version, ok := v[strings.ToLower(name)]
	return version, ok
}

func TestClosedEligibilityProfilesKeepWorkerExtensionsOutOfSynchronousAPI(
	t *testing.T,
) {
	tests := []struct {
		name     string
		metadata models.FlowMetadata
		wantMax  uint64
	}{
		{
			name: "bounded lookup loop and exact-ID write",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "Lookup",
				RecordLookups: []models.FlowRecordLookup{{
					Name: "Lookup", OutputReference: "$records", Object: "Widget",
					Connector: &models.FlowConnector{
						TargetReference: "Loop",
					},
				}},
				Loops: []models.FlowLoop{{
					Name:                       "Loop",
					CollectionReference:        "$records",
					AssignNextValueToReference: "$record",
					NextValueConnector: &models.FlowConnector{
						TargetReference: "Update",
					},
					NoMoreValuesConnector: &models.FlowConnector{
						TargetReference: "Done",
					},
				}},
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", FilterLogic: "and",
					Filters: []models.FlowRecordFilter{{
						Field: "Id", Operator: "EqualTo",
						Value: map[string]interface{}{
							"elementReference": "$record.Id",
						},
					}},
					Connector: &models.FlowConnector{
						TargetReference: "Loop",
					},
				}},
				Assignments: []models.FlowAssignment{{Name: "Done"}},
			},
		},
		{
			name: "input collection path and atomic-safe platform action",
			metadata: models.FlowMetadata{
				RunAtomically: true,
				Variables: []models.FlowVariable{{
					Name: "payload", DataType: "Any", IsInput: true,
				}},
				StartElementReference: "Loop",
				Loops: []models.FlowLoop{{
					Name:                "Loop",
					CollectionReference: "payload.items",
					NextValueConnector: &models.FlowConnector{
						TargetReference: "Retry",
					},
					NoMoreValuesConnector: &models.FlowConnector{
						TargetReference: "Done",
					},
				}},
				ActionCalls: []models.FlowActionCall{{
					Name:       "Retry",
					ActionType: "setRetriable",
					ActionName: "setRetriable",
					Connector: &models.FlowConnector{
						TargetReference: "Loop",
					},
				}},
				Assignments: []models.FlowAssignment{{Name: "Done"}},
			},
		},
		{
			name:    "engine-governed filtered write",
			wantMax: 2,
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "Update",
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", FilterLogic: "and",
					Filters: []models.FlowRecordFilter{{
						Field: "ExternalKey", Operator: "EqualTo",
						Value: map[string]interface{}{
							"elementReference": "externalKey",
						},
					}},
					Connector: &models.FlowConnector{
						TargetReference: "Done",
					},
				}},
				Assignments: []models.FlowAssignment{{Name: "Done"}},
			},
		},
		{
			name: "declared Loop owns its complete strongly-connected body",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "Find",
				Variables: []models.FlowVariable{{
					Name: "payload", DataType: "Any", IsInput: true,
				}},
				RecordLookups: []models.FlowRecordLookup{{
					Name: "Find", Object: "Widget", GetFirstRecordOnly: true,
					Connector: &models.FlowConnector{
						TargetReference: "Handle",
					},
				}},
				Decisions: []models.FlowDecision{{
					Name: "Handle",
					DefaultConnector: &models.FlowConnector{
						TargetReference: "Loop",
					},
				}},
				Loops: []models.FlowLoop{{
					Name:                "Loop",
					CollectionReference: "payload.items",
					NextValueConnector: &models.FlowConnector{
						TargetReference: "Work",
					},
					NoMoreValuesConnector: &models.FlowConnector{
						TargetReference: "Done",
					},
				}},
				Assignments: []models.FlowAssignment{
					{
						Name: "Work",
						Connector: &models.FlowConnector{
							TargetReference: "Find",
						},
					},
					{Name: "Done"},
				},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			view := profileTestView{
				"root": profileTestFlowVersion(test.metadata),
			}
			selection, err := SelectPlatformCommand(
				context.Background(),
				view,
				"Root",
				automation.ValidateAtomicFlowForExecution,
			)
			if err != nil {
				t.Fatalf("SelectPlatformCommand() error = %v", err)
			}
			wantMax := test.wantMax
			if wantMax == 0 {
				wantMax = MaxElementExecutions
			}
			if selection.MaxElementExecutions != wantMax {
				t.Fatalf(
					"Platform-command execution bound = %d, want %d",
					selection.MaxElementExecutions,
					wantMax,
				)
			}
			if _, err := SelectSynchronousAPI(
				context.Background(),
				view,
				"Root",
			); err == nil {
				t.Fatal("SelectSynchronousAPI() accepted a Worker-only shape")
			}
		})
	}
}

func TestPlatformCommandRejectsCycleWithoutDeclaredLoop(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "First",
			Assignments: []models.FlowAssignment{
				{
					Name: "First",
					Connector: &models.FlowConnector{
						TargetReference: "Second",
					},
				},
				{
					Name: "Second",
					Connector: &models.FlowConnector{
						TargetReference: "First",
					},
				},
			},
		}),
	}
	if _, err := SelectPlatformCommand(
		context.Background(),
		view,
		"Root",
		automation.ValidateAtomicFlowForExecution,
	); err == nil || !strings.Contains(err.Error(), "does not pass through a declared Loop") {
		t.Fatalf(
			"SelectPlatformCommand() cycle error = %v, want undeclared cycle rejection",
			err,
		)
	}
}

func TestPlatformCommandRejectsInnerCycleThatBypassesDeclaredLoop(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Loop",
			Variables: []models.FlowVariable{{
				Name: "items", DataType: "Any", IsInput: true, IsCollection: true,
			}},
			Loops: []models.FlowLoop{{
				Name:                "Loop",
				CollectionReference: "items",
				NextValueConnector: &models.FlowConnector{
					TargetReference: "First",
				},
				NoMoreValuesConnector: &models.FlowConnector{
					TargetReference: "Done",
				},
			}},
			Assignments: []models.FlowAssignment{
				{
					Name: "First",
					Connector: &models.FlowConnector{
						TargetReference: "Second",
					},
				},
				{Name: "Done"},
			},
			Decisions: []models.FlowDecision{{
				Name: "Second",
				DefaultConnector: &models.FlowConnector{
					TargetReference: "Loop",
				},
				Rules: []models.FlowDecisionRule{{
					Name: "RepeatWithoutLoop",
					Connector: &models.FlowConnector{
						TargetReference: "First",
					},
				}},
			}},
		}),
	}
	if _, err := SelectPlatformCommand(
		context.Background(),
		view,
		"Root",
		automation.ValidateAtomicFlowForExecution,
	); err == nil || !strings.Contains(err.Error(), "does not pass through a declared Loop") {
		t.Fatalf(
			"SelectPlatformCommand() inner cycle error = %v, want loop-bypass rejection",
			err,
		)
	}
}

func TestPlatformCommandRejectsUnprovenRecordReferenceWrites(t *testing.T) {
	tests := []struct {
		name     string
		metadata models.FlowMetadata
	}{
		{
			name: "declared collection update",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "Update",
				Variables: []models.FlowVariable{{
					Name: "records", DataType: "SObject", IsCollection: true,
				}},
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", InputReference: "records",
				}},
			},
		},
		{
			name: "collection assigned to scalar-named delete alias",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "FindMany",
				RecordLookups: []models.FlowRecordLookup{{
					Name: "FindMany", Object: "Widget", OutputReference: "records",
					Connector: &models.FlowConnector{TargetReference: "Alias"},
				}},
				Assignments: []models.FlowAssignment{{
					Name: "Alias",
					Items: []models.FlowAssignmentItem{{
						AssignToReference: "recordAlias",
						Operator:          "Assign",
						Value: map[string]interface{}{
							"elementReference": "records",
						},
					}},
					Connector: &models.FlowConnector{TargetReference: "Delete"},
				}},
				RecordDeletes: []models.FlowRecordDelete{{
					Name: "Delete", Object: "Widget", InputReference: "recordAlias",
				}},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			view := profileTestView{
				"root": profileTestFlowVersion(test.metadata),
			}
			if _, err := SelectPlatformCommand(
				context.Background(),
				view,
				"Root",
				automation.ValidateAtomicFlowForExecution,
			); err == nil || !strings.Contains(err.Error(), "single-record cardinality") {
				t.Fatalf(
					"SelectPlatformCommand() record-reference error = %v, want proven-single rejection",
					err,
				)
			}
		})
	}
}

func TestPlatformCommandRejectsConflictingRecordReferenceCardinality(
	t *testing.T,
) {
	tests := []struct {
		name     string
		metadata models.FlowMetadata
	}{
		{
			name: "update alias is produced as both one record and a collection",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "FindOne",
				RecordLookups: []models.FlowRecordLookup{
					{
						Name: "FindOne", Object: "Widget", OutputReference: "$target",
						GetFirstRecordOnly: true,
						Connector:          &models.FlowConnector{TargetReference: "FindMany"},
					},
					{
						Name: "FindMany", Object: "Widget", OutputReference: "$target",
						Connector: &models.FlowConnector{TargetReference: "Update"},
					},
				},
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", InputReference: "$target",
				}},
			},
		},
		{
			name: "delete alias is produced as both one record and a collection",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "FindOne",
				RecordLookups: []models.FlowRecordLookup{
					{
						Name: "FindOne", Object: "Widget", OutputReference: "$target",
						GetFirstRecordOnly: true,
						Connector:          &models.FlowConnector{TargetReference: "FindMany"},
					},
					{
						Name: "FindMany", Object: "Widget", OutputReference: "$target",
						Connector: &models.FlowConnector{TargetReference: "Delete"},
					},
				},
				RecordDeletes: []models.FlowRecordDelete{{
					Name: "Delete", Object: "Widget", InputReference: "$target",
				}},
			},
		},
		{
			name: "assignment can replace update alias with a collection",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "FindOne",
				RecordLookups: []models.FlowRecordLookup{
					{
						Name: "FindOne", Object: "Widget", OutputReference: "$target",
						GetFirstRecordOnly: true,
						Connector:          &models.FlowConnector{TargetReference: "FindMany"},
					},
					{
						Name: "FindMany", Object: "Widget", OutputReference: "records",
						Connector: &models.FlowConnector{TargetReference: "ReplaceTarget"},
					},
				},
				Assignments: []models.FlowAssignment{{
					Name: "ReplaceTarget",
					Items: []models.FlowAssignmentItem{{
						AssignToReference: "$target",
						Operator:          "Assign",
						Value: map[string]interface{}{
							"elementReference": "records",
						},
					}},
					Connector: &models.FlowConnector{TargetReference: "Update"},
				}},
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", InputReference: "$target",
				}},
			},
		},
		{
			name: "assignment can replace delete alias with a collection",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "FindOne",
				RecordLookups: []models.FlowRecordLookup{
					{
						Name: "FindOne", Object: "Widget", OutputReference: "$target",
						GetFirstRecordOnly: true,
						Connector:          &models.FlowConnector{TargetReference: "FindMany"},
					},
					{
						Name: "FindMany", Object: "Widget", OutputReference: "records",
						Connector: &models.FlowConnector{TargetReference: "ReplaceTarget"},
					},
				},
				Assignments: []models.FlowAssignment{{
					Name: "ReplaceTarget",
					Items: []models.FlowAssignmentItem{{
						AssignToReference: "$target",
						Operator:          "Assign",
						Value: map[string]interface{}{
							"elementReference": "records",
						},
					}},
					Connector: &models.FlowConnector{TargetReference: "Delete"},
				}},
				RecordDeletes: []models.FlowRecordDelete{{
					Name: "Delete", Object: "Widget", InputReference: "$target",
				}},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			view := profileTestView{
				"root": profileTestFlowVersion(test.metadata),
			}
			if _, err := SelectPlatformCommand(
				context.Background(),
				view,
				"Root",
				automation.ValidateAtomicFlowForExecution,
			); err == nil || !strings.Contains(err.Error(), "single-record cardinality") {
				t.Fatalf(
					"SelectPlatformCommand() cardinality error = %v, want conflict rejection",
					err,
				)
			}
		})
	}
}

func TestPlatformCommandAcceptsProvenSingleLookupRecordReferenceWrites(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "FindOne",
			RecordLookups: []models.FlowRecordLookup{{
				Name:               "FindOne",
				Object:             "Widget",
				OutputReference:    "$record",
				GetFirstRecordOnly: true,
				Connector: &models.FlowConnector{
					TargetReference: "SetName",
				},
			}},
			Assignments: []models.FlowAssignment{{
				Name: "SetName",
				Items: []models.FlowAssignmentItem{{
					AssignToReference: "$record.Name",
					Operator:          "Assign",
					Value:             "updated",
				}},
				Connector: &models.FlowConnector{TargetReference: "Update"},
			}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "Update", Object: "Widget", InputReference: "$record",
				Connector: &models.FlowConnector{
					TargetReference: "Delete",
				},
			}},
			RecordDeletes: []models.FlowRecordDelete{{
				Name: "Delete", Object: "Widget", InputReference: "$record",
			}},
		}),
	}
	if _, err := SelectPlatformCommand(
		context.Background(),
		view,
		"Root",
		automation.ValidateAtomicFlowForExecution,
	); err != nil {
		t.Fatalf(
			"SelectPlatformCommand() rejected proven-single record writes: %v",
			err,
		)
	}
}

func TestPlatformCommandRejectsRecordReferenceWithoutAuthoritativeLookupProvenance(t *testing.T) {
	tests := []struct {
		name     string
		metadata models.FlowMetadata
	}{
		{
			name: "disconnected lookup does not authorize update input",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "Update",
				RecordLookups: []models.FlowRecordLookup{{
					Name: "DisconnectedLookup", Object: "Widget",
					OutputReference: "$target", GetFirstRecordOnly: true,
				}},
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", InputReference: "$target",
				}},
			},
		},
		{
			name: "branch that bypasses lookup does not authorize delete input",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "ChoosePath",
				Decisions: []models.FlowDecision{{
					Name: "ChoosePath",
					Rules: []models.FlowDecisionRule{{
						Name:      "UseLookup",
						Connector: &models.FlowConnector{TargetReference: "FindOne"},
					}},
					DefaultConnector: &models.FlowConnector{TargetReference: "Delete"},
				}},
				RecordLookups: []models.FlowRecordLookup{{
					Name: "FindOne", Object: "Widget", OutputReference: "$target",
					GetFirstRecordOnly: true,
					Connector:          &models.FlowConnector{TargetReference: "Delete"},
				}},
				RecordDeletes: []models.FlowRecordDelete{{
					Name: "Delete", Object: "Widget", InputReference: "$target",
				}},
			},
		},
		{
			name: "lookup fault path does not authorize update input",
			metadata: models.FlowMetadata{
				RunAtomically:         true,
				StartElementReference: "FindOne",
				RecordLookups: []models.FlowRecordLookup{{
					Name: "FindOne", Object: "Widget", OutputReference: "$target",
					GetFirstRecordOnly: true,
					Connector:          &models.FlowConnector{TargetReference: "Update"},
					FaultConnector:     &models.FlowConnector{TargetReference: "Update"},
				}},
				RecordUpdates: []models.FlowRecordUpdate{{
					Name: "Update", Object: "Widget", InputReference: "$target",
				}},
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			view := profileTestView{"root": profileTestFlowVersion(test.metadata)}
			if _, err := SelectPlatformCommand(
				context.Background(),
				view,
				"Root",
				automation.ValidateAtomicFlowForExecution,
			); err == nil || !strings.Contains(err.Error(), "provenance") {
				t.Fatalf(
					"SelectPlatformCommand() error = %v, want lookup provenance rejection",
					err,
				)
			}
		})
	}
}

func TestPlatformCommandRejectsDeferredDependencyAction(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "WaitForIdentity",
			ActionCalls: []models.FlowActionCall{{
				Name: "WaitForIdentity", ActionName: "setIdentityConvergenceDependency",
				ActionType: "setIdentityConvergenceDependency",
			}},
		}),
	}
	if _, err := SelectPlatformCommand(
		context.Background(),
		view,
		"Root",
		automation.ValidateAtomicFlowForExecution,
	); err == nil || !strings.Contains(err.Error(), "no deferred dependency lifecycle") {
		t.Fatalf(
			"SelectPlatformCommand() dependency error = %v, want static rejection",
			err,
		)
	}
	if _, err := SelectPlatformCommandWithDependency(
		context.Background(),
		view,
		"Root",
		automation.ValidateAtomicFlowForExecution,
	); err != nil {
		t.Fatalf(
			"SelectPlatformCommandWithDependency() rejected V2 dependency profile: %v",
			err,
		)
	}
	if _, err := SelectSynchronousAPI(
		context.Background(),
		view,
		"Root",
	); err == nil {
		t.Fatal("SelectSynchronousAPI() accepted deferred dependency action")
	}
}

func TestPlatformCommandAcceptsDirectSameOwnerNativeSubflowClosure(t *testing.T) {
	view := profileTestView{
		"root": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Child",
			Subflows: []models.FlowSubflow{{
				Name: "Child", FlowName: "ProtectedChild",
			}},
		}),
		"protectedchild": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		}),
	}
	selection, err := SelectPlatformCommand(
		context.Background(),
		view,
		"Root",
		automation.ValidateAtomicFlowForExecution,
	)
	if err != nil {
		t.Fatalf("SelectPlatformCommand() error = %v", err)
	}
	if len(selection.Versions) != 2 || selection.MaxElementExecutions != 2 {
		t.Fatalf("SelectPlatformCommand() selection = %#v, want two-member closure with shared bound 2", selection)
	}
}

func TestPlatformCommandDirectNativeSubflowSharesHardTransactionBudget(t *testing.T) {
	view := profileTestView{
		"root": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
			RunAtomically: true,
			Variables: []models.FlowVariable{{
				Name: "items", DataType: "Any", IsInput: true, IsCollection: true,
			}},
			StartElementReference: "Loop",
			Loops: []models.FlowLoop{{
				Name: "Loop", CollectionReference: "items",
				NextValueConnector:    &models.FlowConnector{TargetReference: "Work"},
				NoMoreValuesConnector: &models.FlowConnector{TargetReference: "Child"},
			}},
			Assignments: []models.FlowAssignment{{
				Name: "Work", Connector: &models.FlowConnector{TargetReference: "Loop"},
			}},
			Subflows: []models.FlowSubflow{{
				Name: "Child", FlowName: "ProtectedChild",
			}},
		}),
		"protectedchild": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		}),
	}
	selection, err := SelectPlatformCommand(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	)
	if err != nil {
		t.Fatalf("SelectPlatformCommand() error = %v", err)
	}
	if selection.MaxElementExecutions != MaxElementExecutions {
		t.Fatalf(
			"SelectPlatformCommand() shared bound = %d, want hard transaction bound %d",
			selection.MaxElementExecutions,
			MaxElementExecutions,
		)
	}
}

func TestPlatformCommandRejectsChildFlowWithoutExactOneHopSameOwnerAuthority(t *testing.T) {
	standardChild := func(owner string) *models.FlowVersion {
		return profileTestOwnedFlowVersion(owner, models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Done",
			Assignments:           []models.FlowAssignment{{Name: "Done"}},
		})
	}
	for _, test := range []struct {
		name string
		view profileTestView
	}{
		{
			name: "flow action has no native delegation seam",
			view: profileTestView{
				"root": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
					RunAtomically:         true,
					StartElementReference: "Child",
					ActionCalls: []models.FlowActionCall{{
						Name: "Child", ActionName: "ProtectedChild", ActionType: "flow",
					}},
				}),
				"protectedchild": standardChild("companion-pack"),
			},
		},
		{
			name: "different owner",
			view: profileTestView{
				"root": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
					RunAtomically:         true,
					StartElementReference: "Child",
					Subflows:              []models.FlowSubflow{{Name: "Child", FlowName: "ProtectedChild"}},
				}),
				"protectedchild": standardChild("other-pack"),
			},
		},
		{
			name: "customized child",
			view: func() profileTestView {
				child := standardChild("companion-pack")
				child.DefinitionIsCustomized = true
				return profileTestView{
					"root": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
						RunAtomically:         true,
						StartElementReference: "Child",
						Subflows:              []models.FlowSubflow{{Name: "Child", FlowName: "ProtectedChild"}},
					}),
					"protectedchild": child,
				}
			}(),
		},
		{
			name: "transitive grandchild",
			view: profileTestView{
				"root": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
					RunAtomically:         true,
					StartElementReference: "Child",
					Subflows:              []models.FlowSubflow{{Name: "Child", FlowName: "ProtectedChild"}},
				}),
				"protectedchild": profileTestOwnedFlowVersion("companion-pack", models.FlowMetadata{
					RunAtomically:         true,
					StartElementReference: "Grandchild",
					Subflows:              []models.FlowSubflow{{Name: "Grandchild", FlowName: "Grandchild"}},
				}),
				"grandchild": standardChild("companion-pack"),
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			if _, err := SelectPlatformCommand(
				context.Background(), test.view, "Root",
				automation.ValidateAtomicFlowForExecution,
			); err == nil || !strings.Contains(err.Error(), "direct child Flow authorization") {
				t.Fatalf(
					"SelectPlatformCommand() child error=%v, want one-hop same-owner authority rejection",
					err,
				)
			}
		})
	}
}

func TestCatalogActionAcceptsAtomicPlatformActionAndExactChildClosure(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Mark",
			ActionCalls: []models.FlowActionCall{{
				Name: "Mark", ActionType: "setRetriable", ActionName: "setRetriable",
				Connector: &models.FlowConnector{TargetReference: "Child"},
			}},
			Subflows: []models.FlowSubflow{{
				Name: "Child", FlowName: "Child",
				Connector: &models.FlowConnector{TargetReference: "Done"},
			}},
			Assignments: []models.FlowAssignment{{Name: "Done"}},
		}),
		"child": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically: true, StartElementReference: "Done",
			Assignments: []models.FlowAssignment{{Name: "Done"}},
		}),
	}
	selection, err := SelectCatalogAction(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	)
	if err != nil {
		t.Fatalf("SelectCatalogAction() error = %v", err)
	}
	if len(selection.Versions) != 2 ||
		selection.MaxElementExecutions != MaxElementExecutions {
		t.Fatalf("selection = %#v", selection)
	}
	if _, err := SelectSynchronousAPI(
		context.Background(), view, "Root",
	); err == nil {
		t.Fatal("SelectSynchronousAPI() accepted the Catalog-action profile")
	}
	if _, err := SelectPlatformCommand(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	); err == nil || !strings.Contains(err.Error(), "no direct child Flow authorization") {
		t.Fatalf("SelectPlatformCommand() error = %v, want child authority rejection", err)
	}
}

func TestCatalogActionAcceptsBoundedLoopCollection(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically: true,
			Variables: []models.FlowVariable{{
				Name: "items", DataType: "Any", IsInput: true, IsCollection: true,
			}},
			StartElementReference: "Loop",
			Loops: []models.FlowLoop{{
				Name: "Loop", CollectionReference: "items",
				NextValueConnector:    &models.FlowConnector{TargetReference: "Work"},
				NoMoreValuesConnector: &models.FlowConnector{TargetReference: "Done"},
			}},
			Assignments: []models.FlowAssignment{
				{Name: "Work", Connector: &models.FlowConnector{TargetReference: "Loop"}},
				{Name: "Done"},
			},
		}),
	}
	selection, err := SelectCatalogAction(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	)
	if err != nil {
		t.Fatalf("SelectCatalogAction() error = %v", err)
	}
	if selection.MaxElementExecutions != MaxElementExecutions {
		t.Fatalf("Catalog-action execution bound = %d", selection.MaxElementExecutions)
	}
	if _, err := SelectSynchronousAPI(
		context.Background(), view, "Root",
	); err == nil {
		t.Fatal("SelectSynchronousAPI() accepted the Catalog-action loop profile")
	}
}

func TestCatalogActionAcceptsBoundedLoopInExactChildClosure(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically:         true,
			StartElementReference: "Child",
			Subflows: []models.FlowSubflow{{
				Name: "Child", FlowName: "Child",
				Connector: &models.FlowConnector{TargetReference: "Done"},
			}},
			Assignments: []models.FlowAssignment{{Name: "Done"}},
		}),
		"child": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically: true,
			Variables: []models.FlowVariable{{
				Name: "items", DataType: "Any", IsInput: true, IsCollection: true,
			}},
			StartElementReference: "Loop",
			Loops: []models.FlowLoop{{
				Name: "Loop", CollectionReference: "items",
				NextValueConnector:    &models.FlowConnector{TargetReference: "Work"},
				NoMoreValuesConnector: &models.FlowConnector{TargetReference: "Done"},
			}},
			Assignments: []models.FlowAssignment{
				{Name: "Work", Connector: &models.FlowConnector{TargetReference: "Loop"}},
				{Name: "Done"},
			},
		}),
	}
	selection, err := SelectCatalogAction(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	)
	if err != nil {
		t.Fatalf("SelectCatalogAction() error = %v", err)
	}
	if len(selection.Versions) != 2 ||
		selection.MaxElementExecutions != MaxElementExecutions {
		t.Fatalf("selection = %#v", selection)
	}
}

func TestCatalogActionAcceptsRuntimeBoundedInputWriteAndParentPrelock(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically: true,
			Variables: []models.FlowVariable{{
				Name: "recordId", DataType: "String", IsInput: true,
			}},
			AtomicParentPrelocks: []models.FlowAtomicParentPrelock{{
				Name: "RecordParent", ParentObject: "Widget",
				RecordIDs: []interface{}{
					map[string]interface{}{"elementReference": "recordId"},
				},
			}},
			StartElementReference: "Update",
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "Update", Object: "Widget", InputReference: "recordId",
				Connector: &models.FlowConnector{TargetReference: "Done"},
			}},
			Assignments: []models.FlowAssignment{{Name: "Done"}},
		}),
	}
	if _, err := SelectCatalogAction(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	); err != nil {
		t.Fatalf("SelectCatalogAction() error = %v", err)
	}
	if _, err := SelectPlatformCommand(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	); err == nil || !strings.Contains(err.Error(), "prelock shape") {
		t.Fatalf("SelectPlatformCommand() error = %v, want prelock rejection", err)
	}
}

func TestCatalogActionRejectsDeferredDependencyLifecycle(t *testing.T) {
	view := profileTestView{
		"root": profileTestFlowVersion(models.FlowMetadata{
			RunAtomically: true, StartElementReference: "Wait",
			ActionCalls: []models.FlowActionCall{{
				Name:       "Wait",
				ActionType: "setIdentityConvergenceDependency",
				ActionName: "setIdentityConvergenceDependency",
			}},
		}),
	}
	if _, err := SelectCatalogAction(
		context.Background(), view, "Root",
		automation.ValidateAtomicFlowForExecution,
	); err == nil || !strings.Contains(err.Error(), "no deferred dependency lifecycle") {
		t.Fatalf("SelectCatalogAction() dependency error = %v, want rejection", err)
	}
}

func profileTestFlowVersion(
	metadata models.FlowMetadata,
) *models.FlowVersion {
	return &models.FlowVersion{
		Id:                      "version",
		DefinitionId:            "definition",
		VersionNumber:           1,
		Status:                  models.FlowStatusActive,
		ProcessType:             "AutoLaunchedFlow",
		DefinitionRuntimeLoaded: true,
		Metadata:                metadata,
	}
}

func profileTestOwnedFlowVersion(
	ownerPack string,
	metadata models.FlowMetadata,
) *models.FlowVersion {
	version := profileTestFlowVersion(metadata)
	version.Source = "standard"
	version.DefinitionOwnerPack = ownerPack
	return version
}