package runtimegrpc

import (
	"context"
	"errors"
	"net"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	analyticsauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/analyticsauthoringv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	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"
	authorizationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/authorizationv1"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	externaleffectsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/externaleffectv1"
	federatedloginsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/federatedloginv1"
	filessemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/filesv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	identitysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/identityv1"
	layoutauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv1"
	layoutauthoringv2semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv2"
	listviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/listviewv1"
	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"
	permissionadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionadminv1"
	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"
	recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1"
	relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	reportsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/reportv1"
	resourceaccesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/resourceaccessv1"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	sessionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/sessionv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/catalog/v1"
	commonv1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/common/v1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	_ "google.golang.org/grpc/encoding/gzip"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
)

func TestPolicyRequiresBoundedConfiguration(t *testing.T) {
	authority := AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) { return "caller", nil })
	// This positional literal is an executable source-compatibility guard for
	// the released Config field shape.
	_ = Config{authority, 1, time.Second}
	var typedNil AuthorityFunc
	for _, config := range []Config{
		{},
		{Authority: typedNil, AdmissionLimit: 1, AttemptCap: time.Second},
		{Authority: authority, AdmissionLimit: 0, AttemptCap: time.Second},
		{Authority: authority, AdmissionLimit: 1, AttemptCap: 0},
		{Authority: authority, AdmissionLimit: 1, AttemptCap: semantic.MaxAttemptDuration + time.Nanosecond},
	} {
		if policy, err := NewPolicy(config); err == nil || policy != nil {
			t.Fatalf("NewPolicy(%+v) = %#v, %v; want rejected", config, policy, err)
		}
	}
	if policy, err := NewPolicyWithIdentity(Config{Authority: authority, AdmissionLimit: 1, AttemptCap: time.Second}, -time.Second); err == nil || policy != nil {
		t.Fatalf("NewPolicyWithIdentity accepted a negative attempt cap: %#v, %v", policy, err)
	}
	config := Config{Authority: authority, AdmissionLimit: 1, AttemptCap: time.Second}
	for name, options := range map[string][]PolicyOption{
		"nil":                         {nil},
		"negative":                    {WithRecordAccessMaxAdmissionLease(-time.Nanosecond)},
		"duplicate":                   {WithRecordAccessMaxAdmissionLease(time.Second), WithRecordAccessMaxAdmissionLease(time.Second)},
		"related negative":            {WithRelatedCountMaxAdmissionLease(-time.Nanosecond)},
		"related duplicate":           {WithRelatedCountMaxAdmissionLease(time.Second), WithRelatedCountMaxAdmissionLease(time.Second)},
		"recycle negative":            {WithRecycleBinMaxAdmissionLease(-time.Nanosecond)},
		"recycle duplicate":           {WithRecycleBinMaxAdmissionLease(time.Second), WithRecycleBinMaxAdmissionLease(time.Second)},
		"recycle mutation negative":   {WithRecycleBinMutationMaxAdmissionLease(-time.Nanosecond)},
		"recycle mutation duplicate":  {WithRecycleBinMutationMaxAdmissionLease(time.Second), WithRecycleBinMutationMaxAdmissionLease(time.Second)},
		"record-view negative":        {WithRecordViewMaxAdmissionLease(-time.Nanosecond)},
		"record-view duplicate":       {WithRecordViewMaxAdmissionLease(time.Second), WithRecordViewMaxAdmissionLease(time.Second)},
		"attachment-source negative":  {WithAttachmentSourceMaxAdmissionLease(-time.Nanosecond)},
		"attachment-source duplicate": {WithAttachmentSourceMaxAdmissionLease(time.Second), WithAttachmentSourceMaxAdmissionLease(time.Second)},
		"mutation negative":           {WithRecordMutationAttemptCap(-time.Nanosecond)},
		"mutation duplicate":          {WithRecordMutationAttemptCap(time.Second), WithRecordMutationAttemptCap(time.Second)},
		"platform unknown":            {WithPlatformCapabilityMaxAdmissionLease("runtime.unknown.v1", time.Second)},
		"platform negative":           {WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, -time.Nanosecond)},
		"platform duplicate": {
			WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, time.Second),
			WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, time.Second),
		},
		"platform missing isolated pool": {
			WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, time.Second),
		},
		"resource short class": {WithResourceClassAdmissionLimit(ResourceClassShortUnary, 1)},
		"resource zero":        {WithResourceClassAdmissionLimit(ResourceClassHeavyWork, 0)},
		"resource duplicate": {
			WithResourceClassAdmissionLimit(ResourceClassFileStream, 1),
			WithResourceClassAdmissionLimit(ResourceClassFileStream, 1),
		},
	} {
		if policy, err := NewPolicyWithOptions(config, options...); err == nil || policy != nil {
			t.Fatalf("NewPolicyWithOptions(%s) = %#v, %v; want rejected", name, policy, err)
		}
	}
}

func TestPinnedWirePlatformStatusMappingRoundTripsStableFailure(t *testing.T) {
	for _, test := range []struct {
		name       string
		capability string
		original   *commonsemantic.Failure
	}{
		{
			name: "search admission", capability: searchsemantic.CapabilityName,
			original: &commonsemantic.Failure{
				Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission, Maximum: 1,
			},
		},
		{
			name: "application result missing", capability: applicationresultsemantic.CapabilityName,
			original: &commonsemantic.Failure{Category: commonsemantic.FailureMissing},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			wireErr := mapping.ToPlatformStatusError(test.original, test.capability)
			var decoded *commonsemantic.Failure
			if err := mapping.FromPlatformStatusError(wireErr, test.capability); !errors.As(err, &decoded) {
				t.Fatalf("decode pinned wire Platform status mapping: %v", err)
			}
			if decoded.Category != test.original.Category ||
				decoded.LimitKind != test.original.LimitKind ||
				decoded.Maximum != test.original.Maximum {
				t.Fatalf("pinned wire Platform status round-trip = %#v, want %#v", decoded, test.original)
			}
		})
	}
}

