package controlplane

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"sort"
	"strings"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/records"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	handlerconfig "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundhandlerconfig"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// CandidateFinalVerifier runs against the exact operation-local Runner and
// candidate pack set after candidate Runtime metadata and derived artifacts
// are prepared, but before the candidate generation is activated.
type CandidateFinalVerifier func(context.Context, *Runner, []packs.Pack) error

// ErrOfflineCandidateLiveStateDrift is the explicit verifier signal that the
// immutable candidate identity is correct but its live derived state needs a
// full stopped-Runtime reconciliation. All unmarked verifier errors remain
// terminal so cancellation, timeout, and dependency failures cannot trigger a
// mutating ApplyAll.
var ErrOfflineCandidateLiveStateDrift = errors.New("offline candidate live state drift")

// MarkOfflineCandidateLiveStateDrift marks a verifier diagnostic as safe to
// recover through a full reconciliation of the same immutable candidate.
func MarkOfflineCandidateLiveStateDrift(err error) error {
	if err == nil {
		return nil
	}
	return fmt.Errorf("%w: %w", ErrOfflineCandidateLiveStateDrift, err)
}

type InstallPackOptions struct {
	SkipAttributionPreflight             bool
	SkipPolymorphicDiscriminatorBackfill bool
	// DeferFormulaViewRebuild makes ApplyAll neither create nor replace ANY
	// formula view: replacing views at apply time would swap shared compiler
	// output while a binary with the previous formula semantics may still
	// hold serving authority. The contract for formula-semantics-changing
	// (cutover) releases:
	//
	//   - ENTRY PRECONDITION (read-only, before any mutation): every object
	//     that requires a formula view must already have one, else the apply
	//     fail-fasts with a heal-first error. Restart-safe by construction —
	//     no snapshot or cross-attempt state exists to go stale.
	//   - A cutover release must NOT introduce active-formula objects (pack
	//     or otherwise); ship those in a prior regular release
	//     (roll-forward-only operations).
	//   - A cutover release must not change schema in ways that break
	//     existing view definitions (e.g. dropping/renaming a physical
	//     column an existing view reads): the views are intentionally left
	//     as-is, so such a release fails closed at the post-apply usability
	//     verify.
	//   - The caller MUST rebuild all views through the target runtime
	//     (maintenance rebuild-all) inside its activation window, after old
	//     serving is quiesced and before traffic is activated. A successful
	//     defer apply stamps the completed marker while every view is still
	//     semantically stale BY DESIGN — the durable
	//     formula_view_rebuild_owed receipt (stamped write-ahead at ENTRY,
	//     after the read-only precondition and before the running marker or
	//     any mutation, and idempotently re-stamped before the completed
	//     marker) records that debt with a CLOSED lifecycle: a lock-holding
	//     full view rebuild (the deployment's activation `setup rebuild`
	//     via RebuildDerivedWithLock*, or a later full apply) settles it
	//     automatically — receipt mutations are package-private and
	//     serialized on the control-plane apply lock, so no clear can
	//     interleave with a defer apply's stamp/completed sequence — and
	//     the auto-start path fails closed while it is outstanding: whether
	//     the defer
	//     apply completed, crashed mid-apply, or died right after
	//     committing, a binary can never serve or mutate past the
	//     outstanding debt.
	//
	// Function-registry refresh, view cache refresh, and every other derived
	// rebuild still run. Standalone `rebuild` and ApplySourceConvergence are
	// unaffected.
	DeferFormulaViewRebuild      bool
	RepairLegacyUniqueDuplicates bool
	VerifyAfterApply             bool
	// CandidateFinalVerify lets a distribution-owned checker participate in the
	// Controlplane-owned candidate -> verify -> activate boundary. The callback
	// must inspect the supplied Runner and candidate packs directly; it must not
	// rebuild Runtime metadata from InstalledPack state.
	CandidateFinalVerify  CandidateFinalVerifier
	SourceRevisionAdvance *SourceRevisionAdvanceAuthorization
	// ApplyAllSourceRevisionAdvances is the distribution-owned plan of exact
	// source-revision edges that a full ApplyAll may converge. The control plane
	// owns only this neutral carrier; pack names and revisions remain facts of
	// the calling distribution.
	ApplyAllSourceRevisionAdvances []SourceRevisionAdvanceAuthorization
	// ApplyAllFinalVerify runs inside ApplyAll's final readiness gate before the
	// completed marker is stamped or trusted by apply-if-needed.
	ApplyAllFinalVerify func(context.Context, *database.TiDBConnection) error

	allowInstalledPackSourceStateRepair bool
	// replayStandardActionPersistenceSemantics is minted only by ApplyAll after
	// either a completed-marker receipt proves the persisted semantics are stale
	// or an explicit/full convergence run declines to trust that receipt. Phase
	// 7 consumes it by replaying only source-backed packs that own
	// invocable_action components before strict verification.
	replayStandardActionPersistenceSemantics bool
	offlineCandidateLocalSourceTransition    *offlineCandidateMutableTransition
	// offlineCandidateCompleteSourceRevisionReplays is minted only while an
	// offline candidate converts a failed/installing revision-addressed attempt
	// with an older active generation into an active -> target advance. The
	// mutable installed row will subsequently look like an ordinary advance, so
	// this operation-local fact must carry the retry's partial-write risk to the
	// execution atom selector.
	offlineCandidateCompleteSourceRevisionReplays map[string]struct{}
	// resumeSourceBackedPackInstallID is minted only by phase-7 retry
	// admission after the durable non-active row, exact revision-bound source,
	// and immutable generation state have all been classified read-only. It is
	// never accepted from an external caller.
	resumeSourceBackedPackInstallID string
	// legacySourceIdentityAdoptionPackInstallID is set only by the explicit
	// stopped-upgrade migration for an active app-pack row that predates source
	// identity. It binds that migration to the already-active incarnation.
	legacySourceIdentityAdoptionPackInstallID string
	// releaseSourceIdentityAdoptionPackInstallID is set only by an offline
	// stopped-candidate transition that rebinds an identical active local_path
	// installation to its exact revision-addressed release_repo authority.
	releaseSourceIdentityAdoptionPackInstallID string
	offlineCandidate                           *offlineCandidateExecution
	offlineCandidatePreparation                *offlineCandidateExecution
	// stoppedSourceSetApplyOnly is minted only by ApplyStoppedSourceSet while
	// Runtime is stopped. It selects the apply-only execution and activation
	// boundary. The source-pack implementation still owns dependency and
	// mutation-safety admission; only online schema readiness and post-apply
	// verification are omitted.
	stoppedSourceSetApplyOnly              bool
	stoppedSourceSetAuthority              offlinecandidateauthority.StoppedSourceSetAuthority
	stoppedSourceSetPreviousPackInstallID  string
	stoppedSourceSetCandidatePackInstallID string
}

type SourceRevisionAdvanceAuthorization struct {
	Name         string
	Kind         string
	URI          string
	FromRevision string
	ToRevision   string
}

var (
	ensureInstallPackSchemaReadinessForInstall        = ensureInstallPackSchemaReadiness
	validateLivePackDependenciesForInstall            = validateLivePackDependencies
	samePackAllowedOwnerlessObjectsForInstall         = samePackAllowedOwnerlessObjectsWithOptions
	samePackAllowedOwnerlessObjectsForReapply         = samePackAllowedOwnerlessObjectsWithOptionsAndComponentKindRegistrationSet
	applyResolvedSourceBackedPackForInstall           = applyResolvedSourceBackedPackUnlockedWithPreflightOverrides
	verifyInstalledPackByNameForInstall               = VerifyInstalledPackByNameWithRunner
	reconcileSystemObjectColumnsForReapplyPackInstall = reconcileSystemObjectColumnsForCacheRefresh
	registerInstalledRuntimeMetadataForCompileProof   = packs.RegisterInstalledRuntimeMetadataWithComponentKindRegistrationSet
	newRunnerForFormulaCompileProof                   = NewRunnerWithOptions
	refreshRuntimeFunctionsForCompileProof            = func(r *Runner) error {
		if r != nil && r.SvcMgr != nil && r.SvcMgr.FormulaView() != nil {
			return r.SvcMgr.FormulaView().RefreshRuntimeFunctionRegistry()
		}
		return nil
	}
	refreshRunnerCacheForCompileProof = func(ctx context.Context, r *Runner) error {
		return r.RefreshCache(ctx)
	}
	sourceRevisionAdvanceValidatePayloadHashForInstalledRow = validatePayloadHashForInstalledPackSourceRevisionWithComponentKindRegistrationSet
)

func InstallPackFromSource(ctx context.Context, r *Runner, rawSource string) error {
	return InstallPackFromSourceWithOptions(ctx, r, rawSource, InstallPackOptions{})
}

func InstallPackFromSourceWithOptions(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions) error {
	// Operator install/reapply paths keep always-rebuild semantics: the defer
	// contract belongs to ApplyAll (the deployment path with the entry
	// precondition); honoring the flag here would fail closed on missing
	// views with wrong cause attribution and no entry gate.
	options.DeferFormulaViewRebuild = false
	db := runnerSQLDB(r)
	sourceOpts := packs.DefaultPackSourceOptions()
	registrations := r.componentKindRegistrationSet()
	if err := validateSourceDiagnosticsForOperationWithInstallContextAndOptionsAndComponentKindRegistrationSet(ctx, db, "install-pack", rawSource, sourceOpts, registrations); err != nil {
		return err
	}
	return withControlPlaneMutationLock(ctx, db, "InstallPack", func() error {
		return installPackFromSourceUnlockedWithPreflightOverridesAndComponentKindRegistrationSet(
			ctx, r, rawSource, options, nil, registrations,
		)
	})
}

func ReapplyPackFromSource(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions) error {
	// Operator install/reapply paths keep always-rebuild semantics: the defer
	// contract belongs to ApplyAll (the deployment path with the entry
	// precondition); honoring the flag here would fail closed on missing
	// views with wrong cause attribution and no entry gate.
	options.DeferFormulaViewRebuild = false
	db := runnerSQLDB(r)
	sourceOpts := packs.DefaultPackSourceOptions()
	if err := validateReapplySourceDiagnosticsForOperationWithInstallContextAndOptions(ctx, r, "reapply-pack", rawSource, sourceOpts, options); err != nil {
		return err
	}
	return withReapplyPackLockScope(ctx, db, nil, func(applyLock controlPlaneApplyLockScope) error {
		return reapplyPackFromSourceUnlockedUnderApplyLock(ctx, r, rawSource, options, applyLock)
	})
}

func ReapplyPackFromSourceWithSchemaReadiness(ctx context.Context, db *database.TiDBConnection, rawSource string, options InstallPackOptions) (*database.TiDBConnection, error) {
	return ReapplyPackFromSourceWithSchemaReadinessAndRunnerOptions(ctx, db, rawSource, options, RunnerOptions{})
}

func ReapplyPackFromSourceWithSchemaReadinessAndRunnerOptions(ctx context.Context, db *database.TiDBConnection, rawSource string, options InstallPackOptions, runnerOptions RunnerOptions) (*database.TiDBConnection, error) {
	// Operator install/reapply paths keep always-rebuild semantics: the defer
	// contract belongs to ApplyAll (the deployment path with the entry
	// precondition); honoring the flag here would fail closed on missing
	// views with wrong cause attribution and no entry gate.
	options.DeferFormulaViewRebuild = false
	sourceOpts := packs.DefaultPackSourceOptions()
	if err := validateReapplySourceDiagnosticsForOperationWithInstallContextAndOptionsAndComponentKindRegistrationSet(
		ctx, &Runner{DB: db}, "reapply-pack", rawSource, sourceOpts, options, runnerOptions.ComponentKindRegistrationSet,
	); err != nil {
		return db, err
	}
	currentDB := db
	var r *Runner
	err := withReapplyPackLockScope(ctx, db.DB(), func() *sql.DB {
		if currentDB != nil {
			return currentDB.DB()
		}
		return db.DB()
	}, func(applyLock controlPlaneApplyLockScope) error {
		if err := runLoggedControlPlanePhase("ReapplyPack", "reconcile-system-object-columns-for-cache-refresh", func() error {
			var phaseErr error
			currentDB, phaseErr = reconcileSystemObjectColumnsForReapplyPackInstall(currentDB)
			if phaseErr != nil {
				return fmt.Errorf("reconcile _System_Object schema before reapply-pack cache refresh: %w", phaseErr)
			}
			return nil
		}); err != nil {
			return err
		}
		if err := runLoggedControlPlanePhase("ReapplyPack", "runner-init-and-initial-cache-refresh", func() error {
			var phaseErr error
			r, phaseErr = NewRunnerWithOptions(currentDB, runnerOptions)
			if phaseErr != nil {
				return fmt.Errorf("runner init: %w", phaseErr)
			}
			if phaseErr := r.RefreshCache(ctx); phaseErr != nil {
				return fmt.Errorf("initial cache refresh: %w", phaseErr)
			}
			return nil
		}); err != nil {
			return err
		}
		return runLoggedControlPlanePhase("ReapplyPack", "reapply-pack-from-source", func() error {
			return reapplyPackFromSourceUnlockedUnderApplyLock(ctx, r, rawSource, options, applyLock)
		})
	})
	return currentDB, err
}

func withReapplyPackLock(ctx context.Context, rawDB *sql.DB, markerDB func() *sql.DB, fn func() error) error {
	return withReapplyPackLockScope(
		ctx, rawDB, markerDB,
		func(controlPlaneApplyLockScope) error { return fn() },
	)
}

func withReapplyPackLockScope(
	ctx context.Context,
	rawDB *sql.DB,
	markerDB func() *sql.DB,
	fn func(controlPlaneApplyLockScope) error,
) error {
	// reapply-pack is pack-scoped source convergence, not a full ApplyAll. It
	// must serialize with ApplyAll and downgrade the full-ApplyAll marker before
	// the first write so a crash/SIGKILL cannot let auto-start skip recovery on
	// partial state. Only the full ApplyAll path may restore completed.
	return withControlPlaneApplyLockScope(ctx, rawDB, "ReapplyPack", func(applyLock controlPlaneApplyLockScope) error {
		if err := bootstrapStateStore.MarkControlPlaneApplyAllInvalidated(rawDB, "ReapplyPack"); err != nil {
			return fmt.Errorf("invalidate ApplyAll completion marker before ReapplyPack: %w", err)
		}
		log.Printf("controlplane: invalidated full ApplyAll marker before ReapplyPack; next auto restart must run ApplyAll")

		err := fn(applyLock)
		if err == nil {
			return nil
		}
		if markErr := bootstrapStateStore.MarkControlPlaneApplyAllFailed(selectedReapplyMarkerDB(rawDB, markerDB), err); markErr != nil {
			log.Printf("controlplane: failed to mark ApplyAll failed after reapply-pack error: %v (original error: %v)", markErr, err)
		}
		return err
	})
}

func selectedReapplyMarkerDB(rawDB *sql.DB, markerDB func() *sql.DB) *sql.DB {
	if markerDB != nil {
		if selected := markerDB(); selected != nil {
			return selected
		}
	}
	return rawDB
}

func runnerSQLDB(r *Runner) *sql.DB {
	if r == nil || r.DB == nil {
		return nil
	}
	return r.DB.DB()
}

func reconcileSystemObjectColumnsForCacheRefresh(db *database.TiDBConnection) (*database.TiDBConnection, error) {
	if err := schemaProvisioner.ReconcileSystemObjectTableColumns(db.DB()); err != nil {
		return nil, err
	}
	newDB, err := database.NewConnection()
	if err != nil {
		return nil, fmt.Errorf("re-create database connection after _System_Object DDL: %w", err)
	}
	if err := db.Close(); err != nil {
		log.Printf("   ⚠️  Failed to close old DB connection: %v", err)
	}
	return newDB, nil
}

func installPackFromSourceUnlockedWithOptions(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions) error {
	return installPackFromSourceUnlockedWithPreflightOverrides(ctx, r, rawSource, options, nil)
}

func installPackFromSourceUnlockedWithPreflightOverrides(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions, preflightOverrides *installPackPreflightOverrides) error {
	return installPackFromSourceUnlockedWithPreflightOverridesAndComponentKindRegistrationSet(
		ctx, r, rawSource, options, preflightOverrides, packs.ComponentKindRegistrationSet{},
	)
}

func installPackFromSourceUnlockedWithPreflightOverridesAndComponentKindRegistrationSet(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions, preflightOverrides *installPackPreflightOverrides, registrations packs.ComponentKindRegistrationSet) error {
	return installPackFromSourceUnlockedWithPreflightOverridesAndPackSourceOptionsAndComponentKindRegistrationSet(
		ctx, r, rawSource, options, preflightOverrides, packs.DefaultPackSourceOptions(), registrations,
	)
}

func installPackFromSourceUnlockedWithPreflightOverridesAndPackSourceOptionsAndComponentKindRegistrationSet(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions, preflightOverrides *installPackPreflightOverrides, sourceOpts packs.PackSourceOptions, registrations packs.ComponentKindRegistrationSet) error {
	if err := ensureInstallPackSchemaReadinessForInstall(ctx, r); err != nil {
		return err
	}

	report, runtimeFormulaFunctionBundles, err := validateInstallablePackSourceForInstallWithPreflightOverridesAndOptionsAndComponentKindRegistrationSet(
		ctx, runnerSQLDB(r), rawSource, preflightOverrides, sourceOpts, registrations,
	)
	if err != nil {
		return err
	}
	if err := installableValidationReportError("install-pack", report); err != nil {
		return err
	}
	pack := report.Pack
	if pack == nil {
		return fmt.Errorf("install-pack: source did not produce an installable pack")
	}
	if pack.Category() != packs.AppPack && pack.Category() != packs.ConnectorPack {
		return fmt.Errorf("install-pack supports categories %q and %q, got %q", packs.AppPack, packs.ConnectorPack, pack.Category())
	}

	if err := validateLivePackDependenciesForInstallWithPreflightOverrides(ctx, r.DB.DB(), pack, preflightOverrides); err != nil {
		return err
	}
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		return fmt.Errorf("pack %s payload hash: %w", pack.Name(), err)
	}
	var allowedObjects []string
	if len(registrations.Registrations()) == 0 {
		allowedObjects, err = samePackAllowedOwnerlessObjectsForInstall(ctx, r.DB.DB(), pack, payloadHash, options)
	} else {
		allowedObjects, err = samePackAllowedOwnerlessObjectsForReapply(ctx, r.DB.DB(), pack, payloadHash, options, registrations)
	}
	if err != nil {
		return err
	}
	if len(allowedObjects) > 0 {
		pack, err = packs.NewSourceBackedPack(
			report.Source,
			packs.WithComponentKindRegistrationsFromSourcePack(pack),
			packs.WithRuntimeFormulaFunctionBundlesForValidation(packs.RuntimeFormulaFunctionBundlesExcludingResolvedSource(report.Source, runtimeFormulaFunctionBundles)),
			packs.WithAllowedOwnerlessObjectAPIs(allowedObjects),
			packs.WithFlowActionCallRuntimeFromPackSourceOptions(sourceOpts),
			packs.WithConnectorPushValidationSchemaResolverForSource(report.Source, sourceOpts),
		)
		if err != nil {
			return fmt.Errorf("construct source-backed pack with ownerless policy: %w", err)
		}
	}
	if preflightOverrides == nil {
		preflightOverrides = &installPackPreflightOverrides{}
	}
	preflightOverrides.skipAttributionPreflight = options.SkipAttributionPreflight
	if err := applyResolvedSourceBackedPackForInstall(ctx, r, pack, preflightOverrides, options); err != nil {
		return err
	}
	if options.VerifyAfterApply {
		log.Printf("controlplane: operator verify enabled after applying pack %q", pack.Name())
		if err := verifyInstalledPackByNameForInstall(ctx, r, pack.Name(), true); err != nil {
			return fmt.Errorf("verify applied pack %s: %w", pack.Name(), err)
		}
	}
	return nil
}

