package runtimeworker

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
)

type runtimeFlowCommandEngineProbe struct {
	outcome                  *contractmodels.FlowExecutionTypedOutcome
	err                      error
	calls                    int
	name                     string
	version                  string
	rootOwnerPack            string
	inputs                   map[string]interface{}
	user                     *models.UserSession
	bound                    uint64
	system                   bool
	connectorNamespace       string
	inboundRootAuthorized    bool
	inboundVersionAuthorized bool
	validationErr            error
	validationCalls          int
}

func (e *runtimeFlowCommandEngineProbe) ValidateAtomicFlowForExecution(
	context.Context,
	string,
	*models.FlowMetadata,
) error {
	e.validationCalls++
	return e.validationErr
}

func (e *runtimeFlowCommandEngineProbe) ExecuteBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	name string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
	bound uint64,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	e.calls++
	e.name = name
	if versions[name] != nil {
		e.version = versions[name].Id
		e.rootOwnerPack = versions[name].DefinitionOwnerPack
	}
	e.inputs = inputs
	e.user = user
	e.bound = bound
	e.system = constants.GetSecurityContext(ctx) == constants.SystemContext
	e.connectorNamespace = constants.ConnectorSyncOriginFromContext(ctx)
	e.inboundRootAuthorized = constants.ConnectorInboundExecutionAuthorizesFlow(
		ctx,
		name,
		e.rootOwnerPack,
	)
	e.inboundVersionAuthorized =
		constants.ConnectorInboundCompanionExecutionAuthorizesVersion(
			ctx,
			e.version,
		)
	return e.outcome, e.err
}

type runtimeFlowCommandUserProbe struct {
	user  *models.UserSession
	err   error
	calls int
}

type runtimeFlowDefinitionAuthorityProbe struct {
	allowed        bool
	namespace      string
	payloadVersion uint32
}

func (a runtimeFlowDefinitionAuthorityProbe) CommandExecutionContext(
	kind string,
	version uint32,
	authorizedWorkloadID string,
	definitionRef string,
	definitionVersion string,
) (string, bool) {
	payloadVersion := a.payloadVersion
	if payloadVersion == 0 {
		payloadVersion = RuntimeFlowCommandPayloadFormatVersion
	}
	allowed := a.allowed &&
		kind == RuntimeFlowCommandKind &&
		version == payloadVersion &&
		authorizedWorkloadID == "integration-a" &&
		definitionRef == "ExactChild" &&
		definitionVersion == "version-child"
	namespace := a.namespace
	if namespace == "" {
		namespace = "connector-a"
	}
	return namespace, allowed
}