func TestPlatformMethodsRemainCapabilitySpecificOnOnePolicy(t *testing.T) {
	tests := map[string]struct {
		capability string
		method     MethodClass
		resource   ResourceClass
	}{
		catalogv1.ConsumerViewService_GetAppMetadataView_FullMethodName: {
			catalogsemantic.AppMetadataCapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		catalogv1.ConsumerViewService_WatchAppMetadataView_FullMethodName: {
			catalogsemantic.AppMetadataCapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch,
		},
		catalogv1.ConsumerViewService_GetAppMetadataViewV2_FullMethodName: {
			catalogsemantic.AppMetadataV2CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		catalogv1.ConsumerViewService_WatchAppMetadataViewV2_FullMethodName: {
			catalogsemantic.AppMetadataV2CapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch,
		},
		catalogv1.AppMetadataViewV3Service_GetAppMetadataViewV3_FullMethodName: {
			catalogsemantic.AppMetadataV3CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		catalogv1.AppMetadataViewV3Service_WatchAppMetadataViewV3_FullMethodName: {
			catalogsemantic.AppMetadataV3CapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch,
		},
		catalogv1.ConsumerViewService_GetIntegrationBindingView_FullMethodName: {
			catalogsemantic.IntegrationBindingCapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		catalogv1.ConsumerViewService_WatchIntegrationBindingView_FullMethodName: {
			catalogsemantic.IntegrationBindingCapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch,
		},
		runtimev1.RecordSearchService_SearchRecords_FullMethodName: {
			searchsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.RecordSearchService_ReadSearchScopeOrder_FullMethodName: {
			searchsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.FlowExecutionService_ExecuteFlow_FullMethodName: {
			flowsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.FlowExecutionService_PreviewCatalogAction_FullMethodName: {
			flowsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.ReportReadService_ReadReport_FullMethodName: {
			reportsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.AuditLogReadService_ReadAuditLog_FullMethodName: {
			auditsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PrivateListViewService_ReadPrivateListViews_FullMethodName: {
			listviewsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PrivateListViewService_MutatePrivateListView_FullMethodName: {
			listviewsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.MetadataAdminReadService_ReadObjectAdminIndex_FullMethodName: {
			metadataadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.MetadataAdminReadService_ReadLayoutAdminWorkspace_FullMethodName: {
			metadataadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.MetadataAdminReadService_ReadStoredLayout_FullMethodName: {
			metadataadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.MetadataAdminReadService_ReadValidationRuleAdminIndex_FullMethodName: {
			metadataadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.ConnectedMailboxAuthorizationService_AuthorizeConnectedMailbox_FullMethodName: {
			mailboxauthorizationsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.ConnectedMailboxAuthorizationService_ReauthorizeConnectedMailbox_FullMethodName: {
			mailboxauthorizationsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.AuthorizedFilesService_CreateFile_FullMethodName: {
			filessemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassFileStream,
		},
		runtimev1.AuthorizedFilesService_OpenFile_FullMethodName: {
			filessemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassFileStream,
		},
		runtimev1.AuthorizedFilesService_DeleteFile_FullMethodName: {
			filessemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.BulkIngestService_CreateBulkIngestJob_FullMethodName: {
			bulkingestsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.BulkIngestService_ReadBulkIngestJob_FullMethodName: {
			bulkingestsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.BulkIngestService_CloseBulkIngestJob_FullMethodName: {
			bulkingestsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.BulkIngestService_UploadBulkIngestData_FullMethodName: {
			bulkingestsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassFileStream,
		},
		runtimev1.CommandAcceptanceService_AcceptCommand_FullMethodName: {
			commandsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.ApplicationResultService_GetApplicationResult_FullMethodName: {
			applicationresultsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.ExternalEffectDispatchService_ClaimExternalEffects_FullMethodName: {
			externaleffectsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.ExternalEffectDispatchService_CompleteExternalEffect_FullMethodName: {
			externaleffectsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.OutboundIntentService_ListPendingOutboundIntents_FullMethodName: {
			outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.OutboundIntentService_ReadOutboundIntentGeneration_FullMethodName: {
			outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.OutboundIntentService_AcknowledgeOutboundIntent_FullMethodName: {
			outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.OutboundIntentService_CommitOutboundResult_FullMethodName: {
			outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.OutboundIntentNoEffectService_CommitOutboundNoEffect_FullMethodName: {
			outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary,
		},
		runtimev1.OutboundIntentService_OpenOutboundMailAttachment_FullMethodName: {
			outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassFileStream,
		},
		runtimev1.EffectivePermissionService_ReadEffectivePermissions_FullMethodName: {
			permissionsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PermissionAdministrationService_ReadPermissionAdministration_FullMethodName: {
			permissionadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PermissionAdministrationService_MutatePermissionAdministration_FullMethodName: {
			permissionadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PermissionAuthoringService_ReadPermissionAuthoringSnapshot_FullMethodName: {
			permissionauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PermissionAuthoringService_SavePermissionAuthoring_FullMethodName: {
			permissionauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PermissionAuthoringService_GetPermissionAuthoringStatus_FullMethodName: {
			permissionauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.LayoutAuthoringService_Submit_FullMethodName: {
			layoutauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.LayoutAuthoringService_GetCommand_FullMethodName: {
			layoutauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.LayoutAuthoringService_ReadWorkspace_FullMethodName: {
			layoutauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.LayoutAuthoringService_ReadStoredLayout_FullMethodName: {
			layoutauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.LayoutAuthoringV2Service_ReadWorkspace_FullMethodName: {
			layoutauthoringv2semantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.LayoutAuthoringV2Service_ReadStoredLayout_FullMethodName: {
			layoutauthoringv2semantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.LayoutAuthoringV2Service_Mutate_FullMethodName: {
			layoutauthoringv2semantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.AnalyticsAuthoringService_MutateAnalyticsDefinition_FullMethodName: {
			analyticsauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.ResourceAccessService_AccessResource_FullMethodName: {
			resourceaccesssemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.ApprovalService_ExecuteApproval_FullMethodName: {
			approvalsemantic.ExecuteCapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.ApprovalService_ReadApproval_FullMethodName: {
			approvalsemantic.ReadCapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.ApprovalNotificationFlowReadService_ReadApprovalNotificationFlows_FullMethodName: {
			approvalnotificationsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork,
		},
		runtimev1.PageLayoutResolutionService_ResolvePageLayout_FullMethodName: {
			pagelayoutsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
		runtimev1.PageLayoutResolutionService_ReadObjectListFields_FullMethodName: {
			pagelayoutsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary,
		},
	}
	descriptors := []grpc.ServiceDesc{
		catalogv1.ConsumerViewService_ServiceDesc,
		catalogv1.AppMetadataViewV3Service_ServiceDesc,
		runtimev1.RecordSearchService_ServiceDesc,
		runtimev1.FlowExecutionService_ServiceDesc,
		runtimev1.ReportReadService_ServiceDesc,
		runtimev1.AuditLogReadService_ServiceDesc,
		runtimev1.PrivateListViewService_ServiceDesc,
		runtimev1.MetadataAdminReadService_ServiceDesc,
		runtimev1.LayoutAuthoringV2Service_ServiceDesc,
		runtimev1.ConnectedMailboxAuthorizationService_ServiceDesc,
		runtimev1.AuthorizedFilesService_ServiceDesc,
		runtimev1.BulkIngestService_ServiceDesc,
		runtimev1.CommandAcceptanceService_ServiceDesc,
		runtimev1.ApplicationResultService_ServiceDesc,
		runtimev1.ExternalEffectDispatchService_ServiceDesc,
		runtimev1.OutboundIntentService_ServiceDesc,
		runtimev1.OutboundIntentNoEffectService_ServiceDesc,
		runtimev1.EffectivePermissionService_ServiceDesc,
		runtimev1.PermissionAdministrationService_ServiceDesc,
		runtimev1.PermissionAuthoringService_ServiceDesc,
		runtimev1.LayoutAuthoringService_ServiceDesc,
		runtimev1.AnalyticsAuthoringService_ServiceDesc,
		runtimev1.ResourceAccessService_ServiceDesc,
		runtimev1.ApprovalService_ServiceDesc,
		runtimev1.ApprovalNotificationFlowReadService_ServiceDesc,
		runtimev1.PageLayoutResolutionService_ServiceDesc,
	}
	descriptorMethods := 0
	for _, descriptor := range descriptors {
		for _, method := range descriptor.Methods {
			fullMethod := "/" + descriptor.ServiceName + "/" + method.MethodName
			if _, ok := tests[fullMethod]; !ok {
				t.Fatalf("released Platform unary method %q has no closed policy classification", fullMethod)
			}
			descriptorMethods++
		}
		for _, stream := range descriptor.Streams {
			fullMethod := "/" + descriptor.ServiceName + "/" + stream.StreamName
			if _, ok := tests[fullMethod]; !ok {
				t.Fatalf("released Platform stream method %q has no closed policy classification", fullMethod)
			}
			descriptorMethods++
		}
	}
	if descriptorMethods != len(tests) {
		t.Fatalf("Platform descriptor methods = %d, classifications = %d", descriptorMethods, len(tests))
	}
	for fullMethod, expected := range tests {
		policy, ok := registeredMethod(fullMethod)
		if !ok || policy.capability != expected.capability || policy.methodClass != expected.method ||
			policy.resourceClass != expected.resource || policy.attemptPolicy != attemptPlatform ||
			policy.sensitiveBearer || policy.statusError == nil {
			t.Fatalf("method %q policy = %#v, %v", fullMethod, policy, ok)
		}
	}

	var authorityCalls atomic.Int64
	policy, err := NewPolicyWithOptions(Config{
		Authority: func(ctx context.Context, method string) (VerifiedCaller, error) {
			expected, ok := tests[method]
			if !ok {
				t.Fatalf("authority received unknown method %q", method)
			}
			if class, ok := RequestMethodClass(ctx); !ok || class != expected.method {
				t.Fatalf("authority method class = %v, %v; want %v", class, ok, expected.method)
			}
			if resource, ok := RequestResourceClass(ctx); !ok || resource != expected.resource {
				t.Fatalf("authority resource class = %v, %v; want %v", resource, ok, expected.resource)
			}
			if capability, ok := Capability(ctx); !ok || capability != expected.capability {
				t.Fatalf("authority capability = %q, %v; want %q", capability, ok, expected.capability)
			}
			authorityCalls.Add(1)
			return "caller", nil
		},
		AdmissionLimit: 1, AttemptCap: time.Second,
	},
		WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(commandsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(applicationresultsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(mailboxauthorizationsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(permissionsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(permissionadminsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(permissionauthoringsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(layoutauthoringsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(analyticsauthoringsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(resourceaccesssemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(approvalsemantic.ExecuteCapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(approvalsemantic.ReadCapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(approvalnotificationsemantic.CapabilityName, 500*time.Millisecond),
		WithPlatformCapabilityMaxAdmissionLease(pagelayoutsemantic.CapabilityName, 500*time.Millisecond),
		WithResourceClassAdmissionLimit(ResourceClassHeavyWork, 1),
	)
	if err != nil {
		t.Fatal(err)
	}
	admittedMethods := []string{
		runtimev1.RecordSearchService_SearchRecords_FullMethodName,
		runtimev1.RecordSearchService_ReadSearchScopeOrder_FullMethodName,
		runtimev1.CommandAcceptanceService_AcceptCommand_FullMethodName,
		runtimev1.ApplicationResultService_GetApplicationResult_FullMethodName,
		runtimev1.ConnectedMailboxAuthorizationService_AuthorizeConnectedMailbox_FullMethodName,
		runtimev1.ConnectedMailboxAuthorizationService_ReauthorizeConnectedMailbox_FullMethodName,
		runtimev1.EffectivePermissionService_ReadEffectivePermissions_FullMethodName,
		runtimev1.PermissionAdministrationService_ReadPermissionAdministration_FullMethodName,
		runtimev1.PermissionAdministrationService_MutatePermissionAdministration_FullMethodName,
		runtimev1.PermissionAuthoringService_ReadPermissionAuthoringSnapshot_FullMethodName,
		runtimev1.PermissionAuthoringService_SavePermissionAuthoring_FullMethodName,
		runtimev1.PermissionAuthoringService_GetPermissionAuthoringStatus_FullMethodName,
		runtimev1.LayoutAuthoringService_Submit_FullMethodName,
		runtimev1.LayoutAuthoringService_GetCommand_FullMethodName,
		runtimev1.LayoutAuthoringService_ReadWorkspace_FullMethodName,
		runtimev1.LayoutAuthoringService_ReadStoredLayout_FullMethodName,
		runtimev1.AnalyticsAuthoringService_MutateAnalyticsDefinition_FullMethodName,
		runtimev1.ResourceAccessService_AccessResource_FullMethodName,
		runtimev1.ApprovalService_ExecuteApproval_FullMethodName,
		runtimev1.ApprovalService_ReadApproval_FullMethodName,
		runtimev1.ApprovalNotificationFlowReadService_ReadApprovalNotificationFlows_FullMethodName,
		runtimev1.PageLayoutResolutionService_ResolvePageLayout_FullMethodName,
	}
	for _, fullMethod := range admittedMethods {
		expected := tests[fullMethod]
		ctx, cancel := platformPolicyContext(expected.capability, 200*time.Millisecond)
		handlerCalled := false
		_, err = policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
			FullMethod: fullMethod,
		}, func(ctx context.Context, _ any) (any, error) {
			handlerCalled = true
			if class, ok := RequestMethodClass(ctx); !ok || class != expected.method {
				t.Fatalf("handler method class = %v, %v; want %v", class, ok, expected.method)
			}
			if resource, ok := RequestResourceClass(ctx); !ok || resource != expected.resource {
				t.Fatalf("handler resource class = %v, %v; want %v", resource, ok, expected.resource)
			}
			return nil, nil
		})
		cancel()
		if err != nil || !handlerCalled {
			t.Fatalf("%s admitted call = %v, called=%v", fullMethod, err, handlerCalled)
		}
	}
	expectedAuthorityCalls := int64(len(admittedMethods))
	if got := authorityCalls.Load(); got != expectedAuthorityCalls {
		t.Fatalf("authority calls = %d, want %d", got, expectedAuthorityCalls)
	}

	overlong, cancelOverlong := platformPolicyContext(searchsemantic.CapabilityName, time.Second)
	defer cancelOverlong()
	_, err = policy.UnaryServerInterceptor()(overlong, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordSearchService_SearchRecords_FullMethodName,
	}, func(context.Context, any) (any, error) {
		t.Fatal("overlong Platform attempt reached the handler")
		return nil, nil
	})
	var failure *commonsemantic.Failure
	if !errors.As(mapping.FromPlatformStatusError(err, searchsemantic.CapabilityName), &failure) ||
		failure.Category != commonsemantic.FailureResourceExhausted ||
		failure.LimitKind != commonsemantic.LimitAdmission || authorityCalls.Load() != expectedAuthorityCalls {
		t.Fatalf("overlong Platform admission = %v, %#v; authority calls=%d", err, failure, authorityCalls.Load())
	}
}

func TestSessionLifecycleAuthenticatesWorkloadThenRedactsCredentials(t *testing.T) {
	authority := Authority(func(ctx context.Context, _ string) (VerifiedCaller, error) {
		values, _ := metadata.FromIncomingContext(ctx)
		for _, key := range sessionSensitiveMetadataKeys() {
			if len(values.Get(key)) != 0 {
				t.Fatalf("authority received sensitive metadata %q", key)
			}
		}
		if got := values.Get("authorization"); len(got) != 1 || got[0] != "Bearer workload-secret" {
			t.Fatalf("workload authorization = %#v", got)
		}
		return "sfcrm-workload", nil
	})
	policy := mustIdentityPolicy(t, Config{Authority: authority, AdmissionLimit: 2, AttemptCap: time.Second}, time.Second)

	createCtx, cancelCreate := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(sessionsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(sessionsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, sessionsemantic.CapabilityName,
		"authorization", "Bearer workload-secret",
		SessionIdentityMetadataKey, "user@example.test",
		SessionSecretMetadataKey, "password-secret",
		SessionClientSecretMetadataKey, "client-secret",
	)), time.Second)
	defer cancelCreate()
	result, err := policy.UnaryServerInterceptor()(createCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.SessionLifecycleService_CreateSession_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		if class, ok := RequestMethodClass(ctx); !ok || class != MethodClassWorkloadConsumer {
			t.Fatalf("create method class = %v, %t", class, ok)
		}
		credentials, ok := RuntimeSessionCreateCredentials(ctx)
		if !ok || credentials.Identity != "user@example.test" || credentials.Secret != "password-secret" || credentials.ClientSecret != "client-secret" {
			t.Fatalf("create credentials = %#v, %t", credentials, ok)
		}
		values, _ := metadata.FromIncomingContext(ctx)
		for _, key := range sessionSensitiveMetadataKeys() {
			if len(values.Get(key)) != 0 {
				t.Fatalf("handler metadata retained sensitive key %q", key)
			}
		}
		return "created", nil
	})
	if err != nil || result != "created" {
		t.Fatalf("create result = %#v, %v", result, err)
	}

	readCtx, cancelRead := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(sessionsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(sessionsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, sessionsemantic.CapabilityName,
		"authorization", "Bearer workload-secret",
		EndUserAuthorizationMetadataKey, "Bearer access-token",
		SessionControlProofMetadataKey, "control-proof",
	)), time.Second)
	defer cancelRead()
	result, err = policy.UnaryServerInterceptor()(readCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.SessionLifecycleService_ReadSession_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		if class, ok := RequestMethodClass(ctx); !ok || class != MethodClassWorkloadConsumer {
			t.Fatalf("read method class = %v, %t", class, ok)
		}
		credentials, ok := RuntimeSessionPresentedCredentials(ctx)
		if !ok || credentials.AccessToken != "access-token" || credentials.ControlProof != "control-proof" {
			t.Fatalf("presented credentials = %#v, %t", credentials, ok)
		}
		if stripped := WithoutRuntimeSessionCredentials(ctx); stripped == nil {
			t.Fatal("credential stripping returned nil")
		} else if _, ok := RuntimeSessionPresentedCredentials(stripped); ok {
			t.Fatal("credentials survived explicit owner consumption")
		}
		return "read", nil
	})
	if err != nil || result != "read" {
		t.Fatalf("read result = %#v, %v", result, err)
	}
}

func TestFederatedLoginMethodsUseClosedWorkloadPolicy(t *testing.T) {
	tests := []struct {
		method string
		mapErr func(error) error
		source error
		want   federatedloginsemantic.FailureCategory
	}{
		{
			method: runtimev1.FederatedLoginTransactionService_BeginFederatedLogin_FullMethodName,
			mapErr: mapping.FromFederatedLoginBeginStatusError,
			source: &federatedloginsemantic.Failure{Category: federatedloginsemantic.FailureUnavailable},
			want:   federatedloginsemantic.FailureUnavailable,
		},
		{
			method: runtimev1.FederatedLoginTransactionService_ConsumeFederatedLogin_FullMethodName,
			mapErr: mapping.FromFederatedLoginConsumeStatusError,
			source: &federatedloginsemantic.Failure{Category: federatedloginsemantic.FailureMissing, CanonicalField: "must-not-cross"},
			want:   federatedloginsemantic.FailureUnauthenticated,
		},
	}
	for _, test := range tests {
		policy, ok := registeredMethod(test.method)
		if !ok || policy.capability != federatedloginsemantic.CapabilityName ||
			policy.protocolMajor != federatedloginsemantic.ProtocolMajor || policy.protocolMinor != federatedloginsemantic.ProtocolMinor ||
			policy.methodClass != MethodClassWorkloadConsumer || policy.resourceClass != ResourceClassShortUnary ||
			policy.attemptPolicy != attemptIdentity || policy.statusError == nil {
			t.Fatalf("method %q policy=%#v ok=%t", test.method, policy, ok)
		}
		var failure *federatedloginsemantic.Failure
		if err := test.mapErr(policy.statusError(test.source)); !errors.As(err, &failure) || failure.Category != test.want {
			t.Fatalf("method %q mapped failure=%#v err=%v", test.method, failure, err)
		}
		if failure.CanonicalField != "" {
			t.Fatalf("method %q disclosed state detail: %#v", test.method, failure)
		}
	}
}

func TestSessionSensitiveMetadataCannotReachAnotherRuntimeMethod(t *testing.T) {
	policy := mustIdentityPolicy(t, Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	}, time.Second)
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, semantic.CapabilityName,
		"authorization", "Bearer workload-secret",
		SessionSecretMetadataKey, "must-not-leak",
	)), time.Second)
	defer cancel()
	called := false
	_, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName,
	}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	if status.Code(err) != codes.Unauthenticated || called {
		t.Fatalf("foreign sensitive metadata err=%v called=%t", err, called)
	}
}

func TestSessionCredentialGrammarRejectsForeignAndPartialMetadata(t *testing.T) {
	policy := mustIdentityPolicy(t, Config{
		Authority: func(ctx context.Context, _ string) (VerifiedCaller, error) {
			values, _ := metadata.FromIncomingContext(ctx)
			for _, key := range sessionSensitiveMetadataKeys() {
				if len(values.Get(key)) != 0 {
					t.Fatalf("authority received sensitive metadata %q", key)
				}
			}
			return "sfcrm-workload", nil
		},
		AdmissionLimit: 1, AttemptCap: time.Second,
	}, time.Second)

	tests := []struct {
		name       string
		method     string
		capability string
		metadata   []string
	}{
		{
			name:       "create rejects presented bearer",
			method:     runtimev1.SessionLifecycleService_CreateSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				SessionSecretMetadataKey, "password-secret",
				EndUserAuthorizationMetadataKey, "Bearer foreign-access-token",
			},
		},
		{
			name:       "create rejects malformed presented bearer",
			method:     runtimev1.SessionLifecycleService_CreateSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				SessionSecretMetadataKey, "password-secret",
				EndUserAuthorizationMetadataKey, "malformed-access-token",
			},
		},
		{
			name:       "create rejects control proof",
			method:     runtimev1.SessionLifecycleService_CreateSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				SessionSecretMetadataKey, "password-secret",
				SessionControlProofMetadataKey, "foreign-control-proof",
			},
		},
		{
			name:       "create rejects response credential",
			method:     runtimev1.SessionLifecycleService_CreateSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				SessionSecretMetadataKey, "password-secret",
				IssuedSessionAccessTokenMetadataKey, "foreign-issued-token",
			},
		},
		{
			name:       "create rejects identity without authentication secret",
			method:     runtimev1.SessionLifecycleService_CreateSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata:   []string{SessionIdentityMetadataKey, "user@example.test"},
		},
		{
			name:       "read rejects create identity",
			method:     runtimev1.SessionLifecycleService_ReadSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				EndUserAuthorizationMetadataKey, "Bearer access-token",
				SessionIdentityMetadataKey, "foreign-user@example.test",
			},
		},
		{
			name:       "read rejects create client secret",
			method:     runtimev1.SessionLifecycleService_ReadSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				EndUserAuthorizationMetadataKey, "Bearer access-token",
				SessionClientSecretMetadataKey, "foreign-client-secret",
			},
		},
		{
			name:       "revoke rejects response control proof",
			method:     runtimev1.SessionLifecycleService_RevokeSession_FullMethodName,
			capability: sessionsemantic.CapabilityName,
			metadata: []string{
				EndUserAuthorizationMetadataKey, "Bearer access-token",
				IssuedSessionControlProofMetadataKey, "foreign-issued-proof",
			},
		},
		{
			name:       "identity exchange rejects session create secret",
			method:     runtimev1.IdentityContextService_Exchange_FullMethodName,
			capability: identitysemantic.CapabilityName,
			metadata: []string{
				EndUserAuthorizationMetadataKey, "Bearer access-token",
				SessionSecretMetadataKey, "foreign-secret",
			},
		},
		{
			name:       "identity exchange rejects session control proof",
			method:     runtimev1.IdentityContextService_Exchange_FullMethodName,
			capability: identitysemantic.CapabilityName,
			metadata: []string{
				EndUserAuthorizationMetadataKey, "Bearer access-token",
				SessionControlProofMetadataKey, "foreign-control-proof",
			},
		},
		{
			name:       "authorization exchange rejects verifier without code",
			method:     runtimev1.AuthorizationLifecycleService_ExchangeAuthorizationToken_FullMethodName,
			capability: authorizationsemantic.CapabilityName,
			metadata: []string{
				AuthorizationClientSecretMetadataKey, "client-secret",
				AuthorizationCodeVerifierMetadataKey, "orphan-verifier",
			},
		},
		{
			name:       "authorization exchange rejects code and refresh together",
			method:     runtimev1.AuthorizationLifecycleService_ExchangeAuthorizationToken_FullMethodName,
			capability: authorizationsemantic.CapabilityName,
			metadata: []string{
				AuthorizationCodeMetadataKey, "authorization-code",
				AuthorizationRefreshTokenMetadataKey, "refresh-token",
			},
		},
		{
			name:       "authorization exchange rejects refresh with verifier",
			method:     runtimev1.AuthorizationLifecycleService_ExchangeAuthorizationToken_FullMethodName,
			capability: authorizationsemantic.CapabilityName,
			metadata: []string{
				AuthorizationRefreshTokenMetadataKey, "refresh-token",
				AuthorizationCodeVerifierMetadataKey, "foreign-verifier",
			},
		},
		{
			name:       "ordinary method rejects response credential",
			method:     runtimev1.RecordsService_ReadSelectedRoot_FullMethodName,
			capability: semantic.CapabilityName,
			metadata:   []string{IssuedSessionControlProofMetadataKey, "foreign-issued-proof"},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			pairs := []string{
				envelope.ProtocolMajorMetadataKey, "1",
				envelope.ProtocolMinorMetadataKey, "0",
				envelope.CapabilityMetadataKey, test.capability,
				"authorization", "Bearer workload-secret",
			}
			pairs = append(pairs, test.metadata...)
			ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(pairs...)), time.Second)
			defer cancel()
			handlerCalled := false
			_, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: test.method}, func(context.Context, any) (any, error) {
				handlerCalled = true
				return nil, nil
			})
			if status.Code(err) != codes.Unauthenticated || handlerCalled {
				t.Fatalf("credential grammar err=%v called=%t", err, handlerCalled)
			}
		})
	}
}

func TestMethodClassCredentialGrammarRejectsDelegationOnNonDelegatedMethods(t *testing.T) {
	var authorityCalls atomic.Int64
	policy, err := NewPolicyWithIdentityAndOptions(Config{
		Authority: func(ctx context.Context, method string) (VerifiedCaller, error) {
			values, _ := metadata.FromIncomingContext(ctx)
			switch method {
			case runtimev1.RecordSearchService_SearchRecords_FullMethodName:
				if got := values.Get(envelope.RuntimeDelegationMetadataKey); len(got) != 1 || got[0] != "delegation" {
					t.Fatalf("delegated method authority delegation = %#v", got)
				}
			default:
				if got := values.Get(envelope.RuntimeDelegationMetadataKey); len(got) != 0 {
					t.Fatalf("non-delegated method authority observed delegation = %#v", got)
				}
			}
			authorityCalls.Add(1)
			return "caller", nil
		},
		AdmissionLimit: 1, AttemptCap: time.Second,
	},
		time.Second,
		WithPlatformCapabilityMaxAdmissionLease(commandsemantic.CapabilityName, time.Second),
		WithPlatformCapabilityMaxAdmissionLease(applicationresultsemantic.CapabilityName, time.Second),
		WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, time.Second),
		WithResourceClassAdmissionLimit(ResourceClassHeavyWork, 1),
	)
	if err != nil {
		t.Fatal(err)
	}

	tests := []struct {
		name       string
		method     string
		capability string
		metadata   []string
	}{
		{
			name: "identity exchange", method: runtimev1.IdentityContextService_Exchange_FullMethodName,
			capability: identitysemantic.CapabilityName,
			metadata:   []string{EndUserAuthorizationMetadataKey, "Bearer user-secret"},
		},
		{
			name: "workload command", method: runtimev1.CommandAcceptanceService_AcceptCommand_FullMethodName,
			capability: commandsemantic.CapabilityName,
		},
		{
			name: "workload application result", method: runtimev1.ApplicationResultService_GetApplicationResult_FullMethodName,
			capability: applicationresultsemantic.CapabilityName,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			pairs := []string{
				envelope.ProtocolMajorMetadataKey, "1",
				envelope.ProtocolMinorMetadataKey, "0",
				envelope.CapabilityMetadataKey, test.capability,
				envelope.RuntimeDelegationMetadataKey, "delegation",
			}
			pairs = append(pairs, test.metadata...)
			ctx, cancel := context.WithTimeout(
				metadata.NewIncomingContext(context.Background(), metadata.Pairs(pairs...)),
				500*time.Millisecond,
			)
			defer cancel()
			called := false
			_, callErr := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
				FullMethod: test.method,
			}, func(context.Context, any) (any, error) {
				called = true
				return nil, nil
			})
			if status.Code(callErr) != codes.Unauthenticated || called {
				t.Fatalf("non-delegated credential grammar = %v, called=%v", callErr, called)
			}
		})
	}

	delegatedCtx, cancelDelegated := context.WithTimeout(
		metadata.NewIncomingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1",
			envelope.ProtocolMinorMetadataKey, "0",
			envelope.CapabilityMetadataKey, searchsemantic.CapabilityName,
			envelope.RuntimeDelegationMetadataKey, "delegation",
		)),
		500*time.Millisecond,
	)
	defer cancelDelegated()
	called := false
	_, err = policy.UnaryServerInterceptor()(delegatedCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordSearchService_SearchRecords_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		called = true
		values, _ := metadata.FromIncomingContext(ctx)
		if got := values.Get(envelope.RuntimeDelegationMetadataKey); len(got) != 1 || got[0] != "delegation" {
			t.Fatalf("delegated handler delegation = %#v", got)
		}
		return nil, nil
	})
	if err != nil || !called || authorityCalls.Load() != 4 {
		t.Fatalf("delegated credential grammar = %v, called=%v, authority=%d", err, called, authorityCalls.Load())
	}
}

func TestPlatformCapabilitiesRequireTheirOwningResourcePool(t *testing.T) {
	authority := func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }
	config := Config{Authority: authority, AdmissionLimit: 1, AttemptCap: time.Second}
	tests := []struct {
		capability string
		resource   ResourceClass
	}{
		{catalogsemantic.AppMetadataCapabilityName, ResourceClassCatalogWatch},
		{catalogsemantic.AppMetadataV2CapabilityName, ResourceClassCatalogWatch},
		{catalogsemantic.AppMetadataV3CapabilityName, ResourceClassCatalogWatch},
		{catalogsemantic.IntegrationBindingCapabilityName, ResourceClassCatalogWatch},
		{searchsemantic.CapabilityName, ResourceClassHeavyWork},
		{flowsemantic.CapabilityName, ResourceClassHeavyWork},
		{reportsemantic.CapabilityName, ResourceClassHeavyWork},
		{auditsemantic.CapabilityName, ResourceClassHeavyWork},
		{listviewsemantic.CapabilityName, ResourceClassHeavyWork},
		{metadataadminsemantic.CapabilityName, ResourceClassHeavyWork},
		{permissionsemantic.CapabilityName, ResourceClassHeavyWork},
		{permissionadminsemantic.CapabilityName, ResourceClassHeavyWork},
		{permissionauthoringsemantic.CapabilityName, ResourceClassHeavyWork},
		{layoutauthoringsemantic.CapabilityName, ResourceClassHeavyWork},
		{layoutauthoringv2semantic.CapabilityName, ResourceClassHeavyWork},
		{analyticsauthoringsemantic.CapabilityName, ResourceClassHeavyWork},
		{resourceaccesssemantic.CapabilityName, ResourceClassHeavyWork},
		{approvalsemantic.ExecuteCapabilityName, ResourceClassHeavyWork},
		{approvalsemantic.ReadCapabilityName, ResourceClassHeavyWork},
		{approvalnotificationsemantic.CapabilityName, ResourceClassHeavyWork},
		{filessemantic.CapabilityName, ResourceClassFileStream},
	}
	for _, test := range tests {
		t.Run(test.capability, func(t *testing.T) {
			if class, ok := PlatformCapabilityResourceClass(test.capability); !ok || class != test.resource {
				t.Fatalf("PlatformCapabilityResourceClass() = %v, %v; want %v, true", class, ok, test.resource)
			}
			capability := WithPlatformCapabilityMaxAdmissionLease(test.capability, time.Second)
			if policy, err := NewPolicyWithOptions(config, capability); err == nil || policy != nil {
				t.Fatalf("capability without %v pool = %#v, %v; want rejected", test.resource, policy, err)
			}
			if policy, err := NewPolicyWithOptions(
				config,
				capability,
				WithResourceClassAdmissionLimit(test.resource, 1),
			); err != nil || policy == nil {
				t.Fatalf("capability with %v pool = %#v, %v; want admitted", test.resource, policy, err)
			}
		})
	}
	for _, capability := range []string{
		commandsemantic.CapabilityName,
		applicationresultsemantic.CapabilityName,
		externaleffectsemantic.CapabilityName,
		mailboxauthorizationsemantic.CapabilityName,
		pagelayoutsemantic.CapabilityName,
		outboundintentsemantic.CapabilityName,
	} {
		if policy, err := NewPolicyWithOptions(
			config,
			WithPlatformCapabilityMaxAdmissionLease(capability, time.Second),
		); err != nil || policy == nil {
			t.Fatalf("short-unary %s requires no extra pool: %#v, %v", capability, policy, err)
		}
		if class, ok := PlatformCapabilityResourceClass(capability); !ok || class != ResourceClassShortUnary {
			t.Fatalf("%s resource class = %v, %v", capability, class, ok)
		}
	}
	if class, ok := PlatformCapabilityResourceClass("runtime.unknown.v1"); ok || class != ResourceClassUnspecified {
		t.Fatalf("unknown resource class = %v, %v", class, ok)
	}
}

func TestOutboundIntentUnaryCompatibilityDoesNotEnableAttachmentStreamPool(t *testing.T) {
	policy, err := NewPolicyWithOptions(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1,
		AttemptCap:     time.Second,
	}, WithPlatformCapabilityMaxAdmissionLease(outboundintentsemantic.CapabilityName, time.Second))
	if err != nil {
		t.Fatal(err)
	}

	unaryCtx, cancelUnary := platformPolicyContext(outboundintentsemantic.CapabilityName, 500*time.Millisecond)
	defer cancelUnary()
	called := false
	_, err = policy.UnaryServerInterceptor()(unaryCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.OutboundIntentService_ReadOutboundIntentGeneration_FullMethodName,
	}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	if err != nil || !called {
		t.Fatalf("legacy outbound unary = %v, called=%v", err, called)
	}

	streamCtx, cancelStream := platformPolicyContext(outboundintentsemantic.CapabilityName, 500*time.Millisecond)
	defer cancelStream()
	called = false
	err = policy.StreamServerInterceptor()(
		nil,
		&policyTestServerStream{ctx: streamCtx},
		&grpc.StreamServerInfo{
			FullMethod:     runtimev1.OutboundIntentService_OpenOutboundMailAttachment_FullMethodName,
			IsServerStream: true,
		},
		func(any, grpc.ServerStream) error {
			called = true
			return nil
		},
	)
	if status.Code(err) != codes.Unimplemented || called {
		t.Fatalf("unisolated attachment stream = %v, called=%v", err, called)
	}
}

func platformPolicyContext(capability string, timeout time.Duration) (context.Context, context.CancelFunc) {
	ctx := metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(commonsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(commonsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, capability,
	))
	return context.WithTimeout(ctx, timeout)
}

func TestResourceAdmissionClassesDoNotStarveShortUnary(t *testing.T) {
	policy, err := NewPolicyWithOptions(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 2, AttemptCap: time.Second,
	},
		WithPlatformCapabilityMaxAdmissionLease(searchsemantic.CapabilityName, time.Second),
		WithResourceClassAdmissionLimit(ResourceClassHeavyWork, 1),
	)
	if err != nil {
		t.Fatal(err)
	}
	incoming := func(capability string) (context.Context, context.CancelFunc) {
		ctx := metadata.NewIncomingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(commonsemantic.ProtocolMajor), 10),
			envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(commonsemantic.ProtocolMinor), 10),
			envelope.CapabilityMetadataKey, capability,
		))
		return context.WithTimeout(ctx, 500*time.Millisecond)
	}

	entered := make(chan struct{})
	release := make(chan struct{})
	firstDone := make(chan error, 1)
	firstCtx, cancelFirst := incoming(searchsemantic.CapabilityName)
	defer cancelFirst()
	go func() {
		_, callErr := policy.UnaryServerInterceptor()(firstCtx, nil, &grpc.UnaryServerInfo{
			FullMethod: runtimev1.RecordSearchService_SearchRecords_FullMethodName,
		}, func(context.Context, any) (any, error) {
			close(entered)
			<-release
			return nil, nil
		})
		firstDone <- callErr
	}()
	<-entered

	secondCtx, cancelSecond := incoming(searchsemantic.CapabilityName)
	defer cancelSecond()
	_, err = policy.UnaryServerInterceptor()(secondCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordSearchService_SearchRecords_FullMethodName,
	}, func(context.Context, any) (any, error) {
		t.Fatal("second heavy query reached the handler")
		return nil, nil
	})
	var failure *commonsemantic.Failure
	if !errors.As(mapping.FromPlatformStatusError(err, searchsemantic.CapabilityName), &failure) ||
		failure.Category != commonsemantic.FailureResourceExhausted {
		t.Fatalf("full heavy pool = %v, %#v", err, failure)
	}

	shortCtx, cancelShort := incoming(semantic.CapabilityName)
	defer cancelShort()
	shortCalled := false
	if _, err := policy.UnaryServerInterceptor()(shortCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		shortCalled = true
		if class, _ := RequestResourceClass(ctx); class != ResourceClassShortUnary {
			t.Fatalf("short resource class = %v", class)
		}
		return nil, nil
	}); err != nil || !shortCalled {
		t.Fatalf("short unary while heavy full = %v, called=%v", err, shortCalled)
	}

	close(release)
	if err := <-firstDone; err != nil {
		t.Fatalf("first heavy query = %v", err)
	}
	if snapshot := policy.ResourceAdmissionSnapshot(ResourceClassHeavyWork); snapshot.Limit != 1 ||
		snapshot.Accepted != 1 || snapshot.RejectedFull != 1 {
		t.Fatalf("heavy snapshot = %#v", snapshot)
	}
	if snapshot := policy.ResourceAdmissionSnapshot(ResourceClassShortUnary); snapshot.Accepted != 1 ||
		snapshot.RejectedFull != 0 {
		t.Fatalf("short snapshot = %#v", snapshot)
	}
	if snapshot := policy.AdmissionSnapshot(); snapshot.Limit != 2 || snapshot.Accepted != 2 ||
		snapshot.RejectedFull != 1 {
		t.Fatalf("aggregate compatibility snapshot = %#v", snapshot)
	}
	if snapshot := policy.ResourceAdmissionSnapshot(ResourceClass(255)); snapshot != (AdmissionSnapshot{}) {
		t.Fatalf("unknown resource-class snapshot = %#v", snapshot)
	}
}

func TestLongLivedResourceClassesDoNotStarveShortUnary(t *testing.T) {
	tests := []struct {
		name         string
		capability   string
		resource     ResourceClass
		method       string
		clientStream bool
		serverStream bool
	}{
		{
			name: "file stream", capability: filessemantic.CapabilityName,
			resource: ResourceClassFileStream,
			method:   runtimev1.AuthorizedFilesService_CreateFile_FullMethodName, clientStream: true,
		},
		{
			name: "outbound attachment stream", capability: outboundintentsemantic.CapabilityName,
			resource: ResourceClassFileStream,
			method:   runtimev1.OutboundIntentService_OpenOutboundMailAttachment_FullMethodName, serverStream: true,
		},
		{
			name: "catalog watch", capability: catalogsemantic.AppMetadataCapabilityName,
			resource: ResourceClassCatalogWatch,
			method:   catalogv1.ConsumerViewService_WatchAppMetadataView_FullMethodName, serverStream: true,
		},
		{
			name: "catalog v2 watch", capability: catalogsemantic.AppMetadataV2CapabilityName,
			resource: ResourceClassCatalogWatch,
			method:   catalogv1.ConsumerViewService_WatchAppMetadataViewV2_FullMethodName, serverStream: true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			policy, err := NewPolicyWithOptions(Config{
				Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
				AdmissionLimit: 2,
				AttemptCap:     time.Second,
			},
				WithPlatformCapabilityMaxAdmissionLease(test.capability, time.Second),
				WithResourceClassAdmissionLimit(test.resource, 1),
			)
			if err != nil {
				t.Fatal(err)
			}
			incoming := func(capability string) (context.Context, context.CancelFunc) {
				ctx := metadata.NewIncomingContext(context.Background(), metadata.Pairs(
					envelope.ProtocolMajorMetadataKey, "1",
					envelope.ProtocolMinorMetadataKey, "0",
					envelope.CapabilityMetadataKey, capability,
				))
				return context.WithTimeout(ctx, 500*time.Millisecond)
			}

			entered := make(chan struct{})
			release := make(chan struct{})
			firstDone := make(chan error, 1)
			firstCtx, cancelFirst := incoming(test.capability)
			defer cancelFirst()
			go func() {
				firstDone <- policy.StreamServerInterceptor()(
					nil,
					&policyTestServerStream{ctx: firstCtx},
					&grpc.StreamServerInfo{
						FullMethod:     test.method,
						IsClientStream: test.clientStream,
						IsServerStream: test.serverStream,
					},
					func(any, grpc.ServerStream) error {
						close(entered)
						<-release
						return nil
					},
				)
			}()
			<-entered

			secondCtx, cancelSecond := incoming(test.capability)
			defer cancelSecond()
			err = policy.StreamServerInterceptor()(
				nil,
				&policyTestServerStream{ctx: secondCtx},
				&grpc.StreamServerInfo{
					FullMethod:     test.method,
					IsClientStream: test.clientStream,
					IsServerStream: test.serverStream,
				},
				func(any, grpc.ServerStream) error {
					t.Fatal("second long-lived stream reached the handler")
					return nil
				},
			)
			var failure *commonsemantic.Failure
			if !errors.As(mapping.FromPlatformStatusError(err, test.capability), &failure) ||
				failure.Category != commonsemantic.FailureResourceExhausted {
				t.Fatalf("full %v pool = %v, %#v", test.resource, err, failure)
			}

			shortCtx, cancelShort := incoming(semantic.CapabilityName)
			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 %v full = %v, called=%v", test.resource, err, shortCalled)
			}

			close(release)
			if err := <-firstDone; err != nil {
				t.Fatalf("first %v stream = %v", test.resource, err)
			}
			if snapshot := policy.ResourceAdmissionSnapshot(test.resource); snapshot.Limit != 1 ||
				snapshot.Accepted != 1 || snapshot.RejectedFull != 1 {
				t.Fatalf("%v snapshot = %#v", test.resource, snapshot)
			}
			if snapshot := policy.AdmissionSnapshot(); snapshot.Limit != 2 ||
				snapshot.Accepted != 2 || snapshot.RejectedFull != 1 {
				t.Fatalf("aggregate snapshot = %#v", snapshot)
			}
		})
	}
}

func TestResourceAdmissionClassesRemainInsideTheProcessBound(t *testing.T) {
	policy, err := NewPolicyWithOptions(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 2,
		AttemptCap:     time.Second,
	},
		WithResourceClassAdmissionLimit(ResourceClassHeavyWork, 2),
		WithResourceClassAdmissionLimit(ResourceClassFileStream, 2),
	)
	if err != nil {
		t.Fatal(err)
	}

	result, heavy, maximum := policy.admit(ResourceClassHeavyWork)
	if result != admissionAccepted || maximum != 0 {
		t.Fatalf("heavy admission = %v, maximum=%d", result, maximum)
	}
	result, files, maximum := policy.admit(ResourceClassFileStream)
	if result != admissionAccepted || maximum != 0 {
		t.Fatalf("file admission = %v, maximum=%d", result, maximum)
	}
	result, _, maximum = policy.admit(ResourceClassShortUnary)
	if result != admissionFull || maximum != 2 {
		t.Fatalf("third admission = %v, maximum=%d", result, maximum)
	}
	if snapshot := policy.AdmissionSnapshot(); snapshot.Limit != 2 ||
		snapshot.InFlight != 2 || snapshot.Accepted != 2 || snapshot.RejectedFull != 1 {
		t.Fatalf("process snapshot = %#v", snapshot)
	}
	if snapshot := policy.ResourceAdmissionSnapshot(ResourceClassHeavyWork); snapshot.Limit != 2 ||
		snapshot.InFlight != 1 || snapshot.Accepted != 1 {
		t.Fatalf("heavy snapshot = %#v", snapshot)
	}
	if snapshot := policy.ResourceAdmissionSnapshot(ResourceClassFileStream); snapshot.Limit != 2 ||
		snapshot.InFlight != 1 || snapshot.Accepted != 1 {
		t.Fatalf("file snapshot = %#v", snapshot)
	}

	policy.release(files)
	policy.release(heavy)
	if snapshot := policy.AdmissionSnapshot(); snapshot.InFlight != 0 {
		t.Fatalf("released process snapshot = %#v", snapshot)
	}
}

func TestUnconfiguredResourceClassesPreserveOneSharedD1Pool(t *testing.T) {
	policy, err := NewPolicy(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1,
		AttemptCap:     time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}

	result, heavy, maximum := policy.admit(ResourceClassHeavyWork)
	if result != admissionAccepted || maximum != 0 {
		t.Fatalf("heavy admission = %v, maximum=%d", result, maximum)
	}
	result, _, maximum = policy.admit(ResourceClassShortUnary)
	if result != admissionFull || maximum != 1 {
		t.Fatalf("short admission = %v, maximum=%d", result, maximum)
	}
	if policy.admission[ResourceClassHeavyWork] != policy.admission[ResourceClassShortUnary] {
		t.Fatal("unconfigured resource classes no longer share the released D1 pool")
	}
	if snapshot := policy.AdmissionSnapshot(); snapshot.Limit != 1 ||
		snapshot.InFlight != 1 || snapshot.Accepted != 1 || snapshot.RejectedFull != 1 {
		t.Fatalf("process snapshot = %#v", snapshot)
	}
	policy.release(heavy)
}

func TestResourceAdmissionLimitCannotExceedProcessBound(t *testing.T) {
	_, err := NewPolicyWithOptions(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1,
		AttemptCap:     time.Second,
	}, WithResourceClassAdmissionLimit(ResourceClassHeavyWork, 2))
	if err == nil || !strings.Contains(err.Error(), "cannot exceed the process admission limit") {
		t.Fatalf("NewPolicyWithOptions() error = %v", err)
	}
}

func TestRecordMutationStreamUsesTheSameTrustedAdmissionStateMachine(t *testing.T) {
	policy, err := NewPolicyWithOptions(Config{
		Authority: func(ctx context.Context, method string) (VerifiedCaller, error) {
			if method != runtimev1.RecordMutationsService_MutateRecords_FullMethodName {
				t.Fatalf("authority method = %q", method)
			}
			return "mutation-caller", nil
		},
		AdmissionLimit: 1, AttemptCap: time.Second,
	}, WithRecordMutationAttemptCap(150*time.Millisecond))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)), 100*time.Millisecond)
	defer cancel()
	callerDeadline, _ := ctx.Deadline()
	stream := &policyTestServerStream{ctx: ctx}
	called := false
	err = policy.StreamServerInterceptor()(nil, stream, &grpc.StreamServerInfo{
		FullMethod:     runtimev1.RecordMutationsService_MutateRecords_FullMethodName,
		IsClientStream: true,
	}, func(_ any, admitted grpc.ServerStream) error {
		called = true
		if capability, ok := Capability(admitted.Context()); !ok || capability != mutationsemantic.CapabilityName {
			t.Fatalf("capability = %q, %v", capability, ok)
		}
		if caller, ok := Caller(admitted.Context()); !ok || caller != "mutation-caller" {
			t.Fatalf("caller = %q, %v", caller, ok)
		}
		if policy.InFlight() != 1 {
			t.Fatalf("in-flight inside handler = %d", policy.InFlight())
		}
		deadline, ok := admitted.Context().Deadline()
		if !ok || !deadline.Equal(callerDeadline) {
			t.Fatalf("mutation stream deadline = %v, want original transport deadline %v", deadline, callerDeadline)
		}
		if got := status.Code(StatusError(admitted.Context(), &mutationsemantic.Failure{Category: mutationsemantic.FailureMissing})); got != codes.Internal {
			t.Fatalf("MutateRecords exposed top-level MISSING: %s", got)
		}
		return nil
	})
	if err != nil || !called || policy.InFlight() != 0 {
		t.Fatalf("err=%v called=%v in-flight=%d", err, called, policy.InFlight())
	}
}

func TestRecordMutationStreamRejectsCallerDeadlineBeyondHostBoundBeforeAdmission(t *testing.T) {
	var authorityCalls atomic.Int64
	policy, err := NewPolicyWithOptions(Config{
		Authority: func(context.Context, string) (VerifiedCaller, error) {
			authorityCalls.Add(1)
			return "mutation-caller", nil
		},
		AdmissionLimit: 1, AttemptCap: time.Second,
	}, WithRecordMutationAttemptCap(50*time.Millisecond))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)), time.Second)
	defer cancel()
	called := false
	err = policy.StreamServerInterceptor()(nil, &policyTestServerStream{ctx: ctx}, &grpc.StreamServerInfo{
		FullMethod: runtimev1.RecordMutationsService_MutateRecords_FullMethodName, IsClientStream: true,
	}, func(any, grpc.ServerStream) error { called = true; return nil })
	var failure *mutationsemantic.Failure
	if !errors.As(mapping.FromRecordMutationStatusError(err), &failure) ||
		failure.Category != mutationsemantic.FailureResourceExhausted ||
		failure.LimitKind != mutationsemantic.LimitAdmission || failure.RetryAfterMillis != 0 ||
		called || authorityCalls.Load() != 0 || policy.InFlight() != 0 {
		t.Fatalf("error=%v failure=%#v called=%v authority=%d in-flight=%d", err, failure, called, authorityCalls.Load(), policy.InFlight())
	}
}

func TestAuthorityPanicCannotLeakAdmission(t *testing.T) {
	policy := mustPolicy(t, Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { panic("authority failed") },
		AdmissionLimit: 1, AttemptCap: time.Second,
	})
	ctx, cancel := context.WithTimeout(serverProtocolContext(context.Background()), time.Second)
	defer cancel()
	func() {
		defer func() {
			if recover() == nil {
				t.Fatal("authority panic was not propagated")
			}
		}()
		_, _ = policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
			FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName,
		}, func(context.Context, any) (any, error) { return nil, nil })
	}()
	if got := policy.InFlight(); got != 0 {
		t.Fatalf("in-flight after authority panic = %d, want 0", got)
	}
}

func TestStatusErrorNilContextFailsClosed(t *testing.T) {
	if got := status.Code(StatusError(nil, errors.New("boom"))); got != codes.Internal {
		t.Fatalf("StatusError(nil) = %s, want INTERNAL", got)
	}
}

func TestRecordMutationMethodsAreDisabledWithoutExplicitAttemptCap(t *testing.T) {
	policy := mustPolicy(t, Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	})
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)), time.Second)
	defer cancel()
	called := false
	err := policy.StreamServerInterceptor()(nil, &policyTestServerStream{ctx: ctx}, &grpc.StreamServerInfo{
		FullMethod: runtimev1.RecordMutationsService_MutateRecords_FullMethodName, IsClientStream: true,
	}, func(any, grpc.ServerStream) error { called = true; return nil })
	if status.Code(err) != codes.Unimplemented || called {
		t.Fatalf("disabled stream err=%v called=%v", err, called)
	}
}

func TestMutationReconciliationAloneMayReturnMissing(t *testing.T) {
	policy, err := NewPolicyWithOptions(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	}, WithRecordMutationAttemptCap(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)), time.Second)
	defer cancel()
	_, err = policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordMutationsService_GetMutationStatus_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		return nil, StatusError(ctx, &mutationsemantic.Failure{Category: mutationsemantic.FailureMissing})
	})
	if status.Code(err) != codes.NotFound {
		t.Fatalf("reconciliation MISSING = %v", err)
	}
}

