package runtimeapi

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"testing"

	"github.com/nexuscrm/backend/internal/managedruntime"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	approvalnotificationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalnotificationflowv1"
	approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1"
	auditsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/auditlogv1"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	listviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/listviewv1"
	metadataadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/metadataadminreadv1"
	pagelayoutsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/pagelayoutresolutionv1"
	permissionauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionauthoringv1"
	querysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	attachmentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordattachmentsv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	recordviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordviewv1"
	recyclemutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinmutationv1"
	relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
)

const testSFCRMAppCapabilityID = "io.nexuscrm.sfcrm.ui.record-access.v1"

func TestMetadataAdminReadDoesNotSelectCatalogServingProjection(t *testing.T) {
	if got := servingProjectionForCapability(metadataadminsemantic.CapabilityName); got != "" {
		t.Fatalf("metadata admin-read projection = %q, want no Catalog projection", got)
	}
	if servingCapability(metadataadminsemantic.CapabilityName) {
		t.Fatal("metadata admin-read must use workload admission, not ServingAdmission")
	}
}

func TestApprovalNotificationFlowReadDoesNotSelectCatalogServingProjection(t *testing.T) {
	if got := servingProjectionForCapability(approvalnotificationsemantic.CapabilityName); got != "" {
		t.Fatalf("approval notification Flow-read projection = %q, want no Catalog projection", got)
	}
	if servingCapability(approvalnotificationsemantic.CapabilityName) {
		t.Fatal("approval notification Flow read must use workload admission, not ServingAdmission")
	}
}

type servingAssignmentPolicyEligibilityStagerStub struct{ revision uint64 }

func (*servingAssignmentPolicyEligibilityStagerStub) Stage(
	context.Context,
	*sql.Tx,
	[]permissions.AssignmentPolicyEligibilityMutation,
) error {
	return nil
}

func TestServingAdmissionOwnsFlowInternalAppMetadataScope(t *testing.T) {
	state := &ServingState{}
	state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		flowsemantic.CapabilityName: {
			selection: ServingSelection{
				OrgID: testOrgID, AppInstallationID: testInstallationID,
				RuntimeCapabilityID: flowsemantic.CapabilityName,
			},
			binding: ServingHandlerBinding{WorkloadID: testWorkloadID},
		},
		catalogsemantic.AppMetadataV3CapabilityName: {
			selection: ServingSelection{
				OrgID: testOrgID, AppInstallationID: testInstallationID,
				RuntimeCapabilityID: catalogsemantic.AppMetadataV3CapabilityName,
			},
			binding: ServingHandlerBinding{WorkloadID: testWorkloadID},
		},
	}})
	admission := &ServingAdmission{state: state}
	if !admission.hasFlowAppMetadataAuthority() {
		t.Fatal("prepared Flow + AppMetadataV3 set did not become Flow metadata authority")
	}
	scope, capability, err := admission.flowAppMetadataScope(TrustedCall{
		orgID: testOrgID, appInstallationID: testInstallationID,
		callerWorkloadID: testWorkloadID,
	})
	if err != nil {
		t.Fatal(err)
	}
	if capability != catalogsemantic.AppMetadataV3CapabilityName ||
		scope.OrgID != testOrgID || scope.AppInstallationID != testInstallationID ||
		scope.WorkloadID != testWorkloadID ||
		len(scope.Capabilities) != 1 || scope.Capabilities[0] != capability {
		t.Fatalf("Flow metadata scope = %#v, capability = %q", scope, capability)
	}
	if _, _, err := admission.flowAppMetadataScope(TrustedCall{
		orgID: testOrgID, appInstallationID: testInstallationID,
		callerWorkloadID: "another-workload",
	}); err == nil {
		t.Fatal("Flow metadata scope accepted a different workload")
	}
}

func TestServingProjectionForApprovalCapabilitiesIsExact(t *testing.T) {
	if !servingCapability(approvalsemantic.ExecuteCapabilityName) ||
		!servingCapability(approvalsemantic.ReadCapabilityName) {
		t.Fatal("Approval capabilities bypass immutable serving admission")
	}
	if got := servingProjectionForCapability(approvalsemantic.ExecuteCapabilityName); got != catalogsnapshot.ProjectionApprovalExecutionV1 {
		t.Fatalf("Approval execute profile = %q", got)
	}
	if got := servingProjectionForCapability(approvalsemantic.ReadCapabilityName); got != catalogsnapshot.ProjectionApprovalReadV1 {
		t.Fatalf("Approval read profile = %q", got)
	}
}

func TestServingProjectionForEffectivePermissionsUsesLeastPrivilegeProjection(t *testing.T) {
	if !servingCapability(permissionsemantic.CapabilityName) {
		t.Fatal("effective-permissions capability bypasses immutable serving admission")
	}
	if got := servingProjectionForCapability(permissionsemantic.CapabilityName); got != catalogsnapshot.ProjectionEffectivePermissionsV1 {
		t.Fatalf("effective-permissions profile = %q", got)
	}
}

func TestServingProjectionForAuditLogReadUsesPermissionProjection(t *testing.T) {
	if !servingCapability(auditsemantic.CapabilityName) {
		t.Fatal("audit-log read capability bypasses immutable serving admission")
	}
	if got := servingProjectionForCapability(auditsemantic.CapabilityName); got != catalogsnapshot.ProjectionEffectivePermissionsV1 {
		t.Fatalf("audit-log read profile = %q", got)
	}
}

func TestServingProjectionForBulkIngestUsesMutationProjection(t *testing.T) {
	if !servingCapability(bulkingestsemantic.CapabilityName) {
		t.Fatal("Bulk ingest capability bypasses immutable serving admission")
	}
	if got := servingProjectionForCapability(bulkingestsemantic.CapabilityName); got != catalogsnapshot.ProjectionRecordMutationV1 {
		t.Fatalf("Bulk ingest profile = %q", got)
	}
}

func TestPageLayoutResolutionUsesConsumerOnlyAdmission(t *testing.T) {
	if servingCapability(pagelayoutsemantic.CapabilityName) {
		t.Fatal("page-layout resolution manufactured an L1 HandlerBinding serving path")
	}
	if got := servingProjectionForCapability(pagelayoutsemantic.CapabilityName); got != "" {
		t.Fatalf("page-layout resolution ServingAdmission profile = %q, want none", got)
	}
}

func TestPermissionAuthoringServingSpecOwnsMutationProjectionAndEnvelope(t *testing.T) {
	spec, ok := servingCapabilitySpecFor(permissionauthoringsemantic.CapabilityName)
	if !ok || spec.projection != catalogsnapshot.ProjectionRecordMutationV1 ||
		!spec.bindsMutationEnvelope ||
		len(spec.requiredSiblings) != 1 || spec.requiredSiblings[0] != mutationsemantic.CapabilityName {
		t.Fatalf("permission authoring serving spec = %#v, found=%t", spec, ok)
	}
}

func TestRecordViewAdmissionBindsOnlyMatchingEffectivePermissionSibling(t *testing.T) {
	newContext := func() context.Context {
		trusted := newTrustedCall(delegationv1.Claims{
			DelegationID: "record-workspace", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
			CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-test",
			Capabilities: []string{recordviewsemantic.CapabilityName}, SecurityEpoch: 3,
		}, models.UserSession{ID: "user-1"})
		return context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
	}
	recordView := &catalogsnapshot.RuntimeMetadataView{}
	permissionView := &catalogsnapshot.RuntimeMetadataView{}
	prepared := func(permissionRevision uint64) *ServingAdmission {
		state := &ServingState{}
		state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
			recordviewsemantic.CapabilityName: {
				selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: recordviewsemantic.CapabilityName, SecurityEpoch: 3},
				binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: recordView,
				receipts: semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
			},
			permissionsemantic.CapabilityName: {
				selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: permissionsemantic.CapabilityName, SecurityEpoch: 3},
				binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: permissionView,
				receipts: semantic.Receipts{CatalogRevision: permissionRevision, SecurityEpoch: 3},
			},
		}})
		return &ServingAdmission{state: state}
	}

	bound, err := prepared(11).bind(newContext(), recordviewsemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	got, ok := effectivePermissionMetadataViewFromContext(bound)
	if !ok || got != permissionView {
		t.Fatalf("matching sibling = %p/%v, want %p/true", got, ok, permissionView)
	}

	bound, err = prepared(12).bind(newContext(), recordviewsemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	if got, ok := effectivePermissionMetadataViewFromContext(bound); ok || got != nil {
		t.Fatalf("mixed-revision sibling was bound: %p/%v", got, ok)
	}
}

func TestRelatedCountAdmissionBindsMatchingRecordAccessSibling(t *testing.T) {
	trusted := newTrustedCall(delegationv1.Claims{
		DelegationID: "related-count", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-test",
		Capabilities: []string{relatedsemantic.CapabilityName}, SecurityEpoch: 3,
	}, models.UserSession{ID: "user-1"})
	ctx := context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
	queryView := &catalogsnapshot.RuntimeMetadataView{}
	accessView := &catalogsnapshot.RuntimeMetadataView{}
	state := &ServingState{}
	state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		relatedsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: relatedsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: queryView,
			receipts: semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
		},
		accesssemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: accesssemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: accessView,
			receipts: semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
		},
	}})
	bound, err := (&ServingAdmission{state: state}).bind(ctx, relatedsemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	if got, ok := RuntimeMetadataViewFromContext(bound); !ok || got != queryView {
		t.Fatalf("primary view = %p/%v, want record-query %p/true", got, ok, queryView)
	}
	if got, ok := runtimeRecordAccessMetadataViewFromContext(bound); !ok || got != accessView {
		t.Fatalf("access sibling = %p/%v, want %p/true", got, ok, accessView)
	}
}

