package automation

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"testing"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/ports"
	records "github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	sharedErrors "github.com/nexuscrm/shared/pkg/errors"
	"github.com/nexuscrm/shared/pkg/models"
)

func TestExecuteFlowVersionWithoutInterviewServiceFailsClosed(t *testing.T) {
	executor := NewFlowExecutor(nil, nil, nil, nil, nil, nil, nil)
	version := &models.FlowVersion{
		Id: "flow-version", DefinitionId: "flow-definition", DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{Label: "Record Trigger"},
	}
	if _, err := executor.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()), version, "Record_Trigger", nil, &models.UserSession{ID: "user-1"},
	); err == nil {
		t.Fatal("executeFlowVersion accepted a composition with no FlowInterviewService")
	}
}

func TestExecuteFlowVersionKeepsGenericFlowValidationInternal(t *testing.T) {
	tests := []struct {
		name     string
		metadata models.FlowMetadata
	}{
		{
			name: "missing child flow action name",
			metadata: models.FlowMetadata{
				StartElementReference: "CallChild",
				ActionCalls: []models.FlowActionCall{{
					Name:       "CallChild",
					ActionType: "flow",
				}},
			},
		},
		{
			name:     "missing flow entry",
			metadata: models.FlowMetadata{Start: &models.FlowStart{}},
		},
		{
			name: "malformed wait event",
			metadata: models.FlowMetadata{
				StartElementReference: "BadWait",
				Waits: []models.FlowWait{{
					Name: "BadWait",
					WaitEvents: []models.FlowWaitEvent{{
						Name:      "BadAlarm",
						EventType: "AlarmEvent",
					}},
				}},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			executor := NewFlowExecutor(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
			version := &models.FlowVersion{
				Id:                      "misconfigured-flow-version",
				DefinitionId:            "misconfigured-flow-definition",
				DefinitionRuntimeLoaded: true,
				Metadata:                tt.metadata,
			}

			result, err := executor.executeFlowVersion(
				WithTransientFlowInterviews(context.Background()),
				version,
				"MisconfiguredFlow",
				nil,
				&models.UserSession{ID: "005-test"},
			)
			if err != nil {
				t.Fatalf("executeFlowVersion() error = %v, want terminal result", err)
			}
			if result == nil || result.IsSuccess || result.IsCustomError || len(result.Errors) != 1 || result.Errors[0].StatusCode != "FLOW_ERROR" {
				t.Fatalf("executeFlowVersion() result = %#v, want one non-custom FLOW_ERROR", result)
			}
		})
	}
}

func TestFlowClientValidationForPresentationRequiresTrustedProvenance(t *testing.T) {
	ruleFailures := apperrors.NewValidationRuleFailures([]apperrors.ValidationRuleFailure{
		{RuleName: "RuleA", Message: "first"},
		{RuleName: "RuleB", Message: "second"},
	})
	assertFlowClientValidationSurfaceable(t, ruleFailures)
	assertFlowClientValidationSurfaceable(t, flowRecordMutationValidation(
		apperrors.NewValidationError("Name", "is required"),
	))
	assertFlowClientValidationSurfaceable(t, flowRecordMutationValidation(
		apperrors.NewFLSDeniedError([]string{"Secret__c"}, "field is not writable"),
	))
	aggregate := apperrors.NewClientValidationErrors(
		apperrors.NewValidationError("Name", "is required"),
		apperrors.NewValidationRuleError("RuleA", "first"),
	)
	assertFlowClientValidationSurfaceable(t, flowRecordMutationValidation(aggregate))
	assertFlowClientValidationSurfaceable(t, fmt.Errorf(
		"approval action failed: %w",
		flowRecordMutationValidation(aggregate),
	))

	for _, tt := range []struct {
		name string
		err  error
	}{
		{name: "unmarked generic validation", err: apperrors.NewValidationError("actionName", "is required")},
		{
			name: "unmarked field denial",
			err:  apperrors.NewFLSDeniedError([]string{"Secret__c"}, "field is not writable"),
		},
		{
			name: "unmarked aggregate",
			err:  aggregate,
		},
		{
			name: "rollback failure wrapping marked validation",
			err: apperrors.NewInternalError(
				"rollback failed",
				flowRecordMutationValidation(aggregate),
			),
		},
		{
			name: "rule and terminal sibling",
			err: errors.Join(
				apperrors.NewValidationRuleError("RuleA", "first"),
				errors.New("metadata authority unavailable"),
			),
		},
		{
			name: "joined tree cannot borrow a child provenance marker",
			err: errors.Join(
				flowRecordMutationValidation(aggregate),
				apperrors.NewValidationError("actionName", "is required"),
			),
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			if _, ok := flowClientValidationForPresentation(tt.err); ok {
				t.Fatalf("flowClientValidationForPresentation(%T) surfaced an ineligible tree", tt.err)
			}
		})
	}
}

func TestExecuteFlowVersion_AtomicChildPropagatesHistorySequenceReplaySignal(t *testing.T) {
	executor, version, replayErr := historySequenceReplayChildExecutorForTest()
	ctx := withAtomicFlowBoundary(WithTransientFlowInterviews(context.Background()))
	result, err := executor.executeFlowVersion(ctx, version, "AtomicChild", historySequenceReplayChildInputsForTest(), &models.UserSession{ID: "005-test"})
	if !errors.Is(err, replayErr) {
		t.Fatalf("executeFlowVersion() error = %v, want child history replay signal", err)
	}
	if result != nil && result.IsSuccess {
		t.Fatalf("executeFlowVersion() result = %#v, want failed child attempt", result)
	}
}

func TestExecutePreparedFlowClosurePropagatesParentPrelockReplaySignal(t *testing.T) {
	replayErr := &parentPrelockReplaySignalForTest{}
	persistence := &stubPersistenceCapture{capture: &captureUpdate{
		err: fmt.Errorf("wrapped child record update: %w", replayErr),
	}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, nil, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	version := &models.FlowVersion{
		Id:                      "child-version",
		DefinitionId:            "child-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "UpdateParent"}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name:           "UpdateParent",
				InputReference: "$Record.Parent__r",
				InputAssignments: []models.FlowInputAssignment{{
					Field: "Description",
					Value: map[string]interface{}{"stringValue": "set by child"},
				}},
			}},
		},
	}
	result, err := executor.ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Background(),
		"AtomicChild",
		map[string]*models.FlowVersion{"AtomicChild": version},
		historySequenceReplayChildInputsForTest(),
		&models.UserSession{ID: "005-test"},
		100,
	)
	if !errors.Is(err, replayErr) {
		t.Fatalf("ExecutePreparedFlowClosureWithinOwnedTransactionOutcome() error = %v, want parent prelock replay signal", err)
	}
	if result != nil && result.Result != nil && result.Result.IsSuccess {
		t.Fatalf("ExecutePreparedFlowClosureWithinOwnedTransactionOutcome() result = %#v, want failed attempt", result)
	}
}