type policyTestServerStream struct {
	grpc.ServerStream
	ctx context.Context
}

func (s *policyTestServerStream) Context() context.Context { return s.ctx }

func TestIdentityExchangeRequiresAnExplicitHostAttemptCap(t *testing.T) {
	var authorityCalls atomic.Int64
	policy := mustPolicy(t, Config{
		Authority: func(context.Context, string) (VerifiedCaller, error) {
			authorityCalls.Add(1)
			return "caller", nil
		},
		AdmissionLimit: 1,
		AttemptCap:     time.Second,
	})
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, identitysemantic.CapabilityName,
		EndUserAuthorizationMetadataKey, "Bearer user-secret",
	)), time.Second)
	defer cancel()
	result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.IdentityContextService_Exchange_FullMethodName,
	}, func(context.Context, any) (any, error) {
		t.Fatal("disabled identity method reached handler")
		return nil, nil
	})
	if result != nil || status.Code(err) != codes.Unimplemented || authorityCalls.Load() != 0 {
		t.Fatalf("result=%#v err=%v authority=%d", result, err, authorityCalls.Load())
	}
}

func TestIdentityExchangeAuthenticatesWorkloadBeforeCapturingAndRedactsBearer(t *testing.T) {
	const bearer = "raw-user-session-secret"
	authorityCalls := atomic.Int64{}
	policy := mustIdentityPolicy(t, Config{
		AdmissionLimit: 1,
		AttemptCap:     5 * time.Millisecond,
		Authority: func(ctx context.Context, method string) (VerifiedCaller, error) {
			authorityCalls.Add(1)
			if method != runtimev1.IdentityContextService_Exchange_FullMethodName {
				t.Fatalf("method = %q", method)
			}
			if values, _ := metadata.FromIncomingContext(ctx); len(values.Get(EndUserAuthorizationMetadataKey)) != 0 {
				t.Fatal("workload authority observed end-user bearer")
			} else if got := values.Get("authorization"); len(got) != 1 || got[0] != "Bearer workload-secret" {
				t.Fatalf("workload authority lost its authorization metadata: %#v", got)
			}
			return "workload-handle", nil
		},
	}, 750*time.Millisecond)
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1",
		envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, identitysemantic.CapabilityName,
		"authorization", "Bearer workload-secret",
		EndUserAuthorizationMetadataKey, "Bearer "+bearer,
		"traceparent", "trace-1",
	)), time.Second)
	defer cancel()
	called := false
	result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.IdentityContextService_Exchange_FullMethodName}, func(ctx context.Context, _ any) (any, error) {
		called = true
		if caller, ok := Caller(ctx); !ok || caller != "workload-handle" {
			t.Fatalf("caller = %q, %v", caller, ok)
		}
		if got, ok := EndUserBearer(ctx); !ok || got != bearer {
			t.Fatalf("bearer = %q, %v", got, ok)
		}
		values, _ := metadata.FromIncomingContext(ctx)
		if len(values.Get(EndUserAuthorizationMetadataKey)) != 0 || len(values.Get("traceparent")) != 1 ||
			len(values.Get("authorization")) != 1 || values.Get("authorization")[0] != "Bearer workload-secret" {
			t.Fatalf("sanitized metadata = %#v", values)
		}
		deadline, ok := ctx.Deadline()
		if !ok || time.Until(deadline) < 500*time.Millisecond {
			t.Fatalf("identity deadline was copied from selected-root attempt cap: %v", time.Until(deadline))
		}
		return "ok", nil
	})
	if err != nil || result != "ok" || !called || authorityCalls.Load() != 1 {
		t.Fatalf("result=%#v err=%v called=%v authority=%d", result, err, called, authorityCalls.Load())
	}
}