func TestRelatedCountServingSetRequiresRecordAccessSibling(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 7, "io.nexuscrm.sfcrm.related-list-counts.v1",
		relatedsemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1,
	)
	err := fixture.state.Reconcile(t.Context(), fixture.selection)
	if err == nil || !strings.Contains(err.Error(), "requires sibling capability \""+accesssemantic.CapabilityName+"\"") {
		t.Fatalf("Related Count reconcile error = %v, want missing record-access sibling", err)
	}
	if fixture.state.Ready() {
		t.Fatal("Related Count became ready without its record-access identity authority")
	}
}

func TestServingProjectionForPrivateListViewsUsesRecordMutation(t *testing.T) {
	if !servingCapability(listviewsemantic.CapabilityName) {
		t.Fatal("private-list-view capability bypasses immutable serving admission")
	}
	if got := servingProjectionForCapability(listviewsemantic.CapabilityName); got != catalogsnapshot.ProjectionRecordMutationV1 {
		t.Fatalf("private-list-view profile = %q", got)
	}
}

func TestMutationAdmissionBindsAssignmentPolicyLifecycleToCapturedServingRevision(t *testing.T) {
	newContext := func() context.Context {
		trusted := newTrustedCall(delegationv1.Claims{
			DelegationID: "mutation", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
			CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-test",
			Capabilities: []string{mutationsemantic.CapabilityName}, SecurityEpoch: 3,
		}, models.UserSession{ID: "user-1"})
		return context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
	}
	state := &ServingState{}
	stager11 := &servingAssignmentPolicyEligibilityStagerStub{revision: 11}
	state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		mutationsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: mutationsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: &catalogsnapshot.RuntimeMetadataView{},
			receipts:                    semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
			assignmentPolicyEligibility: stager11,
		},
	}})
	admission := &ServingAdmission{state: state}
	bound11, err := admission.bind(newContext(), mutationsemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	got11, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(bound11)
	if !ok || got11 != stager11 {
		t.Fatalf("revision 11 lifecycle binding = %#v/%v", got11, ok)
	}

	stager12 := &servingAssignmentPolicyEligibilityStagerStub{revision: 12}
	state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		mutationsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: mutationsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: &catalogsnapshot.RuntimeMetadataView{},
			receipts:                    semantic.Receipts{CatalogRevision: 12, SecurityEpoch: 3},
			assignmentPolicyEligibility: stager12,
		},
	}})
	bound12, err := admission.bind(newContext(), mutationsemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	got12, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(bound12)
	if !ok || got12 != stager12 {
		t.Fatalf("revision 12 lifecycle binding = %#v/%v", got12, ok)
	}
	still11, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(bound11)
	if !ok || still11 != stager11 {
		t.Fatalf("in-flight revision 11 lifecycle binding changed = %#v/%v", still11, ok)
	}
}

func TestRecycleBinMutationAdmissionBindsOrdinaryMutationLifecycle(t *testing.T) {
	trusted := newTrustedCall(delegationv1.Claims{
		DelegationID: "recycle-bin-mutation", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-test",
		Capabilities: []string{recyclemutationsemantic.CapabilityName}, SecurityEpoch: 3,
	}, models.UserSession{ID: "user-1"})
	ctx := context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
	stager := &servingAssignmentPolicyEligibilityStagerStub{revision: 11}
	mutationView := &catalogsnapshot.RuntimeMetadataView{}
	recycleMutationView := &catalogsnapshot.RuntimeMetadataView{}
	state := &ServingState{}
	state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		mutationsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: mutationsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: mutationView,
			receipts:                    semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
			assignmentPolicyEligibility: stager,
		},
		recyclemutationsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: recyclemutationsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: recycleMutationView,
			receipts: semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
		},
	}})

	bound, err := (&ServingAdmission{state: state}).bind(ctx, recyclemutationsemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	gotStager, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(bound)
	if !ok || gotStager != stager {
		t.Fatalf("Recycle Bin mutation lifecycle binding = %#v/%v, want Mutation sibling stager", gotStager, ok)
	}
	gotMutationView, ok := RuntimeMutationMetadataViewFromContext(bound)
	if !ok || gotMutationView != mutationView {
		t.Fatalf("Recycle Bin mutation view = %p/%v, want Mutation sibling %p", gotMutationView, ok, mutationView)
	}
	gotPrimaryView, ok := metadata.RuntimeMetadataViewFromContext(bound)
	if !ok || gotPrimaryView != recycleMutationView {
		t.Fatalf("Recycle Bin primary view = %p/%v, want admitted Recycle Bin view %p", gotPrimaryView, ok, recycleMutationView)
	}
}

func TestRecycleBinMutationServingSetRequiresOrdinaryMutationSibling(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 7, "io.nexuscrm.sfcrm.recycle-bin-mutation.v1",
		recyclemutationsemantic.CapabilityName, catalogsnapshot.ProjectionRecordMutationV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil ||
		!strings.Contains(err.Error(), "requires sibling capability \""+mutationsemantic.CapabilityName+"\"") {
		t.Fatalf("Recycle Bin mutation without ordinary Mutation sibling error = %v", err)
	}
	if fixture.state.prepared.Load() != nil {
		t.Fatal("Recycle Bin mutation dependency failure retained serving readiness")
	}
}

func TestApprovalExecutionAdmissionBindsMutationSiblingForOrdinaryDML(t *testing.T) {
	trusted := newTrustedCall(delegationv1.Claims{
		DelegationID: "approval", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-test",
		Capabilities: []string{approvalsemantic.ExecuteCapabilityName}, SecurityEpoch: 3,
	}, models.UserSession{ID: "user-1"})
	ctx := context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
	mutationStager := &servingAssignmentPolicyEligibilityStagerStub{revision: 10}
	approvalStager := &servingAssignmentPolicyEligibilityStagerStub{revision: 11}
	mutationView := &catalogsnapshot.RuntimeMetadataView{}
	approvalView := &catalogsnapshot.RuntimeMetadataView{}
	flowWorkView := &catalogsnapshot.RuntimeMetadataView{}
	state := &ServingState{}
	state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		mutationsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: mutationsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: mutationView,
			receipts:                    semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
			assignmentPolicyEligibility: mutationStager,
		},
		approvalsemantic.ExecuteCapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: approvalsemantic.ExecuteCapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: approvalView,
			receipts:                    semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
			assignmentPolicyEligibility: approvalStager,
		},
		flowsemantic.CapabilityName: {
			selection: ServingSelection{OrgID: testOrgID, AppInstallationID: testInstallationID, RuntimeCapabilityID: flowsemantic.CapabilityName, SecurityEpoch: 3},
			binding:   ServingHandlerBinding{WorkloadID: testWorkloadID}, view: flowWorkView,
			receipts: semantic.Receipts{CatalogRevision: 11, SecurityEpoch: 3},
		},
	}})

	bound, err := (&ServingAdmission{state: state}).bind(ctx, approvalsemantic.ExecuteCapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	got, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(bound)
	if !ok || got != mutationStager {
		t.Fatalf("Approval-owned DML lifecycle binding = %#v/%v, want Mutation sibling stager", got, ok)
	}
	engineView, ok := RuntimeMutationMetadataViewFromContext(bound)
	if !ok || engineView != mutationView {
		t.Fatalf("Approval mutation view = %p/%v, want Mutation sibling %p", engineView, ok, mutationView)
	}
	primaryView, ok := metadata.RuntimeMetadataViewFromContext(bound)
	if !ok || primaryView != approvalView {
		t.Fatalf("Approval primary view = %p/%v, want admitted Approval view %p", primaryView, ok, approvalView)
	}
	gotFlowWork, ok := metadata.RuntimeFlowWorkViewFromContext(bound)
	if !ok || gotFlowWork != flowWorkView {
		t.Fatalf("Approval durable Flow-work sibling = %p/%v, want %p", gotFlowWork, ok, flowWorkView)
	}
}

