package runtimeapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"io"
	"log"
	"strconv"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/automation/standardaction"
	flowexecution "github.com/nexuscrm/backend/internal/runtimeflowexecution"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	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"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	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"
)

const appMetadataStandardFlowKind = "metadata.pack.standard-flow.v1"

type FlowExecutionEngine interface {
	ValidateAtomicFlowForExecution(
		context.Context,
		string,
		*models.FlowMetadata,
	) error
	ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Context,
		string,
		map[string]*models.FlowVersion,
		map[string]interface{},
		*models.UserSession,
		uint64,
	) (*contractmodels.FlowExecutionTypedOutcome, error)
	ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
		context.Context,
		string,
		map[string]*models.FlowVersion,
		map[string]interface{},
		*models.UserSession,
		uint64,
	) (*contractmodels.FlowExecutionTypedOutcome, error)
	StandardActionCreatePreviewSupported(
		context.Context,
		string,
		*models.UserSession,
	) (bool, error)
	PreparedStandardActionCreatePreviewSupported(
		context.Context,
		string,
		map[string]*models.FlowVersion,
		*models.UserSession,
	) (bool, error)
	PreviewStandardActionRecordCreateFields(
		context.Context,
		string,
		map[string]interface{},
		*models.UserSession,
	) (*automation.FlowRecordCreatePreviewResult, error)
	PreviewPreparedStandardActionRecordCreateFields(
		context.Context,
		string,
		map[string]*models.FlowVersion,
		map[string]interface{},
		*models.UserSession,
	) (*automation.FlowRecordCreatePreviewResult, error)
	PreviewFlowOutputs(
		context.Context,
		string,
		map[string]interface{},
		*models.UserSession,
	) (*automation.FlowPreviewOutputsResult, error)
	PreviewPreparedFlowOutputs(
		context.Context,
		string,
		map[string]*models.FlowVersion,
		map[string]interface{},
		*models.UserSession,
	) (*automation.FlowPreviewOutputsResult, error)
}

type transactionBoundEffectivePermissionEngine interface {
	EffectivePermissionEngine
	EffectiveActionPermissions(
		context.Context,
		*models.UserSession,
		permissionstatev1.Version,
	) (*sharedcontract.EffectiveActionPermissions, error)
	EffectiveActionPermissionsWithinTransaction(
		context.Context,
		*models.UserSession,
		permissionstatev1.Version,
	) (*sharedcontract.EffectiveActionPermissions, error)
	EffectivePermissionsWithinTransaction(
		context.Context,
		*models.UserSession,
		string,
		permissionstatev1.Version,
	) (*sharedcontract.EffectivePermissionsResult, error)
	FencePermissionFloorWithinTransaction(
		context.Context,
		permissionstatev1.Version,
	) error
}

type catalogActionObjectPermissionEngine interface {
	CheckObjectPermissionWithUser(
		context.Context,
		string,
		string,
		*models.UserSession,
	) bool
}

type FlowInvocationStore interface {
	Accept(
		context.Context,
		persistence.RuntimeFlowInvocationAcceptance,
	) (persistence.RuntimeFlowInvocationRow, bool, error)
	Complete(context.Context, string, []byte) error
}

type flowAppMetadataAdmission interface {
	flowAppMetadataScope(TrustedCall) (CatalogConsumerScope, string, error)
}

type FlowTransactionRunner interface {
	RunInPessimisticTransactionWithOutcome(
		context.Context,
		func(context.Context) error,
	) (ports.TransactionCompletionOutcome, error)
}

type flowExecutionRequest struct {
	DefinitionRef      string
	DefinitionVersion  string
	IdempotencyKey     string
	CanonicalInputs    []byte
	MinCatalogRevision *uint64
	CatalogAction      *flowsemantic.CatalogActionBinding
}

func flowExecutionRequestFromWire(
	ctx context.Context,
	wireRequest *runtimev1.ExecuteFlowRequest,
) (flowExecutionRequest, error) {
	if wireRequest != nil && wireRequest.GetCatalogAction() != nil {
		request, err := mapping.FromWireExecuteCatalogActionRequest(ctx, wireRequest)
		if err != nil {
			return flowExecutionRequest{}, err
		}
		binding := request.CatalogAction
		return flowExecutionRequest{
			IdempotencyKey: request.IdempotencyKey, CanonicalInputs: request.CanonicalInputs,
			MinCatalogRevision: request.MinCatalogRevision, CatalogAction: &binding,
		}, nil
	}
	request, err := mapping.FromWireExecuteFlowRequest(ctx, wireRequest)
	if err != nil {
		return flowExecutionRequest{}, err
	}
	return flowExecutionRequest{
		DefinitionRef: request.DefinitionRef, DefinitionVersion: request.DefinitionVersion,
		IdempotencyKey: request.IdempotencyKey, CanonicalInputs: request.CanonicalInputs,
		MinCatalogRevision: request.MinCatalogRevision,
	}, nil
}

type flowPresentationRollback struct {
	response *flowsemantic.ResponseV2
	failure  error
}