func TestExecuteFlowAction_FaultConnectorCannotSwallowAtomicChildHistoryReplaySignal(t *testing.T) {
	executor, version, replayErr := historySequenceReplayChildExecutorForTest()
	executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
		return version, nil
	}
	interview := &models.FlowInterview{ContextData: historySequenceReplayChildInputsForTest()}
	action := &models.FlowActionCall{
		ActionName: "AtomicChild",
		InputParameters: []models.FlowActionParameter{{
			Name:  "$Record",
			Value: map[string]interface{}{"elementReference": "$Record"},
		}},
		FaultConnector: &models.FlowConnector{TargetReference: "HandleChildFault"},
	}
	ctx := withAtomicFlowBoundary(WithTransientFlowInterviews(context.Background()))
	next, err := executor.executeFlowAction(ctx, interview, action, &models.UserSession{ID: "005-test"})
	if next != "" || !errors.Is(err, replayErr) {
		t.Fatalf("executeFlowAction() = (%q, %v), want (empty, child replay error)", next, err)
	}
	if _, ok := interview.ContextData["$Flow.FaultMessage"]; ok {
		t.Fatal("parent flow action routed child replay signal to its fault connector")
	}
}

func TestExecuteFlowVersionPreservesAllValidationRuleFailures(t *testing.T) {
	executor, version := validationFailureChildExecutorForTest()

	result, err := executor.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()),
		version,
		"ValidationFlow",
		historySequenceReplayChildInputsForTest(),
		&models.UserSession{ID: "005-test"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion() error = %v, want surfaced result", err)
	}
	if result == nil || result.IsSuccess || !result.IsCustomError || len(result.Errors) != 2 {
		t.Fatalf("executeFlowVersion() result = %#v, want two custom validation errors", result)
	}
	wantMessages := []string{"Existing Database is required", "IQM Completed Date is required"}
	wantFields := []string{"Existing_Database__c", "IQM_Completed_Date__c"}
	for i := range result.Errors {
		if result.Errors[i].StatusCode != contract.FlowExecutionErrorCodeValidation ||
			result.Errors[i].Message != wantMessages[i] ||
			len(result.Errors[i].Fields) != 1 || result.Errors[i].Fields[0] != wantFields[i] {
			t.Fatalf("executeFlowVersion() error[%d] = %#v", i, result.Errors[i])
		}
	}
}

func TestExecuteFlowVersionPreservesStaticAndRuleValidationFailures(t *testing.T) {
	executor, version := mixedValidationFailureChildExecutorForTest()

	result, err := executor.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()),
		version,
		"ValidationFlow",
		historySequenceReplayChildInputsForTest(),
		&models.UserSession{ID: "005-test"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion() error = %v, want surfaced result", err)
	}
	if result == nil || result.IsSuccess || !result.IsCustomError || len(result.Errors) != 3 {
		t.Fatalf("executeFlowVersion() result = %#v, want three ordered validation errors", result)
	}
	wantMessages := []string{"is required", "is required", "Industry is required"}
	wantFields := []string{"LastName", "Company", "Industry"}
	for i := range result.Errors {
		if result.Errors[i].StatusCode != contract.FlowExecutionErrorCodeValidation ||
			result.Errors[i].Message != wantMessages[i] ||
			len(result.Errors[i].Fields) != 1 || result.Errors[i].Fields[0] != wantFields[i] {
			t.Fatalf("executeFlowVersion() error[%d] = %#v", i, result.Errors[i])
		}
	}
}

func TestExecuteFlowVersionRejectsMixedValidationAndTerminalErrorTree(t *testing.T) {
	executor, version := validationFailureChildExecutorWithErrorForTest(errors.Join(
		apperrors.NewValidationError("LastName", "is required"),
		fmt.Errorf("metadata authority unavailable"),
	))

	result, err := executor.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()),
		version,
		"ValidationFlow",
		historySequenceReplayChildInputsForTest(),
		&models.UserSession{ID: "005-test"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion() error = %v, want terminal result", err)
	}
	if result == nil || result.IsSuccess || result.IsCustomError || len(result.Errors) != 1 || result.Errors[0].StatusCode != "FLOW_ERROR" {
		t.Fatalf("executeFlowVersion() result = %#v, want one terminal FLOW_ERROR", result)
	}
}

func TestChildFlowBoundariesPreserveAllValidationRuleFailures(t *testing.T) {
	t.Run("flow action", func(t *testing.T) {
		executor, version := validationFailureChildExecutorForTest()
		executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
			return version, nil
		}
		interview := &models.FlowInterview{ContextData: historySequenceReplayChildInputsForTest()}
		action := &models.FlowActionCall{
			ActionName: "ValidationFlow",
			InputParameters: []models.FlowActionParameter{{
				Name:  "$Record",
				Value: map[string]interface{}{"elementReference": "$Record"},
			}},
		}

		next, err := executor.executeFlowAction(
			WithTransientFlowInterviews(context.Background()),
			interview,
			action,
			&models.UserSession{ID: "005-test"},
		)
		if next != "" {
			t.Fatalf("executeFlowAction() next = %q, want empty", next)
		}
		assertFlowClientValidationSurfaceable(t, err)
		assertAllValidationRuleFailures(t, err)
	})

	t.Run("subflow", func(t *testing.T) {
		executor, version := validationFailureChildExecutorForTest()
		executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
			return version, nil
		}
		interview := &models.FlowInterview{ContextData: historySequenceReplayChildInputsForTest()}
		subflow := &models.FlowSubflow{
			FlowName: "ValidationFlow",
			InputAssignments: []models.FlowSubflowInput{{
				Name:  "$Record",
				Value: map[string]interface{}{"elementReference": "$Record"},
			}},
		}

		next, err := executor.executeSubflow(
			WithTransientFlowInterviews(context.Background()),
			interview,
			subflow,
			&models.UserSession{ID: "005-test"},
		)
		if next != "" {
			t.Fatalf("executeSubflow() next = %q, want empty", next)
		}
		assertFlowClientValidationSurfaceable(t, err)
		assertAllValidationRuleFailures(t, err)
	})
}

