package catalogreconcile

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"slices"
	"strings"
	"testing"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	layoutauthoringv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv1"
	pagelayoutresolutionv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/pagelayoutresolutionv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
)

type fakeAuthority struct {
	operation             catalogapp.Operation
	release               catalogapp.RegisteredRelease
	binding               catalogapp.HandlerBinding
	bindings              map[string]catalogapp.HandlerBinding
	frontend              catalogapp.FrontendBinding
	latestEpoch           uint64
	intent                string
	metadata              string
	hideMetadataReads     int
	receipt               string
	existingByIdempotency bool
	idempotencyReads      int
	stageErr              error
	commitErr             error
	binds                 int
	staged                int
	frontends             int
	published             int
	completed             int
	activateDuringPublish bool
	effectiveComponents   []catalogapp.EffectiveMetadataComponent
	selectedHandlers      []string
	consumerGrants        []catalogapp.ConsumerGrant
	beginConsumerGrants   []catalogapp.ConsumerGrant
	consumerGrantReads    int
}

func TestRuntimeMetadataCapabilitiesIncludeOnlyHandlerGrantedPageLayoutConsumer(t *testing.T) {
	request := fixtureReconcileRequest()
	request.Capabilities = []CapabilitySelection{{
		AppCapabilityID: request.AppCapabilityID, HandlerRef: request.HandlerRef,
		RuntimeCapabilityID: request.RuntimeCapabilityID,
	}}
	request.Operation.ConsumerGrants = []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: "another-workload", CapabilityID: pagelayoutresolutionv1.CapabilityName,
	}}
	if capabilities := runtimeMetadataCapabilityIDs(request); !slices.Equal(capabilities, []string{request.RuntimeCapabilityID}) {
		t.Fatalf("foreign-workload grant changed Runtime metadata capabilities: %v", capabilities)
	}
	request.Operation.ConsumerGrants = append(request.Operation.ConsumerGrants, catalogapp.ConsumerGrant{
		AuthorizedWorkloadID: request.WorkloadID, CapabilityID: pagelayoutresolutionv1.CapabilityName,
	})
	capabilities := runtimeMetadataCapabilityIDs(request)
	if !slices.Contains(capabilities, request.RuntimeCapabilityID) ||
		!slices.Contains(capabilities, pagelayoutresolutionv1.CapabilityName) ||
		len(capabilities) != 2 {
		t.Fatalf("Runtime metadata capabilities = %v", capabilities)
	}
	consumerOnly := consumerOnlyRuntimeCapabilityIDs(request)
	if !slices.Equal(consumerOnly, []string{pagelayoutresolutionv1.CapabilityName}) {
		t.Fatalf("consumer-only Runtime capabilities = %v", consumerOnly)
	}
}

func TestRuntimeMetadataCapabilitiesIncludeLayoutAuthoringOnlyForTheAdmittedWorkload(t *testing.T) {
	request := fixtureReconcileRequest()
	request.Capabilities = []CapabilitySelection{{
		AppCapabilityID: request.AppCapabilityID, HandlerRef: request.HandlerRef,
		RuntimeCapabilityID: request.RuntimeCapabilityID,
	}}
	request.Operation.ConsumerGrants = []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: "another-workload", CapabilityID: layoutauthoringv1.CapabilityName,
	}}
	if capabilities := runtimeMetadataCapabilityIDs(request); slices.Contains(capabilities, layoutauthoringv1.CapabilityName) {
		t.Fatalf("foreign-workload Layout authoring grant changed Runtime metadata capabilities: %v", capabilities)
	}
	request.Operation.ConsumerGrants = append(request.Operation.ConsumerGrants, catalogapp.ConsumerGrant{
		AuthorizedWorkloadID: request.WorkloadID, CapabilityID: layoutauthoringv1.CapabilityName,
	})
	if capabilities := consumerOnlyRuntimeCapabilityIDs(request); !slices.Equal(capabilities, []string{layoutauthoringv1.CapabilityName}) {
		t.Fatalf("consumer-only Layout authoring capabilities = %v", capabilities)
	}
}

type effectiveMetadataFormatRaceAuthority struct {
	*fakeAuthority
	historicalObjectSearch catalogapp.EffectiveMetadataComponent
	committedComponents    []catalogapp.EffectiveMetadataComponent
	objectSearchAttempts   []uint32
	raced                  bool
}

func (a *effectiveMetadataFormatRaceAuthority) CommitOperationMetadataSelectionWithEffectiveComponents(
	_ context.Context,
	operationID string,
	digest string,
	components []catalogapp.EffectiveMetadataComponent,
) (string, error) {
	if operationID != a.operation.OperationID {
		return "", catalogapp.ErrConflict
	}
	foundObjectSearch := false
	for _, component := range components {
		if component.ComponentKind == consumerv1.EffectiveObjectSearchComponentKind &&
			component.BusinessKey == consumerv1.EffectiveObjectSearchComponentBusinessKey {
			a.objectSearchAttempts = append(
				a.objectSearchAttempts,
				component.ComponentFormatVersion,
			)
			foundObjectSearch = true
			break
		}
	}
	if !foundObjectSearch {
		return "", catalogapp.ErrConflict
	}
	if !a.raced {
		a.raced = true
		a.metadata = digest
		a.committedComponents = slices.Clone(components)
		replaced := false
		for index, component := range a.committedComponents {
			if component.ComponentKind == consumerv1.EffectiveObjectSearchComponentKind &&
				component.BusinessKey == consumerv1.EffectiveObjectSearchComponentBusinessKey {
				a.committedComponents[index] = a.historicalObjectSearch
				replaced = true
				break
			}
		}
		if !replaced {
			return "", catalogapp.ErrConflict
		}
		return "", catalogapp.ErrConflict
	}
	if a.metadata != digest || !reflect.DeepEqual(a.committedComponents, components) {
		return "", catalogapp.ErrConflict
	}
	a.effectiveComponents = slices.Clone(components)
	return digest, nil
}

func (a *fakeAuthority) ReadRelease(_ context.Context, appID, digest string) (catalogapp.RegisteredRelease, error) {
	if a.release.AppID != "" {
		if a.release.AppID != appID || a.release.AppReleaseDigest != digest {
			return catalogapp.RegisteredRelease{}, catalogapp.ErrNotFound
		}
		return a.release, nil
	}
	return fixtureReconcileRelease(appID, digest, []CapabilitySelection{{
		AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.ui.record-access.v1",
		RuntimeCapabilityID: "runtime.records.evaluate-access.v1",
	}}, appauthority.FrontendModeRequired), nil
}

func (a *fakeAuthority) CommitOperationMetadataSelection(_ context.Context, operationID, digest string) (string, error) {
	if operationID != a.operation.OperationID {
		return "", catalogapp.ErrConflict
	}
	if a.metadata != "" && a.metadata != digest {
		return "", catalogapp.ErrConflict
	}
	a.metadata = digest
	return digest, nil
}

func (a *fakeAuthority) CommitOperationMetadataSelectionWithEffectiveComponents(
	ctx context.Context,
	operationID string,
	digest string,
	components []catalogapp.EffectiveMetadataComponent,
) (string, error) {
	a.effectiveComponents = append(
		[]catalogapp.EffectiveMetadataComponent(nil), components...,
	)
	return a.CommitOperationMetadataSelection(ctx, operationID, digest)
}

func (a *fakeAuthority) ReadOperationMetadataSelection(_ context.Context, operationID string) (string, error) {
	if a.hideMetadataReads > 0 {
		a.hideMetadataReads--
		return "", catalogapp.ErrNotFound
	}
	if operationID != a.operation.OperationID || a.metadata == "" {
		return "", catalogapp.ErrNotFound
	}
	return a.metadata, nil
}

func (a *fakeAuthority) ReadOperation(_ context.Context, operationID string) (catalogapp.Operation, error) {
	if operationID != a.operation.OperationID {
		return catalogapp.Operation{}, catalogapp.ErrNotFound
	}
	return a.operation, nil
}

func (a *fakeAuthority) ReadOperationByIdempotency(_ context.Context, _, _, _ string) (catalogapp.Operation, error) {
	a.idempotencyReads++
	if !a.existingByIdempotency {
		return catalogapp.Operation{}, catalogapp.ErrNotFound
	}
	return a.operation, nil
}

func (a *fakeAuthority) CommitServingArtifacts(_ context.Context, operationID string) (string, error) {
	if operationID != a.operation.OperationID {
		return "", catalogapp.ErrConflict
	}
	if a.bindings != nil {
		expectedBindings := len(a.release.Package.Manifest.ProvidedCapabilities)
		if len(a.selectedHandlers) != 0 {
			expectedBindings = len(a.selectedHandlers)
		}
		if len(a.bindings) != expectedBindings {
			return "", catalogapp.ErrNotReady
		}
	} else if a.binding.HandlerBindingID == "" {
		return "", catalogapp.ErrNotReady
	}
	a.binds++
	if a.commitErr != nil {
		return "", a.commitErr
	}
	a.receipt = a.binding.CompatibilityReceiptDigest
	if a.operation.Step == "CANDIDATE_COMMITTED" {
		a.operation.Step = catalogapp.OperationStepServingArtifactsCommitted
	}
	return digestOf('7'), nil
}

