package catalogreconcile

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/approvalir"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	layoutauthoringv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
)

const (
	OutputSchemaV1 = "nexuscrm.catalog-reconcile-output/v1"
	OutputSchemaV2 = "nexuscrm.catalog-reconcile-output/v2"
	OutputSchemaV3 = "nexuscrm.catalog-reconcile-output/v3"

	continuationIntentSchemaV1 = "nexuscrm.catalog-reconcile-continuation-intent/v1"
	continuationIntentSchemaV2 = "nexuscrm.catalog-reconcile-continuation-intent/v2"
	continuationIntentSchemaV3 = "nexuscrm.catalog-reconcile-continuation-intent/v3"
)

type reconcileProfile struct {
	outputSchema             string
	continuationIntentSchema string
	multipleCapabilities     bool
}

// Authority is the existing Catalog App owner surface used by reconcile. No
// Catalog table or parallel release/install registry is exposed here.
type Authority interface {
	ReadRelease(context.Context, string, string) (catalogapp.RegisteredRelease, error)
	BeginOperation(context.Context, catalogapp.BeginOperationRequest) (catalogapp.Operation, error)
	CommitOperationMetadataSelection(context.Context, string, string) (string, error)
	CommitOperationMetadataSelectionWithEffectiveComponents(
		context.Context,
		string,
		string,
		[]catalogapp.EffectiveMetadataComponent,
	) (string, error)
	CommitServingArtifacts(context.Context, string) (string, error)
	ReadOperation(context.Context, string) (catalogapp.Operation, error)
	ReadOperationConsumerGrants(context.Context, string) ([]catalogapp.ConsumerGrant, error)
	ReadOperationByIdempotency(context.Context, string, string, string) (catalogapp.Operation, error)
	ReadOperationMetadataSelection(context.Context, string) (string, error)
	ReadOperationHandlerBinding(context.Context, string, string) (catalogapp.HandlerBinding, error)
	ReadOperationFrontendBinding(context.Context, string) (catalogapp.FrontendBinding, error)
	ReadLatestHandlerOwnershipEpoch(context.Context, string, string) (uint64, error)
	ReadActiveFrontendBinding(context.Context, string) (catalogapp.FrontendBinding, error)
	StageHandlerBinding(context.Context, catalogapp.HandlerBindingRequest) (catalogapp.HandlerBinding, error)
	StageFrontendBinding(context.Context, catalogapp.FrontendBindingRequest) (catalogapp.FrontendBinding, error)
	PublishRuntimeSecurityProjection(context.Context, string, catalogapp.RuntimeProjectionWriter) error
	CompleteOperation(context.Context, string, *catalogapp.RemovalGateReceipt) (catalogapp.Operation, error)
}

type Request struct {
	handlerWorkloadCapabilities []string

	Operation            catalogapp.BeginOperationRequest
	Capabilities         []CapabilitySelection
	AppCapabilityID      string
	HandlerRef           string
	RuntimeCapabilityID  string
	WorkloadID           string
	ServiceReleaseDigest catalogsnapshot.Digest
	MetadataSourceDigest catalogsnapshot.Digest
	Frontend             *FrontendSelection
	SchemaGeneration     uint64
	EngineGeneration     uint64
	approvalPublication  *approvalPublicationSelection
	outboundMailRoutes   []OutboundMailRouteConfig
	capabilityClosure    string
	profile              reconcileProfile
}

// approvalPublicationSelection keeps the immutable identity and the decoded
// serving input together. A publication input must never be available to the
// artifact compiler without the same identity entering the durable
// continuation intent.
type approvalPublicationSelection struct {
	bundle *approvalir.PublicationBundle
	digest catalogsnapshot.Digest
}

type CapabilitySelection struct {
	AppCapabilityID     string `json:"appCapabilityId"`
	HandlerRef          string `json:"handlerRef"`
	RuntimeCapabilityID string `json:"runtimeCapabilityId"`
}

type FrontendSelection struct {
	ReleaseDigest              string `json:"releaseDigest"`
	CompatibilityReceiptDigest string `json:"compatibilityReceiptDigest"`
}

// Output is the closed handoff consumed by L2 deployment and Runtime exact
// configuration. It contains identities only, never source metadata payloads.
type Output struct {
	Schema                     string             `json:"schema"`
	OperationID                string             `json:"operationId"`
	OrgID                      string             `json:"orgId"`
	AppInstallationID          string             `json:"appInstallationId"`
	AppReleaseDigest           string             `json:"appReleaseDigest"`
	AppCapabilityID            string             `json:"appCapabilityId,omitempty"`
	RuntimeCapabilityID        string             `json:"runtimeCapabilityId,omitempty"`
	HandlerBindingID           string             `json:"handlerBindingId,omitempty"`
	WorkloadID                 string             `json:"workloadId"`
	ServiceReleaseDigest       string             `json:"serviceReleaseDigest"`
	MetadataSourceDigest       string             `json:"metadataSourceDigest,omitempty"`
	OwnershipEpoch             uint64             `json:"ownershipEpoch,omitempty"`
	CatalogRevision            uint64             `json:"catalogRevision"`
	SchemaGeneration           uint64             `json:"schemaGeneration"`
	EngineGeneration           uint64             `json:"engineGeneration"`
	SecurityEpoch              uint64             `json:"securityEpoch"`
	CatalogSnapshotDigest      string             `json:"catalogSnapshotDigest"`
	RuntimeMetadataViewDigest  string             `json:"runtimeMetadataViewDigest,omitempty"`
	CompatibilityReceiptDigest string             `json:"compatibilityReceiptDigest,omitempty"`
	FrontendBindingID          string             `json:"frontendBindingId,omitempty"`
	FrontendReleaseDigest      string             `json:"frontendReleaseDigest,omitempty"`
	FrontendReceiptDigest      string             `json:"frontendCompatibilityReceiptDigest,omitempty"`
	Capabilities               []CapabilityOutput `json:"capabilities,omitempty"`
	// LayoutAuthoringViewRef is the in-process publication result consumed by
	// Runtime recovery. The versioned L2 JSON handoff remains OutputSchemaV3;
	// extending that closed schema is a separate release-boundary decision.
	LayoutAuthoringViewRef *layoutauthoringv1.CatalogViewRef `json:"-"`
}

type CapabilityOutput struct {
	AppCapabilityID            string `json:"appCapabilityId"`
	RuntimeCapabilityID        string `json:"runtimeCapabilityId"`
	HandlerBindingID           string `json:"handlerBindingId"`
	OwnershipEpoch             uint64 `json:"ownershipEpoch"`
	RuntimeMetadataViewDigest  string `json:"runtimeMetadataViewDigest"`
	CompatibilityReceiptDigest string `json:"compatibilityReceiptDigest"`
}