func reapplyPackFromSourceUnlocked(ctx context.Context, r *Runner, rawSource string, options InstallPackOptions) error {
	return reapplyPackFromSourceUnlockedUnderApplyLock(
		ctx, r, rawSource, options, controlPlaneApplyLockScope{},
	)
}

func reapplyPackFromSourceUnlockedUnderApplyLock(
	ctx context.Context,
	r *Runner,
	rawSource string,
	options InstallPackOptions,
	applyLock controlPlaneApplyLockScope,
) error {
	registrations := r.componentKindRegistrationSet()
	sourceReport, err := reapplySourceReportWithComponentKindRegistrationSet(rawSource, registrations)
	if err != nil {
		return err
	}
	options, revisionAdvance, err := prepareReapplySourceRevisionAdvance(rawSource, sourceReport.Source, options)
	if err != nil {
		return err
	}
	if revisionAdvance {
		status, err := PendingApplyAllSourceRevisionAdvances(ctx, r.DB.DB(), options)
		if err != nil {
			return err
		}
		requiresTransition := false
		for _, pending := range status.Pending {
			if pending.Kind == ApplyAllSourceRevisionAdvancePendingAdvance ||
				pending.Kind == ApplyAllSourceRevisionAdvancePendingRetry {
				requiresTransition = true
			}
		}
		if status.HasPending() {
			if err := applyAllSourceRevisionAdvances(ctx, r, options, applyLock); err != nil {
				return err
			}
			if requiresTransition {
				if options.VerifyAfterApply {
					log.Printf("controlplane: operator verify enabled after applying pack %q", options.SourceRevisionAdvance.Name)
					if err := verifyInstalledPackByNameForInstall(ctx, r, options.SourceRevisionAdvance.Name, true); err != nil {
						return fmt.Errorf("verify applied pack %s: %w", options.SourceRevisionAdvance.Name, err)
					}
				}
				return nil
			}
		}
	}
	row, err := activeInstalledPackRowForReapplySource(ctx, r.DB.DB(), sourceReport)
	if err != nil {
		return err
	}
	preflightOverrides, err := installPackPreflightOverridesForReapplySourceWithComponentKindRegistrationSet(
		ctx, r, row, sourceReport.Source.Manifest.Dependencies, registrations,
	)
	if err != nil {
		return err
	}
	return installPackFromSourceUnlockedWithPreflightOverridesAndComponentKindRegistrationSet(
		ctx, r, rawSource, options, preflightOverrides, registrations,
	)
}

func prepareReapplySourceRevisionAdvance(
	rawSource string,
	source *packs.ResolvedPackSource,
	options InstallPackOptions,
) (InstallPackOptions, bool, error) {
	if options.SourceRevisionAdvance == nil {
		return options, false, nil
	}
	// The public reapply interface authorizes one requested source plus one
	// revision edge. Bind identity to that explicit request before resolving or
	// comparing target bytes; never infer identity from the resolved source.
	requested, err := packs.ParsePackSourceRef(rawSource)
	if err != nil {
		return options, false, fmt.Errorf("reapply-pack source revision advance request: %w", err)
	}
	if requested.Kind != packs.PackSourceKindOfficialRepo {
		return options, false, fmt.Errorf("reapply-pack source revision advance requires an official_repo source")
	}
	authorization := *options.SourceRevisionAdvance
	if strings.TrimSpace(authorization.Name) == "" {
		authorization.Name = requested.URI
	}
	if strings.TrimSpace(authorization.Kind) == "" {
		authorization.Kind = string(requested.Kind)
	}
	if strings.TrimSpace(authorization.URI) == "" {
		authorization.URI = requested.URI
	}
	plan, err := normalizeApplyAllSourceRevisionAdvancePlan([]SourceRevisionAdvanceAuthorization{authorization})
	if err != nil {
		return options, false, err
	}
	authorization = plan[0].SourceRevisionAdvanceAuthorization
	options.SourceRevisionAdvance = &authorization
	options.ApplyAllSourceRevisionAdvances = []SourceRevisionAdvanceAuthorization{authorization}
	if err := verifyReapplySourceRevisionAdvanceResolvedSource(source, authorization); err != nil {
		return options, false, err
	}
	return options, true, nil
}

func verifyReapplySourceRevisionAdvanceResolvedSource(
	source *packs.ResolvedPackSource,
	authorization SourceRevisionAdvanceAuthorization,
) error {
	if source == nil || source.Manifest == nil {
		return fmt.Errorf("reapply-pack source revision advance target is unavailable")
	}
	kind, uri, err := canonicalPackSourceIdentity(source.Kind, source.URI)
	if err != nil ||
		strings.TrimSpace(source.Manifest.Name) != strings.TrimSpace(authorization.Name) ||
		kind != strings.TrimSpace(authorization.Kind) ||
		uri != strings.TrimSpace(authorization.URI) ||
		strings.TrimSpace(source.Revision) != strings.TrimSpace(authorization.ToRevision) {
		return fmt.Errorf("reapply-pack source revision advance target does not match the resolved source")
	}
	return nil
}

func reapplySourceReportAndInstalledRow(ctx context.Context, db *sql.DB, rawSource string) (packs.PackValidationReport, installedPackRow, error) {
	return reapplySourceReportAndInstalledRowWithComponentKindRegistrationSet(ctx, db, rawSource, packs.ComponentKindRegistrationSet{})
}

func reapplySourceReportAndInstalledRowWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, rawSource string, registrations packs.ComponentKindRegistrationSet) (packs.PackValidationReport, installedPackRow, error) {
	sourceReport, err := reapplySourceReportWithComponentKindRegistrationSet(rawSource, registrations)
	if err != nil {
		return sourceReport, installedPackRow{}, err
	}
	row, err := activeInstalledPackRowForReapplySource(ctx, db, sourceReport)
	return sourceReport, row, err
}

func reapplySourceReportWithComponentKindRegistrationSet(rawSource string, registrations packs.ComponentKindRegistrationSet) (packs.PackValidationReport, error) {
	sourceReport := packs.ValidatePackSourceWithComponentKindRegistrationSet(rawSource, packs.DefaultPackSourceOptions(), registrations)
	if err := packValidationReportError("reapply-pack", sourceReport); err != nil {
		return sourceReport, err
	}
	if sourceReport.Source == nil || sourceReport.Source.Manifest == nil {
		return sourceReport, fmt.Errorf("reapply-pack: source did not produce a pack manifest")
	}
	if sourceReport.Source.Manifest.Category != packs.AppPack && sourceReport.Source.Manifest.Category != packs.ConnectorPack {
		return sourceReport, fmt.Errorf("reapply-pack supports categories %q and %q, got %q", packs.AppPack, packs.ConnectorPack, sourceReport.Source.Manifest.Category)
	}
	return sourceReport, nil
}

func activeInstalledPackRowForReapplySource(ctx context.Context, db *sql.DB, sourceReport packs.PackValidationReport) (installedPackRow, error) {
	packName := strings.TrimSpace(sourceReport.Source.Manifest.Name)
	row, ok, err := getInstalledPackRowByName(ctx, db, packName)
	if err != nil {
		return installedPackRow{}, err
	}
	if !ok {
		return installedPackRow{}, fmt.Errorf("reapply-pack: installed pack %s is missing; use install-pack for first install", packName)
	}
	if row.Status != installedPackStatusActive {
		return installedPackRow{}, fmt.Errorf("reapply-pack: installed pack %s status is %q, want active", packName, row.Status)
	}
	if !installedPackHasSourceIdentity(row) {
		return installedPackRow{}, fmt.Errorf("reapply-pack: installed pack %s has no source identity; run source-backed adoption or install-pack first", packName)
	}
	return row, nil
}

func applySourceBackedPackInstallPlan(ctx context.Context, r *Runner, pack *packs.SourceBackedPack, options InstallPackOptions) error {
	plan, err := pack.InstallPlan()
	if err != nil {
		return err
	}
	return plan.Apply(ctx, r.PackDepsWithOptions(packs.ApplyOptions{
		SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
		RepairLegacyUniqueDuplicates:         options.RepairLegacyUniqueDuplicates,
		DeferFormulaViewRebuild:              options.DeferFormulaViewRebuild,
	}), packs.InstallPlanHooks{
		AfterPhase: func(ctx context.Context, phase packs.InstallPlanPhase) error {
			if options.stoppedSourceSetApplyOnly && phase == packs.InstallPlanPhaseAutomation {
				if r == nil || r.SvcMgr == nil {
					return fmt.Errorf("service graph is unavailable")
				}
				return packs.ApplyStoppedSourceSetFlowRuntimeMaintenance(
					ctx,
					options.stoppedSourceSetAuthority,
					r.DB,
					r.SvcMgr.Metadata(),
					pack,
				)
			}
			return runSourceBackedInstallPhaseBoundary(ctx, r, pack, phase)
		},
	})
}

func validateSourceBackedPackInstallPreflights(
	ctx context.Context,
	deps packs.ApplyDeps,
	registrations packs.ComponentKindRegistrationSet,
	pack *packs.SourceBackedPack,
	preflightOverrides *installPackPreflightOverrides,
) error {
	if deps.DB == nil {
		return fmt.Errorf("install-pack preflight database is unavailable")
	}
	if err := validateFormulaStorageTransitionAdmissionPreflight(ctx, deps.DB, pack); err != nil {
		return err
	}
	if offlinecandidateauthority.StoppedSourceSetApplyOnly(ctx, pack.Name()) {
		if err := packs.ValidateStoppedSourceSetObjectSpecCompatibility(ctx, deps, pack); err != nil {
			return fmt.Errorf("stopped source-set object-spec compatibility: %w", err)
		}
	}
	if err := validatePackManagedApprovalOwnership(
		ctx, deps.DB, pack.Name(), pack.Category(), registrations,
	); err != nil {
		return fmt.Errorf("install-pack managed approval ownership: %w", err)
	}
	if preflightOverrides != nil && preflightOverrides.skipAttributionPreflight {
		log.Printf("controlplane: operator override enabled; skipping attribution preflight for pack %q", pack.Name())
	} else if err := validateInstallPackAttributionPreflightWithComponentKindRegistrationSet(
		ctx, deps.DB, pack, preflightOverrides, registrations,
	); err != nil {
		return err
	}
	if err := validateInstallPackChildRelationshipNamespacePreflight(ctx, deps.DB, pack); err != nil {
		return err
	}
	if err := assertConnectorCapabilitiesHaveHandlers(pack); err != nil {
		return err
	}
	if err := validateDefaultChildRowsInstallPreflightWithSnapshots(
		ctx, deps.Metadata, engineOwnedByObjectFromAutoNumber(deps.AutoNumberMetadata), pack,
	); err != nil {
		return err
	}
	if err := validateChildCloneRulesInstallPreflightWithSnapshot(ctx, deps.Metadata, pack); err != nil {
		return err
	}
	return packs.ValidateSourceBackedPermissionOwnerPreflight(ctx, deps.DB, pack)
}

// validateFormulaStorageTransitionAdmissionPreflight rejects an ordinary
// online transition before runtime declarations or InstalledPack state can be
// rewritten. applyObjectSpecs retains the same check as a final defense at the
// physical schema boundary.
func validateFormulaStorageTransitionAdmissionPreflight(
	ctx context.Context,
	db *sql.DB,
	pack *packs.SourceBackedPack,
) error {
	if pack == nil {
		return nil
	}
	writeMaterializedFields, err := packs.WriteMaterializedFormulaStorageFields(pack)
	if err != nil {
		return fmt.Errorf("classify formula storage transition for %s: %w", pack.Name(), err)
	}
	if len(writeMaterializedFields) == 0 {
		return nil
	}
	rows, err := listInstalledSourceBackedPacks(ctx, db)
	if err != nil {
		return fmt.Errorf("classify formula storage transition for %s: %w", pack.Name(), err)
	}
	var row *installedPackRow
	for index := range rows {
		if strings.EqualFold(strings.TrimSpace(rows[index].Name), strings.TrimSpace(pack.Name())) {
			row = &rows[index]
			break
		}
	}
	if row == nil {
		return nil
	}
	if !row.RuntimeObjectSpecDecl.Valid || !row.RuntimeFormulaBundles.Valid {
		return fmt.Errorf("classify formula storage transition for %s: active runtime declarations are unavailable", pack.Name())
	}
	comparison, err := packs.CompareFormulaStorageTransitionRuntimeDeclarations(
		row.RuntimeObjectSpecDecl.String,
		row.RuntimeFormulaBundles.String,
		pack,
	)
	if err != nil {
		return fmt.Errorf("classify formula storage transition for %s: %w", pack.Name(), err)
	}
	for _, transition := range comparison.Transitions {
		if offlinecandidateauthority.FormulaStorageTransitionAllowed(
			ctx, pack.Name(), transition.ObjectAPIName, transition.FieldAPIName,
		) {
			continue
		}
		return fmt.Errorf(
			"formula storage transition for %s.%s requires stopped-Runtime offline candidate execution with an exact Catalog-owned plan",
			transition.ObjectAPIName,
			transition.FieldAPIName,
		)
	}
	return nil
}

func runSourceBackedInstallPhaseBoundary(ctx context.Context, r *Runner, pack *packs.SourceBackedPack, phase packs.InstallPlanPhase) error {
	if r == nil || r.SvcMgr == nil {
		return fmt.Errorf("service graph is unavailable")
	}
	return runSourceBackedInstallPhaseBoundaryWithRuntime(
		ctx, r.DB, r.SvcMgr, pack, phase, "install-pack",
	)
}

// runSourceBackedInstallPhaseBoundaryWithRuntime owns the shared phase-boundary
// behavior for ordinary source installs and exact source-revision advances. The
// exact path passes its proof-frozen database and service graph directly so a
// later mutation of the Runner shell cannot change execution authority.
func runSourceBackedInstallPhaseBoundaryWithRuntime(
	ctx context.Context,
	db *database.TiDBConnection,
	serviceGraph ServiceGraph,
	pack *packs.SourceBackedPack,
	phase packs.InstallPlanPhase,
	flowMaintenanceSource string,
) error {
	if serviceGraph == nil {
		return fmt.Errorf("service graph is unavailable")
	}
	switch phase {
	case packs.InstallPlanPhaseObjectMetadata:
		if err := serviceGraph.RefreshMetadataCache(ctx); err != nil {
			return fmt.Errorf("refresh metadata after object metadata: %w", err)
		}
	case packs.InstallPlanPhaseSecurity:
		if err := serviceGraph.RefreshMetadataCache(ctx); err != nil {
			return fmt.Errorf("refresh metadata before permission materialization: %w", err)
		}
		if len(pack.OwnedObjectAPINames()) > 0 || sourceBackedPackHasComponentKind(pack, packs.ComponentKindPermission) {
			if serviceGraph.Permissions() == nil || serviceGraph.Metadata() == nil {
				return fmt.Errorf("permission or metadata service is unavailable")
			}
			if err := packs.InitializePermissions(serviceGraph.Permissions(), serviceGraph.Metadata()); err != nil {
				return fmt.Errorf("initialize permissions: %w", err)
			}
		}
	case packs.InstallPlanPhaseAutomation:
		hasStandardFlow, err := sourceBackedPackNeedsFlowRuntimeMaintenance(ctx, db, pack)
		if err != nil {
			return err
		}
		if hasStandardFlow {
			if db == nil {
				return fmt.Errorf("runner database connection is unavailable")
			}
			if err := runFlowRuntimeMaintenanceAfterOrphanRepair(
				ctx, db, serviceGraph.Metadata(), flowMaintenanceSource,
			); err != nil {
				return err
			}
		}
	}
	return nil
}

func adoptLegacyInstalledPackSourceIdentities(ctx context.Context, r *Runner, options InstallPackOptions) error {
	rows, err := legacyInstalledAppPacksMissingSourceIdentity(ctx, r.DB.DB())
	if err != nil {
		return err
	}
	registrations := r.componentKindRegistrationSet()
	for _, row := range rows {
		name := strings.ToLower(strings.TrimSpace(row.Name))
		if options.offlineCandidate != nil {
			if _, deferred := options.offlineCandidate.plan.legacySourceAdoptions[name]; deferred {
				packInstallID := strings.TrimSpace(row.PackInstallID)
				if packInstallID == "" {
					return fmt.Errorf("offline candidate legacy installed pack %s has no PackInstallID after repair", row.Name)
				}
				if options.offlineCandidate.legacySourceAdoptionPackInstallIDs == nil {
					options.offlineCandidate.legacySourceAdoptionPackInstallIDs = make(map[string]string)
				}
				options.offlineCandidate.legacySourceAdoptionPackInstallIDs[name] = packInstallID
				continue
			}
		}
		source, err := legacyInstalledPackAdoptionSource(row)
		if err != nil {
			return err
		}
		adoptionOptions := options
		adoptionOptions.legacySourceIdentityAdoptionPackInstallID = row.PackInstallID
		if err := installPackFromSourceUnlockedWithPreflightOverridesAndComponentKindRegistrationSet(ctx, r, source, adoptionOptions, nil, registrations); err != nil {
			return fmt.Errorf("adopt legacy installed pack %s from %s: %w", row.Name, source, err)
		}
	}
	return nil
}

