package runtimeproduction

import (
	"context"
	"crypto/ed25519"
	"encoding/base64"
	"encoding/json"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/backend/internal/runtimeapi"
	"github.com/nexuscrm/backend/internal/runtimeconfig"
	"github.com/nexuscrm/backend/internal/runtimeidentity"
	"github.com/nexuscrm/backend/internal/runtimepermissionstate"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	approvalnotificationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalnotificationflowv1"
	filessemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/filesv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	mailboxauthorizationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/mailboxauthorizationv1"
	metadataadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/metadataadminreadv1"
	outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	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"
	recordsemantic "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"
	recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1"
	sessionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/sessionv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

func TestPermissionStateRoleForServingWriters(t *testing.T) {
	tests := []struct {
		name                                                           string
		recordMutation, recycleMutation, outbound, permissionAuthoring bool
	}{
		{name: mutationsemantic.CapabilityName, recordMutation: true},
		{name: recyclemutationsemantic.CapabilityName, recycleMutation: true},
		{name: "runtime outbound intent consumer", outbound: true},
		{name: permissionauthoringsemantic.CapabilityName, permissionAuthoring: true},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			if got := permissionStateRoleForServing(test.recordMutation, test.recycleMutation, test.outbound, test.permissionAuthoring); got != runtimepermissionstate.RoleReaderWriter {
				t.Fatalf("permission state role = %q, want %q", got, runtimepermissionstate.RoleReaderWriter)
			}
		})
	}
	if got := permissionStateRoleForServing(false, false, false, false); got != runtimepermissionstate.RoleReader {
		t.Fatalf("read-only permission state role = %q, want %q", got, runtimepermissionstate.RoleReader)
	}
}

func TestProductionServingCapabilitiesHaveOneRuntimeAdmissionOwner(t *testing.T) {
	for _, definition := range productionRuntimeCapabilityDefinitions {
		if !runtimeapi.UsesServingAdmission(definition.id) && definition.selectionModes == 0 {
			t.Fatalf("production capability %q has no selectable admission lane", definition.id)
		}
	}
}

func TestFederatedIDTokenVerifiersAreConfiguredByNeutralProviderIdentity(t *testing.T) {
	providers := []runtimeconfig.FederatedOIDCProvider{
		{
			ProviderID: "provider-a", Audience: "client-a", SessionScope: "api",
			TrustedIssuers: []string{"https://issuer-a.example"},
			JWKSURL:        "https://issuer-a.example/jwks.json",
		},
		{
			ProviderID: "provider-b", Audience: "client-b", SessionScope: "api",
			TrustedIssuers: []string{"https://issuer-b.example"},
			JWKSURL:        "https://issuer-b.example/jwks.json",
		},
	}
	verifiers, err := newFederatedIDTokenVerifiers(providers)
	if err != nil {
		t.Fatal(err)
	}
	if len(verifiers) != 2 || verifiers["provider-a"] == nil ||
		verifiers["provider-b"] == nil || verifiers["provider-c"] != nil {
		t.Fatalf("federated verifier registry = %#v", verifiers)
	}
	if scopes := federatedSessionScopes(providers); len(scopes) != 2 ||
		scopes["provider-a"] != "api" || scopes["provider-b"] != "api" {
		t.Fatalf("federated session scopes = %#v", scopes)
	}

	providers[0].Audience = ""
	if _, err := newFederatedIDTokenVerifiers(providers); err == nil ||
		!strings.Contains(err.Error(), `provider-a`) {
		t.Fatalf("invalid provider error = %v", err)
	}
}