// Reconcile prepares and atomically activates one complete Catalog serving
// selection. Immutable artifacts are durable before the binding points at
// them; Runtime security projection is acknowledged before Catalog activation.
func Reconcile(
	ctx context.Context,
	authority Authority,
	writer catalogapp.RuntimeProjectionWriter,
	store catalogsnapshot.Store,
	withinTransaction InTransaction,
	reader MetadataReader,
	request Request,
) (Output, error) {
	if authority == nil {
		return Output{}, fmt.Errorf("Catalog App authority is required")
	}
	if writer == nil {
		return Output{}, fmt.Errorf("Runtime security projection writer is required")
	}
	if store == nil {
		return Output{}, fmt.Errorf("Catalog snapshot store is required")
	}
	request, err := bindContinuationIntent(request)
	if err != nil {
		return Output{}, err
	}
	release, err := authority.ReadRelease(ctx, request.Operation.AppID, request.Operation.TargetAppReleaseDigest)
	if err != nil {
		return Output{}, fmt.Errorf("read verified AppRelease before Catalog reconcile: %w", err)
	}
	if err := validateRequestAgainstRelease(request, release); err != nil {
		return Output{}, err
	}
	if err := validateApprovalDefinitionAvailability(request, &release.Package); err != nil {
		return Output{}, err
	}
	servingErr := validateRecordViewServingSelection(request, release)
	operation, err := beginOperationForServingSelection(ctx, authority, request.Operation, servingErr)
	if err != nil {
		return Output{}, fmt.Errorf("begin Catalog operation: %w", err)
	}
	request.Operation.ConsumerGrants, err = authority.ReadOperationConsumerGrants(
		ctx, operation.OperationID,
	)
	if err != nil {
		return Output{}, fmt.Errorf("read admitted Catalog consumer grants: %w", err)
	}
	initial := operation
	var lastErr error
	for attempt := 0; attempt < 16; attempt++ {
		output, err := reconcileObservedOperation(ctx, authority, writer, store, withinTransaction, reader, &release.Package, request, operation)
		if err == nil {
			return output, nil
		}
		lastErr = err
		if ctx.Err() != nil {
			return Output{}, errors.Join(ctx.Err(), err)
		}
		refreshed, readErr := authority.ReadOperation(ctx, operation.OperationID)
		if readErr != nil {
			return Output{}, fmt.Errorf("refresh Catalog operation after reconcile error: %w", errors.Join(err, readErr))
		}
		if !sameOperationFence(initial, refreshed) {
			return Output{}, fmt.Errorf("%w: Catalog operation identity changed during reconcile", catalogapp.ErrConflict)
		}
		if refreshed.Step == "ACTIVATED" {
			return outputForActiveOperation(ctx, authority, store, refreshed, request)
		}
		if refreshed.State == catalogapp.StateFailed || refreshed.State == catalogapp.StateRemoved {
			return Output{}, err
		}
		if refreshed.Step != operation.Step {
			operation = refreshed
			continue
		}
		return Output{}, err
	}
	return Output{}, fmt.Errorf("Catalog reconcile did not converge after concurrent transitions: %w", lastErr)
}

func validateRequestAgainstRelease(request Request, release catalogapp.RegisteredRelease) error {
	if release.AppID != request.Operation.AppID || release.AppReleaseDigest != request.Operation.TargetAppReleaseDigest {
		return fmt.Errorf("Catalog reconcile request does not match the verified AppRelease identity")
	}
	manifest := release.Package.Manifest
	if request.capabilityClosure == ConfigSchemaV4 {
		if err := validateReleaseCapabilityClosureV4(
			manifest.ProvidedCapabilities,
			manifest.RequiredEngineCapabilities,
			request.Capabilities, request.handlerWorkloadCapabilities...,
		); err != nil {
			return err
		}
	} else {
		if err := validateReleaseCapabilitySelection(
			manifest.ProvidedCapabilities,
			manifest.RequiredEngineCapabilities,
			request.Capabilities,
		); err != nil {
			return err
		}
	}
	switch manifest.FrontendCompatibility.Mode {
	case appauthority.FrontendModeRequired:
		if request.Frontend == nil {
			return fmt.Errorf("Catalog reconcile verified AppRelease requires a frontend binding")
		}
	case appauthority.FrontendModeNone:
		if request.Frontend != nil {
			return fmt.Errorf("Catalog reconcile verified AppRelease does not accept a frontend binding")
		}
	default:
		return fmt.Errorf("Catalog reconcile AppRelease frontend mode %q is unsupported", manifest.FrontendCompatibility.Mode)
	}
	return nil
}

// beginOperationForServingSelection lets historical releases replay their
// durable Catalog result without allowing an unsupported new serving
// activation. Catalog owns the unique operation idempotency key, so Reconcile
// first asks that owner whether the key already exists. BeginOperation still
// validates the full request digest and is the only operation creator.
func beginOperationForServingSelection(
	ctx context.Context,
	authority Authority,
	request catalogapp.BeginOperationRequest,
	servingErr error,
) (catalogapp.Operation, error) {
	if servingErr == nil {
		return authority.BeginOperation(ctx, request)
	}
	_, err := authority.ReadOperationByIdempotency(ctx, request.OrgID, request.AppID, request.IdempotencyKey)
	if errors.Is(err, catalogapp.ErrNotFound) {
		return catalogapp.Operation{}, servingErr
	}
	if err != nil {
		return catalogapp.Operation{}, fmt.Errorf("read Catalog operation for serving eligibility: %w", err)
	}
	operation, err := authority.BeginOperation(ctx, request)
	if err != nil {
		return catalogapp.Operation{}, err
	}
	if operation.Step != "ACTIVATED" {
		return catalogapp.Operation{}, servingErr
	}
	return operation, nil
}

func validateRecordViewServingSelection(request Request, release catalogapp.RegisteredRelease) error {
	if !selectionIncludesRecordViewAttachmentFacts(MetadataSelection{RuntimeCapabilityIDs: runtimeCapabilityIDs(request.Capabilities)}) {
		return nil
	}
	document, err := readRecordViewReleaseDocument(&release.Package)
	if err != nil {
		return fmt.Errorf("read record-view AppRelease serving facts: %w", err)
	}
	if err := validateRecordViewServingDocument(document); err != nil {
		return fmt.Errorf("record-view AppRelease cannot activate a serving capability: %w", err)
	}
	return nil
}