func (a *fakeAuthority) ReadOperationHandlerBinding(_ context.Context, operationID, capabilityID string) (catalogapp.HandlerBinding, error) {
	if a.bindings != nil {
		binding, ok := a.bindings[capabilityID]
		if operationID != a.operation.OperationID || !ok || binding.State == catalogapp.BindingFailed {
			return catalogapp.HandlerBinding{}, catalogapp.ErrNotFound
		}
		return binding, nil
	}
	if operationID != a.operation.OperationID || a.binding.HandlerBindingID == "" ||
		a.binding.CapabilityID != capabilityID || a.binding.State == catalogapp.BindingFailed {
		return catalogapp.HandlerBinding{}, catalogapp.ErrNotFound
	}
	return a.binding, nil
}

func (a *fakeAuthority) BeginOperation(_ context.Context, request catalogapp.BeginOperationRequest) (catalogapp.Operation, error) {
	if a.intent != "" && a.intent != request.ContinuationIntentDigest {
		return catalogapp.Operation{}, catalogapp.ErrConflict
	}
	a.intent = request.ContinuationIntentDigest
	a.selectedHandlers = slices.Clone(request.SelectedHandlerCapabilities)
	a.beginConsumerGrants = slices.Clone(request.ConsumerGrants)
	return a.operation, nil
}

func (a *fakeAuthority) ReadOperationConsumerGrants(
	_ context.Context,
	operationID string,
) ([]catalogapp.ConsumerGrant, error) {
	if operationID != a.operation.OperationID {
		return nil, catalogapp.ErrNotFound
	}
	a.consumerGrantReads++
	if a.consumerGrants != nil {
		return slices.Clone(a.consumerGrants), nil
	}
	return slices.Clone(a.beginConsumerGrants), nil
}

func (a *fakeAuthority) ReadLatestHandlerOwnershipEpoch(context.Context, string, string) (uint64, error) {
	if a.latestEpoch == 0 {
		return 0, catalogapp.ErrNotFound
	}
	return a.latestEpoch, nil
}

func (a *fakeAuthority) ReadActiveFrontendBinding(context.Context, string) (catalogapp.FrontendBinding, error) {
	if a.operation.Step != "ACTIVATED" || a.frontend.FrontendBindingID == "" {
		return catalogapp.FrontendBinding{}, catalogapp.ErrNotFound
	}
	return a.frontend, nil
}

func (a *fakeAuthority) ReadOperationFrontendBinding(_ context.Context, operationID string) (catalogapp.FrontendBinding, error) {
	if operationID != a.operation.OperationID || a.frontend.FrontendBindingID == "" {
		return catalogapp.FrontendBinding{}, catalogapp.ErrNotFound
	}
	return a.frontend, nil
}

func (a *fakeAuthority) StageHandlerBinding(_ context.Context, request catalogapp.HandlerBindingRequest) (catalogapp.HandlerBinding, error) {
	if a.stageErr != nil {
		return catalogapp.HandlerBinding{}, a.stageErr
	}
	a.staged++
	bindingID := "binding-test"
	if a.bindings != nil {
		bindingID = "binding-" + request.CapabilityID
	}
	binding := catalogapp.HandlerBinding{
		HandlerBindingID: bindingID, OrgID: a.operation.OrgID,
		AppInstallationID: a.operation.AppInstallationID,
		AppReleaseDigest:  a.operation.TargetAppReleaseDigest,
		CapabilityID:      request.CapabilityID, HandlerRef: request.HandlerRef,
		WorkloadID: request.WorkloadID, ServiceReleaseDigest: request.ServiceReleaseDigest,
		OwnershipEpoch: request.ExpectedOwnershipEpoch + 1,
		State:          catalogapp.BindingStaged, CompatibilityReceiptDigest: request.CompatibilityReceiptDigest,
	}
	if a.bindings != nil {
		a.bindings[request.CapabilityID] = binding
	} else {
		a.binding = binding
	}
	a.latestEpoch = binding.OwnershipEpoch
	return binding, nil
}

func (a *fakeAuthority) StageFrontendBinding(_ context.Context, request catalogapp.FrontendBindingRequest) (catalogapp.FrontendBinding, error) {
	if a.frontend.FrontendBindingID != "" {
		if a.frontend.FrontendReleaseDigest != request.FrontendReleaseDigest || a.frontend.CompatibilityReceiptDigest != request.CompatibilityReceiptDigest {
			return catalogapp.FrontendBinding{}, catalogapp.ErrConflict
		}
		return a.frontend, nil
	}
	a.frontends++
	a.frontend = catalogapp.FrontendBinding{
		FrontendBindingID: "frontend-binding-test", OrgID: a.operation.OrgID,
		AppInstallationID: a.operation.AppInstallationID, AppReleaseDigest: a.operation.TargetAppReleaseDigest,
		FrontendReleaseDigest: request.FrontendReleaseDigest, State: catalogapp.BindingStaged,
		CompatibilityReceiptDigest: request.CompatibilityReceiptDigest,
	}
	return a.frontend, nil
}

func (a *fakeAuthority) PublishRuntimeSecurityProjection(context.Context, string, catalogapp.RuntimeProjectionWriter) error {
	a.published++
	if a.activateDuringPublish {
		a.activateDuringPublish = false
		a.operation.Step = "ACTIVATED"
		a.operation.State = catalogapp.StateActive
		a.binding.State = catalogapp.BindingActive
		if a.frontend.FrontendBindingID != "" {
			a.frontend.State = catalogapp.BindingActive
		}
		return catalogapp.ErrNotReady
	}
	return nil
}

func (a *fakeAuthority) CompleteOperation(context.Context, string, *catalogapp.RemovalGateReceipt) (catalogapp.Operation, error) {
	a.completed++
	a.operation.Step = "ACTIVATED"
	a.operation.State = catalogapp.StateActive
	a.binding.State = catalogapp.BindingActive
	for capabilityID, binding := range a.bindings {
		binding.State = catalogapp.BindingActive
		a.bindings[capabilityID] = binding
	}
	if a.frontend.FrontendBindingID != "" {
		a.frontend.State = catalogapp.BindingActive
	}
	return a.operation, nil
}

type fakeProjectionWriter struct{}

func (fakeProjectionWriter) ActivateProjection(context.Context, uint64, catalogapp.RuntimeSecurityProjection) error {
	return nil
}

func TestReconcileActivatesSupportedCapabilityFamilyFromOneCatalogSnapshot(t *testing.T) {
	for _, capabilityCount := range []int{2, 3, 4} {
		t.Run(fmt.Sprintf("capabilities=%d", capabilityCount), func(t *testing.T) {
			request := fixtureReconcileRequest()
			request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
			request.Frontend = nil
			request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
			request.Capabilities = []CapabilitySelection{
				{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.ui.record-access.v1", RuntimeCapabilityID: recordaccessv1.CapabilityName},
				{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.query.v1", RuntimeCapabilityID: queryv1.CapabilityName},
			}
			request.Operation.GrantedCapabilities = []string{recordaccessv1.CapabilityName, queryv1.CapabilityName}
			if capabilityCount == 3 {
				request.Capabilities = append(request.Capabilities, CapabilitySelection{
					AppCapabilityID: "io.nexuscrm.sfcrm.related-list-counts.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.related-list-counts.v1",
					RuntimeCapabilityID: relatedcountsv1.CapabilityName,
				})
				request.Operation.GrantedCapabilities = append(request.Operation.GrantedCapabilities, relatedcountsv1.CapabilityName)
			}
			if capabilityCount >= 4 {
				request.Capabilities = append(request.Capabilities,
					CapabilitySelection{
						AppCapabilityID: "io.nexuscrm.sfcrm.related-list-counts.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.related-list-counts.v1",
						RuntimeCapabilityID: relatedcountsv1.CapabilityName,
					},
					CapabilitySelection{
						AppCapabilityID: "io.nexuscrm.sfcrm.record-mutation.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.record-mutation.v1",
						RuntimeCapabilityID: recordmutationv1.CapabilityName,
					},
				)
				request.Operation.GrantedCapabilities = append(request.Operation.GrantedCapabilities,
					relatedcountsv1.CapabilityName, recordmutationv1.CapabilityName)
			}
			authority := &fakeAuthority{operation: catalogapp.Operation{
				OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
				AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
				State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
				TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
				TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
			}, bindings: map[string]catalogapp.HandlerBinding{}}
			authority.release = fixtureReconcileRelease(request.Operation.AppID, request.Operation.TargetAppReleaseDigest,
				request.Capabilities, appauthority.FrontendModeNone)
			transactions := 0
			store := &memoryStore{}
			output, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store,
				func(ctx context.Context, fn func(context.Context) error) error {
					transactions++
					return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
				}, &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}, request)
			if err != nil {
				t.Fatalf("reconcile capability family: %v", err)
			}
			if output.Schema != OutputSchemaV3 || output.AppCapabilityID != "" || len(output.Capabilities) != capabilityCount ||
				transactions != 1 || authority.staged != capabilityCount || authority.published != 1 || authority.completed != 1 || len(store.artifacts) != 1+2*capabilityCount {
				t.Fatalf("atomic family output=%#v transactions=%d staged=%d published=%d completed=%d artifacts=%d",
					output, transactions, authority.staged, authority.published, authority.completed, len(store.artifacts))
			}
			if authority.consumerGrantReads != 1 {
				t.Fatalf("Catalog consumer grant authority reads = %d, want 1", authority.consumerGrantReads)
			}
			if _, _, err := MarshalOutput(output); err != nil {
				t.Fatalf("marshal closed v3 handoff: %v", err)
			}
			for _, capability := range output.Capabilities {
				artifact, err := store.Get(context.Background(), catalogsnapshot.ArtifactRuntimeMetadataView, catalogsnapshot.Digest(capability.RuntimeMetadataViewDigest))
				if err != nil {
					t.Fatal(err)
				}
				view, err := catalogsnapshot.OpenRuntimeMetadataView(artifact)
				if err != nil {
					t.Fatal(err)
				}
				wantProfile := mustProjectionProfile(capability.RuntimeCapabilityID)
				if view.Profile() != wantProfile || view.SourceSnapshotDigest() != catalogsnapshot.Digest(output.CatalogSnapshotDigest) {
					t.Fatalf("capability %q view profile/source = %q/%q, want %q/%q", capability.RuntimeCapabilityID,
						view.Profile(), view.SourceSnapshotDigest(), wantProfile, output.CatalogSnapshotDigest)
				}
			}
		})
	}
}