func TestApprovalExecutionAdmissionSeparatesApprovalAndMutationAuthorities(t *testing.T) {
	snapshotArtifact, views := approvalAuthorityTestArtifactsForCapabilitiesAndMutationDefinitions(
		t,
		testOrgID,
		testInstallationID,
		[]string{approvalsemantic.ExecuteCapabilityName, mutationsemantic.CapabilityName},
		nil,
		approvalBeforeSaveMutationDefinitions(t),
	)
	approvalView, err := catalogsnapshot.OpenRuntimeMetadataView(views[approvalsemantic.ExecuteCapabilityName])
	if err != nil {
		t.Fatal(err)
	}
	if got := approvalView.TriggerSubscriptions("a01", "RecordBeforeSave", "Update"); len(got) != 0 {
		t.Fatalf("Approval projection leaked ordinary mutation subscriptions: %#v", got)
	}
	mutationView, err := catalogsnapshot.OpenRuntimeMetadataView(views[mutationsemantic.CapabilityName])
	if err != nil {
		t.Fatal(err)
	}
	if got := mutationView.TriggerSubscriptions("a01", "RecordBeforeSave", "Update"); len(got) != 1 || got[0].FlowDefinitionID != approvalBeforeSaveFlowDefinitionID {
		t.Fatalf("Mutation projection subscriptions = %#v, want %s", got, approvalBeforeSaveFlowDefinitionID)
	}

	trusted := newTrustedCall(delegationv1.Claims{
		DelegationID: "approval-authority", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: testWorkloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-test",
		Capabilities: []string{approvalsemantic.ExecuteCapabilityName}, SecurityEpoch: 1,
	}, models.UserSession{ID: "user-1"})
	ctx := context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
	bound, err := newApprovalServingFixture(t, snapshotArtifact, views).bind(ctx, approvalsemantic.ExecuteCapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	primaryView, ok := metadata.RuntimeMetadataViewFromContext(bound)
	if !ok || primaryView.Profile() != catalogsnapshot.ProjectionApprovalExecutionV1 {
		t.Fatalf("Approval primary authority = %#v/%v, want Approval execution view", primaryView, ok)
	}
	if got := primaryView.TriggerSubscriptions("a01", "RecordBeforeSave", "Update"); len(got) != 0 {
		t.Fatalf("Approval primary authority leaked ordinary mutation subscriptions: %#v", got)
	}
	boundMutationView, ok := RuntimeMutationMetadataViewFromContext(bound)
	if !ok || boundMutationView.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		t.Fatalf("Approval-owned DML authority = %#v/%v, want record-mutation sibling", boundMutationView, ok)
	}
	if got := boundMutationView.TriggerSubscriptions("a01", "RecordBeforeSave", "Update"); len(got) != 1 || got[0].FlowDefinitionID != approvalBeforeSaveFlowDefinitionID {
		t.Fatalf("Approval-owned DML subscriptions = %#v, want %s", got, approvalBeforeSaveFlowDefinitionID)
	}
	mutationMetadata, err := metadata.NewRuntimeMetadataService(RuntimeMutationMetadataViewFromContext)
	if err != nil {
		t.Fatal(err)
	}
	durablePrimaryView, ok := mutationMetadata.BoundRuntimeMetadataView(bound)
	if !ok || durablePrimaryView != boundMutationView {
		t.Fatalf("durable record-event primary authority = %p/%v, want Mutation sibling %p", durablePrimaryView, ok, boundMutationView)
	}
}

func TestAssignmentPolicyLifecycleBindingReusesOnlyTheSameCatalogMutationView(t *testing.T) {
	prepared := func(source catalogsnapshot.Digest, catalogRevision, securityEpoch uint64) *preparedServingState {
		return &preparedServingState{
			selection: ServingSelection{
				OrgID: testOrgID, AppInstallationID: testInstallationID,
				RuntimeCapabilityID: mutationsemantic.CapabilityName, SecurityEpoch: securityEpoch,
			},
			binding: ServingHandlerBinding{
				AppReleaseDigest: "sha256:app", ServiceReleaseDigest: "sha256:service",
			},
			sourceSnapshotDigest: source,
			receipts: semantic.Receipts{
				CatalogRevision: catalogRevision, SchemaGeneration: 8,
				EngineGeneration: 9, SecurityEpoch: securityEpoch,
			},
		}
	}
	source := catalogsnapshot.Digest("sha256:source")
	if !sameAssignmentPolicyEligibilityServingView(prepared(source, 11, 3), prepared(source, 11, 4)) {
		t.Fatal("security-only refresh did not reuse the immutable Catalog lifecycle binding")
	}
	if sameAssignmentPolicyEligibilityServingView(prepared(source, 11, 3), prepared(catalogsnapshot.Digest("sha256:changed"), 11, 3)) {
		t.Fatal("different Catalog source snapshot reused the old lifecycle binding")
	}
	if sameAssignmentPolicyEligibilityServingView(prepared(source, 11, 3), prepared(source, 12, 3)) {
		t.Fatal("different Catalog revision reused the old lifecycle binding")
	}
}

type servingMemoryStore struct {
	artifacts map[string]catalogsnapshot.Artifact
	gets      atomic.Int64
}

func (s *servingMemoryStore) Put(_ context.Context, artifact catalogsnapshot.Artifact) error {
	if s.artifacts == nil {
		s.artifacts = map[string]catalogsnapshot.Artifact{}
	}
	s.artifacts[servingArtifactKey(artifact.Kind(), artifact.Digest())] = artifact
	return nil
}

func (s *servingMemoryStore) Get(_ context.Context, kind catalogsnapshot.ArtifactKind, digest catalogsnapshot.Digest) (catalogsnapshot.Artifact, error) {
	s.gets.Add(1)
	artifact, ok := s.artifacts[servingArtifactKey(kind, digest)]
	if !ok {
		return catalogsnapshot.Artifact{}, errors.New("artifact unavailable")
	}
	return artifact, nil
}

func servingArtifactKey(kind catalogsnapshot.ArtifactKind, digest catalogsnapshot.Digest) string {
	return string(kind) + "\x00" + string(digest)
}

type servingFixture struct {
	state        *ServingState
	admission    *ServingAdmission
	store        *servingMemoryStore
	catalogCalls atomic.Int64
	catalogFound bool
	installation ServingInstallation
	binding      ServingHandlerBinding
	selection    ServingSelection
}

type servingCapabilitySetEntry struct {
	appCapabilityID     string
	runtimeCapabilityID string
	profile             catalogsnapshot.ProjectionProfile
}

type servingSetFixture struct {
	state      *ServingState
	admission  *ServingAdmission
	selections []ServingSelection
}

func newServingSetFixture(
	t *testing.T,
	revision uint64,
	entries []servingCapabilitySetEntry,
) *servingSetFixture {
	return newServingSetPublicationFixture(t, revision, 1, 0, entries, nil)
}

func newServingSetPublicationFixture(t *testing.T, revision, securityEpoch, ownershipOffset uint64,
	entries, consumers []servingCapabilitySetEntry,
) *servingSetFixture {
	return newServingSetPublicationFixtureWithObjects(t, revision, securityEpoch, ownershipOffset, entries, consumers, nil)
}

