package automation

import (
	"context"
	"testing"

	"github.com/go-sql-driver/mysql"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

type neutralFlowValidationError struct {
	field          string
	message        string
	panicOnMessage bool
}

func (e neutralFlowValidationError) Error() string           { return e.message }
func (e neutralFlowValidationError) ValidationField() string { return e.field }
func (e neutralFlowValidationError) ValidationMessage() string {
	if e.panicOnMessage {
		panic("flow cause classification must not read presentation messages")
	}
	return e.message
}

type neutralFlowWriteDeniedError struct {
	fields         []string
	message        string
	panicOnMessage bool
}

func (e neutralFlowWriteDeniedError) Error() string { return e.message }
func (e neutralFlowWriteDeniedError) FieldWriteDeniedFields() []string {
	return append([]string(nil), e.fields...)
}
func (e neutralFlowWriteDeniedError) FieldWriteDeniedMessage() string {
	if e.panicOnMessage {
		panic("flow cause classification must not read presentation messages")
	}
	return e.message
}

func TestClassifyFlowExecutionCause(t *testing.T) {
	tests := []struct {
		name      string
		err       error
		code      string
		category  contractmodels.FlowExecutionCauseCategory
		retriable bool
	}{
		{
			name: "TiDB lock wait",
			err:  &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"},
			code: "DB_LOCK_WAIT_TIMEOUT", category: contractmodels.FlowExecutionCauseCategoryConcurrency, retriable: true,
		},
		{
			name: "validation rule",
			err:  apperrors.NewValidationRuleError("AE_Rule", "rejected"),
			code: "VALIDATION_RULE", category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name: "unmarked neutral validation",
			err:  neutralFlowValidationError{field: "actionName", message: "is required", panicOnMessage: true},
			code: "FLOW_ERROR", category: contractmodels.FlowExecutionCauseCategoryUnknown,
		},
		{
			name: "unmarked neutral field write denial",
			err:  neutralFlowWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable", panicOnMessage: true},
			code: "FLOW_ERROR", category: contractmodels.FlowExecutionCauseCategoryUnknown,
		},
		{
			name: "marked aggregate follows first neutral validation leaf",
			err: flowRecordMutationValidation(apperrors.NewClientValidationErrors(
				neutralFlowValidationError{field: "Name", message: "is required", panicOnMessage: true},
				neutralFlowWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable", panicOnMessage: true},
			)),
			code: "VALIDATION", category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name: "marked invalid write field remains a validation cause",
			err: flowRecordMutationValidation(
				apperrors.NewInvalidFieldForWriteError([]string{"CatalogNamespaceId"}, "field is catalog-owned"),
			),
			code: "VALIDATION", category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name: "marked aggregate follows first neutral denial leaf",
			err: flowRecordMutationValidation(apperrors.NewClientValidationErrors(
				neutralFlowWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable", panicOnMessage: true},
				neutralFlowValidationError{field: "Name", message: "is required", panicOnMessage: true},
			)),
			code: contract.FlowExecutionErrorCodeFieldWriteDenied, category: contractmodels.FlowExecutionCauseCategoryAuthorization,
		},
		{
			name: "internal wrapper around marked validation stays internal",
			err: apperrors.NewInternalError(
				"rollback failed",
				flowRecordMutationValidation(apperrors.NewClientValidationErrors(
					neutralFlowValidationError{field: "Name", message: "is required", panicOnMessage: true},
					neutralFlowValidationError{field: "Company", message: "is required", panicOnMessage: true},
				)),
			),
			code:     "INTERNAL_ERROR",
			category: contractmodels.FlowExecutionCauseCategoryInternal,
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got := classifyFlowExecutionCause(tc.err, "ActionCall", "ApplyMappings")
			if got == nil || got.Code != tc.code || got.Category != tc.category || got.Retriable != tc.retriable || got.ElementType != "ActionCall" || got.ElementName != "ApplyMappings" {
				t.Fatalf("classifyFlowExecutionCause = %#v", got)
			}
		})
	}
}