func TestIdentityExchangeUsesItsExplicitHostAttemptCap(t *testing.T) {
	policy := mustIdentityPolicy(t, Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1,
		AttemptCap:     time.Second,
	}, 20*time.Millisecond)
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, identitysemantic.CapabilityName,
		EndUserAuthorizationMetadataKey, "Bearer user-secret",
	)), time.Second)
	defer cancel()
	started := time.Now()
	result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.IdentityContextService_Exchange_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		<-ctx.Done()
		if !errors.Is(ctx.Err(), context.DeadlineExceeded) {
			t.Fatalf("handler context error = %v", ctx.Err())
		}
		return "bounded", nil
	})
	if result != "bounded" || err != nil || time.Since(started) > 500*time.Millisecond {
		t.Fatalf("result=%#v err=%v elapsed=%v", result, err, time.Since(started))
	}
}

func TestRecordAccessRequiresExplicitAdmissionLeaseAndPreservesCallerDeadline(t *testing.T) {
	policy, err := NewPolicyWithOptions(Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: 10 * time.Millisecond,
	}, WithRecordAccessMaxAdmissionLease(200*time.Millisecond))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, accesssemantic.CapabilityName,
	)), 150*time.Millisecond)
	defer cancel()
	callerDeadline, _ := ctx.Deadline()
	result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName,
	}, func(ctx context.Context, _ any) (any, error) {
		if capability, ok := Capability(ctx); !ok || capability != accesssemantic.CapabilityName {
			t.Fatalf("capability = %q, %v", capability, ok)
		}
		deadline, ok := ctx.Deadline()
		if !ok || !deadline.Equal(callerDeadline) {
			t.Fatalf("handler deadline = %v, want unchanged caller deadline %v", deadline, callerDeadline)
		}
		mapped := StatusError(ctx, &accesssemantic.Failure{Category: accesssemantic.FailureForbidden})
		if status.Code(mapped) != codes.PermissionDenied {
			t.Fatalf("record-access status = %v", mapped)
		}
		return "ok", nil
	})
	if err != nil || result != "ok" {
		t.Fatalf("result=%#v err=%v", result, err)
	}

	longCtx, longCancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, accesssemantic.CapabilityName,
	)), time.Second)
	defer longCancel()
	authorityCalls := 0
	policy.authority = func(context.Context, string) (VerifiedCaller, error) {
		authorityCalls++
		return "caller", nil
	}
	_, err = policy.UnaryServerInterceptor()(longCtx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName,
	}, func(context.Context, any) (any, error) {
		t.Fatal("over-lease request reached handler")
		return nil, nil
	})
	var failure *accesssemantic.Failure
	if !errors.As(mapping.FromRecordAccessStatusError(err), &failure) || failure.Category != accesssemantic.FailureResourceExhausted ||
		failure.LimitKind != accesssemantic.LimitAdmission || authorityCalls != 0 {
		t.Fatalf("over-lease error=%#v authority calls=%d", err, authorityCalls)
	}
}

