// Package runtimegrpc owns app-blind gRPC host policy for NexusCRM Runtime
// capabilities. It enforces transport, compatibility, deadline, admission,
// authority, and drain invariants before a Runtime handler can execute.
package runtimegrpc

import (
	"context"
	"errors"
	navigationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationnavigationv1"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"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"
	defaultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recorddefaultsv1"
	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"
	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/metadata"
	"google.golang.org/grpc/stats"
	"google.golang.org/grpc/status"
)

// VerifiedCaller is an opaque, authority-issued handle to trusted caller state.
// Its zero value is invalid. Keeping the handle opaque avoids prematurely
// freezing the identity/delegation grammar that the authority implementation
// owns while making nil and typed-nil authority results impossible.
type VerifiedCaller string

// Authority authenticates the transport caller and returns an opaque verified
// caller handle. A function value is intentionally used instead of an interface
// so NewPolicy can reject nil without a typed-nil ambiguity.
type Authority func(context.Context, string) (VerifiedCaller, error)

// AuthorityFunc is retained as a source-compatible name for Authority.
type AuthorityFunc = Authority

type callerContextKey struct{}
type endUserBearerContextKey struct{}
type sessionCreateCredentialsContextKey struct{}
type sessionPresentedCredentialsContextKey struct{}
type authorizationExchangeCredentialsContextKey struct{}
type capabilityContextKey struct{}
type methodContextKey struct{}
type methodClassContextKey struct{}
type resourceClassContextKey struct{}

// MethodClass is the closed transport-authorization shape selected from the
// registered RPC method. It is host policy, never a caller-provided request
// field.
type MethodClass uint8

const (
	MethodClassUnspecified MethodClass = iota
	MethodClassIdentityExchange
	MethodClassDelegatedUserData
	MethodClassWorkloadConsumer
)

// ResourceClass partitions admission on the one private endpoint. It is an
// operational class rather than a new service or protocol capability.
type ResourceClass uint8

const (
	ResourceClassUnspecified ResourceClass = iota
	ResourceClassShortUnary
	ResourceClassHeavyWork
	ResourceClassFileStream
	ResourceClassCatalogWatch
)

// PlatformCapabilityResourceClass returns the transport-owner classification
// for one closed private Platform capability. Hosts use this to configure only
// the isolated classes required by capabilities they actually enable.
func PlatformCapabilityResourceClass(capability string) (ResourceClass, bool) {
	switch capability {
	case catalogsemantic.AppMetadataCapabilityName,
		catalogsemantic.AppMetadataV2CapabilityName,
		catalogsemantic.AppMetadataV3CapabilityName,
		catalogsemantic.IntegrationBindingCapabilityName:
		return ResourceClassCatalogWatch, true
	case approvalsemantic.ExecuteCapabilityName,
		approvalsemantic.ReadCapabilityName,
		approvalnotificationsemantic.CapabilityName,
		auditsemantic.CapabilityName,
		searchsemantic.CapabilityName, flowsemantic.CapabilityName,
		reportsemantic.CapabilityName, permissionsemantic.CapabilityName,
		permissionadminsemantic.CapabilityName,
		permissionauthoringsemantic.CapabilityName,
		layoutauthoringsemantic.CapabilityName,
		layoutauthoringv2semantic.CapabilityName,
		navigationsemantic.CapabilityName,
		analyticsauthoringsemantic.CapabilityName,
		resourceaccesssemantic.CapabilityName,
		listviewsemantic.CapabilityName, metadataadminsemantic.CapabilityName:
		return ResourceClassHeavyWork, true
	case filessemantic.CapabilityName, bulkingestsemantic.CapabilityName:
		return ResourceClassFileStream, true
	case applicationresultsemantic.CapabilityName, commandsemantic.CapabilityName,
		externaleffectsemantic.CapabilityName, mailboxauthorizationsemantic.CapabilityName,
		pagelayoutsemantic.CapabilityName, outboundintentsemantic.CapabilityName:
		return ResourceClassShortUnary, true
	default:
		return ResourceClassUnspecified, false
	}
}

// EndUserAuthorizationMetadataKey is intentionally distinct from gRPC's
// standard authorization key. Authority implementations may already own that
// key for workload authentication; identity exchange must carry both facts
// without overwriting or hiding the workload credential.
const EndUserAuthorizationMetadataKey = "x-nexuscrm-end-user-authorization"

// Caller returns the authority-verified caller fact installed by Policy. The
// private key prevents request metadata or application code from forging it.
func Caller(ctx context.Context) (VerifiedCaller, bool) {
	if ctx == nil {
		return "", false
	}
	caller, ok := ctx.Value(callerContextKey{}).(VerifiedCaller)
	return caller, ok && caller != ""
}

// Capability returns the host-verified Runtime capability for the current
// method. The private context key prevents request metadata or downstream code
// from forging the value after the deny-by-default method gate.
func Capability(ctx context.Context) (string, bool) {
	if ctx == nil {
		return "", false
	}
	capability, ok := ctx.Value(capabilityContextKey{}).(string)
	return capability, ok && capability != ""
}

// RequestMethodClass returns the host-selected authorization class. The
// private key prevents request metadata from choosing a weaker admission path.
func RequestMethodClass(ctx context.Context) (MethodClass, bool) {
	if ctx == nil {
		return MethodClassUnspecified, false
	}
	class, ok := ctx.Value(methodClassContextKey{}).(MethodClass)
	return class, ok && class != MethodClassUnspecified
}

// RequestResourceClass returns the host-selected admission partition.
func RequestResourceClass(ctx context.Context) (ResourceClass, bool) {
	if ctx == nil {
		return ResourceClassUnspecified, false
	}
	class, ok := ctx.Value(resourceClassContextKey{}).(ResourceClass)
	return class, ok && class != ResourceClassUnspecified
}

// StatusError maps an error with the capability grammar already selected by
// Policy. Downstream interceptors use this instead of duplicating the method
// registry or accidentally projecting selected-root details for another RPC.
func StatusError(ctx context.Context, err error) error {
	if ctx == nil {
		return status.Error(codes.Internal, "runtime capability context is unavailable")
	}
	if method, _ := ctx.Value(methodContextKey{}).(string); method != "" {
		switch method {
		case runtimev1.RecordMutationsService_MutateRecords_FullMethodName:
			return mapping.ToRecordMutationStatusError(err)
		case runtimev1.RecordMutationsService_GetMutationStatus_FullMethodName:
			return mapping.ToRecordMutationReconciliationStatusError(err)
		case runtimev1.FederatedLoginTransactionService_BeginFederatedLogin_FullMethodName:
			return mapping.ToFederatedLoginBeginStatusError(err)
		case runtimev1.FederatedLoginTransactionService_ConsumeFederatedLogin_FullMethodName:
			return mapping.ToFederatedLoginConsumeStatusError(err)
		}
	}
	capability, ok := Capability(ctx)
	if !ok {
		return status.Error(codes.Internal, "runtime capability context is unavailable")
	}
	switch capability {
	case semantic.CapabilityName:
		return mapping.ToStatusError(err)
	case accesssemantic.CapabilityName:
		return mapping.ToRecordAccessStatusError(err)
	case defaultsemantic.CapabilityName:
		return mapping.ToCreateDefaultPreviewStatusError(err)
	case querysemantic.CapabilityName:
		return mapping.ToQueryStatusError(err)
	case relatedsemantic.CapabilityName:
		return mapping.ToRelatedCountStatusError(err)
	case recyclesemantic.CapabilityName:
		return mapping.ToRecycleBinStatusError(err)
	case recyclemutationsemantic.CapabilityName:
		return mapping.ToRecycleBinMutationStatusError(err)
	case recordviewsemantic.CapabilityName:
		return mapping.ToRecordViewStatusError(err)
	case attachmentsemantic.CapabilityName:
		return mapping.ToRecordAttachmentSourceStatusError(err)
	case identitysemantic.CapabilityName:
		return mapping.ToIdentityStatusError(err)
	default:
		return mapping.ToPlatformStatusError(err, capability)
	}
}

// EndUserBearer returns the sensitive credential captured only after workload
// authentication for IdentityContextService.Exchange. Policy removes the raw
// metadata before installing this private context value, so ordinary metadata
// consumers, business handlers and observability cannot rediscover it.
func EndUserBearer(ctx context.Context) (string, bool) {
	if ctx == nil {
		return "", false
	}
	bearer, ok := ctx.Value(endUserBearerContextKey{}).(string)
	return bearer, ok && bearer != ""
}

