package runtimeapi

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	navigationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationnavigationv1"
	"time"

	auditengine "github.com/nexuscrm/audit"
	"github.com/nexuscrm/automation"
	backendmiddleware "github.com/nexuscrm/backend/internal/host/adapters/apihost/middleware"
	"github.com/nexuscrm/backend/internal/managedruntime"
	"github.com/nexuscrm/backend/internal/runtimeassembly"
	recordmutation "github.com/nexuscrm/backend/internal/runtimeassembly/recordmutation"
	"github.com/nexuscrm/backend/internal/runtimeconfig"
	"github.com/nexuscrm/backend/internal/runtimeprocess"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/kernel/pkg/telemetry"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	approvalnotificationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalnotificationflowv1"
	approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1"
	auditsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/auditlogv1"
	bulkingestsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/bulkingestv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	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"
	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"
	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"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	sessionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/sessionv1"
	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/prometheus/client_golang/prometheus"
	"github.com/rs/zerolog"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials"
)

const (
	DefaultAdmissionLimit = runtimeconfig.DefaultAPIAdmissionLimit
	MaxAdmissionLimit     = runtimeconfig.MaxAPIAdmissionLimit
)

type ProcessConfig struct {
	ApplicationNavigation                 ApplicationNavigationEngine
	ApplicationNavigationAdmission        *ApplicationNavigationAdmission
	Version                               string
	AdminAddress                          string
	GRPCAddress                           string
	AdmissionLimit                        int
	ProfilingEnabled                      bool
	SQLDiagnosticsEnabled                 bool
	WorkloadAuthority                     runtimegrpc.Authority
	SecurityAdmission                     *SecurityAdmission
	ServingAdmission                      *ServingAdmission
	CatalogConsumerAdmission              *CatalogConsumerAdmission
	CommandConsumerAdmission              *CommandConsumerAdmission
	PageLayoutConsumerAdmission           *PageLayoutConsumerAdmission
	OutboundIntentAdmission               *OutboundIntentAdmission
	ApplicationResultAdmission            *ApplicationResultAdmission
	MailboxAuthorization                  *MailboxAuthorizationProcessConfig
	MetadataAdminReadAdmission            *MetadataAdminReadAdmission
	LayoutAuthoringV2Admission            *LayoutAuthoringV2Admission
	ApprovalNotificationFlowReadAdmission *ApprovalNotificationFlowReadAdmission
	Receipts                              ReceiptSource
	IdentityExchange                      IdentityExchange
	SessionLifecycle                      SessionLifecycleEngine
	AuthorizationLifecycle                AuthorizationLifecycleEngine
	FederatedLoginTransactions            FederatedLoginTransactionEngine
	IdentityAttemptCap                    time.Duration
	// AdditionalMetricsCollectors are composition-owned operational metrics
	// exposed through this process's private registry.
	AdditionalMetricsCollectors []prometheus.Collector
	// RecordAccessMaxAdmissionLease is operational host enablement. It does
	// not replace the caller deadline or approve a semantic/SLO limit.
	RecordAccessMaxAdmissionLease time.Duration
	// QueryMaxAdmissionLease enables the neutral query RPC. It is an
	// operational caller-lease ceiling, not a production SLO.
	QueryMaxAdmissionLease time.Duration
	// RelatedCountMaxAdmissionLease enables one bounded neutral count batch.
	// It is an operational caller-lease ceiling, not a production SLO.
	RelatedCountMaxAdmissionLease time.Duration
	// CreateDefaultPreviewMaxAdmissionLease enables the bounded dynamic-default
	// preview. The eventual create still recomputes authoritative write defaults.
	CreateDefaultPreviewMaxAdmissionLease time.Duration
	// RecycleBinMaxAdmissionLease enables one bounded deleted-record listing.
	// It is an operational caller-lease ceiling, not a production SLO.
	RecycleBinMaxAdmissionLease time.Duration
	// RecycleBinMutationMaxAdmissionLease enables restore and permanent purge.
	// It is an operational caller-lease ceiling, not a production SLO.
	RecycleBinMutationMaxAdmissionLease time.Duration
	// RecordViewMaxAdmissionLease enables the bounded record-view root read.
	// It is a host admission ceiling, not an L1 route SLO.
	RecordViewMaxAdmissionLease time.Duration
	// AttachmentSourceMaxAdmissionLease independently enables the proof-bearing
	// source resolver. It must never be inferred from record-view admission.
	AttachmentSourceMaxAdmissionLease time.Duration
	// RecordMutationAttemptCap enables the coarse durable mutation command and
	// its reconciliation method. It is a host availability bound, not a route
	// SLO or a caller-provided business field.
	RecordMutationAttemptCap time.Duration
	// RecordMutationOutcomeRetention is immutable command-outcome lineage. It
	// is required only when production composition selects record mutation.
	RecordMutationOutcomeRetention time.Duration
	// PlatformCapabilityMaxAdmissionLeases enables coarse phase-one Platform
	// RPC families on this same private endpoint. Missing entries remain
	// unregistered and unadvertised.
	PlatformCapabilityMaxAdmissionLeases map[string]time.Duration
	// PlatformResourceClassAdmissionLimits reserves bounded capacity inside the
	// process-wide AdmissionLimit. Only classes required by enabled Platform
	// capabilities may be configured.
	PlatformResourceClassAdmissionLimits map[runtimegrpc.ResourceClass]int
	// PermissionAuthoringOutcomeRetention freezes the replay window into every
	// accepted aggregate command. It is required when the capability is enabled.
	PermissionAuthoringOutcomeRetention time.Duration
	QueryEngine                         RecordQueryEngine
	CatalogConsumerViews                CatalogConsumerViewEngine
	SearchEngine                        RecordSearchEngine
	SearchScopeOrder                    SearchScopeOrderEngine
	ReportEngine                        ReportReadEngine
	AuditLog                            sharedcontract.AuditLog
	FileEngine                          AuthorizedFileEngine
	BulkIngest                          BulkIngestEngine
	CommandEngine                       CommandAcceptanceEngine
	OutboundIntents                     OutboundIntentEngine
	OutboundResultCommitter             OutboundResultCommitter
	OutboundIntentTransactions          OutboundIntentTransactionRunner
	OutboundMailAttachments             FrozenOutboundMailAttachmentEngine
	ApplicationResults                  ApplicationResultEngine
	EffectivePermissions                EffectivePermissionEngine
	MetadataAdminReads                  MetadataAdminReadEngine
	LayoutAuthoringV2                   LayoutAuthoringV2Engine
	ApprovalNotificationFlows           ApprovalNotificationFlowReadEngine
	PageLayouts                         PageLayoutResolver
	PageLayoutRoleHierarchy             PageLayoutRoleHierarchy
	FlowEngine                          FlowExecutionEngine
	FlowInvocations                     FlowInvocationStore
	FlowTransactions                    FlowTransactionRunner
	ApprovalEngine                      *ApprovalExecutionEngine
	ApprovalInvocations                 ApprovalInvocationStore
	ApprovalTransactions                ApprovalTransactionRunner
	ApprovalPermissionFence             approvalPermissionFence
	RecycleBinLister                    RecycleBinLister
	RecycleBinMutator                   RecycleBinMutator
	RecordView                          RecordViewEngine
	AttachmentSources                   AttachmentSourceEngine
	RecordMutation                      RecordMutationExecutor
	PrivateListViews                    PrivateListViewReader
	PrivateListViewAuthoring            sharedcontract.ListViewAuthoring
	PrivateListViewPermissions          privateListViewPermissionEngine
	PermissionAuthoring                 PermissionAuthoringEngine
	ServingEnabled                      bool
	TransportCredentials                credentials.TransportCredentials
	ServingController                   func(context.Context) error
	// PermissionStateCapabilities is the immutable production capability
	// identity assembled from the startup lineage and exact linked artifacts.
	PermissionStateCapabilities []string
	// PermissionStateReady strongly re-proves that the startup lineage and
	// serving-schema mode contract remain current.
	PermissionStateReady func() bool
}

type MailboxAuthorizationProcessConfig struct {
	Admission          *MailboxAuthorizationAdmission
	Handles            mailboxAuthorizationHandleStore
	Permissions        mailboxPermissionAuthorizer
	Sessions           mailboxSessionReader
	RequiredPermission MailboxPermissionRequirement
}

type runtimeMailboxAuthorizationTransactionOwner struct {
	transactions *persistence.TransactionManager
}

func (owner *runtimeMailboxAuthorizationTransactionOwner) RunInPessimisticTransaction(
	ctx context.Context,
	run func(context.Context) error,
) error {
	if owner == nil || owner.transactions == nil || run == nil {
		return errMailboxAuthorizationUnavailable
	}
	return owner.transactions.WithPessimisticRetry(ctx, run, 3)
}

func (owner *runtimeMailboxAuthorizationTransactionOwner) Executor(
	ctx context.Context,
) (mailboxAuthorizationSQLExecutor, bool) {
	if owner == nil || owner.transactions == nil ||
		!owner.transactions.OwnsPessimisticTx(ctx) {
		return nil, false
	}
	tx, ok := owner.transactions.ExtractTx(ctx).(*sql.Tx)
	return tx, ok && tx != nil
}