func TestRecordAccessIsDisabledWithoutExplicitHostBound(t *testing.T) {
	policy := mustPolicy(t, Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	})
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, accesssemantic.CapabilityName,
	)), time.Second)
	defer cancel()
	var handlerCalls int
	_, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
		FullMethod: runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName,
	}, func(context.Context, any) (any, error) {
		handlerCalls++
		return nil, nil
	})
	if status.Code(err) != codes.Unimplemented || handlerCalls != 0 {
		t.Fatalf("error=%v handler calls=%d, want disabled before handler", err, handlerCalls)
	}
}

func TestQueryRequiresExplicitAdmissionLeaseAndPreservesCallerDeadline(t *testing.T) {
	for _, method := range []string{
		runtimev1.RecordQueryService_QueryRecords_FullMethodName,
		runtimev1.RecordQueryService_QueryHistory_FullMethodName,
	} {
		t.Run(method, func(t *testing.T) {
			policy, err := NewPolicyWithOptions(Config{
				Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
				AdmissionLimit: 1, AttemptCap: 10 * time.Millisecond,
			}, WithQueryMaxAdmissionLease(200*time.Millisecond))
			if err != nil {
				t.Fatal(err)
			}
			ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
				envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
				envelope.CapabilityMetadataKey, querysemantic.CapabilityName,
			)), 150*time.Millisecond)
			defer cancel()
			callerDeadline, _ := ctx.Deadline()
			result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(ctx context.Context, _ any) (any, error) {
				if capability, ok := Capability(ctx); !ok || capability != querysemantic.CapabilityName {
					t.Fatalf("capability = %q, %v", capability, ok)
				}
				deadline, ok := ctx.Deadline()
				if !ok || !deadline.Equal(callerDeadline) {
					t.Fatalf("handler deadline = %v, want unchanged caller deadline %v", deadline, callerDeadline)
				}
				mapped := StatusError(ctx, &querysemantic.Failure{Category: querysemantic.FailureForbidden})
				if status.Code(mapped) != codes.PermissionDenied {
					t.Fatalf("query status = %v", mapped)
				}
				return "ok", nil
			})
			if err != nil || result != "ok" {
				t.Fatalf("result=%#v err=%v", result, err)
			}

			longCtx, longCancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
				envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
				envelope.CapabilityMetadataKey, querysemantic.CapabilityName,
			)), time.Second)
			defer longCancel()
			authorityCalls := 0
			policy.authority = func(context.Context, string) (VerifiedCaller, error) {
				authorityCalls++
				return "caller", nil
			}
			_, err = policy.UnaryServerInterceptor()(longCtx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(context.Context, any) (any, error) {
				t.Fatal("over-lease query reached handler")
				return nil, nil
			})
			var failure *querysemantic.Failure
			if !errors.As(mapping.FromQueryStatusError(err), &failure) || failure.Category != querysemantic.FailureResourceExhausted ||
				failure.LimitKind != querysemantic.LimitAdmission || authorityCalls != 0 {
				t.Fatalf("over-lease error=%#v authority calls=%d", err, authorityCalls)
			}
		})
	}
}