func reconcileObservedOperation(
	ctx context.Context,
	authority Authority,
	writer catalogapp.RuntimeProjectionWriter,
	store catalogsnapshot.Store,
	withinTransaction InTransaction,
	reader MetadataReader,
	release *appauthority.VerifiedPackage,
	request Request,
	operation catalogapp.Operation,
) (Output, error) {
	if operation.Step == "ACTIVATED" {
		return outputForActiveOperation(ctx, authority, store, operation, request)
	}
	if operation.Step != "CANDIDATE_COMMITTED" && operation.Step != catalogapp.OperationStepServingArtifactsCommitted && operation.Step != "PROJECTION_ACKNOWLEDGED" {
		return Output{}, fmt.Errorf("Catalog operation step %q cannot be reconciled", operation.Step)
	}

	states := make([]capabilityReconcileState, 0, len(request.Capabilities))
	committedSnapshot, selectionErr := authority.ReadOperationMetadataSelection(ctx, operation.OperationID)
	if selectionErr != nil && !errors.Is(selectionErr, catalogapp.ErrNotFound) {
		return Output{}, fmt.Errorf("read operation metadata selection: %w", selectionErr)
	}
	for _, capability := range request.Capabilities {
		state, selectedSnapshot, err := reconcileCapability(ctx, authority, store, withinTransaction, reader, release,
			operation, request, capability, committedSnapshot, operation.Step == "CANDIDATE_COMMITTED")
		if err != nil {
			return Output{}, err
		}
		if committedSnapshot == "" {
			committedSnapshot = selectedSnapshot
		}
		states = append(states, state)
	}
	if err := validateCapabilityArtifactSet(states); err != nil {
		return Output{}, err
	}
	consumerViews, err := storeConsumerOnlyRuntimeViews(
		ctx,
		store,
		states[0].artifacts.CatalogSnapshotDigest,
		operation,
		request,
	)
	if err != nil {
		return Output{}, err
	}
	if operation.Step == "CANDIDATE_COMMITTED" {
		if request.Frontend != nil {
			if _, err := authority.StageFrontendBinding(ctx, catalogapp.FrontendBindingRequest{
				OperationID: operation.OperationID, FrontendReleaseDigest: request.Frontend.ReleaseDigest,
				CompatibilityReceiptDigest: request.Frontend.CompatibilityReceiptDigest,
			}); err != nil {
				return Output{}, fmt.Errorf("stage Catalog FrontendBinding: %w", err)
			}
		}
	}
	if _, err := authority.CommitServingArtifacts(ctx, operation.OperationID); err != nil {
		return Output{}, fmt.Errorf("commit complete serving selection: %w", err)
	}
	if err := authority.PublishRuntimeSecurityProjection(ctx, operation.OperationID, writer); err != nil {
		return Output{}, fmt.Errorf("publish Runtime security projection: %w", err)
	}
	completed, err := authority.CompleteOperation(ctx, operation.OperationID, nil)
	if err != nil {
		return Output{}, fmt.Errorf("complete Catalog operation: %w", err)
	}
	if completed.Step != "ACTIVATED" {
		return Output{}, fmt.Errorf("Catalog operation completed at unexpected step %q", completed.Step)
	}
	for i := range states {
		binding, err := authority.ReadOperationHandlerBinding(ctx, completed.OperationID, states[i].selection.AppCapabilityID)
		if err != nil {
			return Output{}, fmt.Errorf("read activated operation capability owner: %w", err)
		}
		if err := validateReconcileBinding(binding, completed, request, states[i].selection, states[i].binding.OwnershipEpoch,
			catalogapp.BindingActive, catalogapp.BindingDraining, catalogapp.BindingRetired); err != nil {
			return Output{}, err
		}
		states[i].binding = binding
	}
	output := Output{
		Schema: request.profile.outputSchema, OperationID: completed.OperationID,
		OrgID: completed.OrgID, AppInstallationID: completed.AppInstallationID,
		AppReleaseDigest: completed.TargetAppReleaseDigest, WorkloadID: request.WorkloadID,
		ServiceReleaseDigest: string(request.ServiceReleaseDigest),
		MetadataSourceDigest: string(request.MetadataSourceDigest),
		CatalogRevision:      completed.TargetCatalogRevision, SchemaGeneration: request.SchemaGeneration,
		EngineGeneration: request.EngineGeneration, SecurityEpoch: completed.TargetSecurityEpoch,
		CatalogSnapshotDigest: string(states[0].artifacts.CatalogSnapshotDigest),
	}
	bindCapabilityOutput(&output, states)
	if err := bindLayoutAuthoringViewRef(&output, request, states, consumerViews); err != nil {
		return Output{}, err
	}
	if request.Frontend != nil {
		frontend, err := authority.ReadOperationFrontendBinding(ctx, completed.OperationID)
		if err != nil {
			return Output{}, fmt.Errorf("read activated frontend owner: %w", err)
		}
		if frontend.FrontendReleaseDigest != request.Frontend.ReleaseDigest || frontend.CompatibilityReceiptDigest != request.Frontend.CompatibilityReceiptDigest {
			return Output{}, fmt.Errorf("activated FrontendBinding disagrees with the continuation intent")
		}
		bindFrontendOutput(&output, frontend)
	}
	return output, nil
}

type capabilityReconcileState struct {
	selection CapabilitySelection
	binding   catalogapp.HandlerBinding
	artifacts ArtifactResult
}