func newServingSetPublicationFixtureWithObjects(t *testing.T, revision, securityEpoch, ownershipOffset uint64,
	entries, consumers []servingCapabilitySetEntry, objects []*models.ObjectMetadata,
) *servingSetFixture {
	t.Helper()
	object := &models.ObjectMetadata{
		ID: "object-widget", APIName: "Widget", Label: "Widget", PluralLabel: "Widgets",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields: []models.FieldMetadata{{
			ID: "field-widget-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString,
			PhysicalName: "id", IsSystem: true,
		}},
	}
	if objects == nil {
		objects = []*models.ObjectMetadata{object}
	}
	input := catalogsnapshot.CompileInput{
		OrgID: testOrgID,
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision:  catalogsnapshot.CatalogRevision(revision),
			SchemaGeneration: 8,
			EngineGeneration: 9,
		},
	}
	var allObjectNames []string
	for _, metadata := range objects {
		input.Objects = append(input.Objects, catalogsnapshot.ObjectInput{
			CatalogRevision: catalogsnapshot.CatalogRevision(revision), Metadata: metadata,
		})
		allObjectNames = append(allObjectNames, metadata.APIName)
	}
	for _, entry := range append(append([]servingCapabilitySetEntry(nil), entries...), consumers...) {
		objectNames := allObjectNames
		var componentIDs []string
		if entry.profile == catalogsnapshot.ProjectionCommandAcceptanceV1 {
			objectNames = nil
			componentIDs = []string{"component.command.flow"}
			input.CommandRegistrations = []catalogsnapshot.CommandRegistrationInput{{CatalogRevision: catalogsnapshot.CatalogRevision(revision), ComponentID: componentIDs[0], DefinitionVersion: "1", ContentDigest: catalogsnapshot.ComputeDigest([]byte("command")), CommandKind: "runtime.flow.execute", PayloadFormatVersion: 1, ExecutionBindings: []catalogsnapshot.CommandExecutionBinding{{AuthorizedWorkloadID: testWorkloadID, DefinitionRef: "flow.integration-ingress", DefinitionVersion: "flow-version-1", ContextNamespace: "connector.test"}}}}
		}
		input.Projections = append(input.Projections, catalogsnapshot.RuntimeProjectionInput{
			CatalogRevision:        catalogsnapshot.CatalogRevision(revision),
			AppInstallationID:      testInstallationID,
			CapabilityID:           entry.runtimeCapabilityID,
			Profile:                entry.profile,
			ObjectAPINames:         objectNames,
			DefinitionComponentIDs: componentIDs,
		})
	}
	for _, entry := range consumers {
		if entry.profile == catalogsnapshot.ProjectionPageLayoutResolutionV1 {
			input.PageLayouts = []catalogsnapshot.PageLayoutInput{{CatalogRevision: catalogsnapshot.CatalogRevision(revision), Layout: &models.PageLayout{ID: "layout-widget", ObjectAPIName: object.APIName, LayoutName: "Widget", CompactLayout: []string{}, Sections: []models.PageSection{}, RelatedLists: []models.RelatedListConfig{}, HeaderActions: []models.ActionConfig{}, QuickActions: []models.ActionConfig{}}}}
			input.LayoutAssignments = []catalogsnapshot.LayoutAssignmentInput{{CatalogRevision: catalogsnapshot.CatalogRevision(revision), Assignment: &models.LayoutAssignment{ID: "assignment-widget", ObjectAPIName: object.APIName, LayoutID: "layout-widget"}}}
		}
	}
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appReleaseDigest := catalogsnapshot.ComputeDigest([]byte("app-release"))
	serviceReleaseDigest := catalogsnapshot.ComputeDigest([]byte("service-release"))
	store := &servingMemoryStore{artifacts: map[string]catalogsnapshot.Artifact{}}
	if err := store.Put(t.Context(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	bindings := make(map[string]ServingHandlerBinding, len(entries))
	selections := make([]ServingSelection, 0, len(entries))
	for index, entry := range entries {
		viewArtifact, projectErr := catalogsnapshot.ProjectRuntimeMetadataView(
			snapshot,
			catalogsnapshot.ProjectionKey{
				AppInstallationID: testInstallationID,
				CapabilityID:      entry.runtimeCapabilityID,
			},
		)
		if projectErr != nil {
			t.Fatal(projectErr)
		}
		ownershipEpoch := uint64(index+3) + ownershipOffset
		receiptArtifact := servingReceiptAtSecurityEpoch(
			t,
			snapshotArtifact,
			viewArtifact,
			revision,
			appReleaseDigest,
			serviceReleaseDigest,
			ownershipEpoch,
			entry.appCapabilityID,
			entry.runtimeCapabilityID, securityEpoch,
		)
		for _, artifact := range []catalogsnapshot.Artifact{viewArtifact, receiptArtifact} {
			if err := store.Put(t.Context(), artifact); err != nil {
				t.Fatal(err)
			}
		}
		bindings[entry.appCapabilityID] = ServingHandlerBinding{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			AppReleaseDigest: string(appReleaseDigest), CapabilityID: entry.appCapabilityID,
			WorkloadID: testWorkloadID, ServiceReleaseDigest: string(serviceReleaseDigest),
			OwnershipEpoch:             ownershipEpoch,
			CompatibilityReceiptDigest: string(receiptArtifact.Digest()),
		}
		selections = append(selections, ServingSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			ExpectedWorkloadID:                 testWorkloadID,
			AppCapabilityID:                    entry.appCapabilityID,
			RuntimeCapabilityID:                entry.runtimeCapabilityID,
			ExpectedServiceReleaseDigest:       serviceReleaseDigest,
			ExpectedOwnershipEpoch:             ownershipEpoch,
			ExpectedCompatibilityReceiptDigest: receiptArtifact.Digest(),
			LocalCompatibility:                 catalogsnapshot.Compatibility{SchemaGeneration: 8, EngineGeneration: 9},
			SecurityEpoch:                      securityEpoch,
		})
	}
	state, err := NewServingState(ServingStateConfig{
		Store: store,
		Catalog: func(_ context.Context, installationID, capabilityID string) (ServingInstallation, ServingHandlerBinding, bool, error) {
			binding, found := bindings[capabilityID]
			if installationID != testInstallationID || !found {
				return ServingInstallation{}, ServingHandlerBinding{}, false, nil
			}
			return ServingInstallation{
				OrgID: testOrgID, AppInstallationID: testInstallationID,
				ActiveReleaseDigest: string(appReleaseDigest), CatalogRevision: revision,
			}, binding, true, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	admission, err := NewServingAdmission(state)
	if err != nil {
		t.Fatal(err)
	}
	return &servingSetFixture{state: state, admission: admission, selections: selections}
}

func newServingFixture(t *testing.T, revision uint64) *servingFixture {
	return newServingFixtureForCapability(t, revision, testSFCRMAppCapabilityID, accesssemantic.CapabilityName, catalogsnapshot.ProjectionRecordAccessV1)
}

func newServingFixtureForCapability(t *testing.T, revision uint64, appCapabilityID, runtimeCapabilityID string, profile catalogsnapshot.ProjectionProfile) *servingFixture {
	t.Helper()
	return newServingFixtureForCapabilityAndObject(t, revision, appCapabilityID, runtimeCapabilityID, profile, &models.ObjectMetadata{
		ID: "object-widget", APIName: "Widget", Label: "Widget", PluralLabel: "Widgets",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields:       []models.FieldMetadata{{ID: "field-widget-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true}},
	})
}

func newServingFixtureForCapabilityAndObject(t *testing.T, revision uint64, appCapabilityID, runtimeCapabilityID string, profile catalogsnapshot.ProjectionProfile, object *models.ObjectMetadata) *servingFixture {
	t.Helper()
	snapshotArtifact, viewArtifact := servingArtifactsForObject(t, revision, runtimeCapabilityID, profile, object)
	return newServingFixtureForArtifacts(
		t, revision, appCapabilityID, runtimeCapabilityID, snapshotArtifact, viewArtifact,
	)
}

func newServingFixtureForArtifacts(
	t *testing.T,
	revision uint64,
	appCapabilityID string,
	runtimeCapabilityID string,
	snapshotArtifact catalogsnapshot.Artifact,
	viewArtifact catalogsnapshot.Artifact,
) *servingFixture {
	t.Helper()
	appReleaseDigest := catalogsnapshot.ComputeDigest([]byte("app-release"))
	serviceReleaseDigest := catalogsnapshot.ComputeDigest([]byte("service-release"))
	receiptArtifact := servingReceipt(t, snapshotArtifact, viewArtifact, revision, appReleaseDigest, serviceReleaseDigest, 3, appCapabilityID, runtimeCapabilityID)
	store := &servingMemoryStore{artifacts: map[string]catalogsnapshot.Artifact{}}
	if err := store.Put(t.Context(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), viewArtifact); err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), receiptArtifact); err != nil {
		t.Fatal(err)
	}
	fixture := &servingFixture{
		store:        store,
		catalogFound: true,
		installation: ServingInstallation{
			OrgID: testOrgID, AppInstallationID: testInstallationID, ActiveReleaseDigest: string(appReleaseDigest), CatalogRevision: revision,
		},
		binding: ServingHandlerBinding{
			OrgID: testOrgID, AppInstallationID: testInstallationID, AppReleaseDigest: string(appReleaseDigest),
			CapabilityID: appCapabilityID, WorkloadID: testWorkloadID,
			ServiceReleaseDigest: string(serviceReleaseDigest), OwnershipEpoch: 3,
			CompatibilityReceiptDigest: string(receiptArtifact.Digest()),
		},
		selection: ServingSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			ExpectedWorkloadID: testWorkloadID,
			AppCapabilityID:    appCapabilityID, RuntimeCapabilityID: runtimeCapabilityID,
			ExpectedServiceReleaseDigest: serviceReleaseDigest, ExpectedOwnershipEpoch: 3,
			ExpectedCompatibilityReceiptDigest: receiptArtifact.Digest(),
			LocalCompatibility:                 catalogsnapshot.Compatibility{SchemaGeneration: 8, EngineGeneration: 9}, SecurityEpoch: 1,
		},
	}
	state, err := NewServingState(ServingStateConfig{
		Store: store,
		Catalog: func(_ context.Context, installationID, capabilityID string) (ServingInstallation, ServingHandlerBinding, bool, error) {
			fixture.catalogCalls.Add(1)
			if installationID != fixture.selection.AppInstallationID || capabilityID != fixture.selection.AppCapabilityID {
				return ServingInstallation{}, ServingHandlerBinding{}, false, nil
			}
			return fixture.installation, fixture.binding, fixture.catalogFound, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	fixture.state = state
	fixture.admission, err = NewServingAdmission(state)
	if err != nil {
		t.Fatal(err)
	}
	return fixture
}

func servingReceipt(t *testing.T, snapshotArtifact, viewArtifact catalogsnapshot.Artifact, revision uint64, appReleaseDigest, serviceReleaseDigest catalogsnapshot.Digest, ownershipEpoch uint64, appCapabilityID, runtimeCapabilityID string) catalogsnapshot.Artifact {
	return servingReceiptAtSecurityEpoch(t, snapshotArtifact, viewArtifact, revision, appReleaseDigest, serviceReleaseDigest, ownershipEpoch, appCapabilityID, runtimeCapabilityID, 1)
}

func servingReceiptAtSecurityEpoch(t *testing.T, snapshotArtifact, viewArtifact catalogsnapshot.Artifact, revision uint64, appReleaseDigest, serviceReleaseDigest catalogsnapshot.Digest, ownershipEpoch uint64, appCapabilityID, runtimeCapabilityID string, securityEpoch uint64) catalogsnapshot.Artifact {
	t.Helper()
	receipt, err := catalogsnapshot.BuildServingCompatibilityReceipt(catalogsnapshot.ServingCompatibilityReceiptInput{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		AppReleaseDigest: appReleaseDigest, AppCapabilityID: appCapabilityID,
		RuntimeCapabilityID: runtimeCapabilityID, WorkloadID: testWorkloadID,
		ServiceReleaseDigest: serviceReleaseDigest, OwnershipEpoch: ownershipEpoch,
		CatalogSnapshotDigest: snapshotArtifact.Digest(), RuntimeMetadataViewDigest: viewArtifact.Digest(),
		Lineage:       catalogsnapshot.Lineage{CatalogRevision: catalogsnapshot.CatalogRevision(revision), SchemaGeneration: 8, EngineGeneration: 9},
		SecurityEpoch: securityEpoch,
	})
	if err != nil {
		t.Fatal(err)
	}
	return receipt
}

func servingArtifacts(t *testing.T, revision uint64, runtimeCapabilityID string, profile catalogsnapshot.ProjectionProfile) (catalogsnapshot.Artifact, catalogsnapshot.Artifact) {
	t.Helper()
	return servingArtifactsForObject(t, revision, runtimeCapabilityID, profile, &models.ObjectMetadata{
		ID: "object-widget", APIName: "Widget", Label: "Widget", PluralLabel: "Widgets",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields:       []models.FieldMetadata{{ID: "field-widget-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true}},
	})
}

func servingArtifactsForObject(t *testing.T, revision uint64, runtimeCapabilityID string, profile catalogsnapshot.ProjectionProfile, object *models.ObjectMetadata) (catalogsnapshot.Artifact, catalogsnapshot.Artifact) {
	t.Helper()
	input := catalogsnapshot.CompileInput{
		OrgID: testOrgID,
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision: catalogsnapshot.CatalogRevision(revision), SchemaGeneration: 8, EngineGeneration: 9,
		},
		Objects: []catalogsnapshot.ObjectInput{{CatalogRevision: catalogsnapshot.CatalogRevision(revision), Metadata: object}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{
			CatalogRevision: catalogsnapshot.CatalogRevision(revision), AppInstallationID: testInstallationID,
			CapabilityID: runtimeCapabilityID, Profile: profile, ObjectAPINames: []string{object.APIName},
		}},
	}
	if profile == catalogsnapshot.ProjectionReportExecutionV1 {
		const componentID = "component.report.widget"
		input.ReportDefinitions = []catalogsnapshot.ReportDefinitionInput{{
			CatalogRevision:   catalogsnapshot.CatalogRevision(revision),
			ComponentID:       componentID,
			DefinitionVersion: "1",
			ContentDigest:     catalogsnapshot.ComputeDigest([]byte("report-definition")),
			Definition: &models.ReportDefinition{
				ID: "report-widget", Name: "Widget report", SourceObject: object.APIName,
				DisplayMode: "TABULAR", Columns: []string{"Id"},
			},
		}}
		input.Projections[0].DefinitionComponentIDs = []string{componentID}
	}
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: testInstallationID, CapabilityID: runtimeCapabilityID,
	})
	if err != nil {
		t.Fatal(err)
	}
	return snapshotArtifact, viewArtifact
}

func trustedServingContext(f *servingFixture) context.Context {
	claims := delegationv1.Claims{
		DelegationID: "delegation-1", OrgID: f.selection.OrgID, PrincipalID: "user-1", SessionID: "session-1",
		CallerWorkloadID: f.binding.WorkloadID, AppInstallationID: f.selection.AppInstallationID,
		RuntimeAudience: "runtime-1", Capabilities: []string{f.selection.RuntimeCapabilityID}, AuthEpoch: 1,
		SecurityEpoch: f.selection.SecurityEpoch, IssuedAt: 1, ExpiresAt: 2,
	}
	trusted := newTrustedCall(claims, models.UserSession{ID: "user-1"})
	return context.WithValue(context.Background(), trustedCallContextKey{}, trusted)
}

func TestServingAdmissionPreparesOnceAndBindsExactRequestView(t *testing.T) {
	fixture := newServingFixture(t, 7)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	preparedGets := fixture.store.gets.Load()
	if !fixture.admission.Ready() || fixture.catalogCalls.Load() != 1 || preparedGets == 0 {
		t.Fatalf("prepare counts: ready=%v catalog=%d artifacts=%d", fixture.admission.Ready(), fixture.catalogCalls.Load(), fixture.store.gets.Load())
	}
	reader, err := metadata.NewRuntimeMetadataService(RuntimeMetadataViewFromContext)
	if err != nil {
		t.Fatal(err)
	}
	var previousRequestSchema *models.ObjectMetadata
	for range 2 {
		ctx, err := fixture.admission.bind(trustedServingContext(fixture), accesssemantic.CapabilityName)
		if err != nil {
			t.Fatal(err)
		}
		scope, ok := runtimeExecutionScopeFromContext(ctx)
		if !ok || scope.view.Digest() == "" || scope.receipts.CatalogRevision != 7 || scope.receipts.SecurityEpoch != 1 {
			t.Fatalf("unexpected execution scope: ok=%v scope=%+v", ok, scope)
		}
		neutralView, ok := metadata.RuntimeMetadataViewFromContext(ctx)
		if !ok || neutralView != scope.view {
			t.Fatalf("neutral Runtime view = %p/%v, want admitted view %p/true", neutralView, ok, scope.view)
		}
		firstEngine, bound, err := metadata.RuntimeFormulaEngineFromContext(ctx)
		if err != nil || !bound {
			t.Fatalf("first Runtime formula engine = %p/%v/%v, want bound engine", firstEngine, bound, err)
		}
		secondEngine, bound, err := metadata.RuntimeFormulaEngineFromContext(ctx)
		if err != nil || !bound || secondEngine != firstEngine {
			t.Fatalf("second Runtime formula engine = %p/%v/%v, want request-cached %p", secondEngine, bound, err, firstEngine)
		}
		first := reader.GetSchema(ctx, "Widget")
		second := reader.GetSchema(ctx, "widget")
		if first == nil || second != first {
			t.Fatalf("admitted request did not reuse one detached schema: first=%#v second=%#v", first, second)
		}
		if previousRequestSchema == first {
			t.Fatal("two admitted requests shared a mutable detached schema")
		}
		previousRequestSchema = first
	}
	if fixture.catalogCalls.Load() != 1 || fixture.store.gets.Load() != preparedGets {
		t.Fatal("request admission consulted Catalog or artifact store")
	}
}

func TestServingStateSeparatesSFCRMHandlerBindingFromRuntimeCapability(t *testing.T) {
	fixture := newServingFixture(t, 7)
	if fixture.binding.CapabilityID != "io.nexuscrm.sfcrm.ui.record-access.v1" ||
		fixture.selection.RuntimeCapabilityID != "runtime.records.evaluate-access.v1" {
		t.Fatalf("fixture no longer models the immutable sfcrm AppRelease capability pair: binding=%q runtime=%q",
			fixture.binding.CapabilityID, fixture.selection.RuntimeCapabilityID)
	}
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatalf("distinct App and Runtime capabilities did not reconcile: %v", err)
	}
	ctx, err := fixture.admission.bind(trustedServingContext(fixture), fixture.selection.RuntimeCapabilityID)
	if err != nil {
		t.Fatal(err)
	}
	scope, ok := runtimeExecutionScopeFromContext(ctx)
	if !ok || scope.view.CapabilityID() != fixture.selection.RuntimeCapabilityID ||
		scope.receipts.CapabilityVersion != fixture.selection.RuntimeCapabilityID {
		t.Fatalf("Runtime execution selected the App handler capability: ok=%v view=%q receipts=%q",
			ok, scope.view.CapabilityID(), scope.receipts.CapabilityVersion)
	}

	// Reusing the Runtime data capability as a Catalog HandlerBinding selector
	// is not an inference rule. The real AppRelease declares these as two
	// different axes, so the selection must fail before artifact access.
	bad := fixture.selection
	bad.AppCapabilityID = bad.RuntimeCapabilityID
	if err := fixture.state.Reconcile(t.Context(), bad); err == nil {
		t.Fatal("Runtime capability was accepted as the sfcrm HandlerBinding capability")
	}
	if fixture.state.Ready() {
		t.Fatal("invalid capability-axis selection retained readiness")
	}
}

func TestServingStateRequiresExactDeploymentAssertionsAndCatalogReceipt(t *testing.T) {
	for _, test := range []struct {
		name   string
		mutate func(*servingFixture)
	}{
		{name: "service release", mutate: func(f *servingFixture) {
			f.binding.ServiceReleaseDigest = string(catalogsnapshot.ComputeDigest([]byte("other-service")))
		}},
		{name: "workload owner", mutate: func(f *servingFixture) { f.binding.WorkloadID = "workload-other" }},
		{name: "ownership epoch", mutate: func(f *servingFixture) { f.binding.OwnershipEpoch++ }},
		{name: "receipt selection", mutate: func(f *servingFixture) {
			f.binding.CompatibilityReceiptDigest = string(catalogsnapshot.ComputeDigest([]byte("other-receipt")))
		}},
	} {
		t.Run(test.name, func(t *testing.T) {
			fixture := newServingFixture(t, 7)
			test.mutate(fixture)
			if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
				t.Fatal("mismatched Catalog ownership was accepted")
			}
			if fixture.state.Ready() {
				t.Fatal("mismatched ownership retained readiness")
			}
		})
	}
}

func TestServingAdmissionFailsClosedForAuthorityAndReceiptMismatch(t *testing.T) {
	fixture := newServingFixture(t, 7)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	tests := []struct {
		name   string
		mutate func(*TrustedCall)
		want   semantic.FailureCategory
	}{
		{name: "wrong org", mutate: func(call *TrustedCall) { call.orgID = "org-other" }, want: semantic.FailureForbidden},
		{name: "wrong installation", mutate: func(call *TrustedCall) { call.appInstallationID = "installation-other" }, want: semantic.FailureForbidden},
		{name: "wrong workload", mutate: func(call *TrustedCall) { call.callerWorkloadID = "workload-other" }, want: semantic.FailureForbidden},
		{name: "missing capability", mutate: func(call *TrustedCall) { call.capabilities = nil }, want: semantic.FailureForbidden},
		{name: "stale security epoch", mutate: func(call *TrustedCall) { call.securityEpoch-- }, want: semantic.FailureStaleState},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			ctx := trustedServingContext(fixture)
			call, _ := trustedCallFromContext(ctx)
			test.mutate(&call)
			ctx = context.WithValue(context.Background(), trustedCallContextKey{}, call)
			_, err := fixture.admission.bind(ctx, accesssemantic.CapabilityName)
			var failure *semantic.Failure
			if !errors.As(err, &failure) || failure.Category != test.want {
				t.Fatalf("error = %v, want %s", err, test.want)
			}
		})
	}
}