// WithoutEndUserBearer preserves cancellation, deadline, peer identity and
// other trusted context while shadowing the sensitive credential after the
// identity owner has consumed it. Runtime Identity must use the returned
// context for every subsequent authority, repository and signing-provider call.
func WithoutEndUserBearer(ctx context.Context) context.Context {
	if ctx == nil {
		return nil
	}
	return context.WithValue(ctx, endUserBearerContextKey{}, "")
}

// RuntimeSessionCreateCredentials returns the sensitive authentication
// material captured only after workload authentication for CreateSession.
// Policy has already removed every raw value from incoming metadata.
func RuntimeSessionCreateCredentials(ctx context.Context) (SessionCreateCredentials, bool) {
	if ctx == nil {
		return SessionCreateCredentials{}, false
	}
	credentials, ok := ctx.Value(sessionCreateCredentialsContextKey{}).(SessionCreateCredentials)
	return credentials, ok && credentials.Secret != ""
}

// RuntimeSessionPresentedCredentials returns the current public bearer and
// optional browser control proof captured for ReadSession or RevokeSession.
func RuntimeSessionPresentedCredentials(ctx context.Context) (SessionPresentedCredentials, bool) {
	if ctx == nil {
		return SessionPresentedCredentials{}, false
	}
	credentials, ok := ctx.Value(sessionPresentedCredentialsContextKey{}).(SessionPresentedCredentials)
	return credentials, ok && credentials.AccessToken != ""
}

// RuntimeAuthorizationExchangeCredentials returns token-endpoint secrets only
// after workload authentication and method selection have completed.
func RuntimeAuthorizationExchangeCredentials(ctx context.Context) (AuthorizationExchangeCredentials, bool) {
	if ctx == nil {
		return AuthorizationExchangeCredentials{}, false
	}
	credentials, ok := ctx.Value(authorizationExchangeCredentialsContextKey{}).(AuthorizationExchangeCredentials)
	return credentials, ok && (credentials.ClientSecret != "" || credentials.AuthorizationCode != "" || credentials.RefreshToken != "")
}

// WithoutRuntimeSessionCredentials shadows every private credential context
// value after the Runtime identity owner has consumed it.
func WithoutRuntimeSessionCredentials(ctx context.Context) context.Context {
	if ctx == nil {
		return nil
	}
	ctx = context.WithValue(ctx, sessionCreateCredentialsContextKey{}, SessionCreateCredentials{})
	ctx = context.WithValue(ctx, sessionPresentedCredentialsContextKey{}, SessionPresentedCredentials{})
	ctx = context.WithValue(ctx, authorizationExchangeCredentialsContextKey{}, AuthorizationExchangeCredentials{})
	return context.WithValue(ctx, endUserBearerContextKey{}, "")
}

// Config declares the released host-owned selected-root policy. Its field
// shape is preserved for source compatibility; identity enablement uses the
// separate NewPolicyWithIdentity constructor.
type Config struct {
	Authority      Authority
	AdmissionLimit int
	AttemptCap     time.Duration
}

// PolicyOption enables one optional Runtime method with its own explicit
// host-owned availability bound. The sealed interface keeps option validation
// in this package while preserving the released Config field shape.
type PolicyOption interface {
	apply(*policyOptions) error
}

type policyOptionFunc func(*policyOptions) error

func (option policyOptionFunc) apply(config *policyOptions) error { return option(config) }

type policyOptions struct {
	recordAccessMaxAdmissionLease       time.Duration
	createDefaultMaxAdmissionLease      time.Duration
	queryMaxAdmissionLease              time.Duration
	relatedCountMaxAdmissionLease       time.Duration
	recycleBinMaxAdmissionLease         time.Duration
	recycleBinMutationMaxAdmissionLease time.Duration
	recordViewMaxAdmissionLease         time.Duration
	attachmentSourceMaxAdmissionLease   time.Duration
	recordMutationAttempt               time.Duration
	identityAttempt                     time.Duration
	platformMaxAdmissionLease           map[string]time.Duration
	resourceAdmissionLimits             map[ResourceClass]int
}

// WithRecordAccessMaxAdmissionLease enables
// RecordsService.EvaluateRecordAccess. The supplied value is the longest
// caller-owned deadline the host will admit, not a semantic deadline or a
// production-approved SLO. Admitted calls retain their original context;
// longer leases fail fast as admission load shedding.
func WithRecordAccessMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: record-access maximum admission lease must be positive")
		}
		if config.recordAccessMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: record-access policy option is duplicated")
		}
		config.recordAccessMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithCreateDefaultPreviewMaxAdmissionLease enables one bounded, Runtime-owned
// create-default preview. The supplied value is host availability policy and
// does not become a semantic deadline or product SLO.
func WithCreateDefaultPreviewMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: create-default preview maximum admission lease must be positive")
		}
		if config.createDefaultMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: create-default preview policy option is duplicated")
		}
		config.createDefaultMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithQueryMaxAdmissionLease enables the bounded RecordQueryService family with a
// caller-owned maximum lease. It is host availability policy, not a semantic
// deadline or production SLO.
func WithQueryMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: query maximum admission lease must be positive")
		}
		if config.queryMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: query policy option is duplicated")
		}
		config.queryMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithRelatedCountMaxAdmissionLease enables the bounded related-record count
// batch with a caller-owned maximum lease. It is host availability policy, not
// a semantic deadline or production SLO.
func WithRelatedCountMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: related-count maximum admission lease must be positive")
		}
		if config.relatedCountMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: related-count policy option is duplicated")
		}
		config.relatedCountMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithRecycleBinMaxAdmissionLease enables one bounded deleted-record listing
// with a caller-owned maximum lease. It is host availability policy, not a
// semantic deadline or production SLO.
func WithRecycleBinMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: recycle-bin maximum admission lease must be positive")
		}
		if config.recycleBinMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: recycle-bin policy option is duplicated")
		}
		config.recycleBinMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithRecycleBinMutationMaxAdmissionLease enables restore and permanent purge
// with one caller-owned maximum lease. Both methods share one neutral mutation
// capability and one heavy-work admission class.
func WithRecycleBinMutationMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: recycle-bin mutation maximum admission lease must be positive")
		}
		if config.recycleBinMutationMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: recycle-bin mutation policy option is duplicated")
		}
		config.recycleBinMutationMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithRecordViewMaxAdmissionLease enables the two bounded record-view read
// methods with one caller-owned maximum lease. Both methods belong to the same
// neutral capability and share one explicit host availability policy.
func WithRecordViewMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: record-view maximum admission lease must be positive")
		}
		if config.recordViewMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: record-view policy option is duplicated")
		}
		config.recordViewMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithAttachmentSourceMaxAdmissionLease enables the proof-bearing attachment
// source resolver. It is deliberately independent from record-view: the two
// RPCs have distinct trusted capabilities and receipt authority even though
// Catalog projects their shared metadata closure from one snapshot.
func WithAttachmentSourceMaxAdmissionLease(maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: attachment-source maximum admission lease must be positive")
		}
		if config.attachmentSourceMaxAdmissionLease > 0 {
			return errors.New("runtimegrpc: attachment-source policy option is duplicated")
		}
		config.attachmentSourceMaxAdmissionLease = maxAdmissionLease
		return nil
	})
}

// WithRecordMutationAttemptCap enables both mutation methods with one explicit
// availability bound. Unary reconciliation receives a host-owned child
// deadline. Client-streaming mutation must instead arrive with a caller-owned
// deadline no longer than this bound: gRPC RecvMsg is owned by the transport's
// original context and cannot be interrupted by merely replacing the handler
// context. MutateRecords and GetMutationStatus share the same trusted
// capability but retain distinct disclosure grammars.
func WithRecordMutationAttemptCap(attemptCap time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if attemptCap <= 0 {
			return errors.New("runtimegrpc: record-mutation attempt cap must be positive")
		}
		if config.recordMutationAttempt > 0 {
			return errors.New("runtimegrpc: record-mutation policy option is duplicated")
		}
		config.recordMutationAttempt = attemptCap
		return nil
	})
}