func reconcileCapability(
	ctx context.Context,
	authority Authority,
	store catalogsnapshot.Store,
	withinTransaction InTransaction,
	reader MetadataReader,
	release *appauthority.VerifiedPackage,
	operation catalogapp.Operation,
	request Request,
	selection CapabilitySelection,
	committedSnapshot string,
	mayStage bool,
) (capabilityReconcileState, string, error) {
	binding, err := authority.ReadOperationHandlerBinding(ctx, operation.OperationID, selection.AppCapabilityID)
	if err == nil {
		if err := validateReconcileBinding(binding, operation, request, selection, binding.OwnershipEpoch, catalogapp.BindingStaged); err != nil {
			return capabilityReconcileState{}, committedSnapshot, err
		}
		artifacts, err := loadServingArtifacts(ctx, store, catalogsnapshot.Digest(binding.CompatibilityReceiptDigest), operation, request, selection, binding.OwnershipEpoch)
		return capabilityReconcileState{selection: selection, binding: binding, artifacts: artifacts}, committedSnapshot, err
	}
	if !errors.Is(err, catalogapp.ErrNotFound) {
		return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("read operation HandlerBinding: %w", err)
	}
	if !mayStage {
		return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("read committed operation HandlerBinding: %w", err)
	}
	previousEpoch, err := authority.ReadLatestHandlerOwnershipEpoch(ctx, operation.AppInstallationID, selection.AppCapabilityID)
	if errors.Is(err, catalogapp.ErrNotFound) {
		previousEpoch = 0
	} else if err != nil {
		return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("read latest capability ownership epoch: %w", err)
	}
	if previousEpoch == ^uint64(0) {
		return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("capability ownership epoch exhausted")
	}
	selectedEpoch := previousEpoch + 1
	metadataSelection := MetadataSelection{
		OrgID: operation.OrgID, AppInstallationID: operation.AppInstallationID,
		RuntimeCapabilityID:  selection.RuntimeCapabilityID,
		RuntimeCapabilityIDs: runtimeMetadataCapabilityIDs(request),
		IntegrationBindings: append(
			[]catalogapp.IntegrationBindingSelection(nil),
			request.Operation.IntegrationBindings...,
		),
		OutboundMailRoutes: append([]OutboundMailRouteConfig(nil), request.outboundMailRoutes...),
		CatalogRevision:    operation.TargetCatalogRevision,
		SchemaGeneration:   request.SchemaGeneration, EngineGeneration: request.EngineGeneration,
	}
	if request.approvalPublication != nil {
		metadataSelection.ApprovalDefinitions = make(
			[]catalogsnapshot.ApprovalDefinitionInput,
			0,
			len(request.approvalPublication.bundle.Definitions),
		)
		for _, definition := range request.approvalPublication.bundle.Definitions {
			envelope, marshalErr := json.Marshal(definition.Envelope)
			if marshalErr != nil {
				return capabilityReconcileState{}, committedSnapshot, fmt.Errorf(
					"encode approval publication definition %q: %w",
					definition.DefinitionID,
					marshalErr,
				)
			}
			metadataSelection.ApprovalDefinitions = append(
				metadataSelection.ApprovalDefinitions,
				catalogsnapshot.ApprovalDefinitionInput{
					CatalogRevision: catalogsnapshot.CatalogRevision(operation.TargetCatalogRevision),
					DefinitionID:    definition.DefinitionID, ObjectAPIName: definition.ObjectID,
					DeveloperName: definition.DefinitionName, Envelope: envelope,
				},
			)
		}
	}
	artifactSelection := ArtifactSelection{
		Metadata: metadataSelection, AppReleaseDigest: catalogsnapshot.Digest(operation.TargetAppReleaseDigest),
		AppCapabilityID: selection.AppCapabilityID, WorkloadID: request.WorkloadID,
		ServiceReleaseDigest: request.ServiceReleaseDigest, MetadataSourceDigest: request.MetadataSourceDigest,
		OwnershipEpoch: selectedEpoch, SecurityEpoch: operation.TargetSecurityEpoch,
	}
	var prepared preparedArtifacts
	if committedSnapshot == "" {
		effectiveMetadataCapability := selectedEffectiveMetadataCapability(request.Operation)
		includeEffectiveMetadata := effectiveMetadataCapability != ""
		effective, readErr := readEffectiveMetadata(
			ctx, withinTransaction, reader, metadataSelection, release,
			effectiveMetadataCapability,
		)
		if readErr != nil {
			return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("prepare committed operation metadata selection: %w", readErr)
		}
		prepared, err = prepareArtifacts(effective.metadata, artifactSelection)
		if err == nil {
			err = storePreparedCatalogSnapshot(ctx, store, prepared)
		}
		if err == nil {
			if includeEffectiveMetadata {
				committedSnapshot, err =
					authority.CommitOperationMetadataSelectionWithEffectiveComponents(
						ctx, operation.OperationID,
						string(prepared.result.CatalogSnapshotDigest),
						effective.components,
					)
			} else {
				committedSnapshot, err =
					authority.CommitOperationMetadataSelection(
						ctx, operation.OperationID,
						string(prepared.result.CatalogSnapshotDigest),
					)
			}
			if errors.Is(err, catalogapp.ErrConflict) {
				committedSnapshot, err = authority.ReadOperationMetadataSelection(ctx, operation.OperationID)
				if err == nil && includeEffectiveMetadata {
					_, err =
						authority.CommitOperationMetadataSelectionWithEffectiveComponents(
							ctx, operation.OperationID, committedSnapshot,
							effective.components,
						)
					if errors.Is(err, catalogapp.ErrConflict) &&
						len(effective.historicalRetryComponentSets) != 0 {
						// A rolling-upgrade predecessor may have won the same V3
						// operation with a historical object-search format. Keep
						// Catalog's exact-set verification and retry each fully
						// compiled predecessor candidate instead of weakening it.
						for _, historicalComponents := range effective.historicalRetryComponentSets {
							_, err =
								authority.CommitOperationMetadataSelectionWithEffectiveComponents(
									ctx, operation.OperationID, committedSnapshot,
									historicalComponents,
								)
							if !errors.Is(err, catalogapp.ErrConflict) {
								break
							}
						}
					}
				}
				if err == nil {
					prepared, err = prepareArtifactsFromCommittedSnapshot(ctx, store, catalogsnapshot.Digest(committedSnapshot), artifactSelection)
				}
			}
		}
	} else {
		prepared, err = prepareArtifactsFromCommittedSnapshot(ctx, store, catalogsnapshot.Digest(committedSnapshot), artifactSelection)
	}
	if err != nil {
		return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("prepare committed operation metadata selection: %w", err)
	}
	if err := storePreparedDerivedArtifacts(ctx, store, prepared); err != nil {
		return capabilityReconcileState{}, committedSnapshot, err
	}
	binding, err = authority.StageHandlerBinding(ctx, catalogapp.HandlerBindingRequest{
		OperationID: operation.OperationID, CapabilityID: selection.AppCapabilityID,
		HandlerRef: selection.HandlerRef, WorkloadID: request.WorkloadID,
		ServiceReleaseDigest: string(request.ServiceReleaseDigest), ExpectedOwnershipEpoch: previousEpoch,
		CompatibilityReceiptDigest: string(prepared.result.CompatibilityReceiptDigest),
	})
	if err != nil {
		return capabilityReconcileState{}, committedSnapshot, fmt.Errorf("stage Catalog HandlerBinding: %w", err)
	}
	return capabilityReconcileState{selection: selection, binding: binding, artifacts: prepared.result}, committedSnapshot, nil
}

// selectedEffectiveMetadataCapability chooses the newest complete metadata
// projection granted to the consumer. An AppRelease may retain V2 while adding
// V3 for compatibility; that must not silently downgrade V3 publication to
// V2's smaller, older response envelope.
func selectedEffectiveMetadataCapability(operation catalogapp.BeginOperationRequest) string {
	if requestHasConsumerCapability(operation, consumerv1.AppMetadataV3CapabilityName) {
		return consumerv1.AppMetadataV3CapabilityName
	}
	if requestHasConsumerCapability(operation, consumerv1.AppMetadataV2CapabilityName) {
		return consumerv1.AppMetadataV2CapabilityName
	}
	return ""
}

func requestHasConsumerCapability(
	request catalogapp.BeginOperationRequest,
	capability string,
) bool {
	for _, grant := range request.ConsumerGrants {
		if grant.CapabilityID == capability {
			return true
		}
	}
	return false
}