func TestClassifyFailedFlowExecutionResultUsesNeutralEngineConditions(t *testing.T) {
	tests := []struct {
		name     string
		code     string
		origin   flowExecutionResultErrorOrigin
		category contractmodels.FlowExecutionCauseCategory
	}{
		{
			name:     "neutral validation condition",
			code:     contract.FlowExecutionErrorCodeValidation,
			origin:   flowExecutionResultErrorOriginNeutralEngine,
			category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name:     "authored validation code collision stays unknown",
			code:     contract.FlowExecutionErrorCodeValidation,
			origin:   flowExecutionResultErrorOriginAuthored,
			category: contractmodels.FlowExecutionCauseCategoryUnknown,
		},
		{
			name:     "authored field-write code collision stays unknown",
			code:     contract.FlowExecutionErrorCodeFieldWriteDenied,
			origin:   flowExecutionResultErrorOriginAuthored,
			category: contractmodels.FlowExecutionCauseCategoryUnknown,
		},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			cause := classifyFailedFlowExecutionResult(&models.FlowExecutionResult{
				IsCustomError: true,
				Errors:        []models.FlowExecutionError{{StatusCode: tc.code}},
			}, tc.origin)
			if cause == nil || cause.Code != tc.code || cause.Category != tc.category {
				t.Fatalf("classifyFailedFlowExecutionResult() = %#v, want code %q category %q", cause, tc.code, tc.category)
			}
		})
	}
}

func TestExecuteFlowOutcomeAlignsValidationProvenanceResultAndCause(t *testing.T) {
	for _, tc := range []struct {
		name         string
		err          error
		wantCustom   bool
		wantErrors   int
		wantStatus   string
		wantCode     string
		wantCategory contractmodels.FlowExecutionCauseCategory
	}{
		{
			name:         "single record field validation is client visible",
			err:          neutralFlowValidationError{field: "Name", message: "is required"},
			wantCustom:   true,
			wantErrors:   1,
			wantStatus:   contract.FlowExecutionErrorCodeValidation,
			wantCode:     "VALIDATION",
			wantCategory: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name:         "single record field write denial is client visible",
			err:          neutralFlowWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable"},
			wantCustom:   true,
			wantErrors:   1,
			wantStatus:   contract.FlowExecutionErrorCodeFieldWriteDenied,
			wantCode:     contract.FlowExecutionErrorCodeFieldWriteDenied,
			wantCategory: contractmodels.FlowExecutionCauseCategoryAuthorization,
		},
		{
			name:         "single validation rule uses neutral result condition",
			err:          apperrors.NewValidationRuleError("Account_Name_Required", "Name is required"),
			wantCustom:   true,
			wantErrors:   1,
			wantStatus:   contract.FlowExecutionErrorCodeValidation,
			wantCode:     "VALIDATION_RULE",
			wantCategory: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name: "aggregate follows first field validation",
			err: apperrors.NewClientValidationErrors(
				neutralFlowValidationError{field: "Name", message: "is required"},
				neutralFlowWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable"},
			),
			wantCustom:   true,
			wantErrors:   2,
			wantStatus:   contract.FlowExecutionErrorCodeValidation,
			wantCode:     "VALIDATION",
			wantCategory: contractmodels.FlowExecutionCauseCategoryValidation,
		},
		{
			name: "aggregate follows first field write denial",
			err: apperrors.NewClientValidationErrors(
				neutralFlowWriteDeniedError{fields: []string{"Secret__c"}, message: "field is not writable"},
				neutralFlowValidationError{field: "Name", message: "is required"},
			),
			wantCustom:   true,
			wantErrors:   2,
			wantStatus:   contract.FlowExecutionErrorCodeFieldWriteDenied,
			wantCode:     contract.FlowExecutionErrorCodeFieldWriteDenied,
			wantCategory: contractmodels.FlowExecutionCauseCategoryAuthorization,
		},
		{
			name: "rollback failure wrapping validation remains internal",
			err: apperrors.NewInternalError(
				"rollback failed",
				neutralFlowValidationError{field: "Name", message: "is required"},
			),
			wantErrors:   1,
			wantStatus:   "FLOW_ERROR",
			wantCode:     "INTERNAL_ERROR",
			wantCategory: contractmodels.FlowExecutionCauseCategoryInternal,
		},
		{
			name: "rollback failure wrapping authored result remains internal",
			err: apperrors.NewInternalError(
				"rollback failed",
				&FlowCustomErrorFault{
					Message:    "authored rejection",
					StatusCode: "DIALECT_ONLY_CODE",
					resultErrors: []models.FlowExecutionError{{
						StatusCode: "DIALECT_ONLY_CODE",
						Message:    "authored rejection",
					}},
				},
			),
			wantErrors:   1,
			wantStatus:   "FLOW_ERROR",
			wantCode:     "INTERNAL_ERROR",
			wantCategory: contractmodels.FlowExecutionCauseCategoryInternal,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			executor, version := validationFailureChildExecutorWithErrorForTest(tc.err)
			executor.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
				return version, nil
			}

			outcome, err := executor.ExecuteFlowOutcome(
				WithTransientFlowInterviews(context.Background()),
				"ValidationFlow",
				historySequenceReplayChildInputsForTest(),
				&models.UserSession{ID: "005-test"},
			)
			if err != nil {
				t.Fatalf("ExecuteFlowOutcome: %v", err)
			}
			if outcome.Result == nil || outcome.Result.IsSuccess || outcome.Result.IsCustomError != tc.wantCustom || len(outcome.Result.Errors) != tc.wantErrors || outcome.Result.Errors[0].StatusCode != tc.wantStatus {
				t.Fatalf("result = %#v, want custom=%t errors=%d first status=%q", outcome.Result, tc.wantCustom, tc.wantErrors, tc.wantStatus)
			}
			if outcome.Cause == nil || outcome.Cause.Code != tc.wantCode || outcome.Cause.Category != tc.wantCategory {
				t.Fatalf("cause = %#v, want %q/%q", outcome.Cause, tc.wantCode, tc.wantCategory)
			}
		})
	}
}