func TestReconcileV4ActivatesCompleteComposedHandlerClosure(t *testing.T) {
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
	request.capabilityClosure = ConfigSchemaV4
	request.Capabilities = []CapabilitySelection{
		{
			AppCapabilityID:     "io.nexuscrm.sfcrm.ui.record-access.v1",
			HandlerRef:          "io.nexuscrm.sfcrm.handler.ui.record-access.v1",
			RuntimeCapabilityID: recordaccessv1.CapabilityName,
		},
		{
			AppCapabilityID:     "io.nexuscrm.sfcrm.query.v1",
			HandlerRef:          "io.nexuscrm.sfcrm.handler.query.v1",
			RuntimeCapabilityID: queryv1.CapabilityName,
		},
		{
			AppCapabilityID:     "io.nexuscrm.sfcrm.related-list-counts.v1",
			HandlerRef:          "io.nexuscrm.sfcrm.handler.related-list-counts.v1",
			RuntimeCapabilityID: relatedcountsv1.CapabilityName,
		},
	}
	request.Operation.GrantedCapabilities = []string{
		recordaccessv1.CapabilityName,
		queryv1.CapabilityName,
		relatedcountsv1.CapabilityName,
	}
	operation := catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test",
		AppID: request.Operation.AppID, State: catalogapp.StatePreparing,
		Step:                   "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}
	authority := &fakeAuthority{
		operation: operation,
		bindings:  map[string]catalogapp.HandlerBinding{},
	}
	authority.release = fixtureReconcileRelease(
		request.Operation.AppID,
		request.Operation.TargetAppReleaseDigest,
		request.Capabilities,
		appauthority.FrontendModeNone,
	)
	for index := range authority.release.Package.Manifest.ProvidedCapabilities {
		capability := &authority.release.Package.Manifest.ProvidedCapabilities[index]
		if capability.CapabilityID == "io.nexuscrm.sfcrm.query.v1" {
			capability.RequiredRuntimeCapabilities = []string{
				queryv1.CapabilityName,
				recordaccessv1.CapabilityName,
			}
		}
	}
	store := &memoryStore{}
	transactions := 0
	output, err := Reconcile(
		context.Background(),
		authority,
		fakeProjectionWriter{},
		store,
		func(ctx context.Context, fn func(context.Context) error) error {
			transactions++
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		},
		&markedReader{
			t:       t,
			objects: []*models.ObjectMetadata{validObject("WorkItem")},
		},
		request,
	)
	if err != nil {
		t.Fatalf("reconcile v4 composed handler closure: %v", err)
	}
	if output.Schema != OutputSchemaV3 ||
		len(output.Capabilities) != len(request.Capabilities) ||
		authority.staged != len(request.Capabilities) ||
		authority.completed != 1 ||
		transactions != 1 {
		t.Fatalf(
			"v4 terminal closure output=%#v staged=%d completed=%d transactions=%d",
			output,
			authority.staged,
			authority.completed,
			transactions,
		)
	}
}

func TestReconcileV4ActivatesOneServingHandlerWithWorkloadOnlyPeer(t *testing.T) {
	const sessionCapability = "runtime.identity.session-lifecycle.v1"
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
	request.capabilityClosure = ConfigSchemaV4
	request.Capabilities = []CapabilitySelection{{
		AppCapabilityID:     "io.nexuscrm.sfcrm.query.v1",
		HandlerRef:          "io.nexuscrm.sfcrm.handler.query.v1",
		RuntimeCapabilityID: queryv1.CapabilityName,
	}}
	request.Operation.SelectedHandlerCapabilities = []string{
		request.Capabilities[0].AppCapabilityID,
	}
	request.Operation.WorkloadHandlerAdmissions = []catalogapp.WorkloadHandlerAdmission{{
		CapabilityID: "io.nexuscrm.sfcrm.scim.v1", AuthorizedWorkloadID: request.WorkloadID,
		RequiredRuntimeCapabilities: []string{sessionCapability},
	}}
	request.Operation.GrantedCapabilities = []string{
		queryv1.CapabilityName,
		sessionCapability,
	}
	slices.Sort(request.Operation.GrantedCapabilities)
	operation := catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test",
		AppID: request.Operation.AppID, State: catalogapp.StatePreparing,
		Step:                   "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}
	authority := &fakeAuthority{
		operation: operation,
		bindings:  map[string]catalogapp.HandlerBinding{},
	}
	authority.release = fixtureReconcileRelease(
		request.Operation.AppID,
		request.Operation.TargetAppReleaseDigest,
		request.Capabilities,
		appauthority.FrontendModeNone,
	)
	authority.release.Package.Manifest.RequiredEngineCapabilities = append(
		authority.release.Package.Manifest.RequiredEngineCapabilities,
		sessionCapability,
	)
	slices.Sort(authority.release.Package.Manifest.RequiredEngineCapabilities)
	authority.release.Package.Manifest.ProvidedCapabilities = append(
		authority.release.Package.Manifest.ProvidedCapabilities,
		appauthority.ProvidedCapability{
			CapabilityID: "io.nexuscrm.sfcrm.scim.v1", Kind: appauthority.CapabilityKindServiceHandler,
			HandlerRef:                  "io.nexuscrm.sfcrm.handler.scim.v1",
			RequiredRuntimeCapabilities: []string{sessionCapability},
		},
	)
	store := &memoryStore{}
	transactions := 0
	output, err := Reconcile(
		context.Background(),
		authority,
		fakeProjectionWriter{},
		store,
		func(ctx context.Context, fn func(context.Context) error) error {
			transactions++
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		},
		&markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}},
		request,
	)
	if err != nil {
		t.Fatalf("reconcile v4 single serving handler: %v", err)
	}
	if output.Schema != OutputSchemaV3 || len(output.Capabilities) != 1 ||
		output.Capabilities[0].AppCapabilityID != request.Capabilities[0].AppCapabilityID ||
		output.AppCapabilityID != "" || authority.staged != 1 || authority.completed != 1 ||
		transactions != 1 {
		t.Fatalf(
			"v4 single-serving output=%#v staged=%d completed=%d transactions=%d",
			output,
			authority.staged,
			authority.completed,
			transactions,
		)
	}
	if _, _, err := MarshalOutput(output); err != nil {
		t.Fatalf("marshal v4 single-serving handoff: %v", err)
	}
}

