// Package runtimeworker owns independently deployed Runtime Worker production
// composition. It contains no L1 App, HTTP business route, or Runtime API
// server. Each registered work family owns its execution and recovery; metadata
// interpretation remains bound to immutable artifacts.
package runtimeworker

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/backend/internal/managedruntime"
	recordmutation "github.com/nexuscrm/backend/internal/runtimeassembly/recordmutation"
	"github.com/nexuscrm/backend/internal/runtimeconfig"
	"github.com/nexuscrm/backend/internal/runtimepermissionstate"
	"github.com/nexuscrm/backend/internal/runtimeprocess"
	database "github.com/nexuscrm/db"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/files"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/storage"
	sharedworker "github.com/nexuscrm/worker"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

var supportedPayloadWorkFamilies = map[string]struct{}{
	audit.AuditHistoryFanoutWorkFamily:                               {},
	persistence.RuntimeRecordCommandLifecycleWorkFamily:              {},
	persistence.RuntimePermissionAuthoringCommandLifecycleWorkFamily: {},
	files.RuntimeFileUploadLifecycleWorkFamily:                       {},
}

var supportedMetadataWorkFamilies = map[string]struct{}{
	eventfeature.RuntimeRecordEventDispatchWorkFamily:     {},
	records.RuntimeRollupDeltaWorkFamily:                  {},
	records.RuntimeBulkIngestWorkFamily:                   {},
	automation.RuntimeScheduledFlowWorkFamily:             {},
	persistence.RuntimePlatformCommandExecutionWorkFamily: {},
}

type permissionStateGateFactory func(
	context.Context,
	*sql.DB,
	runtimepermissionstate.Config,
) (*runtimepermissionstate.Gate, error)

type platformCommandExecutorRegistryFactory func(
	context.Context,
	*sql.DB,
	*recordmutation.RecordsMutationRuntime,
	string,
) (*platformCommandExecutorRegistry, error)

type productionAssemblyDependencies struct {
	permissionStateFactory          permissionStateGateFactory
	platformCommandExecutorRegistry platformCommandExecutorRegistryFactory
}

func bindRuntimeRecordEventOutboundIntents(
	graph *recordmutation.RecordsMutationRuntime,
	attachmentMetadata sharedcontract.RecordAttachmentMetadataAuthorizer,
	sink automation.RuntimeOutboundIntentSink,
) error {
	if graph == nil || attachmentMetadata == nil || sink == nil {
		return errors.New("runtime metadata worker: outbound intent dependency closure is incomplete")
	}
	if err := graph.InstallOutboundMailAttachmentMetadataAuthorizer(attachmentMetadata); err != nil {
		return err
	}
	return graph.Flows.EnableRuntimeOutboundIntents(sink)
}

func newProductionPlatformCommandExecutorRegistry(
	ctx context.Context,
	db *sql.DB,
	graph *recordmutation.RecordsMutationRuntime,
	runtimeCapabilityID string,
) (*platformCommandExecutorRegistry, error) {
	if ctx == nil || db == nil || graph == nil {
		return nil, errors.New("runtime platform command: composition inputs are required")
	}
	switch runtimeCapabilityID {
	case commandsemantic.CapabilityName:
		if graph.Flows == nil || graph.Persistence == nil {
			return nil, errors.New("runtime platform command: Flow engine and system actor loader are required")
		}
		return newPlatformCommandExecutorRegistry(
			map[string]map[uint32]PlatformCommandExecutor{
				RuntimeFlowCommandKind: {
					RuntimeFlowCommandPayloadFormatVersion: &runtimeFlowCommandExecutor{
						flows: graph.Flows,
						users: graph.Persistence,
					},
					RuntimeFlowCommandPayloadFormatVersionV2: &runtimeFlowCommandDependencyExecutor{
						flows: graph.Flows,
						users: graph.Persistence,
					},
				},
			},
			map[string]map[uint32]platformCommandApplicationExecutor{
				RuntimeFlowCommandKind: {
					RuntimeFlowCommandPayloadFormatVersionV3: &runtimeFlowCommandApplicationExecutor{
						flows: graph.Flows,
						users: graph.Persistence,
					},
				},
			},
		)
	default:
		return nil, fmt.Errorf("runtime platform command: unsupported capability %q", runtimeCapabilityID)
	}
}

func activateRuntimeConnectorInboundFinalization(
	graph *recordmutation.RecordsMutationRuntime,
	db *sql.DB,
) (platformCommandRecoveryOwner, error) {
	installed, store, err := activateRuntimeConnectorInboundIdentityExecution(graph, db)
	if err != nil {
		return nil, err
	}
	probe := automation.NewConnectorInboundRemoteIdentityRecoveryCandidateProbe(store, installed)
	graph.Flows.SetConnectorInboundRemoteIdentityRecoveryCandidateProbe(probe)
	return automation.NewConnectorInboundRemoteIdentityRecoveryService(
		graph.Persistence,
		store,
		installed,
		automation.NewConnectorInboundRemoteIdentityRecoverySourceValidator(
			graph.Persistence,
			automation.MetadataBoundConnectorInboundRemoteIdentityRecoveryValidator{},
		),
	), nil
}

// activateRuntimeConnectorInboundIdentityExecution wires the small
// no-provider-effect connector identity surface shared by the scheduled Flow
// and platform-command workers. Recovery remains platform-command-only, but a
// strict AsyncAfterCommit identity handoff executes in the scheduled worker
// and therefore must receive the exact same immutable resolver, durable store,
// and action registrations.
func activateRuntimeConnectorInboundIdentityExecution(
	graph *recordmutation.RecordsMutationRuntime,
	db *sql.DB,
) (runtimeConnectorInboundRecoveryInstalledResolver, *automation.SQLConnectorPushDurableStore, error) {
	if graph == nil || graph.Flows == nil || graph.Transactions == nil || db == nil {
		return runtimeConnectorInboundRecoveryInstalledResolver{}, nil, errors.New(
			"runtime connector inbound finalization: Flow graph, transaction owner and database are required",
		)
	}
	store := automation.NewSQLConnectorPushDurableStoreWithTxExtractor(db, graph.Transactions)
	installed := runtimeConnectorInboundRecoveryInstalledResolver{}
	graph.Flows.SetConnectorInboundRemoteIdentityFinalizationAuthorityResolver(
		runtimeConnectorInboundFinalizationAuthorityResolver{},
	)
	graph.Flows.SetConnectorInboundRemoteIdentityInstalledContractResolver(installed)
	graph.Flows.SetConnectorPushDurableStore(store)
	graph.Flows.RegisterFlowActionCalls(
		automation.ConnectorInboundRemoteIdentityFinalizationFlowActionCallRegistration(),
		automation.IdentityConvergenceDependencyFlowActionCallRegistration(),
		automation.ConnectorInboundRemoteIdentityRecoveryRequestFlowActionCallRegistration(),
	)
	return installed, store, nil
}