func TestRouteFaultCapturesUnderlyingCauseBeforeCustomErrorPresentation(t *testing.T) {
	ctx, collector := withFlowAttemptCauseCollector(context.Background())
	ctx = withFlowCauseElementType(ctx, "ActionCall")
	interview := &models.FlowInterview{CurrentElement: "ApplyMappings", ContextData: map[string]interface{}{}}
	_, err := (&FlowExecutor{}).routeFault(ctx, interview, "FailMalformedEmail", &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"})
	if err != nil {
		t.Fatalf("routeFault: %v", err)
	}
	if collector.cause == nil || collector.cause.Code != "DB_LOCK_WAIT_TIMEOUT" || collector.cause.ElementType != "ActionCall" || collector.cause.ElementName != "ApplyMappings" {
		t.Fatalf("captured cause = %#v", collector.cause)
	}
	presented := &FlowCustomErrorFault{StatusCode: "EMAIL_MALFORMED", Message: "presented email error"}
	terminal := selectTerminalFlowExecutionCause(collector, presented, "CustomError", "FailMalformedEmail")
	if terminal.Code != "DB_LOCK_WAIT_TIMEOUT" || !isRetriableFlowExecutionFailure(presented, terminal) {
		t.Fatalf("terminal typed failure = %#v, retriable = %v", terminal, isRetriableFlowExecutionFailure(presented, terminal))
	}
	rollbackFailure := apperrors.NewInternalError("rollback failed", presented)
	terminal = selectTerminalFlowExecutionCause(collector, rollbackFailure, "CustomError", "FailMalformedEmail")
	if terminal.Code != "INTERNAL_ERROR" || terminal.Category != contractmodels.FlowExecutionCauseCategoryInternal {
		t.Fatalf("rollback failure cause = %#v, want internal owner to supersede authored presentation", terminal)
	}
}

func TestFlowExecutionOutcomeCollectorCarriesExecutionIDAndCause(t *testing.T) {
	ctx, collector := withFlowExecutionOutcomeCollector(context.Background())
	publishFlowExecutionID(ctx, "flow-execution-123")
	publishFlowExecutionCause(ctx, &contractmodels.FlowExecutionCause{Code: "VALIDATION_RULE", Category: contractmodels.FlowExecutionCauseCategoryValidation})
	if collector.executionID != "flow-execution-123" || collector.cause == nil || collector.cause.Code != "VALIDATION_RULE" {
		t.Fatalf("collector = %#v", collector)
	}
}

func TestExecuteFlowOutcomeKeepsApplyMappingsCauseSeparateFromEmailPresentation(t *testing.T) {
	action := buildConnectorMappingAction(map[string]interface{}{
		"source":       "crm-pack",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "upsert",
		"direction":    "inbound",
		"targetId":     "lead-1",
	})
	action.FaultConnector = &models.FlowConnector{TargetReference: "FailMalformedEmail"}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		ProcessType:             "AutoLaunchedFlow",
		Source:                  flowSourceStandard,
		DefinitionOwnerPack:     "crm-pack",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start:       &models.FlowStart{Connector: &models.FlowConnector{TargetReference: action.Name}},
			ActionCalls: []models.FlowActionCall{*action},
			CustomErrors: []models.FlowCustomError{{
				Name: "FailMalformedEmail",
				CustomErrorMessages: []models.FlowCustomErrorMessage{{
					ErrorMessage: "presented email error",
					StatusCode:   "EMAIL_MALFORMED",
				}},
			}},
		},
	}
	exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
	exec.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) { return version, nil }
	exec.connectorMappingResolverFn = func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return nil, &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"}
	}

	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "crm-pack")
	ctx = constants.WithConnectorInboundFlowExecution(ctx, "HubSpot_Contact_Lead_Upsert", "crm-pack")
	outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(ctx), "HubSpot_Contact_Lead_Upsert", nil, &models.UserSession{ID: "system"})
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome: %v", err)
	}
	if outcome.ExecutionID == "" || outcome.Result == nil || outcome.Result.IsSuccess || !outcome.Result.IsRetriable {
		t.Fatalf("outcome result/identity = %#v", outcome)
	}
	if len(outcome.Result.Errors) != 1 || outcome.Result.Errors[0].StatusCode != "EMAIL_MALFORMED" {
		t.Fatalf("presented errors = %#v", outcome.Result.Errors)
	}
	if outcome.Cause == nil || outcome.Cause.Code != "DB_LOCK_WAIT_TIMEOUT" || outcome.Cause.Category != contractmodels.FlowExecutionCauseCategoryConcurrency || outcome.Cause.ElementType != "ActionCall" || outcome.Cause.ElementName != "ApplyMappings" || !outcome.Cause.Retriable {
		t.Fatalf("underlying cause = %#v", outcome.Cause)
	}
}