var (
	listInstalledAppPacksForReconcile                                        = listInstalledSourceBackedPacks
	preflightRetryableInstalledPackSourceStateForReconcile                   = preflightRetryableInstalledPackSourceState
	sourceBackedPackFromInstalledRowForReconcile                             = sourceBackedPackFromInstalledRow
	verifyInstalledPackHashStateForReconcile                                 = verifyInstalledPackHashState
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile                 = rebuildRepairPackWithAllowedOwnerlessObjects
	rebuildRepairPackWithAllowedOwnerlessObjectsAndRegistrationsForReconcile = rebuildRepairPackWithAllowedOwnerlessObjectsAndComponentKindRegistrationSet
	applyResolvedSourceBackedPackUnlockedForReconcile                        = applyResolvedSourceBackedPackUnlockedWithPreflightOverrides
	backfillChildCloneRulePackOwnershipForReconcile                          = packs.BackfillObjectSpecChildCloneRulePackOwnershipFromActivePacksWithComponentKindRegistrationSet
	backfillExclusiveSiblingFieldPackOwnershipForReconcile                   = packs.BackfillObjectSpecExclusiveSiblingFieldPackOwnershipFromActivePacksWithComponentKindRegistrationSet
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile               = packs.InspectLiveInstalledPackMissingPhysicalColumns
	repairPackOwnedMissingPhysicalColumnsForReconcile                        = packs.RepairPackOwnedMissingPhysicalColumns
	verifyResolvedSourceBackedPackForReconcile                               = func(ctx context.Context, row installedPackRow, db *sql.DB, pack *packs.SourceBackedPack) error {
		if err := assertConnectorCapabilitiesHaveHandlersWithInstalledCompat(pack, true); err != nil {
			return fmt.Errorf("installed pack %s capability handler guard: %w", pack.Name(), err)
		}
		if err := verifyInstalledConnectorPackHandlerConfigs(ctx, db, pack); err != nil {
			return fmt.Errorf("installed pack %s live connector inbound handlers: %w", pack.Name(), err)
		}
		return pack.Verify(ctx, packs.VerifyDeps{DB: db, DeepChecks: false})
	}
	markVerifiedInstalledPackConvergedForReconcile = markInstalledPackActive
	publishFirstPackGenerationForReconcile         = publishFirstSourceBackedPackGeneration
	preflightPackGenerationForReconcile            = preflightActiveInstalledPackGenerationAuthorities
)

type installPackPreflightOverrides struct {
	// plannedDesiredPacks owns the declarative target facts used by install
	// preflight. A non-nil empty context suppresses legacy fallback.
	plannedDesiredPacks *installPackPreflightPackContext
	// runtimeEffectivePacks owns the peer facts that are effective in the live
	// runtime. It must not contain merely planned or retryable targets.
	runtimeEffectivePacks *installPackPreflightPackContext
	// runtimeProjectionCompatibilityPacks owns generation-backed persisted peer
	// declarations admitted only for one pending local-pack reapply.
	runtimeProjectionCompatibilityPacks *installPackPreflightRuntimeDeclarationContext
	// activeGenerationAttributionPeers owns the exact object-ownership snapshot
	// for active source-backed peers omitted from one stopped source set. A
	// non-nil empty context is authoritative: attribution must not reopen peer
	// sources or fall back to process-global source authority.
	activeGenerationAttributionPeers *installPackPreflightAttributionPeerContext
	// sourceRevisionAdvanceRetryPeers is an opaque prepared-attempt authority
	// for exact non-active peers in one admitted source-revision tuple. It does
	// not authorize arbitrary failed or installing installed-pack rows.
	sourceRevisionAdvanceRetryPeers *applyAllSourceRevisionAdvanceAttributionRetryAuthority
	// offlineCandidatePrepared is the Catalog operation that owns exact peers
	// already prepared by this same ApplyAll execution. It never admits an
	// unrelated installing row.
	offlineCandidatePrepared *offlineCandidateExecution
	// activeDriftedPacks is the legacy shared context for reapply and repair.
	// New orchestration must populate the two authority-specific maps above.
	activeDriftedPacks       map[string]*packs.SourceBackedPack
	skipAttributionPreflight bool
}

type activeInstalledPackSourceStateDriftKind uint8

const (
	activeInstalledPackSourceStateDriftPayloadHash activeInstalledPackSourceStateDriftKind = iota + 1
	activeInstalledPackSourceStateDriftInconsistentHashes
	activeInstalledPackSourceStateDriftHostRequirements
	activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration
)

func (kind activeInstalledPackSourceStateDriftKind) String() string {
	switch kind {
	case activeInstalledPackSourceStateDriftPayloadHash:
		return "payload hash drift"
	case activeInstalledPackSourceStateDriftInconsistentHashes:
		return "inconsistent installed payload hashes"
	case activeInstalledPackSourceStateDriftHostRequirements:
		return "host requirements digest drift"
	case activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration:
		return "persisted runtime declaration drift"
	default:
		return "unknown source state drift"
	}
}

type activeInstalledPackSourceStateDrift struct {
	row                    installedPackRow
	pack                   *packs.SourceBackedPack
	runtimeDeclarationPack packs.Pack
	kind                   activeInstalledPackSourceStateDriftKind
	storedPayloadHash      string
	currentPayloadHash     string
	storedHostDigest       string
	currentHostDigest      string
}

func installPackPreflightOverridesForReapplySource(ctx context.Context, r *Runner, requestedRow installedPackRow) (*installPackPreflightOverrides, error) {
	return installPackPreflightOverridesForReapplySourceWithComponentKindRegistrationSet(ctx, r, requestedRow, nil, packs.ComponentKindRegistrationSet{})
}

func installPackPreflightOverridesForReapplySourceWithComponentKindRegistrationSet(ctx context.Context, r *Runner, requestedRow installedPackRow, requestedDependencies []packs.PackDependency, registrations packs.ComponentKindRegistrationSet) (*installPackPreflightOverrides, error) {
	if r == nil || r.DB == nil {
		return nil, fmt.Errorf("reapply-pack: database connection is unavailable")
	}
	rows, err := listInstalledAppPacksForReconcile(ctx, r.DB.DB())
	if err != nil {
		return nil, err
	}
	runtimeProjectionPeerAdmission := runtimeProjectionPeerAdmissionForRequestedRow(requestedRow, requestedDependencies)
	drifts, driftedPacks, err := detectActiveInstalledPackSourceStateDriftWithComponentKindRegistrationSet(
		ctx, r, rows, registrations, runtimeProjectionPeerAdmission,
	)
	if err != nil {
		return nil, err
	}
	logReapplySourcePeerDriftDiagnostics(requestedRow.Name, drifts)
	if runtimeProjectionPeerAdmission.pendingLocalPackName == "" {
		return &installPackPreflightOverrides{activeDriftedPacks: driftedPacks}, nil
	}
	plannedDesiredPacks := make(map[string]*packs.SourceBackedPack)
	runtimeProjectionCompatibility := make(map[string]packs.Pack)
	for _, drift := range drifts {
		name := strings.ToLower(strings.TrimSpace(drift.row.Name))
		switch {
		case drift.kind == activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration && runtimeProjectionPeerAdmission.inspectsRuntimeProjectionPeer(drift.row):
			if runtimeProjectionPeerAdmission.permits(name) {
				plannedDesiredPacks[name] = drift.pack
			}
			runtimeProjectionCompatibility[name] = drift.runtimeDeclarationPack
		case runtimeProjectionPeerAdmission.requested(name):
			plannedDesiredPacks[name] = drift.pack
		default:
			return nil, fmt.Errorf("reapply-pack pending local convergence found unsupported active peer drift for %q (%s)", drift.row.Name, drift.kind.String())
		}
	}
	plannedDesired, err := newInstallPackPreflightPackContext(plannedDesiredPacks)
	if err != nil {
		return nil, err
	}
	runtimeEffective, err := newInstallPackPreflightPackContext(map[string]*packs.SourceBackedPack{})
	if err != nil {
		return nil, err
	}
	runtimeCompatibility, err := newInstallPackPreflightRuntimeDeclarationContext(runtimeProjectionCompatibility)
	if err != nil {
		return nil, err
	}
	return &installPackPreflightOverrides{
		plannedDesiredPacks:                 plannedDesired,
		runtimeEffectivePacks:               runtimeEffective,
		runtimeProjectionCompatibilityPacks: runtimeCompatibility,
	}, nil
}

type runtimeProjectionPeerAdmission struct {
	pendingLocalPackName string
	dependencyNames      map[string]struct{}
}

func runtimeProjectionPeerAdmissionForRequestedRow(row installedPackRow, dependencies []packs.PackDependency) runtimeProjectionPeerAdmission {
	if !installedPackIsPendingLocalSourceConvergence(row) {
		return runtimeProjectionPeerAdmission{}
	}
	admission := runtimeProjectionPeerAdmission{
		pendingLocalPackName: strings.TrimSpace(row.Name),
		dependencyNames:      make(map[string]struct{}, len(dependencies)),
	}
	for _, dependency := range dependencies {
		if name := strings.ToLower(strings.TrimSpace(dependency.Name)); name != "" {
			admission.dependencyNames[name] = struct{}{}
		}
	}
	return admission
}

func (a runtimeProjectionPeerAdmission) requested(packName string) bool {
	return a.pendingLocalPackName != "" && strings.EqualFold(strings.TrimSpace(packName), a.pendingLocalPackName)
}

func (a runtimeProjectionPeerAdmission) permits(peerName string) bool {
	peerName = strings.ToLower(strings.TrimSpace(peerName))
	_, dependency := a.dependencyNames[peerName]
	return a.pendingLocalPackName != "" && !a.requested(peerName) && dependency
}

func (a runtimeProjectionPeerAdmission) inspectsRuntimeProjectionPeer(row installedPackRow) bool {
	if a.pendingLocalPackName == "" || a.requested(row.Name) ||
		row.Status != installedPackStatusActive || row.IsDeleted != 0 ||
		!row.SourceKind.Valid || !row.SourceURI.Valid {
		return false
	}
	if a.permits(row.Name) {
		return true
	}
	kind, _, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	return err == nil && kind == string(packs.PackSourceKindOfficialRepo)
}

func installedPackIsPendingLocalSourceConvergence(row installedPackRow) bool {
	if row.Status != installedPackStatusActive || row.IsDeleted != 0 ||
		!row.AppliedPayloadHash.Valid || !row.DesiredPayloadHash.Valid ||
		strings.TrimSpace(row.AppliedPayloadHash.String) == "" ||
		strings.TrimSpace(row.DesiredPayloadHash.String) == "" ||
		row.AppliedPayloadHash.String == row.DesiredPayloadHash.String ||
		!row.SourceKind.Valid || !row.SourceURI.Valid {
		return false
	}
	kind, _, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	return err == nil && kind == string(packs.PackSourceKindLocalPath)
}

func logReapplySourcePeerDriftDiagnostics(requestedPackName string, drifts []activeInstalledPackSourceStateDrift) {
	requestedPackName = strings.ToLower(strings.TrimSpace(requestedPackName))
	for _, drift := range drifts {
		if strings.ToLower(strings.TrimSpace(drift.row.Name)) == requestedPackName {
			continue
		}
		if strings.TrimSpace(drift.storedPayloadHash) != "" || strings.TrimSpace(drift.currentPayloadHash) != "" {
			log.Printf("controlplane: reapply-pack detected active peer source drift for %q (%s: stored=%q current=%q); run setup apply-source-convergence to reapply all drifted source-backed packs", drift.row.Name, drift.kind.String(), drift.storedPayloadHash, drift.currentPayloadHash)
			continue
		}
		if strings.TrimSpace(drift.storedHostDigest) != "" || strings.TrimSpace(drift.currentHostDigest) != "" {
			log.Printf("controlplane: reapply-pack detected active peer source drift for %q (%s: storedHostDigest=%q currentHostDigest=%q); run setup apply-source-convergence to reapply all drifted source-backed packs", drift.row.Name, drift.kind.String(), drift.storedHostDigest, drift.currentHostDigest)
			continue
		}
		log.Printf("controlplane: reapply-pack detected active peer source drift for %q (%s); run setup apply-source-convergence to reapply all drifted source-backed packs", drift.row.Name, drift.kind.String())
	}
}

type retryableInstalledPackSourceStateAdmission struct {
	row  installedPackRow
	pack *packs.SourceBackedPack
}

type sourceBackedPackReconcileLifecycle uint8

const (
	sourceBackedPackReconcileLifecycleActive sourceBackedPackReconcileLifecycle = iota + 1
	sourceBackedPackReconcileLifecycleRetry
)

// classifySourceBackedPackReconcileLifecycle is the single phase-7 lifecycle
// and source-authority classifier. Every enumerated row must be classified
// before any retry, repair, backfill, or schema mutation can run.
func classifySourceBackedPackReconcileLifecycle(
	row installedPackRow,
) (sourceBackedPackReconcileLifecycle, error) {
	var lifecycle sourceBackedPackReconcileLifecycle
	switch row.Status {
	case installedPackStatusActive:
		lifecycle = sourceBackedPackReconcileLifecycleActive
	case installedPackStatusInstalling, installedPackStatusFailed:
		lifecycle = sourceBackedPackReconcileLifecycleRetry
	default:
		return 0, fmt.Errorf(
			"%w: source-backed pack %q has unsupported reconcile lifecycle status %q",
			ErrPackGenerationPrecondition,
			row.Name,
			row.Status,
		)
	}

	if !installedPackHasSourceIdentity(row) {
		return 0, fmt.Errorf(
			"%w: source-backed pack %q status %q has no complete source identity",
			ErrPackGenerationPrecondition,
			row.Name,
			row.Status,
		)
	}
	sourceKind, _, err := canonicalPackSourceIdentity(
		row.SourceKind.String,
		row.SourceURI.String,
	)
	if err != nil {
		return 0, fmt.Errorf(
			"%w: source-backed pack %q status %q has invalid source identity: %v",
			ErrPackGenerationPrecondition,
			row.Name,
			row.Status,
			err,
		)
	}
	revisionRequired := lifecycle == sourceBackedPackReconcileLifecycleRetry ||
		packs.IsRevisionAddressedPackSourceKind(sourceKind)
	if revisionRequired &&
		(!row.SourceRevision.Valid || strings.TrimSpace(row.SourceRevision.String) == "") {
		return 0, fmt.Errorf(
			"%w: source-backed pack %q status %q has no complete revision-bound source authority",
			ErrPackGenerationPrecondition,
			row.Name,
			row.Status,
		)
	}
	return lifecycle, nil
}

func reconcileActiveInstalledPackSourceState(ctx context.Context, r *Runner, options InstallPackOptions) error {
	ctx = withOfflineCandidateOperationContext(ctx, options.offlineCandidate)
	rows, err := listInstalledAppPacksForReconcile(ctx, r.DB.DB())
	if err != nil {
		return err
	}
	registrations := r.componentKindRegistrationSet()
	if err := preflightPackGenerationForReconcile(ctx, r, rows, registrations); err != nil {
		return err
	}
	retries, err := preflightRetryableInstalledPackSourceStateForReconcile(ctx, r, rows, registrations)
	if err != nil {
		return err
	}
	// Retry admission proves the exact non-active attempts, but it does not
	// supersede the global ownership authority of already-active peers. Reject
	// incomplete managed-approval coverage before the first retry mutates
	// shared runtime state.
	if err := validateActiveInstalledPackManagedApprovalOwnership(ctx, r.DB.DB(), rows, registrations); err != nil {
		return err
	}
	for _, retry := range retries {
		retryOptions := options
		retryOptions.resumeSourceBackedPackInstallID = retry.row.PackInstallID
		retryPack, err := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcileWithComponentKindRegistrationSet(
			ctx,
			r.DB.DB(),
			retry.row,
			retry.pack,
			retryOptions,
			nil,
			registrations,
		)
		if err != nil {
			return fmt.Errorf(
				"prepare %s source-backed pack %s exact installation %s for resume: %w",
				retry.row.Status,
				retry.row.Name,
				retry.row.PackInstallID,
				err,
			)
		}
		if err := applyResolvedSourceBackedPackUnlockedForReconcile(
			ctx,
			r,
			retryPack,
			nil,
			retryOptions,
		); err != nil {
			return fmt.Errorf(
				"resume %s source-backed pack %s from exact installation %s: %w",
				retry.row.Status,
				retry.row.Name,
				retry.row.PackInstallID,
				err,
			)
		}
	}
	if len(retries) > 0 {
		rows, err = listInstalledAppPacksForReconcile(ctx, r.DB.DB())
		if err != nil {
			return err
		}
		if err := preflightPackGenerationForReconcile(ctx, r, rows, registrations); err != nil {
			return err
		}
		// Successful attempts are now active and therefore join the same global
		// managed-approval ownership authority.
		if err := validateActiveInstalledPackManagedApprovalOwnership(ctx, r.DB.DB(), rows, registrations); err != nil {
			return err
		}
	}
	if options.allowInstalledPackSourceStateRepair {
		if err := repairActiveInstalledPackSourceHashDrift(ctx, r, rows, options); err != nil {
			return err
		}
	}
	if updated, err := backfillChildCloneRulePackOwnershipForReconcile(ctx, r.DB.DB(), registrations); err != nil {
		return fmt.Errorf("backfill object_spec childCloneRules pack ownership: %w", err)
	} else if updated > 0 {
		log.Printf("repair: backfilled ChildCloneRules pack ownership rows=%d", updated)
	}
	if updated, err := backfillExclusiveSiblingFieldPackOwnershipForReconcile(ctx, r.DB.DB(), registrations); err != nil {
		return fmt.Errorf("backfill object_spec exclusiveSiblingFields pack ownership: %w", err)
	} else if updated > 0 {
		log.Printf("repair: backfilled ExclusiveSiblingFields pack ownership rows=%d", updated)
	}
	rows, err = listInstalledAppPacksForReconcile(ctx, r.DB.DB())
	if err != nil {
		return err
	}
	return verifyActiveInstalledPackSourceState(ctx, r, rows, options)
}