// WithPlatformCapabilityMaxAdmissionLease explicitly enables one additional
// neutral Platform API family. The capability must be one of the released
// private v1 identities; callers retain their original cancellation and
// deadline, which may not exceed this operational ceiling.
func WithPlatformCapabilityMaxAdmissionLease(capability string, maxAdmissionLease time.Duration) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		if _, ok := platformMethodStatus(capability); !ok {
			return errors.New("runtimegrpc: unknown Platform API capability")
		}
		if maxAdmissionLease <= 0 {
			return errors.New("runtimegrpc: Platform API maximum admission lease must be positive")
		}
		if config.platformMaxAdmissionLease == nil {
			config.platformMaxAdmissionLease = make(map[string]time.Duration)
		}
		if _, duplicate := config.platformMaxAdmissionLease[capability]; duplicate {
			return errors.New("runtimegrpc: Platform API policy option is duplicated")
		}
		config.platformMaxAdmissionLease[capability] = maxAdmissionLease
		return nil
	})
}

// WithResourceClassAdmissionLimit gives a long-lived or expensive method class
// a stricter bound inside Config.AdmissionLimit. The process-wide limit remains
// authoritative, so isolating classes never expands total concurrency.
// Previously released methods share that global bound; newly enabled Platform
// methods fail construction unless their non-short resource class is isolated.
func WithResourceClassAdmissionLimit(class ResourceClass, limit int) PolicyOption {
	return policyOptionFunc(func(config *policyOptions) error {
		switch class {
		case ResourceClassHeavyWork, ResourceClassFileStream, ResourceClassCatalogWatch:
		default:
			return errors.New("runtimegrpc: resource class admission limit is not configurable")
		}
		if limit <= 0 {
			return errors.New("runtimegrpc: resource class admission limit must be positive")
		}
		if config.resourceAdmissionLimits == nil {
			config.resourceAdmissionLimits = make(map[ResourceClass]int)
		}
		if _, duplicate := config.resourceAdmissionLimits[class]; duplicate {
			return errors.New("runtimegrpc: resource class admission limit is duplicated")
		}
		config.resourceAdmissionLimits[class] = limit
		return nil
	})
}

// Policy is the authoritative pre-handler transport gate for explicitly
// registered Runtime methods.
type Policy struct {
	authority                           Authority
	attempt                             time.Duration
	recordAccessMaxAdmissionLease       time.Duration
	createDefaultMaxAdmissionLease      time.Duration
	queryMaxAdmissionLease              time.Duration
	relatedCountMaxAdmissionLease       time.Duration
	recycleBinMaxAdmissionLease         time.Duration
	recycleBinMutationMaxAdmissionLease time.Duration
	recordViewMaxAdmissionLease         time.Duration
	attachmentSourceMaxAdmissionLease   time.Duration
	recordMutationAttempt               time.Duration
	identityAttempt                     time.Duration
	platformMaxAdmissionLease           map[string]time.Duration
	mu                                  sync.Mutex
	draining                            bool
	globalAdmission                     *admissionState
	admission                           map[ResourceClass]*admissionState
}

type admissionState struct {
	limit            int
	inFlight         int
	accepted         uint64
	rejectedFull     uint64
	rejectedDraining uint64
}

// AdmissionSnapshot is the released process-wide admission view. Counters are
// process-local telemetry, not protocol facts. Resource-class isolation never
// expands this global bound.
type AdmissionSnapshot struct {
	Limit            int
	InFlight         int
	Accepted         uint64
	RejectedFull     uint64
	RejectedDraining uint64
}