func TestChildFlowBoundariesPreserveMixedNeutralValidationConditions(t *testing.T) {
	tests := []struct {
		name string
		run  func(*FlowExecutor, *models.FlowVersion) error
	}{
		{
			name: "flow action",
			run: func(executor *FlowExecutor, version *models.FlowVersion) error {
				executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
					return version, nil
				}
				interview := &models.FlowInterview{ContextData: historySequenceReplayChildInputsForTest()}
				action := &models.FlowActionCall{
					ActionName: "ValidationFlow",
					InputParameters: []models.FlowActionParameter{{
						Name:  "$Record",
						Value: map[string]interface{}{"elementReference": "$Record"},
					}},
				}
				_, err := executor.executeFlowAction(
					WithTransientFlowInterviews(context.Background()),
					interview,
					action,
					&models.UserSession{ID: "005-test"},
				)
				return err
			},
		},
		{
			name: "subflow",
			run: func(executor *FlowExecutor, version *models.FlowVersion) error {
				executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
					return version, nil
				}
				interview := &models.FlowInterview{ContextData: historySequenceReplayChildInputsForTest()}
				subflow := &models.FlowSubflow{
					FlowName: "ValidationFlow",
					InputAssignments: []models.FlowSubflowInput{{
						Name:  "$Record",
						Value: map[string]interface{}{"elementReference": "$Record"},
					}},
				}
				_, err := executor.executeSubflow(
					WithTransientFlowInterviews(context.Background()),
					interview,
					subflow,
					&models.UserSession{ID: "005-test"},
				)
				return err
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			executor, version := mixedNeutralValidationFailureChildExecutorForTest()
			err := tt.run(executor, version)
			assertFlowClientValidationSurfaceable(t, err)
			projected, ok := flowExecutionErrorsForPresentation(err)
			want := mixedNeutralFlowExecutionErrorsForTest()
			if !ok || !reflect.DeepEqual(projected, want) {
				t.Fatalf("child Flow projection = %#v, %t; want ordered neutral conditions %#v", projected, ok, want)
			}
		})
	}
}

func TestFailedFlowResultDoesNotReclassifyNonValidationCustomError(t *testing.T) {
	result := &models.FlowExecutionResult{
		ActionName:    "EmailFlow",
		IsSuccess:     false,
		IsCustomError: true,
		Errors: []models.FlowExecutionError{{
			StatusCode: "EMAIL_MALFORMED",
			Message:    "Recipient address is malformed",
			Fields:     []string{"RecipientAddress"},
		}},
	}

	err := errorFromFailedFlowExecutionResult(result, "email flow failed", "EmailFlow", flowExecutionResultErrorOriginAuthored)
	if _, ok := apperrors.AsValidationRuleFailures(err); ok {
		t.Fatalf("error = %T %v, must not reclassify EMAIL_MALFORMED as validation-rule failure", err, err)
	}
	if err == nil || err.Error() != "Recipient address is malformed" {
		t.Fatalf("error = %v, want authored custom-error message", err)
	}
}

func TestNestedFlowBoundariesPreserveOpaqueAuthoredCustomError(t *testing.T) {
	const (
		parentAPI = "ParentFlow"
		childAPI  = "ChildFlow"
	)
	child := &models.FlowVersion{
		Id:                      "child-version",
		DefinitionId:            "child-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "RejectChild"}},
			CustomErrors: []models.FlowCustomError{{
				Name: "RejectChild",
				CustomErrorMessages: []models.FlowCustomErrorMessage{{
					ErrorMessage: "authored child rejection",
				}},
			}},
		},
	}

	for _, tt := range []struct {
		name     string
		metadata models.FlowMetadata
	}{
		{
			name: "flow action",
			metadata: models.FlowMetadata{
				Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
				ActionCalls: []models.FlowActionCall{{
					Name:       "CallChild",
					ActionType: "flow",
					ActionName: childAPI,
				}},
			},
		},
		{
			name: "subflow",
			metadata: models.FlowMetadata{
				Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
				Subflows: []models.FlowSubflow{{
					Name:     "CallChild",
					FlowName: childAPI,
				}},
			},
		},
	} {
		for _, statusCode := range []string{
			"DIALECT_ONLY_CODE",
			contract.FlowExecutionErrorCodeValidation,
			contract.FlowExecutionErrorCodeFieldWriteDenied,
		} {
			t.Run(tt.name+"/"+statusCode, func(t *testing.T) {
				child.Metadata.CustomErrors[0].CustomErrorMessages[0].StatusCode = statusCode
				parent := &models.FlowVersion{
					Id:                      "parent-version",
					DefinitionId:            "parent-definition",
					DefinitionRuntimeLoaded: true,
					Metadata:                tt.metadata,
				}
				executor := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
				executor.flowVersionFn = func(_ context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
					switch apiName {
					case parentAPI:
						return parent, nil
					case childAPI:
						return child, nil
					default:
						return nil, nil
					}
				}

				result, err := executor.ExecuteFlow(
					WithTransientFlowInterviews(context.Background()),
					parentAPI,
					nil,
					&models.UserSession{ID: "user-1"},
				)
				if err != nil {
					t.Fatalf("ExecuteFlow(%s) error = %v, want surfaced result", parentAPI, err)
				}
				wantErrors := []models.FlowExecutionError{{
					StatusCode: statusCode,
					Message:    "authored child rejection",
				}}
				if result == nil || result.IsSuccess || !result.IsCustomError || !reflect.DeepEqual(result.Errors, wantErrors) {
					t.Fatalf("nested result = %#v, want opaque authored errors %#v", result, wantErrors)
				}
			})
		}
	}
}

func TestFieldWriteDenialUsesNeutralFlowKindAndRoundTrips(t *testing.T) {
	denied := apperrors.NewFLSDeniedError([]string{"Secret__c"}, "field is not writable")
	projected, ok := flowExecutionErrorsFromClientValidation(denied)
	if !ok || len(projected) != 1 {
		t.Fatalf("flow projection = %#v, %t; want one field-denial error", projected, ok)
	}
	if projected[0].StatusCode != contract.FlowExecutionErrorCodeFieldWriteDenied {
		t.Fatalf("flow error kind = %q, want neutral %q", projected[0].StatusCode, contract.FlowExecutionErrorCodeFieldWriteDenied)
	}

	result := &models.FlowExecutionResult{
		IsCustomError: true,
		Errors:        projected,
	}
	cause := classifyFailedFlowExecutionResult(result, flowExecutionResultErrorOriginNeutralEngine)
	if cause == nil || cause.Code != contract.FlowExecutionErrorCodeFieldWriteDenied || string(cause.Category) != "authorization" {
		t.Fatalf("flow cause = %#v, want neutral field-denial authorization cause", cause)
	}

	roundTripped, ok := clientValidationErrorFromFlowExecutionResult(result, "ValidationFlow", flowExecutionResultErrorOriginNeutralEngine)
	if !ok {
		t.Fatal("neutral field-denial flow result did not reconstruct a client validation error")
	}
	fields, message, ok := sharedErrors.FieldWriteDeniedDetails(roundTripped)
	if !ok || len(fields) != 1 || fields[0] != "Secret__c" || message != "field is not writable" {
		t.Fatalf("round-trip details = (%#v, %q, %t)", fields, message, ok)
	}
}