func TestReconcileV4PublishesConsumerOnlyCommandViewBeforeActivation(t *testing.T) {
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
	request.capabilityClosure = ConfigSchemaV4
	request.Capabilities = []CapabilitySelection{
		{
			AppCapabilityID:     "io.example.app.query.v1",
			HandlerRef:          "io.example.app.handler.query.v1",
			RuntimeCapabilityID: queryv1.CapabilityName,
		},
		{
			AppCapabilityID:     "io.example.app.record-access.v1",
			HandlerRef:          "io.example.app.handler.record-access.v1",
			RuntimeCapabilityID: recordaccessv1.CapabilityName,
		},
	}
	request.Operation.AppID = "io.example.app"
	request.Operation.AuthorizedWorkloads = []string{
		request.WorkloadID,
		"integration-service",
	}
	request.Operation.GrantedCapabilities = []string{
		commandv1.CapabilityName,
		queryv1.CapabilityName,
		recordaccessv1.CapabilityName,
	}
	request.Operation.IntegrationBindings = []catalogapp.IntegrationBindingSelection{{
		AuthorizedWorkloadID: "integration-service",
		Binding: consumerv1.IntegrationBinding{
			OwnerPack: "connector-pack",
			RuntimeCommand: &consumerv1.RuntimeCommandBinding{
				CommandKind:          "provider.event",
				PayloadFormatVersion: 1,
				DefinitionRef:        "ProviderEvent",
				DefinitionVersion:    "flow-version-1",
			},
		},
	}}
	release := fixtureReconcileReleaseWithConsumerCommand(t, request.Capabilities)
	request.Operation.TargetAppReleaseDigest = release.AppReleaseDigest
	operation := catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test",
		AppID: request.Operation.AppID, State: catalogapp.StatePreparing,
		Step:                   "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}
	authority := &fakeAuthority{
		operation: operation,
		release:   release,
		bindings:  map[string]catalogapp.HandlerBinding{},
	}
	store := &memoryStore{}
	output, err := Reconcile(
		context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		},
		&markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}},
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	snapshotArtifact, err := store.Get(
		context.Background(),
		catalogsnapshot.ArtifactCatalogSnapshot,
		catalogsnapshot.Digest(output.CatalogSnapshotDigest),
	)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	expected, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: output.AppInstallationID,
			CapabilityID:      commandv1.CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	published, err := store.Get(
		context.Background(),
		catalogsnapshot.ArtifactRuntimeMetadataView,
		expected.Digest(),
	)
	if err != nil {
		t.Fatalf("consumer-only command view was not durable before activation: %v", err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(published)
	if err != nil {
		t.Fatal(err)
	}
	if namespace, ok := view.CommandExecutionContext(
		"provider.event",
		1,
		"integration-service",
		"ProviderEvent",
		"flow-version-1",
	); !ok || namespace != "connector-pack" {
		t.Fatalf("published command execution context = %q, %t", namespace, ok)
	}
	failingStore := &memoryStore{
		artifacts: []catalogsnapshot.Artifact{snapshotArtifact},
		failAt:    2,
	}
	if _, err := storeConsumerOnlyRuntimeViews(
		context.Background(),
		failingStore,
		snapshotArtifact.Digest(),
		operation,
		request,
	); err == nil || len(failingStore.artifacts) != 1 {
		t.Fatalf(
			"consumer-only view durability failure = %v, artifacts=%d",
			err,
			len(failingStore.artifacts),
		)
	}
	store.artifacts = slices.DeleteFunc(
		store.artifacts,
		func(artifact catalogsnapshot.Artifact) bool {
			return artifact.Kind() == catalogsnapshot.ArtifactRuntimeMetadataView &&
				artifact.Digest() == expected.Digest()
		},
	)
	replayed, err := Reconcile(
		context.Background(),
		authority,
		fakeProjectionWriter{},
		store,
		nil,
		nil,
		request,
	)
	if err != nil {
		t.Fatalf("active operation replay did not restore consumer-only view: %v", err)
	}
	if replayed.CatalogSnapshotDigest != output.CatalogSnapshotDigest {
		t.Fatalf(
			"active replay snapshot = %q, want %q",
			replayed.CatalogSnapshotDigest,
			output.CatalogSnapshotDigest,
		)
	}
	if _, err := store.Get(
		context.Background(),
		catalogsnapshot.ArtifactRuntimeMetadataView,
		expected.Digest(),
	); err != nil {
		t.Fatalf("active operation replay left command view missing: %v", err)
	}
}

func TestReconcileV4ReturnsAndReplaysExactLayoutAuthoringViewRef(t *testing.T) {
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
	request.capabilityClosure = ConfigSchemaV4
	request.Capabilities = []CapabilitySelection{{
		AppCapabilityID:     "io.example.app.layout-authoring.v1",
		HandlerRef:          "io.example.app.handler.layout-authoring.v1",
		RuntimeCapabilityID: layoutauthoringv1.CapabilityName,
	}}
	request.Operation.AppID = "io.example.app"
	request.Operation.GrantedCapabilities = []string{layoutauthoringv1.CapabilityName}
	release := fixtureReconcileRelease(
		request.Operation.AppID,
		request.Operation.TargetAppReleaseDigest,
		request.Capabilities,
		appauthority.FrontendModeNone,
	)
	operation := catalogapp.Operation{
		OperationID: "operation-layout-authoring", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test",
		AppID: request.Operation.AppID, State: catalogapp.StatePreparing,
		Step: "CANDIDATE_COMMITTED", TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision: 7, TargetSecurityEpoch: 11,
	}
	authority := &fakeAuthority{
		operation: operation,
		release:   release,
		bindings:  map[string]catalogapp.HandlerBinding{},
	}
	object := validObject("WorkItem")
	object.ListFields = []string{"Id"}
	layout := &models.PageLayout{
		ID: "layout-work-item", ObjectAPIName: "WorkItem", LayoutName: "Work Item",
		Type: "Create", Sections: []models.PageSection{}, HeaderActions: []models.ActionConfig{},
		QuickActions: []models.ActionConfig{}, RelatedLists: []models.RelatedListConfig{},
	}
	reader := &markedReader{
		t: t, objects: []*models.ObjectMetadata{object},
		layouts: map[string][]*models.PageLayout{"WorkItem": {layout}},
		layoutAssignments: map[string][]*models.LayoutAssignment{
			"WorkItem": {{ID: "assignment-default", ObjectAPIName: "WorkItem", LayoutID: layout.ID}},
		},
	}
	store := &memoryStore{}
	output, err := Reconcile(
		context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		},
		reader,
		request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if output.LayoutAuthoringViewRef == nil ||
		output.LayoutAuthoringViewRef.CatalogRevision != output.CatalogRevision {
		t.Fatalf("Layout authoring view ref = %#v, publication revision = %d", output.LayoutAuthoringViewRef, output.CatalogRevision)
	}
	viewDigest := catalogsnapshot.Digest(output.LayoutAuthoringViewRef.AuthoringViewDigest)
	viewArtifact, err := store.Get(context.Background(), catalogsnapshot.ArtifactRuntimeMetadataView, viewDigest)
	if err != nil {
		t.Fatalf("returned Layout authoring view is not durable: %v", err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	if view.Digest() != viewDigest || view.OrgID() != output.OrgID ||
		view.AppInstallationID() != output.AppInstallationID ||
		view.CapabilityID() != layoutauthoringv1.CapabilityName ||
		view.Profile() != catalogsnapshot.ProjectionLayoutAuthoringV1 ||
		uint64(view.Lineage().CatalogRevision) != output.CatalogRevision {
		t.Fatalf("returned Layout authoring view disagrees with publication: output=%#v view=%#v", output, view)
	}
	if _, _, err := MarshalOutput(output); err != nil {
		t.Fatalf("marshal Layout authoring publication result: %v", err)
	}

	firstRef := *output.LayoutAuthoringViewRef
	replayed, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, nil, nil, request)
	if err != nil {
		t.Fatalf("replay Layout authoring publication: %v", err)
	}
	if replayed.LayoutAuthoringViewRef == nil || *replayed.LayoutAuthoringViewRef != firstRef {
		t.Fatalf("replayed Layout authoring view ref = %#v, want %#v", replayed.LayoutAuthoringViewRef, firstRef)
	}
	store.artifacts = slices.DeleteFunc(store.artifacts, func(artifact catalogsnapshot.Artifact) bool {
		return artifact.Kind() == catalogsnapshot.ArtifactRuntimeMetadataView && artifact.Digest() == viewDigest
	})
	if _, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, nil, nil, request); err == nil ||
		!strings.Contains(err.Error(), "read committed Runtime metadata view") {
		t.Fatalf("missing durable Layout authoring view replay error = %v", err)
	}
}

func TestReconcileCommitsEffectiveObjectSearchWithCatalogSnapshot(
	t *testing.T,
) {
	request := fixtureReconcileRequest()
	request.Operation.ConsumerGrants = []catalogapp.ConsumerGrant{{
		AuthorizedWorkloadID: request.WorkloadID,
		CapabilityID:         consumerv1.AppMetadataV2CapabilityName,
	}}
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test",
		AppID: request.Operation.AppID, State: catalogapp.StatePreparing,
		Step:                   "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}}
	reader := &markedReader{
		t:       t,
		objects: []*models.ObjectMetadata{validObject("WorkItem")},
		searchAllFields: []searchAllDefaultFieldsDeclaration{{
			ObjectAPIName: "WorkItem", Fields: []string{"Id"},
		}},
	}
	store := &memoryStore{}
	_, err := Reconcile(
		context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(
				ctx, transactionMarker{}, "one-snapshot",
			))
		},
		reader, request,
	)
	if err != nil {
		t.Fatal(err)
	}
	if len(authority.effectiveComponents) != 2 ||
		authority.effectiveComponents[0].ComponentKind !=
			consumerv1.EffectiveApplicationNavigationComponentKind ||
		authority.effectiveComponents[0].BusinessKey !=
			consumerv1.EffectiveApplicationNavigationComponentBusinessKey ||
		consumerv1.ValidateEffectiveMetadataComponent(
			authority.effectiveComponents[0],
		) != nil ||
		authority.effectiveComponents[1].ComponentKind !=
			consumerv1.EffectiveObjectSearchComponentKind ||
		authority.effectiveComponents[1].BusinessKey !=
			consumerv1.EffectiveObjectSearchComponentBusinessKey ||
		consumerv1.ValidateEffectiveMetadataComponent(
			authority.effectiveComponents[1],
		) != nil {
		t.Fatalf(
			"effective component commitment = %#v",
			authority.effectiveComponents,
		)
	}
}