// preflightRetryableInstalledPackSourceState admits only durable source-backed
// attempts that phase 7 itself can safely resume after a stopped process:
// installing/failed, exact deterministic row identity and incarnation, and a
// revision-bound source payload equal to DesiredPayloadHash. For an
// official_repo candidate, either no generation authority or the exact active
// immutable generation is also required. A Git-backed local_path retry remains
// retryable but never becomes Catalog publication authority.
func preflightRetryableInstalledPackSourceState(
	ctx context.Context,
	r *Runner,
	rows []installedPackRow,
	registrations packs.ComponentKindRegistrationSet,
) ([]retryableInstalledPackSourceStateAdmission, error) {
	db := runnerSQLDB(r)
	formulaFunctionBundlesForValidation := &activeFormulaFunctionBundlesValidationContext{}
	admissions := make([]retryableInstalledPackSourceStateAdmission, 0)
	for _, summary := range rows {
		lifecycle, err := classifySourceBackedPackReconcileLifecycle(summary)
		if err != nil {
			return nil, err
		}
		if lifecycle == sourceBackedPackReconcileLifecycleActive {
			continue
		}
		row, exists, err := getInstalledPackRowByNameForSourceRevisionAdvance(
			ctx,
			db,
			summary.Name,
		)
		if err != nil {
			return nil, err
		}
		if !exists {
			return nil, fmt.Errorf(
				"source-backed pack %s retry row disappeared during phase-7 admission",
				summary.Name,
			)
		}
		copyInstalledPackRuntimeDeclarationState(&row, summary)
		pack, err := sourceBackedPackFromInstalledRowForReconcileWithFormulaBundlesAndComponentKindRegistrationSet(
			ctx,
			db,
			row,
			formulaFunctionBundlesForValidation,
			registrations,
		)
		if err != nil {
			return nil, fmt.Errorf(
				"source-backed pack %s retry source: %w",
				row.Name,
				err,
			)
		}
		payloadHash, err := packs.PackPayloadHash(pack)
		if err != nil {
			return nil, fmt.Errorf("source-backed pack %s retry payload hash: %w", row.Name, err)
		}
		if err := verifyRetryableInstalledPackSourceState(row, pack, payloadHash); err != nil {
			return nil, err
		}
		publishGeneration, err := packPublishesImmutableGeneration(pack)
		if err != nil {
			return nil, fmt.Errorf(
				"source-backed pack %s retry publication eligibility: %w",
				row.Name,
				err,
			)
		}
		if publishGeneration {
			generationInstallID, err := activatedSourceBackedPackRepairInstallID(
				ctx,
				db,
				appliedAppPack{pack: pack, payloadHash: payloadHash},
			)
			if err != nil {
				return nil, fmt.Errorf(
					"source-backed pack %s retry immutable generation authority: %w",
					row.Name,
					err,
				)
			}
			if generationInstallID != "" && generationInstallID != row.PackInstallID {
				return nil, fmt.Errorf(
					"%w: source-backed pack %q retry incarnation %q differs from active generation %q",
					ErrPackGenerationConflict,
					row.Name,
					row.PackInstallID,
					generationInstallID,
				)
			}
		} else if err := verifyMutableSourceHasNoPublishedGenerationAuthority(
			ctx,
			db,
			row.Name,
		); err != nil {
			return nil, fmt.Errorf(
				"source-backed pack %s retry mutable-source authority: %w",
				row.Name,
				err,
			)
		}
		admissions = append(admissions, retryableInstalledPackSourceStateAdmission{
			row:  row,
			pack: pack,
		})
	}
	return admissions, nil
}

func copyInstalledPackRuntimeDeclarationState(
	target *installedPackRow,
	source installedPackRow,
) {
	target.RuntimeObjectSpecDecl = source.RuntimeObjectSpecDecl
	target.RuntimeRelatedLists = source.RuntimeRelatedLists
	target.RuntimeSearchScope = source.RuntimeSearchScope
	target.RuntimeSearchAllDefault = source.RuntimeSearchAllDefault
	target.RuntimeFormulaBundles = source.RuntimeFormulaBundles
	target.RuntimeStartupShallow = source.RuntimeStartupShallow
	target.HasRuntimeFormula = source.HasRuntimeFormula
	target.HasRuntimeStartupDecl = source.HasRuntimeStartupDecl
}

func verifyRetryableInstalledPackSourceState(
	row installedPackRow,
	pack *packs.SourceBackedPack,
	payloadHash string,
) error {
	if pack == nil ||
		strings.TrimSpace(row.ID) != installedPackID(row.Name) ||
		strings.TrimSpace(row.PackInstallID) == "" ||
		row.IsDeleted != 0 ||
		(row.Status != installedPackStatusInstalling && row.Status != installedPackStatusFailed) ||
		strings.TrimSpace(row.Name) != strings.TrimSpace(pack.Name()) ||
		strings.TrimSpace(row.Category) != strings.TrimSpace(string(pack.Category())) {
		return fmt.Errorf(
			"%w: source-backed pack %q retry lifecycle/incarnation tuple is invalid",
			ErrPackGenerationPrecondition,
			row.Name,
		)
	}
	meta := packs.PackHashMetadataForPack(pack)
	if !row.DesiredPayloadHash.Valid ||
		strings.TrimSpace(row.DesiredPayloadHash.String) != strings.TrimSpace(payloadHash) ||
		!packs.IsPackPayloadHashV1(payloadHash) ||
		!installedPackCommitNullStringEquals(row.PackVersion, meta.PackVersion, true) ||
		!installedPackCommitNullStringEquals(row.ManifestFormatVersion, meta.ManifestFormatVersion, true) ||
		!installedPackCommitNullStringEquals(row.PlatformCompatibility, meta.PlatformCompatibility, meta.PlatformCompatibilityValid) ||
		!installedPackCommitNullStringEquals(row.PayloadHashAlgorithm, meta.PayloadHashAlgorithm, true) ||
		!installedPackCommitNullStringEquals(row.SourceKind, meta.SourceKind, meta.SourceKindValid) ||
		!installedPackCommitNullStringEquals(row.SourceURI, meta.SourceURI, meta.SourceURIValid) ||
		!installedPackCommitNullStringEquals(row.SourceRevision, meta.SourceRevision, meta.SourceRevisionValid) {
		return fmt.Errorf(
			"%w: source-backed pack %q retry facts do not match the exact revision-bound source",
			ErrPackGenerationConflict,
			row.Name,
		)
	}
	if err := verifyInstalledPackHostRequirementsState(row, pack); err != nil {
		return fmt.Errorf(
			"%w: source-backed pack %q retry host requirements: %v",
			ErrPackGenerationConflict,
			row.Name,
			err,
		)
	}
	if err := verifyInstalledPersistedRuntimeDeclarationState(row, pack, true); err != nil {
		return fmt.Errorf(
			"%w: source-backed pack %q retry Runtime declaration: %v",
			ErrPackGenerationConflict,
			row.Name,
			err,
		)
	}
	return nil
}

// preflightActiveInstalledPackGenerationAuthorities freezes the immutable
// authority decision before phase-7 reconciliation performs any repair,
// ownership backfill, compatibility convergence, or physical-schema mutation.
// A legacy active pack with no authority is admissible for explicit stopped
// upgrade adoption. Exact authority is admissible for read-only verification
// or same-generation repair. Mixed or different authority fails closed here.
func preflightActiveInstalledPackGenerationAuthorities(
	ctx context.Context,
	r *Runner,
	rows []installedPackRow,
	registrations packs.ComponentKindRegistrationSet,
) error {
	db := runnerSQLDB(r)
	formulaFunctionBundlesForValidation := &activeFormulaFunctionBundlesValidationContext{}
	for _, row := range rows {
		lifecycle, err := classifySourceBackedPackReconcileLifecycle(row)
		if err != nil {
			return err
		}
		if lifecycle != sourceBackedPackReconcileLifecycleActive {
			continue
		}
		authorityRow, exists, err := getInstalledPackRowByNameForSourceRevisionAdvance(
			ctx,
			db,
			row.Name,
		)
		if err != nil {
			return fmt.Errorf("active source-backed pack %s generation preflight compatibility row: %w", row.Name, err)
		}
		if !exists {
			return fmt.Errorf(
				"%w: active source-backed pack %q compatibility row disappeared during generation preflight",
				ErrPackGenerationConflict,
				row.Name,
			)
		}
		row = authorityRow
		pack, err := sourceBackedPackFromInstalledRowForReconcileWithFormulaBundlesAndComponentKindRegistrationSet(
			ctx,
			db,
			row,
			formulaFunctionBundlesForValidation,
			registrations,
		)
		if err != nil {
			return fmt.Errorf("active source-backed pack %s generation preflight: %w", row.Name, err)
		}
		publishGeneration, err := packPublishesImmutableGeneration(pack)
		if err != nil {
			return fmt.Errorf(
				"active source-backed pack %s generation publication eligibility: %w",
				row.Name,
				err,
			)
		}
		if !publishGeneration {
			if err := verifyMutableSourceHasNoPublishedGenerationAuthority(
				ctx,
				db,
				row.Name,
			); err != nil {
				return fmt.Errorf(
					"active source-backed pack %s mutable-source authority preflight: %w",
					row.Name,
					err,
				)
			}
			continue
		}
		payloadHash, err := packs.PackPayloadHash(pack)
		if err != nil {
			return fmt.Errorf("active source-backed pack %s generation preflight payload hash: %w", row.Name, err)
		}
		generationInstallID, err := verifySourceBackedPackGenerationAuthorityForInstalledRow(
			ctx,
			db,
			appliedAppPack{pack: pack, payloadHash: payloadHash},
			row,
		)
		if err != nil {
			return fmt.Errorf("active source-backed pack %s immutable generation preflight: %w", row.Name, err)
		}
		if generationInstallID != "" && generationInstallID != row.PackInstallID {
			return fmt.Errorf(
				"%w: active source-backed pack %q compatibility incarnation %q differs from active generation %q",
				ErrPackGenerationConflict,
				row.Name,
				row.PackInstallID,
				generationInstallID,
			)
		}
	}
	return nil
}

func repairActiveInstalledPackSourceHashDrift(ctx context.Context, r *Runner, rows []installedPackRow, options InstallPackOptions) error {
	registrations := r.componentKindRegistrationSet()
	drifts, driftedPacks, err := detectActiveInstalledPackSourceStateDriftWithComponentKindRegistrationSet(ctx, r, rows, registrations, runtimeProjectionPeerAdmission{})
	if err != nil {
		return err
	}
	overrides := &installPackPreflightOverrides{activeDriftedPacks: driftedPacks}
	for _, drift := range drifts {
		row := drift.row
		pack := drift.pack
		missingCols, inspectErr := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile(ctx, r.DB.DB(), row.Name)
		if inspectErr != nil {
			return fmt.Errorf("inspect live physical columns for pack %s: %w", row.Name, inspectErr)
		}
		if len(missingCols) > 0 {
			// Attempt metadata-driven repair before failing closed (#1392).
			// Rebuilds missing physical columns from live _System_Field rows
			// for pack-owned fields, then re-inspects.
			repaired, repairErr := repairPackOwnedMissingPhysicalColumnsForReconcile(ctx, r.DB.DB(), row.Name)
			if repairErr != nil {
				return fmt.Errorf("repair pack-owned missing physical columns for %s: %w", row.Name, repairErr)
			}
			if len(repaired) > 0 {
				log.Printf("controlplane: repaired %d pack-owned missing physical column(s) for %s: %v", len(repaired), row.Name, repaired)
				// Re-inspect after repair.
				missingCols, inspectErr = inspectLiveInstalledPackMissingPhysicalColumnsForReconcile(ctx, r.DB.DB(), row.Name)
				if inspectErr != nil {
					return fmt.Errorf("re-inspect live physical columns for pack %s after repair: %w", row.Name, inspectErr)
				}
			}
			if len(missingCols) > 0 {
				return &MissingPhysicalColumnDetectOnlyError{PackName: row.Name, MissingColumns: missingCols}
			}
		}
		var rebuildErr error
		pack, rebuildErr = rebuildRepairPackWithAllowedOwnerlessObjectsForReconcileWithComponentKindRegistrationSet(ctx, r.DB.DB(), row, pack, options, overrides, registrations)
		if rebuildErr != nil {
			return fmt.Errorf("reapply source-backed pack %s ownerless allowance: %w", row.Name, rebuildErr)
		}
		if converged, err := tryConvergeVerifiedInstalledPackSourceStateWithoutReapply(ctx, r, row, pack, drift); err != nil {
			return err
		} else if converged {
			continue
		}
		logSourceConvergenceDriftDiagnostic(drift)
		if err := applyResolvedSourceBackedPackUnlockedForReconcile(ctx, r, pack, overrides, options); err != nil {
			return fmt.Errorf("reapply source-backed pack %s after convergence drift: %w", row.Name, err)
		}
	}
	return nil
}

func tryConvergeVerifiedInstalledPackSourceStateWithoutReapply(ctx context.Context, r *Runner, row installedPackRow, pack *packs.SourceBackedPack, drift activeInstalledPackSourceStateDrift) (bool, error) {
	if drift.kind != activeInstalledPackSourceStateDriftInconsistentHashes {
		return false, nil
	}
	if r == nil || r.DB == nil || pack == nil {
		return false, nil
	}
	// Exact-set approval ownership still requires the apply path after payload
	// hash drift so reconciliation, effects, and evidence run before bookkeeping
	// convergence, even when the fresh complete-set projection already matches.
	if pack.Category() == packs.AppPack && r.componentKindRegistrationSet().HasManagedOwnerPackTable(packs.ComponentKindApprovalProcess, "ProcessDefinition") {
		return false, nil
	}
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		return false, fmt.Errorf("pack %s payload hash for verified convergence: %w", row.Name, err)
	}
	if !canConvergeInconsistentInstalledPayloadHashesWithoutReapply(row, payloadHash) {
		return false, nil
	}
	db := r.DB.DB()
	if err := verifyResolvedSourceBackedPackForReconcile(ctx, row, db, pack); err != nil {
		log.Printf("controlplane: source-backed convergence drift for %q still needs reapply; live verify did not match current source: %v", row.Name, err)
		return false, nil
	}
	if err := markVerifiedInstalledPackConvergedForReconcile(ctx, db, pack, payloadHash); err != nil {
		return false, fmt.Errorf("converge verified installed pack %s hash state: %w", row.Name, err)
	}
	log.Printf("controlplane: converged installed pack %q after source-backed bookkeeping drift without reapply (%s: stored=%q current=%q)", row.Name, drift.kind.String(), drift.storedPayloadHash, payloadHash)
	return true, nil
}

func canConvergeInconsistentInstalledPayloadHashesWithoutReapply(row installedPackRow, currentPayloadHash string) bool {
	current := strings.TrimSpace(currentPayloadHash)
	applied := strings.TrimSpace(row.AppliedPayloadHash.String)
	desired := strings.TrimSpace(row.DesiredPayloadHash.String)
	if current == "" || applied == "" || desired == "" {
		return false
	}
	if desired != current || applied == desired {
		return false
	}
	syntheticStale, ok := deriveSyntheticStalePayloadHashForCurrent(current)
	return ok && applied == syntheticStale
}

func deriveSyntheticStalePayloadHashForCurrent(currentPayloadHash string) (string, bool) {
	current := strings.TrimSpace(currentPayloadHash)
	if !strings.HasPrefix(current, packs.PackPayloadHashV1Prefix) {
		return "", false
	}
	digest := current[len(packs.PackPayloadHashV1Prefix):]
	if len(digest) != 64 {
		return "", false
	}
	for _, char := range digest {
		if !((char >= '0' && char <= '9') || (char >= 'a' && char <= 'f')) {
			return "", false
		}
	}
	replacement := byte('0')
	if digest[len(digest)-1] == '0' {
		replacement = '1'
	}
	return packs.PackPayloadHashV1Prefix + digest[:len(digest)-1] + string(replacement), true
}

func logSourceConvergenceDriftDiagnostic(drift activeInstalledPackSourceStateDrift) {
	if strings.TrimSpace(drift.storedPayloadHash) != "" || strings.TrimSpace(drift.currentPayloadHash) != "" {
		log.Printf("controlplane: reapplying installed pack %q after source-backed convergence drift (%s: stored=%q current=%q)", drift.row.Name, drift.kind.String(), drift.storedPayloadHash, drift.currentPayloadHash)
		return
	}
	if strings.TrimSpace(drift.storedHostDigest) != "" || strings.TrimSpace(drift.currentHostDigest) != "" {
		log.Printf("controlplane: reapplying installed pack %q after source-backed convergence drift (%s: storedHostDigest=%q currentHostDigest=%q)", drift.row.Name, drift.kind.String(), drift.storedHostDigest, drift.currentHostDigest)
		return
	}
	log.Printf("controlplane: reapplying installed pack %q after source-backed convergence drift (%s)", drift.row.Name, drift.kind.String())
}

func detectActiveInstalledPackSourceStateDrift(ctx context.Context, r *Runner, rows []installedPackRow) ([]activeInstalledPackSourceStateDrift, map[string]*packs.SourceBackedPack, error) {
	return detectActiveInstalledPackSourceStateDriftWithComponentKindRegistrationSet(ctx, r, rows, r.componentKindRegistrationSet(), runtimeProjectionPeerAdmission{})
}