func TestServingStateActivationDoesNotMixInFlightRevision(t *testing.T) {
	fixture := newServingFixture(t, 7)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	inFlight, err := fixture.admission.bind(trustedServingContext(fixture), accesssemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}

	newSnapshot, newView := servingArtifacts(t, 8, accesssemantic.CapabilityName, catalogsnapshot.ProjectionRecordAccessV1)
	_ = fixture.store.Put(t.Context(), newSnapshot)
	_ = fixture.store.Put(t.Context(), newView)
	newReceipt := servingReceipt(t, newSnapshot, newView, 8, catalogsnapshot.Digest(fixture.installation.ActiveReleaseDigest), fixture.selection.ExpectedServiceReleaseDigest, fixture.selection.ExpectedOwnershipEpoch, fixture.selection.AppCapabilityID, fixture.selection.RuntimeCapabilityID)
	_ = fixture.store.Put(t.Context(), newReceipt)
	fixture.installation.CatalogRevision = 8
	fixture.binding.CompatibilityReceiptDigest = string(newReceipt.Digest())
	fixture.selection.ExpectedCompatibilityReceiptDigest = newReceipt.Digest()
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	fresh, err := fixture.admission.bind(trustedServingContext(fixture), accesssemantic.CapabilityName)
	if err != nil {
		t.Fatal(err)
	}
	oldScope, _ := runtimeExecutionScopeFromContext(inFlight)
	newScope, _ := runtimeExecutionScopeFromContext(fresh)
	if oldScope.receipts.CatalogRevision != 7 || newScope.receipts.CatalogRevision != 8 || oldScope.view.Digest() == newScope.view.Digest() {
		t.Fatalf("mixed activation scopes: old=%+v new=%+v", oldScope.receipts, newScope.receipts)
	}
}