func TestQueryIsDisabledWithoutExplicitHostBound(t *testing.T) {
	policy := mustPolicy(t, Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	})
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, querysemantic.CapabilityName,
	)), time.Second)
	defer cancel()
	for _, method := range []string{
		runtimev1.RecordQueryService_QueryRecords_FullMethodName,
		runtimev1.RecordQueryService_QueryHistory_FullMethodName,
	} {
		handlerCalls := 0
		_, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(context.Context, any) (any, error) {
			handlerCalls++
			return nil, nil
		})
		if status.Code(err) != codes.Unimplemented || handlerCalls != 0 {
			t.Fatalf("%s error=%v handler calls=%d, want disabled before handler", method, err, handlerCalls)
		}
	}
}

func TestRelatedCountRequiresItsOwnExplicitHostBound(t *testing.T) {
	config := Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	}
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, relatedsemantic.CapabilityName,
	)), 500*time.Millisecond)
	defer cancel()

	disabled := mustPolicy(t, config)
	called := false
	_, err := disabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RelatedRecordCountService_CountRelatedRecords_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	if status.Code(err) != codes.Unimplemented || called {
		t.Fatalf("disabled error=%v called=%v", err, called)
	}

	enabled, err := NewPolicyWithOptions(config, WithRelatedCountMaxAdmissionLease(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	_, err = enabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RelatedRecordCountService_CountRelatedRecords_FullMethodName}, func(handlerCtx context.Context, _ any) (any, error) {
		called = true
		if capability, ok := Capability(handlerCtx); !ok || capability != relatedsemantic.CapabilityName {
			t.Fatalf("capability=%q ok=%v", capability, ok)
		}
		return "ok", nil
	})
	if err != nil || !called {
		t.Fatalf("enabled error=%v called=%v", err, called)
	}

	overLease, cancelOverLease := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, relatedsemantic.CapabilityName,
	)), 2*time.Second)
	defer cancelOverLease()
	called = false
	_, err = enabled.UnaryServerInterceptor()(overLease, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RelatedRecordCountService_CountRelatedRecords_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	var failure *relatedsemantic.Failure
	if !errors.As(mapping.FromRelatedCountStatusError(err), &failure) || failure.Category != relatedsemantic.FailureResourceExhausted || failure.LimitKind != relatedsemantic.LimitAdmission || called {
		t.Fatalf("over-lease error=%#v called=%v", err, called)
	}
}

func TestRecycleBinRequiresItsOwnExplicitHostBound(t *testing.T) {
	config := Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	}
	requestContext := func(duration time.Duration) (context.Context, context.CancelFunc) {
		return context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
			envelope.CapabilityMetadataKey, recyclesemantic.CapabilityName,
		)), duration)
	}
	ctx, cancel := requestContext(500 * time.Millisecond)
	defer cancel()

	disabled := mustPolicy(t, config)
	called := false
	_, err := disabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecycleBinReadService_ListRecycleBin_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	if status.Code(err) != codes.Unimplemented || called {
		t.Fatalf("disabled error=%v called=%v", err, called)
	}

	enabled, err := NewPolicyWithOptions(config, WithRecycleBinMaxAdmissionLease(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	_, err = enabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecycleBinReadService_ListRecycleBin_FullMethodName}, func(handlerCtx context.Context, _ any) (any, error) {
		called = true
		if capability, ok := Capability(handlerCtx); !ok || capability != recyclesemantic.CapabilityName {
			t.Fatalf("capability=%q ok=%v", capability, ok)
		}
		return "ok", nil
	})
	if err != nil || !called {
		t.Fatalf("enabled error=%v called=%v", err, called)
	}

	overLease, cancelOverLease := requestContext(2 * time.Second)
	defer cancelOverLease()
	called = false
	_, err = enabled.UnaryServerInterceptor()(overLease, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecycleBinReadService_ListRecycleBin_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	var failure *recyclesemantic.Failure
	if !errors.As(mapping.FromRecycleBinStatusError(err), &failure) || failure.Category != recyclesemantic.FailureResourceExhausted || failure.LimitKind != recyclesemantic.LimitAdmission || called {
		t.Fatalf("over-lease error=%#v called=%v", err, called)
	}
}

func TestRecycleBinMutationsRequireTheirOwnExplicitHostBound(t *testing.T) {
	config := Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	}
	requestContext := func(duration time.Duration) (context.Context, context.CancelFunc) {
		return context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
			envelope.CapabilityMetadataKey, recyclemutationsemantic.CapabilityName,
		)), duration)
	}
	methods := []string{
		runtimev1.RecycleBinMutationService_RestoreRecycleBin_FullMethodName,
		runtimev1.RecycleBinMutationService_PurgeRecycleBin_FullMethodName,
	}

	disabled := mustPolicy(t, config)
	for _, method := range methods {
		ctx, cancel := requestContext(500 * time.Millisecond)
		called := false
		_, err := disabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(context.Context, any) (any, error) {
			called = true
			return nil, nil
		})
		cancel()
		if status.Code(err) != codes.Unimplemented || called {
			t.Fatalf("disabled method=%q error=%v called=%v", method, err, called)
		}
	}

	enabled, err := NewPolicyWithOptions(config, WithRecycleBinMutationMaxAdmissionLease(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	for _, method := range methods {
		ctx, cancel := requestContext(500 * time.Millisecond)
		called := false
		_, err := enabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(handlerCtx context.Context, _ any) (any, error) {
			called = true
			if capability, ok := Capability(handlerCtx); !ok || capability != recyclemutationsemantic.CapabilityName {
				t.Fatalf("method=%q capability=%q ok=%v", method, capability, ok)
			}
			mapped := StatusError(handlerCtx, &recyclemutationsemantic.Failure{Category: recyclemutationsemantic.FailureConflict})
			var conflict *recyclemutationsemantic.Failure
			if !errors.As(mapping.FromRecycleBinMutationStatusError(mapped), &conflict) || conflict.Category != recyclemutationsemantic.FailureConflict {
				t.Fatalf("method=%q conflict mapping=%#v", method, mapped)
			}
			return "ok", nil
		})
		cancel()
		if err != nil || !called {
			t.Fatalf("enabled method=%q error=%v called=%v", method, err, called)
		}
	}

	overLease, cancelOverLease := requestContext(2 * time.Second)
	defer cancelOverLease()
	called := false
	_, err = enabled.UnaryServerInterceptor()(overLease, nil, &grpc.UnaryServerInfo{FullMethod: methods[0]}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	var failure *recyclemutationsemantic.Failure
	if !errors.As(mapping.FromRecycleBinMutationStatusError(err), &failure) || failure.Category != recyclemutationsemantic.FailureResourceExhausted || failure.LimitKind != recyclemutationsemantic.LimitAdmission || called {
		t.Fatalf("over-lease error=%#v called=%v", err, called)
	}
}

func TestRecordViewAndAttachmentSourceRequireDistinctExplicitHostBounds(t *testing.T) {
	config := Config{
		Authority:      func(context.Context, string) (VerifiedCaller, error) { return "caller", nil },
		AdmissionLimit: 1, AttemptCap: time.Second,
	}
	requestContext := func(capability string, duration time.Duration) (context.Context, context.CancelFunc) {
		return context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
			envelope.CapabilityMetadataKey, capability,
		)), duration)
	}
	recordViewMethods := []string{
		runtimev1.RecordViewReadService_ReadRecordView_FullMethodName,
		runtimev1.RecordViewReadService_ResolveRecordAttachmentSources_FullMethodName,
	}
	disabled := mustPolicy(t, config)
	for _, method := range recordViewMethods {
		ctx, cancel := requestContext(recordviewsemantic.CapabilityName, 500*time.Millisecond)
		called := false
		_, err := disabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(context.Context, any) (any, error) {
			called = true
			return nil, nil
		})
		cancel()
		if status.Code(err) != codes.Unimplemented || called {
			t.Fatalf("disabled %s error=%v called=%v", method, err, called)
		}
	}
	ctx, cancel := requestContext(attachmentsemantic.CapabilityName, 500*time.Millisecond)
	called := false
	_, err := disabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	cancel()
	if status.Code(err) != codes.Unimplemented || called {
		t.Fatalf("disabled attachment source error=%v called=%v", err, called)
	}

	enabled, err := NewPolicyWithOptions(config, WithRecordViewMaxAdmissionLease(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	for _, method := range recordViewMethods {
		ctx, cancel := requestContext(recordviewsemantic.CapabilityName, 500*time.Millisecond)
		called := false
		_, err := enabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: method}, func(handlerCtx context.Context, _ any) (any, error) {
			called = true
			if capability, ok := Capability(handlerCtx); !ok || capability != recordviewsemantic.CapabilityName {
				t.Fatalf("capability=%q ok=%v", capability, ok)
			}
			return "ok", nil
		})
		cancel()
		if err != nil || !called {
			t.Fatalf("enabled %s error=%v called=%v", method, err, called)
		}
	}
	ctx, cancel = requestContext(attachmentsemantic.CapabilityName, 500*time.Millisecond)
	called = false
	_, err = enabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	cancel()
	if status.Code(err) != codes.Unimplemented || called {
		t.Fatalf("record-view option enabled attachment source error=%v called=%v", err, called)
	}

	overLease, cancel := requestContext(recordviewsemantic.CapabilityName, 2*time.Second)
	defer cancel()
	called = false
	_, err = enabled.UnaryServerInterceptor()(overLease, nil, &grpc.UnaryServerInfo{FullMethod: recordViewMethods[0]}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	var failure *recordviewsemantic.Failure
	if !errors.As(mapping.FromRecordViewStatusError(err), &failure) || failure.Category != recordviewsemantic.FailureResourceExhausted || failure.LimitKind != recordviewsemantic.LimitAdmission || called {
		t.Fatalf("over-lease error=%#v called=%v", err, called)
	}

	attachmentEnabled, err := NewPolicyWithOptions(config, WithAttachmentSourceMaxAdmissionLease(time.Second))
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel = requestContext(attachmentsemantic.CapabilityName, 500*time.Millisecond)
	called = false
	_, err = attachmentEnabled.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName}, func(handlerCtx context.Context, _ any) (any, error) {
		called = true
		if capability, ok := Capability(handlerCtx); !ok || capability != attachmentsemantic.CapabilityName {
			t.Fatalf("attachment capability=%q ok=%v", capability, ok)
		}
		return "ok", nil
	})
	cancel()
	if err != nil || !called {
		t.Fatalf("enabled attachment source error=%v called=%v", err, called)
	}

	overLease, cancel = requestContext(attachmentsemantic.CapabilityName, 2*time.Second)
	defer cancel()
	called = false
	_, err = attachmentEnabled.UnaryServerInterceptor()(overLease, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName}, func(context.Context, any) (any, error) {
		called = true
		return nil, nil
	})
	var attachmentFailure *attachmentsemantic.Failure
	if !errors.As(mapping.FromRecordAttachmentSourceStatusError(err), &attachmentFailure) || attachmentFailure.Category != attachmentsemantic.FailureResourceExhausted || attachmentFailure.LimitKind != attachmentsemantic.LimitAdmission || called {
		t.Fatalf("attachment over-lease error=%#v called=%v", err, called)
	}
}