func (e *flowPresentationRollback) Error() string {
	return "runtime flow execution failed before commit"
}

type FlowExecutionServer struct {
	runtimev1.UnimplementedFlowExecutionServiceServer
	engine                  FlowExecutionEngine
	store                   FlowInvocationStore
	transactions            FlowTransactionRunner
	catalog                 AppMetadataViewEngine
	catalogAdmission        flowAppMetadataAdmission
	permissions             transactionBoundEffectivePermissionEngine
	actionObjectPermissions catalogActionObjectPermissionEngine
	actionPreviewAccess     standardaction.PreviewAccessChecker
}

func NewFlowExecutionServer(
	engine FlowExecutionEngine,
	store FlowInvocationStore,
	transactions FlowTransactionRunner,
	catalog AppMetadataViewEngine,
	permissions EffectivePermissionEngine,
	catalogAdmission flowAppMetadataAdmission,
) (*FlowExecutionServer, error) {
	if engine == nil || store == nil || transactions == nil || catalog == nil ||
		catalogAdmission == nil {
		return nil, errors.New(
			"runtime flow execution: engine, durable invocation store, transaction owner, Catalog reader, and workload-scoped Catalog admission are required",
		)
	}
	var transactionBoundPermissions transactionBoundEffectivePermissionEngine
	var actionObjectPermissions catalogActionObjectPermissionEngine
	var actionPreviewAccess standardaction.PreviewAccessChecker
	if permissions != nil {
		var ok bool
		transactionBoundPermissions, ok = permissions.(transactionBoundEffectivePermissionEngine)
		if !ok {
			return nil, errors.New(
				"runtime flow execution: permission engine must support transaction-bound authorization",
			)
		}
		actionObjectPermissions, _ = permissions.(catalogActionObjectPermissionEngine)
		actionPreviewAccess, _ = permissions.(standardaction.PreviewAccessChecker)
	}
	return &FlowExecutionServer{
		engine: engine, store: store, transactions: transactions,
		catalog: catalog, catalogAdmission: catalogAdmission,
		permissions:             transactionBoundPermissions,
		actionObjectPermissions: actionObjectPermissions,
		actionPreviewAccess:     actionPreviewAccess,
	}, nil
}