func TestAssembleRejectsInvalidAuthorityConfigurationBeforeDatabaseOpen(t *testing.T) {
	base := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		Capabilities: []runtimeconfig.ServingCapability{
			{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", RuntimeCapabilityID: accesssemantic.CapabilityName, AdmissionLease: "1s"},
			{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", RuntimeCapabilityID: querysemantic.CapabilityName, AdmissionLease: "1s"},
		},
	}}
	tests := []struct {
		name      string
		mutate    func(*runtimeconfig.API)
		wantError string
	}{
		{
			name: "unsupported capability",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID = "runtime.records.read-selected-root.v1"
			},
			wantError: "unsupported Runtime capability",
		},
		{
			name: "session cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID = sessionsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "application result cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					applicationresultsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "mailbox authorization cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					mailboxauthorizationsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "metadata administration read cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					metadataadminsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "page layout resolution cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					pagelayoutsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "approval notification Flow read cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					approvalnotificationsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "outbound intent cannot select Catalog serving projection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					outboundintentsemantic.CapabilityName
			},
			wantError: "cannot select a Catalog serving projection",
		},
		{
			name: "permission authoring requires record mutation sibling",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID =
					permissionauthoringsemantic.CapabilityName
				config.Serving.Capabilities[0].OutcomeRetention = "1h"
			},
			wantError: "requires sibling capability \"runtime.records.mutate.v1\"",
		},
		{
			name: "unsupported workload capability",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.WorkloadCapabilities = []runtimeconfig.WorkloadCapability{{
					RuntimeCapabilityID: "runtime.unknown.v1", AuthorizedWorkloadID: "sfcrm",
				}}
			},
			wantError: "unsupported workload capability",
		},
		{
			name: "application result requires explicit platform lease",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.WorkloadCapabilities = []runtimeconfig.WorkloadCapability{{
					RuntimeCapabilityID:  applicationresultsemantic.CapabilityName,
					AuthorizedWorkloadID: "integration-service",
				}}
			},
			wantError: "admission lease",
		},
		{
			name: "session cannot declare platform lease",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.WorkloadCapabilities = []runtimeconfig.WorkloadCapability{{
					RuntimeCapabilityID:  sessionsemantic.CapabilityName,
					AuthorizedWorkloadID: "sfcrm-app-service",
					AdmissionLease:       "1s",
				}}
			},
			wantError: "uses the identity attempt cap",
		},
		{
			name: "mailbox authorization requires custom permission",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.WorkloadCapabilities = []runtimeconfig.WorkloadCapability{{
					RuntimeCapabilityID:  mailboxauthorizationsemantic.CapabilityName,
					AuthorizedWorkloadID: "integration-service", AdmissionLease: "1s",
				}}
			},
			wantError: "requires exactly one system or custom permission",
		},
		{
			name: "session cannot declare custom permission",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.WorkloadCapabilities = []runtimeconfig.WorkloadCapability{{
					RuntimeCapabilityID:      sessionsemantic.CapabilityName,
					AuthorizedWorkloadID:     "sfcrm-app-service",
					RequiredCustomPermission: "SendRecordEmail",
				}}
			},
			wantError: "cannot declare a permission requirement",
		},
		{
			name: "federated trust requires session workload selection",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.FederatedOIDCProviders = []runtimeconfig.FederatedOIDCProvider{{
					ProviderID: "provider-a", Audience: "client-a", SessionScope: "api",
					TrustedIssuers: []string{"https://issuer.example"},
					JWKSURL:        "https://issuer.example/jwks.json",
				}}
			},
			wantError: "requires the session workload capability",
		},
		{
			name: "handler capability aliases runtime capability",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].AppCapabilityID = config.Serving.Capabilities[0].RuntimeCapabilityID
			},
			wantError: "App handler and Runtime data capabilities must be distinct",
		},
		{
			name: "invalid identity attempt cap",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.IdentityAttemptCap = "0s"
			},
			wantError: "identity attempt cap is invalid",
		},
		{
			name: "invalid capability lease",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].AdmissionLease = "invalid"
			},
			wantError: "admission lease",
		},
		{
			name: "read capability cannot own outcomes",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].OutcomeRetention = "24h"
			},
			wantError: "cannot own durable outcomes",
		},
		{
			name: "mutation requires explicit outcome retention",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID = mutationsemantic.CapabilityName
			},
			wantError: "outcome retention for capability \"runtime.records.mutate.v1\" is invalid",
		},
		{
			name: "permission authoring requires explicit outcome retention",
			mutate: func(config *runtimeconfig.API) {
				config.Serving.Capabilities[0].RuntimeCapabilityID = permissionauthoringsemantic.CapabilityName
			},
			wantError: "outcome retention for capability \"runtime.permissions.authoring.v1\" is invalid",
		},
		{
			name: "missing workload credential",
			mutate: func(config *runtimeconfig.API) {
				config.WorkloadIdentityTokenFile = "/path/that/does/not/exist"
			},
			wantError: "read workload credential",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			config := base
			config.Serving.Capabilities = append([]runtimeconfig.ServingCapability(nil), base.Serving.Capabilities...)
			test.mutate(&config)
			process, connection, err := Assemble(context.Background(), "test", config)
			if process != nil || connection != nil {
				t.Fatalf("invalid configuration created resources: process=%v connection=%v", process, connection)
			}
			if err == nil || !strings.Contains(err.Error(), test.wantError) {
				t.Fatalf("Assemble() error = %v, want %q", err, test.wantError)
			}
		})
	}
}