func TestRuntimeFlowCommandV2BindsDependencySourceWithoutChangingV1(t *testing.T) {
	_, _, _, flowView := workerRuntimeFlowCommandViewsWithMetadataClosure(
		t,
		[]byte(`{
  "runAtomically": true,
  "startElementReference": "WaitForIdentity",
  "actionCalls": [{
    "name": "WaitForIdentity",
    "actionName": "setIdentityConvergenceDependency",
    "actionType": "setIdentityConvergenceDependency",
    "inputParameters": [
      {"name":"objectApiName","value":{"stringValue":"Lead"}},
      {"name":"recordId","value":{"stringValue":"lead-1"}},
      {"name":"destinationSlotId","value":{"stringValue":"contact"}}
    ]
  }]
}`),
		"connector-a",
		nil,
		nil,
	)
	request := contractmodels.FlowExecutionDependencyRequest{
		Kind:          contractmodels.FlowDependencyKindIdentityConvergence,
		ObjectAPIName: "Lead", RecordID: "lead-1",
		ConnectorOwnerKey: "connector-a", DestinationSlotID: "contact",
	}
	engine := &runtimeFlowCommandEngineProbe{
		outcome: &contractmodels.FlowExecutionTypedOutcome{
			Result: &contractmodels.FlowExecutionResult{
				ActionName: "ExactChild", IsRetriable: true,
			},
			Dependency: &request,
		},
		err: errors.New("flow execution suspended on durable dependency"),
	}
	payload, err := flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
		DefinitionRef: "ExactChild", DefinitionVersion: "version-child",
		CanonicalInputs: []byte(`{"sourceRecord":{"id":"remote-contact-1"}}`),
		DependencySource: flowcommandsemanticv2.DependencySourceScope{
			ObjectKind: "contacts", SourceID: "remote-contact-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	executor := &runtimeFlowCommandDependencyExecutor{
		flows: engine,
		users: &runtimeFlowCommandUserProbe{
			user: &models.UserSession{ID: constants.SystemUserID},
		},
	}
	_, err = executor.ExecutePlatformCommand(
		metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
		PlatformCommandExecution{
			CommandKind:          RuntimeFlowCommandKind,
			PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV2,
			Payload:              payload, CallerWorkloadID: "integration-a",
			DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{
				allowed: true, payloadVersion: RuntimeFlowCommandPayloadFormatVersionV2,
			},
		},
	)
	var dependencyErr *PlatformCommandDependencyError
	if !errors.As(err, &dependencyErr) {
		t.Fatalf("V2 dependency error = %T %v", err, err)
	}
	dependency := dependencyErr.Dependency
	if dependency.Validate() != nil ||
		dependency.ObjectAPIName != "Lead" || dependency.RecordID != "lead-1" ||
		dependency.ConnectorOwnerKey != "connector-a" ||
		dependency.ConnectorNamespace != "connector-a" ||
		dependency.ConnectorSourceObject != "contacts" ||
		dependency.ConnectorRemoteID != "remote-contact-1" ||
		dependency.DestinationSlotID != "contact" {
		t.Fatalf("bound V2 dependency = %#v", dependency)
	}

	v1 := &runtimeFlowCommandExecutor{flows: engine, users: executor.users}
	_, err = v1.ExecutePlatformCommand(
		metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
		PlatformCommandExecution{
			CommandKind:          RuntimeFlowCommandKind,
			PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
			Payload:              []byte(`{"canonicalInputs":{},"definitionRef":"ExactChild","definitionVersion":"version-child"}`),
			CallerWorkloadID:     "integration-a",
			DefinitionAuthority:  runtimeFlowDefinitionAuthorityProbe{allowed: true},
		},
	)
	var executionErr *PlatformCommandExecutionError
	if !errors.As(err, &executionErr) || !executionErr.Permanent {
		t.Fatalf("V1 dependency behavior changed: %T %v", err, err)
	}
}

func TestRuntimeFlowCommandV2PreservesTypedRecoveryRequest(t *testing.T) {
	_, _, _, flowView := workerRuntimeFlowCommandViewsWithMetadataClosure(
		t,
		[]byte(`{
  "runAtomically": true,
  "startElementReference": "RequestRecovery",
  "actionCalls": [{
    "name": "RequestRecovery",
    "actionName": "requestConnectorRemoteIdentityRecovery",
    "actionType": "requestConnectorRemoteIdentityRecovery",
    "inputParameters": [
      {"name":"destinationSlotId","value":{"stringValue":"contact"}},
      {"name":"remoteId","value":{"elementReference":"sourceRecord.id"}}
    ],
    "outputParameters": [
      {"name":"recoveryState","value":{"elementReference":"recoveryState"}}
    ]
  }]
}`),
		"connector-a",
		nil,
		nil,
	)
	request := automation.ConnectorInboundRemoteIdentityRecoveryRequest{
		Selector: automation.ConnectorInboundRemoteIdentityRecoverySelector{
			ConnectorOwnerKey: "connector-a", ConnectorNamespace: "connector-a",
			DestinationSlotID: "contact", RemoteID: "remote-contact-1",
		},
	}
	engine := &runtimeFlowCommandEngineProbe{
		err: &automation.ConnectorInboundRemoteIdentityRecoveryRequestedError{
			Selector: request.Selector,
		},
	}
	payload, err := flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
		DefinitionRef: "ExactChild", DefinitionVersion: "version-child",
		CanonicalInputs: []byte(`{"sourceRecord":{"id":"remote-contact-1"}}`),
		DependencySource: flowcommandsemanticv2.DependencySourceScope{
			ObjectKind: "contacts", SourceID: "remote-contact-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	executor := &runtimeFlowCommandDependencyExecutor{
		flows: engine,
		users: &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}},
	}
	_, err = executor.ExecutePlatformCommand(
		metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
		PlatformCommandExecution{
			CommandKind: RuntimeFlowCommandKind, PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV2,
			Payload: payload, CallerWorkloadID: "integration-a",
			DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{allowed: true, payloadVersion: RuntimeFlowCommandPayloadFormatVersionV2},
		},
	)
	var recoveryErr *PlatformCommandRecoveryError
	if !errors.As(err, &recoveryErr) || recoveryErr.Request.Selector != request.Selector {
		t.Fatalf("V2 recovery error = %T %#v", err, recoveryErr)
	}
}