func TestServingStateRejectsWrongProfileAndDropsReadiness(t *testing.T) {
	fixture := newServingFixture(t, 7)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	snapshot, view := servingArtifacts(t, 7, accesssemantic.CapabilityName, catalogsnapshot.ProjectionSelectedRootReadV1)
	_ = fixture.store.Put(t.Context(), snapshot)
	_ = fixture.store.Put(t.Context(), view)
	receipt := servingReceipt(t, snapshot, view, 7, catalogsnapshot.Digest(fixture.installation.ActiveReleaseDigest), fixture.selection.ExpectedServiceReleaseDigest, fixture.selection.ExpectedOwnershipEpoch, fixture.selection.AppCapabilityID, fixture.selection.RuntimeCapabilityID)
	_ = fixture.store.Put(t.Context(), receipt)
	fixture.binding.CompatibilityReceiptDigest = string(receipt.Digest())
	fixture.selection.ExpectedCompatibilityReceiptDigest = receipt.Digest()
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
		t.Fatal("wrong projection profile was accepted")
	}
	if fixture.state.Ready() {
		t.Fatal("failed reconcile retained readiness")
	}
}

func TestServingStateRejectsHistoricalRecordViewAndDropsReadiness(t *testing.T) {
	tests := []struct {
		name          string
		appCapability string
		runtime       string
	}{
		{name: "record-view", appCapability: "io.nexuscrm.sfcrm.ui.record-view-read.v1", runtime: recordviewsemantic.CapabilityName},
		{name: "attachment-source", appCapability: "io.nexuscrm.sfcrm.ui.record-view-attachments.v1", runtime: attachmentsemantic.CapabilityName},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			assertHistoricalRecordViewRejected(t, test.appCapability, test.runtime)
		})
	}
}

func assertHistoricalRecordViewRejected(t *testing.T, appCapability, runtimeCapability string) {
	t.Helper()
	fixture := newServingFixtureForCapability(
		t, 7, appCapability, runtimeCapability, catalogsnapshot.ProjectionRecordViewV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}

	receiptArtifact, err := fixture.store.Get(
		t.Context(), catalogsnapshot.ArtifactServingCompatibilityReceipt,
		catalogsnapshot.Digest(fixture.binding.CompatibilityReceiptDigest),
	)
	if err != nil {
		t.Fatal(err)
	}
	currentReceipt, err := catalogsnapshot.OpenServingCompatibilityReceipt(receiptArtifact)
	if err != nil {
		t.Fatal(err)
	}
	snapshotArtifact, err := fixture.store.Get(
		t.Context(), catalogsnapshot.ArtifactCatalogSnapshot, currentReceipt.CatalogSnapshotDigest(),
	)
	if err != nil {
		t.Fatal(err)
	}
	currentView, err := fixture.store.Get(
		t.Context(), catalogsnapshot.ArtifactRuntimeMetadataView, currentReceipt.RuntimeMetadataViewDigest(),
	)
	if err != nil {
		t.Fatal(err)
	}
	historicalView := historicalRecordViewV6(t, currentView)
	if err := fixture.store.Put(t.Context(), historicalView); err != nil {
		t.Fatal(err)
	}
	historicalReceipt := servingReceipt(
		t, snapshotArtifact, historicalView, 7,
		catalogsnapshot.Digest(fixture.installation.ActiveReleaseDigest),
		fixture.selection.ExpectedServiceReleaseDigest,
		fixture.selection.ExpectedOwnershipEpoch,
		fixture.selection.AppCapabilityID,
		fixture.selection.RuntimeCapabilityID,
	)
	if err := fixture.store.Put(t.Context(), historicalReceipt); err != nil {
		t.Fatal(err)
	}
	fixture.binding.CompatibilityReceiptDigest = string(historicalReceipt.Digest())
	fixture.selection.ExpectedCompatibilityReceiptDigest = historicalReceipt.Digest()

	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
		t.Fatal("historical record-view metadata was accepted for serving")
	}
	if fixture.state.Ready() {
		t.Fatal("failed historical-view reconcile retained readiness")
	}
}

func historicalRecordViewV6(t *testing.T, current catalogsnapshot.Artifact) catalogsnapshot.Artifact {
	t.Helper()
	currentSchema := []byte(catalogsnapshot.RuntimeMetadataViewSchemaV24)
	historicalSchema := []byte(catalogsnapshot.RuntimeMetadataViewSchemaV6)
	data := bytes.Replace(current.Bytes(), currentSchema, historicalSchema, 1)
	if bytes.Equal(data, current.Bytes()) {
		t.Fatal("current record-view fixture did not contain the v24 schema")
	}
	digest := catalogsnapshot.ComputeDigest(data)
	root := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(root)
	if err != nil {
		t.Fatal(err)
	}
	directory := filepath.Join(root, "runtime-metadata-view-v2")
	if err := os.MkdirAll(directory, 0o750); err != nil {
		t.Fatal(err)
	}
	path := filepath.Join(directory, strings.TrimPrefix(string(digest), "sha256:")+".json")
	if err := os.WriteFile(path, data, 0o440); err != nil {
		t.Fatal(err)
	}
	historical, err := store.Get(t.Context(), catalogsnapshot.ArtifactRuntimeMetadataView, digest)
	if err != nil {
		t.Fatalf("open historical record-view fixture: %v", err)
	}
	return historical
}