func (s *FlowExecutionServer) ExecuteFlow(
	ctx context.Context,
	wireRequest *runtimev1.ExecuteFlowRequest,
) (*runtimev1.ExecuteFlowResponse, error) {
	ctx = withRuntimeFlowProfiling(ctx)
	totalStarted := startRuntimeFlowStage(ctx)
	stageStarted := startRuntimeFlowStage(ctx)
	request, err := flowExecutionRequestFromWire(ctx, wireRequest)
	finishRuntimeFlowStage(ctx, "decode-request", stageStarted, err)
	if err != nil {
		return nil, flowStatus(err)
	}
	stageStarted = startRuntimeFlowStage(ctx)
	scope, scopeOK := runtimeExecutionScopeFromContext(ctx)
	view, viewOK := RuntimeMetadataViewFromContext(ctx)
	trusted, trustedOK := trustedCallFromContext(ctx)
	if !scopeOK || !viewOK || !trustedOK {
		return nil, flowStatus(&recordsemantic.Failure{Category: recordsemantic.FailureUnavailable})
	}
	if scope.capability != flowsemantic.CapabilityName ||
		scope.orgID != trusted.orgID ||
		scope.appInstallationID != trusted.appInstallationID ||
		scope.workloadID != trusted.callerWorkloadID ||
		scope.receipts.SecurityEpoch != trusted.securityEpoch {
		return nil, flowStatus(&recordsemantic.Failure{Category: recordsemantic.FailureForbidden})
	}
	if err := requireCatalogRevision(request.MinCatalogRevision, scope.receipts.CatalogRevision); err != nil {
		return nil, flowStatus(err)
	}
	finishRuntimeFlowStage(ctx, "admitted-context", stageStarted, nil)
	stageStarted = startRuntimeFlowStage(ctx)
	appView, err := s.flowAppMetadataAuthority(ctx, scope, view, trusted)
	finishRuntimeFlowStage(ctx, "app-metadata-authority", stageStarted, err)
	if err != nil {
		return nil, flowStatus(err)
	}
	stageStarted = startRuntimeFlowStage(ctx)
	inputs, err := decodeFlowInputs(request.CanonicalInputs)
	finishRuntimeFlowStage(ctx, "decode-inputs", stageStarted, err)
	if err != nil {
		return nil, flowStatus(err)
	}
	definitionRef := request.DefinitionRef
	definitionVersion := request.DefinitionVersion
	var component catalogsemantic.AppMetadataComponent
	var actionAuthority *flowCatalogActionAuthority
	var preparedAction *standardaction.PreparedCatalogInvocation
	var selection boundedAtomicFlowSelection
	stageStarted = startRuntimeFlowStage(ctx)
	if request.CatalogAction == nil {
		component, err = selectCatalogActionFlowAppMetadataComponent(
			appView.AppComponents, definitionRef,
		)
		if err != nil {
			return nil, flowStatus(err)
		}
		selection, err = selectBoundedAtomicFlowWithValidator(
			ctx,
			view,
			definitionRef,
			s.engine.ValidateAtomicFlowForExecution,
		)
	} else {
		if s.permissions == nil || s.actionObjectPermissions == nil {
			return nil, flowStatus(&recordsemantic.Failure{
				Category: recordsemantic.FailureUnavailable,
			})
		}
		viewer := trusted.UserSession()
		if viewer.ID == "" {
			return nil, flowStatus(&recordsemantic.Failure{
				Category: recordsemantic.FailureForbidden,
			})
		}
		projected, permissionErr := s.permissions.EffectiveActionPermissions(
			ctx, &viewer, trusted.PermissionFloor(),
		)
		if permissionErr != nil {
			return nil, flowStatus(classifyEffectivePermissionError(ctx, permissionErr))
		}
		actionAuthority, err = resolveFlowCatalogActionAuthority(
			appView.AppComponents, appView.EffectiveComponents,
			request.CatalogAction, projected, view,
			s.actionObjectPermissions,
		)
		if err != nil {
			logFlowCatalogActionPreflightFailure(
				"resolve-authority", request.CatalogAction, definitionRef, err,
			)
		}
		if err == nil {
			definitionRef = actionAuthority.invocation.DefinitionRef()
			if !actionAuthority.effective {
				component, err = selectCatalogActionFlowAppMetadataComponent(
					appView.AppComponents, definitionRef,
				)
				if err != nil {
					logFlowCatalogActionPreflightFailure(
						"select-app-flow-authority", request.CatalogAction,
						definitionRef, err,
					)
				}
			}
		}
		if err == nil {
			preparedAction, err = actionAuthority.preparePlan(ctx, inputs, &viewer)
			if err != nil {
				logFlowCatalogActionPreflightFailure(
					"prepare-inputs", request.CatalogAction, definitionRef, err,
				)
			}
		}
		if err == nil {
			selection, err = selectCatalogActionFlow(
				ctx, view, definitionRef,
				s.engine.ValidateAtomicFlowForExecution,
			)
			if err != nil {
				logFlowCatalogActionPreflightFailure(
					"select-bounded-atomic-closure", request.CatalogAction,
					definitionRef, err,
				)
			}
		}
	}
	finishRuntimeFlowStage(ctx, "select-and-prepare-authority", stageStarted, err)
	if err != nil {
		var failure *commonsemantic.Failure
		if errors.As(err, &failure) {
			return nil, flowStatus(err)
		}
		return nil, flowStatus(&recordsemantic.Failure{
			Category:         recordsemantic.FailureInvalidRequest,
			InvalidComponent: commonsemantic.InvalidComponentCommandData,
		})
	}
	var flowAuthority flowAppMetadataAuthority
	requireAppFlowAuthority := actionAuthority == nil || !actionAuthority.effective
	stageStarted = startRuntimeFlowStage(ctx)
	if requireAppFlowAuthority {
		flowAuthority, err = authorizeFlowAppMetadataClosure(
			appView.AppComponents,
			appView.EffectiveComponents,
			selection.versions,
			definitionRef,
			definitionVersion,
		)
		if err != nil {
			return nil, flowStatus(err)
		}
		component = flowAuthority.appComponent
		definitionVersion = flowAuthority.definitionVersion
	} else {
		if selection.root == nil || selection.root.VersionNumber <= 0 {
			return nil, flowStatus(&recordsemantic.Failure{
				Category: recordsemantic.FailureInternal,
			})
		}
		// Catalog actions are authorized by the revision-bound effective action
		// registry. Their exact Flow version comes from the same immutable
		// RuntimeMetadataView, rather than being misclassified as L1
		// AppRelease-owned metadata.
		definitionVersion = strconv.Itoa(selection.root.VersionNumber)
	}
	finishRuntimeFlowStage(ctx, "authorize-flow-closure", stageStarted, nil)

	stageStarted = startRuntimeFlowStage(ctx)
	receipts := receiptsForCapability(scope.receipts, flowsemantic.CapabilityName)
	receiptPayload, err := json.Marshal(receipts)
	if err != nil {
		return nil, flowStatus(&recordsemantic.Failure{Category: recordsemantic.FailureInternal})
	}
	requestAuthority := struct {
		DefinitionRef                string          `json:"definitionRef"`
		DefinitionVersion            string          `json:"definitionVersion"`
		ComponentID                  string          `json:"componentId,omitempty"`
		ComponentDigest              string          `json:"componentDigest,omitempty"`
		AppMetadataDigest            string          `json:"appMetadataDigest"`
		RuntimeClosureDigest         string          `json:"runtimeClosureDigest"`
		CanonicalInputs              json.RawMessage `json:"canonicalInputs"`
		CatalogActionComponentID     string          `json:"catalogActionComponentId,omitempty"`
		CatalogActionComponentDigest string          `json:"catalogActionComponentDigest,omitempty"`
		CatalogActionRef             string          `json:"catalogActionRef,omitempty"`
		EffectiveComponentDigest     string          `json:"effectiveComponentDigest,omitempty"`
	}{
		DefinitionRef: definitionRef, DefinitionVersion: definitionVersion,
		ComponentID: component.ComponentID, ComponentDigest: component.ContentDigest,
		AppMetadataDigest: appView.Lineage.ViewDigest, RuntimeClosureDigest: selection.closureDigest,
		CanonicalInputs: request.CanonicalInputs,
	}
	if requireAppFlowAuthority && flowAuthority.effectiveComponent != nil {
		requestAuthority.EffectiveComponentDigest =
			flowAuthority.effectiveComponent.ContentDigest
	}
	if actionAuthority != nil {
		requestAuthority.CatalogActionComponentID = actionAuthority.component.ComponentID
		requestAuthority.CatalogActionComponentDigest = actionAuthority.component.ContentDigest
		requestAuthority.CatalogActionRef = actionAuthority.invocation.ActionRef()
	}
	requestPayload, err := json.Marshal(requestAuthority)
	if err != nil {
		return nil, flowStatus(&recordsemantic.Failure{Category: recordsemantic.FailureInternal})
	}
	scopeSum := sha256.Sum256([]byte(
		trusted.orgID + "\x00" + trusted.appInstallationID + "\x00" +
			trusted.callerWorkloadID + "\x00" + trusted.principalID + "\x00" +
			flowsemantic.CapabilityName,
	))
	scopeDigest := hex.EncodeToString(scopeSum[:])
	requestSum := sha256.Sum256(requestPayload)
	acceptance := persistence.RuntimeFlowInvocationAcceptance{
		RowID: deterministicFileID(
			scopeDigest,
			definitionRef,
			request.IdempotencyKey,
		),
		ScopeDigest:       scopeDigest,
		IdempotencyKey:    request.IdempotencyKey,
		DefinitionRef:     definitionRef,
		DefinitionVersion: definitionVersion,
		ClosureDigest:     selection.closureDigest,
		RequestDigest:     "sha256:" + hex.EncodeToString(requestSum[:]),
		Receipts:          receiptPayload,
	}
	finishRuntimeFlowStage(ctx, "prepare-invocation-acceptance", stageStarted, nil)

	var response flowsemantic.ResponseV2
	stageStarted = startRuntimeFlowStage(ctx)
	completion, err := s.transactions.RunInPessimisticTransactionWithOutcome(ctx, func(txCtx context.Context) error {
		response = flowsemantic.ResponseV2{}
		txStageStarted := startRuntimeFlowStage(txCtx)
		var inputs map[string]interface{}
		var decodeErr error
		if actionAuthority != nil {
			if preparedAction == nil {
				return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
			}
			inputs, decodeErr = preparedAction.Inputs()
		} else {
			inputs, decodeErr = decodeFlowInputs(request.CanonicalInputs)
		}
		finishRuntimeFlowStage(txCtx, "transaction.decode-inputs", txStageStarted, decodeErr)
		if decodeErr != nil {
			return decodeErr
		}
		user := trusted.UserSession()
		executionCtx := txCtx
		txStageStarted = startRuntimeFlowStage(txCtx)
		if actionAuthority != nil {
			executionCtx = preparedAction.Bind(txCtx)
		}
		finishRuntimeFlowStage(txCtx, "transaction.authorize-and-prepare", txStageStarted, nil)
		// The prepared action plan was authorized against the admitted Permission
		// State floor. Validate that floor with a locking current read at the
		// transaction's serialization point. A revocation that committed while the
		// request was in flight rolls this transaction back; a later revocation
		// waits only for the short terminal-write and commit tail.
		fencePermissionFloor := func() error {
			if actionAuthority == nil {
				return nil
			}
			fenceStarted := startRuntimeFlowStage(txCtx)
			fenceErr := s.permissions.FencePermissionFloorWithinTransaction(
				txCtx, trusted.PermissionFloor(),
			)
			finishRuntimeFlowStage(txCtx, "transaction.fence-permission-floor", fenceStarted, fenceErr)
			if fenceErr != nil {
				return classifyEffectivePermissionError(txCtx, fenceErr)
			}
			return nil
		}
		txStageStarted = startRuntimeFlowStage(txCtx)
		row, duplicate, acceptErr := s.store.Accept(txCtx, acceptance)
		finishRuntimeFlowStage(txCtx, "transaction.accept-invocation", txStageStarted, acceptErr)
		if acceptErr != nil {
			return classifyFlowInvocationError(txCtx, acceptErr, row, receipts)
		}
		if duplicate {
			if row.State != persistence.RuntimeFlowInvocationCompleted || len(row.ResultPayload) == 0 {
				return &recordsemantic.Failure{Category: recordsemantic.FailureUnavailable}
			}
			if err := json.Unmarshal(row.ResultPayload, &response); err != nil ||
				flowsemantic.ValidateResponseV2(response) != nil {
				return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
			}
			return fencePermissionFloor()
		}

		preparedClosure := make(map[string]*models.FlowVersion, len(selection.versions))
		for _, selected := range selection.versions {
			preparedClosure[selected.definitionRef] = selected.version
		}
		var outcome *contractmodels.FlowExecutionTypedOutcome
		var executeErr error
		txStageStarted = startRuntimeFlowStage(txCtx)
		if actionAuthority == nil {
			outcome, executeErr = s.engine.ExecutePreparedFlowClosureWithinOwnedTransactionOutcome(
				executionCtx,
				definitionRef,
				preparedClosure,
				inputs,
				&user,
				selection.maxElementExecutions,
			)
		} else {
			outcome, executeErr = s.engine.ExecuteResourceBoundedPreparedFlowClosureWithinOwnedTransactionOutcome(
				executionCtx,
				definitionRef,
				preparedClosure,
				inputs,
				&user,
				selection.maxElementExecutions,
			)
		}
		finishRuntimeFlowStage(txCtx, "transaction.execute-flow", txStageStarted, executeErr)
		if executeErr != nil {
			return executeErr
		}
		if outcome == nil || outcome.Result == nil {
			return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		}
		if !outcome.Result.IsSuccess {
			// Catalog-action error details are authorized output just like success
			// data. Re-prove the admitted Permission State floor after execution and
			// before constructing the presentation sentinel so a concurrent
			// revocation rolls back without disclosing stale-authority details.
			if err := fencePermissionFloor(); err != nil {
				return err
			}
			presented, presentable, presentationErr := presentFlowOutcomeFailure(outcome, receipts)
			if presentationErr != nil {
				return &flowPresentationRollback{failure: presentationErr}
			}
			if presentable {
				return &flowPresentationRollback{response: &presented}
			}
			return &flowPresentationRollback{failure: classifyFlowOutcomeFailure(outcome)}
		}
		outputPayload, marshalErr := marshalPlatformCanonicalJSON(outcome.Result.OutputValues)
		if marshalErr != nil || outcome.ExecutionID == "" {
			return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		}
		response = flowsemantic.ResponseV2{
			ExecutionID:      outcome.ExecutionID,
			CanonicalOutputs: outputPayload,
			Receipts:         receipts,
		}
		if err := flowsemantic.ValidateResponseV2(response); err != nil {
			return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		}
		if err := txCtx.Err(); err != nil {
			return err
		}
		if err := fencePermissionFloor(); err != nil {
			return err
		}
		resultPayload, marshalErr := json.Marshal(response)
		if marshalErr != nil {
			return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		}
		txStageStarted = startRuntimeFlowStage(txCtx)
		completeErr := s.store.Complete(txCtx, row.RowID, resultPayload)
		finishRuntimeFlowStage(txCtx, "transaction.complete-invocation", txStageStarted, completeErr)
		return completeErr
	})
	finishRuntimeFlowStage(ctx, "transaction.total", stageStarted, err)
	if err != nil {
		if ctxErr := ctx.Err(); ctxErr != nil {
			return nil, flowStatus(ctxErr)
		}
		if presentation, ok := flowPresentationAfterRollback(completion, err); ok {
			if presentation.response == nil {
				return nil, flowStatus(presentation.failure)
			}
			response = *presentation.response
		} else {
			var failure *commonsemantic.Failure
			if !errors.As(err, &failure) {
				err = classifyPlatformEngineError(
					ctx, err, commonsemantic.InvalidComponentCommandData,
				)
			}
			return nil, flowStatus(err)
		}
	}
	stageStarted = startRuntimeFlowStage(ctx)
	wireResponse, err := mapping.ToWireExecuteFlowResponseV2(response)
	finishRuntimeFlowStage(ctx, "encode-response", stageStarted, err)
	if err != nil {
		return nil, flowStatus(err)
	}
	finishRuntimeFlowStage(ctx, "request.total", totalStarted, nil)
	return wireResponse, nil
}