// ResourceAdmissionSnapshot reports one effective pool. When a class was not
// independently configured, it returns the shared short-unary pool snapshot.
func (p *Policy) ResourceAdmissionSnapshot(class ResourceClass) AdmissionSnapshot {
	if p == nil {
		return AdmissionSnapshot{}
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	state := p.admissionState(class)
	if state == nil {
		return AdmissionSnapshot{}
	}
	return admissionSnapshot(state)
}

// NewPolicy validates configuration and creates a fail-fast admission gate.
func NewPolicy(config Config) (*Policy, error) {
	return newPolicy(config, policyOptions{})
}

// NewPolicyWithOptions enables explicitly bounded optional Runtime methods
// without changing the released Config shape.
func NewPolicyWithOptions(config Config, options ...PolicyOption) (*Policy, error) {
	resolved := policyOptions{}
	for _, option := range options {
		if option == nil {
			return nil, errors.New("runtimegrpc: policy option is required")
		}
		if err := option.apply(&resolved); err != nil {
			return nil, err
		}
	}
	return newPolicy(config, resolved)
}

// NewPolicyWithIdentity enables IdentityContext.Exchange with an explicit
// host-owned availability bound. It deliberately does not inherit selected-
// root's two-second semantic cap without identity-specific evidence.
func NewPolicyWithIdentity(config Config, identityAttemptCap time.Duration) (*Policy, error) {
	if identityAttemptCap <= 0 {
		return nil, errors.New("runtimegrpc: identity attempt cap must be positive")
	}
	return newPolicy(config, policyOptions{identityAttempt: identityAttemptCap})
}

// NewPolicyWithIdentityAndOptions preserves the released identity constructor
// while allowing the same host to enable other explicitly bounded methods.
func NewPolicyWithIdentityAndOptions(config Config, identityAttemptCap time.Duration, options ...PolicyOption) (*Policy, error) {
	if identityAttemptCap <= 0 {
		return nil, errors.New("runtimegrpc: identity attempt cap must be positive")
	}
	resolved := policyOptions{identityAttempt: identityAttemptCap}
	for _, option := range options {
		if option == nil {
			return nil, errors.New("runtimegrpc: policy option is required")
		}
		if err := option.apply(&resolved); err != nil {
			return nil, err
		}
	}
	return newPolicy(config, resolved)
}

func newPolicy(config Config, options policyOptions) (*Policy, error) {
	if config.Authority == nil {
		return nil, errors.New("runtimegrpc: authority is required")
	}
	if config.AdmissionLimit <= 0 {
		return nil, errors.New("runtimegrpc: admission limit must be positive")
	}
	if config.AttemptCap <= 0 || config.AttemptCap > semantic.MaxAttemptDuration {
		return nil, errors.New("runtimegrpc: attempt cap must be positive and no greater than the semantic maximum")
	}
	if err := validatePlatformResourceIsolation(options); err != nil {
		return nil, err
	}
	globalAdmission := &admissionState{limit: config.AdmissionLimit}
	shortAdmission := &admissionState{limit: config.AdmissionLimit}
	admission := map[ResourceClass]*admissionState{
		ResourceClassShortUnary:   shortAdmission,
		ResourceClassHeavyWork:    shortAdmission,
		ResourceClassFileStream:   shortAdmission,
		ResourceClassCatalogWatch: shortAdmission,
	}
	for class, limit := range options.resourceAdmissionLimits {
		if limit > config.AdmissionLimit {
			return nil, errors.New("runtimegrpc: resource-class admission limit cannot exceed the process admission limit")
		}
		admission[class] = &admissionState{limit: limit}
	}
	return &Policy{
		authority:                           config.Authority,
		attempt:                             config.AttemptCap,
		recordAccessMaxAdmissionLease:       options.recordAccessMaxAdmissionLease,
		createDefaultMaxAdmissionLease:      options.createDefaultMaxAdmissionLease,
		queryMaxAdmissionLease:              options.queryMaxAdmissionLease,
		relatedCountMaxAdmissionLease:       options.relatedCountMaxAdmissionLease,
		recycleBinMaxAdmissionLease:         options.recycleBinMaxAdmissionLease,
		recycleBinMutationMaxAdmissionLease: options.recycleBinMutationMaxAdmissionLease,
		recordViewMaxAdmissionLease:         options.recordViewMaxAdmissionLease,
		attachmentSourceMaxAdmissionLease:   options.attachmentSourceMaxAdmissionLease,
		recordMutationAttempt:               options.recordMutationAttempt,
		identityAttempt:                     options.identityAttempt,
		platformMaxAdmissionLease:           cloneAdmissionLeases(options.platformMaxAdmissionLease),
		globalAdmission:                     globalAdmission,
		admission:                           admission,
	}, nil
}

func validatePlatformResourceIsolation(options policyOptions) error {
	for capability := range options.platformMaxAdmissionLease {
		class, ok := PlatformCapabilityResourceClass(capability)
		if !ok {
			return errors.New("runtimegrpc: enabled Platform capability has no resource classification")
		}
		if class != ResourceClassShortUnary && options.resourceAdmissionLimits[class] <= 0 {
			return errors.New("runtimegrpc: enabled Platform capability requires an isolated resource class")
		}
	}
	return nil
}

// BeginDrain rejects new Runtime handlers. Complete RPC shutdown, including
// response marshal and transport send, belongs to grpc.Server.GracefulStop.
func (p *Policy) BeginDrain() {
	if p == nil {
		return
	}
	p.mu.Lock()
	p.draining = true
	p.mu.Unlock()
}

// Ready reports whether the policy is accepting new Runtime work.
func (p *Policy) Ready() bool {
	if p == nil {
		return false
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	return !p.draining
}

// InFlight reports the number of admitted calls for readiness/drain tests and
// bounded operational telemetry.
func (p *Policy) InFlight() int {
	if p == nil {
		return 0
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	if p.globalAdmission == nil {
		return 0
	}
	return p.globalAdmission.inFlight
}

func (p *Policy) AdmissionSnapshot() AdmissionSnapshot {
	if p == nil {
		return AdmissionSnapshot{}
	}
	p.mu.Lock()
	defer p.mu.Unlock()
	return admissionSnapshot(p.globalAdmission)
}

func admissionSnapshot(state *admissionState) AdmissionSnapshot {
	if state == nil {
		return AdmissionSnapshot{}
	}
	return AdmissionSnapshot{
		Limit: state.limit, InFlight: state.inFlight, Accepted: state.accepted,
		RejectedFull: state.rejectedFull, RejectedDraining: state.rejectedDraining,
	}
}

// ServerOptions returns the complete Runtime transport envelope. A
// host must use these options rather than recreating individual values.
func (p *Policy) ServerOptions() []grpc.ServerOption {
	return []grpc.ServerOption{
		grpc.MaxRecvMsgSize(envelope.MaxRequestWireBytes),
		grpc.MaxSendMsgSize(envelope.MaxResponseWireBytes),
		grpc.StatsHandler(compressionStatsHandler{}),
		grpc.ChainUnaryInterceptor(p.UnaryServerInterceptor()),
		grpc.ChainStreamInterceptor(p.StreamServerInterceptor()),
	}
}

// UnaryServerInterceptor enforces one deny-by-default ordered Runtime path.
func (p *Policy) UnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		if info == nil {
			return nil, status.Error(codes.Unimplemented, "runtime method is not registered by host policy")
		}
		handlerCtx, release, err := p.prepare(ctx, info.FullMethod)
		if err != nil {
			return nil, err
		}
		defer release()
		return handler(handlerCtx, request)
	}
}

// StreamServerInterceptor applies the same trusted admission state machine to
// client-streaming Runtime capabilities. The admission lease spans the whole
// handler stream; grpc.Server.GracefulStop remains the transport-completion
// owner after the handler returns.
func (p *Policy) StreamServerInterceptor() grpc.StreamServerInterceptor {
	return func(server any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if stream == nil || info == nil {
			return status.Error(codes.Unimplemented, "runtime method is not registered by host policy")
		}
		handlerCtx, release, err := p.prepare(stream.Context(), info.FullMethod)
		if err != nil {
			return err
		}
		defer release()
		return handler(server, &contextServerStream{ServerStream: stream, ctx: handlerCtx})
	}
}

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

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

func (p *Policy) prepare(ctx context.Context, fullMethod string) (context.Context, func(), error) {
	method, ok := registeredMethod(fullMethod)
	if !ok {
		return nil, nil, status.Error(codes.Unimplemented, "runtime method is not registered by host policy")
	}
	if err := ctx.Err(); err != nil {
		return nil, nil, method.statusError(err)
	}
	if !compatible(ctx, method) || compressionUsed(ctx) {
		return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureIncompatible})
	}
	if _, ok := ctx.Deadline(); !ok {
		return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureInvalidRequest, InvalidComponent: commonsemantic.InvalidComponentDeadline})
	}
	if p == nil {
		return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureUnavailable})
	}
	attemptCap := time.Duration(0)
	switch method.attemptPolicy {
	case attemptSelectedRoot:
		attemptCap = p.attempt
	case attemptRecordAccess:
		if p.recordAccessMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.recordAccessMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptCreateDefaultPreview:
		if p.createDefaultMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.createDefaultMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptQuery:
		if p.queryMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.queryMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptRelatedCount:
		if p.relatedCountMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.relatedCountMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptRecycleBin:
		if p.recycleBinMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.recycleBinMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptRecycleBinMutation:
		if p.recycleBinMutationMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.recycleBinMutationMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptRecordView:
		if p.recordViewMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.recordViewMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptAttachmentSource:
		if p.attachmentSourceMaxAdmissionLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > p.attachmentSourceMaxAdmissionLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
	case attemptMutation:
		if p.recordMutationAttempt <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		if method.callerDeadlineOwnsAttempt {
			deadline, _ := ctx.Deadline()
			if time.Until(deadline) > p.recordMutationAttempt {
				return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
			}
		} else {
			attemptCap = p.recordMutationAttempt
		}
	case attemptIdentity:
		if p.identityAttempt <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		attemptCap = p.identityAttempt
	case attemptPlatform:
		maxLease := p.platformMaxAdmissionLease[method.capability]
		if maxLease <= 0 {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method is not enabled by host policy")
		}
		deadline, _ := ctx.Deadline()
		if time.Until(deadline) > maxLease {
			return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission})
		}
		// A capability may add a long-lived method after shipping unary RPCs.
		// Enabling that existing capability must remain source-compatible, but
		// the new method must fail closed until its isolated pool is explicit.
		if method.resourceClass != ResourceClassShortUnary &&
			p.admission[method.resourceClass] == p.admission[ResourceClassShortUnary] {
			return nil, nil, status.Error(codes.Unimplemented, "runtime method resource class is not enabled by host policy")
		}
	}
	admissionResult, admission, admissionMaximum := p.admit(method.resourceClass)
	switch admissionResult {
	case admissionDraining:
		return nil, nil, method.statusError(&commonsemantic.Failure{Category: commonsemantic.FailureUnavailable})
	case admissionFull:
		return nil, nil, method.statusError(&commonsemantic.Failure{
			Category: commonsemantic.FailureResourceExhausted, LimitKind: commonsemantic.LimitAdmission,
			Maximum: uint64(admissionMaximum), RetryAfterMillis: 50,
		})
	}
	releaseOnFailure := true
	cancelOnFailure := func() {}
	defer func() {
		// prepare may unwind through an injected authority panic. Once admit has
		// succeeded, this defer is the only owner capable of preventing a
		// permanent admission-slot leak before a release closure is returned.
		if releaseOnFailure {
			cancelOnFailure()
			p.release(admission)
		}
	}()
	abort := func(err error) (context.Context, func(), error) {
		cancelOnFailure()
		p.release(admission)
		releaseOnFailure = false
		return nil, nil, method.statusError(err)
	}
	attemptCtx := ctx
	cancel := func() {}
	if attemptCap > 0 {
		attemptCtx, cancel = context.WithTimeout(ctx, attemptCap)
		cancelOnFailure = cancel
	}
	release := func() { cancel(); p.release(admission) }
	authorityCtx := trustedMethodContext(authorityInputContext(attemptCtx, method.methodClass), fullMethod, method)
	caller, err := p.authority(authorityCtx, fullMethod)
	if contextErr := requestContextError(ctx, attemptCtx); contextErr != nil {
		cancel()
		return abort(contextErr)
	}
	if err != nil {
		cancel()
		return abort(err)
	}
	if caller == "" {
		cancel()
		return abort(&commonsemantic.Failure{Category: commonsemantic.FailureInternal})
	}
	handlerCtx := context.WithValue(trustedMethodContext(authorityInputContext(attemptCtx, method.methodClass), fullMethod, method), callerContextKey{}, caller)
	bearer, bearerPresent := incomingEndUserBearer(attemptCtx)
	bearerMetadataPresent := incomingEndUserAuthorizationPresent(attemptCtx)
	createCredentials, createCredentialsPresent := incomingSessionCreateCredentials(attemptCtx)
	exchangeCredentials, exchangeCredentialsPresent := incomingAuthorizationExchangeCredentials(attemptCtx)
	exchangeCredentialMetadataPresent := incomingAuthorizationExchangeCredentialMetadataPresent(attemptCtx)
	createCredentialMetadataPresent := incomingSessionCreateCredentialMetadataPresent(attemptCtx)
	controlProof, controlProofPresent, controlProofValid := incomingSessionControlProof(attemptCtx)
	issuedCredentialMetadataPresent := incomingSessionIssuedCredentialMetadataPresent(attemptCtx)
	credentialHeaderPresent := incomingSensitiveSessionMetadataPresent(attemptCtx)
	delegationHeaderPresent := incomingRuntimeDelegationPresent(attemptCtx)
	if method.sensitiveBearer {
		if !bearerPresent || createCredentialMetadataPresent || exchangeCredentialMetadataPresent || controlProofPresent || issuedCredentialMetadataPresent {
			cancel()
			return abort(&commonsemantic.Failure{Category: commonsemantic.FailureUnauthenticated})
		}
		handlerCtx = context.WithValue(handlerCtx, endUserBearerContextKey{}, bearer)
	} else if method.sensitiveSessionCreate {
		if !createCredentialsPresent || bearerMetadataPresent || exchangeCredentialMetadataPresent || controlProofPresent || issuedCredentialMetadataPresent {
			cancel()
			return abort(&commonsemantic.Failure{Category: commonsemantic.FailureUnauthenticated})
		}
		handlerCtx = context.WithValue(handlerCtx, sessionCreateCredentialsContextKey{}, createCredentials)
	} else if method.sensitiveSessionPresented {
		if !bearerPresent || !controlProofValid || createCredentialMetadataPresent || exchangeCredentialMetadataPresent || issuedCredentialMetadataPresent {
			cancel()
			return abort(&commonsemantic.Failure{Category: commonsemantic.FailureUnauthenticated})
		}
		handlerCtx = context.WithValue(handlerCtx, sessionPresentedCredentialsContextKey{}, SessionPresentedCredentials{
			AccessToken: bearer, ControlProof: controlProof,
		})
	} else if method.sensitiveAuthorizationExchange {
		if !exchangeCredentialsPresent || bearerMetadataPresent || createCredentialMetadataPresent || controlProofPresent || issuedCredentialMetadataPresent {
			cancel()
			return abort(&commonsemantic.Failure{Category: commonsemantic.FailureUnauthenticated})
		}
		handlerCtx = context.WithValue(handlerCtx, authorizationExchangeCredentialsContextKey{}, exchangeCredentials)
	} else if credentialHeaderPresent {
		cancel()
		return abort(&commonsemantic.Failure{Category: commonsemantic.FailureUnauthenticated})
	}
	if method.methodClass != MethodClassDelegatedUserData && delegationHeaderPresent {
		cancel()
		return abort(&commonsemantic.Failure{Category: commonsemantic.FailureUnauthenticated})
	}
	if stream := grpc.ServerTransportStreamFromContext(handlerCtx); stream != nil {
		handlerCtx = grpc.NewContextWithServerTransportStream(handlerCtx, identityOnlyStream{ServerTransportStream: stream})
	}
	if err := requestContextError(ctx, handlerCtx); err != nil {
		cancel()
		return abort(err)
	}
	releaseOnFailure = false
	return handlerCtx, release, nil
}