func TestAssembleAcceptsSessionOnlyAsWorkloadCapabilityBeforeDatabaseOpen(t *testing.T) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		WorkloadCapabilities: []runtimeconfig.WorkloadCapability{{
			RuntimeCapabilityID:  sessionsemantic.CapabilityName,
			AuthorizedWorkloadID: "sfcrm-app-service",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf("incomplete configuration created resources: process=%v connection=%v", process, connection)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf("Assemble() error = %v, want post-capability workload-credential failure", err)
	}
}

func TestAssembleAcceptsApplicationResultAsWorkloadCapabilityBeforeDatabaseOpen(
	t *testing.T,
) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		WorkloadCapabilities: []runtimeconfig.WorkloadCapability{{
			RuntimeCapabilityID:  applicationresultsemantic.CapabilityName,
			AuthorizedWorkloadID: "integration-service",
			AdmissionLease:       "1s",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf(
			"Assemble() error = %v, want post-capability workload-credential failure",
			err,
		)
	}
}

func TestAssembleAcceptsMailboxAuthorizationAsWorkloadCapabilityBeforeDatabaseOpen(
	t *testing.T,
) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		WorkloadCapabilities: []runtimeconfig.WorkloadCapability{{
			RuntimeCapabilityID:  mailboxauthorizationsemantic.CapabilityName,
			AuthorizedWorkloadID: "integration-service", AdmissionLease: "1s",
			RequiredCustomPermission: "SendRecordEmail",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf(
			"Assemble() error = %v, want post-capability workload-credential failure",
			err,
		)
	}
}

func TestAssembleAcceptsSystemPermissionMailboxAuthorizationBeforeDatabaseOpen(
	t *testing.T,
) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		WorkloadCapabilities: []runtimeconfig.WorkloadCapability{{
			RuntimeCapabilityID:  mailboxauthorizationsemantic.CapabilityName,
			AuthorizedWorkloadID: "integration-service", AdmissionLease: "1s",
			RequiredSystemPermission: "ApiEnabled",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf(
			"Assemble() error = %v, want post-capability workload-credential failure",
			err,
		)
	}
}

func TestAssembleAcceptsMetadataAdminReadAsWorkloadCapabilityBeforeDatabaseOpen(
	t *testing.T,
) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		WorkloadCapabilities: []runtimeconfig.WorkloadCapability{{
			RuntimeCapabilityID:  metadataadminsemantic.CapabilityName,
			AuthorizedWorkloadID: "sfcrm-app-service",
			AdmissionLease:       "1s",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf(
			"Assemble() error = %v, want post-capability workload-credential failure",
			err,
		)
	}
}

func TestAssembleAcceptsPageLayoutResolutionAsCatalogConsumerBeforeDatabaseOpen(
	t *testing.T,
) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		CatalogConsumerCapabilities: []runtimeconfig.CatalogConsumerCapability{{
			RuntimeCapabilityID:   pagelayoutsemantic.CapabilityName,
			AuthorizedWorkloadID:  "sfcrm-app-service",
			CatalogSnapshotDigest: "sha256:snapshot",
			AdmissionLease:        "1s",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf(
			"Assemble() error = %v, want post-capability workload-credential failure",
			err,
		)
	}
}