func TestReconcileRecoversHistoricalObjectSearchCommitDuringV3Rollout(
	t *testing.T,
) {
	for _, test := range []struct {
		name         string
		format       uint32
		wantAttempts []uint32
	}{
		{"format 2", 2, []uint32{6, 6, 5, 4, 3, 2}},
		{"format 5", 5, []uint32{6, 6, 5}},
	} {
		t.Run(test.name, func(t *testing.T) {

			request := fixtureReconcileRequest()
			request.Frontend = nil
			request.Operation.ConsumerGrants = []catalogapp.ConsumerGrant{{
				AuthorizedWorkloadID: request.WorkloadID,
				CapabilityID:         consumerv1.AppMetadataV3CapabilityName,
			}}
			release := verifiedSingleCapabilityReconcileRelease(
				t,
				request.Operation.AppID,
				CapabilitySelection{
					AppCapabilityID: request.AppCapabilityID,
					HandlerRef:      request.HandlerRef, RuntimeCapabilityID: request.RuntimeCapabilityID,
				},
			)
			request.Operation.TargetAppReleaseDigest = release.AppReleaseDigest
			pathField := "StageName"
			opportunity := validObject("Opportunity")
			opportunity.PathField = &pathField
			opportunity.Fields = append(opportunity.Fields, models.FieldMetadata{
				ID: "field-opportunity-stage", APIName: "StageName", Label: "Stage",
				Type: constants.FieldTypePicklist, PhysicalName: "stage_name",
			})
			declarations := []searchAllDefaultFieldsDeclaration{{
				ObjectAPIName: "Opportunity", Fields: []string{"Id"},
			}}
			historicalObjectSearch, err := buildObjectSearchForFormatForTest(
				[]*models.ObjectMetadata{opportunity},
				declarations,
				test.format,
			)
			if err != nil {
				t.Fatal(err)
			}
			baseAuthority := &fakeAuthority{release: release, operation: catalogapp.Operation{
				OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
				AppInstallationID: "installation-test", OrgID: "org-test",
				AppID: request.Operation.AppID, State: catalogapp.StatePreparing,
				Step:                   "CANDIDATE_COMMITTED",
				TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
				TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
			}}
			authority := &effectiveMetadataFormatRaceAuthority{
				fakeAuthority:          baseAuthority,
				historicalObjectSearch: historicalObjectSearch,
			}
			reader := &completeV3FlowReader{markedReader: &markedReader{
				t: t, objects: []*models.ObjectMetadata{opportunity},
				searchAllFields: declarations,
			}}
			_, err = Reconcile(
				context.Background(), authority, fakeProjectionWriter{}, &memoryStore{},
				func(ctx context.Context, fn func(context.Context) error) error {
					return fn(context.WithValue(
						ctx, transactionMarker{}, "one-snapshot",
					))
				},
				reader,
				request,
			)
			if err != nil {
				t.Fatalf("V3 reconcile did not recover historical format-%d winner: %v", test.format, err)
			}
			if !slices.Equal(authority.objectSearchAttempts, test.wantAttempts) {
				t.Fatalf(
					"object-search commit attempts = %v, want current/current/historical",
					authority.objectSearchAttempts,
				)
			}
			if authority.staged != 1 || authority.completed != 1 {
				t.Fatalf(
					"recovered operation side effects = staged:%d completed:%d, want 1/1",
					authority.staged,
					authority.completed,
				)
			}

		})
	}
}

func TestSelectedEffectiveMetadataCapabilityPrefersNewestGrantedVersion(t *testing.T) {
	tests := []struct {
		name   string
		grants []catalogapp.ConsumerGrant
		want   string
	}{
		{
			name: "v3 and v2",
			grants: []catalogapp.ConsumerGrant{
				{AuthorizedWorkloadID: "workload-test", CapabilityID: consumerv1.AppMetadataV2CapabilityName},
				{AuthorizedWorkloadID: "workload-test", CapabilityID: consumerv1.AppMetadataV3CapabilityName},
			},
			want: consumerv1.AppMetadataV3CapabilityName,
		},
		{
			name: "v2 only",
			grants: []catalogapp.ConsumerGrant{
				{AuthorizedWorkloadID: "workload-test", CapabilityID: consumerv1.AppMetadataV2CapabilityName},
			},
			want: consumerv1.AppMetadataV2CapabilityName,
		},
		{
			name: "no complete metadata grant",
			grants: []catalogapp.ConsumerGrant{
				{AuthorizedWorkloadID: "workload-test", CapabilityID: consumerv1.AppMetadataCapabilityName},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			operation := catalogapp.BeginOperationRequest{ConsumerGrants: test.grants}
			if got := selectedEffectiveMetadataCapability(operation); got != test.want {
				t.Fatalf("selected capability = %q, want %q", got, test.want)
			}
		})
	}
}

func TestReconcileRejectsAppToRuntimeCapabilityMismatchBeforeCatalogMutation(t *testing.T) {
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
	declared := []CapabilitySelection{
		{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.ui.record-access.v1", RuntimeCapabilityID: "runtime.records.evaluate-access.v1"},
		{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.query.v1", RuntimeCapabilityID: "runtime.records.query.v1"},
	}
	request.Capabilities = []CapabilitySelection{
		{AppCapabilityID: declared[0].AppCapabilityID, HandlerRef: declared[0].HandlerRef, RuntimeCapabilityID: declared[1].RuntimeCapabilityID},
		{AppCapabilityID: declared[1].AppCapabilityID, HandlerRef: declared[1].HandlerRef, RuntimeCapabilityID: declared[0].RuntimeCapabilityID},
	}
	request.Operation.GrantedCapabilities = []string{"runtime.records.evaluate-access.v1", "runtime.records.query.v1"}
	authority := &fakeAuthority{release: fixtureReconcileRelease(
		request.Operation.AppID, request.Operation.TargetAppReleaseDigest, declared, appauthority.FrontendModeNone)}

	_, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, &memoryStore{}, nil, nil, request)
	if err == nil || !strings.Contains(err.Error(), "does not match the verified AppRelease") {
		t.Fatalf("swapped capability mapping error = %v", err)
	}
	if authority.intent != "" || authority.staged != 0 || authority.published != 0 || authority.completed != 0 {
		t.Fatal("swapped capability mapping reached Catalog mutation")
	}
}

func TestReconcileRejectsFrontendShapeMismatchBeforeCatalogMutation(t *testing.T) {
	tests := []struct {
		name         string
		frontendMode string
		mutate       func(*Request)
		wantMessage  string
	}{
		{
			name: "required frontend omitted", frontendMode: appauthority.FrontendModeRequired,
			mutate: func(request *Request) { request.Frontend = nil }, wantMessage: "requires a frontend binding",
		},
		{
			name: "frontend supplied for none mode", frontendMode: appauthority.FrontendModeNone,
			mutate: func(*Request) {}, wantMessage: "does not accept a frontend binding",
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			request := fixtureReconcileRequest()
			test.mutate(&request)
			authority := &fakeAuthority{release: fixtureReconcileRelease(
				request.Operation.AppID, request.Operation.TargetAppReleaseDigest,
				[]CapabilitySelection{{AppCapabilityID: request.AppCapabilityID, HandlerRef: request.HandlerRef, RuntimeCapabilityID: request.RuntimeCapabilityID}},
				test.frontendMode)}
			_, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, &memoryStore{}, nil, nil, request)
			if err == nil || !strings.Contains(err.Error(), test.wantMessage) {
				t.Fatalf("frontend shape error = %v", err)
			}
			if authority.intent != "" || authority.staged != 0 || authority.published != 0 || authority.completed != 0 {
				t.Fatal("frontend shape mismatch reached Catalog mutation")
			}
		})
	}
}