type methodPolicy struct {
	capability                     string
	protocolMajor                  uint32
	protocolMinor                  uint32
	methodClass                    MethodClass
	resourceClass                  ResourceClass
	attemptPolicy                  attemptPolicy
	callerDeadlineOwnsAttempt      bool
	sensitiveBearer                bool
	sensitiveSessionCreate         bool
	sensitiveSessionPresented      bool
	sensitiveAuthorizationExchange bool
	statusError                    func(error) error
}

func trustedMethodContext(ctx context.Context, fullMethod string, method methodPolicy) context.Context {
	ctx = context.WithValue(ctx, capabilityContextKey{}, method.capability)
	ctx = context.WithValue(ctx, methodContextKey{}, fullMethod)
	ctx = context.WithValue(ctx, methodClassContextKey{}, method.methodClass)
	return context.WithValue(ctx, resourceClassContextKey{}, method.resourceClass)
}

type attemptPolicy uint8

const (
	attemptSelectedRoot attemptPolicy = iota
	attemptRecordAccess
	attemptCreateDefaultPreview
	attemptQuery
	attemptRelatedCount
	attemptRecycleBin
	attemptRecycleBinMutation
	attemptRecordView
	attemptAttachmentSource
	attemptMutation
	attemptIdentity
	attemptPlatform
)

