package runtimeapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"maps"
	"slices"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/automation"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type flowServerTransactionKey struct{}

type flowServerTestStore struct {
	t          *testing.T
	row        persistence.RuntimeFlowInvocationRow
	acceptance persistence.RuntimeFlowInvocationAcceptance
	present    bool
	acceptErr  error
}

func (s *flowServerTestStore) Accept(
	ctx context.Context,
	input persistence.RuntimeFlowInvocationAcceptance,
) (persistence.RuntimeFlowInvocationRow, bool, error) {
	s.t.Helper()
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); !active {
		s.t.Fatal("Flow invocation acceptance escaped the transaction")
	}
	if s.acceptErr != nil {
		return s.row, true, s.acceptErr
	}
	if s.present {
		if s.acceptance.RowID != input.RowID ||
			s.acceptance.ScopeDigest != input.ScopeDigest ||
			s.acceptance.IdempotencyKey != input.IdempotencyKey ||
			s.acceptance.DefinitionRef != input.DefinitionRef ||
			s.acceptance.DefinitionVersion != input.DefinitionVersion ||
			s.acceptance.ClosureDigest != input.ClosureDigest ||
			s.acceptance.RequestDigest != input.RequestDigest ||
			!bytes.Equal(s.acceptance.Receipts, input.Receipts) {
			return s.row, true, persistence.ErrRuntimeFlowInvocationConflict
		}
		return s.row, true, nil
	}
	s.acceptance = input
	s.row = persistence.RuntimeFlowInvocationRow{
		RowID: input.RowID, ScopeDigest: input.ScopeDigest,
		IdempotencyKey: input.IdempotencyKey, DefinitionRef: input.DefinitionRef,
		DefinitionVersion: input.DefinitionVersion, ClosureDigest: input.ClosureDigest,
		RequestDigest: input.RequestDigest, Receipts: append([]byte(nil), input.Receipts...),
		State: persistence.RuntimeFlowInvocationAccepted,
	}
	s.present = true
	return s.row, false, nil
}

func (s *flowServerTestStore) Complete(ctx context.Context, rowID string, result []byte) error {
	s.t.Helper()
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); !active {
		s.t.Fatal("Flow invocation completion escaped the transaction")
	}
	if !s.present || rowID != s.row.RowID {
		s.t.Fatalf("completed unknown Flow invocation %q", rowID)
	}
	s.row.State = persistence.RuntimeFlowInvocationCompleted
	s.row.ResultPayload = append([]byte(nil), result...)
	return nil
}

type flowServerTestRunner struct {
	store                *flowServerTestStore
	attempts             int
	retry                bool
	wrapError            func(error) error
	completionOverride   *ports.TransactionCompletionOutcome
	preserveEffectsOnErr bool
}

func (r *flowServerTestRunner) RunInPessimisticTransaction(
	ctx context.Context,
	run func(context.Context) error,
) error {
	_, err := r.RunInPessimisticTransactionWithOutcome(ctx, run)
	return err
}

func (r *flowServerTestRunner) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	run func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	r.attempts++
	txCtx := context.WithValue(ctx, flowServerTransactionKey{}, true)
	originalRow, originalAcceptance, originalPresent := r.store.row, r.store.acceptance, r.store.present
	err := run(txCtx)
	if err == nil {
		return ports.TransactionCompletionCommitted, nil
	}
	if !r.preserveEffectsOnErr {
		r.store.row, r.store.acceptance, r.store.present = originalRow, originalAcceptance, originalPresent
	}
	if r.wrapError != nil {
		err = r.wrapError(err)
	}
	completion := ports.TransactionCompletionRolledBack
	if r.completionOverride != nil {
		completion = *r.completionOverride
	}
	if !r.retry {
		return completion, err
	}
	r.retry = false
	r.attempts++
	err = run(context.WithValue(ctx, flowServerTransactionKey{}, true))
	if err != nil {
		if !r.preserveEffectsOnErr {
			r.store.row, r.store.acceptance, r.store.present = originalRow, originalAcceptance, originalPresent
		}
		if r.wrapError != nil {
			err = r.wrapError(err)
		}
		return completion, err
	}
	return ports.TransactionCompletionCommitted, nil
}

type flowServerTestEngine struct {
	t                    *testing.T
	outcome              *contractmodels.FlowExecutionTypedOutcome
	calls                int
	validateCalls        int
	createPreviewCalls   int
	outputPreviewCalls   int
	createPreviewInputs  map[string]interface{}
	outputPreviewInputs  map[string]interface{}
	failFirst            bool
	ownerCancel          bool
	cancelCaller         context.CancelFunc
	wantCatalogAction    bool
	wantPreparedDefault  bool
	resourceBoundedCalls int
}

func (e *flowServerTestEngine) ValidateAtomicFlowForExecution(
	ctx context.Context,
	processType string,
	metadata *models.FlowMetadata,
) error {
	e.validateCalls++
	return automation.ValidateAtomicFlowForExecution(ctx, processType, metadata)
}

func (e *flowServerTestEngine) ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	flowName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	_ *models.UserSession,
	maxElementExecutions uint64,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	return e.executePreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx, flowName, versions, inputs, maxElementExecutions, false,
	)
}

func (e *flowServerTestEngine) ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	flowName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	_ *models.UserSession,
	maxElementExecutions uint64,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	return e.executePreparedFlowClosureWithinOwnedTransactionOutcome(
		ctx, flowName, versions, inputs, maxElementExecutions, true,
	)
}

func (e *flowServerTestEngine) StandardActionCreatePreviewSupported(
	context.Context,
	string,
	*models.UserSession,
) (bool, error) {
	return true, nil
}

func (e *flowServerTestEngine) PreparedStandardActionCreatePreviewSupported(
	ctx context.Context,
	flowName string,
	versions map[string]*models.FlowVersion,
	user *models.UserSession,
) (bool, error) {
	e.requirePreparedPreviewFlow(flowName, versions)
	return e.StandardActionCreatePreviewSupported(ctx, flowName, user)
}

func (e *flowServerTestEngine) PreviewStandardActionRecordCreateFields(
	ctx context.Context,
	flowName string,
	inputs map[string]interface{},
	_ *models.UserSession,
) (*automation.FlowRecordCreatePreviewResult, error) {
	e.t.Helper()
	if flowName != "AppOwnedAction" ||
		!constants.StandardActionExecutionOwnsRootFlow(ctx, flowName) {
		e.t.Fatalf("create preview authority = flow %q", flowName)
	}
	e.createPreviewCalls++
	e.createPreviewInputs = maps.Clone(inputs)
	return &automation.FlowRecordCreatePreviewResult{
		ActionName: flowName, IsSuccess: true,
	}, nil
}

func (e *flowServerTestEngine) PreviewPreparedStandardActionRecordCreateFields(
	ctx context.Context,
	flowName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
) (*automation.FlowRecordCreatePreviewResult, error) {
	e.requirePreparedPreviewFlow(flowName, versions)
	return e.PreviewStandardActionRecordCreateFields(ctx, flowName, inputs, user)
}

func (e *flowServerTestEngine) PreviewFlowOutputs(
	ctx context.Context,
	flowName string,
	inputs map[string]interface{},
	_ *models.UserSession,
) (*automation.FlowPreviewOutputsResult, error) {
	e.t.Helper()
	if flowName != "PreviewAction" ||
		!constants.StandardActionExecutionOwnsRootFlow(ctx, flowName) {
		e.t.Fatalf("output preview authority = flow %q", flowName)
	}
	e.outputPreviewCalls++
	e.outputPreviewInputs = maps.Clone(inputs)
	return &automation.FlowPreviewOutputsResult{
		IsSuccess: true, OutputValues: map[string]interface{}{"result": "ok"},
	}, nil
}

func (e *flowServerTestEngine) PreviewPreparedFlowOutputs(
	ctx context.Context,
	flowName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	user *models.UserSession,
) (*automation.FlowPreviewOutputsResult, error) {
	e.requirePreparedPreviewFlow(flowName, versions)
	return e.PreviewFlowOutputs(ctx, flowName, inputs, user)
}

func (e *flowServerTestEngine) requirePreparedPreviewFlow(
	flowName string,
	versions map[string]*models.FlowVersion,
) {
	e.t.Helper()
	version := versions[flowName]
	if version == nil || version.Id == "" {
		e.t.Fatalf("prepared preview closure[%q] = %#v", flowName, version)
	}
}

func (e *flowServerTestEngine) executePreparedFlowClosureWithinOwnedTransactionOutcome(
	ctx context.Context,
	flowName string,
	versions map[string]*models.FlowVersion,
	inputs map[string]interface{},
	maxElementExecutions uint64,
	resourceBounded bool,
) (*contractmodels.FlowExecutionTypedOutcome, error) {
	e.t.Helper()
	version := versions[flowName]
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); !active {
		e.t.Fatal("Flow engine escaped the transaction")
	}
	if e.wantCatalogAction &&
		(!constants.StandardActionExecutionOwnsRootFlow(ctx, flowName) ||
			!constants.IsScopedInternalAccess(ctx, "WorkItem", constants.PermRead)) {
		e.t.Fatal("Catalog action execution is missing its exact protected root or internal scope")
	}
	if e.wantPreparedDefault && inputs["defaulted"] != "prepared" {
		e.t.Fatalf("prepared Catalog input defaults = %#v", inputs)
	}
	if resourceBounded {
		e.resourceBoundedCalls++
	}
	wantElementExecutions := uint64(2)
	if resourceBounded {
		// Catalog actions can synchronously dispatch record-trigger roots that
		// share the transaction budget without joining the prepared child closure.
		wantElementExecutions = runtimeFlowMaxElementExecutions
	}
	if flowName != "AppOwnedAction" || version == nil || version.Id != "version-action" ||
		versions["ChildAction"] == nil || versions["ChildAction"].Id != "version-child" ||
		maxElementExecutions != wantElementExecutions {
		e.t.Fatalf("prepared Flow identity/bound = %q/%#v/%d", flowName, version, maxElementExecutions)
	}
	e.calls++
	if _, leaked := inputs["failedAttemptLeak"]; leaked {
		e.t.Fatal("rolled-back attempt input leaked into transaction retry")
	}
	if e.failFirst && e.calls == 1 {
		inputs["failedAttemptLeak"] = true
		return nil, errors.New("synthetic transaction retry")
	}
	if e.cancelCaller != nil {
		e.cancelCaller()
		return nil, ctx.Err()
	}
	if e.ownerCancel {
		return nil, context.Canceled
	}
	if e.outcome != nil {
		return e.outcome, nil
	}
	return &contractmodels.FlowExecutionTypedOutcome{
		ExecutionID: "execution-1",
		Result: &contractmodels.FlowExecutionResult{
			ActionName: flowName, IsSuccess: true,
			OutputValues: map[string]interface{}{"result": "ok"},
		},
	}, nil
}