func TestReconcileRejectsIncompleteTwoCapabilityProfileBeforeCatalogMutation(t *testing.T) {
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.Capabilities = []CapabilitySelection{
		{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.ui.record-access.v1", RuntimeCapabilityID: "runtime.records.evaluate-access.v1"},
		{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.query.v1", RuntimeCapabilityID: "runtime.records.query.v1"},
	}
	request.Operation.GrantedCapabilities = []string{"runtime.records.evaluate-access.v1", "runtime.records.query.v1"}
	authority := &fakeAuthority{}

	_, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, &memoryStore{}, nil, nil, request)
	if err == nil || !strings.Contains(err.Error(), "plural capability request requires a metadata source digest") {
		t.Fatalf("incomplete profile error = %v", err)
	}
	if authority.intent != "" || authority.staged != 0 || authority.published != 0 || authority.completed != 0 {
		t.Fatal("incomplete profile reached Catalog mutation")
	}
}

func TestClassifyReconcileProfileAcceptsOnlyClosedVersionShapes(t *testing.T) {
	recordAccess := CapabilitySelection{RuntimeCapabilityID: recordaccessv1.CapabilityName}
	query := CapabilitySelection{RuntimeCapabilityID: queryv1.CapabilityName}
	relatedCounts := CapabilitySelection{RuntimeCapabilityID: relatedcountsv1.CapabilityName}
	metadata := catalogsnapshot.Digest(digestOf('8'))
	tests := []struct {
		name         string
		capabilities []CapabilitySelection
		metadata     catalogsnapshot.Digest
		closure      string
		wantSchema   string
		wantError    bool
	}{
		{name: "v1 record access", capabilities: []CapabilitySelection{recordAccess}, wantSchema: OutputSchemaV1},
		{name: "v2 record access", capabilities: []CapabilitySelection{recordAccess}, metadata: metadata, wantSchema: OutputSchemaV2},
		{name: "singular query without metadata", capabilities: []CapabilitySelection{query}, wantError: true},
		{name: "singular query with metadata", capabilities: []CapabilitySelection{query}, metadata: metadata, wantError: true},
		{name: "pair without metadata", capabilities: []CapabilitySelection{recordAccess, query}, wantError: true},
		{name: "v3 pair", capabilities: []CapabilitySelection{recordAccess, query}, metadata: metadata, wantSchema: OutputSchemaV3},
		{name: "v3 complete current family", capabilities: []CapabilitySelection{recordAccess, query, relatedCounts}, metadata: metadata, wantSchema: OutputSchemaV3},
		{name: "v4 one selected query", capabilities: []CapabilitySelection{query}, metadata: metadata, closure: ConfigSchemaV4, wantSchema: OutputSchemaV3},
		{name: "v4 one selected query without metadata", capabilities: []CapabilitySelection{query}, closure: ConfigSchemaV4, wantError: true},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			profile, err := classifyReconcileProfile(test.capabilities, test.metadata, test.closure)
			if test.wantError {
				if err == nil {
					t.Fatalf("profile = %#v, want error", profile)
				}
				return
			}
			if err != nil || profile.outputSchema != test.wantSchema {
				t.Fatalf("profile = %#v, error = %v, want schema %q", profile, err, test.wantSchema)
			}
		})
	}
}

func TestReconcileRejectsAmbiguousCapabilityRepresentationsBeforeCatalogMutation(t *testing.T) {
	selection := CapabilitySelection{
		AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.ui.record-access.v1",
		RuntimeCapabilityID: "runtime.records.evaluate-access.v1",
	}
	tests := []struct {
		name        string
		mutate      func(*Request)
		wantMessage string
	}{
		{
			name: "query capability in singular representation",
			mutate: func(request *Request) {
				request.AppCapabilityID = "io.nexuscrm.sfcrm.query.v1"
				request.HandlerRef = "io.nexuscrm.sfcrm.handler.query.v1"
				request.RuntimeCapabilityID = "runtime.records.query.v1"
				request.Operation.GrantedCapabilities = []string{"runtime.records.query.v1"}
			},
			wantMessage: "singular profile supports only",
		},
		{
			name: "single capability in plural representation",
			mutate: func(request *Request) {
				request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
				request.Capabilities = []CapabilitySelection{selection}
			},
			wantMessage: "plural capability request requires at least two",
		},
		{
			name: "mixed singular and plural representations",
			mutate: func(request *Request) {
				request.Capabilities = []CapabilitySelection{
					selection,
					{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.query.v1", RuntimeCapabilityID: "runtime.records.query.v1"},
				}
			},
			wantMessage: "cannot mix singular capability fields with capabilities",
		},
		{
			name: "unsupported capability",
			mutate: func(request *Request) {
				request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
				request.Capabilities = []CapabilitySelection{
					selection,
					{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.query.v1", RuntimeCapabilityID: queryv1.CapabilityName},
					{AppCapabilityID: "io.nexuscrm.sfcrm.related-list-counts.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.related-list-counts.v1", RuntimeCapabilityID: relatedcountsv1.CapabilityName},
					{AppCapabilityID: "io.nexuscrm.sfcrm.unsupported.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.unsupported.v1", RuntimeCapabilityID: "runtime.records.unsupported.v1"},
				}
			},
			wantMessage: "does not support Runtime capability",
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			request := fixtureReconcileRequest()
			test.mutate(&request)
			authority := &fakeAuthority{}
			_, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, &memoryStore{}, nil, nil, request)
			if err == nil || !strings.Contains(err.Error(), test.wantMessage) {
				t.Fatalf("ambiguous request error = %v", err)
			}
			if authority.intent != "" || authority.staged != 0 || authority.published != 0 || authority.completed != 0 {
				t.Fatal("ambiguous request reached Catalog mutation")
			}
		})
	}
}

func TestReconcileRejectsMixedCatalogSnapshotsBeforeServingCommit(t *testing.T) {
	request := fixtureReconcileRequest()
	request.AppCapabilityID, request.HandlerRef, request.RuntimeCapabilityID = "", "", ""
	request.Frontend = nil
	request.MetadataSourceDigest = catalogsnapshot.Digest(digestOf('8'))
	request.Capabilities = []CapabilitySelection{
		{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.ui.record-access.v1", RuntimeCapabilityID: "runtime.records.evaluate-access.v1"},
		{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", HandlerRef: "io.nexuscrm.sfcrm.handler.query.v1", RuntimeCapabilityID: "runtime.records.query.v1"},
	}
	request.Operation.GrantedCapabilities = []string{"runtime.records.evaluate-access.v1", "runtime.records.query.v1"}
	operation := catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}
	store := &memoryStore{}
	bindings := make(map[string]catalogapp.HandlerBinding, len(request.Capabilities))
	for i, capability := range request.Capabilities {
		metadata := MetadataSelection{
			OrgID: "org-test", AppInstallationID: "installation-test",
			RuntimeCapabilityID:  capability.RuntimeCapabilityID,
			RuntimeCapabilityIDs: runtimeCapabilityIDs(request.Capabilities),
			CatalogRevision:      7, SchemaGeneration: request.SchemaGeneration, EngineGeneration: request.EngineGeneration,
		}
		effective, err := ReadEffectiveMetadata(context.Background(),
			func(ctx context.Context, fn func(context.Context) error) error {
				return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
			}, &markedReader{t: t, objects: []*models.ObjectMetadata{validObject([]string{"FirstSnapshot", "SecondSnapshot"}[i])}}, metadata, nil)
		if err != nil {
			t.Fatal(err)
		}
		artifacts, err := BuildAndStoreArtifacts(context.Background(), store, effective, ArtifactSelection{
			Metadata: metadata, AppReleaseDigest: catalogsnapshot.Digest(operation.TargetAppReleaseDigest),
			AppCapabilityID: capability.AppCapabilityID, WorkloadID: request.WorkloadID,
			ServiceReleaseDigest: request.ServiceReleaseDigest, MetadataSourceDigest: request.MetadataSourceDigest,
			OwnershipEpoch: 1, SecurityEpoch: operation.TargetSecurityEpoch,
		})
		if err != nil {
			t.Fatal(err)
		}
		bindings[capability.AppCapabilityID] = catalogapp.HandlerBinding{
			HandlerBindingID: "binding-" + capability.AppCapabilityID,
			OrgID:            operation.OrgID, AppInstallationID: operation.AppInstallationID,
			AppReleaseDigest: operation.TargetAppReleaseDigest, CapabilityID: capability.AppCapabilityID,
			HandlerRef: capability.HandlerRef, WorkloadID: request.WorkloadID,
			ServiceReleaseDigest: string(request.ServiceReleaseDigest), OwnershipEpoch: 1,
			State: catalogapp.BindingStaged, CompatibilityReceiptDigest: string(artifacts.CompatibilityReceiptDigest),
		}
	}
	authority := &fakeAuthority{operation: operation, bindings: bindings}
	authority.release = fixtureReconcileRelease(request.Operation.AppID, request.Operation.TargetAppReleaseDigest,
		request.Capabilities, appauthority.FrontendModeNone)
	_, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, nil, nil, request)
	if err == nil || !strings.Contains(err.Error(), "disagree on the committed Catalog snapshot") {
		t.Fatalf("mixed snapshot error = %v", err)
	}
	if authority.binds != 0 || authority.published != 0 || authority.completed != 0 {
		t.Fatalf("mixed snapshot reached serving side effects: commit=%d publish=%d complete=%d",
			authority.binds, authority.published, authority.completed)
	}
}

func TestReconcileActivatesOneClosedSelectionAndReplaysWithoutASecondReadPath(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	transactions := 0
	run := func(ctx context.Context, fn func(context.Context) error) error {
		transactions++
		return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
	}
	store := &memoryStore{}
	first, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request)
	if err != nil {
		t.Fatalf("reconcile: %v", err)
	}
	if first.Schema != OutputSchemaV1 || first.HandlerBindingID == "" || first.CompatibilityReceiptDigest == "" ||
		first.FrontendBindingID == "" || first.FrontendReleaseDigest != request.Frontend.ReleaseDigest {
		t.Fatalf("incomplete reconcile output: %#v", first)
	}
	if authority.binds != 1 || authority.staged != 1 || authority.frontends != 1 || authority.published != 1 || authority.completed != 1 || transactions != 1 {
		t.Fatalf("first reconcile side effects = bind %d stage %d frontend %d publish %d complete %d transactions %d",
			authority.binds, authority.staged, authority.frontends, authority.published, authority.completed, transactions)
	}
	// A later ownership flip may drain or retire these rows. Replaying this
	// operation must still use its immutable operation-bound selection rather
	// than whichever binding is active now.
	authority.binding.State = catalogapp.BindingDraining
	authority.frontend.State = catalogapp.BindingRetired
	second, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request)
	if err != nil {
		t.Fatalf("replay reconcile: %v", err)
	}
	if !reflect.DeepEqual(second, first) {
		t.Fatalf("replay output = %#v, want %#v", second, first)
	}
	if authority.staged != 1 || authority.frontends != 1 || authority.published != 1 || authority.completed != 1 || transactions != 1 {
		t.Fatal("activated replay reopened metadata or mutation paths")
	}
}