func TestAssembleAcceptsApprovalNotificationFlowReadAsWorkloadCapabilityBeforeDatabaseOpen(
	t *testing.T,
) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		WorkloadCapabilities: []runtimeconfig.WorkloadCapability{{
			RuntimeCapabilityID:  approvalnotificationsemantic.CapabilityName,
			AuthorizedWorkloadID: "sfcrm-app-service",
			AdmissionLease:       "1s",
		}},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf(
			"Assemble() error = %v, want post-capability workload-credential failure",
			err,
		)
	}
}

func TestProductionPlatformAdmissionCarriesApplicationResultLease(t *testing.T) {
	platformLeases, resourceLimits, err := productionPlatformAdmission(
		map[string]time.Duration{
			applicationresultsemantic.CapabilityName:    2 * time.Second,
			mailboxauthorizationsemantic.CapabilityName: 2 * time.Second,
			querysemantic.CapabilityName:                3 * time.Second,
		},
		runtimeconfig.DefaultAPIAdmissionLimit,
	)
	if err != nil {
		t.Fatal(err)
	}
	if len(platformLeases) != 2 ||
		platformLeases[applicationresultsemantic.CapabilityName] != 2*time.Second ||
		platformLeases[mailboxauthorizationsemantic.CapabilityName] != 2*time.Second {
		t.Fatalf("Production Platform leases = %#v", platformLeases)
	}
	if len(resourceLimits) != 0 {
		t.Fatalf(
			"ApplicationResult must use the host-owned short-unary pool: %#v",
			resourceLimits,
		)
	}
}

func TestProductionPlatformAdmissionReservesAdditiveOutboundAttachmentStream(t *testing.T) {
	platformLeases, resourceLimits, err := productionPlatformAdmission(
		map[string]time.Duration{outboundintentsemantic.CapabilityName: 2 * time.Second},
		runtimeconfig.DefaultAPIAdmissionLimit,
		runtimegrpc.ResourceClassFileStream,
	)
	if err != nil {
		t.Fatal(err)
	}
	if platformLeases[outboundintentsemantic.CapabilityName] != 2*time.Second {
		t.Fatalf("outbound intent lease = %#v", platformLeases)
	}
	if got := resourceLimits[runtimegrpc.ResourceClassFileStream]; got != 1 {
		t.Fatalf("additive outbound FileStream limit = %d, want 1", got)
	}
}

func TestProductionRuntimeDeclaresMetadataAdminReadAsExistingHeavyWork(t *testing.T) {
	definition, supported := productionRuntimeCapabilityDefinition(metadataadminsemantic.CapabilityName)
	if !supported || definition.id != metadataadminsemantic.CapabilityName || definition.requiresOutcomeRetention {
		t.Fatalf("metadata admin-read production definition = %#v, supported=%v", definition, supported)
	}
	platformLeases, resourceLimits, err := productionPlatformAdmission(
		map[string]time.Duration{metadataadminsemantic.CapabilityName: 2 * time.Second},
		runtimeconfig.DefaultAPIAdmissionLimit,
	)
	if err != nil {
		t.Fatal(err)
	}
	if platformLeases[metadataadminsemantic.CapabilityName] != 2*time.Second ||
		resourceLimits[runtimegrpc.ResourceClassHeavyWork] <= 0 {
		t.Fatalf("metadata admin-read production admission = leases %#v, limits %#v", platformLeases, resourceLimits)
	}
}

func TestProductionRuntimeDeclaresApprovalNotificationFlowReadAsExistingHeavyWork(t *testing.T) {
	definition, supported := productionRuntimeCapabilityDefinition(approvalnotificationsemantic.CapabilityName)
	if !supported || definition.id != approvalnotificationsemantic.CapabilityName || definition.requiresOutcomeRetention {
		t.Fatalf("approval notification Flow-read production definition = %#v, supported=%v", definition, supported)
	}
	platformLeases, resourceLimits, err := productionPlatformAdmission(
		map[string]time.Duration{approvalnotificationsemantic.CapabilityName: 2 * time.Second},
		runtimeconfig.DefaultAPIAdmissionLimit,
	)
	if err != nil {
		t.Fatal(err)
	}
	if platformLeases[approvalnotificationsemantic.CapabilityName] != 2*time.Second ||
		resourceLimits[runtimegrpc.ResourceClassHeavyWork] <= 0 {
		t.Fatalf("approval notification Flow-read production admission = leases %#v, limits %#v", platformLeases, resourceLimits)
	}
}