func TestInvalidFieldForWritePreservesLegacyFlowPresentation(t *testing.T) {
	invalid := apperrors.NewInvalidFieldForWriteError(
		[]string{"CatalogNamespaceId"},
		"field is owned by Catalog",
	)
	projected, ok := flowExecutionErrorsFromClientValidation(invalid)
	want := []models.FlowExecutionError{{
		StatusCode: contract.FlowExecutionErrorCodeValidation,
		Message:    "field is owned by Catalog",
		Fields:     []string{"CatalogNamespaceId"},
	}}
	if !ok || !reflect.DeepEqual(projected, want) {
		t.Fatalf("invalid-write Flow projection = %#v, %t; want %#v", projected, ok, want)
	}
}

func TestFlowClientValidationConditionsPreserveNeutralOrderedProjection(t *testing.T) {
	projected, ok := flowExecutionErrorsFromClientValidation(mixedNeutralClientValidationForTest())
	if !ok {
		t.Fatal("mixed client validation tree was not projected")
	}
	want := mixedNeutralFlowExecutionErrorsForTest()
	if !reflect.DeepEqual(projected, want) {
		t.Fatalf("flow projection = %#v, want ordered neutral leaves %#v", projected, want)
	}

	reconstructed, ok := clientValidationErrorFromFlowExecutionResult(&models.FlowExecutionResult{
		IsCustomError: true,
		Errors:        projected,
	}, "ValidationFlow", flowExecutionResultErrorOriginNeutralEngine)
	if !ok {
		t.Fatal("neutral mixed Flow result was not reconstructed")
	}
	// The neutral boundary condition deliberately does not encode whether a
	// validation originated in a field check or a metadata rule. Re-projection
	// must still preserve the complete ordered boundary payload.
	reprojected, ok := flowExecutionErrorsFromClientValidation(reconstructed)
	if !ok || !reflect.DeepEqual(reprojected, want) {
		t.Fatalf("reprojected Flow errors = %#v, %t; want %#v", reprojected, ok, want)
	}
}

func mixedNeutralClientValidationForTest() error {
	return apperrors.NewClientValidationErrors(
		apperrors.NewValidationError("Summary", "is required"),
		apperrors.NewValidationRuleFailures([]apperrors.ValidationRuleFailure{
			{RuleName: "Severity_Required", Message: "Severity is required", Field: "Severity"},
			{RuleName: "Assignee_Required", Message: "Assignee is required", Field: "Assignee"},
		}),
		apperrors.NewFLSDeniedError([]string{"PrivateNotes", "AuditMarker"}, "fields are not writable"),
	)
}

func mixedNeutralFlowExecutionErrorsForTest() []models.FlowExecutionError {
	return []models.FlowExecutionError{
		{StatusCode: contract.FlowExecutionErrorCodeValidation, Message: "is required", Fields: []string{"Summary"}},
		{StatusCode: contract.FlowExecutionErrorCodeValidation, Message: "Severity is required", Fields: []string{"Severity"}},
		{StatusCode: contract.FlowExecutionErrorCodeValidation, Message: "Assignee is required", Fields: []string{"Assignee"}},
		{StatusCode: contract.FlowExecutionErrorCodeFieldWriteDenied, Message: "fields are not writable", Fields: []string{"PrivateNotes", "AuditMarker"}},
	}
}

func validationFailureChildExecutorForTest() (*FlowExecutor, *models.FlowVersion) {
	validationErr := apperrors.NewValidationRuleFailures([]apperrors.ValidationRuleFailure{
		{RuleName: "IQM_Existing_Database", Message: "Existing Database is required", Field: "Existing_Database__c"},
		{RuleName: "IQM_Completed_Date", Message: "IQM Completed Date is required", Field: "IQM_Completed_Date__c"},
	})
	return validationFailureChildExecutorWithErrorForTest(validationErr)
}

func mixedValidationFailureChildExecutorForTest() (*FlowExecutor, *models.FlowVersion) {
	validationErr := apperrors.NewClientValidationErrors(
		apperrors.NewValidationError("LastName", "is required"),
		apperrors.NewValidationError("Company", "is required"),
		apperrors.NewValidationRuleFailures([]apperrors.ValidationRuleFailure{{
			RuleName: "Industry_Required", Message: "Industry is required", Field: "Industry",
		}}),
	)
	return validationFailureChildExecutorWithErrorForTest(validationErr)
}

func mixedNeutralValidationFailureChildExecutorForTest() (*FlowExecutor, *models.FlowVersion) {
	return validationFailureChildExecutorWithErrorForTest(mixedNeutralClientValidationForTest())
}

func validationFailureChildExecutorWithErrorForTest(validationErr error) (*FlowExecutor, *models.FlowVersion) {
	persistence := &stubPersistenceCapture{capture: &captureUpdate{err: validationErr}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, nil, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	version := &models.FlowVersion{
		Id:                      "validation-flow-version",
		DefinitionId:            "validation-flow-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "UpdateParent"}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name:           "UpdateParent",
				InputReference: "$Record.Parent__r",
				InputAssignments: []models.FlowInputAssignment{{
					Field: "Description",
					Value: map[string]interface{}{"stringValue": "set by child"},
				}},
			}},
		},
	}
	return executor, version
}

func assertFlowClientValidationSurfaceable(t *testing.T, err error) {
	t.Helper()
	if _, ok := flowClientValidationForPresentation(err); !ok {
		t.Fatalf("error = %T %v, want surfaceable Flow client validation", err, err)
	}
}