type scheduledRunTransactionOwner struct {
	records      *records.PersistenceService
	transactions *persistence.TransactionManager
}

func (o scheduledRunTransactionOwner) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	return o.records.RunInPessimisticTransactionWithOutcome(ctx, fn)
}

func (o scheduledRunTransactionOwner) ExtractTx(ctx context.Context) interface{} {
	return o.transactions.ExtractTx(ctx)
}

// AssembleProduction binds exactly one work family to the deployment fence
// already activated in _System_RuntimeWorkOwnership. Metadata-free roles use
// only their payload store; metadata-bound roles reopen the exact immutable
// view carried by each row. Startup can verify authority but cannot create or
// transfer it.
func AssembleProduction(ctx context.Context, version string, config runtimeconfig.Worker) (*runtimeprocess.Process, *sql.DB, error) {
	return assembleProduction(ctx, version, config, runtimepermissionstate.NewGate)
}

func assembleProduction(
	ctx context.Context,
	version string,
	config runtimeconfig.Worker,
	permissionStateFactory permissionStateGateFactory,
) (*runtimeprocess.Process, *sql.DB, error) {
	return assembleProductionWithDependencies(
		ctx,
		version,
		config,
		productionAssemblyDependencies{
			permissionStateFactory:          permissionStateFactory,
			platformCommandExecutorRegistry: newProductionPlatformCommandExecutorRegistry,
		},
	)
}