func registeredMethod(fullMethod string) (methodPolicy, bool) {
	switch fullMethod {
	case runtimev1.RecordsService_ReadSelectedRoot_FullMethodName:
		return delegatedMethod(semantic.CapabilityName, semantic.ProtocolMajor, semantic.ProtocolMinor, attemptSelectedRoot, ResourceClassShortUnary, mapping.ToStatusError), true
	case runtimev1.RecordsService_EvaluateRecordAccess_FullMethodName:
		return delegatedMethod(accesssemantic.CapabilityName, accesssemantic.ProtocolMajor, accesssemantic.ProtocolMinor, attemptRecordAccess, ResourceClassShortUnary, mapping.ToRecordAccessStatusError), true
	case runtimev1.RecordCreateDefaultPreviewService_PreviewCreateDefaults_FullMethodName:
		return delegatedMethod(defaultsemantic.CapabilityName, defaultsemantic.ProtocolMajor, defaultsemantic.ProtocolMinor, attemptCreateDefaultPreview, ResourceClassHeavyWork, mapping.ToCreateDefaultPreviewStatusError), true
	case runtimev1.RecordQueryService_QueryRecords_FullMethodName:
		return delegatedMethod(querysemantic.CapabilityName, querysemantic.ProtocolMajor, querysemantic.ProtocolMinor, attemptQuery, ResourceClassHeavyWork, mapping.ToQueryStatusError), true
	case runtimev1.RecordQueryService_QueryHistory_FullMethodName:
		return delegatedMethod(querysemantic.CapabilityName, querysemantic.ProtocolMajor, querysemantic.ProtocolMinor, attemptQuery, ResourceClassHeavyWork, mapping.ToQueryStatusError), true
	case runtimev1.RelatedRecordCountService_CountRelatedRecords_FullMethodName:
		return delegatedMethod(relatedsemantic.CapabilityName, relatedsemantic.ProtocolMajor, relatedsemantic.ProtocolMinor, attemptRelatedCount, ResourceClassHeavyWork, mapping.ToRelatedCountStatusError), true
	case runtimev1.RecycleBinReadService_ListRecycleBin_FullMethodName:
		return delegatedMethod(recyclesemantic.CapabilityName, recyclesemantic.ProtocolMajor, recyclesemantic.ProtocolMinor, attemptRecycleBin, ResourceClassHeavyWork, mapping.ToRecycleBinStatusError), true
	case runtimev1.RecycleBinMutationService_RestoreRecycleBin_FullMethodName,
		runtimev1.RecycleBinMutationService_PurgeRecycleBin_FullMethodName:
		return delegatedMethod(recyclemutationsemantic.CapabilityName, recyclemutationsemantic.ProtocolMajor, recyclemutationsemantic.ProtocolMinor, attemptRecycleBinMutation, ResourceClassHeavyWork, mapping.ToRecycleBinMutationStatusError), true
	case runtimev1.RecordViewReadService_ReadRecordView_FullMethodName,
		runtimev1.RecordViewReadService_ResolveRecordAttachmentSources_FullMethodName:
		return delegatedMethod(recordviewsemantic.CapabilityName, recordviewsemantic.ProtocolMajor, recordviewsemantic.ProtocolMinor, attemptRecordView, ResourceClassShortUnary, mapping.ToRecordViewStatusError), true
	case runtimev1.RecordAttachmentSourceResolutionService_ResolveAttachmentSources_FullMethodName:
		return delegatedMethod(attachmentsemantic.CapabilityName, attachmentsemantic.ProtocolMajor, attachmentsemantic.ProtocolMinor, attemptAttachmentSource, ResourceClassShortUnary, mapping.ToRecordAttachmentSourceStatusError), true
	case runtimev1.RecordMutationsService_MutateRecords_FullMethodName:
		result := delegatedMethod(mutationsemantic.CapabilityName, mutationsemantic.ProtocolMajor, mutationsemantic.ProtocolMinor, attemptMutation, ResourceClassHeavyWork, mapping.ToRecordMutationStatusError)
		result.callerDeadlineOwnsAttempt = true
		return result, true
	case runtimev1.RecordMutationsService_GetMutationStatus_FullMethodName:
		return delegatedMethod(mutationsemantic.CapabilityName, mutationsemantic.ProtocolMajor, mutationsemantic.ProtocolMinor, attemptMutation, ResourceClassShortUnary, mapping.ToRecordMutationReconciliationStatusError), true
	case runtimev1.IdentityContextService_Exchange_FullMethodName:
		return methodPolicy{
			capability: identitysemantic.CapabilityName, protocolMajor: identitysemantic.ProtocolMajor,
			protocolMinor: identitysemantic.ProtocolMinor, methodClass: MethodClassIdentityExchange,
			resourceClass: ResourceClassShortUnary, attemptPolicy: attemptIdentity,
			sensitiveBearer: true, statusError: mapping.ToIdentityStatusError,
		}, true
	case runtimev1.SessionLifecycleService_CreateSession_FullMethodName:
		return methodPolicy{
			capability: sessionsemantic.CapabilityName, protocolMajor: sessionsemantic.ProtocolMajor,
			protocolMinor: sessionsemantic.ProtocolMinor, methodClass: MethodClassWorkloadConsumer,
			resourceClass: ResourceClassShortUnary, attemptPolicy: attemptIdentity,
			sensitiveSessionCreate: true,
			statusError:            func(err error) error { return mapping.ToPlatformStatusError(err, sessionsemantic.CapabilityName) },
		}, true
	case runtimev1.SessionLifecycleService_ReadSession_FullMethodName,
		runtimev1.SessionLifecycleService_RevokeSession_FullMethodName:
		return methodPolicy{
			capability: sessionsemantic.CapabilityName, protocolMajor: sessionsemantic.ProtocolMajor,
			protocolMinor: sessionsemantic.ProtocolMinor, methodClass: MethodClassWorkloadConsumer,
			resourceClass: ResourceClassShortUnary, attemptPolicy: attemptIdentity,
			sensitiveSessionPresented: true,
			statusError:               func(err error) error { return mapping.ToPlatformStatusError(err, sessionsemantic.CapabilityName) },
		}, true
	case runtimev1.AuthorizationLifecycleService_BeginAuthorization_FullMethodName:
		return authorizationLifecycleMethodPolicy(), true
	case runtimev1.AuthorizationLifecycleService_CompleteAuthorization_FullMethodName:
		result := authorizationLifecycleMethodPolicy()
		result.sensitiveSessionPresented = true
		return result, true
	case runtimev1.AuthorizationLifecycleService_ExchangeAuthorizationToken_FullMethodName:
		result := authorizationLifecycleMethodPolicy()
		result.sensitiveAuthorizationExchange = true
		return result, true
	case runtimev1.FederatedLoginTransactionService_BeginFederatedLogin_FullMethodName:
		return federatedLoginMethodPolicy(mapping.ToFederatedLoginBeginStatusError), true
	case runtimev1.FederatedLoginTransactionService_ConsumeFederatedLogin_FullMethodName:
		return federatedLoginMethodPolicy(mapping.ToFederatedLoginConsumeStatusError), true
	case catalogv1.ConsumerViewService_GetAppMetadataView_FullMethodName:
		return platformMethodPolicy(catalogsemantic.AppMetadataCapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case catalogv1.ConsumerViewService_WatchAppMetadataView_FullMethodName:
		return platformMethodPolicy(catalogsemantic.AppMetadataCapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch), true
	case catalogv1.ConsumerViewService_GetAppMetadataViewV2_FullMethodName:
		return platformMethodPolicy(catalogsemantic.AppMetadataV2CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case catalogv1.ConsumerViewService_WatchAppMetadataViewV2_FullMethodName:
		return platformMethodPolicy(catalogsemantic.AppMetadataV2CapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch), true
	case catalogv1.AppMetadataViewV3Service_GetAppMetadataViewV3_FullMethodName:
		return platformMethodPolicy(catalogsemantic.AppMetadataV3CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case catalogv1.AppMetadataViewV3Service_WatchAppMetadataViewV3_FullMethodName:
		return platformMethodPolicy(catalogsemantic.AppMetadataV3CapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch), true
	case catalogv1.ConsumerViewService_GetIntegrationBindingView_FullMethodName:
		return platformMethodPolicy(catalogsemantic.IntegrationBindingCapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case catalogv1.ConsumerViewService_WatchIntegrationBindingView_FullMethodName:
		return platformMethodPolicy(catalogsemantic.IntegrationBindingCapabilityName, MethodClassWorkloadConsumer, ResourceClassCatalogWatch), true
	case runtimev1.RecordSearchService_SearchRecords_FullMethodName,
		runtimev1.RecordSearchService_ReadSearchScopeOrder_FullMethodName:
		return platformMethodPolicy(searchsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.FlowExecutionService_ExecuteFlow_FullMethodName,
		runtimev1.FlowExecutionService_PreviewCatalogAction_FullMethodName:
		return platformMethodPolicy(flowsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.ReportReadService_ReadReport_FullMethodName:
		return platformMethodPolicy(reportsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.AuditLogReadService_ReadAuditLog_FullMethodName:
		return platformMethodPolicy(auditsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.PrivateListViewService_ReadPrivateListViews_FullMethodName,
		runtimev1.PrivateListViewService_MutatePrivateListView_FullMethodName:
		return platformMethodPolicy(listviewsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.MetadataAdminReadService_ReadObjectAdminIndex_FullMethodName,
		runtimev1.MetadataAdminReadService_ReadLayoutAdminWorkspace_FullMethodName,
		runtimev1.MetadataAdminReadService_ReadStoredLayout_FullMethodName,
		runtimev1.MetadataAdminReadService_ReadValidationRuleAdminIndex_FullMethodName:
		return platformMethodPolicy(metadataadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.ConnectedMailboxAuthorizationService_AuthorizeConnectedMailbox_FullMethodName:
		return platformMethodPolicy(mailboxauthorizationsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary), true
	case runtimev1.ConnectedMailboxAuthorizationService_ReauthorizeConnectedMailbox_FullMethodName:
		return platformMethodPolicy(mailboxauthorizationsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case runtimev1.AuthorizedFilesService_CreateFile_FullMethodName,
		runtimev1.AuthorizedFilesService_OpenFile_FullMethodName:
		return platformMethodPolicy(filessemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassFileStream), true
	case runtimev1.AuthorizedFilesService_DeleteFile_FullMethodName:
		return platformMethodPolicy(filessemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary), true
	case runtimev1.BulkIngestService_UploadBulkIngestData_FullMethodName:
		return platformMethodPolicy(bulkingestsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassFileStream), true
	case runtimev1.BulkIngestService_CreateBulkIngestJob_FullMethodName,
		runtimev1.BulkIngestService_ReadBulkIngestJob_FullMethodName,
		runtimev1.BulkIngestService_CloseBulkIngestJob_FullMethodName:
		return platformMethodPolicy(bulkingestsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary), true
	case runtimev1.CommandAcceptanceService_AcceptCommand_FullMethodName:
		return platformMethodPolicy(commandsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case runtimev1.ApplicationResultService_GetApplicationResult_FullMethodName:
		return platformMethodPolicy(applicationresultsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case runtimev1.ExternalEffectDispatchService_ClaimExternalEffects_FullMethodName,
		runtimev1.ExternalEffectDispatchService_CompleteExternalEffect_FullMethodName:
		return platformMethodPolicy(externaleffectsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case runtimev1.OutboundIntentService_OpenOutboundMailAttachment_FullMethodName:
		return platformMethodPolicy(outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassFileStream), true
	case runtimev1.OutboundIntentService_ReadOutboundIntentGeneration_FullMethodName,
		runtimev1.OutboundIntentService_ListPendingOutboundIntents_FullMethodName,
		runtimev1.OutboundIntentService_AcknowledgeOutboundIntent_FullMethodName,
		runtimev1.OutboundIntentService_CommitOutboundResult_FullMethodName,
		runtimev1.OutboundIntentNoEffectService_CommitOutboundNoEffect_FullMethodName:
		return platformMethodPolicy(outboundintentsemantic.CapabilityName, MethodClassWorkloadConsumer, ResourceClassShortUnary), true
	case runtimev1.EffectivePermissionService_ReadEffectivePermissions_FullMethodName:
		return platformMethodPolicy(permissionsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.PermissionAdministrationService_ReadPermissionAdministration_FullMethodName,
		runtimev1.PermissionAdministrationService_MutatePermissionAdministration_FullMethodName:
		return platformMethodPolicy(permissionadminsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.PermissionAuthoringService_ReadPermissionAuthoringSnapshot_FullMethodName,
		runtimev1.PermissionAuthoringService_SavePermissionAuthoring_FullMethodName:
		return platformMethodPolicy(permissionauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.PermissionAuthoringService_GetPermissionAuthoringStatus_FullMethodName:
		return platformMethodPolicy(permissionauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary), true
	case runtimev1.LayoutAuthoringService_Submit_FullMethodName,
		runtimev1.LayoutAuthoringService_ReadWorkspace_FullMethodName,
		runtimev1.LayoutAuthoringService_ReadStoredLayout_FullMethodName:
		return platformMethodPolicy(layoutauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.LayoutAuthoringService_GetCommand_FullMethodName:
		return platformMethodPolicy(layoutauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary), true
	case runtimev1.ApplicationNavigationService_Read_FullMethodName, runtimev1.ApplicationNavigationService_Resolve_FullMethodName, runtimev1.ApplicationNavigationService_Save_FullMethodName:
		return platformMethodPolicy(navigationsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.LayoutAuthoringV2Service_ReadWorkspace_FullMethodName,
		runtimev1.LayoutAuthoringV2Service_ReadStoredLayout_FullMethodName,
		runtimev1.LayoutAuthoringV2Service_Mutate_FullMethodName:
		return platformMethodPolicy(layoutauthoringv2semantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.AnalyticsAuthoringService_MutateAnalyticsDefinition_FullMethodName:
		return platformMethodPolicy(analyticsauthoringsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.ResourceAccessService_AccessResource_FullMethodName:
		return platformMethodPolicy(resourceaccesssemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.ApprovalService_ExecuteApproval_FullMethodName:
		return platformMethodPolicy(approvalsemantic.ExecuteCapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.ApprovalService_ReadApproval_FullMethodName:
		return platformMethodPolicy(approvalsemantic.ReadCapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.ApprovalNotificationFlowReadService_ReadApprovalNotificationFlows_FullMethodName:
		return platformMethodPolicy(approvalnotificationsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassHeavyWork), true
	case runtimev1.PageLayoutResolutionService_ResolvePageLayout_FullMethodName,
		runtimev1.PageLayoutResolutionService_ReadObjectListFields_FullMethodName:
		return platformMethodPolicy(pagelayoutsemantic.CapabilityName, MethodClassDelegatedUserData, ResourceClassShortUnary), true
	default:
		return methodPolicy{}, false
	}
}

func delegatedMethod(
	capability string,
	protocolMajor, protocolMinor uint32,
	attempt attemptPolicy,
	resource ResourceClass,
	statusError func(error) error,
) methodPolicy {
	return methodPolicy{
		capability: capability, protocolMajor: protocolMajor, protocolMinor: protocolMinor,
		methodClass: MethodClassDelegatedUserData, resourceClass: resource,
		attemptPolicy: attempt, statusError: statusError,
	}
}

func platformMethodPolicy(capability string, class MethodClass, resource ResourceClass) methodPolicy {
	statusError, _ := platformMethodStatus(capability)
	return methodPolicy{
		capability: capability, protocolMajor: commonsemantic.ProtocolMajor, protocolMinor: commonsemantic.ProtocolMinor,
		methodClass: class, resourceClass: resource, attemptPolicy: attemptPlatform, statusError: statusError,
	}
}

func authorizationLifecycleMethodPolicy() methodPolicy {
	return methodPolicy{
		capability: authorizationsemantic.CapabilityName, protocolMajor: authorizationsemantic.ProtocolMajor,
		protocolMinor: authorizationsemantic.ProtocolMinor, methodClass: MethodClassWorkloadConsumer,
		resourceClass: ResourceClassShortUnary, attemptPolicy: attemptIdentity,
		statusError: func(err error) error {
			return mapping.ToPlatformStatusError(err, authorizationsemantic.CapabilityName)
		},
	}
}

func federatedLoginMethodPolicy(statusError func(error) error) methodPolicy {
	return methodPolicy{
		capability: federatedloginsemantic.CapabilityName, protocolMajor: federatedloginsemantic.ProtocolMajor,
		protocolMinor: federatedloginsemantic.ProtocolMinor, methodClass: MethodClassWorkloadConsumer,
		resourceClass: ResourceClassShortUnary, attemptPolicy: attemptIdentity,
		statusError: statusError,
	}
}

func platformMethodStatus(capability string) (func(error) error, bool) {
	switch capability {
	case approvalsemantic.ExecuteCapabilityName,
		approvalsemantic.ReadCapabilityName,
		approvalnotificationsemantic.CapabilityName,
		auditsemantic.CapabilityName,
		catalogsemantic.AppMetadataCapabilityName,
		catalogsemantic.AppMetadataV2CapabilityName,
		catalogsemantic.AppMetadataV3CapabilityName,
		catalogsemantic.IntegrationBindingCapabilityName,
		searchsemantic.CapabilityName,
		flowsemantic.CapabilityName,
		reportsemantic.CapabilityName,
		listviewsemantic.CapabilityName,
		metadataadminsemantic.CapabilityName,
		mailboxauthorizationsemantic.CapabilityName,
		filessemantic.CapabilityName,
		bulkingestsemantic.CapabilityName,
		applicationresultsemantic.CapabilityName,
		externaleffectsemantic.CapabilityName,
		outboundintentsemantic.CapabilityName,
		commandsemantic.CapabilityName,
		permissionsemantic.CapabilityName,
		permissionadminsemantic.CapabilityName,
		permissionauthoringsemantic.CapabilityName,
		layoutauthoringsemantic.CapabilityName,
		layoutauthoringv2semantic.CapabilityName,
		navigationsemantic.CapabilityName,
		analyticsauthoringsemantic.CapabilityName,
		resourceaccesssemantic.CapabilityName,
		pagelayoutsemantic.CapabilityName:
		return func(err error) error { return mapping.ToPlatformStatusError(err, capability) }, true
	default:
		return nil, false
	}
}

func cloneAdmissionLeases(source map[string]time.Duration) map[string]time.Duration {
	if len(source) == 0 {
		return nil
	}
	result := make(map[string]time.Duration, len(source))
	for capability, lease := range source {
		result[capability] = lease
	}
	return result
}

type admissionResult uint8

const (
	admissionAccepted admissionResult = iota
	admissionDraining
	admissionFull
)

func (p *Policy) admissionState(class ResourceClass) *admissionState {
	if p == nil {
		return nil
	}
	switch class {
	case ResourceClassShortUnary, ResourceClassHeavyWork, ResourceClassFileStream, ResourceClassCatalogWatch:
	default:
		return nil
	}
	return p.admission[class]
}

type admissionLease struct {
	global   *admissionState
	resource *admissionState
}

func (p *Policy) admit(class ResourceClass) (admissionResult, admissionLease, int) {
	p.mu.Lock()
	defer p.mu.Unlock()
	global := p.globalAdmission
	resource := p.admissionState(class)
	if global == nil || resource == nil {
		return admissionDraining, admissionLease{}, 0
	}
	lease := admissionLease{global: global, resource: resource}
	if p.draining {
		global.rejectedDraining++
		if resource != global {
			resource.rejectedDraining++
		}
		return admissionDraining, admissionLease{}, 0
	}
	if global.inFlight >= global.limit {
		global.rejectedFull++
		if resource != global {
			resource.rejectedFull++
		}
		return admissionFull, admissionLease{}, global.limit
	}
	if resource != global && resource.inFlight >= resource.limit {
		global.rejectedFull++
		resource.rejectedFull++
		return admissionFull, admissionLease{}, resource.limit
	}
	global.inFlight++
	global.accepted++
	if resource != global {
		resource.inFlight++
		resource.accepted++
	}
	return admissionAccepted, lease, 0
}

func (p *Policy) release(lease admissionLease) {
	p.mu.Lock()
	if lease.global != nil && lease.global.inFlight > 0 {
		lease.global.inFlight--
	}
	if lease.resource != nil && lease.resource != lease.global && lease.resource.inFlight > 0 {
		lease.resource.inFlight--
	}
	p.mu.Unlock()
}

func requestContextError(parent, attempt context.Context) error {
	if err := parent.Err(); err != nil {
		return err
	}
	return attempt.Err()
}

type identityOnlyStream struct{ grpc.ServerTransportStream }

func compatible(ctx context.Context, method methodPolicy) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	return ok &&
		exactly(values.Get(envelope.ProtocolMajorMetadataKey), strconv.FormatUint(uint64(method.protocolMajor), 10)) &&
		exactly(values.Get(envelope.ProtocolMinorMetadataKey), strconv.FormatUint(uint64(method.protocolMinor), 10)) &&
		exactly(values.Get(envelope.CapabilityMetadataKey), method.capability)
}

func withoutIncomingEndUserAuthorization(ctx context.Context) context.Context {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return ctx
	}
	values = values.Copy()
	for _, key := range sessionSensitiveMetadataKeys() {
		values.Delete(key)
	}
	return metadata.NewIncomingContext(ctx, values)
}

func authorityInputContext(ctx context.Context, class MethodClass) context.Context {
	ctx = withoutIncomingEndUserAuthorization(ctx)
	if class == MethodClassDelegatedUserData {
		return ctx
	}
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return ctx
	}
	values = values.Copy()
	values.Delete(envelope.RuntimeDelegationMetadataKey)
	return metadata.NewIncomingContext(ctx, values)
}

func incomingRuntimeDelegationPresent(ctx context.Context) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	return ok && len(values.Get(envelope.RuntimeDelegationMetadataKey)) != 0
}

func incomingEndUserBearer(ctx context.Context) (string, bool) {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return "", false
	}
	authorization := values.Get(EndUserAuthorizationMetadataKey)
	if len(authorization) != 1 || !strings.HasPrefix(authorization[0], "Bearer ") {
		return "", false
	}
	bearer := strings.TrimPrefix(authorization[0], "Bearer ")
	if bearer == "" || strings.TrimSpace(bearer) != bearer || len(bearer) > envelope.MaxRequestWireBytes {
		return "", false
	}
	return bearer, true
}

func incomingEndUserAuthorizationPresent(ctx context.Context) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	return ok && len(values.Get(EndUserAuthorizationMetadataKey)) != 0
}

func sessionSensitiveMetadataKeys() []string {
	return []string{
		EndUserAuthorizationMetadataKey,
		SessionIdentityMetadataKey,
		SessionSecretMetadataKey,
		SessionClientSecretMetadataKey,
		SessionControlProofMetadataKey,
		IssuedSessionAccessTokenMetadataKey,
		IssuedSessionControlProofMetadataKey,
		AuthorizationClientSecretMetadataKey,
		AuthorizationCodeMetadataKey,
		AuthorizationCodeVerifierMetadataKey,
		AuthorizationRefreshTokenMetadataKey,
		IssuedAuthorizationAccessTokenMetadataKey,
		IssuedAuthorizationRefreshTokenMetadataKey,
		IssuedAuthorizationCodeMetadataKey,
	}
}

func incomingAuthorizationExchangeCredentialMetadataPresent(ctx context.Context) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return false
	}
	for _, key := range []string{AuthorizationClientSecretMetadataKey, AuthorizationCodeMetadataKey, AuthorizationCodeVerifierMetadataKey, AuthorizationRefreshTokenMetadataKey} {
		if len(values.Get(key)) != 0 {
			return true
		}
	}
	return false
}

func incomingAuthorizationExchangeCredentials(ctx context.Context) (AuthorizationExchangeCredentials, bool) {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return AuthorizationExchangeCredentials{}, false
	}
	clientSecret, clientOK := exactSensitiveMetadataValue(values, AuthorizationClientSecretMetadataKey, maxAuthorizationClientSecretBytes, false, false)
	code, codeOK := exactSensitiveMetadataValue(values, AuthorizationCodeMetadataKey, maxAuthorizationCodeBytes, false, true)
	verifier, verifierOK := exactSensitiveMetadataValue(values, AuthorizationCodeVerifierMetadataKey, maxAuthorizationCodeVerifierBytes, false, true)
	refresh, refreshOK := exactSensitiveMetadataValue(values, AuthorizationRefreshTokenMetadataKey, maxAuthorizationRefreshTokenBytes, false, true)
	codeProfile := code != "" && refresh == ""
	refreshProfile := refresh != "" && code == "" && verifier == ""
	clientCredentialsProfile := clientSecret != "" && code == "" && verifier == "" && refresh == ""
	if !clientOK || !codeOK || !verifierOK || !refreshOK || (!codeProfile && !refreshProfile && !clientCredentialsProfile) {
		return AuthorizationExchangeCredentials{}, false
	}
	return AuthorizationExchangeCredentials{ClientSecret: clientSecret, AuthorizationCode: code, CodeVerifier: verifier, RefreshToken: refresh}, true
}

func incomingSessionCreateCredentialMetadataPresent(ctx context.Context) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return false
	}
	for _, key := range []string{SessionIdentityMetadataKey, SessionSecretMetadataKey, SessionClientSecretMetadataKey} {
		if len(values.Get(key)) != 0 {
			return true
		}
	}
	return false
}

func incomingSessionIssuedCredentialMetadataPresent(ctx context.Context) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return false
	}
	return len(values.Get(IssuedSessionAccessTokenMetadataKey)) != 0 ||
		len(values.Get(IssuedSessionControlProofMetadataKey)) != 0 ||
		len(values.Get(IssuedAuthorizationAccessTokenMetadataKey)) != 0 ||
		len(values.Get(IssuedAuthorizationRefreshTokenMetadataKey)) != 0 ||
		len(values.Get(IssuedAuthorizationCodeMetadataKey)) != 0
}

func incomingSensitiveSessionMetadataPresent(ctx context.Context) bool {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return false
	}
	for _, key := range sessionSensitiveMetadataKeys() {
		if len(values.Get(key)) != 0 {
			return true
		}
	}
	return false
}

func incomingSessionCreateCredentials(ctx context.Context) (SessionCreateCredentials, bool) {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return SessionCreateCredentials{}, false
	}
	secret, ok := exactSensitiveMetadataValue(values, SessionSecretMetadataKey, maxSessionSecretBytes, true, false)
	if !ok {
		return SessionCreateCredentials{}, false
	}
	identity, identityOK := exactSensitiveMetadataValue(values, SessionIdentityMetadataKey, maxSessionIdentityBytes, false, true)
	clientSecret, clientSecretOK := exactSensitiveMetadataValue(values, SessionClientSecretMetadataKey, maxSessionClientSecretBytes, false, false)
	if !identityOK || !clientSecretOK {
		return SessionCreateCredentials{}, false
	}
	return SessionCreateCredentials{Identity: identity, Secret: secret, ClientSecret: clientSecret}, true
}

func incomingSessionControlProof(ctx context.Context) (string, bool, bool) {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok || len(values.Get(SessionControlProofMetadataKey)) == 0 {
		return "", false, true
	}
	proof, valid := exactSensitiveMetadataValue(values, SessionControlProofMetadataKey, maxSessionControlProofBytes, true, true)
	return proof, true, valid
}

func exactSensitiveMetadataValue(values metadata.MD, key string, maximum int, required, canonical bool) (string, bool) {
	presented := values.Get(key)
	if len(presented) == 0 {
		return "", !required
	}
	if len(presented) != 1 || presented[0] == "" || len(presented[0]) > maximum ||
		(canonical && strings.TrimSpace(presented[0]) != presented[0]) {
		return "", false
	}
	return presented[0], true
}

func exactly(values []string, expected string) bool {
	return len(values) == 1 && values[0] == expected
}

type compressionContextKey struct{}

type compressionState struct{ used atomic.Bool }

type compressionStatsHandler struct{}

func (compressionStatsHandler) TagRPC(ctx context.Context, _ *stats.RPCTagInfo) context.Context {
	return context.WithValue(ctx, compressionContextKey{}, &compressionState{})
}

func (compressionStatsHandler) HandleRPC(ctx context.Context, event stats.RPCStats) {
	header, ok := event.(*stats.InHeader)
	if !ok || header.Compression == "" || header.Compression == "identity" {
		return
	}
	if state, ok := ctx.Value(compressionContextKey{}).(*compressionState); ok {
		state.used.Store(true)
	}
}

func (compressionStatsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context {
	return ctx
}

func (compressionStatsHandler) HandleConn(context.Context, stats.ConnStats) {}

func compressionUsed(ctx context.Context) bool {
	state, ok := ctx.Value(compressionContextKey{}).(*compressionState)
	return ok && state.used.Load()
}