// storeConsumerOnlyRuntimeViews closes the immutable artifact chain for
// Runtime capabilities that are selected by an independent consumer rather
// than an L1 HandlerBinding. Handler-backed views are stored with their
// compatibility receipts in reconcileCapability. A consumer-only command
// view has no HandlerBinding receipt, but durable Runtime work still names its
// exact content address and Workers mount this store read-only.
func storeConsumerOnlyRuntimeViews(
	ctx context.Context,
	store catalogsnapshot.Store,
	snapshotDigest catalogsnapshot.Digest,
	operation catalogapp.Operation,
	request Request,
) (map[string]catalogsnapshot.Digest, error) {
	capabilityIDs := consumerOnlyRuntimeCapabilityIDs(request)
	results := make(map[string]catalogsnapshot.Digest, len(capabilityIDs))
	if len(capabilityIDs) == 0 {
		return results, nil
	}
	if store == nil {
		return nil, fmt.Errorf("Catalog snapshot store is required")
	}
	snapshotArtifact, err := store.Get(
		ctx,
		catalogsnapshot.ArtifactCatalogSnapshot,
		snapshotDigest,
	)
	if err != nil {
		return nil, fmt.Errorf(
			"read committed Catalog snapshot for consumer-only Runtime views: %w",
			err,
		)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		return nil, fmt.Errorf(
			"open committed Catalog snapshot for consumer-only Runtime views: %w",
			err,
		)
	}
	wantLineage := catalogsnapshot.Lineage{
		CatalogRevision: catalogsnapshot.CatalogRevision(
			operation.TargetCatalogRevision,
		),
		SchemaGeneration: catalogsnapshot.SchemaGeneration(
			request.SchemaGeneration,
		),
		EngineGeneration: catalogsnapshot.EngineGeneration(
			request.EngineGeneration,
		),
	}
	if snapshot.Digest() != snapshotDigest ||
		snapshot.OrgID() != operation.OrgID ||
		snapshot.Lineage() != wantLineage {
		return nil, fmt.Errorf(
			"consumer-only Runtime view snapshot disagrees with the Catalog operation",
		)
	}
	for _, capabilityID := range capabilityIDs {
		if _, duplicate := results[capabilityID]; duplicate {
			return nil, fmt.Errorf("consumer-only Runtime capability %q is duplicated", capabilityID)
		}
		viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
			snapshot,
			catalogsnapshot.ProjectionKey{
				AppInstallationID: operation.AppInstallationID,
				CapabilityID:      capabilityID,
			},
		)
		if err != nil {
			return nil, fmt.Errorf(
				"project consumer-only Runtime metadata view %q: %w",
				capabilityID,
				err,
			)
		}
		view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
		if err != nil {
			return nil, fmt.Errorf(
				"open consumer-only Runtime metadata view %q: %w",
				capabilityID,
				err,
			)
		}
		expectedProfile, err := projectionProfile(capabilityID)
		if err != nil {
			return nil, err
		}
		if err := view.VerifySource(snapshot); err != nil {
			return nil, fmt.Errorf(
				"verify consumer-only Runtime metadata view %q source: %w",
				capabilityID,
				err,
			)
		}
		if view.OrgID() != operation.OrgID ||
			view.AppInstallationID() != operation.AppInstallationID ||
			view.CapabilityID() != capabilityID ||
			view.Profile() != expectedProfile ||
			view.Lineage() != wantLineage {
			return nil, fmt.Errorf(
				"consumer-only Runtime metadata view %q disagrees with the Catalog operation",
				capabilityID,
			)
		}
		if err := store.Put(ctx, viewArtifact); err != nil {
			return nil, fmt.Errorf(
				"store consumer-only Runtime metadata view %q: %w",
				capabilityID,
				err,
			)
		}
		results[capabilityID] = viewArtifact.Digest()
	}
	return results, nil
}

func bindLayoutAuthoringViewRef(
	output *Output,
	request Request,
	states []capabilityReconcileState,
	consumerViews map[string]catalogsnapshot.Digest,
) error {
	selected := slices.Contains(
		runtimeMetadataCapabilityIDs(request),
		layoutauthoringv1.CapabilityName,
	)
	if !selected {
		return nil
	}
	digests := make([]catalogsnapshot.Digest, 0, 1)
	for _, state := range states {
		if state.selection.RuntimeCapabilityID == layoutauthoringv1.CapabilityName {
			digests = append(digests, state.artifacts.RuntimeMetadataViewDigest)
		}
	}
	if digest, ok := consumerViews[layoutauthoringv1.CapabilityName]; ok {
		digests = append(digests, digest)
	}
	if len(digests) != 1 {
		return fmt.Errorf("published Layout authoring Runtime view count is %d, want one", len(digests))
	}
	ref := layoutauthoringv1.CatalogViewRef{
		CatalogRevision:     output.CatalogRevision,
		AuthoringViewDigest: string(digests[0]),
	}
	if err := layoutauthoringv1.ValidateCatalogViewRef(ref); err != nil {
		return fmt.Errorf("published Layout authoring view ref is invalid: %w", err)
	}
	output.LayoutAuthoringViewRef = &ref
	return nil
}

func consumerOnlyRuntimeCapabilityIDs(request Request) []string {
	handlerIDs := runtimeCapabilityIDs(request.Capabilities)
	handlers := make(map[string]struct{}, len(handlerIDs))
	for _, capabilityID := range handlerIDs {
		handlers[capabilityID] = struct{}{}
	}
	selected := metadataCapabilityIDs(MetadataSelection{
		RuntimeCapabilityIDs: runtimeMetadataCapabilityIDs(request),
		IntegrationBindings: append(
			[]catalogapp.IntegrationBindingSelection(nil),
			request.Operation.IntegrationBindings...,
		),
		OutboundMailRoutes: append(
			[]OutboundMailRouteConfig(nil),
			request.outboundMailRoutes...,
		),
	})
	result := make([]string, 0, len(selected))
	for _, capabilityID := range selected {
		if _, handlerOwned := handlers[capabilityID]; handlerOwned {
			continue
		}
		result = append(result, capabilityID)
	}
	slices.Sort(result)
	return result
}

// runtimeMetadataCapabilityIDs is the one capability set used both to compile
// the immutable Catalog closure and to store its Runtime views. Most Runtime
// projections are selected by an L1 HandlerBinding. PageLayout resolution and
// Layout authoring are consumer-only dependencies shared by handlers and the
// Runtime command worker, so their workload grants enter the same set.
func runtimeMetadataCapabilityIDs(request Request) []string {
	result := runtimeCapabilityIDs(request.Capabilities)
	for _, grant := range request.Operation.ConsumerGrants {
		if !consumerGrantSelectsRuntimeMetadataCapability(
			request.WorkloadID, grant.CapabilityID, grant,
		) ||
			slices.Contains(result, grant.CapabilityID) {
			continue
		}
		result = append(result, grant.CapabilityID)
	}
	slices.Sort(result)
	return result
}

func sameOperationFence(initial, current catalogapp.Operation) bool {
	return current.OperationID == initial.OperationID && current.Kind == initial.Kind &&
		current.AppInstallationID == initial.AppInstallationID && current.OrgID == initial.OrgID &&
		current.AppID == initial.AppID && current.TargetAppReleaseDigest == initial.TargetAppReleaseDigest &&
		current.TargetCatalogRevision == initial.TargetCatalogRevision &&
		current.TargetSecurityEpoch == initial.TargetSecurityEpoch &&
		current.ExpectedProjectionDigest == initial.ExpectedProjectionDigest
}

func validateReconcileBinding(binding catalogapp.HandlerBinding, operation catalogapp.Operation, request Request, selection CapabilitySelection, ownershipEpoch uint64, allowedStates ...catalogapp.BindingState) error {
	if binding.OrgID != operation.OrgID || binding.AppInstallationID != operation.AppInstallationID ||
		binding.AppReleaseDigest != operation.TargetAppReleaseDigest || binding.CapabilityID != selection.AppCapabilityID ||
		binding.HandlerRef != selection.HandlerRef || binding.WorkloadID != request.WorkloadID ||
		binding.ServiceReleaseDigest != string(request.ServiceReleaseDigest) || binding.OwnershipEpoch != ownershipEpoch ||
		!slices.Contains(allowedStates, binding.State) {
		return fmt.Errorf("operation HandlerBinding disagrees with the reconcile intent")
	}
	return nil
}