func assembleProductionWithDependencies(
	ctx context.Context,
	version string,
	config runtimeconfig.Worker,
	dependencies productionAssemblyDependencies,
) (*runtimeprocess.Process, *sql.DB, error) {
	startup := newWorkerStartupObservation()
	defer startup.complete()
	if ctx == nil {
		return nil, nil, errors.New("runtime worker: startup context is required")
	}
	if dependencies.permissionStateFactory == nil {
		return nil, nil, errors.New("runtime worker: production dependencies are incomplete")
	}
	if err := config.Role.Validate(); err != nil {
		return nil, nil, err
	}
	if err := validateWorkerBinary(version, config.Role); err != nil {
		return nil, nil, err
	}
	_, payloadFamily := supportedPayloadWorkFamilies[config.Role.WorkFamily]
	_, metadataFamily := supportedMetadataWorkFamilies[config.Role.WorkFamily]
	if !payloadFamily && !metadataFamily {
		return nil, nil, fmt.Errorf("runtime worker: unsupported work family %q", config.Role.WorkFamily)
	}
	if config.Role.WorkFamily == persistence.RuntimePlatformCommandExecutionWorkFamily &&
		dependencies.platformCommandExecutorRegistry == nil {
		return nil, nil, errors.New(
			"runtime worker: Platform-command role requires a release-owned neutral executor registry",
		)
	}
	if payloadFamily &&
		(config.Role.Schema == runtimeconfig.WorkerConfigSchemaV2 || config.Role.HasMetadataScope()) {
		return nil, nil, errors.New("runtime worker: payload work family requires payload schema v1 or v3")
	}
	var metadataScope runtimeconfig.WorkerMetadataScope
	var timedFlowOwnerScope *runtimeconfig.WorkerMetadataScope
	if metadataFamily {
		var err error
		metadataScope, err = config.Role.MetadataScope()
		if err != nil {
			return nil, nil, fmt.Errorf("runtime worker: metadata work family requires an immutable-view scope: %w", err)
		}
		switch config.Role.WorkFamily {
		case automation.RuntimeScheduledFlowWorkFamily:
			if metadataScope.AppCapabilityID == "" ||
				metadataScope.RuntimeCapabilityID != flowsemantic.CapabilityName ||
				metadataScope.ExpectedCompatibilityReceiptDigest == "" {
				return nil, nil, errors.New("runtime worker: scheduled Flow role requires App capability selection, neutral Flow metadata scope, and exact compatibility receipt selection")
			}
		case persistence.RuntimePlatformCommandExecutionWorkFamily:
			if metadataScope.AppCapabilityID != "" ||
				metadataScope.RuntimeCapabilityID != commandsemantic.CapabilityName {
				return nil, nil, errors.New(
					"runtime worker: Platform-command role requires neutral command metadata scope without App capability selection",
				)
			}
		case records.RuntimeBulkIngestWorkFamily:
			if metadataScope.AppCapabilityID == "" ||
				metadataScope.RuntimeCapabilityID != mutationsemantic.CapabilityName ||
				metadataScope.ExpectedCompatibilityReceiptDigest == "" {
				return nil, nil, errors.New("runtime worker: Bulk ingest role requires App capability selection, neutral record-mutation metadata scope, and exact compatibility receipt selection")
			}
		default:
			if metadataScope.AppCapabilityID != "" {
				return nil, nil, fmt.Errorf(
					"runtime worker: work family %q does not own App capability selection",
					config.Role.WorkFamily,
				)
			}
		}
	}
	if config.TimedFlowOwner != nil {
		if config.Role.WorkFamily != eventfeature.RuntimeRecordEventDispatchWorkFamily {
			return nil, nil, errors.New("runtime worker: timed Flow owner assertion belongs only to record-event work")
		}
		scope, err := validateTimedFlowOwnerRole(config.Role, *config.TimedFlowOwner)
		if err != nil {
			return nil, nil, err
		}
		timedFlowOwnerScope = &scope
	}
	var processedRetention time.Duration
	var processedCleanupInterval time.Duration
	if config.Role.ProcessedRetention != "" {
		var parseErr error
		processedRetention, parseErr = time.ParseDuration(config.Role.ProcessedRetention)
		if parseErr != nil || processedRetention < time.Second {
			return nil, nil, errors.New("runtime worker: processed retention is invalid")
		}
	}
	if config.Role.ProcessedCleanupInterval != "" {
		var parseErr error
		processedCleanupInterval, parseErr = time.ParseDuration(config.Role.ProcessedCleanupInterval)
		if parseErr != nil || processedCleanupInterval < time.Second {
			return nil, nil, errors.New("runtime worker: processed cleanup interval is invalid")
		}
	}
	if config.Role.WorkFamily == eventfeature.RuntimeRecordEventDispatchWorkFamily {
		if processedRetention < time.Second || processedCleanupInterval < time.Second {
			return nil, nil, errors.New("runtime worker: record-event work requires explicit processed retention and cleanup interval")
		}
	} else if processedRetention != 0 || processedCleanupInterval != 0 {
		return nil, nil, fmt.Errorf("runtime worker: work family %q does not own processed-row retention", config.Role.WorkFamily)
	}
	pollInterval, err := time.ParseDuration(config.Role.PollInterval)
	if err != nil {
		return nil, nil, fmt.Errorf("runtime worker: parse poll interval: %w", err)
	}
	leaseDuration, err := time.ParseDuration(config.Role.LeaseDuration)
	if err != nil {
		return nil, nil, fmt.Errorf("runtime worker: parse lease duration: %w", err)
	}
	retryJitterMax, err := time.ParseDuration(config.Role.RetryJitterMax)
	if err != nil {
		return nil, nil, fmt.Errorf("runtime worker: parse retry jitter: %w", err)
	}
	ownershipHealthInterval := config.Role.EffectiveOwnershipHealthInterval()
	if metadataFamily {
		return assembleMetadataProduction(
			ctx, version, config, metadataScope, pollInterval, leaseDuration, retryJitterMax,
			processedRetention, processedCleanupInterval, timedFlowOwnerScope,
			ownershipHealthInterval,
			dependencies.permissionStateFactory,
			dependencies.platformCommandExecutorRegistry,
			startup,
		)
	}
	stageStarted := time.Now()
	connection, err := openWorkerConnection(ctx, productionWorkerPool(config.Role), "runtime worker")
	startup.observe(startupStageDatabase, stageStarted)
	if err != nil {
		return nil, nil, err
	}
	db := connection.DB()
	fail := func(cause error) (*runtimeprocess.Process, *sql.DB, error) {
		_ = db.Close()
		return nil, nil, cause
	}
	stageStarted = time.Now()
	permissionStateGate, err := newWorkerPermissionStateGate(
		ctx, db, version, config.Role, dependencies.permissionStateFactory,
	)
	startup.observe(startupStagePermissionState, stageStarted)
	if err != nil {
		return fail(fmt.Errorf("runtime worker: Permission State gate: %w", err))
	}
	ownership := configuredWorkOwnership(config.Role)
	authority, err := persistence.NewRuntimeWorkOwnershipRepository(db, ownership)
	if err != nil {
		return fail(err)
	}
	var controller func(context.Context) error
	var snapshot workerSnapshotFunc
	observationPools := &workerAuxiliaryPoolTracker{}
	stageStarted = time.Now()
	switch config.Role.WorkFamily {
	case audit.AuditHistoryFanoutWorkFamily:
		store, err := audit.ResolveFencedOutboxStore(db, ownership)
		if err != nil {
			return fail(err)
		}
		if err := store.CheckReady(ctx); err != nil {
			return fail(fmt.Errorf("runtime worker: audit-history readiness: %w", err))
		}
		worker := audit.NewOutboxWorker(store, audit.OutboxWorkerConfig{
			BatchSize: config.Role.BatchSize, PollInterval: pollInterval, LeaseDuration: leaseDuration,
			RetryJitterMax: retryJitterMax, MaxAttempts: config.Role.MaxAttempts, HolderID: "rw-" + uuid.NewString(),
		})
		runtimeWorker, err := newAuditOutboxRuntimeWorker(
			worker,
			newAuditOutboxObservationFactory(connection.OpenIndependentDB, ownership, observationPools),
		)
		if err != nil {
			return fail(err)
		}
		if err := runtimeWorker.Prepare(ctx); err != nil {
			return fail(fmt.Errorf("runtime worker: initial audit-history claim: %w", err))
		}
		controller = runtimeWorker.Run
		snapshot = runtimeWorker.snapshot
	case persistence.RuntimeRecordCommandLifecycleWorkFamily:
		store, err := persistence.NewRuntimeRecordCommandLifecycleStore(db, ownership)
		if err != nil {
			return fail(err)
		}
		commands := persistence.NewRuntimeRecordCommandRepository(db)
		lifecycle, err := records.NewRuntimeRecordCommandLifecycle(
			commands, persistence.NewTransactionManager(database.NewConnectionFromDB(db)), store,
		)
		if err != nil {
			return fail(err)
		}
		worker, err := newRecordCommandLifecycleWorker(
			&composedRecordCommandLifecycleStore{state: store, resolver: lifecycle},
			config.Role.BatchSize, pollInterval, retryJitterMax,
			newRecordCommandLifecycleObservationFactory(connection.OpenIndependentDB, ownership, observationPools),
		)
		if err != nil {
			return fail(err)
		}
		if err := worker.Prepare(ctx); err != nil {
			return fail(fmt.Errorf("runtime worker: record-command lifecycle readiness: %w", err))
		}
		controller = worker.Run
		snapshot = worker.snapshot
	case persistence.RuntimePermissionAuthoringCommandLifecycleWorkFamily:
		if err := persistence.VerifyPermissionAuthoringRuntimeSchema(ctx, db); err != nil {
			return fail(fmt.Errorf("runtime worker: permission-authoring runtime schema: %w", err))
		}
		store, err := persistence.NewPermissionAuthoringCommandLifecycleStore(db, ownership)
		if err != nil {
			return fail(err)
		}
		convergenceStore, err := persistence.NewFencedPermissionAuthoringConvergenceRepository(db, ownership)
		if err != nil {
			return fail(err)
		}
		readback, err := permissions.NewPermissionAuthoringReadbackConvergence(
			persistence.NewPermissionRepository(db),
		)
		if err != nil {
			return fail(err)
		}
		convergence, err := permissions.NewPermissionAuthoringConvergenceWorker(
			convergenceStore,
			readback,
			permissions.PermissionAuthoringConvergenceWorkerOptions{
				LeaseDuration: leaseDuration,
				RetryDelay:    retryJitterMax,
				BatchSize:     config.Role.BatchSize,
			},
		)
		if err != nil {
			return fail(err)
		}
		worker, err := newPermissionAuthoringCommandLifecycleWorker(
			store, convergence, "rw-"+uuid.NewString(),
			config.Role.BatchSize, pollInterval, retryJitterMax,
			newPermissionAuthoringCommandLifecycleObservationFactory(connection.OpenIndependentDB, ownership, observationPools),
		)
		if err != nil {
			return fail(err)
		}
		if err := worker.Prepare(ctx); err != nil {
			return fail(fmt.Errorf("runtime worker: permission-authoring-command lifecycle readiness: %w", err))
		}
		controller = worker.Run
		snapshot = worker.snapshot
	case files.RuntimeFileUploadLifecycleWorkFamily:
		objectStore, err := storage.NewStoreFromEnv()
		if err != nil {
			return fail(fmt.Errorf("runtime worker: file upload object store: %w", err))
		}
		lifecycle, err := files.NewRuntimeUploadLifecycle(db, objectStore, ownership)
		if err != nil {
			return fail(err)
		}
		worker, err := newFileUploadLifecycleWorker(
			lifecycle, config.Role.BatchSize, pollInterval, leaseDuration, retryJitterMax,
			newFileUploadLifecycleObservationFactory(connection.OpenIndependentDB, observationPools),
		)
		if err != nil {
			return fail(err)
		}
		if err := worker.Prepare(ctx); err != nil {
			return fail(fmt.Errorf("runtime worker: file upload lifecycle readiness: %w", err))
		}
		controller = worker.Run
		snapshot = worker.snapshot
	}
	startup.observe(startupStageRolePrepare, stageStarted)
	guardedController, err := newOwnershipController(authority, controller, ownershipHealthInterval)
	if err != nil {
		return fail(err)
	}
	capabilities, ready, err := workerPermissionStateProcessInputs(
		config.Role.WorkFamily,
		permissionStateGate,
		guardedController.Ready,
	)
	if err != nil {
		return fail(err)
	}
	process, err := runtimeprocess.Assemble(runtimeprocess.Config{
		Role: "runtime-worker", Version: version, AdminAddress: config.AdminAddress,
		Capabilities: capabilities,
		Ready:        ready, MetricsCollector: newWorkerMetricsCollector(
			config.Role.WorkFamily, workerExecutionClassPayload, snapshot, db,
			startup, permissionStateGate, guardedController,
			workerDatabaseObservations{queueObservation: observationPools},
		), Controller: guardedController.Run,
		ExposeDiagnostics: runtimeprocess.IsLoopbackListenAddress(config.AdminAddress),
	})
	if err != nil {
		return fail(err)
	}
	return process, db, nil
}