type flowServerTestCatalog struct {
	view            catalogsemantic.AppMetadataViewV2
	viewV3          catalogsemantic.AppMetadataViewV3
	scopes          []CatalogConsumerScope
	v2Calls         int
	v3Calls         int
	requireDeadline bool
}

type flowServerV2OnlyTestCatalog struct {
	view  catalogsemantic.AppMetadataViewV2
	calls int
}

type flowServerTestPermissionEngine struct {
	result            *sharedcontract.EffectivePermissionsResult
	transactionResult *sharedcontract.EffectivePermissionsResult
	calls             int
	transactionCalls  int
	completeCalls     int
	completeTxCalls   int
	actionCalls       int
	actionTxCalls     int
	fenceCalls        int
	fenceErr          error
	fenceCheck        func() error
	objectDecision    func(context.Context, string, string, *models.UserSession) bool
}

func (*flowServerTestPermissionEngine) CheckFieldVisibilityWithUser(
	context.Context,
	string,
	string,
	*models.UserSession,
) (bool, error) {
	return true, nil
}

func (engine *flowServerTestPermissionEngine) CheckObjectPermissionWithUser(
	ctx context.Context,
	objectAPIName string,
	operation string,
	user *models.UserSession,
) bool {
	if engine.objectDecision != nil {
		return engine.objectDecision(ctx, objectAPIName, operation, user)
	}
	projected := engine.result
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); active &&
		engine.transactionResult != nil {
		projected = engine.transactionResult
	}
	if projected == nil {
		return false
	}
	for _, object := range projected.ObjectPermissions {
		if !strings.EqualFold(object.ResourceType, objectAPIName) {
			continue
		}
		switch operation {
		case constants.PermRead:
			return object.CanRead
		case constants.PermCreate:
			return object.CanCreate
		case constants.PermEdit:
			return object.CanEdit
		case constants.PermDelete:
			return object.CanDelete
		}
	}
	return false
}

func (engine *flowServerTestPermissionEngine) FencePermissionFloorWithinTransaction(
	ctx context.Context,
	_ permissionstatev1.Version,
) error {
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); !active {
		return errors.New("permission fence escaped the Flow transaction")
	}
	engine.fenceCalls++
	if engine.fenceCheck != nil {
		return engine.fenceCheck()
	}
	return engine.fenceErr
}

func (engine *flowServerTestPermissionEngine) EffectivePermissionsWithinTransaction(
	ctx context.Context,
	_ *models.UserSession,
	_ string,
	_ permissionstatev1.Version,
) (*sharedcontract.EffectivePermissionsResult, error) {
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); !active {
		return nil, errors.New("permission projection escaped the Flow transaction")
	}
	engine.transactionCalls++
	engine.completeTxCalls++
	if engine.transactionResult != nil {
		return engine.transactionResult, nil
	}
	return engine.result, nil
}

func (engine *flowServerTestPermissionEngine) EffectivePermissions(
	context.Context,
	*models.UserSession,
	string,
	permissionstatev1.Version,
) (*sharedcontract.EffectivePermissionsResult, error) {
	engine.calls++
	engine.completeCalls++
	return engine.result, nil
}

func (engine *flowServerTestPermissionEngine) EffectiveActionPermissions(
	context.Context,
	*models.UserSession,
	permissionstatev1.Version,
) (*sharedcontract.EffectiveActionPermissions, error) {
	engine.calls++
	engine.actionCalls++
	return actionPermissionsForFlowServerTest(engine.result), nil
}

func (engine *flowServerTestPermissionEngine) EffectiveActionPermissionsWithinTransaction(
	ctx context.Context,
	_ *models.UserSession,
	_ permissionstatev1.Version,
) (*sharedcontract.EffectiveActionPermissions, error) {
	if active, _ := ctx.Value(flowServerTransactionKey{}).(bool); !active {
		return nil, errors.New("action permission projection escaped the Flow transaction")
	}
	engine.transactionCalls++
	engine.actionTxCalls++
	projected := engine.transactionResult
	if projected == nil {
		projected = engine.result
	}
	return actionPermissionsForFlowServerTest(projected), nil
}

func actionPermissionsForFlowServerTest(
	projected *sharedcontract.EffectivePermissionsResult,
) *sharedcontract.EffectiveActionPermissions {
	if projected == nil {
		return nil
	}
	return &sharedcontract.EffectiveActionPermissions{
		SystemPermissions: projected.SystemPermissions,
		CustomPermissions: projected.CustomPermissions,
	}
}

func (c *flowServerTestCatalog) GetAppMetadataViewV2(
	ctx context.Context,
	scope CatalogConsumerScope,
	_ catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV2, error) {
	if c.requireDeadline {
		if _, ok := ctx.Deadline(); !ok {
			return catalogsemantic.AppMetadataViewV2{}, errors.New("Catalog V2 read has no deadline")
		}
	}
	c.v2Calls++
	c.scopes = append(c.scopes, scope)
	return c.view, nil
}

func (c *flowServerTestCatalog) GetAppMetadataViewV3(
	ctx context.Context,
	scope CatalogConsumerScope,
	_ catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV3, error) {
	if c.requireDeadline {
		if _, ok := ctx.Deadline(); !ok {
			return catalogsemantic.AppMetadataViewV3{}, errors.New("Catalog V3 read has no deadline")
		}
	}
	c.v3Calls++
	c.scopes = append(c.scopes, scope)
	return c.viewV3, nil
}

func (c *flowServerV2OnlyTestCatalog) GetAppMetadataViewV2(
	context.Context,
	CatalogConsumerScope,
	catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV2, error) {
	c.calls++
	return c.view, nil
}