func loadServingArtifacts(ctx context.Context, store catalogsnapshot.Store, receiptDigest catalogsnapshot.Digest, operation catalogapp.Operation, request Request, selection CapabilitySelection, ownershipEpoch uint64) (ArtifactResult, error) {
	if store == nil {
		return ArtifactResult{}, fmt.Errorf("Catalog snapshot store is required")
	}
	receiptArtifact, err := store.Get(ctx, catalogsnapshot.ArtifactServingCompatibilityReceipt, receiptDigest)
	if err != nil {
		return ArtifactResult{}, fmt.Errorf("read committed compatibility receipt: %w", err)
	}
	receipt, err := catalogsnapshot.OpenServingCompatibilityReceipt(receiptArtifact)
	if err != nil {
		return ArtifactResult{}, fmt.Errorf("open committed compatibility receipt: %w", err)
	}
	wantLineage := catalogsnapshot.Lineage{
		CatalogRevision:  catalogsnapshot.CatalogRevision(operation.TargetCatalogRevision),
		SchemaGeneration: catalogsnapshot.SchemaGeneration(request.SchemaGeneration),
		EngineGeneration: catalogsnapshot.EngineGeneration(request.EngineGeneration),
	}
	if receipt.OrgID() != operation.OrgID || receipt.AppInstallationID() != operation.AppInstallationID ||
		receipt.AppReleaseDigest() != catalogsnapshot.Digest(operation.TargetAppReleaseDigest) ||
		receipt.AppCapabilityID() != selection.AppCapabilityID || receipt.RuntimeCapabilityID() != selection.RuntimeCapabilityID ||
		receipt.WorkloadID() != request.WorkloadID || receipt.ServiceReleaseDigest() != request.ServiceReleaseDigest ||
		receipt.MetadataSourceDigest() != request.MetadataSourceDigest ||
		receipt.OwnershipEpoch() != ownershipEpoch || receipt.SecurityEpoch() != operation.TargetSecurityEpoch || receipt.Lineage() != wantLineage {
		return ArtifactResult{}, fmt.Errorf("committed serving compatibility receipt disagrees with the Catalog operation")
	}
	snapshotArtifact, err := store.Get(ctx, catalogsnapshot.ArtifactCatalogSnapshot, receipt.CatalogSnapshotDigest())
	if err != nil {
		return ArtifactResult{}, fmt.Errorf("read committed Catalog snapshot: %w", err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		return ArtifactResult{}, fmt.Errorf("open committed Catalog snapshot: %w", err)
	}
	viewArtifact, err := store.Get(ctx, catalogsnapshot.ArtifactRuntimeMetadataView, receipt.RuntimeMetadataViewDigest())
	if err != nil {
		return ArtifactResult{}, fmt.Errorf("read committed Runtime metadata view: %w", err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		return ArtifactResult{}, fmt.Errorf("open committed Runtime metadata view: %w", err)
	}
	if err := view.VerifySource(snapshot); err != nil {
		return ArtifactResult{}, fmt.Errorf("verify committed Runtime view source: %w", err)
	}
	if snapshot.OrgID() != operation.OrgID || snapshot.Lineage() != wantLineage ||
		view.OrgID() != operation.OrgID || view.AppInstallationID() != operation.AppInstallationID ||
		view.CapabilityID() != selection.RuntimeCapabilityID || view.Profile() != mustProjectionProfile(selection.RuntimeCapabilityID) || view.Lineage() != wantLineage {
		return ArtifactResult{}, fmt.Errorf("committed serving artifacts disagree with the Catalog operation")
	}
	return ArtifactResult{
		CatalogSnapshotDigest: snapshotArtifact.Digest(), RuntimeMetadataViewDigest: viewArtifact.Digest(),
		CompatibilityReceiptDigest: receiptArtifact.Digest(),
	}, nil
}

func bindContinuationIntent(request Request) (Request, error) {
	capabilities, err := requestCapabilitySelections(request)
	if err != nil {
		return request, err
	}
	request.Capabilities = capabilities
	if len(request.Operation.SelectedHandlerCapabilities) != 0 &&
		!slices.Equal(
			request.Operation.SelectedHandlerCapabilities,
			capabilitySelectionIDs(capabilities),
		) {
		return request, fmt.Errorf(
			"Catalog reconcile durable handler selection disagrees with its capability manifest",
		)
	}
	if len(request.Operation.SelectedHandlerCapabilities) == 0 &&
		len(request.Operation.WorkloadHandlerAdmissions) != 0 {
		return request, fmt.Errorf(
			"Catalog reconcile legacy all-handler selection cannot carry workload admissions",
		)
	}
	request.outboundMailRoutes = canonicalOutboundMailRouteConfigs(request.outboundMailRoutes)
	if strings.TrimSpace(request.WorkloadID) == "" || strings.TrimSpace(request.WorkloadID) != request.WorkloadID || len(request.WorkloadID) > 255 {
		return request, fmt.Errorf("Catalog reconcile workload ID is invalid")
	}
	if err := appauthority.ValidateDigest(string(request.ServiceReleaseDigest)); err != nil {
		return request, fmt.Errorf("Catalog reconcile service release digest: %w", err)
	}
	if request.MetadataSourceDigest != "" {
		if err := appauthority.ValidateDigest(string(request.MetadataSourceDigest)); err != nil {
			return request, fmt.Errorf("Catalog reconcile metadata source digest: %w", err)
		}
	}
	profile, err := classifyReconcileProfile(
		request.Capabilities,
		request.MetadataSourceDigest,
		request.capabilityClosure,
	)
	if err != nil {
		return request, err
	}
	request.profile = profile
	if request.Frontend != nil {
		if err := appauthority.ValidateDigest(request.Frontend.ReleaseDigest); err != nil {
			return request, fmt.Errorf("Catalog reconcile frontend release digest: %w", err)
		}
		if err := appauthority.ValidateDigest(request.Frontend.CompatibilityReceiptDigest); err != nil {
			return request, fmt.Errorf("Catalog reconcile frontend compatibility receipt: %w", err)
		}
	}
	if request.SchemaGeneration == 0 || request.EngineGeneration == 0 {
		return request, fmt.Errorf("Catalog reconcile schema and engine generations must be non-zero")
	}
	if !slices.Contains(request.Operation.AuthorizedWorkloads, request.WorkloadID) {
		return request, fmt.Errorf("Catalog reconcile workload is not authorized by the operation")
	}
	if request.Operation.HandlerWorkloadID != request.WorkloadID {
		return request, fmt.Errorf("Catalog reconcile workload disagrees with the operation handler workload")
	}
	for _, capability := range request.Capabilities {
		if !slices.Contains(request.Operation.GrantedCapabilities, capability.RuntimeCapabilityID) {
			return request, fmt.Errorf("Catalog reconcile Runtime capability %q is not granted by the operation", capability.RuntimeCapabilityID)
		}
	}
	if len(request.outboundMailRoutes) != 0 &&
		!slices.Contains(request.Operation.GrantedCapabilities, outboundintentv1.CapabilityName) {
		return request, fmt.Errorf(
			"Catalog reconcile outbound mail routes require granted Runtime capability %q",
			outboundintentv1.CapabilityName,
		)
	}
	approvalPublicationDigest, err := approvalPublicationIdentity(request)
	if err != nil {
		return request, err
	}
	var intentCapabilities []CapabilitySelection
	if request.profile.multipleCapabilities {
		intentCapabilities = request.Capabilities
	}
	document := struct {
		Schema                    string                    `json:"schema"`
		CapabilityClosure         string                    `json:"capabilityClosure,omitempty"`
		AppCapabilityID           string                    `json:"appCapabilityId"`
		HandlerRef                string                    `json:"handlerRef"`
		RuntimeCapabilityID       string                    `json:"runtimeCapabilityId"`
		WorkloadID                string                    `json:"workloadId"`
		ServiceReleaseDigest      catalogsnapshot.Digest    `json:"serviceReleaseDigest"`
		MetadataSourceDigest      catalogsnapshot.Digest    `json:"metadataSourceDigest,omitempty"`
		SchemaGeneration          uint64                    `json:"schemaGeneration"`
		EngineGeneration          uint64                    `json:"engineGeneration"`
		Frontend                  *FrontendSelection        `json:"frontend,omitempty"`
		Capabilities              []CapabilitySelection     `json:"capabilities,omitempty"`
		ApprovalPublicationDigest catalogsnapshot.Digest    `json:"approvalPublicationDigest,omitempty"`
		OutboundMailRoutes        []OutboundMailRouteConfig `json:"outboundMailRoutes,omitempty"`
	}{
		Schema:            request.profile.continuationIntentSchema,
		CapabilityClosure: request.capabilityClosure,
		AppCapabilityID:   request.AppCapabilityID, HandlerRef: request.HandlerRef,
		RuntimeCapabilityID: request.RuntimeCapabilityID, WorkloadID: request.WorkloadID,
		ServiceReleaseDigest: request.ServiceReleaseDigest,
		MetadataSourceDigest: request.MetadataSourceDigest,
		SchemaGeneration:     request.SchemaGeneration, EngineGeneration: request.EngineGeneration,
		Frontend: request.Frontend, Capabilities: intentCapabilities,
		ApprovalPublicationDigest: approvalPublicationDigest,
		OutboundMailRoutes:        append([]OutboundMailRouteConfig(nil), request.outboundMailRoutes...),
	}
	encoded, err := json.Marshal(document)
	if err != nil {
		return request, fmt.Errorf("encode Catalog reconcile continuation intent: %w", err)
	}
	request.Operation.ContinuationIntentDigest = string(catalogsnapshot.ComputeDigest(encoded))
	return request, nil
}

func outputForActiveOperation(ctx context.Context, authority Authority, store catalogsnapshot.Store, operation catalogapp.Operation, request Request) (Output, error) {
	if _, err := authority.CommitServingArtifacts(ctx, operation.OperationID); err != nil {
		return Output{}, fmt.Errorf("verify activated serving selection commitment: %w", err)
	}
	states := make([]capabilityReconcileState, 0, len(request.Capabilities))
	var lineage catalogsnapshot.Lineage
	for i, selection := range request.Capabilities {
		binding, err := authority.ReadOperationHandlerBinding(ctx, operation.OperationID, selection.AppCapabilityID)
		if err != nil {
			return Output{}, fmt.Errorf("read activated operation capability owner: %w", err)
		}
		if err := validateReconcileBinding(binding, operation, request, selection, binding.OwnershipEpoch,
			catalogapp.BindingActive, catalogapp.BindingDraining, catalogapp.BindingRetired); err != nil {
			return Output{}, err
		}
		artifacts, err := loadServingArtifacts(ctx, store, catalogsnapshot.Digest(binding.CompatibilityReceiptDigest), operation, request, selection, binding.OwnershipEpoch)
		if err != nil {
			return Output{}, err
		}
		receiptArtifact, err := store.Get(ctx, catalogsnapshot.ArtifactServingCompatibilityReceipt, catalogsnapshot.Digest(binding.CompatibilityReceiptDigest))
		if err != nil {
			return Output{}, fmt.Errorf("read activated compatibility receipt: %w", err)
		}
		receipt, err := catalogsnapshot.OpenServingCompatibilityReceipt(receiptArtifact)
		if err != nil || receipt.MetadataSourceDigest() != request.MetadataSourceDigest {
			return Output{}, fmt.Errorf("activated HandlerBinding and compatibility receipt disagree")
		}
		if i == 0 {
			lineage = receipt.Lineage()
		} else if receipt.Lineage() != lineage {
			return Output{}, fmt.Errorf("activated capability receipts disagree on metadata lineage")
		}
		states = append(states, capabilityReconcileState{selection: selection, binding: binding, artifacts: artifacts})
	}
	if err := validateCapabilityArtifactSet(states); err != nil {
		return Output{}, err
	}
	consumerViews, err := storeConsumerOnlyRuntimeViews(
		ctx,
		store,
		states[0].artifacts.CatalogSnapshotDigest,
		operation,
		request,
	)
	if err != nil {
		return Output{}, err
	}
	output := Output{
		Schema: request.profile.outputSchema, OperationID: operation.OperationID,
		OrgID: operation.OrgID, AppInstallationID: operation.AppInstallationID,
		AppReleaseDigest: operation.TargetAppReleaseDigest, WorkloadID: request.WorkloadID,
		ServiceReleaseDigest: string(request.ServiceReleaseDigest),
		MetadataSourceDigest: string(request.MetadataSourceDigest),
		CatalogRevision:      uint64(lineage.CatalogRevision), SchemaGeneration: uint64(lineage.SchemaGeneration),
		EngineGeneration: uint64(lineage.EngineGeneration), SecurityEpoch: operation.TargetSecurityEpoch,
		CatalogSnapshotDigest: string(states[0].artifacts.CatalogSnapshotDigest),
	}
	bindCapabilityOutput(&output, states)
	if err := bindLayoutAuthoringViewRef(&output, request, states, consumerViews); err != nil {
		return Output{}, err
	}
	if request.Frontend != nil {
		frontend, err := authority.ReadOperationFrontendBinding(ctx, operation.OperationID)
		if err != nil {
			return Output{}, fmt.Errorf("read activated frontend owner: %w", err)
		}
		if !slices.Contains([]catalogapp.BindingState{catalogapp.BindingActive, catalogapp.BindingDraining, catalogapp.BindingRetired}, frontend.State) || frontend.OrgID != operation.OrgID ||
			frontend.AppInstallationID != operation.AppInstallationID ||
			frontend.AppReleaseDigest != operation.TargetAppReleaseDigest ||
			frontend.FrontendReleaseDigest != request.Frontend.ReleaseDigest ||
			frontend.CompatibilityReceiptDigest != request.Frontend.CompatibilityReceiptDigest {
			return Output{}, fmt.Errorf("activated FrontendBinding disagrees with the continuation intent")
		}
		bindFrontendOutput(&output, frontend)
	}
	return output, nil
}

func validateCapabilityArtifactSet(states []capabilityReconcileState) error {
	if len(states) == 0 {
		return fmt.Errorf("Catalog reconcile produced no capability artifacts")
	}
	snapshot := states[0].artifacts.CatalogSnapshotDigest
	for _, state := range states[1:] {
		if state.artifacts.CatalogSnapshotDigest != snapshot {
			return fmt.Errorf("capability compatibility receipts disagree on the committed Catalog snapshot")
		}
	}
	return nil
}

func requestCapabilitySelections(request Request) ([]CapabilitySelection, error) {
	if len(request.Capabilities) != 0 {
		if request.AppCapabilityID != "" || request.HandlerRef != "" || request.RuntimeCapabilityID != "" {
			return nil, fmt.Errorf("Catalog reconcile request cannot mix singular capability fields with capabilities")
		}
		if len(request.Capabilities) < 2 && request.capabilityClosure != ConfigSchemaV4 {
			return nil, fmt.Errorf("Catalog reconcile plural capability request requires at least two capabilities")
		}
		return normalizeCapabilitySelections(request.Capabilities)
	}
	return normalizeCapabilitySelections([]CapabilitySelection{{
		AppCapabilityID:     request.AppCapabilityID,
		HandlerRef:          request.HandlerRef,
		RuntimeCapabilityID: request.RuntimeCapabilityID,
	}})
}

func classifyReconcileProfile(
	capabilities []CapabilitySelection,
	metadataSourceDigest catalogsnapshot.Digest,
	capabilityClosure string,
) (reconcileProfile, error) {
	if capabilityClosure != "" && capabilityClosure != ConfigSchemaV4 {
		return reconcileProfile{}, fmt.Errorf(
			"Catalog reconcile capability closure %q is unsupported",
			capabilityClosure,
		)
	}
	if capabilityClosure == ConfigSchemaV4 {
		if len(capabilities) == 0 || metadataSourceDigest == "" {
			return reconcileProfile{}, fmt.Errorf(
				"Catalog reconcile v4 capability request requires a metadata source digest",
			)
		}
		return reconcileProfile{
			outputSchema: OutputSchemaV3, continuationIntentSchema: continuationIntentSchemaV3,
			multipleCapabilities: true,
		}, nil
	}
	switch {
	case len(capabilities) == 1 && capabilities[0].RuntimeCapabilityID != recordaccessv1.CapabilityName:
		return reconcileProfile{}, fmt.Errorf("Catalog reconcile singular profile supports only Runtime capability %q", recordaccessv1.CapabilityName)
	case len(capabilities) == 1 && metadataSourceDigest == "":
		return reconcileProfile{outputSchema: OutputSchemaV1, continuationIntentSchema: continuationIntentSchemaV1}, nil
	case len(capabilities) == 1:
		return reconcileProfile{outputSchema: OutputSchemaV2, continuationIntentSchema: continuationIntentSchemaV2}, nil
	case len(capabilities) >= 2 && metadataSourceDigest != "":
		return reconcileProfile{outputSchema: OutputSchemaV3, continuationIntentSchema: continuationIntentSchemaV3, multipleCapabilities: true}, nil
	case len(capabilities) >= 2:
		return reconcileProfile{}, fmt.Errorf("Catalog reconcile plural capability request requires a metadata source digest")
	default:
		return reconcileProfile{}, fmt.Errorf("Catalog reconcile request has no supported closed profile")
	}
}

func normalizeCapabilitySelections(capabilities []CapabilitySelection) ([]CapabilitySelection, error) {
	if len(capabilities) == 0 {
		return nil, fmt.Errorf("Catalog reconcile requires at least one supported serving capability")
	}
	result := append([]CapabilitySelection(nil), capabilities...)
	sort.Slice(result, func(i, j int) bool { return result[i].AppCapabilityID < result[j].AppCapabilityID })
	seenApp := make(map[string]struct{}, len(result))
	seenRuntime := make(map[string]struct{}, len(result))
	for _, capability := range result {
		if err := appauthority.ValidateCapabilityID(capability.AppCapabilityID); err != nil {
			return nil, err
		}
		if err := appauthority.ValidateHandlerRef(capability.HandlerRef); err != nil {
			return nil, err
		}
		if err := appauthority.ValidateCapabilityID(capability.RuntimeCapabilityID); err != nil {
			return nil, err
		}
		if _, err := projectionProfile(capability.RuntimeCapabilityID); err != nil {
			return nil, err
		}
		if _, duplicate := seenApp[capability.AppCapabilityID]; duplicate {
			return nil, fmt.Errorf("duplicate App capability %q", capability.AppCapabilityID)
		}
		if _, duplicate := seenRuntime[capability.RuntimeCapabilityID]; duplicate {
			return nil, fmt.Errorf("duplicate Runtime capability %q", capability.RuntimeCapabilityID)
		}
		seenApp[capability.AppCapabilityID] = struct{}{}
		seenRuntime[capability.RuntimeCapabilityID] = struct{}{}
	}
	return result, nil
}

func runtimeCapabilityIDs(capabilities []CapabilitySelection) []string {
	result := make([]string, 0, len(capabilities))
	for _, capability := range capabilities {
		result = append(result, capability.RuntimeCapabilityID)
	}
	return result
}

func mustProjectionProfile(runtimeCapabilityID string) catalogsnapshot.ProjectionProfile {
	profile, err := projectionProfile(runtimeCapabilityID)
	if err != nil {
		panic(err)
	}
	return profile
}

func bindCapabilityOutput(output *Output, states []capabilityReconcileState) {
	if len(states) == 1 && output.Schema != OutputSchemaV3 {
		state := states[0]
		output.AppCapabilityID = state.selection.AppCapabilityID
		output.RuntimeCapabilityID = state.selection.RuntimeCapabilityID
		output.HandlerBindingID = state.binding.HandlerBindingID
		output.OwnershipEpoch = state.binding.OwnershipEpoch
		output.RuntimeMetadataViewDigest = string(state.artifacts.RuntimeMetadataViewDigest)
		output.CompatibilityReceiptDigest = state.binding.CompatibilityReceiptDigest
		return
	}
	output.Capabilities = make([]CapabilityOutput, 0, len(states))
	for _, state := range states {
		output.Capabilities = append(output.Capabilities, CapabilityOutput{
			AppCapabilityID:            state.selection.AppCapabilityID,
			RuntimeCapabilityID:        state.selection.RuntimeCapabilityID,
			HandlerBindingID:           state.binding.HandlerBindingID,
			OwnershipEpoch:             state.binding.OwnershipEpoch,
			RuntimeMetadataViewDigest:  string(state.artifacts.RuntimeMetadataViewDigest),
			CompatibilityReceiptDigest: state.binding.CompatibilityReceiptDigest,
		})
	}
}

func bindFrontendOutput(output *Output, binding catalogapp.FrontendBinding) {
	if binding.FrontendBindingID == "" {
		return
	}
	output.FrontendBindingID = binding.FrontendBindingID
	output.FrontendReleaseDigest = binding.FrontendReleaseDigest
	output.FrontendReceiptDigest = binding.CompatibilityReceiptDigest
}