func TestReconcileOverlappingActivationStillEmitsExactOutput(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}, activateDuringPublish: true}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	transactions := 0
	store := &memoryStore{}
	output, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			transactions++
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		}, reader, request)
	if err != nil {
		t.Fatalf("overlapping activation: %v", err)
	}
	if output.OperationID != authority.operation.OperationID || output.HandlerBindingID != authority.binding.HandlerBindingID ||
		output.CompatibilityReceiptDigest != authority.binding.CompatibilityReceiptDigest || transactions != 1 ||
		authority.published != 1 || authority.completed != 0 {
		t.Fatalf("overlap output=%#v operation=%#v binding=%#v transactions=%d", output, authority.operation, authority.binding, transactions)
	}
}

func TestReconcileRetryConsumesDurablyStagedArtifactsInsteadOfLiveMetadata(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}, commitErr: errors.New("injected crash after binding selection")}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	transactions := 0
	run := func(ctx context.Context, fn func(context.Context) error) error {
		transactions++
		return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
	}
	store := &memoryStore{}
	if _, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request); err == nil {
		t.Fatal("injected post-commit crash was not observed")
	}
	if authority.receipt != "" || authority.binds != 1 || transactions != 1 || authority.staged != 1 {
		t.Fatalf("first attempt receipt=%q binds=%d transactions=%d staged=%d", authority.receipt, authority.binds, transactions, authority.staged)
	}
	// The live repository now contains different metadata. A retry must not
	// reopen it; the durable operation-bound binding already owns the receipt
	// that the complete binding-set commitment must select.
	reader.objects = []*models.ObjectMetadata{validObject("ChangedAfterCrash")}
	authority.commitErr = nil
	output, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request)
	if err != nil {
		t.Fatalf("retry staged selection: %v", err)
	}
	if output.CompatibilityReceiptDigest != authority.binding.CompatibilityReceiptDigest || transactions != 1 || authority.binds != 2 {
		t.Fatalf("retry output=%#v transactions=%d binds=%d", output, transactions, authority.binds)
	}
}

func TestReconcileRetryRecoversCommittedSnapshotWithoutLiveMetadataRead(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}, stageErr: errors.New("injected crash before binding stage")}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	transactions := 0
	run := func(ctx context.Context, fn func(context.Context) error) error {
		transactions++
		return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
	}
	store := &memoryStore{}
	if _, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request); err == nil {
		t.Fatal("injected pre-binding crash was not observed")
	}
	if authority.metadata == "" || authority.binding.HandlerBindingID != "" || len(store.artifacts) != 3 {
		t.Fatalf("crash prefix metadata=%q binding=%#v artifacts=%d", authority.metadata, authority.binding, len(store.artifacts))
	}
	durableArtifacts := len(store.artifacts)
	reader.objects = []*models.ObjectMetadata{validObject("ChangedAfterCrash")}
	authority.stageErr = nil
	output, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request)
	if err != nil {
		t.Fatalf("retry from committed snapshot: %v", err)
	}
	if transactions != 1 || output.CatalogSnapshotDigest != authority.metadata ||
		output.CompatibilityReceiptDigest == "" || authority.binding.HandlerBindingID == "" ||
		authority.published != 1 || authority.completed != 1 || len(store.artifacts) != durableArtifacts+2 {
		t.Fatalf("snapshot recovery output=%#v transactions=%d artifacts=%d binding=%#v", output, transactions, len(store.artifacts), authority.binding)
	}
}

func TestReconcileConcurrentSnapshotSelectionUsesCatalogWinner(t *testing.T) {
	request := fixtureReconcileRequest()
	operation := catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}
	selection := MetadataSelection{
		OrgID: operation.OrgID, AppInstallationID: operation.AppInstallationID,
		RuntimeCapabilityID: request.RuntimeCapabilityID,
		CatalogRevision:     operation.TargetCatalogRevision, SchemaGeneration: request.SchemaGeneration,
		EngineGeneration: request.EngineGeneration,
	}
	winnerMetadata, err := ReadEffectiveMetadata(context.Background(),
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		}, &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("CatalogWinner")}}, selection, nil)
	if err != nil {
		t.Fatal(err)
	}
	winner, err := prepareArtifacts(winnerMetadata, ArtifactSelection{
		Metadata: selection, AppReleaseDigest: catalogsnapshot.Digest(operation.TargetAppReleaseDigest),
		AppCapabilityID: request.AppCapabilityID, WorkloadID: request.WorkloadID,
		ServiceReleaseDigest: request.ServiceReleaseDigest, OwnershipEpoch: 1,
		SecurityEpoch: operation.TargetSecurityEpoch,
	})
	if err != nil {
		t.Fatal(err)
	}
	store := &memoryStore{}
	if err := storePreparedCatalogSnapshot(context.Background(), store, winner); err != nil {
		t.Fatal(err)
	}
	authority := &fakeAuthority{
		operation: operation, metadata: string(winner.result.CatalogSnapshotDigest), hideMetadataReads: 1,
	}
	liveReader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("ConcurrentLoser")}}
	output, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		}, liveReader, request)
	if err != nil {
		t.Fatalf("concurrent snapshot selection: %v", err)
	}
	if output.CatalogSnapshotDigest != string(winner.result.CatalogSnapshotDigest) ||
		authority.binding.CompatibilityReceiptDigest != output.CompatibilityReceiptDigest {
		t.Fatalf("Catalog winner was not authoritative: output=%#v winner=%q", output, winner.result.CatalogSnapshotDigest)
	}
}

func TestReconcileRetryAdvancesPastFailedOperationBindingEpoch(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}, binding: catalogapp.HandlerBinding{
		HandlerBindingID: "failed-binding", CapabilityID: request.AppCapabilityID,
		OwnershipEpoch: 2, State: catalogapp.BindingFailed,
	}, latestEpoch: 2}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	store := &memoryStore{}
	output, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		}, reader, request)
	if err != nil {
		t.Fatalf("recover failed operation binding: %v", err)
	}
	if output.OwnershipEpoch != 3 || authority.binding.OwnershipEpoch != 3 || authority.latestEpoch != 3 {
		t.Fatalf("recovered ownership epoch: output=%d binding=%d latest=%d; want 3",
			output.OwnershipEpoch, authority.binding.OwnershipEpoch, authority.latestEpoch)
	}
}

func TestReconcileNeverStagesBindingBeforeAllArtifactsAreDurable(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	store := &memoryStore{failAt: 3}
	_, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store,
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		}, reader, request)
	if err == nil || authority.staged != 0 || authority.published != 0 || authority.completed != 0 {
		t.Fatalf("artifact failure = %v; stage %d publish %d complete %d", err, authority.staged, authority.published, authority.completed)
	}
	if authority.metadata == "" {
		t.Fatal("durable Catalog snapshot was not selected before derived artifact publication")
	}
}

func TestReconcileJournalsTheCompleteContinuationIntentBeforeSideEffects(t *testing.T) {
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{operation: catalogapp.Operation{
		OperationID: "operation-test", Kind: catalogapp.OperationPrepare,
		AppInstallationID: "installation-test", OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
		State: catalogapp.StatePreparing, Step: "CANDIDATE_COMMITTED",
		TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
		TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
	}}
	reader := &markedReader{t: t, objects: []*models.ObjectMetadata{validObject("WorkItem")}}
	store := &memoryStore{failAt: 1}
	run := func(ctx context.Context, fn func(context.Context) error) error {
		return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
	}
	if _, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, request); err == nil {
		t.Fatal("first reconcile unexpectedly survived artifact failure")
	}
	if authority.intent == "" {
		t.Fatal("Catalog operation was not bound to a continuation intent")
	}
	if authority.metadata != "" {
		t.Fatal("Catalog selected a snapshot whose durable store write failed")
	}
	changed := request
	changed.ServiceReleaseDigest = catalogsnapshot.Digest(digestOf('e'))
	if _, err := Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, run, reader, changed); !errors.Is(err, catalogapp.ErrConflict) {
		t.Fatalf("changed continuation retry error = %v", err)
	}
	if authority.staged != 0 || authority.published != 0 || authority.completed != 0 {
		t.Fatal("conflicting continuation retry reached a side effect")
	}
}