func detectActiveInstalledPackSourceStateDriftWithComponentKindRegistrationSet(ctx context.Context, r *Runner, rows []installedPackRow, registrations packs.ComponentKindRegistrationSet, runtimeProjectionPeerAdmission runtimeProjectionPeerAdmission) ([]activeInstalledPackSourceStateDrift, map[string]*packs.SourceBackedPack, error) {
	db := runnerSQLDB(r)
	drifts := make([]activeInstalledPackSourceStateDrift, 0)
	driftedPacks := make(map[string]*packs.SourceBackedPack)
	deferredRows := make([]installedPackRow, 0)
	for _, row := range rows {
		if row.Status != installedPackStatusActive || !installedPackHasSourceIdentity(row) {
			continue
		}
		pack, err := sourceBackedPackFromInstalledRowForReapplyPreflight(ctx, db, row, registrations)
		if err != nil {
			if isMissingRuntimeFormulaFunctionError(err) {
				deferredRows = append(deferredRows, row)
				continue
			}
			return nil, nil, fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
		}
		drift, ok, err := detectInstalledPackSourceStateDriftForContext(row, pack, runtimeProjectionPeerAdmission)
		if err != nil {
			return nil, nil, fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
		}
		if ok {
			if err := appendAdmittedActiveInstalledPackSourceStateDrift(ctx, db, &drifts, driftedPacks, drift); err != nil {
				return nil, nil, fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
			}
		}
	}
	if len(deferredRows) > 0 {
		for len(deferredRows) > 0 {
			formulaFunctionBundlesForValidation, err := formulaFunctionBundlesValidationContextForReconcileDriftedPacks(ctx, db, driftedPacks)
			if err != nil {
				return nil, nil, err
			}
			nextDeferredRows := make([]installedPackRow, 0, len(deferredRows))
			progressed := false
			var firstDeferredErr error
			for _, row := range deferredRows {
				pack, err := sourceBackedPackFromInstalledRowForReapplyPreflightWithFormulaBundles(ctx, db, row, formulaFunctionBundlesForValidation, registrations)
				if err != nil {
					if isMissingRuntimeFormulaFunctionError(err) {
						if firstDeferredErr == nil {
							firstDeferredErr = err
						}
						nextDeferredRows = append(nextDeferredRows, row)
						continue
					}
					return nil, nil, fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
				}
				progressed = true
				drift, ok, err := detectInstalledPackSourceStateDriftForContext(row, pack, runtimeProjectionPeerAdmission)
				if err != nil {
					return nil, nil, fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
				}
				if ok {
					if err := appendAdmittedActiveInstalledPackSourceStateDrift(ctx, db, &drifts, driftedPacks, drift); err != nil {
						return nil, nil, fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
					}
				}
			}
			if len(nextDeferredRows) == 0 {
				break
			}
			if !progressed {
				return nil, nil, fmt.Errorf("active source-backed pack %s: %w", nextDeferredRows[0].Name, firstDeferredErr)
			}
			deferredRows = nextDeferredRows
		}
	}
	return drifts, driftedPacks, nil
}

func appendAdmittedActiveInstalledPackSourceStateDrift(ctx context.Context, db *sql.DB, drifts *[]activeInstalledPackSourceStateDrift, driftedPacks map[string]*packs.SourceBackedPack, drift activeInstalledPackSourceStateDrift) error {
	if drift.kind == activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration {
		runtimePack, err := loadActivePackGenerationRuntimeForInstalledRow(ctx, db, drift.row.Name, drift.pack)
		if err != nil {
			return err
		}
		drift.runtimeDeclarationPack = runtimePack
	}
	appendActiveInstalledPackSourceStateDrift(drifts, driftedPacks, drift)
	return nil
}

func appendActiveInstalledPackSourceStateDrift(drifts *[]activeInstalledPackSourceStateDrift, driftedPacks map[string]*packs.SourceBackedPack, drift activeInstalledPackSourceStateDrift) {
	key := strings.ToLower(strings.TrimSpace(drift.row.Name))
	if key == "" {
		return
	}
	driftedPacks[key] = drift.pack
	*drifts = append(*drifts, drift)
}

func detectInstalledPackSourceStateDriftForContext(row installedPackRow, pack *packs.SourceBackedPack, runtimeProjectionPeerAdmission runtimeProjectionPeerAdmission) (activeInstalledPackSourceStateDrift, bool, error) {
	drift, ok, err := detectInstalledPackSourceStateDrift(row, pack)
	if err != nil {
		return drift, false, err
	}
	if ok {
		if runtimeProjectionPeerAdmission.permits(row.Name) {
			return activeInstalledPackSourceStateDrift{}, false, fmt.Errorf("runtime projection dependency peer %q is not source/hash/host exact (%s)", row.Name, drift.kind.String())
		}
		return drift, true, nil
	}
	if !runtimeProjectionPeerAdmission.inspectsRuntimeProjectionPeer(row) {
		return activeInstalledPackSourceStateDrift{}, false, nil
	}
	return detectInstalledPackRuntimeProjectionPeerDrift(row, pack)
}

func detectInstalledPackSourceStateDrift(row installedPackRow, pack *packs.SourceBackedPack) (activeInstalledPackSourceStateDrift, bool, error) {
	if err := verifyInstalledPackHashStateForReconcile(row, pack); err != nil {
		var hashDrift *HashDriftError
		if errors.As(err, &hashDrift) {
			return activeInstalledPackSourceStateDrift{
				row:                row,
				pack:               pack,
				kind:               activeInstalledPackSourceStateDriftPayloadHash,
				storedPayloadHash:  hashDrift.StoredHash,
				currentPayloadHash: hashDrift.CurrentHash,
			}, true, nil
		}
		if shouldReapplyForInconsistentInstalledPackHashes(row, err) {
			currentHash := ""
			if pack != nil {
				if hash, hashErr := packs.PackPayloadHash(pack); hashErr == nil {
					currentHash = hash
				}
			}
			return activeInstalledPackSourceStateDrift{
				row:                row,
				pack:               pack,
				kind:               activeInstalledPackSourceStateDriftInconsistentHashes,
				storedPayloadHash:  strings.TrimSpace(row.AppliedPayloadHash.String),
				currentPayloadHash: currentHash,
			}, true, nil
		}
		return activeInstalledPackSourceStateDrift{}, false, err
	}
	if err := verifyInstalledPackHostRequirementsState(row, pack); err != nil {
		var hostDrift *HostRequirementsDigestDriftError
		if !errors.As(err, &hostDrift) {
			return activeInstalledPackSourceStateDrift{}, false, err
		}
		return activeInstalledPackSourceStateDrift{
			row:               row,
			pack:              pack,
			kind:              activeInstalledPackSourceStateDriftHostRequirements,
			storedHostDigest:  hostDrift.StoredDigest,
			currentHostDigest: hostDrift.CurrentDigest,
		}, true, nil
	}
	return activeInstalledPackSourceStateDrift{}, false, nil
}

func detectInstalledPackRuntimeProjectionPeerDrift(row installedPackRow, pack *packs.SourceBackedPack) (activeInstalledPackSourceStateDrift, bool, error) {
	matches, err := classifyCompleteInstalledPersistedRuntimeDeclarationState(row, pack)
	if err == nil && matches {
		return activeInstalledPackSourceStateDrift{}, false, nil
	}
	if exactErr := verifyInstalledPackRuntimeProjectionPeerExactSource(row, pack); exactErr != nil {
		return activeInstalledPackSourceStateDrift{}, false, exactErr
	}
	if err != nil {
		declaration, declarationErr := completeInstalledPersistedRuntimeDeclaration(row)
		if declarationErr != nil {
			return activeInstalledPackSourceStateDrift{}, false, declarationErr
		}
		if _, compatibilityErr := declaration.LoadPackWithLegacyConnectorCalloutProfileProjectionSource(row.Name, pack); compatibilityErr != nil {
			return activeInstalledPackSourceStateDrift{}, false, err
		}
	}
	return activeInstalledPackSourceStateDrift{
		row:  row,
		pack: pack,
		kind: activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration,
	}, true, nil
}

func verifyInstalledPackRuntimeProjectionPeerExactSource(row installedPackRow, pack *packs.SourceBackedPack) error {
	meta := packs.PackHashMetadataForPack(pack)
	kind, _, identityErr := canonicalPackSourceIdentity(meta.SourceKind, meta.SourceURI)
	if identityErr != nil || kind != string(packs.PackSourceKindOfficialRepo) {
		return fmt.Errorf("installed source-backed pack %s runtime projection peer is not an official_repo source", row.Name)
	}
	if !sourceIdentityMatches(row, meta) {
		return fmt.Errorf("installed source-backed pack %s runtime projection peer source identity is not exact", row.Name)
	}
	hostDigest, err := packs.PackHostRequirementsDigest(pack)
	if err != nil {
		return err
	}
	if !row.HostRequirementsDigest.Valid || strings.TrimSpace(row.HostRequirementsDigest.String) != hostDigest {
		return fmt.Errorf("installed source-backed pack %s runtime projection peer host requirements digest is not exact", row.Name)
	}
	return nil
}

func shouldReapplyForInconsistentInstalledPackHashes(row installedPackRow, err error) bool {
	if err == nil || !installedPackHasSourceIdentity(row) {
		return false
	}
	if !row.AppliedPayloadHash.Valid || !row.DesiredPayloadHash.Valid {
		return false
	}
	applied := strings.TrimSpace(row.AppliedPayloadHash.String)
	desired := strings.TrimSpace(row.DesiredPayloadHash.String)
	if applied == "" || desired == "" || applied == desired {
		return false
	}
	return strings.Contains(err.Error(), "active hash state is inconsistent")
}

func verifyActiveInstalledPackSourceState(ctx context.Context, r *Runner, rows []installedPackRow, options InstallPackOptions) error {
	db := runnerSQLDB(r)
	registrations := r.componentKindRegistrationSet()
	formulaFunctionBundlesForValidation := &activeFormulaFunctionBundlesValidationContext{}
	for _, row := range rows {
		if row.Status != installedPackStatusActive || !installedPackHasSourceIdentity(row) {
			continue
		}
		pack, err := sourceBackedPackFromInstalledRowForReconcileWithFormulaBundlesAndComponentKindRegistrationSet(ctx, db, row, formulaFunctionBundlesForValidation, registrations)
		if err != nil {
			return fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
		}
		if err := verifyInstalledPackHashStateForReconcile(row, pack); err != nil {
			return fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
		}
		if err := verifyInstalledPackHostRequirementsState(row, pack); err != nil {
			return fmt.Errorf("active source-backed pack %s: %w", row.Name, err)
		}
		verifyErr := verifyResolvedSourceBackedPackForReconcile(ctx, row, db, pack)
		if verifyErr != nil && options.replayStandardActionPersistenceSemantics &&
			sourceBackedPackHasComponentKind(pack, packs.ComponentKindInvocableAction) &&
			isStandardActionPersistenceReplayError(verifyErr) {
			pack, err = rebuildRepairPackWithAllowedOwnerlessObjectsAndRegistrationsForReconcile(
				ctx, db, row, pack, options, nil, registrations,
			)
			if err != nil {
				return fmt.Errorf("active source-backed pack %s standard-action persistence replay preparation: %w", row.Name, err)
			}
			if err := applyResolvedSourceBackedPackUnlockedForReconcile(ctx, r, pack, nil, options); err != nil {
				return fmt.Errorf("replay active source-backed pack %s standard-action persistence from %s: %w", row.Name, installedPackSourceLabel(row), err)
			}
			continue
		}
		if verifyErr != nil {
			if !options.allowInstalledPackSourceStateRepair {
				return fmt.Errorf("active source-backed pack %s verify: %w", row.Name, verifyErr)
			}
			if !isRepairableInstalledPackApplyAllError(verifyErr) {
				return fmt.Errorf("active source-backed pack %s verify: %w", row.Name, verifyErr)
			}
			pack, err = rebuildRepairPackWithAllowedOwnerlessObjectsForReconcileWithComponentKindRegistrationSet(ctx, db, row, pack, options, nil, registrations)
			if err != nil {
				return fmt.Errorf("active source-backed pack %s ownerless allowance: %w", row.Name, err)
			}
			if err := applyResolvedSourceBackedPackUnlockedForReconcile(ctx, r, pack, nil, options); err != nil {
				return fmt.Errorf("reconcile active source-backed pack %s from %s: %w", row.Name, installedPackSourceLabel(row), err)
			}
			continue
		}
		publishGeneration, err := packPublishesImmutableGeneration(pack)
		if err != nil {
			return fmt.Errorf(
				"active source-backed pack %s generation publication eligibility: %w",
				row.Name,
				err,
			)
		}
		if !publishGeneration {
			continue
		}
		payloadHash, err := packs.PackPayloadHash(pack)
		if err != nil {
			return fmt.Errorf("active source-backed pack %s generation payload hash: %w", row.Name, err)
		}
		if err := publishFirstPackGenerationForReconcile(
			ctx,
			db,
			appliedAppPack{pack: pack, payloadHash: payloadHash},
		); err != nil {
			return fmt.Errorf(
				"active source-backed pack %s publish immutable generation: %w",
				row.Name,
				err,
			)
		}
	}
	return nil
}

func isStandardActionPersistenceReplayError(err error) bool {
	componentErrs := packs.ComponentVerifyErrorList(err)
	if len(componentErrs) == 0 {
		return false
	}
	for _, componentErr := range componentErrs {
		if componentErr == nil || componentErr.Kind != packs.ComponentKindInvocableAction ||
			!packs.IsStandardActionPersistenceDrift(componentErr.Err) {
			return false
		}
	}
	return true
}

type activeFormulaFunctionBundlesValidationContext struct {
	loaded  bool
	bundles []formula.FunctionBundle
	has     bool
	err     error
}

func (c *activeFormulaFunctionBundlesValidationContext) load(ctx context.Context, db *sql.DB) ([]formula.FunctionBundle, bool, error) {
	if c == nil {
		return nil, false, nil
	}
	if c.loaded {
		return c.bundles, c.has, c.err
	}
	c.loaded = true
	c.bundles, c.has, c.err = packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidation(ctx, db)
	return c.bundles, c.has, c.err
}

func sourceBackedPackFromInstalledRowForReconcileWithFormulaBundles(ctx context.Context, db *sql.DB, row installedPackRow, formulaFunctionBundlesForValidation *activeFormulaFunctionBundlesValidationContext, opts ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
	return sourceBackedPackFromInstalledRowForReconcileWithFormulaBundlesAndComponentKindRegistrationSet(
		ctx, db, row, formulaFunctionBundlesForValidation, packs.ComponentKindRegistrationSet{}, opts...,
	)
}

func sourceBackedPackFromInstalledRowForReconcileWithFormulaBundlesAndComponentKindRegistrationSet(ctx context.Context, db *sql.DB, row installedPackRow, formulaFunctionBundlesForValidation *activeFormulaFunctionBundlesValidationContext, registrations packs.ComponentKindRegistrationSet, opts ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
	if len(opts) == 0 {
		if candidatePack, ok, err := offlineCandidatePackFromContext(ctx, row); err != nil || ok {
			return candidatePack, err
		}
	}
	if len(registrations.Registrations()) == 0 {
		pack, err := sourceBackedPackFromInstalledRowForReconcile(ctx, db, row, opts...)
		if err == nil || !isMissingRuntimeFormulaFunctionError(err) {
			return pack, err
		}
		bundles, hasBundles, loadErr := formulaFunctionBundlesForValidation.load(ctx, db)
		if loadErr != nil {
			return nil, fmt.Errorf("active runtime formula function bundles: %w", loadErr)
		}
		if hasBundles {
			return sourceBackedPackFromInstalledRowWithFormulaBundles(ctx, db, row, bundles, true, opts...)
		}
		return nil, err
	}

	pack, err := sourceBackedPackFromInstalledRowWithComponentKindRegistrationSet(ctx, db, row, registrations, opts...)
	if err == nil || !isMissingRuntimeFormulaFunctionError(err) {
		return pack, err
	}
	bundles, hasBundles, loadErr := formulaFunctionBundlesForValidation.load(ctx, db)
	if loadErr != nil {
		return nil, fmt.Errorf("active runtime formula function bundles: %w", loadErr)
	}
	if hasBundles {
		return sourceBackedPackFromInstalledRowWithFormulaBundlesAndComponentKindRegistrationSet(ctx, db, row, bundles, true, registrations, opts...)
	}
	return nil, err
}

func sourceBackedPackFromInstalledRowForReapplyPreflight(ctx context.Context, db *sql.DB, row installedPackRow, registrations packs.ComponentKindRegistrationSet) (*packs.SourceBackedPack, error) {
	if candidatePack, ok, err := offlineCandidatePackFromContext(ctx, row); err != nil || ok {
		return candidatePack, err
	}
	if len(registrations.Registrations()) == 0 {
		return sourceBackedPackFromInstalledRowForReconcile(ctx, db, row)
	}
	return sourceBackedPackFromInstalledRowWithComponentKindRegistrationSet(ctx, db, row, registrations)
}

func sourceBackedPackFromInstalledRowForReapplyPreflightWithFormulaBundles(ctx context.Context, db *sql.DB, row installedPackRow, formulaFunctionBundlesForValidation *activeFormulaFunctionBundlesValidationContext, registrations packs.ComponentKindRegistrationSet) (*packs.SourceBackedPack, error) {
	if candidatePack, ok, err := offlineCandidatePackFromContext(ctx, row); err != nil || ok {
		return candidatePack, err
	}
	if len(registrations.Registrations()) == 0 {
		return sourceBackedPackFromInstalledRowForReconcileWithFormulaBundles(ctx, db, row, formulaFunctionBundlesForValidation)
	}
	pack, err := sourceBackedPackFromInstalledRowWithComponentKindRegistrationSet(ctx, db, row, registrations)
	if err == nil || !isMissingRuntimeFormulaFunctionError(err) {
		return pack, err
	}
	bundles, hasBundles, loadErr := formulaFunctionBundlesForValidation.load(ctx, db)
	if loadErr != nil {
		return nil, fmt.Errorf("active runtime formula function bundles: %w", loadErr)
	}
	if !hasBundles {
		return nil, err
	}
	return sourceBackedPackFromInstalledRowWithFormulaBundlesAndComponentKindRegistrationSet(ctx, db, row, bundles, true, registrations)
}

func isMissingRuntimeFormulaFunctionError(err error) bool {
	return err != nil && strings.Contains(err.Error(), "formula references function(s) not implemented by the runtime evaluator")
}

func formulaFunctionBundlesValidationContextForReconcileDriftedPacks(ctx context.Context, db *sql.DB, driftedPacks map[string]*packs.SourceBackedPack) (*activeFormulaFunctionBundlesValidationContext, error) {
	if len(driftedPacks) == 0 {
		return &activeFormulaFunctionBundlesValidationContext{}, nil
	}
	names := make([]string, 0, len(driftedPacks))
	for name := range driftedPacks {
		names = append(names, name)
	}
	sort.Strings(names)
	installing := make([]packs.Pack, 0, len(names))
	for _, name := range names {
		if pack := driftedPacks[name]; pack != nil {
			installing = append(installing, pack)
		}
	}
	bundles, hasBundles, err := packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidationWithOverrides(ctx, db, installing)
	if err != nil {
		return nil, fmt.Errorf("reconcile source-backed formula function bundles: %w", err)
	}
	return &activeFormulaFunctionBundlesValidationContext{
		loaded:  true,
		bundles: bundles,
		has:     hasBundles,
	}, nil
}