// AssembleProductionWithConnection builds the serving graph on the same
// Runtime-owned pool used by Identity, Catalog admission, and records. The
// caller owns the connection lifecycle.
func AssembleProductionWithConnection(ctx context.Context, config ProcessConfig, connection *database.TiDBConnection) (*runtimeprocess.Process, error) {
	if config.ServingAdmission == nil || config.Receipts != nil || !config.ServingEnabled ||
		config.TransportCredentials == nil || config.PermissionStateReady == nil {
		return nil, errors.New("runtime-api production composition requires serving admission, request-bound receipts, and explicit serving enablement")
	}
	if connection == nil {
		return nil, errors.New("runtime-api production composition requires a Runtime-owned database connection")
	}
	readDependencies := runtimeassembly.BuildRecordsReadDependencies(connection)
	readGraph, err := runtimeassembly.NewServingRecordsReadRuntime(
		readDependencies, runtimeRecordsMetadataViewFromContext,
	)
	if err != nil {
		return nil, err
	}
	if err := readGraph.Prepare(ctx); err != nil {
		return nil, err
	}
	if config.PlatformCapabilityMaxAdmissionLeases[bulkingestsemantic.CapabilityName] > 0 {
		transactions := persistence.NewTransactionManager(connection)
		config.BulkIngest = records.NewBulkIngestContract(
			records.NewBulkService(persistence.NewBulkRepository(connection.DB(), transactions), logging.New(ctx)),
			readGraph.Metadata,
			readGraph.Permissions,
		)
	}
	config.QueryEngine = readGraph.Query
	config.SearchEngine = readGraph.Query
	config.ReportEngine = readGraph.Reports
	if config.PlatformCapabilityMaxAdmissionLeases[pagelayoutsemantic.CapabilityName] > 0 {
		if config.PageLayouts != nil || config.PageLayoutRoleHierarchy != nil {
			return nil, errors.New("runtime-api production page Layout resolution does not accept an injected owner")
		}
		pageLayouts, pageLayoutErr := runtimeassembly.NewPageLayoutRuntime(connection, readGraph.Permissions)
		if pageLayoutErr != nil {
			return nil, pageLayoutErr
		}
		config.PageLayouts = pageLayouts.Engine
		config.PageLayoutRoleHierarchy = pageLayouts.Engine
	}
	if config.PlatformCapabilityMaxAdmissionLeases[auditsemantic.CapabilityName] > 0 {
		config.AuditLog = auditengine.NewReader(connection.DB())
	}
	if config.PlatformCapabilityMaxAdmissionLeases[searchsemantic.CapabilityName] > 0 {
		config.SearchScopeOrder, err = runtimeassembly.NewSearchScopeOrderProjection(
			connection, readGraph.Permissions,
		)
		if err != nil {
			return nil, err
		}
	}
	if config.PlatformCapabilityMaxAdmissionLeases[permissionsemantic.CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[metadataadminsemantic.CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[layoutauthoringv2semantic.CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[flowsemantic.CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[approvalsemantic.ExecuteCapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[auditsemantic.CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[mailboxauthorizationsemantic.CapabilityName] > 0 {
		config.EffectivePermissions, err = runtimeassembly.NewEffectivePermissionProjection(
			connection, readGraph.Permissions,
		)
		if err != nil {
			return nil, err
		}
	}
	if config.PlatformCapabilityMaxAdmissionLeases[navigationsemantic.CapabilityName] > 0 {
		if config.ApplicationNavigation != nil {
			return nil, errors.New("production navigation cannot accept an injected owner")
		}
		if err := persistence.VerifyApplicationNavigationSchema(ctx, connection.DB()); err != nil {
			return nil, err
		}
		config.ApplicationNavigation, err = runtimeassembly.NewApplicationNavigationRuntime(connection, readGraph.Permissions)
		if err != nil {
			return nil, err
		}
	}

	if config.PlatformCapabilityMaxAdmissionLeases[metadataadminsemantic.CapabilityName] > 0 {
		if config.MetadataAdminReads != nil {
			return nil, errors.New("runtime-api production metadata administration reads do not accept an injected owner")
		}
		metadataAdminReads, metadataErr := runtimeassembly.NewMetadataAdminReadRuntime(
			readDependencies.MetadataRepo,
		)
		if metadataErr != nil {
			return nil, metadataErr
		}
		if metadataErr := metadataAdminReads.Prepare(ctx); metadataErr != nil {
			return nil, metadataErr
		}
		config.MetadataAdminReads = metadataAdminReads.Reads
	}
	if config.PlatformCapabilityMaxAdmissionLeases[layoutauthoringv2semantic.CapabilityName] > 0 {
		if config.LayoutAuthoringV2 != nil {
			return nil, errors.New("runtime-api production Layout authoring v2 does not accept an injected owner")
		}
		layoutRuntime, layoutErr := runtimeassembly.NewLayoutAuthoringRuntime(ctx, connection, readGraph.Permissions)
		if layoutErr != nil {
			return nil, layoutErr
		}
		if layoutErr := layoutRuntime.Prepare(ctx); layoutErr != nil {
			return nil, layoutErr
		}
		config.LayoutAuthoringV2 = layoutRuntime.Engine
	}
	if config.PlatformCapabilityMaxAdmissionLeases[approvalnotificationsemantic.CapabilityName] > 0 {
		if config.ApprovalNotificationFlows != nil {
			return nil, errors.New("runtime-api production approval notification Flow reads do not accept an injected owner")
		}
		approvalNotificationFlows, readErr := runtimeassembly.NewApprovalNotificationFlowReadRuntime(
			readDependencies.MetadataRepo,
			automation.NewProtectedFlowCatalog(connection.DB()),
		)
		if readErr != nil {
			return nil, readErr
		}
		config.ApprovalNotificationFlows = approvalNotificationFlows
	}
	filesEnabled := config.PlatformCapabilityMaxAdmissionLeases[filessemantic.CapabilityName] > 0
	outboundIntentEnabled := config.PlatformCapabilityMaxAdmissionLeases[outboundintentsemantic.CapabilityName] > 0
	var outboundMailFiles sharedcontract.Files
	if filesEnabled || outboundIntentEnabled {
		// The outbound-mail Flow action and attachment stream share this exact
		// permission/storage authority. Selecting outbound intents must not also
		// expose the separately selectable public Files RPC family.
		if filesEnabled {
			if err := persistence.VerifyFileUploadOperationSchema(ctx, connection.DB()); err != nil {
				return nil, err
			}
		} else if outboundIntentEnabled {
			if err := persistence.VerifyRecordAttachmentReadSchema(ctx, connection.DB()); err != nil {
				return nil, err
			}
		}
		fileService, fileErr := readGraph.NewFileService()
		if fileErr != nil {
			return nil, fileErr
		}
		if filesEnabled {
			config.FileEngine = fileService
		}
		if outboundIntentEnabled {
			config.OutboundMailAttachments = fileService
			outboundMailFiles = fileService
		}
	}
	commandEnabled := config.PlatformCapabilityMaxAdmissionLeases[commandsemantic.CapabilityName] > 0
	mailboxAuthorizationEnabled := config.PlatformCapabilityMaxAdmissionLeases[mailboxauthorizationsemantic.CapabilityName] > 0
	applicationResultEnabled := config.PlatformCapabilityMaxAdmissionLeases[applicationresultsemantic.CapabilityName] > 0
	needsPlatformCommandRepository := commandEnabled || applicationResultEnabled
	if needsPlatformCommandRepository {
		if err := persistence.VerifyRuntimePlatformCommandSchema(ctx, connection.DB()); err != nil {
			return nil, err
		}
	}
	if mailboxAuthorizationEnabled {
		if config.MailboxAuthorization == nil ||
			config.MailboxAuthorization.Admission == nil {
			return nil, errors.New(
				"runtime-api mailbox authorization service requires exact Integration consumer admission",
			)
		}
		mailboxConfig := *config.MailboxAuthorization
		if mailboxConfig.Handles != nil || mailboxConfig.Permissions != nil ||
			mailboxConfig.Sessions != nil {
			return nil, errors.New(
				"runtime-api production mailbox authorization composition does not accept injected owners",
			)
		}
		transactions := &runtimeMailboxAuthorizationTransactionOwner{
			transactions: persistence.NewTransactionManager(connection),
		}
		handles, handleErr := NewSQLMailboxAuthorizationHandleStore(
			connection.DB(),
			transactions,
		)
		if handleErr != nil {
			return nil, handleErr
		}
		if handleErr := handles.VerifySchema(ctx); handleErr != nil {
			return nil, handleErr
		}
		if _, handleErr := handles.CleanupExpired(
			ctx,
			mailboxAuthorizationCleanupBatch,
		); handleErr != nil {
			return nil, handleErr
		}
		permissionFencer, ok := config.EffectivePermissions.(effectivePermissionFloorFencer)
		if !ok {
			return nil, errors.New(
				"runtime-api production mailbox authorization requires the transaction-bound permission fence",
			)
		}
		permissionAuthorizer, permissionErr := newCurrentMailboxPermissionAuthorizer(
			readGraph.Permissions,
			permissionFencer,
		)
		if permissionErr != nil {
			return nil, permissionErr
		}
		mailboxConfig.Handles = handles
		mailboxConfig.Permissions = permissionAuthorizer
		mailboxConfig.Sessions = config.SecurityAdmission
		config.MailboxAuthorization = &mailboxConfig
	}
	config.RecycleBinLister = readGraph.Records
	config.RecordView = readGraph.RecordView
	config.AttachmentSources = readGraph.AttachmentSources
	needsFlowGraph := config.PlatformCapabilityMaxAdmissionLeases[flowsemantic.CapabilityName] > 0
	needsApprovalExecute := config.PlatformCapabilityMaxAdmissionLeases[approvalsemantic.ExecuteCapabilityName] > 0
	needsApprovalRead := config.PlatformCapabilityMaxAdmissionLeases[approvalsemantic.ReadCapabilityName] > 0
	needsApprovalGraph := needsApprovalExecute || needsApprovalRead
	needsPrivateListViews := config.PlatformCapabilityMaxAdmissionLeases[listviewsemantic.CapabilityName] > 0
	permissionAuthoringEnabled := config.PlatformCapabilityMaxAdmissionLeases[permissionauthoringsemantic.CapabilityName] > 0
	recycleBinMutationEnabled := config.RecycleBinMutationMaxAdmissionLease > 0
	needsMutationWriters := config.RecordMutationOutcomeRetention > 0 || needsFlowGraph || needsApprovalExecute || outboundIntentEnabled || needsPrivateListViews || recycleBinMutationEnabled || permissionAuthoringEnabled
	needsMutationGraph := needsMutationWriters || needsApprovalGraph
	if needsMutationGraph {
		if config.RecycleBinMutator != nil || config.RecordMutation != nil || config.FlowEngine != nil ||
			config.FlowInvocations != nil || config.FlowTransactions != nil || config.ApprovalEngine != nil ||
			config.ApprovalInvocations != nil || config.ApprovalTransactions != nil || config.ApprovalPermissionFence != nil ||
			config.OutboundIntents != nil || config.OutboundResultCommitter != nil || config.OutboundIntentTransactions != nil ||
			config.PermissionAuthoring != nil ||
			(config.RecordMutationOutcomeRetention > 0 && config.RecordMutationAttemptCap <= 0) {
			return nil, errors.New("runtime-api production mutation composition requires an explicit host attempt cap and no injected executor")
		}
		deps, err := recordmutation.BuildRecordsMutationDependencies(connection)
		if err != nil {
			return nil, err
		}
		mutationGraph, err := recordmutation.NewRecordsMutationEngineRuntime(
			deps,
			RuntimeMutationMetadataViewFromContext,
		)
		if err != nil {
			return nil, err
		}
		if err := managedruntime.InstallIndependentAssignmentPolicyEligibilityLifecycle(
			mutationGraph.Transactions,
			mutationGraph,
		); err != nil {
			return nil, err
		}
		if err := mutationGraph.PrepareEngine(ctx); err != nil {
			return nil, err
		}
		// Producers become active only after all of their durable schemas and
		// independently fenced worker-family owners are proven. The mutation
		// graph constructor deliberately cannot perform this flip.
		if needsMutationWriters {
			if err := mutationGraph.ActivateDurableMutationProducers(ctx); err != nil {
				return nil, err
			}
		}
		if config.RecordMutationOutcomeRetention > 0 {
			mutation, err := mutationGraph.NewRuntimeRecordMutationService(config.RecordMutationOutcomeRetention)
			if err != nil {
				return nil, err
			}
			config.RecordMutation = mutation
		}
		if recycleBinMutationEnabled {
			config.RecycleBinMutator = mutationGraph.Records
		}
		if needsPrivateListViews {
			config.PrivateListViews = mutationGraph.ListViews
			config.PrivateListViewAuthoring = mutationGraph.ListViewAuthoring
			config.PrivateListViewPermissions = mutationGraph.Permissions
		}
		if permissionAuthoringEnabled {
			if config.PermissionAuthoringOutcomeRetention <= 0 {
				return nil, errors.New("runtime-api production permission authoring requires explicit terminal outcome retention")
			}
			if err := persistence.VerifyPermissionAuthoringRuntimeSchema(ctx, connection.DB()); err != nil {
				return nil, err
			}
			if err := persistence.VerifyPermissionAuthoringSemanticCutover(
				ctx, connection.DB(), permissionauthoringsemantic.SemanticMappingRevision,
			); err != nil {
				return nil, err
			}
			if err := persistence.VerifyActiveRuntimeWorkFamily(
				ctx,
				connection.DB(),
				persistence.RuntimePermissionAuthoringCommandLifecycleWorkFamily,
			); err != nil {
				return nil, fmt.Errorf("runtime-api permission authoring recovery owner: %w", err)
			}
			authoringRuntime, authoringErr := mutationGraph.NewPermissionAuthoringRuntime(
				runtimePermissionAuthoringReceiptSource{},
				permissions.PermissionAuthoringServiceOptions{
					OutcomeRetention: config.PermissionAuthoringOutcomeRetention,
				},
			)
			if authoringErr != nil {
				return nil, authoringErr
			}
			config.PermissionAuthoring, authoringErr = newProductionPermissionAuthoringEngine(
				authoringRuntime.Snapshot,
				authoringRuntime.Save,
			)
			if authoringErr != nil {
				return nil, authoringErr
			}
		}
		if needsFlowGraph {
			if config.CatalogConsumerViews == nil {
				return nil, errors.New("runtime-api production Flow composition requires the Catalog consumer authority")
			}
			if err := persistence.VerifyRuntimeFlowInvocationSchema(ctx, connection.DB()); err != nil {
				return nil, err
			}
			config.FlowEngine = mutationGraph.Flows
			config.FlowTransactions = mutationGraph.Persistence
			config.FlowInvocations = persistence.NewRuntimeFlowInvocationRepository(
				connection.DB(), mutationGraph.Transactions,
			)
		}
		if needsApprovalGraph {
			config.ApprovalEngine, err = NewApprovalExecutionEngine(
				mutationGraph.Approval,
				persistence.NewRuntimeApprovalIdentityAnchorRepository(
					connection.DB(), mutationGraph.Transactions,
				),
			)
			if err != nil {
				return nil, err
			}
		}
		if needsApprovalExecute {
			if err := persistence.VerifyRuntimeApprovalInvocationSchema(ctx, connection.DB()); err != nil {
				return nil, err
			}
			config.ApprovalInvocations = persistence.NewRuntimeApprovalInvocationRepository(
				connection.DB(), mutationGraph.Transactions,
			)
			config.ApprovalTransactions = mutationGraph.Persistence
			var ok bool
			config.ApprovalPermissionFence, ok = config.EffectivePermissions.(approvalPermissionFence)
			if !ok {
				return nil, errors.New("runtime-api production Approval execution requires the transaction-bound permission fence")
			}
		}
		if outboundIntentEnabled {
			if err := mutationGraph.InstallOutboundMailAttachmentMetadataAuthorizer(outboundMailFiles); err != nil {
				return nil, err
			}
			if err := persistence.VerifyRuntimeOutboundIntentSchema(ctx, connection.DB()); err != nil {
				return nil, err
			}
			repository := persistence.NewRuntimeOutboundIntentRepository(
				connection.DB(), mutationGraph.Transactions,
			)
			if err := mutationGraph.Flows.EnableRuntimeOutboundIntents(repository); err != nil {
				return nil, err
			}
			config.OutboundIntents = repository
			config.OutboundResultCommitter, err = automation.NewRuntimeOutboundResultCommitter(
				mutationGraph.Flows, repository,
			)
			if err != nil {
				return nil, err
			}
			config.OutboundIntentTransactions = mutationGraph.Persistence
		}
	} else if config.RecycleBinMutator != nil || config.RecordMutationAttemptCap != 0 || config.RecordMutation != nil ||
		config.FlowEngine != nil || config.FlowInvocations != nil || config.FlowTransactions != nil ||
		config.ApprovalEngine != nil || config.ApprovalInvocations != nil || config.ApprovalTransactions != nil ||
		config.ApprovalPermissionFence != nil || config.OutboundIntents != nil ||
		config.OutboundResultCommitter != nil || config.OutboundIntentTransactions != nil ||
		config.PrivateListViews != nil || config.PrivateListViewAuthoring != nil || config.PrivateListViewPermissions != nil ||
		config.PermissionAuthoring != nil || config.PermissionAuthoringOutcomeRetention != 0 {
		return nil, errors.New("runtime-api production read composition cannot configure record mutation")
	}
	if needsPlatformCommandRepository {
		repository := persistence.NewRuntimePlatformCommandRepository(connection.DB())
		if commandEnabled {
			config.CommandEngine = repository
		}
		if applicationResultEnabled {
			config.ApplicationResults = repository
		}
	}
	process, err := assemble(config, readGraph.Records, connection.DB())
	if err != nil {
		return nil, err
	}
	return process, nil
}

func Assemble(config ProcessConfig, reader RecordsReader) (*runtimeprocess.Process, error) {
	return assemble(config, reader, nil)
}

func platformRequiredResourceClasses(
	capabilities map[string]time.Duration,
) (map[runtimegrpc.ResourceClass]struct{}, error) {
	required := make(map[runtimegrpc.ResourceClass]struct{})
	for capability, lease := range capabilities {
		if lease <= 0 {
			continue
		}
		class, ok := runtimegrpc.PlatformCapabilityResourceClass(capability)
		if !ok {
			return nil, fmt.Errorf("runtime-api Platform capability %q has no host-owned resource classification", capability)
		}
		if class != runtimegrpc.ResourceClassShortUnary {
			required[class] = struct{}{}
		}
	}
	return required, nil
}

func assemble(config ProcessConfig, reader RecordsReader, pool *sql.DB) (*runtimeprocess.Process, error) {
	if (config.IdentityExchange == nil) != (config.IdentityAttemptCap == 0) {
		return nil, errors.New("runtime-api identity exchange and its explicit attempt cap must be configured together")
	}
	if config.SessionLifecycle != nil && config.IdentityExchange == nil {
		return nil, errors.New("runtime-api session lifecycle requires identity exchange and its shared attempt cap")
	}
	if config.AuthorizationLifecycle != nil && config.SessionLifecycle == nil {
		return nil, errors.New("runtime-api authorization lifecycle requires the session lifecycle owner")
	}
	if config.AdmissionLimit == 0 {
		config.AdmissionLimit = DefaultAdmissionLimit
	}
	if config.AdmissionLimit < 0 || config.AdmissionLimit > MaxAdmissionLimit {
		return nil, fmt.Errorf(
			"runtime-api admission limit must be between 1 and %d",
			MaxAdmissionLimit,
		)
	}
	policyConfig := runtimegrpc.Config{
		Authority: config.WorkloadAuthority, AdmissionLimit: config.AdmissionLimit,
		AttemptCap: semantic.MaxAttemptDuration,
	}
	var policyOptions []runtimegrpc.PolicyOption
	if config.RecordAccessMaxAdmissionLease != 0 {
		policyOptions = append(policyOptions, runtimegrpc.WithRecordAccessMaxAdmissionLease(config.RecordAccessMaxAdmissionLease))
	}
	if config.QueryMaxAdmissionLease != 0 {
		if config.QueryEngine == nil {
			return nil, errors.New("runtime-api query admission requires the query engine")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithQueryMaxAdmissionLease(config.QueryMaxAdmissionLease))
	}
	if config.RelatedCountMaxAdmissionLease != 0 {
		if config.QueryEngine == nil {
			return nil, errors.New("runtime-api related-count admission requires the query engine")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithRelatedCountMaxAdmissionLease(config.RelatedCountMaxAdmissionLease))
	}
	if config.CreateDefaultPreviewMaxAdmissionLease != 0 {
		policyOptions = append(policyOptions, runtimegrpc.WithCreateDefaultPreviewMaxAdmissionLease(config.CreateDefaultPreviewMaxAdmissionLease))
	}
	if config.RecycleBinMaxAdmissionLease != 0 {
		if config.RecycleBinLister == nil {
			return nil, errors.New("runtime-api recycle-bin admission requires the recycle-bin lister")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithRecycleBinMaxAdmissionLease(config.RecycleBinMaxAdmissionLease))
	}
	if config.RecycleBinMutationMaxAdmissionLease != 0 {
		if config.RecycleBinMutator == nil {
			return nil, errors.New("runtime-api recycle-bin mutation admission requires the recycle-bin mutator")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithRecycleBinMutationMaxAdmissionLease(config.RecycleBinMutationMaxAdmissionLease))
	}
	if config.RecordViewMaxAdmissionLease != 0 {
		if config.RecordView == nil {
			return nil, errors.New("runtime-api record-view admission requires the record-view engine")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithRecordViewMaxAdmissionLease(config.RecordViewMaxAdmissionLease))
	}
	if config.AttachmentSourceMaxAdmissionLease != 0 {
		if config.AttachmentSources == nil {
			return nil, errors.New("runtime-api attachment-source admission requires the attachment source engine")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithAttachmentSourceMaxAdmissionLease(config.AttachmentSourceMaxAdmissionLease))
	}
	if (config.RecordMutation == nil) != (config.RecordMutationAttemptCap == 0) {
		return nil, errors.New("runtime-api record-mutation executor and attempt cap must be configured together")
	}
	if config.RecordMutation != nil {
		policyOptions = append(policyOptions, runtimegrpc.WithRecordMutationAttemptCap(config.RecordMutationAttemptCap))
	}
	for capability, maxLease := range config.PlatformCapabilityMaxAdmissionLeases {
		policyOptions = append(policyOptions, runtimegrpc.WithPlatformCapabilityMaxAdmissionLease(capability, maxLease))
	}
	requiredResourceClasses, err := platformRequiredResourceClasses(config.PlatformCapabilityMaxAdmissionLeases)
	if err != nil {
		return nil, err
	}
	// Outbound intents predate their attachment stream. The capability remains
	// short-unary compatible, while enabling the additive stream server requires
	// an explicit FileStream pool.
	if config.OutboundMailAttachments != nil {
		requiredResourceClasses[runtimegrpc.ResourceClassFileStream] = struct{}{}
	}
	for class, limit := range config.PlatformResourceClassAdmissionLimits {
		if _, required := requiredResourceClasses[class]; !required {
			return nil, errors.New("runtime-api resource-class admission cannot configure an unused Platform class")
		}
		policyOptions = append(policyOptions, runtimegrpc.WithResourceClassAdmissionLimit(class, limit))
	}
	for class := range requiredResourceClasses {
		if config.PlatformResourceClassAdmissionLimits[class] <= 0 {
			return nil, errors.New("runtime-api enabled Platform capability requires an explicit resource-class admission limit")
		}
	}
	var policy *runtimegrpc.Policy
	if config.IdentityExchange == nil {
		if len(policyOptions) == 0 {
			policy, err = runtimegrpc.NewPolicy(policyConfig)
		} else {
			policy, err = runtimegrpc.NewPolicyWithOptions(policyConfig, policyOptions...)
		}
	} else {
		if len(policyOptions) == 0 {
			policy, err = runtimegrpc.NewPolicyWithIdentity(policyConfig, config.IdentityAttemptCap)
		} else {
			policy, err = runtimegrpc.NewPolicyWithIdentityAndOptions(policyConfig, config.IdentityAttemptCap, policyOptions...)
		}
	}
	if err != nil {
		return nil, err
	}
	if config.SecurityAdmission == nil {
		return nil, errors.New("runtime-api security admission is required")
	}
	recordsServer, err := NewRecordsServer(reader, config.Receipts)
	if err != nil {
		return nil, err
	}
	recordsServer.metrics.pool = pool
	var queryServer *QueryServer
	if config.QueryMaxAdmissionLease > 0 {
		queryServer, err = NewQueryServer(config.QueryEngine, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var relatedCountServer *RelatedCountServer
	if config.RelatedCountMaxAdmissionLease > 0 {
		readAccess, ok := reader.(RecordReadAccessEvaluator)
		if !ok {
			return nil, errors.New("runtime-api: related-count read access evaluator is required")
		}
		relatedCountServer, err = NewRelatedCountServer(readAccess, config.QueryEngine, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var createDefaultPreviewServer *CreateDefaultPreviewServer
	if config.CreateDefaultPreviewMaxAdmissionLease > 0 {
		engine, ok := reader.(CreateDefaultPreviewEngine)
		if !ok {
			return nil, errors.New("runtime-api: create-default preview engine is required")
		}
		createDefaultPreviewServer, err = NewCreateDefaultPreviewServer(engine, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var recycleBinServer *RecycleBinServer
	if config.RecycleBinMaxAdmissionLease > 0 {
		recycleBinServer, err = NewRecycleBinServer(config.RecycleBinLister, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var recycleBinMutationServer *RecycleBinMutationServer
	if config.RecycleBinMutationMaxAdmissionLease > 0 {
		recycleBinMutationServer, err = NewRecycleBinMutationServer(config.RecycleBinMutator, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var recordViewServer *RecordViewServer
	if config.RecordViewMaxAdmissionLease > 0 {
		recordViewServer, err = NewRecordViewServer(config.RecordView, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var attachmentSourceServer *AttachmentSourceServer
	if config.AttachmentSourceMaxAdmissionLease > 0 {
		attachmentSourceServer, err = NewAttachmentSourceServer(config.AttachmentSources, config.Receipts, recordsServer.metrics)
		if err != nil {
			return nil, err
		}
	}
	var mutationServer *RecordMutationsServer
	if config.RecordMutation != nil {
		mutationServer, err = NewRecordMutationsServer(config.RecordMutation)
		if err != nil {
			return nil, err
		}
	}
	var permissionAuthoringServer *PermissionAuthoringServer
	if config.PlatformCapabilityMaxAdmissionLeases[permissionauthoringsemantic.CapabilityName] > 0 {
		permissionAuthoringServer, err = NewPermissionAuthoringServer(config.PermissionAuthoring)
		if err != nil {
			return nil, err
		}
	} else if config.PermissionAuthoring != nil || config.PermissionAuthoringOutcomeRetention != 0 {
		return nil, errors.New("runtime-api permission authoring engine requires explicit admission enablement")
	}
	var applicationNavigationServer *ApplicationNavigationServer
	if config.PlatformCapabilityMaxAdmissionLeases[navigationsemantic.CapabilityName] > 0 {
		if config.ApplicationNavigationAdmission == nil {
			return nil, errors.New("navigation requires exact workload and application owner admission")
		}
		applicationNavigationServer, err = NewApplicationNavigationServer(config.ApplicationNavigation)
		if err != nil {
			return nil, err
		}
	} else if config.ApplicationNavigation != nil || config.ApplicationNavigationAdmission != nil {
		return nil, errors.New("navigation requires explicit admission enablement")
	}

	var catalogConsumerViewServer *CatalogConsumerViewServer
	var metadataAdminReadServer *MetadataAdminReadServer
	var layoutAuthoringV2Server *LayoutAuthoringV2Server
	var approvalNotificationFlowReadServer *ApprovalNotificationFlowReadServer
	var pageLayoutResolutionServer *PageLayoutResolutionServer
	if config.PlatformCapabilityMaxAdmissionLeases[pagelayoutsemantic.CapabilityName] > 0 {
		if config.PageLayoutConsumerAdmission == nil {
			return nil, errors.New(
				"runtime-api page-layout resolution requires exact consumer admission",
			)
		}
		pageLayoutResolutionServer, err = NewPageLayoutResolutionServer(
			config.PageLayouts, config.PageLayoutRoleHierarchy,
		)
		if err != nil {
			return nil, err
		}
	} else if config.PageLayouts != nil || config.PageLayoutRoleHierarchy != nil || config.PageLayoutConsumerAdmission != nil {
		return nil, errors.New("runtime-api page-layout resolution requires explicit admission enablement")
	}
	if config.PlatformCapabilityMaxAdmissionLeases[metadataadminsemantic.CapabilityName] > 0 {
		if config.MetadataAdminReadAdmission == nil {
			return nil, errors.New("runtime-api metadata administration reads require exact workload admission")
		}
		metadataAdminReadServer, err = NewMetadataAdminReadServer(
			config.MetadataAdminReads, config.EffectivePermissions,
		)
		if err != nil {
			return nil, err
		}
	} else if config.MetadataAdminReads != nil || config.MetadataAdminReadAdmission != nil {
		return nil, errors.New("runtime-api metadata administration reads require explicit admission enablement")
	}
	if config.PlatformCapabilityMaxAdmissionLeases[layoutauthoringv2semantic.CapabilityName] > 0 {
		if config.LayoutAuthoringV2Admission == nil {
			return nil, errors.New("runtime-api Layout authoring v2 requires exact workload admission")
		}
		layoutAuthoringV2Server, err = NewLayoutAuthoringV2Server(
			config.LayoutAuthoringV2, config.EffectivePermissions,
		)
		if err != nil {
			return nil, err
		}
	} else if config.LayoutAuthoringV2 != nil || config.LayoutAuthoringV2Admission != nil {
		return nil, errors.New("runtime-api Layout authoring v2 requires explicit admission enablement")
	}
	if config.PlatformCapabilityMaxAdmissionLeases[approvalnotificationsemantic.CapabilityName] > 0 {
		if config.ApprovalNotificationFlowReadAdmission == nil {
			return nil, errors.New("runtime-api approval notification Flow reads require exact workload admission")
		}
		approvalNotificationFlowReadServer, err = NewApprovalNotificationFlowReadServer(
			config.ApprovalNotificationFlows,
		)
		if err != nil {
			return nil, err
		}
	} else if config.ApprovalNotificationFlows != nil || config.ApprovalNotificationFlowReadAdmission != nil {
		return nil, errors.New("runtime-api approval notification Flow reads require explicit admission enablement")
	}
	if config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.AppMetadataCapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.AppMetadataV2CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.AppMetadataV3CapabilityName] > 0 ||
		config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.IntegrationBindingCapabilityName] > 0 {
		if (config.CatalogConsumerAdmission == nil &&
			config.ServingAdmission == nil) ||
			config.CatalogConsumerViews == nil {
			return nil, errors.New("runtime-api Catalog consumer views require admission and the durable Catalog reader")
		}
		catalogConsumerViewServer, err = NewCatalogConsumerViewServer(config.CatalogConsumerViews)
		if err != nil {
			return nil, err
		}
	}
	var recordSearchServer *RecordSearchServer
	if config.PlatformCapabilityMaxAdmissionLeases[searchsemantic.CapabilityName] > 0 {
		if config.SearchEngine == nil || config.SearchScopeOrder == nil {
			return nil, errors.New(
				"runtime-api search admission requires record and scope-order engines",
			)
		}
		recordSearchServer, err = NewRecordSearchServer(
			config.SearchEngine, config.SearchScopeOrder,
		)
		if err != nil {
			return nil, err
		}
	}
	var reportReadServer *ReportReadServer
	if config.PlatformCapabilityMaxAdmissionLeases[reportsemantic.CapabilityName] > 0 {
		if config.ReportEngine == nil {
			return nil, errors.New("runtime-api report admission requires the report engine")
		}
		reportReadServer, err = NewReportReadServer(config.ReportEngine)
		if err != nil {
			return nil, err
		}
	}
	var privateListViewServer *PrivateListViewServer
	if config.PlatformCapabilityMaxAdmissionLeases[listviewsemantic.CapabilityName] > 0 {
		privateListViewServer, err = NewPrivateListViewServer(
			config.PrivateListViews, config.PrivateListViewAuthoring, config.PrivateListViewPermissions,
		)
		if err != nil {
			return nil, err
		}
	} else if config.PrivateListViews != nil || config.PrivateListViewAuthoring != nil || config.PrivateListViewPermissions != nil {
		return nil, errors.New("runtime-api private list views require explicit admission enablement")
	}
	var authorizedFilesServer *AuthorizedFilesServer
	if config.PlatformCapabilityMaxAdmissionLeases[filessemantic.CapabilityName] > 0 {
		if config.FileEngine == nil {
			return nil, errors.New("runtime-api files admission requires the file engine")
		}
		authorizedFilesServer, err = NewAuthorizedFilesServer(config.FileEngine)
		if err != nil {
			return nil, err
		}
	}
	var bulkIngestServer *BulkIngestServer
	if config.PlatformCapabilityMaxAdmissionLeases[bulkingestsemantic.CapabilityName] > 0 {
		if config.BulkIngest == nil {
			return nil, errors.New("runtime-api bulk ingest admission requires the durable bulk ingest engine")
		}
		bulkIngestServer, err = NewBulkIngestServer(config.BulkIngest)
		if err != nil {
			return nil, err
		}
	} else if config.BulkIngest != nil {
		return nil, errors.New("runtime-api bulk ingest engine requires explicit admission enablement")
	}
	var commandAcceptanceServer *CommandAcceptanceServer
	if config.PlatformCapabilityMaxAdmissionLeases[commandsemantic.CapabilityName] > 0 {
		if config.CommandEngine == nil {
			return nil, errors.New("runtime-api command admission requires the durable command engine")
		}
		commandAcceptanceServer, err = NewCommandAcceptanceServer(config.CommandEngine)
		if err != nil {
			return nil, err
		}
	}
	var applicationResultServer *ApplicationResultServer
	if config.PlatformCapabilityMaxAdmissionLeases[applicationresultsemantic.CapabilityName] > 0 {
		if config.ApplicationResults == nil || config.ApplicationResultAdmission == nil {
			return nil, errors.New(
				"runtime-api application-result admission requires the durable result engine and scope authority",
			)
		}
		applicationResultServer, err = NewApplicationResultServer(
			config.ApplicationResults,
		)
		if err != nil {
			return nil, err
		}
	} else if config.ApplicationResults != nil || config.ApplicationResultAdmission != nil {
		return nil, errors.New(
			"runtime-api application-result authority requires explicit admission enablement",
		)
	}
	var outboundIntentServer *OutboundIntentServer
	if config.PlatformCapabilityMaxAdmissionLeases[outboundintentsemantic.CapabilityName] > 0 {
		if config.OutboundIntentAdmission == nil {
			return nil, errors.New(
				"runtime-api outbound intent service requires exact Integration consumer admission",
			)
		}
		var attachmentEngines []FrozenOutboundMailAttachmentEngine
		if config.OutboundMailAttachments != nil {
			attachmentEngines = append(attachmentEngines, config.OutboundMailAttachments)
		}
		outboundIntentServer, err = NewOutboundIntentServer(
			config.OutboundIntents,
			config.OutboundResultCommitter,
			config.OutboundIntentTransactions,
			attachmentEngines...,
		)
		if err != nil {
			return nil, err
		}
	} else if config.OutboundIntentAdmission != nil || config.OutboundIntents != nil ||
		config.OutboundResultCommitter != nil || config.OutboundIntentTransactions != nil ||
		config.OutboundMailAttachments != nil {
		return nil, errors.New(
			"runtime-api outbound intent authority requires explicit admission enablement",
		)
	}
	var mailboxAuthorizationServer *MailboxAuthorizationServer
	if config.PlatformCapabilityMaxAdmissionLeases[mailboxauthorizationsemantic.CapabilityName] > 0 {
		if config.MailboxAuthorization == nil ||
			config.MailboxAuthorization.Admission == nil {
			return nil, errors.New(
				"runtime-api mailbox authorization service requires exact Integration consumer admission",
			)
		}
		mailboxAuthorizationServer, err = NewMailboxAuthorizationServer(
			config.MailboxAuthorization.Handles,
			config.MailboxAuthorization.Permissions,
			config.MailboxAuthorization.Sessions,
			config.MailboxAuthorization.RequiredPermission,
		)
		if err != nil {
			return nil, err
		}
	} else if config.MailboxAuthorization != nil {
		return nil, errors.New(
			"runtime-api mailbox authorization authority requires explicit admission enablement",
		)
	}
	var flowExecutionServer *FlowExecutionServer
	if config.PlatformCapabilityMaxAdmissionLeases[flowsemantic.CapabilityName] > 0 {
		var flowMetadataAdmission flowAppMetadataAdmission = config.CatalogConsumerAdmission
		if config.ServingAdmission != nil && config.ServingAdmission.hasFlowAppMetadataAuthority() {
			flowMetadataAdmission = config.ServingAdmission
		}
		if flowMetadataAdmission == nil {
			return nil, errors.New(
				"runtime-api Flow execution requires workload-scoped Catalog consumer admission",
			)
		}
		flowExecutionServer, err = NewFlowExecutionServer(
			config.FlowEngine,
			config.FlowInvocations,
			config.FlowTransactions,
			config.CatalogConsumerViews,
			config.EffectivePermissions,
			flowMetadataAdmission,
		)
		if err != nil {
			return nil, err
		}
	}
	var approvalServer *ApprovalServer
	approvalExecuteEnabled := config.PlatformCapabilityMaxAdmissionLeases[approvalsemantic.ExecuteCapabilityName] > 0
	approvalReadEnabled := config.PlatformCapabilityMaxAdmissionLeases[approvalsemantic.ReadCapabilityName] > 0
	if approvalExecuteEnabled || approvalReadEnabled {
		approvalServer, err = NewApprovalServer(
			config.ApprovalEngine, config.ApprovalInvocations, config.ApprovalTransactions,
			config.ApprovalPermissionFence,
			approvalExecuteEnabled, approvalReadEnabled,
		)
		if err != nil {
			return nil, err
		}
	}
	var effectivePermissionServer *EffectivePermissionServer
	if config.PlatformCapabilityMaxAdmissionLeases[permissionsemantic.CapabilityName] > 0 {
		if config.EffectivePermissions == nil {
			return nil, errors.New(
				"runtime-api effective-permission admission requires the permission projection engine",
			)
		}
		effectivePermissionServer, err = NewEffectivePermissionServer(
			config.EffectivePermissions,
		)
		if err != nil {
			return nil, err
		}
	}
	var auditLogReadServer *AuditLogReadServer
	if config.PlatformCapabilityMaxAdmissionLeases[auditsemantic.CapabilityName] > 0 {
		auditLogReadServer, err = NewAuditLogReadServer(config.AuditLog, config.EffectivePermissions)
		if err != nil {
			return nil, err
		}
	}
	if recordViewServer != nil {
		recordViewServer.permissions = effectivePermissionServer
	}
	var sessionLifecycleServer *SessionLifecycleServer
	if config.SessionLifecycle != nil {
		sessionLifecycleServer, err = NewSessionLifecycleServer(config.SessionLifecycle)
		if err != nil {
			return nil, err
		}
	}
	var authorizationLifecycleServer *AuthorizationLifecycleServer
	if config.AuthorizationLifecycle != nil {
		authorizationLifecycleServer, err = NewAuthorizationLifecycleServer(config.AuthorizationLifecycle)
		if err != nil {
			return nil, err
		}
	}
	var federatedLoginTransactionServer *FederatedLoginTransactionServer
	if config.FederatedLoginTransactions != nil {
		federatedLoginTransactionServer, err = NewFederatedLoginTransactionServer(config.FederatedLoginTransactions)
		if err != nil {
			return nil, err
		}
	}
	interceptors := []grpc.UnaryServerInterceptor{
		runtimeSQLDiagnosticsInterceptor(config.SQLDiagnosticsEnabled, recordsServer.metrics),
		recordsServer.metrics.sqlObservationInterceptor(),
		config.SecurityAdmission.UnaryServerInterceptor(),
		recordsServer.metrics.postSecurityObservationInterceptor(),
	}
	if config.CatalogConsumerAdmission != nil {
		interceptors = append(
			interceptors,
			config.CatalogConsumerAdmission.UnaryServerInterceptor(),
		)
	}
	if config.CommandConsumerAdmission != nil {
		interceptors = append(
			interceptors,
			config.CommandConsumerAdmission.UnaryServerInterceptor(),
		)
	}
	if config.PageLayoutConsumerAdmission != nil {
		interceptors = append(
			interceptors,
			config.PageLayoutConsumerAdmission.UnaryServerInterceptor(),
		)
	}
	if config.OutboundIntentAdmission != nil {
		interceptors = append(
			interceptors,
			config.OutboundIntentAdmission.UnaryServerInterceptor(),
		)
	}
	if config.ApplicationResultAdmission != nil {
		interceptors = append(
			interceptors,
			config.ApplicationResultAdmission.UnaryServerInterceptor(),
		)
	}
	if config.MailboxAuthorization != nil {
		interceptors = append(
			interceptors,
			config.MailboxAuthorization.Admission.UnaryServerInterceptor(),
		)
	}
	if config.ApplicationNavigationAdmission != nil {
		interceptors = append(interceptors, config.ApplicationNavigationAdmission.UnaryServerInterceptor())
	}

	if config.MetadataAdminReadAdmission != nil {
		interceptors = append(
			interceptors,
			config.MetadataAdminReadAdmission.UnaryServerInterceptor(),
		)
	}
	if config.LayoutAuthoringV2Admission != nil {
		interceptors = append(
			interceptors,
			config.LayoutAuthoringV2Admission.UnaryServerInterceptor(),
		)
	}
	if config.ApprovalNotificationFlowReadAdmission != nil {
		interceptors = append(
			interceptors,
			config.ApprovalNotificationFlowReadAdmission.UnaryServerInterceptor(),
		)
	}
	if config.ServingAdmission != nil {
		interceptors = append(interceptors, config.ServingAdmission.UnaryServerInterceptor())
	}
	interceptors = append(interceptors, runtimeRequestContextInterceptor())
	interceptors = append(interceptors, runtimeHandlerDiagnosticsInterceptor(config.SQLDiagnosticsEnabled))
	streamInterceptors := []grpc.StreamServerInterceptor{
		runtimeSQLDiagnosticsStreamInterceptor(config.SQLDiagnosticsEnabled, recordsServer.metrics),
		recordsServer.metrics.sqlObservationStreamInterceptor(),
		config.SecurityAdmission.StreamServerInterceptor(),
		recordsServer.metrics.postSecurityObservationStreamInterceptor(),
	}
	if config.CatalogConsumerAdmission != nil {
		streamInterceptors = append(
			streamInterceptors,
			config.CatalogConsumerAdmission.StreamServerInterceptor(),
		)
	}
	if config.OutboundIntentAdmission != nil {
		streamInterceptors = append(
			streamInterceptors,
			config.OutboundIntentAdmission.StreamServerInterceptor(),
		)
	}
	if config.ServingAdmission != nil {
		streamInterceptors = append(streamInterceptors, config.ServingAdmission.StreamServerInterceptor())
	}
	streamInterceptors = append(streamInterceptors, runtimeRequestContextStreamInterceptor())
	streamInterceptors = append(streamInterceptors, runtimeHandlerDiagnosticsStreamInterceptor(config.SQLDiagnosticsEnabled))
	serverOptions := append(policy.ServerOptions(),
		grpc.ChainUnaryInterceptor(interceptors...),
		grpc.ChainStreamInterceptor(streamInterceptors...),
	)
	if config.TransportCredentials != nil {
		serverOptions = append(serverOptions, grpc.Creds(config.TransportCredentials))
	}
	grpcServer := grpc.NewServer(serverOptions...)
	runtimev1.RegisterRecordsServiceServer(grpcServer, recordsServer)
	if queryServer != nil {
		runtimev1.RegisterRecordQueryServiceServer(grpcServer, queryServer)
	}
	if relatedCountServer != nil {
		runtimev1.RegisterRelatedRecordCountServiceServer(grpcServer, relatedCountServer)
	}
	if createDefaultPreviewServer != nil {
		runtimev1.RegisterRecordCreateDefaultPreviewServiceServer(grpcServer, createDefaultPreviewServer)
	}
	if recycleBinServer != nil {
		runtimev1.RegisterRecycleBinReadServiceServer(grpcServer, recycleBinServer)
	}
	if recycleBinMutationServer != nil {
		runtimev1.RegisterRecycleBinMutationServiceServer(grpcServer, recycleBinMutationServer)
	}
	if recordViewServer != nil {
		runtimev1.RegisterRecordViewReadServiceServer(grpcServer, recordViewServer)
	}
	if attachmentSourceServer != nil {
		runtimev1.RegisterRecordAttachmentSourceResolutionServiceServer(grpcServer, attachmentSourceServer)
	}
	if mutationServer != nil {
		runtimev1.RegisterRecordMutationsServiceServer(grpcServer, mutationServer)
	}
	if permissionAuthoringServer != nil {
		runtimev1.RegisterPermissionAuthoringServiceServer(grpcServer, permissionAuthoringServer)
	}
	if catalogConsumerViewServer != nil {
		catalogv1.RegisterConsumerViewServiceServer(grpcServer, catalogConsumerViewServer)
		catalogv1.RegisterAppMetadataViewV3ServiceServer(grpcServer, catalogConsumerViewServer)
	}
	if applicationNavigationServer != nil {
		runtimev1.RegisterApplicationNavigationServiceServer(grpcServer, applicationNavigationServer)
	}

	if metadataAdminReadServer != nil {
		runtimev1.RegisterMetadataAdminReadServiceServer(grpcServer, metadataAdminReadServer)
	}
	if layoutAuthoringV2Server != nil {
		runtimev1.RegisterLayoutAuthoringV2ServiceServer(grpcServer, layoutAuthoringV2Server)
	}
	if approvalNotificationFlowReadServer != nil {
		runtimev1.RegisterApprovalNotificationFlowReadServiceServer(grpcServer, approvalNotificationFlowReadServer)
	}
	if pageLayoutResolutionServer != nil {
		runtimev1.RegisterPageLayoutResolutionServiceServer(grpcServer, pageLayoutResolutionServer)
	}
	if recordSearchServer != nil {
		runtimev1.RegisterRecordSearchServiceServer(grpcServer, recordSearchServer)
	}
	if reportReadServer != nil {
		runtimev1.RegisterReportReadServiceServer(grpcServer, reportReadServer)
	}
	if auditLogReadServer != nil {
		runtimev1.RegisterAuditLogReadServiceServer(grpcServer, auditLogReadServer)
	}
	if privateListViewServer != nil {
		runtimev1.RegisterPrivateListViewServiceServer(grpcServer, privateListViewServer)
	}
	if authorizedFilesServer != nil {
		runtimev1.RegisterAuthorizedFilesServiceServer(grpcServer, authorizedFilesServer)
	}
	if bulkIngestServer != nil {
		runtimev1.RegisterBulkIngestServiceServer(grpcServer, bulkIngestServer)
	}
	if commandAcceptanceServer != nil {
		runtimev1.RegisterCommandAcceptanceServiceServer(grpcServer, commandAcceptanceServer)
	}
	if applicationResultServer != nil {
		runtimev1.RegisterApplicationResultServiceServer(
			grpcServer,
			applicationResultServer,
		)
	}
	if outboundIntentServer != nil {
		runtimev1.RegisterOutboundIntentServiceServer(grpcServer, outboundIntentServer)
	}
	if mailboxAuthorizationServer != nil {
		runtimev1.RegisterConnectedMailboxAuthorizationServiceServer(
			grpcServer,
			mailboxAuthorizationServer,
		)
	}
	if flowExecutionServer != nil {
		runtimev1.RegisterFlowExecutionServiceServer(grpcServer, flowExecutionServer)
	}
	if approvalServer != nil {
		runtimev1.RegisterApprovalServiceServer(grpcServer, approvalServer)
	}
	if effectivePermissionServer != nil {
		runtimev1.RegisterEffectivePermissionServiceServer(
			grpcServer, effectivePermissionServer,
		)
	}
	if sessionLifecycleServer != nil {
		runtimev1.RegisterSessionLifecycleServiceServer(grpcServer, sessionLifecycleServer)
	}
	if authorizationLifecycleServer != nil {
		runtimev1.RegisterAuthorizationLifecycleServiceServer(grpcServer, authorizationLifecycleServer)
	}
	if federatedLoginTransactionServer != nil {
		runtimev1.RegisterFederatedLoginTransactionServiceServer(grpcServer, federatedLoginTransactionServer)
	}
	capabilities := []string{}
	if config.ServingAdmission == nil || config.Receipts != nil {
		capabilities = append(capabilities, semantic.CapabilityName)
	}
	if config.RecordAccessMaxAdmissionLease > 0 {
		capabilities = append(capabilities, accesssemantic.CapabilityName)
	}
	if config.QueryMaxAdmissionLease > 0 {
		capabilities = append(capabilities, querysemantic.CapabilityName)
	}
	if config.RelatedCountMaxAdmissionLease > 0 {
		capabilities = append(capabilities, relatedsemantic.CapabilityName)
	}
	if config.CreateDefaultPreviewMaxAdmissionLease > 0 {
		capabilities = append(capabilities, defaultsemantic.CapabilityName)
	}
	if config.RecycleBinMaxAdmissionLease > 0 {
		capabilities = append(capabilities, recyclesemantic.CapabilityName)
	}
	if config.RecycleBinMutationMaxAdmissionLease > 0 {
		capabilities = append(capabilities, recyclemutationsemantic.CapabilityName)
	}
	if config.RecordViewMaxAdmissionLease > 0 {
		capabilities = append(capabilities, recordviewsemantic.CapabilityName)
	}
	if config.AttachmentSourceMaxAdmissionLease > 0 {
		capabilities = append(capabilities, attachmentsemantic.CapabilityName)
	}
	if config.RecordMutation != nil {
		capabilities = append(capabilities, mutationsemantic.CapabilityName)
	}
	if permissionAuthoringServer != nil {
		capabilities = append(capabilities, permissionauthoringsemantic.CapabilityName)
	}
	if approvalNotificationFlowReadServer != nil {
		capabilities = append(capabilities, approvalnotificationsemantic.CapabilityName)
	}
	if pageLayoutResolutionServer != nil {
		capabilities = append(capabilities, pagelayoutsemantic.CapabilityName)
	}
	if catalogConsumerViewServer != nil {
		if config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.AppMetadataCapabilityName] > 0 {
			capabilities = append(capabilities, catalogsemantic.AppMetadataCapabilityName)
		}
		if config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.AppMetadataV2CapabilityName] > 0 {
			capabilities = append(capabilities, catalogsemantic.AppMetadataV2CapabilityName)
		}
		if config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.AppMetadataV3CapabilityName] > 0 {
			capabilities = append(capabilities, catalogsemantic.AppMetadataV3CapabilityName)
		}
		if config.PlatformCapabilityMaxAdmissionLeases[catalogsemantic.IntegrationBindingCapabilityName] > 0 {
			capabilities = append(capabilities, catalogsemantic.IntegrationBindingCapabilityName)
		}
	}
	if layoutAuthoringV2Server != nil {
		capabilities = append(capabilities, layoutauthoringv2semantic.CapabilityName)
	}
	if applicationNavigationServer != nil {
		capabilities = append(capabilities, navigationsemantic.CapabilityName)
	}

	if metadataAdminReadServer != nil {
		capabilities = append(capabilities, metadataadminsemantic.CapabilityName)
	}
	if recordSearchServer != nil {
		capabilities = append(capabilities, searchsemantic.CapabilityName)
	}
	if reportReadServer != nil {
		capabilities = append(capabilities, reportsemantic.CapabilityName)
	}
	if auditLogReadServer != nil {
		capabilities = append(capabilities, auditsemantic.CapabilityName)
	}
	if privateListViewServer != nil {
		capabilities = append(capabilities, listviewsemantic.CapabilityName)
	}
	if authorizedFilesServer != nil {
		capabilities = append(capabilities, filessemantic.CapabilityName)
	}
	if bulkIngestServer != nil {
		capabilities = append(capabilities, bulkingestsemantic.CapabilityName)
	}
	if commandAcceptanceServer != nil {
		capabilities = append(capabilities, commandsemantic.CapabilityName)
	}
	if applicationResultServer != nil {
		capabilities = append(
			capabilities,
			applicationresultsemantic.CapabilityName,
		)
	}
	if outboundIntentServer != nil {
		capabilities = append(capabilities, outboundintentsemantic.CapabilityName)
	}
	if mailboxAuthorizationServer != nil {
		capabilities = append(capabilities, mailboxauthorizationsemantic.CapabilityName)
	}
	if flowExecutionServer != nil {
		capabilities = append(capabilities, flowsemantic.CapabilityName)
	}
	if approvalExecuteEnabled {
		capabilities = append(capabilities, approvalsemantic.ExecuteCapabilityName)
	}
	if approvalReadEnabled {
		capabilities = append(capabilities, approvalsemantic.ReadCapabilityName)
	}
	if effectivePermissionServer != nil {
		capabilities = append(
			capabilities, permissionsemantic.CapabilityName,
		)
	}
	if config.IdentityExchange != nil {
		identityServer, err := NewIdentityServer(config.IdentityExchange)
		if err != nil {
			return nil, err
		}
		runtimev1.RegisterIdentityContextServiceServer(grpcServer, identityServer)
		capabilities = append(capabilities, identitysemantic.CapabilityName)
	}
	if sessionLifecycleServer != nil {
		capabilities = append(capabilities, sessionsemantic.CapabilityName)
	}
	capabilities = append(capabilities, config.PermissionStateCapabilities...)
	metricsCollectors := append(
		backendmiddleware.CredentialScopeInventoryCollectors(),
		config.AdditionalMetricsCollectors...,
	)
	if effectivePermissionServer != nil {
		metricsCollectors = append(metricsCollectors, effectivePermissionServer)
	}
	return runtimeprocess.Assemble(runtimeprocess.Config{
		Role: "runtime-api", Version: config.Version,
		AdminAddress: config.AdminAddress, GRPCAddress: config.GRPCAddress, GRPCServer: grpcServer,
		Capabilities: capabilities,
		Ready: func() bool {
			servingReady := config.ServingAdmission == nil || config.ServingAdmission.Ready()
			consumerReady := config.CatalogConsumerAdmission == nil ||
				config.CatalogConsumerAdmission.Ready()
			permissionStateReady := config.PermissionStateReady == nil ||
				config.PermissionStateReady()
			return config.ServingEnabled && policy.Ready() &&
				config.SecurityAdmission.Ready() && servingReady && consumerReady &&
				permissionStateReady
		},
		BeginDrain:       policy.BeginDrain,
		MetricsCollector: recordsServer.metrics,
		// This process serves a private registry, so the empty-scope inventory counters are
		// invisible here unless handed over. The Runtime admissions this process performs are
		// exactly the ones an operator would otherwise conclude do not exist.
		AdditionalMetricsCollectors: metricsCollectors,
		Controller:                  config.ServingController,
		// Test2 deliberately binds its admin listener to loopback, where bounded
		// CPU/heap/goroutine capture is safe without continuous contention
		// sampling. Generic Kubernetes binds probes and metrics to the pod network;
		// keep pprof absent there instead of making the whole process fail startup.
		ExposeDiagnostics: runtimeprocess.IsLoopbackListenAddress(config.AdminAddress),
		EnableProfiling:   config.ProfilingEnabled,
		Admission: func() runtimeprocess.AdmissionSnapshot {
			snapshot := policy.AdmissionSnapshot()
			return runtimeprocess.AdmissionSnapshot{
				Limit: snapshot.Limit, InFlight: snapshot.InFlight, Accepted: snapshot.Accepted,
				RejectedFull: snapshot.RejectedFull, RejectedDraining: snapshot.RejectedDraining,
			}
		},
	})
}

// runtimeSQLDiagnosticsInterceptor opts one Runtime RPC into the existing
// detailed engine timings plus driver-level SQL statement timings. It is
// independent from pprof and disabled by default. SQL arguments are never
// logged; the driver emits only the statement text/template that it receives.
func runtimeSQLDiagnosticsInterceptor(enabled bool, metrics *recordsMetrics) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		if !enabled {
			return handler(ctx, request)
		}
		if requestID, ok := runtimegrpc.IncomingRequestID(ctx); ok {
			ctx = context.WithValue(ctx, constants.ContextKeyRequestID, requestID)
		}
		ctx, sqlCounter := telemetry.WithSQLStatementCounter(perftiming.WithEnabled(ctx))
		poolBefore := runtimeDiagnosticsPoolStats(metrics)
		started := time.Now()
		response, err := handler(ctx, request)
		poolAfter := runtimeDiagnosticsPoolStats(metrics)
		operation := "unobserved"
		if observed, ok := runtimeMetricOperation(info); ok {
			operation = observed
		} else if info != nil && info.FullMethod != "" {
			operation = info.FullMethod
		}
		event := perftiming.Event(ctx, "runtime_rpc_timing", started, err).
			Str("operation", operation).
			Uint64("sql_statements", sqlCounter.Count()).
			Int64("sql_elapsed_ms", sqlCounter.TotalDuration().Milliseconds()).
			Int64("sql_max_ms", sqlCounter.MaxDuration().Milliseconds())
		logRuntimeDiagnosticsPoolDelta(event, poolBefore, poolAfter).Msg("Runtime RPC timing")
		return response, err
	}
}

func runtimeHandlerDiagnosticsInterceptor(enabled bool) grpc.UnaryServerInterceptor {
	return func(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		if !enabled {
			return handler(ctx, request)
		}
		ctx, sqlCounter := telemetry.WithSQLStatementCounter(ctx)
		started := time.Now()
		response, err := handler(ctx, request)
		operation := "unobserved"
		if observed, ok := runtimeMetricOperation(info); ok {
			operation = observed
		} else if info != nil && info.FullMethod != "" {
			operation = info.FullMethod
		}
		perftiming.Event(ctx, "runtime_handler_timing", started, err).
			Str("operation", operation).
			Uint64("sql_statements", sqlCounter.Count()).
			Int64("sql_elapsed_ms", sqlCounter.TotalDuration().Milliseconds()).
			Int64("sql_max_ms", sqlCounter.MaxDuration().Milliseconds()).
			Msg("Runtime handler timing")
		return response, err
	}
}

func runtimeSQLDiagnosticsStreamInterceptor(enabled bool, metrics *recordsMetrics) grpc.StreamServerInterceptor {
	return func(server any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if !enabled || stream == nil {
			return handler(server, stream)
		}
		ctx := stream.Context()
		if requestID, ok := runtimegrpc.IncomingRequestID(ctx); ok {
			ctx = context.WithValue(ctx, constants.ContextKeyRequestID, requestID)
		}
		ctx, sqlCounter := telemetry.WithSQLStatementCounter(perftiming.WithEnabled(ctx))
		poolBefore := runtimeDiagnosticsPoolStats(metrics)
		started := time.Now()
		operation := "unobserved"
		if info != nil {
			if observed, ok := runtimeMetricOperationName(info.FullMethod); ok {
				operation = observed
			} else if info.FullMethod != "" {
				operation = info.FullMethod
			}
		}
		err := handler(server, &runtimeContextServerStream{ServerStream: stream, ctx: ctx})
		poolAfter := runtimeDiagnosticsPoolStats(metrics)
		event := perftiming.Event(ctx, "runtime_rpc_timing", started, err).
			Str("operation", operation).
			Uint64("sql_statements", sqlCounter.Count()).
			Int64("sql_elapsed_ms", sqlCounter.TotalDuration().Milliseconds()).
			Int64("sql_max_ms", sqlCounter.MaxDuration().Milliseconds())
		logRuntimeDiagnosticsPoolDelta(event, poolBefore, poolAfter).Msg("Runtime stream RPC timing")
		return err
	}
}

func runtimeDiagnosticsPoolStats(metrics *recordsMetrics) sql.DBStats {
	if metrics == nil || metrics.pool == nil {
		return sql.DBStats{}
	}
	return metrics.pool.Stats()
}

func logRuntimeDiagnosticsPoolDelta(event *zerolog.Event, before, after sql.DBStats) *zerolog.Event {
	if event == nil {
		return event
	}
	waitCount := after.WaitCount - before.WaitCount
	waitDuration := after.WaitDuration - before.WaitDuration
	return event.
		Int("db_pool_open_before", before.OpenConnections).
		Int("db_pool_open_after", after.OpenConnections).
		Int("db_pool_in_use_before", before.InUse).
		Int("db_pool_in_use_after", after.InUse).
		Int64("db_pool_wait_count_delta", waitCount).
		Int64("db_pool_wait_elapsed_ms_delta", waitDuration.Milliseconds())
}

func runtimeHandlerDiagnosticsStreamInterceptor(enabled bool) grpc.StreamServerInterceptor {
	return func(server any, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if !enabled || stream == nil {
			return handler(server, stream)
		}
		ctx, sqlCounter := telemetry.WithSQLStatementCounter(stream.Context())
		started := time.Now()
		operation := "unobserved"
		if info != nil {
			if observed, ok := runtimeMetricOperationName(info.FullMethod); ok {
				operation = observed
			} else if info.FullMethod != "" {
				operation = info.FullMethod
			}
		}
		err := handler(server, &runtimeContextServerStream{ServerStream: stream, ctx: ctx})
		perftiming.Event(ctx, "runtime_handler_timing", started, err).
			Str("operation", operation).
			Uint64("sql_statements", sqlCounter.Count()).
			Int64("sql_elapsed_ms", sqlCounter.TotalDuration().Milliseconds()).
			Int64("sql_max_ms", sqlCounter.MaxDuration().Milliseconds()).
			Msg("Runtime stream handler timing")
		return err
	}
}

func runtimeRequestContextInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, request any, _ *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		if requestID, ok := runtimegrpc.IncomingRequestID(ctx); ok {
			ctx = context.WithValue(ctx, constants.ContextKeyRequestID, requestID)
		}
		// The legacy HTTP host installs these caches once at request ingress.
		// Runtime is a separate ingress owner and must establish the same
		// request-local input snapshot before any records/query engine runs.
		ctx = permissions.WithAccessInputCache(ctx)
		ctx = records.EnsureRecordReadCache(ctx)
		return handler(ctx, request)
	}
}

func runtimeRequestContextStreamInterceptor() grpc.StreamServerInterceptor {
	return func(server any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if stream == nil {
			return errors.New("runtime-api stream context is unavailable")
		}
		ctx := stream.Context()
		if requestID, ok := runtimegrpc.IncomingRequestID(ctx); ok {
			ctx = context.WithValue(ctx, constants.ContextKeyRequestID, requestID)
		}
		ctx = permissions.WithAccessInputCache(ctx)
		ctx = records.EnsureRecordReadCache(ctx)
		return handler(server, &runtimeContextServerStream{ServerStream: stream, ctx: ctx})
	}
}

// RejectingAuthority is the explicit metadata-independent deployment
// authority. The default serving composition never references it.
func RejectingAuthority(context.Context, string) (runtimegrpc.VerifiedCaller, error) {
	return "", &semantic.Failure{Category: semantic.FailureUnauthenticated}
}

func ParseAdmissionLimit(raw string) (int, error) {
	return runtimeconfig.ParseAdmissionLimit(raw)
}

const ShutdownTimeout = 10 * time.Second