func TestIdentityBearerFailsClosedOutsideExchangeAndAfterWorkloadFailure(t *testing.T) {
	for _, test := range []struct {
		name       string
		method     string
		capability string
		authority  Authority
	}{
		{name: "data method", method: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName, capability: semantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "record-access data method", method: runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName, capability: accesssemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "query history data method", method: runtimev1.RecordQueryService_QueryHistory_FullMethodName, capability: querysemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "related-count data method", method: runtimev1.RelatedRecordCountService_CountRelatedRecords_FullMethodName, capability: relatedsemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "recycle-bin data method", method: runtimev1.RecycleBinReadService_ListRecycleBin_FullMethodName, capability: recyclesemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "recycle-bin restore data method", method: runtimev1.RecycleBinMutationService_RestoreRecycleBin_FullMethodName, capability: recyclemutationsemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "recycle-bin purge data method", method: runtimev1.RecycleBinMutationService_PurgeRecycleBin_FullMethodName, capability: recyclemutationsemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "record-view data method", method: runtimev1.RecordViewReadService_ReadRecordView_FullMethodName, capability: recordviewsemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "attachment-source data method", method: runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName, capability: attachmentsemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }},
		{name: "workload rejected", method: runtimev1.IdentityContextService_Exchange_FullMethodName, capability: identitysemantic.CapabilityName,
			authority: func(context.Context, string) (VerifiedCaller, error) {
				return "", &identitysemantic.Failure{Category: identitysemantic.FailureUnauthenticated}
			}},
	} {
		t.Run(test.name, func(t *testing.T) {
			policy, err := NewPolicyWithIdentityAndOptions(Config{
				AdmissionLimit: 1, AttemptCap: time.Second, Authority: test.authority,
			}, time.Second, WithRecordAccessMaxAdmissionLease(time.Second), WithQueryMaxAdmissionLease(time.Second), WithRelatedCountMaxAdmissionLease(time.Second), WithRecycleBinMaxAdmissionLease(time.Second), WithRecycleBinMutationMaxAdmissionLease(time.Second), WithRecordViewMaxAdmissionLease(time.Second), WithAttachmentSourceMaxAdmissionLease(time.Second))
			if err != nil {
				t.Fatal(err)
			}
			ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
				envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0", envelope.CapabilityMetadataKey, test.capability,
				EndUserAuthorizationMetadataKey, "Bearer secret",
			)), time.Second)
			defer cancel()
			called := false
			result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: test.method}, func(context.Context, any) (any, error) {
				called = true
				return "bad", nil
			})
			if result != nil || status.Code(err) != codes.Unauthenticated || called {
				t.Fatalf("result=%#v err=%v called=%v", result, err, called)
			}
		})
	}
}

func TestReservedEndUserCredentialKeyAlwaysFailsClosedOutsideExchange(t *testing.T) {
	for _, test := range []struct {
		name   string
		values []string
	}{
		{name: "malformed", values: []string{"not-a-bearer"}},
		{name: "multiple", values: []string{"Bearer one", "Bearer two"}},
	} {
		t.Run(test.name, func(t *testing.T) {
			policy := mustPolicy(t, Config{AdmissionLimit: 1, AttemptCap: time.Second, Authority: func(context.Context, string) (VerifiedCaller, error) {
				return "caller", nil
			}})
			md := metadata.Pairs(
				envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
				envelope.CapabilityMetadataKey, semantic.CapabilityName,
			)
			md[EndUserAuthorizationMetadataKey] = append([]string(nil), test.values...)
			ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), md), time.Second)
			defer cancel()
			called := false
			result, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{
				FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName,
			}, func(context.Context, any) (any, error) {
				called = true
				return nil, nil
			})
			if result != nil || status.Code(err) != codes.Unauthenticated || called {
				t.Fatalf("result=%#v err=%v called=%v", result, err, called)
			}
		})
	}
}

func TestIdentityTransportFailuresDecodeThroughIdentityContract(t *testing.T) {
	tests := []struct {
		name      string
		ctx       func(*testing.T) context.Context
		authority Authority
		want      identitysemantic.FailureCategory
	}{
		{
			name: "missing bearer",
			ctx:  identityPolicyContext,
			authority: func(context.Context, string) (VerifiedCaller, error) {
				return "workload", nil
			},
			want: identitysemantic.FailureUnauthenticated,
		},
		{
			name: "legacy records authority rejection",
			ctx: func(t *testing.T) context.Context {
				ctx := identityPolicyContext(t)
				md, _ := metadata.FromIncomingContext(ctx)
				md = md.Copy()
				md.Set(EndUserAuthorizationMetadataKey, "Bearer user-secret")
				return metadata.NewIncomingContext(ctx, md)
			},
			authority: func(context.Context, string) (VerifiedCaller, error) {
				return "", &semantic.Failure{Category: semantic.FailureForbidden}
			},
			want: identitysemantic.FailureForbidden,
		},
		{
			name: "incompatible capability",
			ctx: func(t *testing.T) context.Context {
				ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
					envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
					envelope.CapabilityMetadataKey, semantic.CapabilityName,
				)), time.Second)
				t.Cleanup(cancel)
				return ctx
			},
			authority: func(context.Context, string) (VerifiedCaller, error) { return "workload", nil },
			want:      identitysemantic.FailureIncompatible,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			policy := mustIdentityPolicy(t, Config{AdmissionLimit: 1, AttemptCap: time.Second, Authority: test.authority}, time.Second)
			called := false
			result, err := policy.UnaryServerInterceptor()(test.ctx(t), nil, &grpc.UnaryServerInfo{
				FullMethod: runtimev1.IdentityContextService_Exchange_FullMethodName,
			}, func(context.Context, any) (any, error) {
				called = true
				return nil, nil
			})
			var failure *identitysemantic.Failure
			decoded := mapping.FromIdentityStatusError(err)
			if result != nil || called || !errors.As(decoded, &failure) || failure.Category != test.want {
				t.Fatalf("result=%#v called=%v decoded=%#v err=%v", result, called, failure, err)
			}
		})
	}
}

func identityPolicyContext(t *testing.T) context.Context {
	t.Helper()
	ctx, cancel := context.WithTimeout(metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, identitysemantic.CapabilityName,
	)), time.Second)
	t.Cleanup(cancel)
	return ctx
}

func TestMutationRetryInstructionRejectsLegacyProtocolBeforeHandler(t *testing.T) {
	method, ok := registeredMethod(runtimev1.RecordMutationsService_MutateRecords_FullMethodName)
	if !ok {
		t.Fatal("record mutation method is not registered")
	}
	legacy := metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1",
		envelope.ProtocolMinorMetadataKey, "0",
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	))
	current := metadata.NewIncomingContext(context.Background(), metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, "1",
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	))
	if mutationsemantic.ProtocolMinor != 2 || compatible(legacy, method) || !compatible(current, method) {
		t.Fatalf("mutation protocol fence = 1.%d, legacy compatible=%v, current compatible=%v",
			mutationsemantic.ProtocolMinor, compatible(legacy, method), compatible(current, method))
	}
}

func TestParentCancellationAlwaysWinsAfterAuthority(t *testing.T) {
	for range 1000 {
		var cancelParent context.CancelFunc
		policy := mustPolicy(t, Config{
			AdmissionLimit: 1, AttemptCap: time.Second,
			Authority: func(context.Context, string) (VerifiedCaller, error) {
				cancelParent()
				return "", &semantic.Failure{Category: semantic.FailureUnauthenticated}
			},
		})
		deadlineCtx, deadlineCancel := context.WithTimeout(serverProtocolContext(context.Background()), time.Second)
		parentCtx, cancel := context.WithCancel(deadlineCtx)
		cancelParent = cancel
		called := false
		result, err := policy.UnaryServerInterceptor()(parentCtx, nil, &grpc.UnaryServerInfo{
			FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName,
		}, func(context.Context, any) (any, error) {
			called = true
			return nil, nil
		})
		cancel()
		deadlineCancel()
		if result != nil || status.Code(err) != codes.Canceled || called {
			t.Fatalf("result=%#v error=%v handler_called=%v", result, err, called)
		}
	}
}

func TestPolicyEnforcesCompatibilityDeadlineAuthorityAndAttemptCap(t *testing.T) {
	authorityCalls := atomic.Int64{}
	policy := mustPolicy(t, Config{
		AdmissionLimit: 2,
		AttemptCap:     40 * time.Millisecond,
		Authority: AuthorityFunc(func(_ context.Context, method string) (VerifiedCaller, error) {
			authorityCalls.Add(1)
			if method != runtimev1.RecordsService_ReadSelectedRoot_FullMethodName {
				t.Fatalf("method = %q", method)
			}
			return "trusted", nil
		}),
	})
	client, service, cleanup := startServer(t, policy)
	defer cleanup()

	for _, tc := range []struct {
		name string
		ctx  context.Context
		code codes.Code
	}{
		{name: "missing protocol", ctx: context.Background(), code: codes.FailedPrecondition},
		{name: "missing deadline", ctx: withProtocol(context.Background()), code: codes.InvalidArgument},
		{name: "wrong capability", ctx: metadata.NewOutgoingContext(context.Background(), metadata.Pairs(
			envelope.ProtocolMajorMetadataKey, "1", envelope.ProtocolMinorMetadataKey, "0", envelope.CapabilityMetadataKey, "runtime.records.other.v1",
		)), code: codes.FailedPrecondition},
	} {
		t.Run(tc.name, func(t *testing.T) {
			before := service.calls.Load()
			result, err := client.ReadSelectedRoot(tc.ctx, request("I-1"))
			if result != nil || status.Code(err) != tc.code || service.calls.Load() != before {
				t.Fatalf("result=%#v error=%v calls=%d", result, err, service.calls.Load())
			}
		})
	}
	if authorityCalls.Load() != 0 {
		t.Fatalf("invalid calls reached authority: %d", authorityCalls.Load())
	}

	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()
	response, err := client.ReadSelectedRoot(ctx, request("I-1"))
	if err != nil || response.GetRecordId() != "I-1" || !service.trusted.Load() {
		t.Fatalf("response=%#v error=%v trusted=%v", response, err, service.trusted.Load())
	}
	if got := time.Duration(service.deadlineRemaining.Load()); got <= 0 || got > 50*time.Millisecond {
		t.Fatalf("handler deadline remaining = %v, want attempt-capped", got)
	}
	if authorityCalls.Load() != 1 {
		t.Fatalf("authority calls = %d, want 1", authorityCalls.Load())
	}
}