func legacyInstalledAppPacksMissingSourceIdentity(ctx context.Context, db *sql.DB) ([]installedPackRow, error) {
	rows, err := db.QueryContext(ctx, `
		SELECT Id, Name, COALESCE(PackInstallId, ''), Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision
		  FROM _System_InstalledPack
		 WHERE Category = ?
		   AND Status = ?
		   AND COALESCE(IsDeleted, 0) = 0
		   AND (COALESCE(SourceKind, '') = '' OR COALESCE(SourceURI, '') = '')
		 ORDER BY Name`,
		string(packs.AppPack), installedPackStatusActive,
	)
	if err != nil {
		return nil, fmt.Errorf("query legacy installed app packs missing source identity: %w", err)
	}
	defer rows.Close()

	var out []installedPackRow
	for rows.Next() {
		var row installedPackRow
		if err := rows.Scan(
			&row.ID,
			&row.Name,
			&row.PackInstallID,
			&row.Category,
			&row.Status,
			&row.PackVersion,
			&row.ManifestFormatVersion,
			&row.PlatformCompatibility,
			&row.DesiredPayloadHash,
			&row.AppliedPayloadHash,
			&row.PayloadHashAlgorithm,
			&row.SourceKind,
			&row.SourceURI,
			&row.SourceRevision,
		); err != nil {
			return nil, fmt.Errorf("scan legacy installed app pack: %w", err)
		}
		row.Name = strings.TrimSpace(row.Name)
		if row.Name != "" {
			out = append(out, row)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate legacy installed app packs: %w", err)
	}
	return out, nil
}

func legacyInstalledPackAdoptionSource(row installedPackRow) (string, error) {
	name := strings.TrimSpace(row.Name)
	if name == "" {
		return "", fmt.Errorf("adopt legacy installed pack: missing installed pack name")
	}
	sourceKind := strings.TrimSpace(row.SourceKind.String)
	sourceURI := strings.TrimSpace(row.SourceURI.String)
	officialRepoSource := string(packs.PackSourceKindOfficialRepo) + ":" + name

	switch {
	case sourceKind == "" && sourceURI == "":
		return officialRepoSource, nil
	case sourceKind == string(packs.PackSourceKindOfficialRepo) && sourceURI == "":
		return officialRepoSource, nil
	case sourceKind == "" && sourceURI == name:
		return officialRepoSource, nil
	default:
		return "", fmt.Errorf("installed pack %s has incomplete source identity (SourceKind=%q, SourceURI=%q)", name, sourceKind, sourceURI)
	}
}

func sourceBackedPackHasComponentKind(pack *packs.SourceBackedPack, kind packs.ComponentKind) bool {
	for _, component := range pack.Components() {
		if component.Kind() == kind {
			return true
		}
	}
	return false
}

func sourceBackedPackNeedsFlowRuntimeMaintenance(ctx context.Context, db *database.TiDBConnection, pack *packs.SourceBackedPack) (bool, error) {
	if sourceBackedPackHasComponentKind(pack, packs.ComponentKindStandardFlow) {
		return true, nil
	}
	if db == nil {
		return false, fmt.Errorf("runner database connection is unavailable")
	}
	var marker int
	err := db.DB().QueryRowContext(ctx, `
SELECT 1
  FROM FlowDefinition
 WHERE COALESCE(OwnerPack, '') = ?
 LIMIT 1`, strings.TrimSpace(pack.Name())).Scan(&marker)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("inspect flow runtime ownership for pack %s: %w", pack.Name(), err)
	}
	return true, nil
}

func verifyInstalledConnectorPackHandlerConfigs(ctx context.Context, db *sql.DB, pack *packs.SourceBackedPack) error {
	if pack == nil || pack.Category() != packs.ConnectorPack {
		return nil
	}
	return handlerconfig.VerifyHandlerConfigsForPack(ctx, db, pack.Name())
}

func VerifyInstalledPackByName(ctx context.Context, db *sql.DB, name string, deep bool) error {
	return VerifyInstalledPackByNameWithRunnerOptions(ctx, db, name, deep, RunnerOptions{})
}

// VerifyInstalledPackByNameWithRunnerOptions verifies one installed source-backed
// pack with a lazily prepared runner that owns the immutable distribution options.
func VerifyInstalledPackByNameWithRunnerOptions(ctx context.Context, db *sql.DB, name string, deep bool, options RunnerOptions) error {
	return verifyInstalledPackByName(ctx, db, nil, name, deep, options)
}

// VerifyInstalledPackByNameWithRunner verifies one installed source-backed pack
// with the Runner's immutable distribution registrations and formula compiler.
func VerifyInstalledPackByNameWithRunner(ctx context.Context, r *Runner, name string, deep bool) error {
	if r == nil || r.DB == nil {
		return fmt.Errorf("installed pack %s verify: runner database connection is unavailable", name)
	}
	return verifyInstalledPackByName(ctx, r.DB.DB(), r, name, deep, RunnerOptions{ComponentKindRegistrationSet: r.componentKindRegistrationSet()})
}

func verifyInstalledPackByName(ctx context.Context, db *sql.DB, r *Runner, name string, deep bool, options RunnerOptions) error {
	row, ok, err := getInstalledPackRowByName(ctx, db, name)
	if err != nil {
		return err
	}
	if !ok {
		return fmt.Errorf("installed pack %s is missing", name)
	}
	active, err := packs.IsPackActive(ctx, db, name)
	if err != nil {
		return err
	}
	if !active {
		return fmt.Errorf("installed pack %s status is %q, want active", name, row.Status)
	}
	registrations := options.ComponentKindRegistrationSet
	if r != nil {
		registrations = r.componentKindRegistrationSet()
	}
	if err := validatePackManagedApprovalOwnership(ctx, db, row.Name, packs.Category(strings.TrimSpace(row.Category)), registrations); err != nil {
		return fmt.Errorf("installed pack %s managed approval coverage: %w", name, err)
	}
	var pack packs.Pack
	if r == nil {
		resolverOptions := installedPackResolverOptionsWithSchema(ctx, db)
		resolverOptions.ComponentKindRegistrationSet = &registrations
		pack, err = packs.ResolveInstalledPack(
			packs.InstalledPackResolverInput{
				Name: row.Name, Category: packs.Category(strings.TrimSpace(row.Category)), AppliedPayloadHash: row.AppliedPayloadHash.String,
				ManifestFormatVersion: row.ManifestFormatVersion.String, SourceKind: row.SourceKind.String,
				SourceURI: row.SourceURI.String, SourceRevision: row.SourceRevision.String,
			}, resolverOptions,
		)
		if err != nil && installedPackHasSourceIdentity(row) && isMissingRuntimeFormulaFunctionError(err) {
			formulaFunctionBundles, hasFormulaFunctionBundles, loadErr := packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidation(ctx, db)
			if loadErr != nil {
				return fmt.Errorf("installed pack %s runtime formula function bundles: %w", name, loadErr)
			}
			if hasFormulaFunctionBundles {
				resolverOptions.RuntimeFormulaFunctionBundlesForValidation = formulaFunctionBundles
				resolverOptions.HasRuntimeFormulaFunctionBundlesForValidation = true
				pack, err = packs.ResolveInstalledPack(
					packs.InstalledPackResolverInput{
						Name: row.Name, Category: packs.Category(strings.TrimSpace(row.Category)), AppliedPayloadHash: row.AppliedPayloadHash.String,
						ManifestFormatVersion: row.ManifestFormatVersion.String, SourceKind: row.SourceKind.String,
						SourceURI: row.SourceURI.String, SourceRevision: row.SourceRevision.String,
					}, resolverOptions,
				)
			}
		}
	} else {
		pack, err = sourceBackedPackFromInstalledRowWithComponentKindRegistrationSet(ctx, db, row, registrations)
		if err != nil && isMissingRuntimeFormulaFunctionError(err) {
			formulaFunctionBundles, hasBundles, loadErr := packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidation(ctx, db)
			if loadErr != nil {
				return fmt.Errorf("installed pack %s runtime formula function bundles: %w", name, loadErr)
			}
			if hasBundles {
				pack, err = sourceBackedPackFromInstalledRowWithFormulaBundlesAndComponentKindRegistrationSet(ctx, db, row, formulaFunctionBundles, true, registrations)
			}
		}
	}
	if err != nil {
		return fmt.Errorf("installed pack %s resolution: %w", name, err)
	}
	if err := verifyInstalledPackHashState(row, pack); err != nil {
		return err
	}
	if err := verifyInstalledPackHostRequirementsState(row, pack); err != nil {
		return err
	}
	sourceBackedPack, ok := pack.(*packs.SourceBackedPack)
	if !ok {
		return fmt.Errorf("installed pack %s resolution: expected source-backed pack, got %T", name, pack)
	}
	if err := validateLivePackDependencies(ctx, db, sourceBackedPack); err != nil {
		return fmt.Errorf("installed pack %s dependencies: %w", pack.Name(), err)
	}
	if err := assertConnectorCapabilitiesHaveHandlersWithInstalledCompat(sourceBackedPack, true); err != nil {
		return fmt.Errorf("installed pack %s capability handler guard: %w", pack.Name(), err)
	}
	if err := verifyInstalledConnectorPackHandlerConfigs(ctx, db, sourceBackedPack); err != nil {
		return fmt.Errorf("installed pack %s live connector inbound handlers: %w", pack.Name(), err)
	}
	if err := pack.Verify(ctx, packs.VerifyDeps{DB: db, DeepChecks: deep}); err != nil {
		return fmt.Errorf("installed pack %s verify: %w", pack.Name(), err)
	}
	compileProof := formulaViewCompileProofForDBWithRunnerOptions(db, options)
	if r != nil {
		compileProof = formulaViewCompileProofForRunner(r)
	}
	if err := verifySourceBackedDerivedArtifactsWithCompileProof(ctx, db, sourceBackedPack, compileProof); err != nil {
		return fmt.Errorf("installed pack %s derived artifacts: %w", pack.Name(), err)
	}
	return nil
}

func validateSourceDiagnostics(rawSource string) error {
	return validateSourceDiagnosticsForOperation("install-pack", rawSource)
}

func validateSourceDiagnosticsForOperation(operation string, rawSource string) error {
	return validateSourceDiagnosticsForOperationWithOptions(operation, rawSource, packs.DefaultPackSourceOptions())
}

func validateSourceDiagnosticsWithOptions(rawSource string, opts packs.PackSourceOptions) error {
	return validateSourceDiagnosticsForOperationWithOptions("install-pack", rawSource, opts)
}

func validateSourceDiagnosticsForOperationWithOptions(operation string, rawSource string, opts packs.PackSourceOptions) error {
	return installableValidationReportError(operation, packs.ValidateInstallablePackSourceWithOptions(rawSource, opts))
}

func validateSourceDiagnosticsForOperationWithInstallContext(ctx context.Context, db *sql.DB, operation string, rawSource string) error {
	return validateSourceDiagnosticsForOperationWithInstallContextAndOptions(ctx, db, operation, rawSource, packs.DefaultPackSourceOptions())
}

func validateSourceDiagnosticsForOperationWithInstallContextAndOptions(ctx context.Context, db *sql.DB, operation string, rawSource string, opts packs.PackSourceOptions) error {
	return validateSourceDiagnosticsForOperationWithInstallContextAndOptionsAndComponentKindRegistrationSet(
		ctx, db, operation, rawSource, opts, packs.ComponentKindRegistrationSet{},
	)
}

func validateSourceDiagnosticsForOperationWithInstallContextAndOptionsAndComponentKindRegistrationSet(ctx context.Context, db *sql.DB, operation string, rawSource string, opts packs.PackSourceOptions, registrations packs.ComponentKindRegistrationSet) error {
	report, _, err := validateInstallablePackSourceForInstallWithPreflightOverridesAndOptionsAndComponentKindRegistrationSet(
		ctx, db, rawSource, nil, opts, registrations,
	)
	if err != nil {
		return err
	}
	return installableValidationReportError(operation, report)
}

func validateReapplySourceDiagnosticsForOperationWithInstallContextAndOptions(ctx context.Context, r *Runner, operation string, rawSource string, opts packs.PackSourceOptions, options InstallPackOptions) error {
	registrations := packs.ComponentKindRegistrationSet{}
	if r != nil {
		registrations = r.componentKindRegistrationSet()
	}
	return validateReapplySourceDiagnosticsForOperationWithInstallContextAndOptionsAndComponentKindRegistrationSet(ctx, r, operation, rawSource, opts, options, registrations)
}

func validateReapplySourceDiagnosticsForOperationWithInstallContextAndOptionsAndComponentKindRegistrationSet(ctx context.Context, r *Runner, operation string, rawSource string, opts packs.PackSourceOptions, options InstallPackOptions, registrations packs.ComponentKindRegistrationSet) error {
	if r == nil || r.DB == nil {
		return fmt.Errorf("%s: database connection is unavailable", operation)
	}
	sourceReport, err := reapplySourceReportWithComponentKindRegistrationSet(rawSource, registrations)
	if err != nil {
		return err
	}
	prepared, revisionAdvance, err := prepareReapplySourceRevisionAdvance(rawSource, sourceReport.Source, options)
	if err != nil {
		return err
	}
	if revisionAdvance {
		_, err := PendingApplyAllSourceRevisionAdvances(ctx, r.DB.DB(), prepared)
		return err
	}
	row, err := activeInstalledPackRowForReapplySource(ctx, r.DB.DB(), sourceReport)
	if err != nil {
		return err
	}
	preflightOverrides, err := installPackPreflightOverridesForReapplySourceWithComponentKindRegistrationSet(
		ctx, r, row, sourceReport.Source.Manifest.Dependencies, registrations,
	)
	if err != nil {
		return err
	}
	report, _, err := validateInstallablePackSourceForInstallWithPreflightOverridesAndOptionsAndComponentKindRegistrationSet(
		ctx, r.DB.DB(), rawSource, preflightOverrides, opts, registrations,
	)
	if err != nil {
		return err
	}
	return installableValidationReportError(operation, report)
}

func validateInstallablePackSourceForInstall(ctx context.Context, db *sql.DB, rawSource string) (packs.InstallablePackSourceReport, []formula.FunctionBundle, error) {
	return validateInstallablePackSourceForInstallWithOptions(ctx, db, rawSource, packs.DefaultPackSourceOptions())
}

func validateInstallablePackSourceForInstallWithOptions(ctx context.Context, db *sql.DB, rawSource string, opts packs.PackSourceOptions) (packs.InstallablePackSourceReport, []formula.FunctionBundle, error) {
	return validateInstallablePackSourceForInstallWithPreflightOverridesAndOptions(ctx, db, rawSource, nil, opts)
}

func validateInstallablePackSourceForInstallWithPreflightOverridesAndOptions(ctx context.Context, db *sql.DB, rawSource string, preflightOverrides *installPackPreflightOverrides, opts packs.PackSourceOptions) (packs.InstallablePackSourceReport, []formula.FunctionBundle, error) {
	return validateInstallablePackSourceForInstallWithPreflightOverridesAndOptionsAndComponentKindRegistrationSet(
		ctx, db, rawSource, preflightOverrides, opts, packs.ComponentKindRegistrationSet{},
	)
}

func validateInstallablePackSourceForInstallWithPreflightOverridesAndOptionsAndComponentKindRegistrationSet(ctx context.Context, db *sql.DB, rawSource string, preflightOverrides *installPackPreflightOverrides, opts packs.PackSourceOptions, registrations packs.ComponentKindRegistrationSet) (packs.InstallablePackSourceReport, []formula.FunctionBundle, error) {
	sourceOpts := packSourceOptionsForRequestedSourceValidation(opts)
	sourceReport := packs.ValidatePackSourceWithComponentKindRegistrationSet(rawSource, sourceOpts, registrations)
	if sourceReport.HasErrors() || sourceReport.Source == nil || sourceReport.Source.Manifest == nil {
		return packs.InstallablePackSourceReport{
			Source:      sourceReport.Source,
			Diagnostics: sourceReport.Diagnostics,
		}, nil, nil
	}
	sourceFormulaFunctionBundles, err := packs.RuntimeFormulaFunctionBundlesFromResolvedSource(sourceReport.Source)
	if err != nil {
		return packs.InstallablePackSourceReport{}, nil, fmt.Errorf("install-pack source formula function bundles: %w", err)
	}
	report := packs.ValidateInstallablePackSourceWithRuntimeFormulaFunctionBundlesAndComponentKindRegistrationSet(
		rawSource,
		sourceOpts,
		sourceFormulaFunctionBundles,
		registrations,
	)
	if !report.HasErrors() || !strings.Contains(report.Error(), "formula references function(s) not implemented by the runtime evaluator") {
		return report, formula.CloneFunctionBundles(sourceFormulaFunctionBundles), nil
	}
	formulaFunctionBundles, err := packs.RuntimeFormulaFunctionBundlesForInstallSourceWithOverridesAndComponentKindRegistrationSet(
		ctx,
		db,
		sourceReport.Source,
		runtimeDeclarationOverridesFromPreflightExcludingPack(preflightOverrides, sourceReport.Source.Manifest.Name),
		registrations,
	)
	if err != nil {
		return packs.InstallablePackSourceReport{}, nil, fmt.Errorf("install-pack runtime formula function bundles: %w", err)
	}
	report = packs.ValidateInstallablePackSourceWithRuntimeFormulaFunctionBundlesAndComponentKindRegistrationSet(
		rawSource,
		sourceOpts,
		formulaFunctionBundles,
		registrations,
	)
	return report, formula.CloneFunctionBundles(formulaFunctionBundles), nil
}