func assembleMetadataProduction(
	ctx context.Context,
	version string,
	config runtimeconfig.Worker,
	metadataScope runtimeconfig.WorkerMetadataScope,
	pollInterval, leaseDuration, retryJitterMax, processedRetention, processedCleanupInterval time.Duration,
	timedFlowOwnerScope *runtimeconfig.WorkerMetadataScope,
	ownershipHealthInterval time.Duration,
	permissionStateFactory permissionStateGateFactory,
	platformCommandExecutors platformCommandExecutorRegistryFactory,
	startup *workerStartupObservation,
) (*runtimeprocess.Process, *sql.DB, error) {
	stageStarted := time.Now()
	connection, err := openWorkerConnection(ctx, productionWorkerPool(config.Role), "runtime metadata worker")
	startup.observe(startupStageDatabase, stageStarted)
	if err != nil {
		return nil, nil, err
	}
	db := connection.DB()
	var graph *recordmutation.RecordsMutationRuntime
	fail := func(cause error) (*runtimeprocess.Process, *sql.DB, error) {
		if graph != nil {
			_ = graph.CloseRejectedSaveAuditPool()
		}
		_ = connection.Close()
		return nil, nil, cause
	}
	stageStarted = time.Now()
	permissionStateGate, err := newWorkerPermissionStateGate(
		ctx, db, version, config.Role, permissionStateFactory,
	)
	startup.observe(startupStagePermissionState, stageStarted)
	if err != nil {
		return fail(fmt.Errorf("runtime metadata worker: Permission State gate: %w", err))
	}
	scope, err := configuredMetadataWorkScope(config.Role)
	if err != nil {
		return fail(err)
	}
	stageStarted = time.Now()
	store, err := catalogsnapshot.NewDirectoryStore(metadataScope.CatalogSnapshotStoreRoot)
	if err != nil {
		return fail(err)
	}
	binder, err := newMutationArtifactBinder(store, scope, catalogsnapshot.Compatibility{
		SchemaGeneration: catalogsnapshot.SchemaGeneration(metadataScope.SchemaGeneration),
		EngineGeneration: catalogsnapshot.EngineGeneration(metadataScope.EngineGeneration),
	}, func(
		prepareCtx context.Context,
		view *catalogsnapshot.RuntimeMetadataView,
	) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) {
		return managedruntime.NewIndependentAssignmentPolicyEligibilityAmbientStager(
			prepareCtx,
			db,
			view,
		)
	})
	if err != nil {
		return fail(err)
	}
	startup.observe(startupStageCatalogBind, stageStarted)
	stageStarted = time.Now()
	deps, err := recordmutation.BuildRecordsMutationDependencies(connection)
	if err != nil {
		return fail(err)
	}
	graph, err = recordmutation.NewRecordsMutationEngineRuntime(
		deps,
		metadata.RuntimeMetadataViewFromContext,
	)
	if err != nil {
		return fail(err)
	}
	if err := managedruntime.InstallIndependentAssignmentPolicyEligibilityLifecycle(
		graph.Transactions,
		graph,
	); err != nil {
		return fail(err)
	}
	startup.observe(startupStageMutationGraphBuild, stageStarted)
	stageStarted = time.Now()
	if err := graph.PrepareEngine(ctx); err != nil {
		return fail(fmt.Errorf("runtime metadata worker: prepare engine graph: %w", err))
	}
	startup.observe(startupStageMutationGraphPrepare, stageStarted)
	stageStarted = time.Now()
	if err := graph.ActivateDurableMutationProducers(ctx); err != nil {
		return fail(fmt.Errorf("runtime metadata worker: activate durable mutation producers: %w", err))
	}
	startup.observe(startupStageDurableProducerActivation, stageStarted)
	if config.Role.WorkFamily == eventfeature.RuntimeRecordEventDispatchWorkFamily {
		if err := persistence.VerifyRecordAttachmentReadSchema(ctx, db); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: outbound mail attachment schema: %w", err))
		}
		if err := persistence.VerifyRuntimeOutboundIntentSchema(ctx, db); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: outbound intent schema: %w", err))
		}
		attachmentMetadata, err := graph.NewOutboundMailAttachmentMetadataAuthorizer()
		if err != nil {
			return fail(fmt.Errorf("runtime metadata worker: outbound mail attachment metadata: %w", err))
		}
		if err := bindRuntimeRecordEventOutboundIntents(
			graph,
			attachmentMetadata,
			persistence.NewRuntimeOutboundIntentRepository(db, graph.Transactions),
		); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: enable outbound intent handoff: %w", err))
		}
	}
	ownership := configuredWorkOwnership(config.Role)
	authority, err := persistence.NewRuntimeWorkOwnershipRepository(db, ownership)
	if err != nil {
		return fail(err)
	}
	var processor metadataWorkProcessor
	observationPools := &workerAuxiliaryPoolTracker{}
	parallelism := 1
	if config.Role.EffectiveParallelism() > 1 &&
		config.Role.WorkFamily != eventfeature.RuntimeRecordEventDispatchWorkFamily &&
		config.Role.WorkFamily != persistence.RuntimePlatformCommandExecutionWorkFamily &&
		config.Role.WorkFamily != records.RuntimeBulkIngestWorkFamily {
		return fail(fmt.Errorf("runtime metadata worker: parallel lanes are supported only by keyed record-event, partitioned Platform-command, or Bulk work"))
	}
	stageStarted = time.Now()
	switch config.Role.WorkFamily {
	case records.RuntimeBulkIngestWorkFamily:
		if err := authority.CheckFencesReady(ctx); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: Bulk effect fences: %w", err))
		}
		bulkBinding, _, err := resolveBulkIngestAuthority(
			ctx,
			db,
			store,
			metadataScope,
		)
		if err != nil {
			return fail(err)
		}
		encodedBulkBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(bulkBinding)
		if err != nil {
			return fail(fmt.Errorf("runtime metadata worker: encode Bulk candidate binding: %w", err))
		}
		bulkStorageConfig := storage.LoadConfigFromEnv()
		objectStore, err := storage.NewStore(bulkStorageConfig)
		if err != nil {
			return fail(fmt.Errorf("runtime metadata worker: Bulk object store: %w", err))
		}
		bulkStorageDigest, err := storage.ConfigAuthorityDigest(bulkStorageConfig)
		if err != nil {
			return fail(fmt.Errorf("runtime metadata worker: Bulk storage authority: %w", err))
		}
		holderID := "bulk-rw-" + uuid.NewString()
		bulkStore, err := persistence.NewBulkWorkStore(
			db, graph.Transactions, authority, holderID, ownership.OwnershipEpoch,
			encodedBulkBinding, leaseDuration, config.Role.MaxAttempts,
		)
		if err != nil {
			return fail(err)
		}
		bulkSource, err := records.NewBulkWorkSource(
			bulkStore, graph.Persistence, runtimeBulkObjectReader{store: objectStore},
			records.BulkStorageAuthority{
				Provider: bulkStorageConfig.Provider,
				Bucket:   bulkStorageConfig.Bucket,
				Digest:   bulkStorageDigest,
			},
			func(bindCtx context.Context, binding catalogsnapshot.RuntimeWorkBinding) (context.Context, bool, error) {
				boundCtx, bindErr := binder.bindMutation(bindCtx, binding)
				return boundCtx, isPermanentMetadataBindingError(bindErr), bindErr
			},
			records.DefaultBulkIngestCheckpointRows, config.Role.MaxAttempts,
		)
		if err != nil {
			return fail(err)
		}
		shutdownTimeout, err := time.ParseDuration(config.Role.ShutdownTimeout)
		if err != nil {
			return fail(fmt.Errorf("runtime metadata worker: parse Bulk shutdown timeout: %w", err))
		}
		generationInterval := max(pollInterval, 15*time.Second)
		reconcileInterval := max(pollInterval, time.Minute)
		bulkController, err := sharedworker.NewController(
			bulkSource,
			bulkSource,
			sharedworker.Config{
				Parallelism: config.Role.EffectiveParallelism(), ClaimBatchSize: config.Role.BatchSize,
				ReconcileInterval: reconcileInterval, GenerationInterval: generationInterval,
				ShutdownTimeout: shutdownTimeout,
			},
		)
		if err != nil {
			return fail(err)
		}
		controller := withMutationGraphCleanup(bulkController.Run, graph)
		guardedController, err := newOwnershipController(authority, controller, ownershipHealthInterval)
		if err != nil {
			return fail(err)
		}
		startup.observe(startupStageRolePrepare, stageStarted)
		capabilities, ready, err := workerPermissionStateProcessInputs(
			config.Role.WorkFamily, permissionStateGate, func() bool {
				return guardedController.Ready() && bulkController.Ready()
			},
		)
		if err != nil {
			return fail(err)
		}
		process, err := runtimeprocess.Assemble(runtimeprocess.Config{
			Role: "runtime-worker", Version: version, AdminAddress: config.AdminAddress,
			Capabilities: capabilities, Ready: ready,
			MetricsCollector: newWorkerMetricsCollector(
				config.Role.WorkFamily, workerExecutionClassMetadata,
				func() workerMetricsSnapshot { return bulkControllerMetricsSnapshot(bulkController) },
				db, startup, permissionStateGate, guardedController,
				newWorkerDatabaseObservations(observationPools, graph),
			),
			Controller:        guardedController.Run,
			ExposeDiagnostics: runtimeprocess.IsLoopbackListenAddress(config.AdminAddress),
		})
		if err != nil {
			return fail(err)
		}
		return process, db, nil
	case automation.RuntimeScheduledFlowWorkFamily:
		if _, _, err := activateRuntimeConnectorInboundIdentityExecution(graph, db); err != nil {
			return fail(fmt.Errorf(
				"runtime metadata worker: activate scheduled connector identity execution: %w",
				err,
			))
		}
		binding, view, err := resolveScheduledFlowAuthority(
			ctx,
			db,
			store,
			metadataScope,
		)
		if err != nil {
			return fail(err)
		}
		reservations, err := automation.NewFencedScheduledRunReservationService(
			db,
			scheduledRunTransactionOwner{
				records:      graph.Persistence,
				transactions: graph.Transactions,
			},
			configuredWorkOwnership(config.Role),
			binding,
			leaseDuration,
			config.Role.MaxAttempts,
		)
		if err != nil {
			return fail(err)
		}
		if err := reservations.CheckReady(ctx); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: scheduled Flow reservation readiness: %w", err))
		}
		scheduler := automation.NewSchedulerService(
			graph.Persistence,
			graph.Query,
			graph.Metadata,
			graph.Flows,
		)
		if err := scheduler.SetWorkerDiscoveryInterval(pollInterval); err != nil {
			return fail(err)
		}
		if err := scheduler.SetFencedReservationStore(reservations); err != nil {
			return fail(err)
		}
		if err := scheduler.SetFencedExecutionAuthority(scheduledFlowViewBinderAdapter{binder: binder}); err != nil {
			return fail(err)
		}
		controller := func(controllerCtx context.Context) error {
			controllerCtx = metadata.WithRuntimeMetadataView(controllerCtx, view)
			return scheduler.RunWorkerRole(controllerCtx)
		}
		controller = withMutationGraphCleanup(controller, graph)
		guardedController, err := newOwnershipController(authority, controller, ownershipHealthInterval)
		if err != nil {
			return fail(err)
		}
		startup.observe(startupStageRolePrepare, stageStarted)
		capabilities, ready, err := workerPermissionStateProcessInputs(
			config.Role.WorkFamily,
			permissionStateGate,
			guardedController.Ready,
		)
		if err != nil {
			return fail(err)
		}
		process, err := runtimeprocess.Assemble(runtimeprocess.Config{
			Role: "runtime-worker", Version: version, AdminAddress: config.AdminAddress,
			Capabilities: capabilities, Ready: ready,
			MetricsCollector: newWorkerMetricsCollector(
				config.Role.WorkFamily, workerExecutionClassMetadata,
				func() workerMetricsSnapshot { return scheduledFlowWorkerSnapshot(scheduler) },
				db, startup, permissionStateGate, guardedController,
				newWorkerDatabaseObservations(observationPools, graph)),
			Controller:        guardedController.Run,
			ExposeDiagnostics: runtimeprocess.IsLoopbackListenAddress(config.AdminAddress),
		})
		if err != nil {
			return fail(err)
		}
		return process, db, nil
	case eventfeature.RuntimeRecordEventDispatchWorkFamily:
		if err := graph.RequireDurableTimedFlowHandoff(); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: require durable timed Flow handoff: %w", err))
		}
		if timedFlowOwnerScope != nil {
			owner := configuredWorkOwnership(*config.TimedFlowOwner)
			ownerAuthority, err := persistence.NewRuntimeWorkOwnershipRepository(db, owner)
			if err != nil {
				return fail(fmt.Errorf("runtime metadata worker: scheduled Flow owner assertion: %w", err))
			}
			if err := ownerAuthority.CheckReady(ctx); err != nil {
				return fail(fmt.Errorf("runtime metadata worker: exact scheduled Flow owner is not ready: %w", err))
			}
			if _, _, err := resolveScheduledFlowAuthority(
				ctx,
				db,
				store,
				*timedFlowOwnerScope,
			); err != nil {
				return fail(fmt.Errorf("runtime metadata worker: scheduled Flow handoff authority: %w", err))
			}
			if err := graph.ActivateTimedFlowDueProducer(ctx, owner); err != nil {
				return fail(fmt.Errorf("runtime metadata worker: activate durable timed Flow handoff: %w", err))
			}
		}
		// Without the optional exact owner assertion the rejecting accepter
		// remains installed. There is no global-owner fallback.
		workerRepo, err := persistence.NewScopedRecordEventOutboxRepository(db, scope, leaseDuration)
		if err != nil {
			return fail(err)
		}
		events, err := graph.RecordEvents.RecordEventProcessor()
		if err != nil {
			return fail(err)
		}
		recordEvents := &recordEventProcessor{
			events: events, retention: processedRetention,
			observationFactory: newRecordEventObservationFactory(
				connection.OpenIndependentDB, scope, leaseDuration, observationPools,
			),
		}
		events.WithWorkerRepository(workerRepo).
			WithDurableExecutionBinder(binder.bindEncoded).
			WithDurableAdmissionFence(recordEvents.measureAdmissionFence(authority.LockActiveContext)).
			WithDurableKeyedAttemptFence(authority.LockActiveFenceContext).
			WithMaxRetryAttempts(config.Role.MaxAttempts)
		if err := authority.CheckFencesReady(ctx); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: record-event effect fences: %w", err))
		}
		repair, err := persistence.NewRuntimeRecordEventRepairStore(db, authority, scope)
		if err != nil {
			return fail(err)
		}
		if err := repair.CheckReady(ctx); err != nil {
			return fail(fmt.Errorf("runtime metadata worker: record-event repair/readiness: %w", err))
		}
		processor = recordEvents
		parallelism = config.Role.EffectiveParallelism()
	case records.RuntimeRollupDeltaWorkFamily:
		if err := graph.Persistence.ConfigureRuntimeRollupDeltaWorker(
			db, authority, scope, binder.bindMutation, config.Role.MaxAttempts,
		); err != nil {
			return fail(err)
		}
		processor = &rollupDeltaProcessor{
			runtime: graph,
			observationFactory: newRollupDeltaObservationFactory(
				connection.OpenIndependentDB,
				ownership,
				scope,
				observationPools,
			),
		}
	case persistence.RuntimePlatformCommandExecutionWorkFamily:
		processorOptions := []platformCommandProcessorOption{}
		executionStoreOptions := []persistence.RuntimePlatformCommandExecutionOption{}
		if metadataScope.RuntimeCapabilityID == commandsemantic.CapabilityName {
			recoveryOwner, err := activateRuntimeConnectorInboundFinalization(graph, db)
			if err != nil {
				return fail(fmt.Errorf(
					"runtime metadata worker: activate connector inbound finalization: %w",
					err,
				))
			}
			processorOptions = append(processorOptions, withPlatformCommandRecoveryOwner(recoveryOwner))
		}
		executors, err := platformCommandExecutors(ctx, db, graph, metadataScope.RuntimeCapabilityID)
		if err != nil {
			return fail(fmt.Errorf(
				"runtime metadata worker: assemble Platform-command executors: %w",
				err,
			))
		}
		commandStore, err := persistence.NewRuntimePlatformCommandExecutionStore(
			db,
			scheduledRunTransactionOwner{
				records:      graph.Persistence,
				transactions: graph.Transactions,
			},
			configuredWorkOwnership(config.Role),
			scope,
			leaseDuration,
			config.Role.MaxAttempts,
			executionStoreOptions...,
		)
		if err != nil {
			return fail(err)
		}
		parallelism = config.Role.EffectiveParallelism()
		processor, err = newPlatformCommandProcessor(
			commandStore,
			newPlatformCommandObservationFactory(
				connection.OpenIndependentDB, ownership, scope, observationPools,
			),
			commandViewBinderAdapter{binder: binder},
			executors,
			pollInterval,
			processorOptions...,
		)
		if err != nil {
			return fail(err)
		}
	default:
		return fail(fmt.Errorf("runtime metadata worker: unsupported work family %q", config.Role.WorkFamily))
	}
	worker, err := newMetadataWorker(processor, config.Role.BatchSize, pollInterval, retryJitterMax, processedCleanupInterval, parallelism)
	if err != nil {
		return fail(err)
	}
	if err := worker.Prepare(ctx); err != nil {
		return fail(fmt.Errorf("runtime metadata worker: readiness: %w", err))
	}
	startup.observe(startupStageRolePrepare, stageStarted)
	guardedController, err := newOwnershipController(
		authority, withMutationGraphCleanup(worker.Run, graph), ownershipHealthInterval,
	)
	if err != nil {
		return fail(err)
	}
	capabilities, ready, err := workerPermissionStateProcessInputs(
		config.Role.WorkFamily,
		permissionStateGate,
		guardedController.Ready,
	)
	if err != nil {
		return fail(err)
	}
	process, err := runtimeprocess.Assemble(runtimeprocess.Config{
		Role: "runtime-worker", Version: version, AdminAddress: config.AdminAddress,
		Capabilities: capabilities, Ready: ready,
		MetricsCollector: newWorkerMetricsCollector(
			config.Role.WorkFamily, workerExecutionClassMetadata, worker.snapshot, db,
			startup, permissionStateGate, guardedController,
			newWorkerDatabaseObservations(observationPools, graph),
		), Controller: guardedController.Run,
		ExposeDiagnostics: runtimeprocess.IsLoopbackListenAddress(config.AdminAddress),
	})
	if err != nil {
		return fail(err)
	}
	return process, db, nil
}