func TestPolicyFailsClosedForAuthorityAndCompression(t *testing.T) {
	policy := mustPolicy(t, Config{
		AdmissionLimit: 1, AttemptCap: time.Second,
		Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) {
			return "", &semantic.Failure{Category: semantic.FailureUnauthenticated}
		}),
	})
	client, service, cleanup := startServer(t, policy)
	defer cleanup()
	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()
	if result, err := client.ReadSelectedRoot(ctx, request("I-1")); result != nil || status.Code(err) != codes.Unauthenticated {
		t.Fatalf("authority result=%#v error=%v", result, err)
	}
	if service.calls.Load() != 0 {
		t.Fatal("unauthenticated request reached handler")
	}
	if result, err := client.ReadSelectedRoot(ctx, request("I-1"), grpc.UseCompressor("gzip")); result != nil || status.Code(err) == codes.OK {
		t.Fatalf("compressed result=%#v error=%v", result, err)
	}
	if service.calls.Load() != 0 {
		t.Fatal("compressed request reached handler")
	}
}

func TestPolicyRejectsZeroCallerUnknownMethodsAndCanceledContexts(t *testing.T) {
	policy := mustPolicy(t, Config{
		AdmissionLimit: 1, AttemptCap: time.Second,
		Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) { return "", nil }),
	})
	interceptor := policy.UnaryServerInterceptor()
	called := atomic.Bool{}
	handler := func(context.Context, any) (any, error) {
		called.Store(true)
		return "unexpected", nil
	}

	deadlineCtx, cancelDeadline := context.WithTimeout(serverProtocolContext(context.Background()), time.Second)
	defer cancelDeadline()
	if result, err := interceptor(deadlineCtx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName}, handler); result != nil || status.Code(err) != codes.Internal {
		t.Fatalf("zero caller result=%#v error=%v", result, err)
	}
	if called.Load() {
		t.Fatal("zero caller reached handler")
	}

	if result, err := interceptor(context.Background(), nil, &grpc.UnaryServerInfo{FullMethod: "/runtime.v1.Admin/Unknown"}, handler); result != nil || status.Code(err) != codes.Unimplemented {
		t.Fatalf("unknown method result=%#v error=%v", result, err)
	}
	if called.Load() {
		t.Fatal("unknown method reached handler")
	}

	canceledCtx, cancel := context.WithCancel(context.Background())
	cancel()
	if result, err := interceptor(canceledCtx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName}, handler); result != nil || status.Code(err) != codes.Canceled {
		t.Fatalf("canceled result=%#v error=%v", result, err)
	}
	if called.Load() {
		t.Fatal("already canceled context reached handler")
	}
}

func TestPolicyRechecksContextAfterAuthority(t *testing.T) {
	for _, tc := range []struct {
		name   string
		cancel bool
		code   codes.Code
	}{
		{name: "deadline", code: codes.DeadlineExceeded},
		{name: "cancel", cancel: true, code: codes.Canceled},
	} {
		t.Run(tc.name, func(t *testing.T) {
			entered := make(chan struct{})
			release := make(chan struct{})
			policy := mustPolicy(t, Config{
				AdmissionLimit: 1, AttemptCap: time.Second,
				Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) {
					close(entered)
					<-release
					return "", &semantic.Failure{Category: semantic.FailureUnauthenticated}
				}),
			})
			base := serverProtocolContext(context.Background())
			deadline := time.Second
			if !tc.cancel {
				deadline = 10 * time.Millisecond
			}
			deadlineCtx, deadlineCancel := context.WithTimeout(base, deadline)
			defer deadlineCancel()
			ctx, cancel := context.WithCancel(deadlineCtx)
			defer cancel()
			called := atomic.Bool{}
			done := make(chan error, 1)
			go func() {
				_, err := policy.UnaryServerInterceptor()(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName}, func(context.Context, any) (any, error) {
					called.Store(true)
					return nil, nil
				})
				done <- err
			}()
			<-entered
			if tc.cancel {
				cancel()
			} else {
				<-ctx.Done()
			}
			close(release)
			if err := <-done; status.Code(err) != tc.code {
				t.Fatalf("error=%v, want %v", err, tc.code)
			}
			if called.Load() {
				t.Fatal("expired context reached handler")
			}
		})
	}
}

func TestPolicyPreventsHandlerSelectedResponseCompression(t *testing.T) {
	policy := mustPolicy(t, Config{
		AdmissionLimit: 1, AttemptCap: time.Second,
		Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }),
	})
	client, service, cleanup := startServer(t, policy)
	defer cleanup()
	service.trySendCompressor = true
	var headers metadata.MD
	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()
	if _, err := client.ReadSelectedRoot(ctx, request("I-1"), grpc.Header(&headers)); err != nil {
		t.Fatal(err)
	}
	if service.sendCompressorSucceeded.Load() {
		t.Fatal("handler selected a response compressor")
	}
	if got := headers.Get("grpc-encoding"); len(got) != 0 {
		t.Fatalf("response grpc-encoding = %v, want identity", got)
	}
}

func TestPolicyAdmissionAndDrainAreFailFast(t *testing.T) {
	release := make(chan struct{})
	entered := make(chan struct{}, 1)
	policy := mustPolicy(t, Config{
		AdmissionLimit: 1, AttemptCap: time.Second,
		Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }),
	})
	client, service, cleanup := startServer(t, policy)
	defer cleanup()
	service.release = release
	service.entered = entered

	firstCtx, firstCancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer firstCancel()
	firstDone := make(chan error, 1)
	go func() {
		_, err := client.ReadSelectedRoot(firstCtx, request("blocking"))
		firstDone <- err
	}()
	select {
	case <-entered:
	case <-time.After(time.Second):
		t.Fatal("first call did not enter")
	}
	if policy.InFlight() != 1 {
		t.Fatalf("in flight = %d", policy.InFlight())
	}
	secondCtx, secondCancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer secondCancel()
	if result, err := client.ReadSelectedRoot(secondCtx, request("I-2")); result != nil || status.Code(err) != codes.ResourceExhausted {
		t.Fatalf("overload result=%#v error=%v", result, err)
	} else {
		var failure *semantic.Failure
		if decoded := mapping.FromStatusError(err); !errors.As(decoded, &failure) ||
			failure.LimitKind != semantic.LimitAdmission || failure.Maximum != 1 ||
			failure.RetryAfterMillis != 50 {
			t.Fatalf("overload detail = %#v", decoded)
		}
	}

	policy.BeginDrain()
	if policy.Ready() {
		t.Fatal("draining policy remained ready")
	}
	close(release)
	if err := <-firstDone; err != nil {
		t.Fatalf("in-flight call failed during drain: %v", err)
	}
	if policy.InFlight() != 0 {
		t.Fatalf("handlers in flight after completion = %d", policy.InFlight())
	}
	if result, err := client.ReadSelectedRoot(secondCtx, request("I-3")); result != nil || status.Code(err) != codes.Unavailable {
		t.Fatalf("draining result=%#v error=%v", result, err)
	}
	snapshot := policy.AdmissionSnapshot()
	if snapshot.Limit != 1 || snapshot.InFlight != 0 || snapshot.Accepted != 1 || snapshot.RejectedFull != 1 || snapshot.RejectedDraining != 1 {
		t.Fatalf("admission snapshot = %+v", snapshot)
	}
}

func TestDrainBarrierCannotRaceAdmission(t *testing.T) {
	policy := mustPolicy(t, Config{
		AdmissionLimit: 64, AttemptCap: time.Second,
		Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }),
	})
	interceptor := policy.UnaryServerInterceptor()
	ctx, cancel := context.WithTimeout(serverProtocolContext(context.Background()), time.Second)
	defer cancel()
	var wg sync.WaitGroup
	for range 64 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_, _ = interceptor(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName}, func(context.Context, any) (any, error) { return nil, nil })
		}()
	}
	policy.BeginDrain()
	for range 64 {
		result, err := interceptor(ctx, nil, &grpc.UnaryServerInfo{FullMethod: runtimev1.RecordsService_ReadSelectedRoot_FullMethodName}, func(context.Context, any) (any, error) {
			t.Fatal("post-drain request reached handler")
			return nil, nil
		})
		if result != nil || status.Code(err) != codes.Unavailable {
			t.Fatalf("post-drain result=%#v error=%v", result, err)
		}
	}
	wg.Wait()
	if policy.InFlight() != 0 {
		t.Fatalf("handlers in flight = %d", policy.InFlight())
	}
}

func TestPolicyEnforcesGrossTransportEnvelope(t *testing.T) {
	policy := mustPolicy(t, Config{
		AdmissionLimit: 1, AttemptCap: time.Second,
		Authority: AuthorityFunc(func(context.Context, string) (VerifiedCaller, error) { return "caller", nil }),
	})
	client, service, cleanup := startServer(t, policy)
	defer cleanup()
	ctx, cancel := context.WithTimeout(withProtocol(context.Background()), time.Second)
	defer cancel()
	result, err := client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: strings.Repeat("X", envelope.MaxRequestWireBytes), RecordId: "I-1", SelectedRootFields: []string{"Subject"},
	})
	if result != nil || status.Code(err) != codes.ResourceExhausted || service.calls.Load() != 0 {
		t.Fatalf("oversize result=%#v error=%v calls=%d", result, err, service.calls.Load())
	}
}

type testService struct {
	runtimev1.UnimplementedRecordsServiceServer
	calls                   atomic.Int64
	trusted                 atomic.Bool
	deadlineRemaining       atomic.Int64
	release                 <-chan struct{}
	entered                 chan<- struct{}
	trySendCompressor       bool
	sendCompressorSucceeded atomic.Bool
}

func (s *testService) ReadSelectedRoot(ctx context.Context, request *runtimev1.ReadSelectedRootRequest) (*runtimev1.ReadSelectedRootResponse, error) {
	s.calls.Add(1)
	if s.trySendCompressor && grpc.SetSendCompressor(ctx, "gzip") == nil {
		s.sendCompressorSucceeded.Store(true)
	}
	caller, ok := Caller(ctx)
	s.trusted.Store(ok && caller == "trusted")
	if deadline, ok := ctx.Deadline(); ok {
		s.deadlineRemaining.Store(int64(time.Until(deadline)))
	}
	if request.GetRecordId() == "blocking" {
		s.entered <- struct{}{}
		select {
		case <-s.release:
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
	value := "value"
	return &runtimev1.ReadSelectedRootResponse{
		RecordId: request.GetRecordId(),
		Fields:   []*runtimev1.SelectedRootField{{Name: "Subject", DeclaredKind: runtimev1.ValueKind_VALUE_KIND_STRING, Value: &runtimev1.SelectedRootField_StringValue{StringValue: value}}},
		Receipts: &commonv1.ConsistencyReceipts{
			ProtocolVersion:   &commonv1.ProtocolVersion{Major: 1},
			CapabilityVersion: &commonv1.CapabilityVersion{Name: semantic.CapabilityName, Major: 1},
		},
	}, nil
}

func mustPolicy(t *testing.T, config Config) *Policy {
	t.Helper()
	policy, err := NewPolicy(config)
	if err != nil {
		t.Fatal(err)
	}
	return policy
}

func mustIdentityPolicy(t *testing.T, config Config, attemptCap time.Duration) *Policy {
	t.Helper()
	policy, err := NewPolicyWithIdentity(config, attemptCap)
	if err != nil {
		t.Fatal(err)
	}
	return policy
}

func startServer(t *testing.T, policy *Policy) (runtimev1.RecordsServiceClient, *testService, func()) {
	t.Helper()
	listener := bufconn.Listen(4 * envelope.MaxResponseWireBytes)
	server := grpc.NewServer(policy.ServerOptions()...)
	service := &testService{}
	runtimev1.RegisterRecordsServiceServer(server, service)
	go func() { _ = server.Serve(listener) }()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	connection, err := grpc.DialContext(ctx, "passthrough:///bufnet",
		grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
			grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
			grpc.WaitForReady(false),
		),
	)
	cancel()
	if err != nil {
		server.Stop()
		_ = listener.Close()
		t.Fatal(err)
	}
	return runtimev1.NewRecordsServiceClient(connection), service, func() {
		_ = connection.Close()
		server.Stop()
		_ = listener.Close()
	}
}

func request(id string) *runtimev1.ReadSelectedRootRequest {
	return &runtimev1.ReadSelectedRootRequest{ObjectId: "Issue", RecordId: id, SelectedRootFields: []string{"Subject"}}
}

func withProtocol(ctx context.Context) context.Context {
	return metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, semantic.CapabilityName,
	)
}

func serverProtocolContext(ctx context.Context) context.Context {
	return metadata.NewIncomingContext(ctx, metadata.Pairs(
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, semantic.CapabilityName,
	))
}