func TestExecuteFlowOutcomePreservesNestedFlowCauseAndRetry(t *testing.T) {
	childAction := buildConnectorMappingAction(map[string]interface{}{
		"source":       "crm-pack",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "upsert",
		"direction":    "inbound",
		"targetId":     "lead-1",
	})
	child := &models.FlowVersion{
		Id:                      "child-version",
		DefinitionId:            "child-definition",
		ProcessType:             "AutoLaunchedFlow",
		Source:                  flowSourceStandard,
		DefinitionOwnerPack:     "crm-pack",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start:       &models.FlowStart{Connector: &models.FlowConnector{TargetReference: childAction.Name}},
			ActionCalls: []models.FlowActionCall{*childAction},
		},
	}
	parent := &models.FlowVersion{
		Id:                      "parent-version",
		DefinitionId:            "parent-definition",
		ProcessType:             "AutoLaunchedFlow",
		Source:                  flowSourceStandard,
		DefinitionOwnerPack:     "crm-pack",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
			ActionCalls: []models.FlowActionCall{{
				Name:       "CallChild",
				ActionType: "flow",
				ActionName: "Child",
			}},
		},
	}
	exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
	exec.flowVersionFn = func(_ context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
		if apiName == "Child" {
			return child, nil
		}
		return parent, nil
	}
	exec.connectorMappingResolverFn = func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return nil, &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"}
	}

	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "crm-pack")
	ctx = constants.WithConnectorInboundFlowExecution(ctx, "Parent", "crm-pack")
	ctx = constants.WithStandardActionProtectedFlowExecution(ctx, "Parent")
	outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(ctx), "Parent", nil, &models.UserSession{ID: "system"})
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome: %v", err)
	}
	if outcome.Result == nil || outcome.Result.IsSuccess || !outcome.Result.IsRetriable {
		t.Fatalf("nested result = %#v", outcome.Result)
	}
	if outcome.Cause == nil || outcome.Cause.Code != "DB_LOCK_WAIT_TIMEOUT" || !outcome.Cause.Retriable {
		t.Fatalf("nested cause = %#v", outcome.Cause)
	}
}