func assertAllValidationRuleFailures(t *testing.T, err error) {
	t.Helper()
	failures, ok := apperrors.AsValidationRuleFailures(err)
	if !ok {
		t.Fatalf("error = %T %v, want validation-rule failures", err, err)
	}
	if len(failures) != 2 ||
		failures[0].Message != "Existing Database is required" || failures[0].Field != "Existing_Database__c" ||
		failures[1].Message != "IQM Completed Date is required" || failures[1].Field != "IQM_Completed_Date__c" {
		t.Fatalf("validation-rule failures = %#v, want both messages and fields", failures)
	}
}

func TestExecuteSubflow_FaultConnectorCannotSwallowAtomicChildHistoryReplaySignal(t *testing.T) {
	executor, version, replayErr := historySequenceReplayChildExecutorForTest()
	executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
		return version, nil
	}
	interview := &models.FlowInterview{ContextData: historySequenceReplayChildInputsForTest()}
	subflow := &models.FlowSubflow{
		FlowName: "AtomicChild",
		InputAssignments: []models.FlowSubflowInput{{
			Name:  "$Record",
			Value: map[string]interface{}{"elementReference": "$Record"},
		}},
		FaultConnector: &models.FlowConnector{TargetReference: "HandleChildFault"},
	}
	ctx := withAtomicFlowBoundary(WithTransientFlowInterviews(context.Background()))
	next, err := executor.executeSubflow(ctx, interview, subflow, &models.UserSession{ID: "005-test"})
	if next != "" || !errors.Is(err, replayErr) {
		t.Fatalf("executeSubflow() = (%q, %v), want (empty, child replay error)", next, err)
	}
	if _, ok := interview.ContextData["$Flow.FaultMessage"]; ok {
		t.Fatal("parent subflow routed child replay signal to its fault connector")
	}
}

func TestExecuteSubflow_TypesWrappedRecordPriorInputForChildRecordOperations(t *testing.T) {
	updateCapture := &captureUpdate{}
	executor := minimalPersistenceWithCapture(updateCapture)
	executor.interviews = NewFlowInterviewService(executor.persistence, nil)
	executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
		return &models.FlowVersion{
			Id:                      "child-version",
			DefinitionId:            "child-definition",
			DefinitionRuntimeLoaded: true,
			Status:                  models.FlowStatusActive,
			Metadata: models.FlowMetadata{
				StartElementReference: "UpdatePriorRecord",
				RecordUpdates: []models.FlowRecordUpdate{
					{
						Name:           "UpdatePriorRecord",
						InputReference: "$Record__Prior",
						InputAssignments: []models.FlowInputAssignment{
							{Field: "Description", Value: stringValue("updated by child")},
						},
					},
				},
			},
		}, nil
	}

	parentInterview := &models.FlowInterview{ContextData: map[string]interface{}{
		"$Record__Prior": models.SObject{constants.FieldID: "006000000000001AAA"},
	}}
	subflow := &models.FlowSubflow{
		FlowName: "UpdatePriorChild",
		InputAssignments: []models.FlowSubflowInput{
			{
				Name:  "$Record__Prior",
				Value: map[string]interface{}{"elementReference": "$Record__Prior"},
			},
		},
	}
	user := &models.UserSession{ID: "005-test"}
	ctx := WithTransientFlowInterviews(context.Background())
	runtimeContext := NewFlowRuntimeContext(ctx, user)
	runtimeContext.SetObjectType("Opportunity")
	ctx = withFlowRuntimeContext(ctx, runtimeContext)

	next, err := executor.executeSubflow(ctx, parentInterview, subflow, user)
	if err != nil {
		t.Fatalf("executeSubflow() error = %v", err)
	}
	if next != "" {
		t.Fatalf("executeSubflow() next = %q, want empty", next)
	}
	if len(updateCapture.calls) != 1 {
		t.Fatalf("child Update calls = %d, want 1", len(updateCapture.calls))
	}
	if got := updateCapture.calls[0].objectName; got != "Opportunity" {
		t.Fatalf("child Update object = %q, want Opportunity inferred from wrapped $Record__Prior input", got)
	}
}

func historySequenceReplayChildExecutorForTest() (*FlowExecutor, *models.FlowVersion, *historySequenceReplaySignalForTest) {
	replayErr := &historySequenceReplaySignalForTest{}
	persistence := &stubPersistenceCapture{capture: &captureUpdate{
		err: fmt.Errorf("wrapped child record update: %w", replayErr),
	}}
	executor := NewFlowExecutor(nil, nil, nil, persistence, nil, nil, nil)
	executor.interviews = NewFlowInterviewService(persistence, nil)
	version := &models.FlowVersion{
		Id:                      "child-version",
		DefinitionId:            "child-definition",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "UpdateParent"}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name:           "UpdateParent",
				InputReference: "$Record.Parent__r",
				InputAssignments: []models.FlowInputAssignment{{
					Field: "Description",
					Value: map[string]interface{}{"stringValue": "set by child"},
				}},
			}},
		},
	}
	return executor, version, replayErr
}

func historySequenceReplayChildInputsForTest() map[string]interface{} {
	return map[string]interface{}{
		"$Record": models.SObject{
			constants.FieldID: "child-001",
			"Parent__c":       "acc-001",
			"Parent__r":       parentRecord("Account", "acc-001"),
			"attributes":      map[string]interface{}{"type": "Child__c"},
		},
	}
}

func TestNewFlowExecutorOwnsRecordLookupCoordinator(t *testing.T) {
	executor := NewFlowExecutor(nil, nil, nil, nil, nil, nil, nil)
	if executor.recordLookupQueryGroup == nil {
		t.Fatal("NewFlowExecutor returned no record lookup coordinator")
	}

	literal := &FlowExecutor{}
	result, shared, err := literal.queryRecordLookupWithSingleflight("key", true, func() (*models.QueryResult, error) {
		return &models.QueryResult{Done: true}, nil
	})
	if err != nil {
		t.Fatalf("literal executor fallback: %v", err)
	}
	if shared {
		t.Fatal("literal executor unexpectedly reported a shared query")
	}
	if result == nil || !result.Done {
		t.Fatalf("literal executor result = %#v, want completed query result", result)
	}
}

// ---------------------------------------------------------------------------
// Stub for TestRunPossiblyAtomic_PostUpsertRegistryResetPerRetryAttempt
// ---------------------------------------------------------------------------

// retryTwicePersistenceStub is a minimal PersistenceProvider whose
// RunInPessimisticTransaction runs the callback TWICE in sequence (simulating
// one TiDB 1205/1213 lock-contention retry). The first call's error is
// discarded; the second call's error is returned. IsTransactionActive always returns false so
// runPossiblyAtomic takes the RunInPessimisticTransaction branch.
type retryTwicePersistenceStub struct {
	insertFn func(context.Context, string, models.SObject, *models.UserSession) (models.SObject, error)
}