func (u *runtimeFlowCommandUserProbe) LoadSystemUserSession(
	context.Context,
) (*models.UserSession, error) {
	u.calls++
	return u.user, u.err
}

func TestRuntimeFlowCommandExecutesExactImmutableFlowAsSystemActor(t *testing.T) {
	_, _, flowView := workerScheduledFlowView(t)
	engine := &runtimeFlowCommandEngineProbe{
		outcome: &contractmodels.FlowExecutionTypedOutcome{
			ExecutionID: "execution-1",
			Result: &contractmodels.FlowExecutionResult{
				ActionName: "ExactChild",
				IsSuccess:  true,
				OutputValues: map[string]interface{}{
					"z": "last",
					"a": "first",
				},
			},
		},
	}
	users := &runtimeFlowCommandUserProbe{
		user: &models.UserSession{ID: constants.SystemUserID},
	}
	executor := &runtimeFlowCommandExecutor{flows: engine, users: users}
	ctx := metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
	result, err := executor.ExecutePlatformCommand(ctx, PlatformCommandExecution{
		CommandKind:          RuntimeFlowCommandKind,
		PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
		CallerWorkloadID:     "integration-a",
		DefinitionAuthority:  runtimeFlowDefinitionAuthorityProbe{allowed: true},
		Payload: []byte(
			`{"canonicalInputs":{"count":2,"name":"input"},"definitionRef":"ExactChild","definitionVersion":"version-child"}`,
		),
	})
	if err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 || engine.name != "ExactChild" ||
		engine.version != "version-child" || engine.bound != 1 ||
		engine.user == nil || engine.user.ID != constants.SystemUserID ||
		!engine.system || engine.connectorNamespace != "connector-a" ||
		!engine.inboundRootAuthorized || !engine.inboundVersionAuthorized ||
		users.calls != 1 {
		t.Fatalf("execution = %#v users=%d", engine, users.calls)
	}
	if number, ok := engine.inputs["count"].(json.Number); !ok ||
		number.String() != "2" {
		t.Fatalf("canonical numeric input = %#v", engine.inputs["count"])
	}
	if string(result) !=
		`{"canonicalOutputs":{"a":"first","z":"last"},"executionId":"execution-1"}` {
		t.Fatalf("result = %s", result)
	}
}