func TestServingStatePublishesCompleteCapabilitySetAtomically(t *testing.T) {
	const queryAppCapability = "io.nexuscrm.sfcrm.query.v1"
	const recordViewAppCapability = "io.nexuscrm.sfcrm.ui.record-view-read.v1"
	const attachmentSourceAppCapability = "io.nexuscrm.sfcrm.ui.record-view-attachments.v1"
	const mutationAppCapability = "io.nexuscrm.sfcrm.record-mutation.v1"
	const recycleMutationAppCapability = "io.nexuscrm.sfcrm.recycle-bin-mutation.v1"
	const flowAppCapability = "io.nexuscrm.sfcrm.flow.v1"
	const appMetadataAppCapability = "io.nexuscrm.sfcrm.ui-apps.v1"
	revision := uint64(11)
	lineage := catalogsnapshot.Lineage{CatalogRevision: catalogsnapshot.CatalogRevision(revision), SchemaGeneration: 8, EngineGeneration: 9}
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: testOrgID, Lineage: lineage,
		Objects: []catalogsnapshot.ObjectInput{{CatalogRevision: lineage.CatalogRevision, Metadata: &models.ObjectMetadata{
			ID: "object-widget", APIName: "Widget", Label: "Widget", PluralLabel: "Widgets",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields:       []models.FieldMetadata{{ID: "field-widget-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true}},
		}}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: accesssemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordAccessV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: querysemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordQueryV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: recordviewsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordViewV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: attachmentsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordViewV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: mutationsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordMutationV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: recyclemutationsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordMutationV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: flowsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordMutationV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: catalogsemantic.AppMetadataV3CapabilityName, Profile: catalogsnapshot.ProjectionRecordQueryV1, ObjectAPINames: []string{"Widget"}},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appRelease := catalogsnapshot.ComputeDigest([]byte("app-release"))
	serviceRelease := catalogsnapshot.ComputeDigest([]byte("service-release"))
	store := &servingMemoryStore{artifacts: map[string]catalogsnapshot.Artifact{}}
	if err := store.Put(t.Context(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	type capabilityFixture struct {
		appCapability string
		runtime       string
		workload      string
		epoch         uint64
		view          catalogsnapshot.Artifact
		receipt       catalogsnapshot.Artifact
	}
	capabilities := []capabilityFixture{
		{appCapability: testSFCRMAppCapabilityID, runtime: accesssemantic.CapabilityName, workload: testWorkloadID, epoch: 3},
		{appCapability: queryAppCapability, runtime: querysemantic.CapabilityName, workload: "integration-workload", epoch: 4},
		{appCapability: recordViewAppCapability, runtime: recordviewsemantic.CapabilityName, workload: testWorkloadID, epoch: 5},
		// A binding carries one exact compatibility receipt, so the proof-bearing
		// attachment call needs its own explicit L1 capability even though it
		// shares the record-view metadata profile and AppService workload.
		{appCapability: attachmentSourceAppCapability, runtime: attachmentsemantic.CapabilityName, workload: testWorkloadID, epoch: 6},
		{appCapability: mutationAppCapability, runtime: mutationsemantic.CapabilityName, workload: testWorkloadID, epoch: 7},
		{appCapability: recycleMutationAppCapability, runtime: recyclemutationsemantic.CapabilityName, workload: testWorkloadID, epoch: 8},
		{appCapability: flowAppCapability, runtime: flowsemantic.CapabilityName, workload: testWorkloadID, epoch: 9},
		{appCapability: appMetadataAppCapability, runtime: catalogsemantic.AppMetadataV3CapabilityName, workload: testWorkloadID, epoch: 10},
	}
	bindings := map[string]ServingHandlerBinding{}
	selections := make([]ServingSelection, 0, len(capabilities))
	for i := range capabilities {
		entry := &capabilities[i]
		entry.view, err = catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{AppInstallationID: testInstallationID, CapabilityID: entry.runtime})
		if err != nil {
			t.Fatal(err)
		}
		entry.receipt, err = catalogsnapshot.BuildServingCompatibilityReceipt(catalogsnapshot.ServingCompatibilityReceiptInput{
			OrgID: testOrgID, AppInstallationID: testInstallationID, AppReleaseDigest: appRelease,
			AppCapabilityID: entry.appCapability, RuntimeCapabilityID: entry.runtime, WorkloadID: entry.workload,
			ServiceReleaseDigest: serviceRelease, OwnershipEpoch: entry.epoch,
			CatalogSnapshotDigest: snapshotArtifact.Digest(), RuntimeMetadataViewDigest: entry.view.Digest(), Lineage: lineage, SecurityEpoch: 1,
		})
		if err != nil {
			t.Fatal(err)
		}
		if err := store.Put(t.Context(), entry.view); err != nil {
			t.Fatal(err)
		}
		if err := store.Put(t.Context(), entry.receipt); err != nil {
			t.Fatal(err)
		}
		bindings[entry.appCapability] = ServingHandlerBinding{
			OrgID: testOrgID, AppInstallationID: testInstallationID, AppReleaseDigest: string(appRelease), CapabilityID: entry.appCapability,
			WorkloadID: entry.workload, ServiceReleaseDigest: string(serviceRelease), OwnershipEpoch: entry.epoch,
			CompatibilityReceiptDigest: string(entry.receipt.Digest()),
		}
		selections = append(selections, ServingSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID, ExpectedWorkloadID: entry.workload,
			AppCapabilityID: entry.appCapability, RuntimeCapabilityID: entry.runtime,
			ExpectedServiceReleaseDigest: serviceRelease, ExpectedOwnershipEpoch: entry.epoch,
			ExpectedCompatibilityReceiptDigest: entry.receipt.Digest(),
			LocalCompatibility:                 catalogsnapshot.Compatibility{SchemaGeneration: 8, EngineGeneration: 9}, SecurityEpoch: 1,
		})
	}
	appView := catalogsemantic.AppMetadataViewV3{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		AppID: "io.nexuscrm.sfcrm", AppReleaseDigest: string(appRelease),
		Lineage: catalogsemantic.ViewLineage{
			CatalogRevision: revision, SchemaGeneration: 8, EngineGeneration: 9,
			SourceSnapshotDigest: string(snapshotArtifact.Digest()),
		},
		Receipts: catalogsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: catalogsemantic.AppMetadataV3CapabilityName,
			CatalogRevision: revision, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 1,
		},
	}
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(appView)
	appMetadata := &flowServerTestCatalog{viewV3: appView, requireDeadline: true}
	state, err := NewServingState(ServingStateConfig{
		Store: store,
		Catalog: func(_ context.Context, _, capability string) (ServingInstallation, ServingHandlerBinding, bool, error) {
			binding, ok := bindings[capability]
			return ServingInstallation{OrgID: testOrgID, AppInstallationID: testInstallationID, ActiveReleaseDigest: string(appRelease), CatalogRevision: revision}, binding, ok, nil
		},
		AppMetadata: appMetadata,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := state.ReconcileAll(t.Context(), selections); err != nil {
		t.Fatal(err)
	}
	if appMetadata.v3Calls != 1 {
		t.Fatalf("initial serving reconcile AppMetadata V3 reads = %d, want 1", appMetadata.v3Calls)
	}
	admission, err := NewServingAdmission(state)
	if err != nil {
		t.Fatal(err)
	}
	mutationMetadata, err := metadata.NewRuntimeMetadataService(RuntimeMutationMetadataViewFromContext)
	if err != nil {
		t.Fatal(err)
	}
	var boundFlowContext context.Context
	for _, capability := range []string{accesssemantic.CapabilityName, querysemantic.CapabilityName, recordviewsemantic.CapabilityName, attachmentsemantic.CapabilityName, mutationsemantic.CapabilityName, recyclemutationsemantic.CapabilityName, flowsemantic.CapabilityName} {
		workloadID := testWorkloadID
		if capability == querysemantic.CapabilityName {
			workloadID = "integration-workload"
		}
		claims := delegationv1.Claims{
			DelegationID: "delegation-1", OrgID: testOrgID, PrincipalID: "user-1", SessionID: "session-1",
			CallerWorkloadID: workloadID, AppInstallationID: testInstallationID, RuntimeAudience: "runtime-1",
			Capabilities: []string{accesssemantic.CapabilityName, querysemantic.CapabilityName, recordviewsemantic.CapabilityName, attachmentsemantic.CapabilityName, mutationsemantic.CapabilityName, recyclemutationsemantic.CapabilityName, flowsemantic.CapabilityName}, AuthEpoch: 1, SecurityEpoch: 1, IssuedAt: 1, ExpiresAt: 2,
		}
		ctx := context.WithValue(context.Background(), trustedCallContextKey{}, newTrustedCall(claims, models.UserSession{ID: "user-1"}))
		bound, err := admission.bind(ctx, capability)
		if err != nil {
			t.Fatalf("bind %s: %v", capability, err)
		}
		scope, ok := runtimeExecutionScopeFromContext(bound)
		if !ok || scope.view.CapabilityID() != capability || scope.receipts.CapabilityVersion != capability {
			t.Fatalf("capability %s borrowed scope %#v", capability, scope)
		}
		flowView, hasFlowView := metadata.RuntimeFlowWorkViewFromContext(bound)
		if capability == mutationsemantic.CapabilityName || capability == recyclemutationsemantic.CapabilityName || capability == flowsemantic.CapabilityName {
			if !hasFlowView || flowView.CapabilityID() != flowsemantic.CapabilityName ||
				flowView.Lineage() != scope.view.Lineage() {
				t.Fatalf("mutation auxiliary Flow view = (%#v, %v), mutation lineage = %#v", flowView, hasFlowView, scope.view.Lineage())
			}
		} else if hasFlowView {
			t.Fatalf("capability %s received mutation-only auxiliary Flow authority", capability)
		}
		preparedAppView, hasPreparedAppView := runtimeFlowAppMetadataFromContext(bound)
		if capability == flowsemantic.CapabilityName {
			boundFlowContext = bound
			if !hasPreparedAppView || preparedAppView.Lineage.CatalogRevision != revision ||
				preparedAppView.Receipts.SecurityEpoch != 1 {
				t.Fatalf(
					"Flow prepared AppMetadata = (%#v, %v), want revision %d security epoch 1",
					preparedAppView, hasPreparedAppView, revision,
				)
			}
		} else if hasPreparedAppView {
			t.Fatalf("capability %s received Flow-only AppMetadata authority", capability)
		}
		engineView, engineViewOK := RuntimeMutationMetadataViewFromContext(bound)
		if !engineViewOK {
			t.Fatalf("capability %s has no mutation engine view", capability)
		}
		wantEngineCapability := capability
		if capability == recyclemutationsemantic.CapabilityName || capability == flowsemantic.CapabilityName {
			wantEngineCapability = mutationsemantic.CapabilityName
		}
		if engineView.CapabilityID() != wantEngineCapability {
			t.Fatalf("capability %s mutation engine view = %s, want %s", capability, engineView.CapabilityID(), wantEngineCapability)
		}
		if capability == recyclemutationsemantic.CapabilityName {
			durableView, durable := mutationMetadata.BoundRuntimeMetadataView(bound)
			if !durable || durableView != engineView {
				t.Fatalf("Recycle Bin durable work view = %p/%v, want ordinary Mutation engine view %p", durableView, durable, engineView)
			}
		}
	}
	if appMetadata.v3Calls != 1 {
		t.Fatalf("request admission caused AppMetadata reads: got %d, want startup-only 1", appMetadata.v3Calls)
	}
	originalAppView := appMetadata.viewV3
	appMetadata.viewV3.AppReleaseDigest = "sha256:" + strings.Repeat("e", 64)
	appMetadata.viewV3.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV3Digest(appMetadata.viewV3)
	if err := state.ReconcileAll(t.Context(), selections); err == nil {
		t.Fatal("serving reconcile accepted AppMetadata from a different AppRelease")
	}
	if state.Ready() {
		t.Fatal("failed AppMetadata reconcile retained stale readiness")
	}
	oldRequestView, oldRequestOK := runtimeFlowAppMetadataFromContext(boundFlowContext)
	if !oldRequestOK || oldRequestView.AppInstallationID != testInstallationID ||
		oldRequestView.Lineage.CatalogRevision != revision {
		t.Fatalf("in-flight request lost its captured immutable AppMetadata: (%#v, %v)", oldRequestView, oldRequestOK)
	}
	appMetadata.viewV3 = originalAppView
	if err := state.ReconcileAll(t.Context(), selections); err != nil {
		t.Fatalf("serving reconcile did not recover after valid AppMetadata returned: %v", err)
	}
	if appMetadata.v3Calls != 3 {
		t.Fatalf("AppMetadata reads across initial/failure/recovery reconcile = %d, want 3", appMetadata.v3Calls)
	}
	broken := append([]ServingSelection(nil), selections...)
	broken[1].ExpectedCompatibilityReceiptDigest = catalogsnapshot.ComputeDigest([]byte("missing"))
	if err := state.ReconcileAll(t.Context(), broken); err == nil {
		t.Fatal("partial capability refresh succeeded")
	}
	if state.Ready() {
		t.Fatal("failed complete-set refresh retained a partial or stale ready set")
	}

	// A Catalog revision number is not a content identity. Two projections
	// prepared from different snapshots at the same revision must not be
	// published as one serving set.
	changedSnapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: testOrgID, Lineage: lineage,
		Objects: []catalogsnapshot.ObjectInput{{CatalogRevision: lineage.CatalogRevision, Metadata: &models.ObjectMetadata{
			ID: "object-widget", APIName: "Widget", Label: "Changed Widget", PluralLabel: "Widgets",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields:       []models.FieldMetadata{{ID: "field-widget-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true}},
		}}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: accesssemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordAccessV1, ObjectAPINames: []string{"Widget"}},
			{CatalogRevision: lineage.CatalogRevision, AppInstallationID: testInstallationID, CapabilityID: querysemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordQueryV1, ObjectAPINames: []string{"Widget"}},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	changedSnapshot, err := catalogsnapshot.OpenCatalogSnapshot(changedSnapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	changedQueryView, err := catalogsnapshot.ProjectRuntimeMetadataView(changedSnapshot, catalogsnapshot.ProjectionKey{AppInstallationID: testInstallationID, CapabilityID: querysemantic.CapabilityName})
	if err != nil {
		t.Fatal(err)
	}
	changedQueryReceipt, err := catalogsnapshot.BuildServingCompatibilityReceipt(catalogsnapshot.ServingCompatibilityReceiptInput{
		OrgID: testOrgID, AppInstallationID: testInstallationID, AppReleaseDigest: appRelease,
		AppCapabilityID: queryAppCapability, RuntimeCapabilityID: querysemantic.CapabilityName, WorkloadID: "integration-workload",
		ServiceReleaseDigest: serviceRelease, OwnershipEpoch: capabilities[1].epoch,
		CatalogSnapshotDigest: changedSnapshotArtifact.Digest(), RuntimeMetadataViewDigest: changedQueryView.Digest(), Lineage: lineage, SecurityEpoch: 1,
	})
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range []catalogsnapshot.Artifact{changedSnapshotArtifact, changedQueryView, changedQueryReceipt} {
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	mixed := append([]ServingSelection(nil), selections...)
	mixed[1].ExpectedCompatibilityReceiptDigest = changedQueryReceipt.Digest()
	queryBinding := bindings[queryAppCapability]
	queryBinding.CompatibilityReceiptDigest = string(changedQueryReceipt.Digest())
	bindings[queryAppCapability] = queryBinding
	if err := state.ReconcileAll(t.Context(), mixed); err == nil {
		t.Fatal("capability set mixed two Catalog snapshot digests at one revision")
	}
	if state.Ready() {
		t.Fatal("mixed-snapshot capability set became ready")
	}
}

func TestServingStateFailsClosedWhenFlowHasNoMutationSibling(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t,
		12,
		"io.nexuscrm.sfcrm.automation.execute-flow.v1",
		flowsemantic.CapabilityName,
		catalogsnapshot.ProjectionRecordMutationV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
		t.Fatal("Flow-only serving set became ready without record-mutation authority")
	}
	if fixture.state.Ready() {
		t.Fatal("failed Flow-only reconcile retained readiness")
	}
}

func TestServingStateFailsClosedWhenApprovalHasNoMutationSibling(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t,
		12,
		"io.nexuscrm.sfcrm.approval.execute.v1",
		approvalsemantic.ExecuteCapabilityName,
		catalogsnapshot.ProjectionApprovalExecutionV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
		t.Fatal("Approval-only serving set became ready without record-mutation authority")
	}
	if fixture.state.Ready() {
		t.Fatal("failed Approval-only reconcile retained readiness")
	}
}

func TestServingStateFailsClosedWhenPermissionAuthoringHasNoMutationSibling(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t,
		12,
		"io.nexuscrm.sfcrm.permissions.authoring.v1",
		permissionauthoringsemantic.CapabilityName,
		catalogsnapshot.ProjectionRecordMutationV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
		t.Fatal("Permission Authoring-only serving set became ready without record-mutation authority")
	}
	if fixture.state.Ready() {
		t.Fatal("failed Permission Authoring-only reconcile retained readiness")
	}
}