func flowPresentationAfterRollback(
	completion ports.TransactionCompletionOutcome,
	err error,
) (*flowPresentationRollback, bool) {
	if completion != ports.TransactionCompletionRolledBack || err == nil || apperrors.IsInternal(err) {
		return nil, false
	}
	// A joined chain can pair a presentation sentinel with an independent
	// transaction failure. Error ancestry alone must not suppress that failure.
	for current := err; current != nil; current = errors.Unwrap(current) {
		if _, joined := current.(interface{ Unwrap() []error }); joined {
			return nil, false
		}
	}
	var presentation *flowPresentationRollback
	if !errors.As(err, &presentation) || presentation == nil {
		return nil, false
	}
	return presentation, true
}

func logFlowCatalogActionPreflightFailure(
	stage string,
	binding *flowsemantic.CatalogActionBinding,
	definitionRef string,
	err error,
) {
	componentRef, actionRef := "", ""
	if binding != nil {
		componentRef = binding.ComponentRef
		actionRef = binding.ActionRef
	}
	errorClass := "other"
	switch {
	case errors.Is(err, context.Canceled):
		errorClass = "canceled"
	case errors.Is(err, context.DeadlineExceeded):
		errorClass = "deadline"
	case errors.Is(err, flowexecution.ErrNotBoundedAtomic):
		errorClass = "not-bounded-atomic"
	default:
		var failure *commonsemantic.Failure
		if errors.As(err, &failure) {
			errorClass = "classified-contract-failure"
		}
	}
	log.Printf(
		"runtime Flow catalog action preflight failed: stage=%q component_ref=%q action_ref=%q definition_ref=%q error_class=%q",
		stage, componentRef, actionRef, definitionRef, errorClass,
	)
}