func TestRuntimeFlowCommandExecutesExactCrossPackCompanionAuthority(t *testing.T) {
	_, _, _, flowView := workerRuntimeFlowCommandViewsWithMetadataClosure(
		t,
		[]byte(
			`{"runAtomically":true,"startElementReference":"Assign","assignments":[{"name":"Assign"}]}`,
		),
		"companion-pack",
		nil,
		nil,
	)
	engine := &runtimeFlowCommandEngineProbe{
		outcome: &contractmodels.FlowExecutionTypedOutcome{
			ExecutionID: "execution-cross-pack",
			Result: &contractmodels.FlowExecutionResult{
				ActionName:   "ExactChild",
				IsSuccess:    true,
				OutputValues: map[string]interface{}{},
			},
		},
	}
	executor := &runtimeFlowCommandExecutor{
		flows: engine,
		users: &runtimeFlowCommandUserProbe{
			user: &models.UserSession{ID: constants.SystemUserID},
		},
	}
	ctx := metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
	result, err := executor.ExecutePlatformCommand(ctx, PlatformCommandExecution{
		CommandKind:          RuntimeFlowCommandKind,
		PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
		CallerWorkloadID:     "integration-a",
		DefinitionAuthority:  runtimeFlowDefinitionAuthorityProbe{allowed: true},
		Payload: []byte(
			`{"canonicalInputs":{},"definitionRef":"ExactChild","definitionVersion":"version-child"}`,
		),
	})
	if err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 || engine.connectorNamespace != "connector-a" ||
		engine.inboundRootAuthorized || !engine.inboundVersionAuthorized {
		t.Fatalf("cross-pack companion execution = %#v", engine)
	}
	if string(result) !=
		`{"canonicalOutputs":{},"executionId":"execution-cross-pack"}` {
		t.Fatalf("result = %s", result)
	}
}

func TestRuntimeFlowCommandFailsClosedBeforeExecution(t *testing.T) {
	_, _, flowView := workerScheduledFlowView(t)
	tests := []struct {
		name          string
		ctx           func() context.Context
		payload       string
		kind          string
		version       uint32
		user          *models.UserSession
		workload      string
		validationErr error
	}{
		{
			name: "missing immutable command authority",
			ctx: func() context.Context {
				return metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
			},
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-child"}`,
		},
		{
			name: "wrong authorized workload",
			ctx: func() context.Context {
				return metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
			},
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-child"}`,
			user:     &models.UserSession{ID: constants.SystemUserID},
			workload: "integration-other",
		},
		{
			name: "missing immutable Flow view",
			ctx:  func() context.Context { return t.Context() },
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-child"}`,
		},
		{
			name: "wrong exact version",
			ctx: func() context.Context {
				return metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
			},
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-other"}`,
		},
		{
			name: "unknown payload field",
			ctx: func() context.Context {
				return metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
			},
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-child","extra":true}`,
		},
		{
			name: "invalid system actor",
			ctx: func() context.Context {
				return metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
			},
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-child"}`,
			user: &models.UserSession{ID: "not-system"},
		},
		{
			name: "action unavailable in worker executor",
			ctx: func() context.Context {
				return metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
			},
			payload: `{"canonicalInputs":{},"definitionRef":"ExactChild",` +
				`"definitionVersion":"version-child"}`,
			validationErr: errors.New("worker action registry does not own action"),
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			engine := &runtimeFlowCommandEngineProbe{validationErr: test.validationErr}
			users := &runtimeFlowCommandUserProbe{user: test.user}
			if test.user == nil {
				users.user = &models.UserSession{ID: constants.SystemUserID}
			}
			executor := &runtimeFlowCommandExecutor{flows: engine, users: users}
			kind := test.kind
			if kind == "" {
				kind = RuntimeFlowCommandKind
			}
			version := test.version
			if version == 0 {
				version = RuntimeFlowCommandPayloadFormatVersion
			}
			workload := test.workload
			if workload == "" {
				workload = "integration-a"
			}
			_, err := executor.ExecutePlatformCommand(
				test.ctx(),
				PlatformCommandExecution{
					CommandKind: kind, PayloadFormatVersion: version,
					Payload:          []byte(test.payload),
					CallerWorkloadID: workload,
					DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{
						allowed: test.name != "missing immutable command authority",
					},
				},
			)
			var executionErr *PlatformCommandExecutionError
			if !errors.As(err, &executionErr) || !executionErr.Permanent ||
				engine.calls != 0 {
				t.Fatalf("err=%v calls=%d", err, engine.calls)
			}
		})
	}
}