func bulkControllerMetricsSnapshot(controller *sharedworker.Controller) workerMetricsSnapshot {
	if controller == nil {
		return workerMetricsSnapshot{}
	}
	snapshot := controller.Snapshot()
	return workerMetricsSnapshot{
		InFlight: snapshot.InFlight, Attempts: snapshot.Claimed,
		Failures: snapshot.AttemptFailures, Processed: snapshot.Completed,
		WorkLoop: workLoopSnapshot{
			PollChecks:    snapshot.GenerationProbes + snapshot.Reconciliations,
			ConfirmedIdle: snapshot.EmptyClaims, ClaimAttempts: snapshot.ClaimCalls,
			EmptyClaims: snapshot.EmptyClaims, Claimed: snapshot.Claimed,
		},
	}
}

func withMutationGraphCleanup(
	controller func(context.Context) error,
	graph *recordmutation.RecordsMutationRuntime,
) func(context.Context) error {
	return func(ctx context.Context) (runErr error) {
		defer func() {
			runErr = errors.Join(runErr, graph.CloseRejectedSaveAuditPool())
		}()
		return controller(ctx)
	}
}

func newWorkerDatabaseObservations(
	queueObservation *workerAuxiliaryPoolTracker,
	graph *recordmutation.RecordsMutationRuntime,
) workerDatabaseObservations {
	observations := workerDatabaseObservations{queueObservation: queueObservation}
	if graph == nil {
		return observations
	}
	observations.rejectedSave = func() auxiliaryDatabasePoolSnapshot {
		snapshot := graph.DatabaseObservationSnapshot().RejectedSaveAudit
		return auxiliaryDatabasePoolSnapshot{
			Purpose: workerPoolPurposeRejectedSave, Active: snapshot.Active, Stats: snapshot.Stats,
			OpenAttempts: snapshot.OpenAttempts, OpenFailures: snapshot.OpenFailures, Closes: snapshot.Closes,
		}
	}
	observations.objectIDProbe = func() transientDatabaseSnapshot {
		snapshot := graph.DatabaseObservationSnapshot().ObjectIDProbe
		return transientDatabaseSnapshot{
			Purpose: workerTransientPurposeObjectID, OpenAttempts: snapshot.OpenAttempts,
			OpenFailures: snapshot.OpenFailures, OpenDuration: snapshot.OpenDuration,
		}
	}
	return observations
}