func TestExecuteFlowOutcomePreservesNestedFailedResultCause(t *testing.T) {
	for _, tc := range []struct {
		name     string
		child    *models.FlowVersion
		wantCode string
		wantKind contractmodels.FlowExecutionCauseCategory
	}{
		{
			name:     "not found",
			wantCode: "NOT_FOUND",
			wantKind: contractmodels.FlowExecutionCauseCategoryNotFound,
		},
		{
			name: "invalid flow",
			child: &models.FlowVersion{
				Id:           "invalid-child-version",
				DefinitionId: "invalid-child-definition",
				ProcessType:  "AutoLaunchedFlow",
				Metadata: models.FlowMetadata{
					RunAtomically: true,
					ActionCalls: []models.FlowActionCall{{
						Name:       "UnsupportedAtomicAction",
						ActionType: "unsupportedAtomicAction",
					}},
				},
			},
			wantCode: statusCodeInvalidFlow,
			wantKind: contractmodels.FlowExecutionCauseCategoryConfiguration,
		},
		{
			name: "protected flow denial",
			child: &models.FlowVersion{
				Id:                      "protected-child-version",
				DefinitionId:            "protected-child-definition",
				ProcessType:             "AutoLaunchedFlow",
				Source:                  flowSourceStandard,
				DefinitionOwnerPack:     "crm-pack",
				DefinitionRuntimeLoaded: true,
			},
			wantCode: "FLOW_EXECUTION_ERROR",
			wantKind: contractmodels.FlowExecutionCauseCategoryUnknown,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			const parentAPI = "Parent"
			const childAPI = "Child"
			parent := &models.FlowVersion{
				Id:           "parent-version",
				DefinitionId: "parent-definition",
				ProcessType:  "AutoLaunchedFlow",
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
					ActionCalls: []models.FlowActionCall{{
						Name:           "CallChild",
						ActionType:     "flow",
						ActionName:     childAPI,
						FaultConnector: &models.FlowConnector{TargetReference: "FailMalformedEmail"},
					}},
					CustomErrors: []models.FlowCustomError{{
						Name: "FailMalformedEmail",
						CustomErrorMessages: []models.FlowCustomErrorMessage{{
							ErrorMessage: "presented email error",
							StatusCode:   "EMAIL_MALFORMED",
						}},
					}},
				},
			}
			exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
			exec.flowVersionFn = func(_ context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
				switch apiName {
				case parentAPI:
					return parent, nil
				case childAPI:
					return tc.child, nil
				default:
					return nil, nil
				}
			}

			ctx := constants.WithConnectorInboundFlowExecution(context.Background(), parentAPI, "crm-pack")
			outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(ctx), parentAPI, nil, &models.UserSession{ID: "system"})
			if err != nil {
				t.Fatalf("ExecuteFlowOutcome: %v", err)
			}
			if outcome.Result == nil || outcome.Result.IsSuccess {
				t.Fatalf("nested result = %#v", outcome.Result)
			}
			if len(outcome.Result.Errors) != 1 || outcome.Result.Errors[0].StatusCode != "EMAIL_MALFORMED" {
				t.Fatalf("presented errors = %#v", outcome.Result.Errors)
			}
			if outcome.Cause == nil || outcome.Cause.Code != tc.wantCode || outcome.Cause.Category != tc.wantKind || outcome.Cause.ElementType != "ActionCall" || outcome.Cause.ElementName != "CallChild" {
				t.Fatalf("nested result cause = %#v", outcome.Cause)
			}
		})
	}
}