func TestProductionPlatformAdmissionPreservesConcurrentHeavyWorkAndShortUnary(
	t *testing.T,
) {
	const admissionLimit = runtimeconfig.DefaultAPIAdmissionLimit
	platformLeases, resourceLimits, err := productionPlatformAdmission(
		map[string]time.Duration{
			flowsemantic.CapabilityName:                 10 * time.Second,
			filessemantic.CapabilityName:                10 * time.Second,
			catalogsemantic.AppMetadataV3CapabilityName: 10 * time.Second,
		},
		admissionLimit,
	)
	if err != nil {
		t.Fatal(err)
	}
	if got := resourceLimits[runtimegrpc.ResourceClassHeavyWork]; got != 13 {
		t.Fatalf("production HeavyWork limit = %d, want 13", got)
	}
	if got := resourceLimits[runtimegrpc.ResourceClassFileStream]; got != 1 {
		t.Fatalf("production FileStream limit = %d, want 1", got)
	}
	if got := resourceLimits[runtimegrpc.ResourceClassCatalogWatch]; got != 1 {
		t.Fatalf("production CatalogWatch limit = %d, want 1", got)
	}

	options := []runtimegrpc.PolicyOption{
		runtimegrpc.WithQueryMaxAdmissionLease(10 * time.Second),
		runtimegrpc.WithRecordMutationAttemptCap(10 * time.Second),
	}
	for capability, lease := range platformLeases {
		options = append(
			options,
			runtimegrpc.WithPlatformCapabilityMaxAdmissionLease(capability, lease),
		)
	}
	for resourceClass, limit := range resourceLimits {
		options = append(
			options,
			runtimegrpc.WithResourceClassAdmissionLimit(resourceClass, limit),
		)
	}
	policy, err := runtimegrpc.NewPolicyWithOptions(runtimegrpc.Config{
		Authority: func(context.Context, string) (runtimegrpc.VerifiedCaller, error) {
			return "caller", nil
		},
		AdmissionLimit: admissionLimit,
		AttemptCap:     time.Second,
	}, options...)
	if err != nil {
		t.Fatal(err)
	}

	incoming := func(capability string, major, minor uint32) (context.Context, context.CancelFunc) {
		ctx := metadata.NewIncomingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(major), 10),
			envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(minor), 10),
			envelope.CapabilityMetadataKey, capability,
		))
		return context.WithTimeout(ctx, 5*time.Second)
	}

	heavyLimit := resourceLimits[runtimegrpc.ResourceClassHeavyWork]
	entered := make(chan struct{}, heavyLimit)
	done := make(chan error, heavyLimit)
	release := make(chan struct{})
	released := false
	defer func() {
		if !released {
			close(release)
		}
	}()
	for i := 0; i < heavyLimit; i++ {
		method := runtimev1.RecordQueryService_QueryRecords_FullMethodName
		capability := querysemantic.CapabilityName
		major, minor := querysemantic.ProtocolMajor, querysemantic.ProtocolMinor
		if i%2 != 0 {
			method = runtimev1.RecordMutationsService_MutateRecords_FullMethodName
			capability = mutationsemantic.CapabilityName
			major, minor = mutationsemantic.ProtocolMajor, mutationsemantic.ProtocolMinor
		}
		go func() {
			ctx, cancel := incoming(capability, major, minor)
			defer cancel()
			_, callErr := policy.UnaryServerInterceptor()(
				ctx,
				nil,
				&grpc.UnaryServerInfo{FullMethod: method},
				func(context.Context, any) (any, error) {
					entered <- struct{}{}
					<-release
					return nil, nil
				},
			)
			done <- callErr
		}()
	}
	wait := time.NewTimer(5 * time.Second)
	defer wait.Stop()
	for i := 0; i < heavyLimit; i++ {
		select {
		case <-entered:
		case <-wait.C:
			t.Fatalf("only %d/%d Query/Mutation calls entered concurrently", i, heavyLimit)
		}
	}

	overflowCtx, cancelOverflow := incoming(
		querysemantic.CapabilityName,
		querysemantic.ProtocolMajor,
		querysemantic.ProtocolMinor,
	)
	defer cancelOverflow()
	overflowCalled := false
	_, err = policy.UnaryServerInterceptor()(
		overflowCtx,
		nil,
		&grpc.UnaryServerInfo{FullMethod: runtimev1.RecordQueryService_QueryRecords_FullMethodName},
		func(context.Context, any) (any, error) {
			overflowCalled = true
			return nil, nil
		},
	)
	if status.Code(err) != codes.ResourceExhausted || overflowCalled {
		t.Fatalf("Query above production HeavyWork limit = %v, called=%v", err, overflowCalled)
	}

	shortCtx, cancelShort := incoming(
		recordsemantic.CapabilityName,
		recordsemantic.ProtocolMajor,
		recordsemantic.ProtocolMinor,
	)
	defer cancelShort()
	shortCalled := false
	_, err = policy.UnaryServerInterceptor()(
		shortCtx,
		nil,
		&grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName},
		func(context.Context, any) (any, error) {
			shortCalled = true
			return nil, nil
		},
	)
	if err != nil || !shortCalled {
		t.Fatalf("short unary while production HeavyWork is saturated = %v, called=%v", err, shortCalled)
	}

	if snapshot := policy.ResourceAdmissionSnapshot(runtimegrpc.ResourceClassHeavyWork); snapshot.Limit != heavyLimit ||
		snapshot.InFlight != heavyLimit || snapshot.Accepted != uint64(heavyLimit) || snapshot.RejectedFull != 1 {
		t.Fatalf("production HeavyWork snapshot = %#v", snapshot)
	}
	if snapshot := policy.AdmissionSnapshot(); snapshot.Limit != admissionLimit ||
		snapshot.InFlight != heavyLimit || snapshot.Accepted != uint64(heavyLimit+1) {
		t.Fatalf("production global admission snapshot = %#v", snapshot)
	}

	close(release)
	released = true
	for i := 0; i < heavyLimit; i++ {
		if callErr := <-done; callErr != nil {
			t.Fatalf("concurrent Query/Mutation call %d = %v", i, callErr)
		}
	}
}