func scheduledFlowWorkerSnapshot(scheduler *automation.SchedulerService) workerMetricsSnapshot {
	if scheduler == nil {
		return workerMetricsSnapshot{}
	}
	snapshot := scheduler.WorkerObservationSnapshot()
	var inFlight int64
	if snapshot.ExecutionStarted > snapshot.ExecutionCompleted {
		inFlight = int64(snapshot.ExecutionStarted - snapshot.ExecutionCompleted)
	}
	var succeeded uint64
	if snapshot.ExecutionCompleted > snapshot.ExecutionFailed {
		succeeded = snapshot.ExecutionCompleted - snapshot.ExecutionFailed
	}
	return workerMetricsSnapshot{
		InFlight: inFlight, Attempts: snapshot.ExecutionStarted,
		Failures: snapshot.ExecutionFailed, Processed: succeeded,
		WorkLoop: workLoopSnapshot{
			PollChecks: snapshot.DiscoveryPasses, ConfirmedIdle: snapshot.DiscoveryIdle,
			ClaimAttempts: snapshot.AcceptanceAttempts, EmptyClaims: snapshot.AcceptanceEmpty,
			Claimed: snapshot.OccurrencesAccepted, PollDuration: snapshot.DiscoveryDuration,
			ClaimDuration: snapshot.AcceptanceDuration,
		},
		ScheduledFlow: scheduledFlowMetricsSnapshot{
			Enabled: true, DiscoveryPasses: snapshot.DiscoveryPasses,
			DefinitionsConsidered: snapshot.DefinitionsConsidered,
			OccurrencesAccepted:   snapshot.OccurrencesAccepted,
			RecoveryScanned:       snapshot.RecoveryScanned, RecoveryRecovered: snapshot.RecoveryRecovered,
			ExecutionStarted: snapshot.ExecutionStarted, ExecutionCompleted: snapshot.ExecutionCompleted,
			ExecutionFailed: snapshot.ExecutionFailed, CapacityDeferrals: snapshot.CapacityDeferrals,
			DiscoveryFailures: snapshot.DiscoveryFailures,
			DiscoveryDuration: snapshot.DiscoveryDuration, RecoveryDuration: snapshot.RecoveryDuration,
			ExecutionDuration: snapshot.ExecutionDuration,
		},
	}
}