func TestExecuteFlowOutcomePreservesNativeSubflowFailedResultCause(t *testing.T) {
	const parentAPI = "Parent"
	const childAPI = "MissingChild"
	parent := &models.FlowVersion{
		Id:           "parent-version",
		DefinitionId: "parent-definition",
		ProcessType:  "AutoLaunchedFlow",
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
			Subflows: []models.FlowSubflow{{
				Name:           "CallChild",
				FlowName:       childAPI,
				FaultConnector: &models.FlowConnector{TargetReference: "FailMalformedEmail"},
			}},
			CustomErrors: []models.FlowCustomError{{
				Name: "FailMalformedEmail",
				CustomErrorMessages: []models.FlowCustomErrorMessage{{
					ErrorMessage: "presented email error",
					StatusCode:   "EMAIL_MALFORMED",
				}},
			}},
		},
	}
	exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
	exec.flowVersionFn = func(_ context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
		if apiName == parentAPI {
			return parent, nil
		}
		return nil, nil
	}

	ctx := constants.WithConnectorInboundFlowExecution(context.Background(), parentAPI, "crm-pack")
	outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(ctx), parentAPI, nil, &models.UserSession{ID: "system"})
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome: %v", err)
	}
	if outcome.Result == nil || outcome.Result.IsSuccess || len(outcome.Result.Errors) != 1 || outcome.Result.Errors[0].StatusCode != "EMAIL_MALFORMED" {
		t.Fatalf("result = %#v", outcome.Result)
	}
	if outcome.Cause == nil || outcome.Cause.Code != "NOT_FOUND" || outcome.Cause.Category != contractmodels.FlowExecutionCauseCategoryNotFound || outcome.Cause.ElementType != "Subflow" || outcome.Cause.ElementName != "CallChild" {
		t.Fatalf("native subflow cause = %#v", outcome.Cause)
	}
}

func TestExecuteFlowOutcomeReservesExecutionIDForRootPreflightFailure(t *testing.T) {
	for _, tc := range []struct {
		name     string
		version  *models.FlowVersion
		wantCode string
		wantKind contractmodels.FlowExecutionCauseCategory
	}{
		{
			name:     "not found",
			wantCode: "NOT_FOUND",
			wantKind: contractmodels.FlowExecutionCauseCategoryNotFound,
		},
		{
			name: "invalid flow",
			version: &models.FlowVersion{
				Id:           "invalid-root-version",
				DefinitionId: "invalid-root-definition",
				ProcessType:  "AutoLaunchedFlow",
				Metadata: models.FlowMetadata{
					RunAtomically: true,
					ActionCalls:   []models.FlowActionCall{{ActionType: "unsupportedAtomicAction"}},
				},
			},
			wantCode: statusCodeInvalidFlow,
			wantKind: contractmodels.FlowExecutionCauseCategoryConfiguration,
		},
		{
			name: "protected flow",
			version: &models.FlowVersion{
				Id:                      "protected-root-version",
				DefinitionId:            "protected-root-definition",
				ProcessType:             "AutoLaunchedFlow",
				Source:                  flowSourceStandard,
				DefinitionOwnerPack:     "crm-pack",
				DefinitionRuntimeLoaded: true,
			},
			wantCode: "FLOW_EXECUTION_ERROR",
			wantKind: contractmodels.FlowExecutionCauseCategoryUnknown,
		},
	} {
		t.Run(tc.name, func(t *testing.T) {
			exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
			exec.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) {
				return tc.version, nil
			}

			outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(context.Background()), "Root", nil, &models.UserSession{ID: "system"})
			if err != nil {
				t.Fatalf("ExecuteFlowOutcome: %v", err)
			}
			if outcome.ExecutionID == "" || outcome.Result == nil || outcome.Result.IsSuccess {
				t.Fatalf("outcome result/identity = %#v", outcome)
			}
			if outcome.Cause == nil || outcome.Cause.Code != tc.wantCode || outcome.Cause.Category != tc.wantKind {
				t.Fatalf("preflight cause = %#v", outcome.Cause)
			}
		})
	}
}