type flowAppMetadataView struct {
	OrgID               string
	AppInstallationID   string
	Lineage             catalogsemantic.ViewLineage
	AppComponents       []catalogsemantic.AppMetadataComponent
	EffectiveComponents []catalogsemantic.EffectiveMetadataComponent
	Receipts            catalogsemantic.Receipts
}

func (s *FlowExecutionServer) flowAppMetadataAuthority(
	ctx context.Context,
	scope RuntimeExecutionScope,
	runtimeView *catalogsnapshot.RuntimeMetadataView,
	trusted TrustedCall,
) (flowAppMetadataView, error) {
	if prepared, ok := runtimeFlowAppMetadataFromContext(ctx); ok {
		if err := validateFlowAppMetadataAuthority(scope, runtimeView, prepared); err != nil {
			return flowAppMetadataView{}, err
		}
		return prepared, nil
	}
	minimum := scope.receipts.CatalogRevision
	catalogScope, catalogCapability, err :=
		s.catalogAdmission.flowAppMetadataScope(trusted)
	if err != nil {
		return flowAppMetadataView{}, err
	}
	request := catalogsemantic.GetRequest{MinCatalogRevision: &minimum}
	var normalized flowAppMetadataView
	switch catalogCapability {
	case catalogsemantic.AppMetadataV3CapabilityName:
		engine, ok := s.catalog.(appMetadataViewV3Engine)
		if !ok {
			return normalized, &recordsemantic.Failure{Category: recordsemantic.FailureUnavailable}
		}
		view, err := engine.GetAppMetadataViewV3(ctx, catalogScope, request)
		if err != nil {
			return normalized, classifyPlatformEngineError(
				ctx, err, commonsemantic.InvalidComponentCommandData,
			)
		}
		if catalogsemantic.ValidateAppMetadataViewV3(view) != nil {
			return normalized, &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		}
		normalized = normalizeFlowAppMetadataV3(view)
	case catalogsemantic.AppMetadataV2CapabilityName:
		view, err := s.catalog.GetAppMetadataViewV2(ctx, catalogScope, request)
		if err != nil {
			return normalized, classifyPlatformEngineError(
				ctx, err, commonsemantic.InvalidComponentCommandData,
			)
		}
		if catalogsemantic.ValidateAppMetadataViewV2(view) != nil {
			return normalized, &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
		}
		normalized = normalizeFlowAppMetadataV2(view)
	default:
		return normalized, &recordsemantic.Failure{Category: recordsemantic.FailureUnavailable}
	}
	if err := validateFlowAppMetadataAuthority(scope, runtimeView, normalized); err != nil {
		return flowAppMetadataView{}, err
	}
	return normalized, nil
}