func TestProductionPlatformAdmissionRejectsConfigurationWithoutShortUnaryReserve(
	t *testing.T,
) {
	_, _, err := productionPlatformAdmission(
		map[string]time.Duration{
			flowsemantic.CapabilityName:                 time.Second,
			filessemantic.CapabilityName:                time.Second,
			catalogsemantic.AppMetadataV3CapabilityName: time.Second,
		},
		3,
	)
	if err == nil || !strings.Contains(err.Error(), "reserve short-unary capacity") {
		t.Fatalf("undersized production admission = %v, want fail-closed reserve error", err)
	}
}

func TestAssembleAcceptsSeparatelyAdmittedRecordViewCapabilitiesBeforeDatabaseOpen(t *testing.T) {
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		Capabilities: []runtimeconfig.ServingCapability{
			{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-view-read.v1", RuntimeCapabilityID: recordviewsemantic.CapabilityName, AdmissionLease: "1s"},
			{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-view-attachments.v1", RuntimeCapabilityID: attachmentsemantic.CapabilityName, AdmissionLease: "1s"},
		},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf("incomplete configuration created resources: process=%v connection=%v", process, connection)
	}
	if err == nil || !strings.Contains(err.Error(), "read workload credential") {
		t.Fatalf("Assemble() error = %v, want post-capability workload-credential failure", err)
	}
}

func TestAssembleRequiresFlowAppMetadataCapabilityBeforeDatabaseOpen(t *testing.T) {
	base := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap: "1s",
		Capabilities: []runtimeconfig.ServingCapability{
			{
				AppCapabilityID:     "io.nexuscrm.sfcrm.record-mutation.v1",
				RuntimeCapabilityID: mutationsemantic.CapabilityName,
				AdmissionLease:      "1s",
				OutcomeRetention:    "1h",
			},
			{
				AppCapabilityID:     "io.nexuscrm.sfcrm.flow.v1",
				RuntimeCapabilityID: flowsemantic.CapabilityName,
				AdmissionLease:      "1s",
			},
		},
	}, WorkloadIdentityTokenFile: "/path/that/does/not/exist"}

	process, connection, err := Assemble(context.Background(), "test", base)
	if process != nil || connection != nil {
		t.Fatalf("incomplete configuration created resources: process=%v connection=%v", process, connection)
	}
	if err == nil ||
		!strings.Contains(err.Error(), flowsemantic.CapabilityName) ||
		!strings.Contains(err.Error(), catalogsemantic.AppMetadataV3CapabilityName) ||
		!strings.Contains(err.Error(), catalogsemantic.AppMetadataV2CapabilityName) {
		t.Fatalf("Flow-only Assemble() error = %v, want explicit AppMetadata capability dependency", err)
	}

	withLegacyAppMetadata := base
	withLegacyAppMetadata.Serving.Capabilities = append(
		append([]runtimeconfig.ServingCapability(nil), base.Serving.Capabilities...),
		runtimeconfig.ServingCapability{
			AppCapabilityID:     "io.nexuscrm.sfcrm.app-metadata.v1",
			RuntimeCapabilityID: catalogsemantic.AppMetadataCapabilityName,
			AdmissionLease:      "1s",
		},
	)
	process, connection, err = Assemble(context.Background(), "test", withLegacyAppMetadata)
	if process != nil || connection != nil {
		t.Fatalf("incomplete configuration created resources: process=%v connection=%v", process, connection)
	}
	if err == nil ||
		!strings.Contains(err.Error(), catalogsemantic.AppMetadataV3CapabilityName) ||
		!strings.Contains(err.Error(), catalogsemantic.AppMetadataV2CapabilityName) {
		t.Fatalf("Flow+legacy AppMetadata Assemble() error = %v, want V3/V2 capability dependency", err)
	}

	for _, metadataCapability := range []struct {
		name string
		id   string
	}{
		{name: "V3", id: catalogsemantic.AppMetadataV3CapabilityName},
		{name: "V2", id: catalogsemantic.AppMetadataV2CapabilityName},
	} {
		t.Run(metadataCapability.name, func(t *testing.T) {
			withAppMetadata := base
			withAppMetadata.Serving.Capabilities = append(
				append([]runtimeconfig.ServingCapability(nil), base.Serving.Capabilities...),
				runtimeconfig.ServingCapability{
					AppCapabilityID:     "io.nexuscrm.sfcrm.app-metadata." + strings.ToLower(metadataCapability.name),
					RuntimeCapabilityID: metadataCapability.id,
					AdmissionLease:      "1s",
				},
			)
			process, connection, err := Assemble(context.Background(), "test", withAppMetadata)
			if process != nil || connection != nil {
				t.Fatalf("incomplete configuration created resources: process=%v connection=%v", process, connection)
			}
			if err == nil || !strings.Contains(err.Error(), "read workload credential") {
				t.Fatalf("Flow+AppMetadata %s Assemble() error = %v, want post-capability workload-credential failure", metadataCapability.name, err)
			}
		})
	}
}