type missingCompanionFlowVersionQuery struct {
	ports.QueryService
	calls []queryspec.QuerySpec
}

func (q *missingCompanionFlowVersionQuery) QueryWithSpec(_ context.Context, spec queryspec.QuerySpec, _ *models.UserSession) (*models.QueryResult, error) {
	q.calls = append(q.calls, spec)
	return &models.QueryResult{}, nil
}

func TestExecuteFlowVersionByIDOutcomeReservesExecutionIDForMissingCompanion(t *testing.T) {
	query := &missingCompanionFlowVersionQuery{}
	exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, query, nil, nil)

	outcome, err := exec.ExecuteFlowVersionByIdOutcome(
		WithTransientFlowInterviews(context.Background()),
		"HubSpot_Contact_Lead_Upsert",
		"missing-frozen-companion-version",
		nil,
		&models.UserSession{ID: "system"},
	)
	if err != nil {
		t.Fatalf("ExecuteFlowVersionByIdOutcome: %v", err)
	}
	if outcome.ExecutionID == "" || outcome.Result == nil || outcome.Result.IsSuccess {
		t.Fatalf("outcome result/identity = %#v", outcome)
	}
	if outcome.Cause == nil || outcome.Cause.Code != "NOT_FOUND" || outcome.Cause.Category != contractmodels.FlowExecutionCauseCategoryNotFound {
		t.Fatalf("missing companion cause = %#v", outcome.Cause)
	}
	if len(query.calls) != 1 || query.calls[0].ObjectAPIName != constants.TableFlow || query.calls[0].Limit != 1 {
		t.Fatalf("companion lookup = %#v", query.calls)
	}
}

func TestExecuteFlowOutcomeDropsCauseWhenParentFaultRecovers(t *testing.T) {
	childAction := buildConnectorMappingAction(map[string]interface{}{
		"source":       "crm-pack",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "upsert",
		"direction":    "inbound",
		"targetId":     "lead-1",
	})
	child := &models.FlowVersion{
		Id:                      "child-version",
		DefinitionId:            "child-definition",
		ProcessType:             "AutoLaunchedFlow",
		Source:                  flowSourceStandard,
		DefinitionOwnerPack:     "crm-pack",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start:       &models.FlowStart{Connector: &models.FlowConnector{TargetReference: childAction.Name}},
			ActionCalls: []models.FlowActionCall{*childAction},
		},
	}
	parent := &models.FlowVersion{
		Id:                      "parent-version",
		DefinitionId:            "parent-definition",
		ProcessType:             "AutoLaunchedFlow",
		Source:                  flowSourceStandard,
		DefinitionOwnerPack:     "crm-pack",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: "CallChild"}},
			ActionCalls: []models.FlowActionCall{{
				Name:           "CallChild",
				ActionType:     "flow",
				ActionName:     "Child",
				FaultConnector: &models.FlowConnector{TargetReference: "Recovered"},
			}},
			Assignments: []models.FlowAssignment{{Name: "Recovered"}},
		},
	}
	exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
	exec.flowVersionFn = func(_ context.Context, apiName string, _ *models.UserSession) (*models.FlowVersion, error) {
		if apiName == "Child" {
			return child, nil
		}
		return parent, nil
	}
	exec.connectorMappingResolverFn = func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return nil, &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"}
	}

	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "crm-pack")
	ctx = constants.WithConnectorInboundFlowExecution(ctx, "Parent", "crm-pack")
	ctx = constants.WithStandardActionProtectedFlowExecution(ctx, "Parent")
	outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(ctx), "Parent", nil, &models.UserSession{ID: "system"})
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome: %v", err)
	}
	if outcome.Result == nil || !outcome.Result.IsSuccess {
		t.Fatalf("recovered result = %#v", outcome.Result)
	}
	if outcome.Cause != nil {
		t.Fatalf("successful recovery retained cause = %#v", outcome.Cause)
	}
}