func normalizeFlowAppMetadataV2(view catalogsemantic.AppMetadataViewV2) flowAppMetadataView {
	return flowAppMetadataView{
		OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
		Lineage: view.Lineage, AppComponents: view.AppComponents,
		EffectiveComponents: view.EffectiveComponents, Receipts: view.Receipts,
	}
}

func normalizeFlowAppMetadataV3(view catalogsemantic.AppMetadataViewV3) flowAppMetadataView {
	return flowAppMetadataView{
		OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
		Lineage: view.Lineage, AppComponents: view.AppComponents,
		EffectiveComponents: view.EffectiveComponents, Receipts: view.Receipts,
	}
}

func validateFlowAppMetadataAuthority(
	scope RuntimeExecutionScope,
	runtimeView *catalogsnapshot.RuntimeMetadataView,
	appView flowAppMetadataView,
) error {
	if runtimeView == nil ||
		appView.OrgID != scope.orgID ||
		appView.AppInstallationID != scope.appInstallationID ||
		appView.Lineage.CatalogRevision != scope.receipts.CatalogRevision ||
		appView.Lineage.SchemaGeneration != scope.receipts.SchemaGeneration ||
		appView.Lineage.EngineGeneration != scope.receipts.EngineGeneration ||
		appView.Lineage.SourceSnapshotDigest != string(runtimeView.SourceSnapshotDigest()) ||
		appView.Receipts.CatalogRevision != scope.receipts.CatalogRevision ||
		appView.Receipts.SchemaGeneration != scope.receipts.SchemaGeneration ||
		appView.Receipts.EngineGeneration != scope.receipts.EngineGeneration ||
		appView.Receipts.SecurityEpoch != scope.receipts.SecurityEpoch {
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	return nil
}

func selectRequestedFlowAppMetadataComponent(
	components []catalogsemantic.AppMetadataComponent,
	definitionRef string,
	definitionVersion string,
) (catalogsemantic.AppMetadataComponent, error) {
	var selected catalogsemantic.AppMetadataComponent
	found := 0
	for _, component := range components {
		if component.ComponentKind != appMetadataStandardFlowKind ||
			component.BusinessKey != definitionRef ||
			component.DefinitionVersion != definitionVersion {
			continue
		}
		selected = component
		found++
	}
	if found == 0 {
		return catalogsemantic.AppMetadataComponent{}, &recordsemantic.Failure{
			Category: recordsemantic.FailureMissing,
		}
	}
	if found != 1 {
		return catalogsemantic.AppMetadataComponent{}, &recordsemantic.Failure{
			Category: recordsemantic.FailureInternal,
		}
	}
	return selected, nil
}

func selectCatalogActionFlowAppMetadataComponent(
	components []catalogsemantic.AppMetadataComponent,
	definitionRef string,
) (catalogsemantic.AppMetadataComponent, error) {
	var selected catalogsemantic.AppMetadataComponent
	found := 0
	for _, component := range components {
		if component.ComponentKind != appMetadataStandardFlowKind ||
			component.BusinessKey != definitionRef {
			continue
		}
		selected = component
		found++
	}
	if found == 0 {
		return catalogsemantic.AppMetadataComponent{}, &recordsemantic.Failure{
			Category: recordsemantic.FailureMissing,
		}
	}
	if found != 1 || selected.DefinitionVersion == "" {
		return catalogsemantic.AppMetadataComponent{}, &recordsemantic.Failure{
			Category: recordsemantic.FailureInternal,
		}
	}
	return selected, nil
}

func authorizeFlowAppMetadataClosure(
	components []catalogsemantic.AppMetadataComponent,
	effectiveComponents []catalogsemantic.EffectiveMetadataComponent,
	versions []boundedAtomicFlowVersion,
	rootDefinitionRef string,
	rootDefinitionVersion string,
) (flowAppMetadataAuthority, error) {
	var root flowAppMetadataAuthority
	for _, selected := range versions {
		found := 0
		var matched catalogsemantic.AppMetadataComponent
		for _, component := range components {
			if component.ComponentKind != appMetadataStandardFlowKind ||
				component.BusinessKey != selected.definitionRef {
				continue
			}
			matched = component
			found++
		}
		if found == 0 {
			return flowAppMetadataAuthority{}, &recordsemantic.Failure{
				Category: recordsemantic.FailureMissing,
			}
		}
		authority, matches := appMetadataComponentMatchesRuntimeFlow(
			matched, effectiveComponents, selected.version,
		)
		if found != 1 || !matches {
			return flowAppMetadataAuthority{}, &recordsemantic.Failure{
				Category: recordsemantic.FailureInternal,
			}
		}
		if selected.definitionRef == rootDefinitionRef &&
			rootDefinitionVersion != "" &&
			authority.definitionVersion != rootDefinitionVersion {
			return flowAppMetadataAuthority{}, &recordsemantic.Failure{
				Category: recordsemantic.FailureMissing,
			}
		}
		if selected.definitionRef == rootDefinitionRef {
			root = authority
		}
	}
	if root.appComponent.ComponentID == "" {
		return flowAppMetadataAuthority{}, &recordsemantic.Failure{
			Category: recordsemantic.FailureInternal,
		}
	}
	return root, nil
}

type flowAppMetadataAuthority struct {
	appComponent       catalogsemantic.AppMetadataComponent
	effectiveComponent *catalogsemantic.EffectiveMetadataComponent
	definitionVersion  string
}

func appMetadataComponentMatchesRuntimeFlow(
	component catalogsemantic.AppMetadataComponent,
	effectiveComponents []catalogsemantic.EffectiveMetadataComponent,
	version *models.FlowVersion,
) (flowAppMetadataAuthority, bool) {
	authority := flowAppMetadataAuthority{
		appComponent:      component,
		definitionVersion: component.DefinitionVersion,
	}
	if version == nil || len(component.CanonicalBehavior) == 0 {
		return flowAppMetadataAuthority{}, false
	}
	digest, err := appauthority.ComponentDigest(
		component.ComponentKind,
		component.ComponentFormatVersion,
		component.CanonicalBehavior,
	)
	if err != nil || digest != component.ContentDigest {
		return flowAppMetadataAuthority{}, false
	}
	publishedBehavior := component.CanonicalBehavior
	foundEffective := 0
	for index := range effectiveComponents {
		effective := &effectiveComponents[index]
		if effective.ComponentKind != catalogsemantic.EffectiveStandardFlowComponentKind ||
			effective.BusinessKey != component.BusinessKey {
			continue
		}
		foundEffective++
		document, err := catalogsemantic.DecodeEffectiveStandardFlowDocument(
			effective.CanonicalBehavior,
		)
		if err != nil {
			return flowAppMetadataAuthority{}, false
		}
		authority.effectiveComponent = effective
		authority.definitionVersion = document.DefinitionVersion
		publishedBehavior = document.CanonicalBehavior
	}
	if foundEffective > 1 {
		return flowAppMetadataAuthority{}, false
	}
	var published models.FlowMetadata
	decoder := json.NewDecoder(bytes.NewReader(publishedBehavior))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&published); err != nil {
		return flowAppMetadataAuthority{}, false
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		return flowAppMetadataAuthority{}, false
	}
	publishedBytes, err := json.Marshal(published)
	if err != nil {
		return flowAppMetadataAuthority{}, false
	}
	runtimeBytes, err := json.Marshal(version.Metadata)
	if err != nil {
		return flowAppMetadataAuthority{}, false
	}
	if authority.definitionVersion != strconv.Itoa(version.VersionNumber) ||
		!bytes.Equal(publishedBytes, runtimeBytes) {
		return flowAppMetadataAuthority{}, false
	}
	return authority, true
}