func TestReconcileFailsClosedOnReceiptBindingMismatchDuringReplay(t *testing.T) {
	effective, selection := fixtureEffectiveMetadata(t)
	store := &memoryStore{}
	artifacts, err := BuildAndStoreArtifacts(context.Background(), store, effective, selection)
	if err != nil {
		t.Fatal(err)
	}
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{
		operation: catalogapp.Operation{
			OperationID: "operation-test", AppInstallationID: "installation-test", OrgID: "org-test",
			Step: "ACTIVATED", State: catalogapp.StateActive,
			TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
			TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
		},
		binding: catalogapp.HandlerBinding{
			HandlerBindingID: "binding-test", AppInstallationID: "installation-test", OrgID: "org-test",
			CapabilityID: request.AppCapabilityID, WorkloadID: "wrong-workload",
			ServiceReleaseDigest: string(request.ServiceReleaseDigest), OwnershipEpoch: 1,
			CompatibilityReceiptDigest: string(artifacts.CompatibilityReceiptDigest), State: catalogapp.BindingActive,
		},
	}
	_, err = Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, nil, nil, request)
	if err == nil || !strings.Contains(err.Error(), "disagree") {
		t.Fatalf("binding/receipt mismatch error = %v", err)
	}
}

func TestReconcileFailsClosedOnFrontendBindingMismatchDuringReplay(t *testing.T) {
	effective, selection := fixtureEffectiveMetadata(t)
	store := &memoryStore{}
	artifacts, err := BuildAndStoreArtifacts(context.Background(), store, effective, selection)
	if err != nil {
		t.Fatal(err)
	}
	request := fixtureReconcileRequest()
	authority := &fakeAuthority{
		operation: catalogapp.Operation{
			OperationID: "operation-test", AppInstallationID: "installation-test", OrgID: "org-test",
			Step: "ACTIVATED", State: catalogapp.StateActive,
			TargetAppReleaseDigest: request.Operation.TargetAppReleaseDigest,
			TargetCatalogRevision:  7, TargetSecurityEpoch: 11,
		},
		binding: catalogapp.HandlerBinding{
			HandlerBindingID: "binding-test", AppInstallationID: "installation-test", OrgID: "org-test",
			AppReleaseDigest: request.Operation.TargetAppReleaseDigest,
			CapabilityID:     request.AppCapabilityID, HandlerRef: request.HandlerRef, WorkloadID: request.WorkloadID,
			ServiceReleaseDigest: string(request.ServiceReleaseDigest), OwnershipEpoch: 1,
			CompatibilityReceiptDigest: string(artifacts.CompatibilityReceiptDigest), State: catalogapp.BindingActive,
		},
		frontend: catalogapp.FrontendBinding{
			FrontendBindingID: "frontend-test", AppInstallationID: "installation-test", OrgID: "org-test",
			AppReleaseDigest:           request.Operation.TargetAppReleaseDigest,
			FrontendReleaseDigest:      "sha256:eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
			CompatibilityReceiptDigest: request.Frontend.CompatibilityReceiptDigest, State: catalogapp.BindingActive,
		},
	}
	_, err = Reconcile(context.Background(), authority, fakeProjectionWriter{}, store, nil, nil, request)
	if err == nil || !strings.Contains(err.Error(), "FrontendBinding disagrees") {
		t.Fatalf("frontend mismatch error = %v", err)
	}
}

func fixtureReconcileRequest() Request {
	return Request{
		Operation: catalogapp.BeginOperationRequest{
			Kind: catalogapp.OperationPrepare, OrgID: "org-test", AppID: "io.nexuscrm.sfcrm",
			TargetAppReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
			IdempotencyKey:         "d1-test", RuntimeAudience: "runtime-test",
			AuthorizedWorkloads: []string{"workload-sfcrm-test"},
			GrantedCapabilities: []string{"runtime.records.evaluate-access.v1"},
			HandlerWorkloadID:   "workload-sfcrm-test",
		},
		AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1",
		HandlerRef:      "io.nexuscrm.sfcrm.handler.ui.record-access.v1", RuntimeCapabilityID: "runtime.records.evaluate-access.v1",
		WorkloadID:           "workload-sfcrm-test",
		ServiceReleaseDigest: catalogsnapshot.Digest("sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"),
		Frontend: &FrontendSelection{
			ReleaseDigest:              "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
			CompatibilityReceiptDigest: "sha256:dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
		},
		SchemaGeneration: 8, EngineGeneration: 9,
	}
}

func fixtureReconcileRelease(appID, digest string, selections []CapabilitySelection, frontendMode string) catalogapp.RegisteredRelease {
	required := make([]string, 0, len(selections))
	provided := make([]appauthority.ProvidedCapability, 0, len(selections))
	for _, selection := range selections {
		required = append(required, selection.RuntimeCapabilityID)
		provided = append(provided, appauthority.ProvidedCapability{
			CapabilityID: selection.AppCapabilityID, Kind: appauthority.CapabilityKindServiceHandler,
			HandlerRef: selection.HandlerRef, RequiredRuntimeCapabilities: []string{selection.RuntimeCapabilityID},
		})
	}
	slices.Sort(required)
	return catalogapp.RegisteredRelease{
		AppID: appID, AppReleaseDigest: digest,
		Package: appauthority.VerifiedPackage{Package: appauthority.Package{Manifest: appauthority.AppManifest{
			AppID: appID, RequiredEngineCapabilities: required, ProvidedCapabilities: provided,
			FrontendCompatibility: appauthority.FrontendCompatibility{Mode: frontendMode},
		}}},
	}
}

func verifiedSingleCapabilityReconcileRelease(
	t *testing.T,
	appID string,
	selection CapabilitySelection,
) catalogapp.RegisteredRelease {
	t.Helper()
	bundle := appauthority.MetadataBundle{
		SchemaVersion: appauthority.MetadataBundleSchemaVersion,
		Components:    []appauthority.Component{},
	}
	manifest := appauthority.AppManifest{
		SchemaVersion: appauthority.AppManifestSchemaVersion,
		AppID:         appID, PublisherID: "io.nexuscrm", SemanticVersion: "1.0.0",
		MetadataBundleDigest: "sha256:" + strings.Repeat("0", 64),
		Dependencies:         []appauthority.Dependency{},
		OwnedNamespaces:      []string{appID},
		RequiredProtocols: []appauthority.ProtocolRequirement{{
			ProtocolID: "nexuscrm.runtime.v1", Major: 1,
		}},
		RequiredEngineCapabilities: []string{selection.RuntimeCapabilityID},
		ProvidedCapabilities: []appauthority.ProvidedCapability{{
			CapabilityID: selection.AppCapabilityID,
			Kind:         appauthority.CapabilityKindServiceHandler,
			HandlerRef:   selection.HandlerRef,
			RequiredRuntimeCapabilities: []string{
				selection.RuntimeCapabilityID,
			},
		}},
		FrontendCompatibility: appauthority.FrontendCompatibility{
			Mode: appauthority.FrontendModeNone,
		},
	}
	digest, err := appauthority.MetadataBundleDigest(bundle, manifest)
	if err != nil {
		t.Fatal(err)
	}
	manifest.MetadataBundleDigest = digest
	verified, err := appauthority.VerifyPackage(appauthority.Package{
		Manifest: manifest, Bundle: bundle, ComponentBlobs: []appauthority.ComponentBlob{},
	})
	if err != nil {
		t.Fatal(err)
	}
	return catalogapp.RegisteredRelease{
		AppID: verified.Manifest.AppID, AppReleaseDigest: verified.AppReleaseDigest,
		Package: verified,
	}
}

func fixtureReconcileReleaseWithConsumerCommand(
	t *testing.T,
	selections []CapabilitySelection,
) catalogapp.RegisteredRelease {
	t.Helper()
	verified := verifiedPlatformRelease(t, []platformReleaseComponentFixture{{
		kind: commandReleaseComponentKind, businessKey: "provider-event",
		behavior: []byte(
			`{"commandKind":"provider.event","payloadFormatVersion":1}`,
		),
	}})
	pkg := verified.Package
	pkg.Manifest.RequiredEngineCapabilities = []string{
		commandv1.CapabilityName,
		queryv1.CapabilityName,
		recordaccessv1.CapabilityName,
	}
	slices.Sort(pkg.Manifest.RequiredEngineCapabilities)
	pkg.Manifest.ProvidedCapabilities = make(
		[]appauthority.ProvidedCapability,
		0,
		len(selections),
	)
	for _, selection := range selections {
		pkg.Manifest.ProvidedCapabilities = append(
			pkg.Manifest.ProvidedCapabilities,
			appauthority.ProvidedCapability{
				CapabilityID: selection.AppCapabilityID,
				Kind:         appauthority.CapabilityKindServiceHandler,
				HandlerRef:   selection.HandlerRef,
				RequiredRuntimeCapabilities: []string{
					selection.RuntimeCapabilityID,
				},
			},
		)
	}
	digest, err := appauthority.MetadataBundleDigest(pkg.Bundle, pkg.Manifest)
	if err != nil {
		t.Fatal(err)
	}
	pkg.Manifest.MetadataBundleDigest = digest
	verified, err = appauthority.VerifyPackage(pkg)
	if err != nil {
		t.Fatal(err)
	}
	return catalogapp.RegisteredRelease{
		AppID:            pkg.Manifest.AppID,
		AppReleaseDigest: verified.AppReleaseDigest,
		Package:          verified,
	}
}