func applyResolvedSourceBackedPackUnlockedWithPreflightOverrides(ctx context.Context, r *Runner, pack *packs.SourceBackedPack, preflightOverrides *installPackPreflightOverrides, options InstallPackOptions) (err error) {
	if pack == nil {
		return fmt.Errorf("install-pack: source did not produce an installable pack")
	}
	if options.stoppedSourceSetApplyOnly && !options.stoppedSourceSetAuthority.Valid() {
		return fmt.Errorf("stopped source-set apply authority is required for pack %s", pack.Name())
	}
	if options.stoppedSourceSetApplyOnly && strings.TrimSpace(options.stoppedSourceSetPreviousPackInstallID) == "" {
		return fmt.Errorf("stopped source-set previous installation identity is required for pack %s", pack.Name())
	}
	if options.stoppedSourceSetApplyOnly && strings.TrimSpace(options.stoppedSourceSetCandidatePackInstallID) == "" {
		return fmt.Errorf("stopped source-set candidate installation identity is required for pack %s", pack.Name())
	}
	var payloadHash string
	candidateMutationStarted := false
	defer func() {
		if err == nil || !options.stoppedSourceSetApplyOnly {
			return
		}
		if markErr := markStoppedSourceSetPackGenerationFailed(
			context.WithoutCancel(ctx),
			r.DB.DB(),
			pack.Name(),
			options.stoppedSourceSetCandidatePackInstallID,
			candidateMutationStarted,
			err,
		); markErr != nil {
			err = errors.Join(err, markErr)
		}
	}()
	markInstallFailed := func(applyErr error) error {
		if options.stoppedSourceSetApplyOnly {
			return nil
		}
		return markInstalledPackFailed(ctx, r.DB.DB(), pack, payloadHash, applyErr)
	}
	if pack.Category() != packs.AppPack && pack.Category() != packs.ConnectorPack {
		return fmt.Errorf("install-pack supports categories %q and %q, got %q", packs.AppPack, packs.ConnectorPack, pack.Category())
	}
	packPhasePrefix := fmt.Sprintf("SourcePack/%s", pack.Name())
	if !options.stoppedSourceSetApplyOnly {
		if err := runLoggedControlPlanePhase(packPhasePrefix, "schema-readiness", func() error {
			return ensureInstallPackSchemaReadiness(ctx, r)
		}); err != nil {
			return err
		}
	}
	if err := runLoggedControlPlanePhase(packPhasePrefix, "validate-live-pack-dependencies", func() error {
		return validateLivePackDependenciesWithPreflightOverrides(ctx, r.DB.DB(), pack, preflightOverrides)
	}); err != nil {
		return err
	}
	payloadHash, err = packs.PackPayloadHash(pack)
	if err != nil {
		return fmt.Errorf("pack %s payload hash: %w", pack.Name(), err)
	}
	applied := appliedAppPack{pack: pack, payloadHash: payloadHash}
	var repairPackInstallID string
	publishGeneration, err := packPublishesImmutableGeneration(pack)
	if err != nil {
		return err
	}
	if options.stoppedSourceSetApplyOnly {
		if !publishGeneration {
			return fmt.Errorf("stopped source-set pack %s must use a revision-addressed source", pack.Name())
		}
	} else if publishGeneration {
		adoptionPackInstallID := strings.TrimSpace(options.legacySourceIdentityAdoptionPackInstallID)
		if releaseAdoptionPackInstallID := strings.TrimSpace(options.releaseSourceIdentityAdoptionPackInstallID); releaseAdoptionPackInstallID != "" {
			if adoptionPackInstallID != "" && adoptionPackInstallID != releaseAdoptionPackInstallID {
				return fmt.Errorf("%w: pack %q has conflicting source adoption identities", ErrPackGenerationConflict, pack.Name())
			}
			adoptionPackInstallID = releaseAdoptionPackInstallID
		}
		if adoptionPackInstallID != "" {
			repairPackInstallID, err = sourceBackedPackInstallIDForExactSourceAuthorityAdoption(
				ctx,
				r.DB.DB(),
				applied,
				adoptionPackInstallID,
			)
		} else {
			repairPackInstallID, err = activatedSourceBackedPackRepairInstallID(
				ctx,
				r.DB.DB(),
				applied,
			)
		}
		if err != nil {
			return err
		}
	} else if err := verifyMutableSourceHasNoPublishedGenerationAuthority(
		ctx,
		r.DB.DB(),
		pack.Name(),
	); err != nil {
		return fmt.Errorf(
			"pack %s mutable-source authority preflight: %w",
			pack.Name(),
			err,
		)
	}
	if !options.stoppedSourceSetApplyOnly {
		if resumePackInstallID := strings.TrimSpace(options.resumeSourceBackedPackInstallID); resumePackInstallID != "" {
			if repairPackInstallID != "" && repairPackInstallID != resumePackInstallID {
				return fmt.Errorf(
					"%w: pack %q retry incarnation %q differs from immutable generation %q",
					ErrPackGenerationConflict,
					pack.Name(),
					resumePackInstallID,
					repairPackInstallID,
				)
			}
			repairPackInstallID = resumePackInstallID
		}
	}
	if err := runLoggedControlPlanePhase(packPhasePrefix, "preflight-validation", func() error {
		validate := func(preflightCtx context.Context) error {
			return validateSourceBackedPackInstallPreflights(
				preflightCtx, r.PackDeps(), r.componentKindRegistrationSet(), pack, preflightOverrides,
			)
		}
		if options.stoppedSourceSetApplyOnly {
			return offlinecandidateauthority.RunWithStoppedSourceSetApplyScope(
				ctx, pack.Name(), options.stoppedSourceSetCandidatePackInstallID, validate,
			)
		}
		return validate(ctx)
	}); err != nil {
		return err
	}
	if err := runLoggedControlPlanePhase(packPhasePrefix, "register-runtime-metadata-for-install", func() error {
		if options.stoppedSourceSetApplyOnly {
			return packs.RegisterPersistedRuntimeMetadataForInstallingPacksWithOverrides(
				ctx,
				options.stoppedSourceSetAuthority,
				r.DB.DB(),
				[]packs.Pack{pack},
				runtimeDeclarationOverridesFromPreflight(preflightOverrides),
			)
		}
		if err := packs.RegisterRuntimeMetadataForInstallWithOverridesAndComponentKindRegistrationSet(ctx, r.DB.DB(), pack, runtimeDeclarationOverridesFromPreflight(preflightOverrides), r.componentKindRegistrationSet()); err != nil {
			return fmt.Errorf("register runtime metadata for installing pack %s: %w", pack.Name(), err)
		}
		return nil
	}); err != nil {
		return err
	}
	applyOptions := packs.ApplyOptions{
		SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
		RepairLegacyUniqueDuplicates:         options.RepairLegacyUniqueDuplicates,
		DeferFormulaViewRebuild:              options.DeferFormulaViewRebuild,
	}
	restoreSourceBackedFailureSnapshots := func(applyErr error) error {
		return applyErr
	}
	if !options.stoppedSourceSetApplyOnly {
		defer func() {
			if err == nil {
				return
			}
			if restoreErr := restoreInstalledRuntimeMetadataAfterInstallFailure(ctx, r, r.DB.DB()); restoreErr != nil {
				err = errors.Join(err, fmt.Errorf("restore installed runtime metadata after failed install: %w", restoreErr))
			}
		}()
		childCloneRulesOwnershipSnapshot, snapshotErr := packs.SnapshotObjectSpecChildCloneRulePackOwnershipForPack(ctx, r.DB.DB(), pack)
		if snapshotErr != nil {
			return fmt.Errorf("snapshot %s ChildCloneRules ownership sidecars before install: %w", pack.Name(), snapshotErr)
		}
		exclusiveSiblingFieldsOwnershipSnapshot, snapshotErr := packs.SnapshotObjectSpecExclusiveSiblingFieldPackOwnershipForPack(ctx, r.DB.DB(), pack)
		if snapshotErr != nil {
			return fmt.Errorf("snapshot %s ExclusiveSiblingFields ownership sidecars before install: %w", pack.Name(), snapshotErr)
		}
		connectedAppSnapshot, snapshotErr := packs.SnapshotConnectedAppPackState(ctx, r.DB.DB(), pack.Name())
		if snapshotErr != nil {
			return fmt.Errorf("snapshot %s ConnectedApp source-backed state before install: %w", pack.Name(), snapshotErr)
		}
		restoreSourceBackedFailureSnapshots = func(applyErr error) error {
			var restoreErrs []error
			if restoreErr := connectedAppSnapshot.Restore(ctx, r.PackDepsWithOptions(applyOptions)); restoreErr != nil {
				restoreErrs = append(restoreErrs, fmt.Errorf("restore ConnectedApp source-backed state: %w", restoreErr))
			}
			if restoreErr := childCloneRulesOwnershipSnapshot.Restore(ctx, r.DB.DB()); restoreErr != nil {
				restoreErrs = append(restoreErrs, fmt.Errorf("restore ChildCloneRules pack ownership sidecars: %w", restoreErr))
			}
			if restoreErr := exclusiveSiblingFieldsOwnershipSnapshot.Restore(ctx, r.DB.DB()); restoreErr != nil {
				restoreErrs = append(restoreErrs, fmt.Errorf("restore ExclusiveSiblingFields pack ownership sidecars: %w", restoreErr))
			}
			if len(restoreErrs) > 0 {
				return fmt.Errorf("%w; additionally failed to restore source-backed failure snapshots: %v", applyErr, errors.Join(restoreErrs...))
			}
			return applyErr
		}
	}
	if !options.stoppedSourceSetApplyOnly {
		if err := runLoggedControlPlanePhase(packPhasePrefix, "mark-installed-pack-installing", func() error {
			return markInstalledPackInstallingWithPackInstallID(
				ctx,
				r.DB.DB(),
				pack,
				payloadHash,
				repairPackInstallID,
			)
		}); err != nil {
			return err
		}
	}
	if err := runLoggedControlPlanePhase(packPhasePrefix, "apply-source-backed-pack-install-plan", func() error {
		candidateMutationStarted = options.stoppedSourceSetApplyOnly
		applyPlan := func(applyCtx context.Context) error {
			return applySourceBackedPackInstallPlan(applyCtx, r, pack, options)
		}
		var applyErr error
		if options.stoppedSourceSetApplyOnly {
			applyErr = offlinecandidateauthority.RunWithStoppedSourceSetApplyScope(
				ctx, pack.Name(), options.stoppedSourceSetCandidatePackInstallID, applyPlan,
			)
		} else {
			applyErr = applyPlan(ctx)
		}
		if applyErr != nil {
			restoredErr := restoreSourceBackedFailureSnapshots(applyErr)
			if markErr := markInstallFailed(restoredErr); markErr != nil {
				return fmt.Errorf("pack %s: %w", pack.Name(), errors.Join(restoredErr, fmt.Errorf("mark installed pack failed: %w", markErr)))
			}
			return fmt.Errorf("pack %s: %w", pack.Name(), restoredErr)
		}
		return nil
	}); err != nil {
		return err
	}
	if err := runLoggedControlPlanePhase(packPhasePrefix, "refresh-cache-after-apply", func() error {
		if err := r.RefreshCache(ctx); err != nil {
			applyErr := restoreSourceBackedFailureSnapshots(err)
			if markErr := markInstallFailed(applyErr); markErr != nil {
				return fmt.Errorf("cache refresh after %s failed: %w", pack.Name(), errors.Join(applyErr, fmt.Errorf("mark installed pack failed: %w", markErr)))
			}
			return fmt.Errorf("cache refresh after %s failed: %w", pack.Name(), applyErr)
		}
		return nil
	}); err != nil {
		return err
	}
	if !options.stoppedSourceSetApplyOnly {
		if err := runLoggedControlPlanePhase(packPhasePrefix, "validate-insert-rules-after-apply", func() error {
			if err := runInsertRulesValidators(ctx, r); err != nil {
				applyErr := restoreSourceBackedFailureSnapshots(err)
				if markErr := markInstallFailed(applyErr); markErr != nil {
					return fmt.Errorf("insert-rules validation after %s: %w", pack.Name(), errors.Join(applyErr, fmt.Errorf("mark installed pack failed: %w", markErr)))
				}
				return fmt.Errorf("insert-rules validation after %s: %w", pack.Name(), applyErr)
			}
			return nil
		}); err != nil {
			return err
		}
	}
	verifyCtx := withRuntimeEffectivePeerDeclarationOverrides(ctx, preflightOverrides)
	if options.offlineCandidatePreparation != nil {
		if err := stageOfflineCandidatePreparedMaterialization(
			options.offlineCandidatePreparation,
			applied,
			restoreSourceBackedFailureSnapshots,
		); err != nil {
			applyErr := restoreSourceBackedFailureSnapshots(err)
			if markErr := markInstallFailed(applyErr); markErr != nil {
				return errors.Join(applyErr, fmt.Errorf("mark installed pack failed: %w", markErr))
			}
			return applyErr
		}
		return nil
	}
	var connectedAppChangedIDs []string
	if err := runLoggedControlPlanePhase(packPhasePrefix, "finalize-applied-pack", func() error {
		registerRuntime := func(ctx context.Context, db *sql.DB, installing []packs.Pack) error {
			if options.stoppedSourceSetApplyOnly {
				return packs.RegisterPersistedRuntimeMetadataForInstallingPacksWithOverrides(
					ctx,
					options.stoppedSourceSetAuthority,
					db,
					installing,
					runtimeDeclarationOverridesFromPreflight(preflightOverrides),
				)
			}
			return packs.RegisterRuntimeMetadataForInstallingPacksWithOverridesAndComponentKindRegistrationSet(
				ctx,
				db,
				installing,
				runtimeDeclarationOverridesFromPreflight(preflightOverrides),
				r.componentKindRegistrationSet(),
			)
		}
		publishConnectedApps := func(ctx context.Context, tx *sql.Tx) error {
			if options.stoppedSourceSetApplyOnly {
				if err := packs.FinalizeStoppedSourceSetConnectorIncarnationTransition(
					ctx,
					tx,
					pack.Name(),
					options.stoppedSourceSetPreviousPackInstallID,
					options.stoppedSourceSetCandidatePackInstallID,
				); err != nil {
					return err
				}
			}
			changed, err := packs.PublishSourceBackedConnectedAppsForPackInTx(ctx, tx, pack.Name())
			if err != nil {
				return err
			}
			connectedAppChangedIDs = append(connectedAppChangedIDs, changed...)
			return nil
		}
		var finalizeErr error
		if options.stoppedSourceSetApplyOnly {
			finalizeErr = offlinecandidateauthority.RunWithStoppedSourceSetApplyScope(
				verifyCtx,
				pack.Name(),
				options.stoppedSourceSetCandidatePackInstallID,
				func(candidateCtx context.Context) error {
					return prepareAppliedAppPacksForActivationWithVerifier(
						candidateCtx,
						r.DB.DB(),
						[]appliedAppPack{applied},
						registerRuntime,
						func(*sql.DB) error { return nil },
						func(ctx context.Context) error {
							return materializeSourceBackedPackDerivedApplyOnly(
								ctx, r, pack, options.DeferFormulaViewRebuild,
							)
						},
						nil,
						func(context.Context, appliedAppPack) error { return nil },
					)
				},
			)
			if finalizeErr == nil {
				finalizeErr = markInstalledPackActiveForStoppedSourceSet(
					ctx,
					r.DB.DB(),
					pack,
					payloadHash,
					options.stoppedSourceSetPreviousPackInstallID,
					options.stoppedSourceSetCandidatePackInstallID,
					publishConnectedApps,
				)
			}
		} else {
			finalizeErr = finalizeAppliedAppPacksWithActiveHook(
				verifyCtx,
				r.DB.DB(),
				[]appliedAppPack{applied},
				registerRuntime,
				runLayoutMaintenance,
				func(ctx context.Context) error {
					if err := materializeSourceBackedPackDerived(ctx, r, pack, options.DeferFormulaViewRebuild); err != nil {
						return err
					}
					return handlerconfig.RegisterHandlerConfigsForPack(ctx, r.DB.DB(), connectorcapabilities.Default(), pack.Name())
				},
				candidateFinalVerifyWithRunner(r, options.CandidateFinalVerify),
				func(ctx context.Context, tx *sql.Tx, _ appliedAppPack) error {
					return publishConnectedApps(ctx, tx)
				},
			)
		}
		if finalizeErr != nil {
			err := finalizeErr
			if errors.Is(err, ErrInstalledPackActivationIndeterminate) {
				return fmt.Errorf(
					"finalize applied pack %s without compensation: %w",
					pack.Name(),
					err,
				)
			}
			applyErr := restoreSourceBackedFailureSnapshots(err)
			if markErr := markInstallFailed(applyErr); markErr != nil {
				return fmt.Errorf("finalize applied pack %s: %w", pack.Name(), errors.Join(applyErr, fmt.Errorf("mark installed pack failed: %w", markErr)))
			}
			return fmt.Errorf("finalize applied pack %s: %w", pack.Name(), applyErr)
		}
		return nil
	}); err != nil {
		return err
	}
	packs.NotifySourceBackedConnectedAppChanges(r.PackDepsWithOptions(applyOptions), connectedAppChangedIDs)
	if options.stoppedSourceSetApplyOnly {
		return nil
	}
	if err := runLoggedControlPlanePhase(packPhasePrefix, "reconcile-connector-inbound-workers", func() error {
		return reconcileConnectorInboundWorkersAfterPackLifecycle(ctx, r)
	}); err != nil {
		return err
	}
	return nil
}

func restoreInstalledRuntimeMetadataAfterInstallFailure(ctx context.Context, r *Runner, db *sql.DB) error {
	if err := packs.RegisterInstalledRuntimeMetadataWithComponentKindRegistrationSet(ctx, db, r.componentKindRegistrationSet()); err != nil {
		return fmt.Errorf("register installed runtime metadata: %w", err)
	}
	if r != nil && r.SvcMgr != nil && r.SvcMgr.FormulaView() != nil {
		if err := r.SvcMgr.FormulaView().RefreshRuntimeFunctionRegistry(); err != nil {
			return fmt.Errorf("refresh formula view function registry: %w", err)
		}
	}
	return nil
}

func reconcileConnectorInboundWorkersAfterPackLifecycle(ctx context.Context, r *Runner) error {
	if r == nil || r.SvcMgr == nil {
		if r == nil || r.DB == nil {
			return nil
		}
		return reconcileRegisteredConnectorInboundWorkers(ctx, r.DB.DB())
	}
	if r.SvcMgr.ConnectorInbound() != nil {
		return r.SvcMgr.ReconcileChangedConnectorInboundWorkers(ctx)
	}
	if r.DB == nil {
		return nil
	}
	return reconcileRegisteredConnectorInboundWorkers(ctx, r.DB.DB())
}

func ensureInstallPackSchemaReadiness(_ context.Context, r *Runner) error {
	if r == nil {
		return fmt.Errorf("schema readiness: runner is unavailable")
	}
	return ensureInstallPackSchemaReadinessForConnection(r.DB)
}