func decodeFlowInputs(value []byte) (map[string]interface{}, error) {
	inputs := map[string]interface{}{}
	decoder := json.NewDecoder(bytes.NewReader(value))
	decoder.UseNumber()
	if err := decoder.Decode(&inputs); err != nil {
		return nil, &recordsemantic.Failure{
			Category:         recordsemantic.FailureInvalidRequest,
			InvalidComponent: commonsemantic.InvalidComponentCommandData,
		}
	}
	var trailing interface{}
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		return nil, &recordsemantic.Failure{
			Category:         recordsemantic.FailureInvalidRequest,
			InvalidComponent: commonsemantic.InvalidComponentCommandData,
		}
	}
	return inputs, nil
}

func marshalPlatformCanonicalJSON(value any) ([]byte, error) {
	if value == nil {
		value = map[string]interface{}{}
	}
	encoded, err := json.Marshal(value)
	if err != nil {
		return nil, err
	}
	canonical, err := mutationsemantic.CanonicalSourceJSON(string(encoded))
	if err != nil {
		return nil, err
	}
	return []byte(canonical), nil
}

func classifyFlowInvocationError(
	ctx context.Context,
	err error,
	row persistence.RuntimeFlowInvocationRow,
	receipts flowsemantic.Receipts,
) error {
	switch {
	case ctx.Err() != nil:
		return ctx.Err()
	case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
		return err
	case errors.Is(err, persistence.ErrRuntimeFlowInvocationConflict):
		return &recordsemantic.Failure{
			Category:         recordsemantic.FailureInvalidRequest,
			InvalidComponent: commonsemantic.InvalidComponentCommandData,
		}
	case errors.Is(err, persistence.ErrRuntimeFlowInvocationServingBindingChanged):
		return servingBindingFailure(row.Receipts, receipts)
	default:
		return err
	}
}