type connectorInstalledRetryScopeTestKey struct{}

type connectorInstalledRetryScopeResolverStub struct {
	scopeCalls int
}

func (s *connectorInstalledRetryScopeResolverStub) WithConnectorInboundRemoteIdentityInstalledContractRetryScope(ctx context.Context) context.Context {
	s.scopeCalls++
	scopeID := s.scopeCalls
	return context.WithValue(ctx, connectorInstalledRetryScopeTestKey{}, &scopeID)
}

func (*connectorInstalledRetryScopeResolverStub) ResolveConnectorInboundRemoteIdentityInstalledContracts(
	context.Context,
	string,
) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return ConnectorInboundRemoteIdentityInstalledContractSet{}, nil
}

func (*connectorInstalledRetryScopeResolverStub) ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(
	context.Context,
	string,
) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return ConnectorInboundRemoteIdentityInstalledContractSet{}, nil
}

func (*connectorInstalledRetryScopeResolverStub) ResolveConnectorInboundRemoteIdentityInstalledContractSetsInAmbientTx(
	context.Context,
	[]string,
) (map[string]ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	return nil, nil
}

func (s *retryTwicePersistenceStub) IsTransactionActive(context.Context) bool { return false }
func (s *retryTwicePersistenceStub) RunInPessimisticTransaction(ctx context.Context, fn func(context.Context) error) error {
	// First attempt — run but ignore the error (tx rolled back on contention).
	_ = fn(ctx)
	// Second attempt — return its result.
	return fn(ctx)
}

// Remaining PersistenceProvider methods — not exercised; no-op / nil-return.
func (s *retryTwicePersistenceStub) GetLock(_ context.Context, _ string, _ string) (models.SObject, error) {
	return nil, nil
}
func (s *retryTwicePersistenceStub) Insert(ctx context.Context, tableName string, r models.SObject, user *models.UserSession) (models.SObject, error) {
	if s.insertFn != nil {
		return s.insertFn(ctx, tableName, r, user)
	}
	return r, nil
}
func (s *retryTwicePersistenceStub) PreviewBeforeInsert(_ context.Context, _ string, r models.SObject, _ *models.UserSession) (models.SObject, error) {
	return r, nil
}
func (s *retryTwicePersistenceStub) PrelockUpdateParents(_ context.Context, _ string, _ string, _ models.SObject, _ *models.UserSession) error {
	return nil
}
func (s *retryTwicePersistenceStub) Update(_ context.Context, _ string, _ string, _ models.SObject, _ *models.UserSession) error {
	return nil
}
func (s *retryTwicePersistenceStub) PreviewBeforeUpdate(_ context.Context, _ string, _ string, _ models.SObject, _ models.SObject, _ *models.UserSession) (*models.PreparedRecordUpdate, error) {
	return nil, nil
}
func (s *retryTwicePersistenceStub) CommitPreparedUpdate(_ context.Context, _ string, _ string, _ *models.PreparedRecordUpdate, _ models.SObject, _ *models.UserSession) error {
	return nil
}
func (s *retryTwicePersistenceStub) Delete(_ context.Context, _ string, _ string, _ *models.UserSession) error {
	return nil
}
func (s *retryTwicePersistenceStub) GetRaw(_ context.Context, _ string, _ string) (models.SObject, error) {
	return nil, nil
}
func (s *retryTwicePersistenceStub) GetRecordType(_ context.Context, _ string) (string, error) {
	return "", nil
}
func (s *retryTwicePersistenceStub) RunInTransaction(_ context.Context, fn func(context.Context) error) error {
	return fn(context.Background())
}
func (s *retryTwicePersistenceStub) DetachTx(ctx context.Context) context.Context { return ctx }
func (s *retryTwicePersistenceStub) RecomputeActiveFormulaProjection(_ context.Context, _ *models.UserSession, _, _ string, _ []string) (models.SObject, error) {
	return nil, nil
}

type flowRetrySessionRecordRepo struct {
	ports.RecordRepository
	findOneCalls int
}

func (r *flowRetrySessionRecordRepo) FindOne(_ context.Context, tableName string, id string) (models.SObject, error) {
	if tableName != constants.TableUser || id != "005-flow-user" {
		return nil, nil
	}
	r.findOneCalls++
	return models.SObject{
		constants.FieldID:       id,
		constants.FieldUsername: "flow-user@example.com",
		constants.FieldIsActive: true,
	}, nil
}

// ---------------------------------------------------------------------------
// Test
// ---------------------------------------------------------------------------

// TestRunPossiblyAtomic_PostUpsertRegistryResetPerRetryAttempt is the
// regression test for #2080. It verifies that runPossiblyAtomic gives each
// pessimistic-transaction attempt a FRESH post-upsert registry so that a
// rolled-back attempt's AddPostUpsertRecord calls do not leak into the retry.
//
// The fix lives in flow_executor.go runPossiblyAtomic (~line 603):
//
//	attemptFrc := flowRuntimeContextFromCtx(txCtx).CloneWithFreshPostUpsert()
//	txCtx = withFlowRuntimeContext(txCtx, attemptFrc)
//
// Without those two lines the second attempt's entry-length would be 1
// (containing the first attempt's record), not 0.
func TestRunPossiblyAtomic_PostUpsertRegistryResetPerRetryAttempt(t *testing.T) {
	exec := &FlowExecutor{
		persistence: &retryTwicePersistenceStub{},
	}

	// Seed the outer ctx with a FlowRuntimeContext, as a real caller would.
	outerCtx := withFlowRuntimeContext(
		context.Background(),
		NewFlowRuntimeContext(context.Background(), nil),
	)

	// Per-attempt observations.
	var entryLengths []int // len(PostUpsertRecords("Contact")) at entry, per attempt
	var retryOwners []bool
	var callCount int

	err := exec.runPossiblyAtomic(outerCtx, true, func(attemptCtx context.Context) error {
		attemptNo := callCount
		callCount++

		// (a) Record the registry length AT ENTRY into this attempt.
		frc := flowRuntimeContextFromCtx(attemptCtx)
		entryLengths = append(entryLengths, len(frc.PostUpsertRecords("Contact")))
		retryOwners = append(retryOwners, hasAtomicFlowPessimisticRetryOwner(attemptCtx))

		// (b) Publish an attempt-specific record — simulates what
		//     executeRecordCreate / applyRecordUpdatePersistence does.
		frc.AddPostUpsertRecord("Contact", models.SObject{
			"Id": fmt.Sprintf("003attempt-%d", attemptNo),
		})

		return nil
	})
	if err != nil {
		t.Fatalf("runPossiblyAtomic returned unexpected error: %v", err)
	}

	// The stub runs the callback twice.
	if callCount != 2 {
		t.Fatalf("expected callback to run 2 times (one retry), got %d", callCount)
	}

	// KEY ASSERTION — both attempts must start with an EMPTY registry.
	// If CloneWithFreshPostUpsert is removed, the second attempt's entry
	// length would be 1 (the first attempt's "003attempt-0" survives).
	for i, l := range entryLengths {
		if l != 0 {
			t.Errorf("attempt %d: expected entry-length 0 (fresh registry), got %d — "+
				"rolled-back attempt's record leaked into retry", i, l)
		}
	}
	for i, owned := range retryOwners {
		if !owned {
			t.Errorf("attempt %d: full-flow pessimistic retry-owner marker = false, want true", i)
		}
	}
}