func ensureInstallPackSchemaReadinessForConnection(db *database.TiDBConnection) error {
	if db == nil {
		return fmt.Errorf("schema readiness: database connection is unavailable")
	}
	if err := schemaProvisioner.VerifySchemaReadiness(db); err != nil {
		return fmt.Errorf("schema readiness: %w", err)
	}
	return nil
}

func validateDefaultChildRowsInstallPreflight(ctx context.Context, r *Runner, pack *packs.SourceBackedPack) error {
	var fallback records.DefaultChildRowsSchemaSnapshot
	var fallbackEngineOwned func(string) map[string]bool
	if r != nil && r.SvcMgr != nil {
		fallback = r.SvcMgr.Metadata()
		fallbackEngineOwned = engineOwnedByObjectFromAutoNumber(r.SvcMgr.AutoNumberMetadata())
	}
	return validateDefaultChildRowsInstallPreflightWithSnapshots(
		ctx, fallback, fallbackEngineOwned, pack,
	)
}

func validateDefaultChildRowsInstallPreflightWithSnapshots(
	ctx context.Context,
	fallback records.DefaultChildRowsSchemaSnapshot,
	fallbackEngineOwned func(string) map[string]bool,
	pack *packs.SourceBackedPack,
) error {
	if pack == nil {
		return nil
	}
	specs, err := packs.ParseObjectSpecsPayloads(pack.RuntimeObjectSpecPayloads(), "runtime-object-spec")
	if err != nil {
		return fmt.Errorf("parse runtime object specs for defaultChildRows preflight: %w", err)
	}
	if len(specs) == 0 {
		return nil
	}
	sourceEngineOwned, err := packs.DefaultChildRowsEngineOwnedBySourcePack(pack)
	if err != nil {
		return fmt.Errorf("collect source-backed autonumber ownership for defaultChildRows preflight: %w", err)
	}
	fallbackEngineOwned = packs.MergeDefaultChildRowsEngineOwnedLookups(fallbackEngineOwned, sourceEngineOwned)
	if err := packs.ValidateDefaultChildRowsForSpecs(ctx, specs, fallback, fallbackEngineOwned, false); err != nil {
		return fmt.Errorf("install-pack defaultChildRows preflight: %w", err)
	}
	return nil
}

func validateChildCloneRulesInstallPreflight(ctx context.Context, r *Runner, pack *packs.SourceBackedPack) error {
	var fallback records.DefaultChildRowsSchemaSnapshot
	if r != nil && r.SvcMgr != nil {
		fallback = r.SvcMgr.Metadata()
	}
	return validateChildCloneRulesInstallPreflightWithSnapshot(ctx, fallback, pack)
}

func validateChildCloneRulesInstallPreflightWithSnapshot(
	ctx context.Context,
	fallback records.DefaultChildRowsSchemaSnapshot,
	pack *packs.SourceBackedPack,
) error {
	if pack == nil {
		return nil
	}
	specs, err := packs.ParseObjectSpecsPayloads(pack.RuntimeObjectSpecPayloads(), "runtime-object-spec")
	if err != nil {
		return fmt.Errorf("parse runtime object specs for childCloneRules preflight: %w", err)
	}
	if len(specs) == 0 {
		return nil
	}
	if err := packs.ValidateChildCloneRulesForSpecs(ctx, specs, fallback, fallback == nil); err != nil {
		return fmt.Errorf("install-pack childCloneRules preflight: %w", err)
	}
	return nil
}

func rebuildRepairPackWithAllowedOwnerlessObjectsForReconcileWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, row installedPackRow, pack *packs.SourceBackedPack, options InstallPackOptions, preflightOverrides *installPackPreflightOverrides, registrations packs.ComponentKindRegistrationSet) (*packs.SourceBackedPack, error) {
	if len(registrations.Registrations()) == 0 {
		return rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile(ctx, db, row, pack, options, preflightOverrides)
	}
	return rebuildRepairPackWithAllowedOwnerlessObjectsAndComponentKindRegistrationSet(ctx, db, row, pack, options, preflightOverrides, registrations)
}

func rebuildRepairPackWithAllowedOwnerlessObjects(ctx context.Context, db *sql.DB, row installedPackRow, pack *packs.SourceBackedPack, options InstallPackOptions, preflightOverrides *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
	return rebuildRepairPackWithAllowedOwnerlessObjectsAndComponentKindRegistrationSet(
		ctx, db, row, pack, options, preflightOverrides, packs.ComponentKindRegistrationSet{},
	)
}

func rebuildRepairPackWithAllowedOwnerlessObjectsAndComponentKindRegistrationSet(ctx context.Context, db *sql.DB, row installedPackRow, pack *packs.SourceBackedPack, options InstallPackOptions, preflightOverrides *installPackPreflightOverrides, registrations packs.ComponentKindRegistrationSet) (*packs.SourceBackedPack, error) {
	if pack == nil {
		return nil, fmt.Errorf("source did not produce an installable pack")
	}
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		return nil, fmt.Errorf("pack %s payload hash: %w", pack.Name(), err)
	}
	allowedObjects, err := samePackAllowedOwnerlessObjectsWithOptionsAndComponentKindRegistrationSet(ctx, db, pack, payloadHash, options, registrations)
	if err != nil {
		return nil, err
	}
	if len(allowedObjects) == 0 {
		return pack, nil
	}
	var rebuilt *packs.SourceBackedPack
	if len(registrations.Registrations()) == 0 {
		rebuilt, err = sourceBackedPackFromInstalledRow(ctx, db, row, packs.WithAllowedOwnerlessObjectAPIs(allowedObjects))
	} else {
		rebuilt, err = sourceBackedPackFromInstalledRowWithComponentKindRegistrationSet(ctx, db, row, registrations, packs.WithAllowedOwnerlessObjectAPIs(allowedObjects))
	}
	if err == nil || !isMissingRuntimeFormulaFunctionError(err) {
		return rebuilt, err
	}
	formulaFunctionBundles, hasFormulaFunctionBundles, loadErr := runtimeFormulaFunctionBundlesForOwnerlessRebuild(ctx, db, preflightOverrides)
	if loadErr != nil {
		return nil, fmt.Errorf("active runtime formula function bundles: %w", loadErr)
	}
	if !hasFormulaFunctionBundles {
		return nil, err
	}
	if len(registrations.Registrations()) == 0 {
		rebuilt, err = sourceBackedPackFromInstalledRowWithFormulaBundles(ctx, db, row, formulaFunctionBundles, true, packs.WithAllowedOwnerlessObjectAPIs(allowedObjects))
	} else {
		rebuilt, err = sourceBackedPackFromInstalledRowWithFormulaBundlesAndComponentKindRegistrationSet(ctx, db, row, formulaFunctionBundles, true, registrations, packs.WithAllowedOwnerlessObjectAPIs(allowedObjects))
	}
	if err != nil {
		return nil, err
	}
	return rebuilt, nil
}

func runtimeFormulaFunctionBundlesForOwnerlessRebuild(ctx context.Context, db *sql.DB, preflightOverrides *installPackPreflightOverrides) ([]formula.FunctionBundle, bool, error) {
	overridePacks := runtimeEffectiveDeclarationPacksFromPreflight(preflightOverrides)
	if len(overridePacks) == 0 {
		return packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidation(ctx, db)
	}
	return packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidationWithOverrides(ctx, db, overridePacks)
}

// isRepairableInstalledPackApplyAllError is the permissive repair classifier
// for ApplyAll's step-6 reconcile path (reconcileActiveInstalledPackSourceState).
// It returns true when every component error is one that pack re-apply can
// automatically heal: source replay of removed fields, key-prefix drift,
// curated page-layout config/routing drift, and missing connector handler-config
// rows when the current source has already declared those handler-config
// components. A missing physical column whose _System_Field row still exists is
// intentionally not repairable here because replaying it would hide structural
// data loss.
//
// This function is intentionally ApplyAll-specific. Background verify and
// pod-readiness paths (Verify, VerifyDeep, VerifyInstalledPackByName) must
// NOT use this predicate — they are fail-closed by design and route errors
// directly to the operator without repair.
func isRepairableInstalledPackApplyAllError(err error) bool {
	if handlerconfig.IsMissingHandlerConfig(err) {
		return true
	}
	componentErrs := packs.ComponentVerifyErrorList(err)
	if len(componentErrs) == 0 {
		return false
	}
	for _, componentErr := range componentErrs {
		if componentErr == nil {
			return false
		}
		switch componentErr.Kind {
		case packs.ComponentKindPageLayout:
			if !packs.IsCuratedPageLayoutConfigDrift(componentErr.Err) &&
				!packs.IsCuratedPageLayoutMissing(componentErr.Err) &&
				!packs.IsCuratedPageLayoutRoutingDrift(componentErr.Err) &&
				!packs.IsSourceBackedPageLayoutRepairableDrift(componentErr.Err) {
				return false
			}
		case packs.ComponentKindObjectSpec:
			// Fail closed when a physical column is missing but its _System_Field
			// row still exists. This indicates structural corruption (schema mutated
			// outside pack management): the metadata still expects the column but the
			// column is gone. Auto-repair would recreate it silently, hiding the
			// corruption. When both the column and the field row are gone together,
			// that is a clean full-field removal and IS repairable by ApplyAll.
			if packs.IsPackObjectColumnMissingAloneWithoutFieldRowMissing(componentErr.Err) {
				return false
			}
			if !packs.IsPackObjectInstalledPackRepairableDrift(componentErr.Err) {
				return false
			}
		default:
			return false
		}
	}
	return true
}

func installedPackSourceLabel(row installedPackRow) string {
	return strings.TrimSpace(row.SourceKind.String) + ":" + strings.TrimSpace(row.SourceURI.String)
}

func installableValidationReportError(operation string, report packs.InstallablePackSourceReport) error {
	if !report.HasErrors() {
		if report.Source == nil || report.Source.Manifest == nil || report.Pack == nil {
			return fmt.Errorf("%s: source did not produce an installable pack", operation)
		}
		return nil
	}
	lines := make([]string, 0, len(report.Diagnostics))
	for _, diagnostic := range report.Diagnostics {
		lines = append(lines, diagnostic.String())
	}
	return fmt.Errorf("%s source diagnostics failed:\n  - %s", operation, strings.Join(lines, "\n  - "))
}

func packValidationReportError(operation string, report packs.PackValidationReport) error {
	if !report.HasErrors() {
		if report.Source == nil || report.Source.Manifest == nil {
			return fmt.Errorf("%s: source did not produce a pack manifest", operation)
		}
		return nil
	}
	lines := make([]string, 0, len(report.Diagnostics))
	for _, diagnostic := range report.Diagnostics {
		lines = append(lines, diagnostic.String())
	}
	return fmt.Errorf("%s source diagnostics failed:\n  - %s", operation, strings.Join(lines, "\n  - "))
}

func samePackAllowedOwnerlessObjects(ctx context.Context, db *sql.DB, pack *packs.SourceBackedPack, payloadHash string) ([]string, error) {
	return samePackAllowedOwnerlessObjectsWithOptions(ctx, db, pack, payloadHash, InstallPackOptions{})
}

func samePackAllowedOwnerlessObjectsWithOptions(ctx context.Context, db *sql.DB, pack *packs.SourceBackedPack, payloadHash string, options InstallPackOptions) ([]string, error) {
	return samePackAllowedOwnerlessObjectsWithOptionsAndComponentKindRegistrationSet(ctx, db, pack, payloadHash, options, packs.ComponentKindRegistrationSet{})
}

func samePackAllowedOwnerlessObjectsWithOptionsAndComponentKindRegistrationSet(ctx context.Context, db *sql.DB, pack *packs.SourceBackedPack, payloadHash string, options InstallPackOptions, registrations packs.ComponentKindRegistrationSet) ([]string, error) {
	row, ok, err := getInstalledPackRowByNameForRetryAllowance(ctx, db, pack.Name())
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, nil
	}
	hasSourceIdentity := installedPackHasSourceIdentity(row)
	meta := packs.PackHashMetadataForPack(pack)
	if options.offlineCandidateLocalSourceTransition != nil {
		if err := validateOfflineCandidateSourceTransition(
			row,
			pack,
			payloadHash,
			options.offlineCandidateLocalSourceTransition,
		); err != nil {
			return nil, fmt.Errorf("installed pack %s Catalog candidate source transition: %w", pack.Name(), err)
		}
	} else if hasSourceIdentity && !sourceIdentityMatches(row, meta) {
		if err := sourceRevisionAdvanceAuthorizationErrorWithComponentKindRegistrationSet(ctx, db, row, pack.Name(), meta, options.SourceRevisionAdvance, registrations); err != nil {
			return nil, fmt.Errorf("installed pack %s source identity differs from requested source: %w", pack.Name(), err)
		}
	}
	switch row.Status {
	case installedPackStatusActive:
		if row.IsDeleted != 0 {
			return nil, nil
		}
		if !hasSourceIdentity {
			return pack.OwnedObjectAPINames(), nil
		}
		return pack.OwnedObjectAPINames(), nil
	case installedPackStatusFailed, installedPackStatusInstalling:
		if !hasSourceIdentity {
			return pack.OwnedObjectAPINames(), nil
		}
		return pack.OwnedObjectAPINames(), nil
	default:
		return nil, fmt.Errorf("installed pack %s status is %q; cannot install source-backed pack", pack.Name(), row.Status)
	}
}

func sourceRevisionAdvanceAuthorizationError(ctx context.Context, db *sql.DB, row installedPackRow, packName string, meta packs.PackHashMetadata, auth *SourceRevisionAdvanceAuthorization) error {
	return sourceRevisionAdvanceAuthorizationErrorWithComponentKindRegistrationSet(ctx, db, row, packName, meta, auth, packs.ComponentKindRegistrationSet{})
}

func sourceRevisionAdvanceAuthorizationErrorWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, row installedPackRow, packName string, meta packs.PackHashMetadata, auth *SourceRevisionAdvanceAuthorization, registrations packs.ComponentKindRegistrationSet) error {
	rowRevision := strings.TrimSpace(row.SourceRevision.String)
	targetRevision := strings.TrimSpace(meta.SourceRevision)
	if auth == nil {
		return fmt.Errorf("source revision advance from %q to %q requires supported operator path: rerun setup reapply-pack with --allow-source-revision-advance --from-revision %s --to-revision %s --verify-pack", rowRevision, targetRevision, rowRevision, targetRevision)
	}
	name := strings.TrimSpace(auth.Name)
	kind, uri, identityErr := canonicalPackSourceIdentity(auth.Kind, auth.URI)
	fromRevision := strings.TrimSpace(auth.FromRevision)
	toRevision := strings.TrimSpace(auth.ToRevision)
	if name == "" || identityErr != nil || fromRevision == "" || toRevision == "" || fromRevision == toRevision {
		return fmt.Errorf("source revision advance authorization is incomplete")
	}
	if strings.TrimSpace(row.Name) != name || strings.TrimSpace(packName) != name {
		return fmt.Errorf("source revision advance authorization pack name mismatch")
	}
	rowKind, rowURI, rowIdentityErr := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	if !row.SourceKind.Valid || !row.SourceURI.Valid || rowIdentityErr != nil || rowKind != kind {
		return fmt.Errorf("source revision advance authorization source kind mismatch")
	}
	if rowURI != uri {
		return fmt.Errorf("source revision advance authorization source URI mismatch")
	}
	resolvedKind, resolvedURI, resolvedIdentityErr := canonicalPackSourceIdentity(meta.SourceKind, meta.SourceURI)
	if resolvedIdentityErr != nil || resolvedKind != kind || resolvedURI != uri {
		return fmt.Errorf("source revision advance authorization resolved source mismatch")
	}
	if !row.SourceRevision.Valid || strings.TrimSpace(row.SourceRevision.String) != fromRevision {
		return fmt.Errorf("source revision advance authorization from-revision mismatch")
	}
	if !meta.SourceRevisionValid || strings.TrimSpace(meta.SourceRevision) != toRevision {
		return fmt.Errorf("source revision advance authorization to-revision mismatch")
	}
	appliedPayloadHash := strings.TrimSpace(row.AppliedPayloadHash.String)
	if !row.AppliedPayloadHash.Valid || appliedPayloadHash == "" {
		return fmt.Errorf("source revision advance precondition failed: installed pack %s AppliedPayloadHash is empty for --from-revision %q", name, fromRevision)
	}
	if err := sourceRevisionAdvanceValidatePayloadHashForInstalledRow(ctx, db, row, fromRevision, appliedPayloadHash, registrations); err != nil {
		return fmt.Errorf("source revision advance precondition failed: installed pack %s AppliedPayloadHash = %q does not match --from-revision %q payload: %w; refusing mutation", name, appliedPayloadHash, fromRevision, err)
	}
	return nil
}

func validatePayloadHashForInstalledPackSourceRevision(ctx context.Context, db *sql.DB, row installedPackRow, sourceRevision string, storedHash string) error {
	return validatePayloadHashForInstalledPackSourceRevisionWithComponentKindRegistrationSet(ctx, db, row, sourceRevision, storedHash, packs.ComponentKindRegistrationSet{})
}

func validatePayloadHashForInstalledPackSourceRevisionWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, row installedPackRow, sourceRevision string, storedHash string, registrations packs.ComponentKindRegistrationSet) error {
	pack, err := sourceBackedPackFromInstalledRowAtSourceRevisionWithComponentKindRegistrationSet(ctx, db, row, sourceRevision, registrations)
	if err != nil && isMissingRuntimeFormulaFunctionError(err) {
		bundles, hasBundles, loadErr := packs.RuntimeFormulaFunctionBundlesForActiveInstalledValidation(ctx, db)
		if loadErr != nil {
			return fmt.Errorf("active runtime formula function bundles: %w", loadErr)
		}
		if hasBundles {
			pack, err = sourceBackedPackFromInstalledRowAtSourceRevisionWithFormulaBundlesAndComponentKindRegistrationSet(ctx, db, row, sourceRevision, bundles, true, registrations)
		}
	}
	if err != nil {
		return err
	}
	if _, err := packs.ValidateStoredPackPayloadHash(pack, storedHash); err != nil {
		return err
	}
	return nil
}

func lowerSet(values []string) map[string]bool {
	out := make(map[string]bool, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value != "" {
			out[strings.ToLower(value)] = true
		}
	}
	return out
}