func TestExecuteFlowOutcomeReplacesRecoveredFaultCauseWithLaterFailure(t *testing.T) {
	first := buildConnectorMappingAction(map[string]interface{}{
		"source":       "crm-pack",
		"sourceObject": "Contact",
		"targetObject": "Lead",
		"actionType":   "upsert",
		"direction":    "inbound",
		"targetId":     "lead-1",
	})
	first.FaultConnector = &models.FlowConnector{TargetReference: "Recover"}
	version := &models.FlowVersion{
		Id:                      "flow-version",
		DefinitionId:            "flow-definition",
		ProcessType:             "AutoLaunchedFlow",
		Source:                  flowSourceStandard,
		DefinitionOwnerPack:     "crm-pack",
		DefinitionRuntimeLoaded: true,
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{Connector: &models.FlowConnector{TargetReference: first.Name}},
			ActionCalls: []models.FlowActionCall{
				*first,
				{Name: "LaterFailure", ActionType: "flow"},
			},
			Assignments: []models.FlowAssignment{{
				Name:      "Recover",
				Connector: &models.FlowConnector{TargetReference: "LaterFailure"},
			}},
		},
	}
	exec := newFlowExecutorWithPlatformActionCalls(nil, nil, NewFlowInterviewService(nil, nil), nil, nil, nil, nil)
	exec.flowVersionFn = func(context.Context, string, *models.UserSession) (*models.FlowVersion, error) { return version, nil }
	exec.connectorMappingResolverFn = func(context.Context, connector.ResolveQuery) ([]connector.ConnectorFieldMappingInput, error) {
		return nil, &mysql.MySQLError{Number: 1205, Message: "Lock wait timeout exceeded"}
	}

	ctx := context.WithValue(context.Background(), constants.ContextKeyConnectorSyncOrigin, "crm-pack")
	ctx = constants.WithConnectorInboundFlowExecution(ctx, "HubSpot_Contact_Lead_Upsert", "crm-pack")
	outcome, err := exec.ExecuteFlowOutcome(WithTransientFlowInterviews(ctx), "HubSpot_Contact_Lead_Upsert", nil, &models.UserSession{ID: "system"})
	if err != nil {
		t.Fatalf("ExecuteFlowOutcome: %v", err)
	}
	if outcome.Result == nil || outcome.Result.IsSuccess || outcome.Result.IsRetriable {
		t.Fatalf("later failure result = %#v", outcome.Result)
	}
	if outcome.Result.IsCustomError || len(outcome.Result.Errors) != 1 || outcome.Result.Errors[0].StatusCode != "FLOW_ERROR" {
		t.Fatalf("later configuration failure result = %#v, want one non-custom FLOW_ERROR", outcome.Result)
	}
	if outcome.Cause == nil || outcome.Cause.Code != "INTERNAL_ERROR" || outcome.Cause.Category != contractmodels.FlowExecutionCauseCategoryInternal || outcome.Cause.Retriable || outcome.Cause.ElementName != "LaterFailure" {
		t.Fatalf("later failure cause = %#v", outcome.Cause)
	}
}