func TestRunPossiblyAtomicSharesInstalledContractRetryScopeOnlyWithinOneExecution(t *testing.T) {
	resolver := &connectorInstalledRetryScopeResolverStub{}
	exec := &FlowExecutor{
		persistence:               &retryTwicePersistenceStub{},
		connectorInboundInstalled: resolver,
	}
	run := func() []any {
		var scopes []any
		ctx := withFlowRuntimeContext(
			context.Background(),
			NewFlowRuntimeContext(context.Background(), nil),
		)
		_, err := exec.runPossiblyAtomicWithCompletion(ctx, true, func(attemptCtx context.Context) error {
			scope := attemptCtx.Value(connectorInstalledRetryScopeTestKey{})
			if scope == nil {
				t.Fatal("atomic retry attempt is missing the installed-contract retry scope")
			}
			scopes = append(scopes, scope)
			return nil
		})
		if err != nil {
			t.Fatalf("runPossiblyAtomicWithCompletion: %v", err)
		}
		return scopes
	}

	first := run()
	second := run()
	if len(first) != 2 || first[0] != first[1] {
		t.Fatalf("first execution scopes = %#v, want one scope shared by both attempts", first)
	}
	if len(second) != 2 || second[0] != second[1] {
		t.Fatalf("second execution scopes = %#v, want one scope shared by both attempts", second)
	}
	if first[0] == second[0] {
		t.Fatal("separate atomic executions unexpectedly shared an installed-contract retry scope")
	}
	if resolver.scopeCalls != 2 {
		t.Fatalf("scope creations = %d, want one per atomic execution", resolver.scopeCalls)
	}
}

func TestFlowAttemptRecordReadCacheShadowsOuterCacheOncePerRootAttempt(t *testing.T) {
	outerCtx := records.WithRecordReadCache(context.Background())
	attempt1 := withFlowAttemptRecordReadCache(outerCtx)
	nested := withFlowAttemptRecordReadCache(attempt1)
	attempt2 := withFlowAttemptRecordReadCache(outerCtx)

	if attempt1 == outerCtx {
		t.Fatal("attempt cache reused outer flow/request cache; want fresh attempt-local cache")
	}
	if nested != attempt1 {
		t.Fatal("nested flow received a private records read cache; want root attempt cache shared through subflows")
	}
	if attempt1 == attempt2 {
		t.Fatal("attempt cache context reused across attempts; want a fresh cache per attempt")
	}
	if records.EnsureRecordReadCache(attempt1) != attempt1 {
		t.Fatal("attempt1 does not carry a records read cache")
	}
	if records.EnsureRecordReadCache(attempt2) != attempt2 {
		t.Fatal("attempt2 does not carry a records read cache")
	}
}

func TestFlowAttemptRecordReadCacheNestedInvalidationReachesParent(t *testing.T) {
	sessionRepo := &flowRetrySessionRecordRepo{}
	sessionStore := records.NewPersistenceService(sessionRepo, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)

	parentCtx := withFlowAttemptRecordReadCache(records.WithRecordReadCache(context.Background()))
	if _, err := sessionStore.LoadUserSession(parentCtx, "005-flow-user"); err != nil {
		t.Fatalf("initial LoadUserSession returned error: %v", err)
	}
	if sessionRepo.findOneCalls != 1 {
		t.Fatalf("initial FindOne calls = %d, want 1", sessionRepo.findOneCalls)
	}

	childCtx := withFlowAttemptRecordReadCache(parentCtx)
	records.InvalidateRecordReadCache(childCtx, constants.TableUser, "005-flow-user")

	if _, err := sessionStore.LoadUserSession(parentCtx, "005-flow-user"); err != nil {
		t.Fatalf("second LoadUserSession returned error: %v", err)
	}
	if sessionRepo.findOneCalls != 2 {
		t.Fatalf("FindOne calls after child invalidation = %d, want 2; nested flow invalidated a private cache instead of the parent attempt cache", sessionRepo.findOneCalls)
	}
}

func TestExecuteFlowVersion_AtomicRetryRecordReadCacheRereadsPerAttempt(t *testing.T) {
	sessionRepo := &flowRetrySessionRecordRepo{}
	sessionStore := records.NewPersistenceService(sessionRepo, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)

	persistence := &retryTwicePersistenceStub{}
	var insertCalls int
	persistence.insertFn = func(ctx context.Context, tableName string, record models.SObject, user *models.UserSession) (models.SObject, error) {
		insertCalls++
		if records.EnsureRecordReadCache(ctx) != ctx {
			t.Fatal("RecordCreate insert ctx does not carry an attempt-local records read cache")
		}
		if _, err := sessionStore.LoadUserSession(ctx, "005-flow-user"); err != nil {
			return nil, err
		}
		return models.SObject{constants.FieldID: fmt.Sprintf("003-attempt-%d", insertCalls)}, nil
	}

	engine := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:          persistence,
		formulaEngine:        engine,
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(engine),
		interviews:           NewFlowInterviewService(persistence, nil),
	}

	flowMeta := models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "CreateContact"},
		},
		Variables: []models.FlowVariable{
			{Name: "createdId", DataType: "String", IsOutput: true},
		},
		RecordCreates: []models.FlowRecordCreate{
			{
				Name:                      "CreateContact",
				Object:                    "Contact",
				AssignRecordIdToReference: "createdId",
			},
		},
	}
	flowVersion := &models.FlowVersion{
		Id:           "flow-retry-cache-version",
		DefinitionId: "flow-retry-cache-definition",
		ProcessType:  "Workflow",
		Metadata:     flowMeta,
	}

	outerCtx := records.WithRecordReadCache(WithTransientFlowInterviews(context.Background()))
	result, err := exec.executeFlowVersion(
		outerCtx,
		flowVersion,
		"RetryCacheReset",
		nil,
		&models.UserSession{ID: "005-test"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion returned unexpected error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("executeFlowVersion result = %#v, want success", result)
	}
	if insertCalls != 2 {
		t.Fatalf("Insert calls = %d, want 2 retry attempts", insertCalls)
	}
	if sessionRepo.findOneCalls != 2 {
		t.Fatalf("LoadUserSession FindOne calls = %d, want 2; retry attempt reused stale records read cache", sessionRepo.findOneCalls)
	}
	if got := result.OutputValues["createdId"]; got != "003-attempt-2" {
		t.Fatalf("createdId = %v, want second attempt record id", got)
	}
}