func TestRuntimeFlowCommandRetainsBoundedBusinessRejectionEvidence(t *testing.T) {
	_, _, flowView := workerScheduledFlowView(t)
	for _, version := range []uint32{RuntimeFlowCommandPayloadFormatVersion, RuntimeFlowCommandPayloadFormatVersionV2} {
		for _, test := range []struct {
			name       string
			code       string
			custom     bool
			retriable  bool
			wantSuffix string
		}{
			{name: "authored rejection", code: "RECIPIENT_ADDRESS_REJECTED", custom: true, wantSuffix: "RECIPIENT_ADDRESS_REJECTED"},
			{name: "custom evidence does not change retry policy", code: "RETRY_LATER", custom: true, retriable: true, wantSuffix: "RETRY_LATER"},
			{name: "ordinary error remains typed", code: "UNTRUSTED_PRESENTATION_CODE", wantSuffix: "unknown"},
			{name: "missing code", custom: true, wantSuffix: "unknown"},
			{name: "raw provider value is not a code", code: "invalid address private@example.invalid", custom: true, wantSuffix: "unknown"},
			{name: "line break is not a code", code: "REJECTED\nprivate", custom: true, wantSuffix: "unknown"},
			{name: "oversized code", code: strings.Repeat("A", 129), custom: true, wantSuffix: "unknown"},
		} {
			t.Run(fmt.Sprintf("v%d/%s", version, test.name), func(t *testing.T) {
				engine := &runtimeFlowCommandEngineProbe{
					outcome: &contractmodels.FlowExecutionTypedOutcome{
						Result: &contractmodels.FlowExecutionResult{
							IsCustomError: test.custom,
							Errors: []contractmodels.FlowExecutionError{{
								StatusCode: test.code, Message: "private provider payload must not be retained",
							}},
						},
						Cause: &contractmodels.FlowExecutionCause{
							Category:  contractmodels.FlowExecutionCauseCategoryUnknown,
							Retriable: test.retriable,
						},
					},
				}
				users := &runtimeFlowCommandUserProbe{user: &models.UserSession{ID: constants.SystemUserID}}
				command := PlatformCommandExecution{
					CommandKind: RuntimeFlowCommandKind, PayloadFormatVersion: version,
					Payload:             []byte(`{"canonicalInputs":{},"definitionRef":"ExactChild","definitionVersion":"version-child"}`),
					CallerWorkloadID:    "integration-a",
					DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{allowed: true, payloadVersion: version},
				}
				ctx := metadata.WithRuntimeFlowWorkView(t.Context(), flowView)
				var result []byte
				var err error
				if version == RuntimeFlowCommandPayloadFormatVersionV2 {
					command.Payload, err = flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
						DefinitionRef: "ExactChild", DefinitionVersion: "version-child", CanonicalInputs: []byte(`{}`),
						DependencySource: flowcommandsemanticv2.DependencySourceScope{ObjectKind: "contacts", SourceID: "remote-1"},
					})
					if err != nil {
						t.Fatal(err)
					}
					result, err = (&runtimeFlowCommandDependencyExecutor{flows: engine, users: users}).ExecutePlatformCommand(ctx, command)
				} else {
					result, err = (&runtimeFlowCommandExecutor{flows: engine, users: users}).ExecutePlatformCommand(ctx, command)
				}
				var terminal *PlatformCommandExecutionError
				if !errors.As(err, &terminal) || terminal.Permanent == test.retriable || len(result) != 0 {
					t.Fatalf("result=%s err=%v; want failed command with unchanged retry policy", result, err)
				}
				if got, want := terminal.Cause.Error(), "runtime Flow command failed: "+test.wantSuffix; got != want {
					t.Fatalf("terminal evidence=%q, want %q", got, want)
				}
			})
		}
	}
}