func classifyFlowOutcomeFailure(outcome *contractmodels.FlowExecutionTypedOutcome) error {
	if outcome == nil || outcome.Cause == nil {
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
	switch outcome.Cause.Category {
	case contractmodels.FlowExecutionCauseCategoryValidation:
		return &recordsemantic.Failure{
			Category:         recordsemantic.FailureInvalidRequest,
			InvalidComponent: commonsemantic.InvalidComponentCommandData,
		}
	case contractmodels.FlowExecutionCauseCategoryAuthorization:
		return &recordsemantic.Failure{Category: recordsemantic.FailureForbidden}
	case contractmodels.FlowExecutionCauseCategoryNotFound:
		return &recordsemantic.Failure{Category: recordsemantic.FailureMissing}
	case contractmodels.FlowExecutionCauseCategoryConcurrency:
		return &recordsemantic.Failure{Category: recordsemantic.FailureUnavailable}
	default:
		if outcome.Cause.Retriable {
			return &recordsemantic.Failure{Category: recordsemantic.FailureUnavailable}
		}
		return &recordsemantic.Failure{Category: recordsemantic.FailureInternal}
	}
}

func presentFlowOutcomeFailure(
	outcome *contractmodels.FlowExecutionTypedOutcome,
	receipts flowsemantic.Receipts,
) (flowsemantic.ResponseV2, bool, error) {
	if outcome == nil || outcome.Result == nil || outcome.Result.IsSuccess ||
		!outcome.Result.IsCustomError || len(outcome.Result.Errors) == 0 {
		return flowsemantic.ResponseV2{}, false, nil
	}
	presentedErrors := make([]flowsemantic.ExecutionError, len(outcome.Result.Errors))
	for index, executionError := range outcome.Result.Errors {
		presentedErrors[index] = flowsemantic.ExecutionError{
			Code:    executionError.StatusCode,
			Message: executionError.Message,
			Fields:  append([]string(nil), executionError.Fields...),
		}
	}
	response := flowsemantic.ResponseV2{
		ExecutionID: outcome.ExecutionID,
		Errors:      presentedErrors,
		Receipts:    receipts,
	}
	if err := flowsemantic.ValidateResponseV2(response); err != nil {
		return flowsemantic.ResponseV2{}, false, &recordsemantic.Failure{
			Category: recordsemantic.FailureInternal,
		}
	}
	return response, true, nil
}

func flowStatus(err error) error {
	return mapping.ToPlatformStatusError(err, flowsemantic.CapabilityName)
}

var _ runtimev1.FlowExecutionServiceServer = (*FlowExecutionServer)(nil)