func validateTimedFlowOwnerRole(
	recordEventRole runtimeconfig.WorkerRole,
	scheduledRole runtimeconfig.WorkerRole,
) (runtimeconfig.WorkerMetadataScope, error) {
	if err := scheduledRole.Validate(); err != nil {
		return runtimeconfig.WorkerMetadataScope{}, fmt.Errorf("runtime worker: timed Flow owner role: %w", err)
	}
	if recordEventRole.WorkFamily != eventfeature.RuntimeRecordEventDispatchWorkFamily ||
		scheduledRole.WorkFamily != automation.RuntimeScheduledFlowWorkFamily {
		return runtimeconfig.WorkerMetadataScope{}, errors.New("runtime worker: timed Flow handoff requires record-event and scheduled-Flow roles")
	}
	if scheduledRole.WorkerReleaseDigest != recordEventRole.WorkerReleaseDigest ||
		scheduledRole.WorkerSourceRevision != recordEventRole.WorkerSourceRevision {
		return runtimeconfig.WorkerMetadataScope{}, errors.New("runtime worker: timed Flow roles must use one Worker artifact and source revision")
	}
	recordEventScope, err := recordEventRole.MetadataScope()
	if err != nil {
		return runtimeconfig.WorkerMetadataScope{}, fmt.Errorf("runtime worker: record-event timed Flow scope: %w", err)
	}
	scheduledScope, err := scheduledRole.MetadataScope()
	if err != nil {
		return runtimeconfig.WorkerMetadataScope{}, fmt.Errorf("runtime worker: scheduled timed Flow scope: %w", err)
	}
	if recordEventScope.RuntimeCapabilityID != mutationsemantic.CapabilityName ||
		scheduledScope.RuntimeCapabilityID != flowsemantic.CapabilityName ||
		scheduledScope.AppCapabilityID == "" ||
		scheduledScope.ExpectedCompatibilityReceiptDigest == "" {
		return runtimeconfig.WorkerMetadataScope{}, errors.New("runtime worker: timed Flow handoff requires mutation and Catalog-selected Flow capability scopes")
	}
	if recordEventScope.OrgID != scheduledScope.OrgID ||
		recordEventScope.AppInstallationID != scheduledScope.AppInstallationID ||
		recordEventScope.CatalogSnapshotStoreRoot != scheduledScope.CatalogSnapshotStoreRoot ||
		recordEventScope.SchemaGeneration != scheduledScope.SchemaGeneration ||
		recordEventScope.EngineGeneration != scheduledScope.EngineGeneration {
		return runtimeconfig.WorkerMetadataScope{}, errors.New("runtime worker: timed Flow roles must share one AppInstallation and Catalog compatibility scope")
	}
	return scheduledScope, nil
}

// validateWorkerBinary closes every worker-image execution path, including
// short-lived repair jobs. An OCI digest cannot be embedded in the binary
// whose bytes contribute to that digest, so the process binds itself to the
// exact source revision; owner-signed provenance binds that revision to the
// configured WorkerReleaseDigest, and the durable ownership row fences the
// digest.
func validateWorkerBinary(version string, role runtimeconfig.WorkerRole) error {
	if version != role.WorkerSourceRevision {
		return fmt.Errorf("runtime worker: binary source revision %q does not match configured worker source revision %q", version, role.WorkerSourceRevision)
	}
	return nil
}