func TestRuntimeFlowCommandPreservesRetryAndCancellationClassification(t *testing.T) {
	_, _, flowView := workerScheduledFlowView(t)
	payload := []byte(
		`{"canonicalInputs":{},"definitionRef":"ExactChild","definitionVersion":"version-child"}`,
	)
	for _, test := range []struct {
		name          string
		outcome       *contractmodels.FlowExecutionTypedOutcome
		engineErr     error
		cancelBefore  bool
		wantPermanent bool
		wantCanceled  bool
	}{
		{
			name: "typed retriable failure",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				Result: &contractmodels.FlowExecutionResult{},
				Cause: &contractmodels.FlowExecutionCause{
					Category:  contractmodels.FlowExecutionCauseCategoryConcurrency,
					Retriable: true,
				},
			},
		},
		{
			name: "typed permanent failure",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				Result: &contractmodels.FlowExecutionResult{},
				Cause: &contractmodels.FlowExecutionCause{
					Category: contractmodels.FlowExecutionCauseCategoryConfiguration,
				},
			},
			wantPermanent: true,
		},
		{
			name: "typed budget failure returned with engine error",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				Result: &contractmodels.FlowExecutionResult{},
				Cause: &contractmodels.FlowExecutionCause{
					Category: contractmodels.FlowExecutionCauseCategoryUnknown,
				},
			},
			engineErr:     automation.ErrFlowExecutionBudgetExceeded,
			wantPermanent: true,
		},
		{
			name:          "untyped permanent result",
			outcome:       &contractmodels.FlowExecutionTypedOutcome{Result: &contractmodels.FlowExecutionResult{}},
			wantPermanent: true,
		},
		{
			name:          "missing execution identity",
			outcome:       &contractmodels.FlowExecutionTypedOutcome{Result: &contractmodels.FlowExecutionResult{IsSuccess: true}},
			wantPermanent: true,
		},
		{
			name: "non-canonical output",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				ExecutionID: "execution-invalid-output",
				Result: &contractmodels.FlowExecutionResult{
					IsSuccess:    true,
					OutputValues: map[string]interface{}{"invalid": make(chan struct{})},
				},
			},
			wantPermanent: true,
		},
		{
			name: "successful retriable result",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				Result: &contractmodels.FlowExecutionResult{
					IsSuccess: true, IsRetriable: true,
				},
			},
		},
		{
			name: "deferred dependency is not silently completed",
			outcome: &contractmodels.FlowExecutionTypedOutcome{
				Result: &contractmodels.FlowExecutionResult{IsSuccess: true},
				Dependency: &contractmodels.FlowExecutionDependencyRequest{
					Kind:          contractmodels.FlowDependencyKindIdentityConvergence,
					ObjectAPIName: "Account", RecordID: "account-1",
					ConnectorOwnerKey: "owner-pack",
					DestinationSlotID: "destination",
				},
			},
			wantPermanent: true,
		},
		{name: "engine cancellation", engineErr: context.Canceled, wantCanceled: true},
		{name: "owner already canceled", cancelBefore: true, wantCanceled: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			ctx, cancel := context.WithCancel(
				metadata.WithRuntimeFlowWorkView(t.Context(), flowView),
			)
			if test.cancelBefore {
				cancel()
			} else {
				defer cancel()
			}
			engine := &runtimeFlowCommandEngineProbe{
				outcome: test.outcome,
				err:     test.engineErr,
			}
			executor := &runtimeFlowCommandExecutor{
				flows: engine,
				users: &runtimeFlowCommandUserProbe{
					user: &models.UserSession{ID: constants.SystemUserID},
				},
			}
			_, err := executor.ExecutePlatformCommand(ctx, PlatformCommandExecution{
				CommandKind:          RuntimeFlowCommandKind,
				PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
				Payload:              payload,
				CallerWorkloadID:     "integration-a",
				DefinitionAuthority: runtimeFlowDefinitionAuthorityProbe{
					allowed: true,
				},
			})
			if test.wantCanceled {
				if !errors.Is(err, context.Canceled) {
					t.Fatalf("err=%v", err)
				}
				return
			}
			var executionErr *PlatformCommandExecutionError
			if !errors.As(err, &executionErr) ||
				executionErr.Permanent != test.wantPermanent {
				t.Fatalf("err=%v", err)
			}
		})
	}
}