func TestAssembleCanonicalizesRuntimeCapabilitySetIndependentlyOfAppOrder(t *testing.T) {
	secretRoot := t.TempDir()
	workloadTokenFile := filepath.Join(secretRoot, "workload-token")
	if err := os.WriteFile(workloadTokenFile, []byte("test-token"), 0o600); err != nil {
		t.Fatal(err)
	}
	privateKey := ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize))
	signingKeyFile := filepath.Join(secretRoot, "signing-keys.json")
	signingKeyDocument, err := json.Marshal(map[string]any{
		"schema": runtimeidentity.SigningKeyFileSchemaV1,
		"active": map[string]any{
			"signing_key_id":     "test-current",
			"private_key_base64": base64.StdEncoding.EncodeToString(privateKey),
		},
		"overlap": []any{},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(signingKeyFile, signingKeyDocument, 0o600); err != nil {
		t.Fatal(err)
	}
	config := runtimeconfig.API{Serving: runtimeconfig.Serving{
		IdentityAttemptCap:   "1s",
		AuthorizedWorkloadID: "sfcrm-app-service",
		// Catalog orders this document by App capability, which is intentionally
		// independent of the delegation verifier's Runtime capability order.
		Capabilities: []runtimeconfig.ServingCapability{
			{AppCapabilityID: "io.nexuscrm.sfcrm.query.v1", RuntimeCapabilityID: querysemantic.CapabilityName, AdmissionLease: "1s"},
			{AppCapabilityID: "io.nexuscrm.sfcrm.records.mutate.v1", RuntimeCapabilityID: mutationsemantic.CapabilityName, AdmissionLease: "1s", OutcomeRetention: "24h"},
			{AppCapabilityID: "io.nexuscrm.sfcrm.recycle-bin-read.v1", RuntimeCapabilityID: recyclesemantic.CapabilityName, AdmissionLease: "1s"},
			{AppCapabilityID: "io.nexuscrm.sfcrm.ui.record-access.v1", RuntimeCapabilityID: accesssemantic.CapabilityName, AdmissionLease: "1s"},
		},
	}, WorkloadIdentityTokenFile: workloadTokenFile, SigningKeyFile: signingKeyFile}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf("incomplete configuration created resources: process=%v connection=%v", process, connection)
	}
	if err == nil || !strings.Contains(err.Error(), "load gRPC server identity") {
		t.Fatalf("Assemble() error = %v, want post-verifier TLS identity failure", err)
	}
}