func TestExecuteFlowVersion_AtomicRetryRestoresInitialContextData(t *testing.T) {
	persistence := &retryTwicePersistenceStub{}
	engine := formula.NewEngine()
	exec := &FlowExecutor{
		persistence:          persistence,
		formulaEngine:        engine,
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(engine),
		interviews:           NewFlowInterviewService(persistence, nil),
	}

	flowMeta := models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "BranchOnRecordStage"},
		},
		Variables: []models.FlowVariable{
			{Name: "branch", DataType: "String", IsOutput: true},
		},
		Decisions: []models.FlowDecision{
			{
				Name:                  "BranchOnRecordStage",
				DefaultConnectorLabel: "CleanInput",
				DefaultConnector:      &models.FlowConnector{TargetReference: "MarkCleanAttempt"},
				Rules: []models.FlowDecisionRule{
					{
						Name:           "SawDirtyStageFromEarlierAttempt",
						Label:          "SawDirtyStageFromEarlierAttempt",
						ConditionLogic: "and",
						Connector:      &models.FlowConnector{TargetReference: "MarkLeakedRetryState"},
						Conditions: []models.FlowCondition{
							{
								LeftValueReference: "$Record.Stage__c",
								Operator:           "EqualTo",
								RightValue:         stringValue("dirty"),
							},
						},
					},
				},
			},
		},
		Assignments: []models.FlowAssignment{
			{
				Name: "MarkCleanAttempt",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "$Record.Stage__c", Operator: "Assign", Value: stringValue("dirty")},
					{AssignToReference: "branch", Operator: "Assign", Value: stringValue("clean")},
				},
			},
			{
				Name: "MarkLeakedRetryState",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "branch", Operator: "Assign", Value: stringValue("dirty_seen")},
				},
			},
		},
	}

	flowVersion := &models.FlowVersion{
		Id:           "flow-retry-context-version",
		DefinitionId: "flow-retry-context-definition",
		ProcessType:  "Workflow",
		Metadata:     flowMeta,
	}

	result, err := exec.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()),
		flowVersion,
		"RetryContextReset",
		map[string]interface{}{
			"$Record": models.SObject{
				"Id":       "lead-001",
				"Stage__c": "clean",
			},
		},
		&models.UserSession{ID: "005-test"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion returned unexpected error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("executeFlowVersion result = %#v, want success", result)
	}
	if got := result.OutputValues["branch"]; got != "clean" {
		t.Fatalf("branch = %v, want clean; retry attempt saw leaked ContextData from the rolled-back attempt", got)
	}
}

func TestExecuteFlowVersion_WorkflowAliasesRecordVariablesOnFirstAttempt(t *testing.T) {
	exec := &FlowExecutor{
		formulaEngine:        formula.NewEngine(),
		fieldDefaultResolver: NewDefaultFieldDefaultResolver(formula.NewEngine()),
		interviews:           NewFlowInterviewService(nil, nil),
	}

	flowMeta := models.FlowMetadata{
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "MutateRecord"},
		},
		Variables: []models.FlowVariable{
			{Name: "myVariable_current", DataType: "SObject"},
			{Name: "branch", DataType: "String", IsOutput: true},
		},
		Assignments: []models.FlowAssignment{
			{
				Name:      "MutateRecord",
				Connector: &models.FlowConnector{TargetReference: "CheckAliasView"},
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "$Record.Stage__c", Operator: "Assign", Value: stringValue("dirty")},
				},
			},
			{
				Name: "AliasSawMutation",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "branch", Operator: "Assign", Value: stringValue("alias-sees-dirty")},
				},
			},
			{
				Name: "AliasWentStale",
				Items: []models.FlowAssignmentItem{
					{AssignToReference: "branch", Operator: "Assign", Value: stringValue("alias-stale")},
				},
			},
		},
		Decisions: []models.FlowDecision{
			{
				Name:                  "CheckAliasView",
				DefaultConnectorLabel: "AliasWentStale",
				DefaultConnector:      &models.FlowConnector{TargetReference: "AliasWentStale"},
				Rules: []models.FlowDecisionRule{
					{
						Name:           "AliasSawMutation",
						Label:          "AliasSawMutation",
						ConditionLogic: "and",
						Connector:      &models.FlowConnector{TargetReference: "AliasSawMutation"},
						Conditions: []models.FlowCondition{
							{
								LeftValueReference: "myVariable_current.Stage__c",
								Operator:           "EqualTo",
								RightValue:         stringValue("dirty"),
							},
						},
					},
				},
			},
		},
	}

	flowVersion := &models.FlowVersion{
		Id:           "flow-workflow-alias-version",
		DefinitionId: "flow-workflow-alias-definition",
		ProcessType:  "Workflow",
		Metadata:     flowMeta,
	}

	result, err := exec.executeFlowVersion(
		WithTransientFlowInterviews(context.Background()),
		flowVersion,
		"WorkflowAliasPreserved",
		map[string]interface{}{
			"$Record": models.SObject{
				"Id":       "lead-002",
				"Stage__c": "clean",
			},
		},
		&models.UserSession{ID: "005-test"},
	)
	if err != nil {
		t.Fatalf("executeFlowVersion returned unexpected error: %v", err)
	}
	if result == nil || !result.IsSuccess {
		t.Fatalf("executeFlowVersion result = %#v, want success", result)
	}
	if got := result.OutputValues["branch"]; got != "alias-sees-dirty" {
		t.Fatalf("branch = %v, want alias-sees-dirty; Workflow alias lost shared $Record view on first attempt", got)
	}
}