func TestFlowExecutionServerCommitsReplaysAndBindsBothImmutableAuthorities(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t}
	view := flowServerTestView(t)
	appView := flowServerTestAppView(view)
	catalog := &flowServerTestCatalog{view: appView}
	server, err := newFlowServerTest(t, engine, store, runner, catalog, nil)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "user-1", view)
	request := flowServerTestRequest(t, ctx, "request-1", "1")

	first, err := server.ExecuteFlow(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	second, err := server.ExecuteFlow(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	firstResult, err := mapping.FromWireExecuteFlowResponse(first)
	if err != nil {
		t.Fatal(err)
	}
	secondResult, err := mapping.FromWireExecuteFlowResponse(second)
	if err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 || engine.validateCalls != 4 || runner.attempts != 2 ||
		firstResult.ExecutionID != "execution-1" ||
		string(firstResult.CanonicalOutputs) != `{"result":"ok"}` ||
		string(secondResult.CanonicalOutputs) != string(firstResult.CanonicalOutputs) {
		t.Fatalf("Flow execution/replay = calls %d validations %d attempts %d first %#v second %#v",
			engine.calls, engine.validateCalls, runner.attempts, firstResult, secondResult)
	}
	if !strings.HasPrefix(store.acceptance.ClosureDigest, "sha256:") ||
		store.acceptance.DefinitionVersion != "1" ||
		store.row.State != persistence.RuntimeFlowInvocationCompleted ||
		len(catalog.scopes) != 2 ||
		!slicesEqual(catalog.scopes[0].Capabilities, []string{
			catalogsemantic.AppMetadataV2CapabilityName,
		}) {
		t.Fatalf("durable invocation/Catalog scope = %#v / %#v / %#v", store.acceptance, store.row, catalog.scopes)
	}
	rootComponent, err := selectRequestedFlowAppMetadataComponent(
		appView.AppComponents, "AppOwnedAction", "1",
	)
	if err != nil {
		t.Fatal(err)
	}
	legacyRequestPayload, err := json.Marshal(struct {
		DefinitionRef        string          `json:"definitionRef"`
		DefinitionVersion    string          `json:"definitionVersion"`
		ComponentID          string          `json:"componentId"`
		ComponentDigest      string          `json:"componentDigest"`
		AppMetadataDigest    string          `json:"appMetadataDigest"`
		RuntimeClosureDigest string          `json:"runtimeClosureDigest"`
		CanonicalInputs      json.RawMessage `json:"canonicalInputs"`
	}{
		DefinitionRef: "AppOwnedAction", DefinitionVersion: "1",
		ComponentID: rootComponent.ComponentID, ComponentDigest: rootComponent.ContentDigest,
		AppMetadataDigest:    appView.Lineage.ViewDigest,
		RuntimeClosureDigest: store.acceptance.ClosureDigest,
		CanonicalInputs:      []byte(`{"input":"value"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	legacySum := sha256.Sum256(legacyRequestPayload)
	legacyDigest := "sha256:" + hex.EncodeToString(legacySum[:])
	if store.acceptance.RequestDigest != legacyDigest {
		t.Fatalf(
			"ordinary request digest = %q, want pre-Catalog-action digest %q",
			store.acceptance.RequestDigest, legacyDigest,
		)
	}
}

func TestFlowExecutionServerRollsBackAndPresentsApprovedEngineErrors(t *testing.T) {
	view := flowServerTestView(t)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, outcome: &contractmodels.FlowExecutionTypedOutcome{
		ExecutionID: "execution-validation",
		Result: &contractmodels.FlowExecutionResult{
			ActionName: "AppOwnedAction",
			IsSuccess:  false,
			Errors: []contractmodels.FlowExecutionError{
				{
					StatusCode: "POLICY_VALIDATION_ERROR",
					Message:    "Correct RequestedValue before retrying.",
					Fields:     []string{"RequestedValue"},
				},
				{
					StatusCode: "RELATED_VALUE_REJECTED",
					Message:    "Correct RelatedValue before retrying.",
					Fields:     []string{"RelatedValue"},
				},
			},
			IsCustomError: true,
		},
		Cause: &contractmodels.FlowExecutionCause{
			Code:     "VALIDATION",
			Category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
	}}
	server, err := newFlowServerTest(
		t, engine, store, runner,
		&flowServerTestCatalog{view: flowServerTestAppView(view)},
		&flowServerTestPermissionEngine{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "validation", view)
	wireResponse, err := server.ExecuteFlow(
		ctx,
		flowServerTestRequest(t, ctx, "validation-request", "1"),
	)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireExecuteFlowResponseV2(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if response.ExecutionID != "execution-validation" || len(response.Errors) != 2 ||
		response.Errors[0].Code != "POLICY_VALIDATION_ERROR" ||
		response.Errors[0].Message != "Correct RequestedValue before retrying." ||
		!slices.Equal(response.Errors[0].Fields, []string{"RequestedValue"}) ||
		response.Errors[1].Code != "RELATED_VALUE_REJECTED" ||
		response.Errors[1].Message != "Correct RelatedValue before retrying." ||
		!slices.Equal(response.Errors[1].Fields, []string{"RelatedValue"}) {
		t.Fatalf("presented Flow response = %#v", response)
	}
	if store.present || runner.attempts != 1 || engine.calls != 1 {
		t.Fatalf(
			"failed Flow transaction = row present %t, attempts %d, engine calls %d",
			store.present, runner.attempts, engine.calls,
		)
	}
}

func TestFlowExecutionServerCatalogActionDoesNotPresentErrorsAfterPermissionRevocation(t *testing.T) {
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	permissions := &flowServerTestPermissionEngine{
		result: &sharedcontract.EffectivePermissionsResult{
			SystemPermissions: map[string]bool{"RunBoundedAction": true},
			ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
				ResourceType: "WorkItem", CanRead: true,
			}},
		},
		fenceErr: &permissionsemantic.Failure{
			Category: permissionsemantic.FailureUnavailable,
		},
	}
	engine := &flowServerTestEngine{
		t: t, wantCatalogAction: true,
		outcome: &contractmodels.FlowExecutionTypedOutcome{
			ExecutionID: "execution-revoked",
			Result: &contractmodels.FlowExecutionResult{
				ActionName: "AppOwnedAction",
				IsSuccess:  false,
				Errors: []contractmodels.FlowExecutionError{{
					StatusCode: "POLICY_VALIDATION_ERROR",
					Message:    "This message must not escape after permission revocation.",
					Fields:     []string{"RequestedValue"},
				}},
				IsCustomError: true,
			},
			Cause: &contractmodels.FlowExecutionCause{
				Code:     "VALIDATION",
				Category: contractmodels.FlowExecutionCauseCategoryValidation,
			},
		},
	}
	server, err := newFlowServerTest(
		t, engine, store, runner,
		&flowServerTestCatalog{view: appView}, permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "revoked-failure", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "revoked-failure", component.ComponentID, "runBounded",
	)
	response, executeErr := server.ExecuteFlow(ctx, request)
	if status.Code(executeErr) != codes.Unavailable || response != nil || store.present ||
		engine.calls != 1 || engine.resourceBoundedCalls != 1 || permissions.fenceCalls != 1 ||
		strings.Contains(executeErr.Error(), "must not escape") {
		t.Fatalf(
			"revoked failed action response=%#v error=%v stored=%v engine=%d/%d fence=%d",
			response, executeErr, store.present, engine.calls, engine.resourceBoundedCalls,
			permissions.fenceCalls,
		)
	}
}

func TestFlowExecutionServerDoesNotPresentApprovedEngineErrorsWhenRollbackFails(t *testing.T) {
	view := flowServerTestView(t)
	store := &flowServerTestStore{t: t}
	rollbackError := ports.TransactionCompletionRollbackError
	runner := &flowServerTestRunner{
		store: store, completionOverride: &rollbackError,
		wrapError: func(err error) error {
			return apperrors.NewInternalError("rollback failed", err)
		},
		preserveEffectsOnErr: true,
	}
	engine := &flowServerTestEngine{t: t, outcome: &contractmodels.FlowExecutionTypedOutcome{
		ExecutionID: "execution-rollback-failed",
		Result: &contractmodels.FlowExecutionResult{
			ActionName: "AppOwnedAction",
			IsSuccess:  false,
			Errors: []contractmodels.FlowExecutionError{{
				StatusCode: "POLICY_VALIDATION_ERROR",
				Message:    "This message must not escape an uncertain rollback.",
				Fields:     []string{"RequestedValue"},
			}},
			IsCustomError: true,
		},
		Cause: &contractmodels.FlowExecutionCause{
			Code:     "VALIDATION",
			Category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
	}}
	server, err := newFlowServerTest(
		t, engine, store, runner,
		&flowServerTestCatalog{view: flowServerTestAppView(view)},
		&flowServerTestPermissionEngine{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "rollback-failed", view)
	response, executeErr := server.ExecuteFlow(
		ctx,
		flowServerTestRequest(t, ctx, "rollback-failed-request", "1"),
	)
	if status.Code(executeErr) != codes.Internal || response != nil || !store.present ||
		strings.Contains(executeErr.Error(), "must not escape") {
		t.Fatalf(
			"rollback failure response=%#v error=%v stored=%v, want opaque Internal with uncertain effects",
			response, executeErr, store.present,
		)
	}
}

func TestFlowPresentationAfterRollbackRejectsUncertainOutcomes(t *testing.T) {
	presentation := &flowPresentationRollback{
		response: &flowsemantic.ResponseV2{ExecutionID: "execution-1"},
	}
	for _, test := range []struct {
		name       string
		completion ports.TransactionCompletionOutcome
		err        error
		want       bool
	}{
		{
			name: "direct presentation", completion: ports.TransactionCompletionRolledBack,
			err: presentation, want: true,
		},
		{
			name: "transparent wrapper", completion: ports.TransactionCompletionRolledBack,
			err: fmt.Errorf("transaction: %w", presentation), want: true,
		},
		{
			name: "internal rollback failure", completion: ports.TransactionCompletionRollbackError,
			err: apperrors.NewInternalError("rollback failed", presentation),
		},
		{
			name: "joined rollback outcome", completion: ports.TransactionCompletionRolledBack,
			err: errors.Join(presentation, errors.New("rollback outcome unknown")),
		},
		{name: "committed", completion: ports.TransactionCompletionCommitted, err: presentation},
		{name: "ambient transaction without terminal outcome", err: presentation},
	} {
		t.Run(test.name, func(t *testing.T) {
			got, ok := flowPresentationAfterRollback(test.completion, test.err)
			if ok != test.want || (ok && got != presentation) {
				t.Fatalf("presentation=%#v mapped=%v, want presentation=%#v mapped=%v", got, ok, presentation, test.want)
			}
		})
	}
}

func TestFlowExecutionServerDoesNotPresentUnapprovedEngineErrors(t *testing.T) {
	view := flowServerTestView(t)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, outcome: &contractmodels.FlowExecutionTypedOutcome{
		ExecutionID: "execution-internal",
		Result: &contractmodels.FlowExecutionResult{
			ActionName: "AppOwnedAction",
			IsSuccess:  false,
			Errors: []contractmodels.FlowExecutionError{{
				StatusCode: "FLOW_ERROR",
				Message:    "sql: failed to connect with password secret-value",
			}},
			IsCustomError: false,
		},
		Cause: &contractmodels.FlowExecutionCause{
			Code:     "PERSISTENCE_FAILURE",
			Category: contractmodels.FlowExecutionCauseCategoryInternal,
		},
	}}
	server, err := newFlowServerTest(
		t, engine, store, runner,
		&flowServerTestCatalog{view: flowServerTestAppView(view)},
		&flowServerTestPermissionEngine{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "internal", view)
	response, err := server.ExecuteFlow(
		ctx,
		flowServerTestRequest(t, ctx, "internal-request", "1"),
	)
	if status.Code(err) != codes.Internal || response != nil || store.present ||
		runner.attempts != 1 || engine.calls != 1 {
		t.Fatalf(
			"unapproved Flow response = %#v, error = %v, row present = %t, attempts = %d, engine calls = %d",
			response, err, store.present, runner.attempts, engine.calls,
		)
	}
	if strings.Contains(err.Error(), "secret-value") {
		t.Fatalf("unapproved Engine detail escaped Runtime boundary: %v", err)
	}
}

func TestFlowExecutionServerRejectsUnsafeApprovedEngineErrors(t *testing.T) {
	view := flowServerTestView(t)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, outcome: &contractmodels.FlowExecutionTypedOutcome{
		ExecutionID: "execution-unsafe",
		Result: &contractmodels.FlowExecutionResult{
			ActionName: "AppOwnedAction",
			IsSuccess:  false,
			Errors: []contractmodels.FlowExecutionError{{
				StatusCode: "POLICY_VALIDATION_ERROR",
				Message:    "Correct RequestedValue.\ninternal secret-value",
				Fields:     []string{"RequestedValue"},
			}},
			IsCustomError: true,
		},
		Cause: &contractmodels.FlowExecutionCause{
			Code:     "VALIDATION",
			Category: contractmodels.FlowExecutionCauseCategoryValidation,
		},
	}}
	server, err := newFlowServerTest(
		t, engine, store, runner,
		&flowServerTestCatalog{view: flowServerTestAppView(view)},
		&flowServerTestPermissionEngine{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "unsafe", view)
	response, err := server.ExecuteFlow(
		ctx,
		flowServerTestRequest(t, ctx, "unsafe-request", "1"),
	)
	if status.Code(err) != codes.Internal || response != nil || store.present ||
		runner.attempts != 1 || engine.calls != 1 {
		t.Fatalf(
			"unsafe approved Flow response = %#v, error = %v, row present = %t, attempts = %d, engine calls = %d",
			response, err, store.present, runner.attempts, engine.calls,
		)
	}
	if strings.Contains(err.Error(), "secret-value") {
		t.Fatalf("unsafe approved Engine detail escaped Runtime boundary: %v", err)
	}
}

func TestFlowExecutionServerUsesPreparedAppMetadataWithoutCatalogRead(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t}
	view := flowServerTestView(t)
	prepared := normalizeFlowAppMetadataV2(flowServerTestAppView(view))
	catalog := &flowServerTestCatalog{view: flowServerTestAppView(view)}
	server, err := newFlowServerTest(t, engine, store, runner, catalog, nil)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "user-1", view)
	ctx = context.WithValue(ctx, runtimeFlowAppMetadataContextKey{}, prepared)
	if _, err := server.ExecuteFlow(ctx, flowServerTestRequest(t, ctx, "prepared-request", "1")); err != nil {
		t.Fatal(err)
	}
	if catalog.v2Calls != 0 || catalog.v3Calls != 0 {
		t.Fatalf(
			"prepared ExecuteFlow performed Catalog reads: v2=%d v3=%d",
			catalog.v2Calls, catalog.v3Calls,
		)
	}
}

func TestFlowExecutionServerUsesAdmittedAppMetadataV3Authority(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t}
	view := flowServerTestView(t)
	catalog := &flowServerTestCatalog{viewV3: flowServerTestAppViewV3(view)}
	server, err := NewFlowExecutionServer(
		engine, store, runner, catalog, nil,
		flowServerTestCatalogAdmission(t, catalogsemantic.AppMetadataV3CapabilityName),
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContextWithAppMetadataCapability(
		t, context.Background(), "v3-user", view,
		"",
	)
	if _, err := server.ExecuteFlow(
		ctx, flowServerTestRequest(t, ctx, "v3-request", "1"),
	); err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 || catalog.v3Calls != 1 || catalog.v2Calls != 0 ||
		len(catalog.scopes) != 1 ||
		!slicesEqual(catalog.scopes[0].Capabilities, []string{
			catalogsemantic.AppMetadataV3CapabilityName,
		}) {
		t.Fatalf(
			"V3 authority use = engine %d v3 %d v2 %d scopes %#v",
			engine.calls, catalog.v3Calls, catalog.v2Calls, catalog.scopes,
		)
	}
}

func TestFlowExecutionServerDoesNotFallBackFromAdmittedV3ToV2(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t}
	view := flowServerTestView(t)
	catalog := &flowServerV2OnlyTestCatalog{view: flowServerTestAppView(view)}
	server, err := NewFlowExecutionServer(
		engine, store, runner, catalog, nil,
		flowServerTestCatalogAdmission(t, catalogsemantic.AppMetadataV3CapabilityName),
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContextWithAppMetadataCapability(
		t, context.Background(), "v3-no-engine", view,
		"",
	)
	_, err = server.ExecuteFlow(
		ctx, flowServerTestRequest(t, ctx, "v3-no-engine", "1"),
	)
	if status.Code(err) != codes.Unavailable || catalog.calls != 0 ||
		engine.calls != 0 || runner.attempts != 0 {
		t.Fatalf(
			"V3 fail-closed = %v v2 calls %d engine %d tx %d",
			err, catalog.calls, engine.calls, runner.attempts,
		)
	}
}

func TestFlowExecutionServerCatalogActionAuthorizesExecutesAndReplays(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, wantCatalogAction: true, wantPreparedDefault: true}
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	permissions := &flowServerTestPermissionEngine{result: &sharedcontract.EffectivePermissionsResult{
		SystemPermissions: map[string]bool{"RunBoundedAction": true},
		ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
			ResourceType: "WorkItem", CanRead: true,
		}},
	}}
	server, err := newFlowServerTest(t,
		engine, store, runner, &flowServerTestCatalog{view: appView}, permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "catalog-action", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "catalog-action-key", component.ComponentID, "runBounded",
	)
	first, err := server.ExecuteFlow(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	second, err := server.ExecuteFlow(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	firstResult, err := mapping.FromWireExecuteFlowResponse(first)
	if err != nil {
		t.Fatal(err)
	}
	secondResult, err := mapping.FromWireExecuteFlowResponse(second)
	if err != nil {
		t.Fatal(err)
	}
	if engine.calls != 1 || engine.resourceBoundedCalls != 1 ||
		permissions.calls != 2 || permissions.transactionCalls != 0 ||
		permissions.actionCalls != 2 || permissions.actionTxCalls != 0 ||
		permissions.completeCalls != 0 || permissions.completeTxCalls != 0 ||
		permissions.fenceCalls != 2 || runner.attempts != 2 ||
		firstResult.ExecutionID != secondResult.ExecutionID ||
		store.row.State != persistence.RuntimeFlowInvocationCompleted {
		t.Fatalf(
			"Catalog action execution = engine %d bounded %d permission %d/%d action %d/%d complete %d/%d fence %d attempts %d first %#v second %#v row %#v",
			engine.calls, engine.resourceBoundedCalls, permissions.calls,
			permissions.transactionCalls, permissions.actionCalls,
			permissions.actionTxCalls, permissions.completeCalls,
			permissions.completeTxCalls, permissions.fenceCalls, runner.attempts,
			firstResult, secondResult, store.row,
		)
	}

	ordinary := flowServerTestRequest(t, ctx, "catalog-action-key", "1")
	if _, err := server.ExecuteFlow(ctx, ordinary); status.Code(err) != codes.InvalidArgument {
		t.Fatalf("same idempotency key without Catalog action = %v, want conflict", err)
	}
}

func TestFlowExecutionServerCatalogActionUsesRevisionBoundEffectiveFlowVersion(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, wantCatalogAction: true}
	view := flowServerTestViewWithRootVersion(t, 2)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	root, ok := view.ActiveFlowVersion("AppOwnedAction")
	if !ok {
		t.Fatal("missing effective root Flow")
	}
	envelope, err := catalogsemantic.EncodeEffectiveStandardFlowDocument(
		"2", root.MetadataRaw,
	)
	if err != nil {
		t.Fatal(err)
	}
	appView.EffectiveComponents = []catalogsemantic.EffectiveMetadataComponent{{
		ComponentKind:          catalogsemantic.EffectiveStandardFlowComponentKind,
		ComponentFormatVersion: 1,
		BusinessKey:            "AppOwnedAction",
		ContentDigest: catalogsemantic.ComputeEffectiveMetadataComponentDigest(
			envelope,
		),
		CanonicalBehavior: envelope,
	}}
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
	permissions := &flowServerTestPermissionEngine{result: &sharedcontract.EffectivePermissionsResult{
		SystemPermissions: map[string]bool{"RunBoundedAction": true},
		ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
			ResourceType: "WorkItem", CanRead: true,
		}},
	}}
	server, err := newFlowServerTest(t,
		engine, store, runner, &flowServerTestCatalog{view: appView}, permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "catalog-action-effective", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "catalog-action-effective-key", component.ComponentID, "runBounded",
	)
	if _, err := server.ExecuteFlow(ctx, request); err != nil {
		t.Fatal(err)
	}
	if store.acceptance.DefinitionVersion != "2" {
		t.Fatalf("accepted definition version = %q, want effective version 2", store.acceptance.DefinitionVersion)
	}
}

func TestFlowExecutionServerEffectiveCatalogActionUsesRevisionBoundRuntimeFlowWithoutL1FlowComponent(
	t *testing.T,
) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, wantCatalogAction: true}
	runtimeView := flowServerTestView(t)
	appV2, component := flowServerTestAppViewWithCatalogAction(t, runtimeView, true)
	appV3 := catalogsemantic.AppMetadataViewV3{
		OrgID: appV2.OrgID, AppInstallationID: appV2.AppInstallationID,
		AppID: appV2.AppID, AppReleaseDigest: appV2.AppReleaseDigest,
		Lineage: appV2.Lineage, Receipts: appV2.Receipts,
	}
	for _, candidate := range appV2.AppComponents {
		if candidate.ComponentKind == appMetadataStandardFlowKind &&
			candidate.BusinessKey == "AppOwnedAction" {
			continue
		}
		appV3.AppComponents = append(appV3.AppComponents, candidate)
	}
	effectiveBehavior := []byte(`[{"name":"runBounded","flowDeveloperName":"AppOwnedAction","requiredPermissions":["RunBoundedAction"],"isAtomic":true,"inputs":[{"apiName":"input","dataType":"string","isRequired":true}],"inputGuards":[{"condition":"true","requiredObject":"WorkItem","requiredOperation":"Read"}],"internalAccessScopes":[{"objectApiName":"WorkItem","operation":"Read"}]}]`)
	appV3.EffectiveComponents = []catalogsemantic.EffectiveMetadataComponent{{
		ComponentKind:          catalogsemantic.EffectiveStandardActionComponentKind,
		ComponentFormatVersion: 1, BusinessKey: component.BusinessKey,
		ContentDigest:     catalogsemantic.ComputeEffectiveMetadataComponentDigest(effectiveBehavior),
		CanonicalBehavior: effectiveBehavior,
	}}
	appV3.Receipts.CapabilityVersion = catalogsemantic.AppMetadataV3CapabilityName
	appV3.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(appV3)
	permissions := &flowServerTestPermissionEngine{result: &sharedcontract.EffectivePermissionsResult{
		SystemPermissions: map[string]bool{"RunBoundedAction": true},
		ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
			ResourceType: "WorkItem", CanRead: true,
		}},
	}}
	catalog := &flowServerTestCatalog{viewV3: appV3}
	server, err := NewFlowExecutionServer(
		engine, store, runner, catalog, permissions,
		flowServerTestCatalogAdmission(t, catalogsemantic.AppMetadataV3CapabilityName),
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContextWithAppMetadataCapability(
		t, context.Background(), "effective-catalog-action", runtimeView,
		catalogsemantic.AppMetadataV3CapabilityName,
	)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "effective-catalog-action-key", component.ComponentID, "runBounded",
	)
	if _, err := server.ExecuteFlow(ctx, request); err != nil {
		t.Fatal(err)
	}
	if store.acceptance.DefinitionVersion != "1" || engine.calls != 1 {
		t.Fatalf("effective action execution = acceptance %#v engine calls %d", store.acceptance, engine.calls)
	}
}

func TestFlowExecutionServerPreviewsCatalogActionWithoutDurableOwnership(t *testing.T) {
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t}
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithPreviewCatalogAction(t, view)
	catalog := &flowServerTestCatalog{view: appView}
	permissions := &flowServerTestPermissionEngine{result: &sharedcontract.EffectivePermissionsResult{
		SystemPermissions: map[string]bool{"RunBoundedAction": true},
		ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
			ResourceType: "WorkItem", CanRead: true,
		}},
	}}
	server, err := newFlowServerTest(t, engine, store, runner, catalog, permissions)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "preview-action", view)

	createWire, err := server.PreviewCatalogAction(
		ctx,
		flowServerTestCatalogActionPreviewRequest(
			t, ctx, flowsemantic.CatalogActionPreviewCreateRecords, component.ComponentID,
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	createRequest, err := mapping.FromWirePreviewCatalogActionRequest(ctx, flowServerTestCatalogActionPreviewRequest(
		t, ctx, flowsemantic.CatalogActionPreviewCreateRecords, component.ComponentID,
	))
	if err != nil {
		t.Fatal(err)
	}
	create, err := mapping.FromWirePreviewCatalogActionResponse(createRequest, createWire)
	if err != nil {
		t.Fatal(err)
	}
	outputWire, err := server.PreviewCatalogAction(
		ctx,
		flowServerTestCatalogActionPreviewRequest(
			t, ctx, flowsemantic.CatalogActionPreviewOutputs, component.ComponentID,
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	outputRequest, err := mapping.FromWirePreviewCatalogActionRequest(ctx, flowServerTestCatalogActionPreviewRequest(
		t, ctx, flowsemantic.CatalogActionPreviewOutputs, component.ComponentID,
	))
	if err != nil {
		t.Fatal(err)
	}
	output, err := mapping.FromWirePreviewCatalogActionResponse(outputRequest, outputWire)
	if err != nil {
		t.Fatal(err)
	}

	if string(create.CanonicalResult) != `{"actionName":"runBounded","isSuccess":true}` ||
		string(output.CanonicalResult) != `{"isSuccess":true,"outputValues":{"result":"ok"}}` {
		t.Fatalf(
			"canonical previews = create %s output %s",
			create.CanonicalResult, output.CanonicalResult,
		)
	}
	if engine.createPreviewCalls != 1 || engine.outputPreviewCalls != 1 ||
		engine.createPreviewInputs["recordId"] != "a99000000000001" ||
		engine.outputPreviewInputs["recordId"] != "a99000000000001" {
		t.Fatalf(
			"preview engine calls = create %d/%#v output %d/%#v",
			engine.createPreviewCalls, engine.createPreviewInputs,
			engine.outputPreviewCalls, engine.outputPreviewInputs,
		)
	}
	if runner.attempts != 0 || store.present || permissions.calls != 2 ||
		permissions.actionCalls != 2 || permissions.completeCalls != 0 ||
		len(catalog.scopes) != 2 ||
		create.Receipts.CatalogRevision != 17 || output.Receipts.CatalogRevision != 17 {
		t.Fatalf(
			"preview ownership/receipts = tx %d stored %v permission %d action %d complete %d scopes %d create %#v output %#v",
			runner.attempts, store.present, permissions.calls,
			permissions.actionCalls, permissions.completeCalls, len(catalog.scopes),
			create.Receipts, output.Receipts,
		)
	}
}

func TestFlowExecutionServerPreviewFailsClosedBeforeNoEffectEngine(t *testing.T) {
	tests := []struct {
		name       string
		mutate     func(*catalogsemantic.AppMetadataViewV2)
		wantCode   codes.Code
		cancelCtx  bool
		denyAction bool
	}{
		{name: "action permission denial is concealed", wantCode: codes.NotFound, denyAction: true},
		{
			name: "missing exact preview Flow authority",
			mutate: func(view *catalogsemantic.AppMetadataViewV2) {
				view.AppComponents = slices.DeleteFunc(
					view.AppComponents,
					func(component catalogsemantic.AppMetadataComponent) bool {
						return component.ComponentKind == appMetadataStandardFlowKind &&
							component.BusinessKey == "PreviewAction"
					},
				)
				view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(*view)
			},
			wantCode: codes.NotFound,
		},
		{
			name: "mixed Catalog revision",
			mutate: func(view *catalogsemantic.AppMetadataViewV2) {
				view.Lineage.CatalogRevision++
				view.Receipts.CatalogRevision++
				view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(*view)
			},
			wantCode: codes.Internal,
		},
		{name: "caller cancellation", wantCode: codes.Canceled, cancelCtx: true},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			store := &flowServerTestStore{t: t}
			runner := &flowServerTestRunner{store: store}
			engine := &flowServerTestEngine{t: t}
			view := flowServerTestView(t)
			appView, component := flowServerTestAppViewWithPreviewCatalogAction(t, view)
			if test.mutate != nil {
				test.mutate(&appView)
			}
			permissions := &flowServerTestPermissionEngine{result: &sharedcontract.EffectivePermissionsResult{
				SystemPermissions: map[string]bool{"RunBoundedAction": !test.denyAction},
				ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
					ResourceType: "WorkItem", CanRead: true,
				}},
			}}
			server, err := newFlowServerTest(t,
				engine, store, runner, &flowServerTestCatalog{view: appView}, permissions,
			)
			if err != nil {
				t.Fatal(err)
			}
			ctx := flowServerTestContext(t, context.Background(), test.name, view)
			request := flowServerTestCatalogActionPreviewRequest(
				t, ctx, flowsemantic.CatalogActionPreviewOutputs, component.ComponentID,
			)
			if test.cancelCtx {
				canceled, cancel := context.WithCancel(ctx)
				cancel()
				ctx = canceled
			}
			_, err = server.PreviewCatalogAction(ctx, request)
			if status.Code(err) != test.wantCode || engine.outputPreviewCalls != 0 ||
				runner.attempts != 0 || store.present {
				t.Fatalf(
					"fail closed = %v preview=%d tx=%d stored=%v",
					err, engine.outputPreviewCalls, runner.attempts, store.present,
				)
			}
		})
	}
}

func TestFlowExecutionServerCatalogActionFailsBeforeDurableAcceptance(t *testing.T) {
	tests := []struct {
		name             string
		atomic           bool
		permission       bool
		objectRead       bool
		definitionRef    string
		actionRef        string
		missingComponent bool
		wantCode         codes.Code
	}{
		{name: "action permission denied is concealed", atomic: true, objectRead: true, wantCode: codes.NotFound},
		{name: "input guard denied stays forbidden", atomic: true, permission: true, wantCode: codes.PermissionDenied},
		{name: "non atomic", permission: true, objectRead: true, wantCode: codes.InvalidArgument},
		{
			name: "wrong Flow binding", atomic: true, permission: true, objectRead: true,
			definitionRef: "OtherFlow", wantCode: codes.NotFound,
		},
		{
			name: "missing action", atomic: true, permission: true, objectRead: true,
			actionRef: "missing", wantCode: codes.NotFound,
		},
		{
			name: "missing component", atomic: true, permission: true, objectRead: true,
			missingComponent: true, wantCode: codes.NotFound,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			view := flowServerTestView(t)
			appView, component := flowServerTestAppViewWithCatalogAction(t, view, test.atomic)
			if test.definitionRef != "" {
				componentIndex := slices.IndexFunc(
					appView.AppComponents,
					func(candidate catalogsemantic.AppMetadataComponent) bool {
						return candidate.ComponentID == component.ComponentID
					},
				)
				behavior := []byte(strings.ReplaceAll(
					string(component.CanonicalBehavior), "AppOwnedAction", test.definitionRef,
				))
				component.CanonicalBehavior = behavior
				component.ContentDigest, _ = appauthority.ComponentDigest(
					component.ComponentKind, component.ComponentFormatVersion, behavior,
				)
				appView.AppComponents[componentIndex] = component
				appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
			}
			store := &flowServerTestStore{t: t}
			runner := &flowServerTestRunner{store: store}
			permissions := &flowServerTestPermissionEngine{result: &sharedcontract.EffectivePermissionsResult{
				SystemPermissions: map[string]bool{"RunBoundedAction": test.permission},
				ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
					ResourceType: "WorkItem", CanRead: test.objectRead,
				}},
			}}
			server, err := newFlowServerTest(t,
				&flowServerTestEngine{t: t}, store, runner,
				&flowServerTestCatalog{view: appView}, permissions,
			)
			if err != nil {
				t.Fatal(err)
			}
			ctx := flowServerTestContext(t, context.Background(), test.name, view)
			actionRef := test.actionRef
			if actionRef == "" {
				actionRef = "runBounded"
			}
			componentRef := component.ComponentID
			if test.missingComponent {
				componentRef = "sha256:missing-component"
			}
			request := flowServerTestCatalogActionRequest(
				t, ctx, test.name, componentRef, actionRef,
			)
			_, err = server.ExecuteFlow(ctx, request)
			if status.Code(err) != test.wantCode || runner.attempts != 0 || store.present {
				t.Fatalf("fail-closed result = %v attempts=%d present=%v", err, runner.attempts, store.present)
			}
		})
	}
}

func TestFlowExecutionServerCatalogActionRequiresOptionalPermissionProfile(t *testing.T) {
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	server, err := newFlowServerTest(t,
		&flowServerTestEngine{t: t}, store, runner,
		&flowServerTestCatalog{view: appView}, nil,
	)
	if err != nil {
		t.Fatalf("ordinary Flow composition rejected optional Catalog-action profile: %v", err)
	}
	ctx := flowServerTestContext(t, context.Background(), "missing-action-profile", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "missing-action-profile", component.ComponentID, "runBounded",
	)
	_, err = server.ExecuteFlow(ctx, request)
	if status.Code(err) != codes.Unavailable || runner.attempts != 0 || store.present {
		t.Fatalf(
			"missing Catalog-action permission profile = %v attempts=%d present=%v",
			err, runner.attempts, store.present,
		)
	}
}

func TestFlowExecutionServerCatalogActionReusesPreparedPlanWithLatePermissionFloorFence(t *testing.T) {
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	engine := &flowServerTestEngine{t: t, wantCatalogAction: true}
	permissions := &flowServerTestPermissionEngine{
		result: &sharedcontract.EffectivePermissionsResult{
			SystemPermissions: map[string]bool{"RunBoundedAction": true},
			ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
				ResourceType: "WorkItem", CanRead: true,
			}},
		},
		transactionResult: &sharedcontract.EffectivePermissionsResult{
			SystemPermissions: map[string]bool{"RunBoundedAction": false},
			ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
				ResourceType: "WorkItem", CanRead: true,
			}},
		},
	}
	permissions.fenceCheck = func() error {
		if engine.calls != 1 || !store.present || store.row.State != persistence.RuntimeFlowInvocationAccepted {
			return fmt.Errorf(
				"permission fence ran before effect: engine=%d present=%t state=%s",
				engine.calls, store.present, store.row.State,
			)
		}
		return nil
	}
	server, err := newFlowServerTest(t,
		engine, store, runner, &flowServerTestCatalog{view: appView}, permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "revoked-action", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "revoked-action", component.ComponentID, "runBounded",
	)
	_, err = server.ExecuteFlow(ctx, request)
	if err != nil || runner.attempts != 1 || !store.present ||
		engine.calls != 1 || permissions.calls != 1 || permissions.transactionCalls != 0 ||
		permissions.fenceCalls != 1 {
		t.Fatalf(
			"revoked action = %v attempts=%d present=%v engine=%d permission=%d/%d",
			err, runner.attempts, store.present, engine.calls,
			permissions.calls, permissions.transactionCalls,
		)
	}
}

func TestFlowExecutionServerCatalogActionDoesNotRepeatPreparedObjectGuard(t *testing.T) {
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	objectCalls := 0
	permissions := &flowServerTestPermissionEngine{
		result: &sharedcontract.EffectivePermissionsResult{
			SystemPermissions: map[string]bool{"RunBoundedAction": true},
			// This raw bit deliberately disagrees with the authoritative
			// decision below, as happens for RequiredReadPermission/policy gates.
			ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
				ResourceType: "WorkItem", CanRead: true,
			}},
		},
		objectDecision: func(
			ctx context.Context,
			objectAPIName string,
			operation string,
			user *models.UserSession,
		) bool {
			objectCalls++
			if objectAPIName != "WorkItem" || operation != constants.PermRead || user.ID == "" {
				t.Fatalf("object decision authority = %q/%q/%#v", objectAPIName, operation, user)
			}
			active, _ := ctx.Value(flowServerTransactionKey{}).(bool)
			return !active
		},
	}
	engine := &flowServerTestEngine{t: t, wantCatalogAction: true}
	server, err := newFlowServerTest(t,
		engine, store, runner, &flowServerTestCatalog{view: appView}, permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "gated-object", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "gated-object", component.ComponentID, "runBounded",
	)
	_, err = server.ExecuteFlow(ctx, request)
	if err != nil || runner.attempts != 1 ||
		!store.present || engine.calls != 1 || objectCalls != 1 || permissions.fenceCalls != 1 {
		t.Fatalf(
			"object-gated action = %v attempts=%d present=%v engine=%d objectCalls=%d",
			err, runner.attempts, store.present, engine.calls, objectCalls,
		)
	}
}

func TestFlowCatalogActionSuperUserRequiresSystemCapability(t *testing.T) {
	permissions := &flowServerTestPermissionEngine{}
	checker, err := newFlowCatalogActionPermissionChecker(
		&sharedcontract.EffectiveActionPermissions{
			SystemPermissions: map[string]bool{string(constants.PermModifyAllData): false},
			CustomPermissions: map[string]bool{string(constants.PermModifyAllData): true},
		},
		permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	user := &models.UserSession{ID: "user-1"}
	if !checker.HasPermission(t.Context(), user, string(constants.PermModifyAllData)) {
		t.Fatal("unified ordinary permission lookup lost the custom grant")
	}
	if checker.HasSuperUserCapability(t.Context(), user) {
		t.Fatal("custom permission collision granted the system superuser capability")
	}
}

func TestFlowExecutionServerCatalogActionPermissionFenceRollsBackEffect(t *testing.T) {
	view := flowServerTestView(t)
	appView, component := flowServerTestAppViewWithCatalogAction(t, view, true)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	permissions := &flowServerTestPermissionEngine{
		result: &sharedcontract.EffectivePermissionsResult{
			SystemPermissions: map[string]bool{"RunBoundedAction": true},
			ObjectPermissions: []sharedcontract.ObjectAccessProjection{{
				ResourceType: "WorkItem", CanRead: true,
			}},
		},
		fenceErr: &permissionsemantic.Failure{
			Category: permissionsemantic.FailureUnavailable,
		},
	}
	engine := &flowServerTestEngine{t: t, wantCatalogAction: true}
	server, err := newFlowServerTest(t,
		engine, store, runner, &flowServerTestCatalog{view: appView}, permissions,
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "fenced-action", view)
	request := flowServerTestCatalogActionRequest(
		t, ctx, "fenced-action", component.ComponentID, "runBounded",
	)
	_, err = server.ExecuteFlow(ctx, request)
	if status.Code(err) != codes.Unavailable || runner.attempts != 1 || store.present ||
		engine.calls != 1 || engine.resourceBoundedCalls != 1 ||
		permissions.transactionCalls != 0 || permissions.fenceCalls != 1 {
		t.Fatalf(
			"fenced action = %v attempts=%d present=%v engine=%d/%d permission=%d fence=%d",
			err, runner.attempts, store.present, engine.calls, engine.resourceBoundedCalls,
			permissions.transactionCalls, permissions.fenceCalls,
		)
	}
}

func TestFlowExecutionServerFailsClosedOnMixedCatalogAuthorityBeforeTransaction(t *testing.T) {
	view := flowServerTestView(t)
	appView := flowServerTestAppView(view)
	appView.Lineage.CatalogRevision++
	appView.Receipts.CatalogRevision++
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	server, err := newFlowServerTest(t,
		&flowServerTestEngine{t: t}, store, runner,
		&flowServerTestCatalog{view: appView},
		&flowServerTestPermissionEngine{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "mixed", view)
	_, err = server.ExecuteFlow(ctx, flowServerTestRequest(t, ctx, "mixed", "1"))
	if status.Code(err) != codes.Internal || runner.attempts != 0 || store.present {
		t.Fatalf("mixed authority = %v attempts=%d present=%v", err, runner.attempts, store.present)
	}
}

func TestFlowExecutionServerDoesNotDiscloseUnauthorizedRuntimeFlowExistence(t *testing.T) {
	view := flowServerTestView(t)
	appView := flowServerTestAppView(view)
	appView.AppComponents = nil
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
	for _, definitionRef := range []string{"AppOwnedAction", "DoesNotExist"} {
		t.Run(definitionRef, func(t *testing.T) {
			store := &flowServerTestStore{t: t}
			runner := &flowServerTestRunner{store: store}
			server, err := newFlowServerTest(t,
				&flowServerTestEngine{t: t},
				store,
				runner,
				&flowServerTestCatalog{view: appView},
				&flowServerTestPermissionEngine{},
			)
			if err != nil {
				t.Fatal(err)
			}
			ctx := flowServerTestContext(
				t,
				context.Background(),
				"undisclosed-"+definitionRef,
				view,
			)
			request, err := mapping.ToWireExecuteFlowRequest(
				ctx,
				flowsemantic.Request{
					DefinitionRef: definitionRef, DefinitionVersion: "1",
					IdempotencyKey:  "undisclosed-" + definitionRef,
					CanonicalInputs: []byte(`{}`),
				},
			)
			if err != nil {
				t.Fatal(err)
			}
			_, err = server.ExecuteFlow(ctx, request)
			if status.Code(err) != codes.NotFound ||
				runner.attempts != 0 ||
				store.present {
				t.Fatalf(
					"unauthorized Flow %q = %v attempts=%d present=%v",
					definitionRef,
					err,
					runner.attempts,
					store.present,
				)
			}
		})
	}
}

func TestFlowExecutionServerRejectsAppComponentRuntimeMetadataMismatch(t *testing.T) {
	view := flowServerTestView(t)
	appView := flowServerTestAppView(view)
	appView.AppComponents[0].CanonicalBehavior = []byte(`{"runAtomically":true}`)
	digest, err := appauthority.ComponentDigest(
		appView.AppComponents[0].ComponentKind,
		appView.AppComponents[0].ComponentFormatVersion,
		appView.AppComponents[0].CanonicalBehavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	appView.AppComponents[0].ContentDigest = digest
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
	store := &flowServerTestStore{t: t}
	runner := &flowServerTestRunner{store: store}
	server, err := newFlowServerTest(t,
		&flowServerTestEngine{t: t},
		store,
		runner,
		&flowServerTestCatalog{view: appView},
		&flowServerTestPermissionEngine{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := flowServerTestContext(t, context.Background(), "mismatch", view)
	_, err = server.ExecuteFlow(
		ctx,
		flowServerTestRequest(t, ctx, "mismatch", "1"),
	)
	if status.Code(err) != codes.Internal || runner.attempts != 0 || store.present {
		t.Fatalf(
			"metadata mismatch = %v attempts=%d present=%v",
			err,
			runner.attempts,
			store.present,
		)
	}
}

func TestAuthorizeFlowAppMetadataClosureRequiresEverySubflowComponent(t *testing.T) {
	view := flowServerTestView(t)
	root, ok := view.ActiveFlowVersion("AppOwnedAction")
	if !ok {
		t.Fatal("missing root Flow")
	}
	child, ok := view.ActiveFlowVersion("ChildAction")
	if !ok {
		t.Fatal("missing child Flow")
	}
	appView := flowServerTestAppView(view)
	appView.AppComponents = slices.DeleteFunc(
		appView.AppComponents,
		func(component catalogsemantic.AppMetadataComponent) bool {
			return component.BusinessKey == "ChildAction"
		},
	)
	_, err := authorizeFlowAppMetadataClosure(
		appView.AppComponents,
		nil,
		[]boundedAtomicFlowVersion{
			{definitionRef: "AppOwnedAction", version: root},
			{definitionRef: "ChildAction", version: child},
		},
		"AppOwnedAction",
		"1",
	)
	var failure *recordsemantic.Failure
	if !errors.As(err, &failure) ||
		failure.Category != recordsemantic.FailureMissing {
		t.Fatalf("missing child authority error = %v", err)
	}
}

func TestFlowAppMetadataAuthorityUsesRevisionBoundEffectiveOverride(t *testing.T) {
	baseBehavior := []byte(`{"status":"Active","label":"Base","runAtomically":true}`)
	baseDigest, err := appauthority.ComponentDigest(
		appMetadataStandardFlowKind, 1, baseBehavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	component := catalogsemantic.AppMetadataComponent{
		ComponentID: appauthority.ComponentID(
			"io.neutral", appMetadataStandardFlowKind, "AppFlow",
		),
		ComponentKind: appMetadataStandardFlowKind, ComponentFormatVersion: 1,
		NamespaceID: "io.neutral", BusinessKey: "AppFlow", DefinitionVersion: "1",
		ContentDigest: baseDigest, CanonicalBehavior: baseBehavior,
	}
	effectiveMetadata := models.FlowMetadata{
		Status: "Active", Label: "Tenant override", RunAtomically: true,
	}
	effectiveBehavior, err := json.Marshal(effectiveMetadata)
	if err != nil {
		t.Fatal(err)
	}
	envelope, err := catalogsemantic.EncodeEffectiveStandardFlowDocument(
		"2", effectiveBehavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	effective := catalogsemantic.EffectiveMetadataComponent{
		ComponentKind:          catalogsemantic.EffectiveStandardFlowComponentKind,
		ComponentFormatVersion: 1,
		BusinessKey:            "AppFlow",
		ContentDigest: catalogsemantic.ComputeEffectiveMetadataComponentDigest(
			envelope,
		),
		CanonicalBehavior: envelope,
	}
	version := &models.FlowVersion{VersionNumber: 2, Metadata: effectiveMetadata}
	authority, matches := appMetadataComponentMatchesRuntimeFlow(
		component, []catalogsemantic.EffectiveMetadataComponent{effective}, version,
	)
	if !matches || authority.definitionVersion != "2" ||
		authority.effectiveComponent == nil ||
		authority.effectiveComponent.ContentDigest != effective.ContentDigest {
		t.Fatalf("effective Flow authority = %#v, matches=%t", authority, matches)
	}

	version.Metadata.Label = "Different runtime Flow"
	if _, matches := appMetadataComponentMatchesRuntimeFlow(
		component, []catalogsemantic.EffectiveMetadataComponent{effective}, version,
	); matches {
		t.Fatal("mixed effective Flow authority was accepted")
	}
}

func TestFlowExecutionServerRetryRebuildsInputsAndCancellationKeepsNoReceipt(t *testing.T) {
	t.Run("proven rollback retry", func(t *testing.T) {
		view := flowServerTestView(t)
		store := &flowServerTestStore{t: t}
		runner := &flowServerTestRunner{store: store, retry: true}
		engine := &flowServerTestEngine{t: t, failFirst: true}
		server, err := newFlowServerTest(t,
			engine, store, runner, &flowServerTestCatalog{view: flowServerTestAppView(view)},
			&flowServerTestPermissionEngine{},
		)
		if err != nil {
			t.Fatal(err)
		}
		ctx := flowServerTestContext(t, context.Background(), "retry", view)
		if _, err := server.ExecuteFlow(ctx, flowServerTestRequest(t, ctx, "retry", "1")); err != nil {
			t.Fatal(err)
		}
		if runner.attempts != 2 || engine.calls != 2 ||
			store.row.State != persistence.RuntimeFlowInvocationCompleted {
			t.Fatalf("retried Flow = attempts %d calls %d row %#v", runner.attempts, engine.calls, store.row)
		}
	})
	for _, test := range []struct {
		name       string
		engine     func(*testing.T, context.CancelFunc) *flowServerTestEngine
		wantStatus codes.Code
	}{
		{
			name: "caller",
			engine: func(t *testing.T, cancel context.CancelFunc) *flowServerTestEngine {
				return &flowServerTestEngine{t: t, cancelCaller: cancel}
			},
			wantStatus: codes.Canceled,
		},
		{
			name: "owner",
			engine: func(t *testing.T, _ context.CancelFunc) *flowServerTestEngine {
				return &flowServerTestEngine{t: t, ownerCancel: true}
			},
			wantStatus: codes.Unavailable,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			view := flowServerTestView(t)
			base, cancel := context.WithCancel(context.Background())
			t.Cleanup(cancel)
			ctx := flowServerTestContext(t, base, test.name, view)
			store := &flowServerTestStore{t: t}
			runner := &flowServerTestRunner{store: store}
			server, err := newFlowServerTest(t,
				test.engine(t, cancel), store, runner,
				&flowServerTestCatalog{view: flowServerTestAppView(view)},
				&flowServerTestPermissionEngine{},
			)
			if err != nil {
				t.Fatal(err)
			}
			_, err = server.ExecuteFlow(ctx, flowServerTestRequest(t, ctx, test.name, "1"))
			if status.Code(err) != test.wantStatus || store.present {
				t.Fatalf("cancellation status = %v, row present=%v", err, store.present)
			}
		})
	}
}

func TestFlowExecutionServerPreservesInvocationConflictAndStaleBindingFailures(t *testing.T) {
	storedReceipts, err := json.Marshal(recordsemantic.Receipts{
		ProtocolVersion: "1.0", CapabilityVersion: flowsemantic.CapabilityName,
		CatalogRevision: 16, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
	})
	if err != nil {
		t.Fatal(err)
	}
	tests := []struct {
		name         string
		store        *flowServerTestStore
		wantCode     codes.Code
		wantCategory recordsemantic.FailureCategory
		wantAxis     recordsemantic.StaleAxis
	}{
		{
			name: "idempotency conflict",
			store: &flowServerTestStore{
				acceptErr: persistence.ErrRuntimeFlowInvocationConflict,
			},
			wantCode:     codes.InvalidArgument,
			wantCategory: recordsemantic.FailureInvalidRequest,
		},
		{
			name: "stale catalog binding",
			store: &flowServerTestStore{
				row:       persistence.RuntimeFlowInvocationRow{Receipts: storedReceipts},
				acceptErr: persistence.ErrRuntimeFlowInvocationServingBindingChanged,
			},
			wantCode:     codes.FailedPrecondition,
			wantCategory: recordsemantic.FailureStaleState,
			wantAxis:     recordsemantic.StaleAxisCatalog,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			test.store.t = t
			view := flowServerTestView(t)
			runner := &flowServerTestRunner{store: test.store}
			engine := &flowServerTestEngine{t: t}
			server, err := newFlowServerTest(t,
				engine,
				test.store,
				runner,
				&flowServerTestCatalog{view: flowServerTestAppView(view)},
				&flowServerTestPermissionEngine{},
			)
			if err != nil {
				t.Fatal(err)
			}
			ctx := flowServerTestContext(t, context.Background(), test.name, view)
			_, err = server.ExecuteFlow(
				ctx,
				flowServerTestRequest(t, ctx, "same-idempotency-key", "1"),
			)
			if status.Code(err) != test.wantCode {
				t.Fatalf("status = %v, want %v", err, test.wantCode)
			}
			decoded := mapping.FromPlatformStatusError(err, flowsemantic.CapabilityName)
			var failure *recordsemantic.Failure
			if !errors.As(decoded, &failure) ||
				failure.Category != test.wantCategory ||
				failure.StaleAxis != test.wantAxis {
				t.Fatalf("typed failure = %#v from %v", failure, decoded)
			}
			if engine.calls != 0 || runner.attempts != 1 {
				t.Fatalf("conflict reached engine: calls=%d attempts=%d", engine.calls, runner.attempts)
			}
		})
	}
}

func flowServerTestContext(
	t *testing.T,
	parent context.Context,
	principalID string,
	view *catalogsnapshot.RuntimeMetadataView,
) context.Context {
	return flowServerTestContextWithAppMetadataCapability(
		t, parent, principalID, view, catalogsemantic.AppMetadataV2CapabilityName,
	)
}

func flowServerTestContextWithAppMetadataCapability(
	t *testing.T,
	parent context.Context,
	principalID string,
	view *catalogsnapshot.RuntimeMetadataView,
	appMetadataCapability string,
) context.Context {
	t.Helper()
	capabilities := []string{flowsemantic.CapabilityName}
	if appMetadataCapability != "" {
		capabilities = append([]string{appMetadataCapability}, capabilities...)
	}
	ctx, cancel := context.WithTimeout(parent, time.Second)
	t.Cleanup(cancel)
	ctx = context.WithValue(ctx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: view, orgID: testOrgID, appInstallationID: testInstallationID,
		workloadID: testWorkloadID, capability: flowsemantic.CapabilityName,
		receipts: recordsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: flowsemantic.CapabilityName,
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
		},
	})
	return context.WithValue(ctx, trustedCallContextKey{}, newTrustedCall(delegationv1.Claims{
		DelegationID: "delegation-" + principalID, OrgID: testOrgID,
		PrincipalID: principalID, SessionID: "session-" + principalID,
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID,
		RuntimeAudience: "runtime-test", Capabilities: capabilities,
		AuthEpoch: 2, SecurityEpoch: 3,
	}, models.UserSession{ID: principalID}))
}

func flowServerTestCatalogAdmission(
	t *testing.T,
	capability string,
) *CatalogConsumerAdmission {
	t.Helper()
	admission, err := NewCatalogConsumerAdmission([]CatalogConsumerSelection{{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		WorkloadID: testWorkloadID, CapabilityID: capability,
	}})
	if err != nil {
		t.Fatal(err)
	}
	return admission
}

func newFlowServerTest(
	t *testing.T,
	engine FlowExecutionEngine,
	store FlowInvocationStore,
	transactions FlowTransactionRunner,
	catalog AppMetadataViewEngine,
	permissions EffectivePermissionEngine,
) (*FlowExecutionServer, error) {
	t.Helper()
	return NewFlowExecutionServer(
		engine, store, transactions, catalog, permissions,
		flowServerTestCatalogAdmission(t, catalogsemantic.AppMetadataV2CapabilityName),
	)
}

func flowServerTestRequest(
	t *testing.T,
	ctx context.Context,
	idempotencyKey string,
	version string,
) *runtimev1.ExecuteFlowRequest {
	t.Helper()
	request, err := mapping.ToWireExecuteFlowRequest(ctx, flowsemantic.Request{
		DefinitionRef: "AppOwnedAction", DefinitionVersion: version,
		IdempotencyKey: idempotencyKey, CanonicalInputs: []byte(`{"input":"value"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	return request
}

func flowServerTestCatalogActionRequest(
	t *testing.T,
	ctx context.Context,
	idempotencyKey string,
	componentRef string,
	actionRef string,
) *runtimev1.ExecuteFlowRequest {
	t.Helper()
	request, err := mapping.ToWireExecuteCatalogActionRequest(ctx, flowsemantic.CatalogActionRequest{
		IdempotencyKey: idempotencyKey, CanonicalInputs: []byte(`{"input":"value"}`),
		CatalogAction: flowsemantic.CatalogActionBinding{
			ComponentRef: componentRef, ActionRef: actionRef,
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	return request
}

func flowServerTestCatalogActionPreviewRequest(
	t *testing.T,
	ctx context.Context,
	mode flowsemantic.CatalogActionPreviewMode,
	componentRef string,
) *runtimev1.PreviewCatalogActionRequest {
	t.Helper()
	request := flowsemantic.CatalogActionPreviewRequest{
		Mode: mode,
		CatalogAction: flowsemantic.CatalogActionBinding{
			ComponentRef: componentRef, ActionRef: "runBounded",
		},
	}
	switch mode {
	case flowsemantic.CatalogActionPreviewCreateRecords:
		request.CanonicalInputs = []byte(`{"recordId":"a99000000000001"}`)
	case flowsemantic.CatalogActionPreviewOutputs:
		request.RecordID = "a99000000000001"
	}
	wireRequest, err := mapping.ToWirePreviewCatalogActionRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	return wireRequest
}

func flowServerTestView(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	return flowServerTestViewWithRootVersion(t, 1)
}

func flowServerTestViewWithRootVersion(
	t *testing.T,
	rootVersion int,
) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	_, viewArtifact, _ := flowServerTestArtifactsForMutation(t, false, rootVersion)
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	return view
}

func flowServerTestArtifactsForMutation(
	t *testing.T,
	mutate bool,
	rootVersions ...int,
) (catalogsnapshot.Artifact, catalogsnapshot.Artifact, catalogsnapshot.Artifact) {
	t.Helper()
	rootVersion := 1
	if len(rootVersions) > 0 {
		rootVersion = rootVersions[0]
	}
	rootMetadata, err := json.Marshal(models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "CallChild"},
		},
		Subflows: []models.FlowSubflow{{
			Name: "CallChild", FlowName: "ChildAction",
		}},
	})
	if err != nil {
		t.Fatal(err)
	}
	child := models.FlowMetadata{
		RunAtomically: true,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "Done"},
		},
		Assignments: []models.FlowAssignment{{Name: "Done"}},
	}
	if mutate {
		rootMetadata, err = json.Marshal(models.FlowMetadata{
			RunAtomically: true,
			Start: &models.FlowStart{
				Connector: &models.FlowConnector{TargetReference: "LoadWorkItem"},
			},
			RecordLookups: []models.FlowRecordLookup{{
				Name: "LoadWorkItem", Object: "WorkItem",
				GetFirstRecordOnly: true, OutputReference: "workItem",
				Filters: []models.FlowRecordFilter{{
					Field: "Id", Operator: "EqualTo",
					Value: map[string]any{"stringValue": "a99000000000001"},
				}},
				Connector: &models.FlowConnector{TargetReference: "UpdateWorkItem"},
			}},
			RecordUpdates: []models.FlowRecordUpdate{{
				Name: "UpdateWorkItem", Object: "WorkItem", InputReference: "workItem",
				InputAssignments: []models.FlowInputAssignment{{
					Field: "Status", Value: map[string]any{"stringValue": "after"},
				}},
				Connector: &models.FlowConnector{TargetReference: "CallChild"},
			}},
			Subflows: []models.FlowSubflow{{
				Name: "CallChild", FlowName: "ChildAction",
			}},
		})
		if err != nil {
			t.Fatal(err)
		}
	}
	childMetadata, err := json.Marshal(child)
	if err != nil {
		t.Fatal(err)
	}
	previewMetadata, err := json.Marshal(models.FlowMetadata{
		RunAtomically: false,
		Start: &models.FlowStart{
			Connector: &models.FlowConnector{TargetReference: "Done"},
		},
		Assignments: []models.FlowAssignment{{Name: "Done"}},
	})
	if err != nil {
		t.Fatal(err)
	}
	revision := catalogsnapshot.CatalogRevision(17)
	workItemKeyPrefix := "a99"
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: testOrgID,
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision: revision, SchemaGeneration: 8, EngineGeneration: 9,
		},
		Objects: []catalogsnapshot.ObjectInput{{
			CatalogRevision: revision,
			Metadata: &models.ObjectMetadata{
				ID: "object-work-item", APIName: "WorkItem",
				Label: "Work Item", PluralLabel: "Work Items",
				KeyPrefix:    &workItemKeyPrefix,
				SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{
					{
						ID: "field-work-item-id", APIName: constants.FieldID,
						Label: "ID", Type: constants.FieldTypeString,
						PhysicalName: "Id", IsSystem: true,
					},
					{
						ID: "field-work-item-status", APIName: "Status",
						Label: "Status", Type: constants.FieldTypeString,
						PhysicalName: "Status", Updateable: true,
					},
				},
			},
		}},
		MutationDefinitions: []catalogsnapshot.MutationDefinitionInput{
			{
				CatalogRevision: revision,
				FlowDefinition: &models.FlowDefinition{
					Id: "definition-action", DeveloperName: "AppOwnedAction",
					ProcessType: "AutoLaunchedFlow", ActiveVersionId: "version-action",
					OwnerPack: "app-pack",
				},
			},
			{
				CatalogRevision: revision,
				FlowVersion: &models.FlowVersion{
					Id: "version-action", DefinitionId: "definition-action",
					VersionNumber: rootVersion, Status: models.FlowStatusActive,
					ProcessType: "AutoLaunchedFlow", Source: "standard",
					MetadataRaw: rootMetadata,
				},
			},
			{
				CatalogRevision: revision,
				FlowDefinition: &models.FlowDefinition{
					Id: "definition-child", DeveloperName: "ChildAction",
					ProcessType: "AutoLaunchedFlow", ActiveVersionId: "version-child",
					OwnerPack: "app-pack",
				},
			},
			{
				CatalogRevision: revision,
				FlowVersion: &models.FlowVersion{
					Id: "version-child", DefinitionId: "definition-child",
					VersionNumber: 1, Status: models.FlowStatusActive,
					ProcessType: "AutoLaunchedFlow", Source: "standard",
					MetadataRaw: childMetadata,
				},
			},
			{
				CatalogRevision: revision,
				FlowDefinition: &models.FlowDefinition{
					Id: "definition-preview", DeveloperName: "PreviewAction",
					ProcessType: "AutoLaunchedFlow", ActiveVersionId: "version-preview",
					OwnerPack: "app-pack",
				},
			},
			{
				CatalogRevision: revision,
				FlowVersion: &models.FlowVersion{
					Id: "version-preview", DefinitionId: "definition-preview",
					VersionNumber: 1, Status: models.FlowStatusActive,
					ProcessType: "AutoLaunchedFlow", Source: "standard",
					MetadataRaw: previewMetadata,
				},
			},
		},
		Projections: []catalogsnapshot.RuntimeProjectionInput{
			{
				CatalogRevision: revision, AppInstallationID: testInstallationID,
				CapabilityID: flowsemantic.CapabilityName,
				Profile:      catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{
					"WorkItem",
				},
			},
			{
				CatalogRevision: revision, AppInstallationID: testInstallationID,
				CapabilityID: mutationsemantic.CapabilityName,
				Profile:      catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{
					"WorkItem",
				},
			},
			{
				CatalogRevision: revision, AppInstallationID: testInstallationID,
				CapabilityID: catalogsemantic.AppMetadataV2CapabilityName,
				Profile:      catalogsnapshot.ProjectionRecordQueryV1,
				ObjectAPINames: []string{
					"WorkItem",
				},
			},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: testInstallationID,
			CapabilityID:      flowsemantic.CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	mutationArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: testInstallationID,
			CapabilityID:      mutationsemantic.CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	return snapshotArtifact, viewArtifact, mutationArtifact
}

func flowServerTestAppView(runtimeView *catalogsnapshot.RuntimeMetadataView) catalogsemantic.AppMetadataViewV2 {
	namespaceID := "io.neutral"
	components := make([]catalogsemantic.AppMetadataComponent, 0, 3)
	for _, definitionRef := range []string{"AppOwnedAction", "ChildAction", "PreviewAction"} {
		version, ok := runtimeView.ActiveFlowVersion(definitionRef)
		if !ok {
			panic("test Runtime view has no " + definitionRef)
		}
		behavior := append([]byte(nil), version.MetadataRaw...)
		contentDigest, err := appauthority.ComponentDigest(
			appMetadataStandardFlowKind, 1, behavior,
		)
		if err != nil {
			panic(err)
		}
		components = append(components, catalogsemantic.AppMetadataComponent{
			ComponentID: appauthority.ComponentID(
				namespaceID, appMetadataStandardFlowKind, definitionRef,
			),
			ComponentKind:          appMetadataStandardFlowKind,
			ComponentFormatVersion: 1, NamespaceID: namespaceID,
			BusinessKey: definitionRef, DefinitionVersion: "1",
			ContentDigest: contentDigest, CanonicalBehavior: behavior,
		})
	}
	view := catalogsemantic.AppMetadataViewV2{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		AppID: "io.nexuscrm.neutral", AppReleaseDigest: "sha256:" + strings.Repeat("d", 64),
		Lineage: catalogsemantic.ViewLineage{
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9,
			SourceSnapshotDigest: string(runtimeView.SourceSnapshotDigest()),
		},
		AppComponents: components,
		Receipts: catalogsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: catalogsemantic.AppMetadataV2CapabilityName,
			CatalogRevision: 17, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 3,
		},
	}
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(view)
	return view
}

func flowServerTestAppViewV3(runtimeView *catalogsnapshot.RuntimeMetadataView) catalogsemantic.AppMetadataViewV3 {
	v2 := flowServerTestAppView(runtimeView)
	v3 := catalogsemantic.AppMetadataViewV3{
		OrgID: v2.OrgID, AppInstallationID: v2.AppInstallationID,
		AppID: v2.AppID, AppReleaseDigest: v2.AppReleaseDigest,
		Lineage: v2.Lineage, AppComponents: v2.AppComponents,
		EffectiveComponents: v2.EffectiveComponents, Receipts: v2.Receipts,
	}
	v3.Receipts.CapabilityVersion = catalogsemantic.AppMetadataV3CapabilityName
	v3.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(v3)
	return v3
}

func flowServerTestAppViewWithCatalogAction(
	t *testing.T,
	runtimeView *catalogsnapshot.RuntimeMetadataView,
	atomic bool,
) (catalogsemantic.AppMetadataViewV2, catalogsemantic.AppMetadataComponent) {
	t.Helper()
	view := flowServerTestAppView(runtimeView)
	behavior := []byte(`[{"name":"runBounded","flowDeveloperName":"AppOwnedAction","requiredPermissions":["RunBoundedAction"],"isAtomic":` +
		fmt.Sprintf("%t", atomic) +
		`,"inputs":[{"apiName":"defaulted","dataType":"string"},{"apiName":"input","dataType":"string","isRequired":true}],"fieldDefaults":[{"targetObject":"$Input","fieldAPIName":"defaulted","valueFormula":"\"prepared\"","priority":0}],"inputGuards":[{"condition":"true","requiredObject":"WorkItem","requiredOperation":"Read"}],"internalAccessScopes":[{"objectApiName":"WorkItem","operation":"Read"}]}]`)
	component := catalogsemantic.AppMetadataComponent{
		ComponentID: appauthority.ComponentID(
			"io.neutral", appMetadataInvocableActionKind, "bootstrap-actions",
		),
		ComponentKind: appMetadataInvocableActionKind, ComponentFormatVersion: 1,
		NamespaceID: "io.neutral", BusinessKey: "bootstrap-actions", DefinitionVersion: "1",
		CanonicalBehavior: behavior,
	}
	var err error
	component.ContentDigest, err = appauthority.ComponentDigest(
		component.ComponentKind, component.ComponentFormatVersion, behavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	view.AppComponents = append(view.AppComponents, component)
	sort.Slice(view.AppComponents, func(i, j int) bool {
		return view.AppComponents[i].ComponentID < view.AppComponents[j].ComponentID
	})
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(view)
	return view, component
}

func flowServerTestAppViewWithPreviewCatalogAction(
	t *testing.T,
	runtimeView *catalogsnapshot.RuntimeMetadataView,
) (catalogsemantic.AppMetadataViewV2, catalogsemantic.AppMetadataComponent) {
	t.Helper()
	view := flowServerTestAppView(runtimeView)
	behavior := []byte(`[{"name":"runBounded","flowDeveloperName":"AppOwnedAction","previewFlowDeveloperName":"PreviewAction","requiredPermissions":["RunBoundedAction"],"isAtomic":true,"inputs":[{"apiName":"recordId","dataType":"id","isRequired":true,"uiMode":"hidden","targetObject":"WorkItem","targetFieldAPIName":"Id"}],"inputGuards":[{"condition":"true","requiredObject":"WorkItem","requiredOperation":"Read"}]}]`)
	component := catalogsemantic.AppMetadataComponent{
		ComponentID: appauthority.ComponentID(
			"io.neutral", appMetadataInvocableActionKind, "preview-actions",
		),
		ComponentKind: appMetadataInvocableActionKind, ComponentFormatVersion: 1,
		NamespaceID: "io.neutral", BusinessKey: "preview-actions", DefinitionVersion: "1",
		CanonicalBehavior: behavior,
	}
	var err error
	component.ContentDigest, err = appauthority.ComponentDigest(
		component.ComponentKind, component.ComponentFormatVersion, behavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	view.AppComponents = append(view.AppComponents, component)
	sort.Slice(view.AppComponents, func(i, j int) bool {
		return view.AppComponents[i].ComponentID < view.AppComponents[j].ComponentID
	})
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(view)
	return view, component
}

func slicesEqual(left, right []string) bool {
	if len(left) != len(right) {
		return false
	}
	for i := range left {
		if left[i] != right[i] {
			return false
		}
	}
	return true
}