func TestAssembleV4LoadsExactMultiWorkloadIdentitySetBeforeDatabaseOpen(t *testing.T) {
	secretRoot := t.TempDir()
	workloadIdentityFile := filepath.Join(secretRoot, "workload-identities.json")
	identityDocument, err := json.Marshal(map[string]any{
		"schema": runtimeidentity.WorkloadIdentityFileSchemaV1,
		"identities": []map[string]string{
			{"workloadId": "integration-service", "token": "integration-token"},
			{"workloadId": "sfcrm-app-service", "token": "sfcrm-token"},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(workloadIdentityFile, identityDocument, 0o600); err != nil {
		t.Fatal(err)
	}
	privateKey := ed25519.NewKeyFromSeed(make([]byte, ed25519.SeedSize))
	signingKeyFile := filepath.Join(secretRoot, "signing-keys.json")
	signingKeyDocument, err := json.Marshal(map[string]any{
		"schema": runtimeidentity.SigningKeyFileSchemaV1,
		"active": map[string]any{
			"signing_key_id":     "test-current",
			"private_key_base64": base64.StdEncoding.EncodeToString(privateKey),
		},
		"overlap": []any{},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(signingKeyFile, signingKeyDocument, 0o600); err != nil {
		t.Fatal(err)
	}
	config := runtimeconfig.API{
		Serving: runtimeconfig.Serving{
			Schema:                  runtimeconfig.ServingConfigSchemaV4,
			IdentityAttemptCap:      "1s",
			AuthorizedWorkloadIDs:   []string{"integration-service", "sfcrm-app-service"},
			PermissionStateProtocol: runtimeconfig.PermissionStateProtocolV1,
			CatalogConsumerCapabilities: []runtimeconfig.CatalogConsumerCapability{{
				RuntimeCapabilityID:  catalogsemantic.IntegrationBindingCapabilityName,
				AuthorizedWorkloadID: "integration-service",
				AdmissionLease:       "1s",
			}},
			Capabilities: []runtimeconfig.ServingCapability{{
				AppCapabilityID:      "io.nexuscrm.sfcrm.query.v1",
				RuntimeCapabilityID:  querysemantic.CapabilityName,
				AuthorizedWorkloadID: "sfcrm-app-service",
				AdmissionLease:       "1s",
			}},
		},
		WorkloadIdentityTokenFile: workloadIdentityFile,
		SigningKeyFile:            signingKeyFile,
	}

	process, connection, err := Assemble(context.Background(), "test", config)
	if process != nil || connection != nil {
		t.Fatalf(
			"incomplete configuration created resources: process=%v connection=%v",
			process,
			connection,
		)
	}
	if err == nil || !strings.Contains(err.Error(), "load gRPC server identity") {
		t.Fatalf(
			"Assemble() error = %v, want post-identity-set TLS failure",
			err,
		)
	}
}