package controlplane

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"path/filepath"
	"sort"
	"strings"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/formula"
	connectorcapabilities "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"
)

// UpgradeMode is the minimum execution policy admitted by a Catalog-owned
// candidate operation. This first vertical slice deliberately implements only
// the stopped-Runtime executor.
type UpgradeMode string

const UpgradeModeOfflineRequired UpgradeMode = "OFFLINE_REQUIRED"

// OfflineCandidateRequest is one immutable desired source set. Pack order and
// install/reapply/revision-advance classification are intentionally absent:
// those are Controlplane-owned consequences of the target and current state.
type OfflineCandidateRequest struct {
	OperationID        string
	UpgradeMode        UpgradeMode
	PackSources        []string
	ReleaseRepoSources []OfflineCandidateReleaseRepoSource
}

// OfflineCandidateReleaseRepoSource binds one stable release_repo URI to the
// exact repository revision and pack directory used by this stopped candidate.
// Root is operation-local input and is never persisted as source identity.
type OfflineCandidateReleaseRepoSource struct {
	URI      string `json:"uri"`
	PackRoot string `json:"pack_root"`
	Revision string `json:"revision"`
}

// OfflineCandidateReceipt binds the submitted operation to the exact resolved
// target and the state observed before and after the full verify gate.
type OfflineCandidateReceipt struct {
	OperationID          string      `json:"operation_id"`
	UpgradeMode          UpgradeMode `json:"upgrade_mode"`
	PreviousStateDigest  string      `json:"previous_state_digest"`
	TargetStateDigest    string      `json:"target_state_digest"`
	ActivatedStateDigest string      `json:"activated_state_digest"`
	Verified             bool        `json:"verified"`
	Activated            bool        `json:"activated"`
}

type offlineCandidateTarget struct {
	pack        *packs.SourceBackedPack
	payloadHash string
}

type resolvedOfflineCandidate struct {
	targets       []offlineCandidateTarget
	byName        map[string]offlineCandidateTarget
	targetDigest  string
	sourceOptions packs.PackSourceOptions
}

type offlineCandidateExecution struct {
	candidate                           resolvedOfflineCandidate
	receipt                             OfflineCandidateReceipt
	verify                              CandidateFinalVerifier
	plan                                offlineCandidateTransitionPlan
	formulaStorageTransitions           map[string][]packs.FormulaStorageTransitionKey
	preflight                           *installPackPreflightOverrides
	legacySourceAdoptionPackInstallIDs  map[string]string
	releaseSourceAdoptionPackInstallIDs map[string]string
	prepared                            []offlineCandidatePreparedMaterialization
	verified                            bool
}

type offlineCandidatePreparedMaterialization struct {
	applied appliedAppPack
	pack    *packs.SourceBackedPack
	restore func(error) error
}

type offlineCandidateTransitionPlan struct {
	materialize                   []offlineCandidateMaterializationTransition
	sourceRevisionAdvances        []SourceRevisionAdvanceAuthorization
	completeSourceRevisionReplays map[string]struct{}
	legacySourceAdoptions         map[string]struct{}
	releaseSourceAdoptions        map[string]struct{}
}

// isSourceRevisionDelta reports whether the candidate can be completed by the
// immutable pack-generation path alone. This is deliberately narrow: first
// materialization and source-identity adoption still use the full bootstrap
// path, while an exact active revision -> target revision transition already
// has its own begin/execute/verify/activate/compensate lifecycle.
func (plan offlineCandidateTransitionPlan) isSourceRevisionDelta() bool {
	return len(plan.sourceRevisionAdvances) > 0 &&
		len(plan.materialize) == 0 &&
		len(plan.legacySourceAdoptions) == 0 &&
		len(plan.releaseSourceAdoptions) == 0
}

func (plan offlineCandidateTransitionPlan) isMixedTransition() bool {
	return len(plan.materialize) > 0 && len(plan.sourceRevisionAdvances) > 0
}

func (plan offlineCandidateTransitionPlan) deferredSourceBackedLegacyBackfillPacks() map[string]struct{} {
	deferred := make(map[string]struct{}, len(plan.materialize)+len(plan.sourceRevisionAdvances)+len(plan.legacySourceAdoptions)+len(plan.releaseSourceAdoptions))
	for _, transition := range plan.materialize {
		if transition.target.pack != nil {
			deferred[strings.ToLower(strings.TrimSpace(transition.target.pack.Name()))] = struct{}{}
		}
	}
	for _, advance := range plan.sourceRevisionAdvances {
		deferred[strings.ToLower(strings.TrimSpace(advance.Name))] = struct{}{}
	}
	for name := range plan.legacySourceAdoptions {
		deferred[strings.ToLower(strings.TrimSpace(name))] = struct{}{}
	}
	for name := range plan.releaseSourceAdoptions {
		deferred[strings.ToLower(strings.TrimSpace(name))] = struct{}{}
	}
	return deferred
}

func (execution *offlineCandidateExecution) deferredSourceBackedLegacyBackfillPacks() map[string]struct{} {
	if execution == nil {
		return nil
	}
	return execution.plan.deferredSourceBackedLegacyBackfillPacks()
}

type offlineCandidateMaterializationTransition struct {
	target              offlineCandidateTarget
	current             *installedPackRow
	resumePackInstallID string
}

type offlineCandidateRetryAuthority struct {
	activeSourceRevision  string
	hasActiveGeneration   bool
	runtimeObjectSpecs    sql.NullString
	runtimeFormulaBundles sql.NullString
}

func (execution *offlineCandidateExecution) formulaStorageTransitionScope(packName string) []offlinecandidateauthority.FormulaStorageTransition {
	if execution == nil {
		return nil
	}
	keys := execution.formulaStorageTransitions[strings.ToLower(strings.TrimSpace(packName))]
	out := make([]offlinecandidateauthority.FormulaStorageTransition, 0, len(keys))
	for _, key := range keys {
		out = append(out, offlinecandidateauthority.FormulaStorageTransition{
			ObjectAPIName: key.ObjectAPIName,
			FieldAPIName:  key.FieldAPIName,
		})
	}
	return out
}

func runOfflineCandidateFormulaStorageTransition(
	ctx context.Context,
	execution *offlineCandidateExecution,
	packName string,
	run func(context.Context) error,
) error {
	if run == nil {
		return fmt.Errorf("offline candidate formula storage transition callback is unavailable")
	}
	scope := execution.formulaStorageTransitionScope(packName)
	if len(scope) == 0 {
		return run(ctx)
	}
	return offlinecandidateauthority.RunWithFormulaStorageTransitionScope(ctx, packName, scope, run)
}

func offlineCandidateDeclaresWriteMaterializedFormulaStorage(execution *offlineCandidateExecution) (bool, error) {
	if execution == nil {
		return false, nil
	}
	for _, target := range execution.candidate.targets {
		fields, err := packs.WriteMaterializedFormulaStorageFields(target.pack)
		if err != nil {
			return false, err
		}
		if len(fields) != 0 {
			return true, nil
		}
	}
	return false, nil
}

type offlineCandidateContextKey struct{}

func withOfflineCandidateContext(ctx context.Context, execution *offlineCandidateExecution) context.Context {
	if ctx == nil || execution == nil {
		return ctx
	}
	return context.WithValue(ctx, offlineCandidateContextKey{}, execution)
}

// withOfflineCandidateOperationContext binds both halves of one resolved
// candidate snapshot: the frozen pack instances used by installed-row recovery
// and the exact source roots/revisions used by release_repo resolution. This is
// operation-local authority; it never changes process-global source defaults.
func withOfflineCandidateOperationContext(ctx context.Context, execution *offlineCandidateExecution) context.Context {
	if ctx == nil || execution == nil {
		return ctx
	}
	ctx = packs.WithOperationPackSourceOptions(ctx, execution.candidate.sourceOptions)
	return withOfflineCandidateContext(ctx, execution)
}

func offlineCandidatePackFromContext(ctx context.Context, row installedPackRow) (*packs.SourceBackedPack, bool, error) {
	if ctx == nil {
		return nil, false, nil
	}
	execution, _ := ctx.Value(offlineCandidateContextKey{}).(*offlineCandidateExecution)
	if execution == nil {
		return nil, false, nil
	}
	target, ok := execution.candidate.byName[strings.ToLower(strings.TrimSpace(row.Name))]
	if !ok || target.pack == nil {
		return nil, false, nil
	}
	meta := packs.PackHashMetadataForPack(target.pack)
	currentKind, currentURI, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	if err != nil {
		return nil, false, fmt.Errorf("offline candidate pack %s stored source identity: %w", row.Name, err)
	}
	targetKind, targetURI, err := canonicalPackSourceIdentity(meta.SourceKind, meta.SourceURI)
	if err != nil || currentKind != targetKind || currentURI != targetURI ||
		strings.TrimSpace(row.SourceRevision.String) != strings.TrimSpace(meta.SourceRevision) ||
		strings.TrimSpace(row.DesiredPayloadHash.String) != target.payloadHash {
		return nil, false, fmt.Errorf("offline candidate pack %s stored authority does not match the exact candidate", row.Name)
	}
	if row.Status == installedPackStatusActive && strings.TrimSpace(row.AppliedPayloadHash.String) != target.payloadHash {
		return nil, false, fmt.Errorf("offline candidate active pack %s applied payload does not match the exact candidate", row.Name)
	}
	return target.pack, true, nil
}

type offlineCandidateMutableTransition struct {
	current installedPackRow
	target  offlineCandidateTarget
}

func validateOfflineCandidateRequest(request OfflineCandidateRequest) error {
	if strings.TrimSpace(request.OperationID) == "" {
		return fmt.Errorf("offline candidate operation ID is required")
	}
	if request.UpgradeMode != UpgradeModeOfflineRequired {
		return fmt.Errorf("offline candidate upgrade mode must be %q", UpgradeModeOfflineRequired)
	}
	if len(request.PackSources) == 0 {
		return fmt.Errorf("offline candidate requires a complete non-empty pack source set")
	}
	requiredReleaseURIs := make(map[string]struct{})
	for index, source := range request.PackSources {
		if strings.TrimSpace(source) == "" {
			return fmt.Errorf("offline candidate pack source %d is empty", index)
		}
		ref, err := packs.ParsePackSourceRef(source)
		if err != nil {
			return fmt.Errorf("offline candidate pack source %d: %w", index, err)
		}
		if ref.Kind == packs.PackSourceKindReleaseRepo {
			if _, duplicate := requiredReleaseURIs[ref.URI]; duplicate {
				return fmt.Errorf("offline candidate release source URI %q is duplicated", ref.URI)
			}
			requiredReleaseURIs[ref.URI] = struct{}{}
		}
	}
	providedReleaseURIs := make(map[string]struct{}, len(request.ReleaseRepoSources))
	for index, source := range request.ReleaseRepoSources {
		uri := strings.TrimSpace(source.URI)
		ref, err := packs.ParsePackSourceRef(string(packs.PackSourceKindReleaseRepo) + ":" + uri)
		if err != nil || ref.URI != uri {
			return fmt.Errorf("offline candidate release source %d has an invalid canonical URI", index)
		}
		if _, duplicate := providedReleaseURIs[uri]; duplicate {
			return fmt.Errorf("offline candidate release source authority %q is duplicated", uri)
		}
		if _, required := requiredReleaseURIs[uri]; !required {
			return fmt.Errorf("offline candidate release source authority %q is not selected by the candidate", uri)
		}
		packRoot := strings.TrimSpace(source.PackRoot)
		if packRoot == "" || !filepath.IsAbs(packRoot) || filepath.Clean(packRoot) != packRoot || packRoot != source.PackRoot {
			return fmt.Errorf("offline candidate release source %q pack root must be one canonical absolute path", uri)
		}
		if err := packs.ValidateReleaseRepoRevision(source.Revision); err != nil {
			return fmt.Errorf("offline candidate release source %q: %w", uri, err)
		}
		providedReleaseURIs[uri] = struct{}{}
	}
	for uri := range requiredReleaseURIs {
		if _, provided := providedReleaseURIs[uri]; !provided {
			return fmt.Errorf("offline candidate release source %q has no exact source authority", uri)
		}
	}
	return nil
}

func resolveOfflineCandidate(request OfflineCandidateRequest, registrations packs.ComponentKindRegistrationSet) (resolvedOfflineCandidate, error) {
	if err := validateOfflineCandidateRequest(request); err != nil {
		return resolvedOfflineCandidate{}, err
	}
	sourceOptions := packs.DefaultPackSourceOptions()
	if len(request.ReleaseRepoSources) != 0 {
		sourceOptions.ReleaseRepoSources = make(map[string]packs.ReleaseRepoSource, len(request.ReleaseRepoSources))
		for _, source := range request.ReleaseRepoSources {
			sourceOptions.ReleaseRepoSources[source.URI] = packs.ReleaseRepoSource{
				Root: source.PackRoot, Revision: source.Revision,
			}
		}
	}
	resolvedSources := make([]*packs.ResolvedPackSource, 0, len(request.PackSources))
	var formulaBundles []formula.FunctionBundle
	for index, rawSource := range request.PackSources {
		report := packs.ValidatePackSourceWithComponentKindRegistrationSet(rawSource, sourceOptions, registrations)
		if report.HasErrors() || report.Source == nil || report.Source.Manifest == nil {
			return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate source %d: %s", index, report.Error())
		}
		bundles, err := packs.RuntimeFormulaFunctionBundlesFromResolvedSource(report.Source)
		if err != nil {
			return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate source %d formula functions: %w", index, err)
		}
		formulaBundles = append(formulaBundles, bundles...)
		resolvedSources = append(resolvedSources, report.Source)
	}

	candidate := resolvedOfflineCandidate{
		targets:       make([]offlineCandidateTarget, 0, len(request.PackSources)),
		byName:        make(map[string]offlineCandidateTarget, len(request.PackSources)),
		sourceOptions: sourceOptions,
	}
	for index, rawSource := range request.PackSources {
		report := packs.ValidateInstallablePackSourceWithRuntimeFormulaFunctionBundlesAndComponentKindRegistrationSet(
			rawSource, sourceOptions, formulaBundles, registrations,
		)
		if report.HasErrors() || report.Pack == nil {
			return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate source %d is not installable: %s", index, report.Error())
		}
		if report.Source == nil || report.Source != resolvedSources[index] {
			// Both validations resolve the same immutable request. Compare exact
			// source facts below rather than trusting pointer identity.
			if report.Source == nil || !sameResolvedOfflineCandidateSource(report.Source, resolvedSources[index]) {
				return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate source %d changed while resolving", index)
			}
		}
		name := strings.ToLower(strings.TrimSpace(report.Pack.Name()))
		if name == "" {
			return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate source %d resolved an empty pack name", index)
		}
		if _, duplicate := candidate.byName[name]; duplicate {
			return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate contains duplicate pack %q", report.Pack.Name())
		}
		payloadHash, err := packs.PackPayloadHash(report.Pack)
		if err != nil {
			return resolvedOfflineCandidate{}, fmt.Errorf("offline candidate pack %s payload hash: %w", report.Pack.Name(), err)
		}
		target := offlineCandidateTarget{pack: report.Pack, payloadHash: payloadHash}
		candidate.targets = append(candidate.targets, target)
		candidate.byName[name] = target
	}
	ordered, err := orderOfflineCandidateTargets(candidate)
	if err != nil {
		return resolvedOfflineCandidate{}, err
	}
	candidate.targets = ordered
	candidate.targetDigest = digestOfflineCandidateTargets(ordered)
	return candidate, nil
}

func sameResolvedOfflineCandidateSource(left, right *packs.ResolvedPackSource) bool {
	if left == nil || right == nil || left.Manifest == nil || right.Manifest == nil {
		return false
	}
	if strings.TrimSpace(left.Kind) != strings.TrimSpace(right.Kind) ||
		strings.TrimSpace(left.URI) != strings.TrimSpace(right.URI) ||
		strings.TrimSpace(left.Revision) != strings.TrimSpace(right.Revision) ||
		strings.TrimSpace(left.Manifest.Name) != strings.TrimSpace(right.Manifest.Name) ||
		!bytes.Equal(left.ManifestBytes, right.ManifestBytes) ||
		len(left.FileRefs) != len(right.FileRefs) {
		return false
	}
	for index, ref := range left.FileRefs {
		if ref != right.FileRefs[index] || !bytes.Equal(left.Files[ref], right.Files[ref]) {
			return false
		}
	}
	return true
}

func orderOfflineCandidateTargets(candidate resolvedOfflineCandidate) ([]offlineCandidateTarget, error) {
	if len(candidate.targets) != len(candidate.byName) {
		return nil, fmt.Errorf("offline candidate target set is internally inconsistent")
	}
	indegree := make(map[string]int, len(candidate.targets))
	dependents := make(map[string][]string, len(candidate.targets))
	for name := range candidate.byName {
		indegree[name] = 0
	}
	for name, target := range candidate.byName {
		for _, dependency := range target.pack.PackDependencies() {
			dependencyName := strings.ToLower(strings.TrimSpace(dependency.Name))
			if dependencyName == "" {
				return nil, fmt.Errorf("offline candidate pack %s has an empty dependency", target.pack.Name())
			}
			if _, ok := candidate.byName[dependencyName]; !ok {
				return nil, fmt.Errorf("offline candidate is incomplete: pack %s dependency %s is absent", target.pack.Name(), dependency.Name)
			}
			indegree[name]++
			dependents[dependencyName] = append(dependents[dependencyName], name)
		}
	}
	ready := make([]string, 0, len(candidate.targets))
	for name, degree := range indegree {
		if degree == 0 {
			ready = append(ready, name)
		}
	}
	sort.Strings(ready)
	ordered := make([]offlineCandidateTarget, 0, len(candidate.targets))
	for len(ready) > 0 {
		name := ready[0]
		ready = ready[1:]
		ordered = append(ordered, candidate.byName[name])
		for _, dependent := range dependents[name] {
			indegree[dependent]--
			if indegree[dependent] == 0 {
				ready = append(ready, dependent)
				sort.Strings(ready)
			}
		}
	}
	if len(ordered) != len(candidate.targets) {
		return nil, fmt.Errorf("offline candidate pack dependency graph contains a cycle")
	}
	return ordered, nil
}

func planOfflineCandidateTransitions(
	candidate resolvedOfflineCandidate,
	installed []installedPackRow,
	retryAuthorities map[string]offlineCandidateRetryAuthority,
) (offlineCandidateTransitionPlan, error) {
	ordered, err := orderOfflineCandidateTargets(candidate)
	if err != nil {
		return offlineCandidateTransitionPlan{}, err
	}
	installedByName := make(map[string]installedPackRow, len(installed))
	for _, row := range installed {
		name := strings.ToLower(strings.TrimSpace(row.Name))
		if name == "" {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("installed source-backed pack has an empty name")
		}
		if _, duplicate := installedByName[name]; duplicate {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("installed source-backed pack %q is duplicated", row.Name)
		}
		installedByName[name] = row
		if _, present := candidate.byName[name]; !present {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate is incomplete: active installed pack %s is absent", row.Name)
		}
	}

	plan := offlineCandidateTransitionPlan{}
	for _, target := range ordered {
		name := strings.ToLower(strings.TrimSpace(target.pack.Name()))
		row, exists := installedByName[name]
		if !exists {
			plan.materialize = append(plan.materialize, offlineCandidateMaterializationTransition{target: target})
			continue
		}
		if strings.TrimSpace(row.Category) != strings.TrimSpace(string(target.pack.Category())) {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s category changed from %q to %q", row.Name, row.Category, target.pack.Category())
		}
		meta := packs.PackHashMetadataForPack(target.pack)
		targetKind, targetURI, err := canonicalPackSourceIdentity(meta.SourceKind, meta.SourceURI)
		if err != nil {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s target source: %w", row.Name, err)
		}
		currentKind, currentURI, legacySourceAdoption, err := canonicalOfflineCandidateInstalledSourceIdentity(row)
		if err != nil {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s current source: %w", row.Name, err)
		}
		if row.IsDeleted != 0 {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s current lifecycle is deleted", row.Name)
		}
		if legacySourceAdoption {
			if targetKind == string(packs.PackSourceKindReleaseRepo) {
				return offlineCandidateTransitionPlan{}, fmt.Errorf(
					"offline candidate release pack %s cannot adopt a source-less legacy installation; exact local_path source proof is required",
					row.Name,
				)
			}
			if row.Status != installedPackStatusActive || row.Category != string(packs.AppPack) {
				return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s legacy source adoption requires an active app-pack", row.Name)
			}
			if plan.legacySourceAdoptions == nil {
				plan.legacySourceAdoptions = make(map[string]struct{})
			}
			plan.legacySourceAdoptions[name] = struct{}{}
			current := row
			plan.materialize = append(plan.materialize, offlineCandidateMaterializationTransition{
				target:  target,
				current: &current,
			})
			continue
		}
		if row.Status == installedPackStatusInstalling || row.Status == installedPackStatusFailed {
			if err := validateOfflineCandidateRetryTarget(row, target); err != nil {
				return offlineCandidateTransitionPlan{}, err
			}
			switch targetKind {
			case string(packs.PackSourceKindLocalPath):
				if currentKind != targetKind {
					return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s status %q is not an exact local-source retry", row.Name, row.Status)
				}
				current := row
				plan.materialize = append(plan.materialize, offlineCandidateMaterializationTransition{
					target:              target,
					current:             &current,
					resumePackInstallID: strings.TrimSpace(row.PackInstallID),
				})
				continue
			case string(packs.PackSourceKindOfficialRepo), string(packs.PackSourceKindReleaseRepo):
				if currentKind != targetKind || currentURI != targetURI {
					return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate revision-addressed pack %s retry source identity changed", row.Name)
				}
				authority, ok := retryAuthorities[name]
				if !ok {
					return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate revision-addressed pack %s retry authority is unavailable", row.Name)
				}
				if !authority.hasActiveGeneration {
					plan.materialize = append(plan.materialize, offlineCandidateMaterializationTransition{
						target:              target,
						resumePackInstallID: strings.TrimSpace(row.PackInstallID),
					})
					continue
				}
				fromRevision := strings.TrimSpace(authority.activeSourceRevision)
				toRevision := strings.TrimSpace(meta.SourceRevision)
				if fromRevision == "" || toRevision == "" || fromRevision == toRevision {
					return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate revision-addressed pack %s retry requires different active and target revisions", row.Name)
				}
				plan.sourceRevisionAdvances = append(plan.sourceRevisionAdvances, SourceRevisionAdvanceAuthorization{
					Name: row.Name, Kind: targetKind, URI: targetURI,
					FromRevision: fromRevision, ToRevision: toRevision,
				})
				if plan.completeSourceRevisionReplays == nil {
					plan.completeSourceRevisionReplays = make(map[string]struct{})
				}
				plan.completeSourceRevisionReplays[name] = struct{}{}
				continue
			default:
				return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s status %q has unsupported retry source kind %q", row.Name, row.Status, targetKind)
			}
		}
		if row.Status != installedPackStatusActive {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s current lifecycle status %q is unsupported", row.Name, row.Status)
		}
		if currentKind == string(packs.PackSourceKindLocalPath) && targetKind == string(packs.PackSourceKindReleaseRepo) {
			if strings.TrimSpace(row.PackInstallID) == "" ||
				!row.SourceRevision.Valid ||
				strings.TrimSpace(row.SourceRevision.String) == "" ||
				strings.TrimSpace(row.SourceRevision.String) != strings.TrimSpace(meta.SourceRevision) ||
				strings.TrimSpace(row.DesiredPayloadHash.String) != target.payloadHash ||
				strings.TrimSpace(row.AppliedPayloadHash.String) != target.payloadHash {
				return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate release pack %s cannot adopt mutable source: exact active installation, revision, and payload are required", row.Name)
			}
			if plan.releaseSourceAdoptions == nil {
				plan.releaseSourceAdoptions = make(map[string]struct{})
			}
			plan.releaseSourceAdoptions[name] = struct{}{}
			current := row
			plan.materialize = append(plan.materialize, offlineCandidateMaterializationTransition{
				target:  target,
				current: &current,
			})
			continue
		}
		if packs.IsRevisionAddressedPackSourceKind(targetKind) {
			if currentKind != targetKind || currentURI != targetURI {
				return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate revision-addressed pack %s source identity changed", row.Name)
			}
			fromRevision := strings.TrimSpace(row.SourceRevision.String)
			toRevision := strings.TrimSpace(meta.SourceRevision)
			if fromRevision == "" || toRevision == "" {
				return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate revision-addressed pack %s requires exact current and target revisions", row.Name)
			}
			if fromRevision != toRevision {
				plan.sourceRevisionAdvances = append(plan.sourceRevisionAdvances, SourceRevisionAdvanceAuthorization{
					Name: row.Name, Kind: targetKind, URI: targetURI,
					FromRevision: fromRevision, ToRevision: toRevision,
				})
				continue
			}
			if strings.TrimSpace(row.AppliedPayloadHash.String) != target.payloadHash {
				return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate revision-addressed pack %s changed payload at immutable revision %s", row.Name, toRevision)
			}
			continue
		}
		if targetKind != string(packs.PackSourceKindLocalPath) {
			return offlineCandidateTransitionPlan{}, fmt.Errorf("offline candidate pack %s has unsupported source kind %q", row.Name, targetKind)
		}
		if currentKind != targetKind || currentURI != targetURI ||
			strings.TrimSpace(row.SourceRevision.String) != strings.TrimSpace(meta.SourceRevision) ||
			strings.TrimSpace(row.AppliedPayloadHash.String) != target.payloadHash ||
			strings.TrimSpace(row.DesiredPayloadHash.String) != target.payloadHash {
			current := row
			plan.materialize = append(plan.materialize, offlineCandidateMaterializationTransition{
				target:  target,
				current: &current,
			})
		}
	}
	return plan, nil
}

func canonicalOfflineCandidateInstalledSourceIdentity(row installedPackRow) (kind string, uri string, legacyAdoption bool, err error) {
	if strings.TrimSpace(row.SourceKind.String) != "" && strings.TrimSpace(row.SourceURI.String) != "" {
		kind, uri, err = canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
		return kind, uri, false, err
	}
	source, err := legacyInstalledPackAdoptionSource(row)
	if err != nil {
		return "", "", false, err
	}
	ref, err := packs.ParsePackSourceRef(source)
	if err != nil {
		return "", "", false, err
	}
	return strings.TrimSpace(string(ref.Kind)), strings.TrimSpace(ref.URI), true, nil
}

func validateOfflineCandidateRetryTarget(row installedPackRow, target offlineCandidateTarget) error {
	if target.pack == nil {
		return fmt.Errorf("offline candidate retry target is unavailable")
	}
	adjusted := row
	meta := packs.PackHashMetadataForPack(target.pack)
	adjusted.SourceURI = sql.NullString{String: meta.SourceURI, Valid: meta.SourceURIValid}
	if err := verifyRetryableInstalledPackSourceState(adjusted, target.pack, target.payloadHash); err != nil {
		return fmt.Errorf("offline candidate pack %s exact retry: %w", row.Name, err)
	}
	return nil
}

func validateOfflineCandidateSourceTransition(
	row installedPackRow,
	pack *packs.SourceBackedPack,
	payloadHash string,
	transition *offlineCandidateMutableTransition,
) error {
	if transition == nil || pack == nil {
		return fmt.Errorf("transition proof and target pack are required")
	}
	current := transition.current
	target := transition.target
	if target.pack == nil ||
		strings.TrimSpace(row.Name) != strings.TrimSpace(current.Name) ||
		strings.TrimSpace(row.Name) != strings.TrimSpace(pack.Name()) ||
		strings.TrimSpace(target.pack.Name()) != strings.TrimSpace(pack.Name()) ||
		strings.TrimSpace(row.Category) != strings.TrimSpace(current.Category) ||
		strings.TrimSpace(row.Category) != strings.TrimSpace(string(pack.Category())) ||
		row.Status != current.Status ||
		row.IsDeleted != 0 || current.IsDeleted != 0 {
		return fmt.Errorf("installed lifecycle facts do not match the planned pack")
	}
	if current.Status != installedPackStatusActive {
		if err := validateOfflineCandidateRetryTarget(current, target); err != nil {
			return err
		}
	}
	if !offlineCandidateInstalledTransitionFactsMatch(row, current) {
		return fmt.Errorf("installed source/hash facts changed after candidate planning")
	}
	targetMeta := packs.PackHashMetadataForPack(target.pack)
	requestedMeta := packs.PackHashMetadataForPack(pack)
	targetKind, targetURI, err := canonicalPackSourceIdentity(targetMeta.SourceKind, targetMeta.SourceURI)
	if err != nil {
		return fmt.Errorf("planned target source is invalid")
	}
	requestedKind, requestedURI, err := canonicalPackSourceIdentity(requestedMeta.SourceKind, requestedMeta.SourceURI)
	if err != nil || requestedKind != targetKind || requestedURI != targetURI ||
		requestedMeta.SourceRevisionValid != targetMeta.SourceRevisionValid ||
		strings.TrimSpace(requestedMeta.SourceRevision) != strings.TrimSpace(targetMeta.SourceRevision) ||
		strings.TrimSpace(payloadHash) != strings.TrimSpace(target.payloadHash) {
		return fmt.Errorf("requested source does not match the exact candidate target")
	}
	if strings.TrimSpace(current.SourceKind.String) == "" || strings.TrimSpace(current.SourceURI.String) == "" {
		if _, err := legacyInstalledPackAdoptionSource(current); err != nil {
			return fmt.Errorf("planned current source is not an admitted legacy identity")
		}
		return nil
	}
	currentKind, _, err := canonicalPackSourceIdentity(current.SourceKind.String, current.SourceURI.String)
	if err != nil || currentKind != string(packs.PackSourceKindLocalPath) {
		return fmt.Errorf("planned current source is not local_path")
	}
	switch targetKind {
	case string(packs.PackSourceKindLocalPath):
		return nil
	case string(packs.PackSourceKindReleaseRepo):
		if current.Status != installedPackStatusActive ||
			strings.TrimSpace(current.PackInstallID) == "" ||
			!current.SourceRevision.Valid ||
			strings.TrimSpace(current.SourceRevision.String) == "" ||
			strings.TrimSpace(current.SourceRevision.String) != strings.TrimSpace(targetMeta.SourceRevision) ||
			strings.TrimSpace(current.DesiredPayloadHash.String) != strings.TrimSpace(target.payloadHash) ||
			strings.TrimSpace(current.AppliedPayloadHash.String) != strings.TrimSpace(target.payloadHash) {
			return fmt.Errorf("planned release source adoption is not the exact active installation")
		}
		return nil
	default:
		return fmt.Errorf("planned target must be local_path or release_repo")
	}
}

func offlineCandidateInstalledTransitionFactsMatch(left, right installedPackRow) bool {
	return installedPackCommitNullStringEquals(left.PackVersion, right.PackVersion.String, right.PackVersion.Valid) &&
		installedPackCommitNullStringEquals(left.ManifestFormatVersion, right.ManifestFormatVersion.String, right.ManifestFormatVersion.Valid) &&
		installedPackCommitNullStringEquals(left.PlatformCompatibility, right.PlatformCompatibility.String, right.PlatformCompatibility.Valid) &&
		installedPackCommitNullStringEquals(left.DesiredPayloadHash, right.DesiredPayloadHash.String, right.DesiredPayloadHash.Valid) &&
		installedPackCommitNullStringEquals(left.AppliedPayloadHash, right.AppliedPayloadHash.String, right.AppliedPayloadHash.Valid) &&
		installedPackCommitNullStringEquals(left.PayloadHashAlgorithm, right.PayloadHashAlgorithm.String, right.PayloadHashAlgorithm.Valid) &&
		installedPackCommitNullStringEquals(left.SourceKind, right.SourceKind.String, right.SourceKind.Valid) &&
		installedPackCommitNullStringEquals(left.SourceURI, right.SourceURI.String, right.SourceURI.Valid) &&
		installedPackCommitNullStringEquals(left.SourceRevision, right.SourceRevision.String, right.SourceRevision.Valid) &&
		installedPackCommitNullStringEquals(left.HostRequirementsDigest, right.HostRequirementsDigest.String, right.HostRequirementsDigest.Valid)
}

func digestOfflineCandidateTargets(targets []offlineCandidateTarget) string {
	hash := sha256.New()
	for _, target := range targets {
		meta := packs.PackHashMetadataForPack(target.pack)
		_, _ = fmt.Fprintf(hash, "%s\x00%s\x00%s\x00%s\x00%s\n",
			strings.TrimSpace(target.pack.Name()), strings.TrimSpace(meta.SourceKind),
			strings.TrimSpace(meta.SourceURI), strings.TrimSpace(meta.SourceRevision),
			strings.TrimSpace(target.payloadHash))
	}
	return "sha256:" + hex.EncodeToString(hash.Sum(nil))
}

func digestOfflineCandidateInstalledRows(rows []installedPackRow) string {
	rows = append([]installedPackRow(nil), rows...)
	sort.Slice(rows, func(i, j int) bool { return strings.ToLower(rows[i].Name) < strings.ToLower(rows[j].Name) })
	hash := sha256.New()
	for _, row := range rows {
		_, _ = fmt.Fprintf(hash, "%s\x00%s\x00%s\x00%s\x00%s\n",
			strings.TrimSpace(row.Name), strings.TrimSpace(row.SourceKind.String),
			strings.TrimSpace(row.SourceURI.String), strings.TrimSpace(row.SourceRevision.String),
			strings.TrimSpace(row.AppliedPayloadHash.String))
	}
	return "sha256:" + hex.EncodeToString(hash.Sum(nil))
}

// ApplyOfflineCandidate is the Catalog-owned stopped-Runtime vertical slice.
// UpgradeModeOfflineRequired declares the deployment precondition; it does not
// stop Runtime, so the deployment must quiesce every Runtime writer before
// calling. Formula-storage authority is minted later around each exact internal
// candidate-pack apply and never reaches public verification hooks.
func ApplyOfflineCandidate(ctx context.Context, db *database.TiDBConnection, request OfflineCandidateRequest, options InstallPackOptions) (OfflineCandidateReceipt, error) {
	return ApplyOfflineCandidateWithRunnerOptions(ctx, db, request, options, RunnerOptions{})
}

func ApplyOfflineCandidateWithRunnerOptions(ctx context.Context, db *database.TiDBConnection, request OfflineCandidateRequest, options InstallPackOptions, runnerOptions RunnerOptions) (OfflineCandidateReceipt, error) {
	if err := validateOfflineCandidateRequest(request); err != nil {
		return OfflineCandidateReceipt{}, err
	}
	if db == nil {
		return OfflineCandidateReceipt{}, fmt.Errorf("offline candidate database connection is unavailable")
	}
	return applyOfflineCandidate(ctx, db, request, options, runnerOptions)
}

func applyOfflineCandidate(
	ctx context.Context,
	db *database.TiDBConnection,
	request OfflineCandidateRequest,
	options InstallPackOptions,
	runnerOptions RunnerOptions,
) (OfflineCandidateReceipt, error) {
	if options.SkipPolymorphicDiscriminatorBackfill {
		return OfflineCandidateReceipt{}, fmt.Errorf("offline candidate requires a complete ApplyAll; skip polymorphic discriminator backfill is not supported")
	}
	if options.SourceRevisionAdvance != nil || len(options.ApplyAllSourceRevisionAdvances) != 0 {
		return OfflineCandidateReceipt{}, fmt.Errorf("offline candidate callers must not submit source transition classification")
	}
	candidate, err := resolveOfflineCandidate(request, runnerOptions.ComponentKindRegistrationSet)
	if err != nil {
		return OfflineCandidateReceipt{}, err
	}
	execution := &offlineCandidateExecution{
		candidate: candidate,
		verify:    options.CandidateFinalVerify,
		receipt: OfflineCandidateReceipt{
			OperationID:       strings.TrimSpace(request.OperationID),
			UpgradeMode:       request.UpgradeMode,
			TargetStateDigest: candidate.targetDigest,
		},
	}
	options.offlineCandidate = execution
	_, err = applyAllWithOptions(ctx, db, options, runnerOptions, applyAllCompletionPolicyExactCandidateRetry)
	if err != nil {
		return execution.receipt, err
	}
	if !execution.receipt.Verified {
		return execution.receipt, fmt.Errorf("offline candidate ApplyAll completed without an exact verified candidate")
	}
	execution.receipt.ActivatedStateDigest = candidate.targetDigest
	execution.receipt.Activated = true
	return execution.receipt, nil
}

// offlineCandidateMatchesActivatedState classifies only the immutable
// source-backed target. Its caller owns the apply lock and must separately
// prove the completed ApplyAll marker, release gates, rebuild-debt state, and
// distribution verification before treating this match as a no-op.
func offlineCandidateMatchesActivatedState(
	ctx context.Context,
	db *sql.DB,
	execution *offlineCandidateExecution,
) (bool, []installedPackRow, error) {
	if execution == nil {
		return false, nil, nil
	}
	rows, err := listInstalledSourceBackedPacks(ctx, db)
	if err != nil {
		return false, nil, fmt.Errorf("read active source-backed pack set: %w", err)
	}
	execution.receipt.PreviousStateDigest = digestOfflineCandidateInstalledRows(rows)
	if len(rows) != len(execution.candidate.targets) {
		return false, rows, nil
	}
	for _, row := range rows {
		target, ok := execution.candidate.byName[strings.ToLower(strings.TrimSpace(row.Name))]
		if !ok || verifyOfflineCandidateInstalledTarget(row, target) != nil {
			return false, rows, nil
		}
	}
	return true, rows, nil
}

func prepareOfflineCandidateUnderApplyLock(
	ctx context.Context,
	db *sql.DB,
	options *InstallPackOptions,
) error {
	if options == nil || options.offlineCandidate == nil {
		return nil
	}
	rows, err := listInstalledSourceBackedPacks(ctx, db)
	if err != nil {
		return fmt.Errorf("read current source-backed pack set: %w", err)
	}
	return prepareOfflineCandidateFromInstalledRowsUnderApplyLock(ctx, db, options, rows)
}

func prepareOfflineCandidateFromInstalledRowsUnderApplyLock(
	ctx context.Context,
	db *sql.DB,
	options *InstallPackOptions,
	rows []installedPackRow,
) error {
	if options == nil || options.offlineCandidate == nil {
		return nil
	}
	execution := options.offlineCandidate
	rows, retryAuthorities, err := hydrateOfflineCandidateAuthorityRows(ctx, db, rows, execution.candidate)
	if err != nil {
		return err
	}
	execution.receipt.PreviousStateDigest = digestOfflineCandidateInstalledRows(rows)
	plan, err := planOfflineCandidateTransitions(execution.candidate, rows, retryAuthorities)
	if err != nil {
		return err
	}
	formulaStorageTransitions, err := planOfflineCandidateFormulaStorageTransitions(
		rows,
		execution.candidate,
		retryAuthorities,
	)
	if err != nil {
		return err
	}
	preflight, err := offlineCandidatePreflightOverrides(execution.candidate)
	if err != nil {
		return err
	}
	preflight.offlineCandidatePrepared = execution
	execution.plan = plan
	for name := range plan.releaseSourceAdoptions {
		for _, transition := range plan.materialize {
			if transition.current == nil || !strings.EqualFold(transition.current.Name, name) {
				continue
			}
			if execution.releaseSourceAdoptionPackInstallIDs == nil {
				execution.releaseSourceAdoptionPackInstallIDs = make(map[string]string)
			}
			execution.releaseSourceAdoptionPackInstallIDs[name] = strings.TrimSpace(transition.current.PackInstallID)
		}
	}
	execution.preflight = preflight
	execution.formulaStorageTransitions = formulaStorageTransitions
	options.ApplyAllSourceRevisionAdvances = append(
		[]SourceRevisionAdvanceAuthorization(nil),
		plan.sourceRevisionAdvances...,
	)
	options.offlineCandidateCompleteSourceRevisionReplays = make(
		map[string]struct{}, len(plan.completeSourceRevisionReplays),
	)
	for name := range plan.completeSourceRevisionReplays {
		options.offlineCandidateCompleteSourceRevisionReplays[name] = struct{}{}
	}
	if len(plan.materialize) == 0 && execution.verify != nil {
		options.CandidateFinalVerify = func(ctx context.Context, runner *Runner, _ []packs.Pack) error {
			if execution.verified {
				return nil
			}
			ready, err := offlineCandidateReadyForPreActivationFinalVerify(ctx, runner, execution)
			if err != nil {
				return err
			}
			if !ready {
				return nil
			}
			if err := execution.verify(ctx, runner, execution.candidate.packList()); err != nil {
				return err
			}
			execution.verified = true
			return nil
		}
	} else {
		options.CandidateFinalVerify = nil
	}
	return nil
}

func offlineCandidateReadyForPreActivationFinalVerify(
	ctx context.Context,
	r *Runner,
	execution *offlineCandidateExecution,
) (bool, error) {
	if execution == nil {
		return false, fmt.Errorf("offline candidate execution is unavailable")
	}
	rows, err := listInstalledSourceBackedPacks(ctx, runnerSQLDB(r))
	if err != nil {
		return false, fmt.Errorf("read pre-activation candidate state: %w", err)
	}
	if len(rows) != len(execution.candidate.targets) {
		return false, nil
	}
	return offlineCandidateRowsReadyForPreActivationFinalVerify(rows, execution.candidate), nil
}

func offlineCandidateRowsReadyForPreActivationFinalVerify(
	rows []installedPackRow,
	candidate resolvedOfflineCandidate,
) bool {
	if len(rows) != len(candidate.targets) {
		return false
	}
	installing := 0
	for _, row := range rows {
		target, ok := candidate.byName[strings.ToLower(strings.TrimSpace(row.Name))]
		if !ok {
			return false
		}
		switch row.Status {
		case installedPackStatusActive:
			if err := verifyOfflineCandidateInstalledTarget(row, target); err != nil {
				return false
			}
		case installedPackStatusInstalling:
			installing++
			if !offlineCandidateInstallingRowMatchesTarget(row, target) {
				return false
			}
		default:
			return false
		}
	}
	return installing == 1
}

func offlineCandidateInstallingRowMatchesTarget(row installedPackRow, target offlineCandidateTarget) bool {
	if target.pack == nil || row.IsDeleted != 0 || row.Status != installedPackStatusInstalling {
		return false
	}
	meta := packs.PackHashMetadataForPack(target.pack)
	currentKind, currentURI, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	if err != nil {
		return false
	}
	targetKind, targetURI, err := canonicalPackSourceIdentity(meta.SourceKind, meta.SourceURI)
	if err != nil {
		return false
	}
	return strings.TrimSpace(row.Category) == strings.TrimSpace(string(target.pack.Category())) &&
		strings.TrimSpace(row.DesiredPayloadHash.String) == target.payloadHash &&
		currentKind == targetKind && currentURI == targetURI &&
		strings.TrimSpace(row.SourceRevision.String) == strings.TrimSpace(meta.SourceRevision)
}

func requireOfflineCandidatePreActivationVerification(execution *offlineCandidateExecution) error {
	if execution == nil || execution.verify == nil || len(execution.plan.materialize) != 0 || len(execution.plan.sourceRevisionAdvances) == 0 {
		return nil
	}
	if !execution.verified {
		return fmt.Errorf("offline candidate source revision transitions completed without final pre-activation verification")
	}
	return nil
}

func hydrateOfflineCandidateAuthorityRows(
	ctx context.Context,
	db *sql.DB,
	rows []installedPackRow,
	candidate resolvedOfflineCandidate,
) ([]installedPackRow, map[string]offlineCandidateRetryAuthority, error) {
	out := append([]installedPackRow(nil), rows...)
	authorities := map[string]offlineCandidateRetryAuthority{}
	for index, row := range out {
		isRetry := row.Status == installedPackStatusInstalling || row.Status == installedPackStatusFailed
		needsReleaseAdoptionAuthority := false
		if !isRetry && row.Status == installedPackStatusActive {
			target, ok := candidate.byName[strings.ToLower(strings.TrimSpace(row.Name))]
			if ok && target.pack != nil {
				currentKind, _, currentErr := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
				targetMeta := packs.PackHashMetadataForPack(target.pack)
				targetKind, _, targetErr := canonicalPackSourceIdentity(targetMeta.SourceKind, targetMeta.SourceURI)
				needsReleaseAdoptionAuthority = currentErr == nil && targetErr == nil &&
					currentKind == string(packs.PackSourceKindLocalPath) &&
					targetKind == string(packs.PackSourceKindReleaseRepo)
			}
		}
		if !isRetry && !needsReleaseAdoptionAuthority {
			continue
		}
		full, exists, err := getInstalledPackRowByNameForSourceRevisionAdvance(ctx, db, row.Name)
		if err != nil {
			return nil, nil, fmt.Errorf("read offline candidate authority row %s: %w", row.Name, err)
		}
		if !exists {
			return nil, nil, fmt.Errorf("offline candidate authority row %s disappeared", row.Name)
		}
		copyInstalledPackRuntimeDeclarationState(&full, row)
		out[index] = full
		kind, _, err := canonicalPackSourceIdentity(full.SourceKind.String, full.SourceURI.String)
		if err != nil {
			return nil, nil, fmt.Errorf("offline candidate authority row %s source: %w", row.Name, err)
		}
		if isRetry && packs.IsRevisionAddressedPackSourceKind(kind) {
			authority, err := readOfflineCandidateRevisionRetryAuthority(ctx, db, full)
			if err != nil {
				return nil, nil, err
			}
			authorities[strings.ToLower(strings.TrimSpace(full.Name))] = authority
		}
	}
	return out, authorities, nil
}

func readOfflineCandidateRevisionRetryAuthority(
	ctx context.Context,
	db *sql.DB,
	row installedPackRow,
) (offlineCandidateRetryAuthority, error) {
	if db == nil {
		return offlineCandidateRetryAuthority{}, fmt.Errorf("offline candidate revision-addressed retry database is unavailable")
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return offlineCandidateRetryAuthority{}, fmt.Errorf("read offline candidate revision-addressed pack %s retry authority: begin transaction: %w", row.Name, err)
	}
	defer func() { _ = tx.Rollback() }()

	compatibility, exists, err := readPackGenerationCompatibilityForUpdate(ctx, tx, row.Name)
	if err != nil {
		return offlineCandidateRetryAuthority{}, err
	}
	if !exists ||
		compatibility.packInstallID != strings.TrimSpace(row.PackInstallID) ||
		compatibility.status != strings.TrimSpace(row.Status) ||
		compatibility.isDeleted != row.IsDeleted ||
		compatibility.appliedPayloadHash != row.AppliedPayloadHash {
		return offlineCandidateRetryAuthority{}, fmt.Errorf("offline candidate revision-addressed pack %s retry row changed during authority admission", row.Name)
	}
	activation, exists, err := readPackActivationForUpdate(ctx, tx, row.Name)
	if err != nil {
		return offlineCandidateRetryAuthority{}, err
	}
	if !exists {
		installIDs, err := readNonTerminalPackGenerationInstallIDsForUpdate(ctx, tx, row.Name)
		if err != nil {
			return offlineCandidateRetryAuthority{}, err
		}
		if len(installIDs) != 0 {
			return offlineCandidateRetryAuthority{}, fmt.Errorf("offline candidate revision-addressed pack %s has generation %s without activation", row.Name, installIDs[0])
		}
		return offlineCandidateRetryAuthority{}, nil
	}
	active, err := verifyActivePackGeneration(ctx, tx, activation)
	if err != nil {
		return offlineCandidateRetryAuthority{}, fmt.Errorf("offline candidate revision-addressed pack %s active generation: %w", row.Name, err)
	}
	currentKind, currentURI, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	if err != nil {
		return offlineCandidateRetryAuthority{}, err
	}
	activeKind, activeURI, err := canonicalPackSourceIdentity(active.sourceKind.String, active.sourceURI.String)
	if err != nil || activeKind != currentKind || activeURI != currentURI {
		return offlineCandidateRetryAuthority{}, fmt.Errorf("offline candidate revision-addressed pack %s active generation source identity changed", row.Name)
	}
	if strings.TrimSpace(active.sourceRevision.String) == "" || strings.TrimSpace(active.packInstallID) == strings.TrimSpace(row.PackInstallID) {
		return offlineCandidateRetryAuthority{}, fmt.Errorf("offline candidate revision-addressed pack %s active generation retry authority is invalid", row.Name)
	}
	return offlineCandidateRetryAuthority{
		activeSourceRevision:  strings.TrimSpace(active.sourceRevision.String),
		hasActiveGeneration:   true,
		runtimeObjectSpecs:    active.runtimeObjectSpecDecl,
		runtimeFormulaBundles: active.runtimeFormulaBundles,
	}, nil
}

func planOfflineCandidateFormulaStorageTransitions(
	rows []installedPackRow,
	candidate resolvedOfflineCandidate,
	retryAuthorities map[string]offlineCandidateRetryAuthority,
) (map[string][]packs.FormulaStorageTransitionKey, error) {
	type comparisonResult struct {
		name       string
		comparison packs.FormulaStorageTransitionComparison
		err        error
	}
	results := make([]comparisonResult, 0, len(rows))
	matched := make(map[string]struct{}, len(rows))
	transitions := make(map[string][]packs.FormulaStorageTransitionKey)
	for _, row := range rows {
		name := strings.ToLower(strings.TrimSpace(row.Name))
		target, ok := candidate.byName[name]
		if !ok || target.pack == nil {
			results = append(results, comparisonResult{name: row.Name})
			continue
		}
		matched[name] = struct{}{}
		activeObjectSpecs := row.RuntimeObjectSpecDecl
		activeFormulaBundles := row.RuntimeFormulaBundles
		if authority := retryAuthorities[name]; authority.hasActiveGeneration {
			activeObjectSpecs = authority.runtimeObjectSpecs
			activeFormulaBundles = authority.runtimeFormulaBundles
		}
		if !activeObjectSpecs.Valid || !activeFormulaBundles.Valid {
			results = append(results, comparisonResult{
				name: row.Name,
				err:  fmt.Errorf("active runtime declarations are unavailable"),
			})
			continue
		}
		comparison, compareErr := packs.CompareFormulaStorageTransitionRuntimeDeclarations(
			activeObjectSpecs.String,
			activeFormulaBundles.String,
			target.pack,
		)
		results = append(results, comparisonResult{name: row.Name, comparison: comparison, err: compareErr})
		if compareErr == nil && len(comparison.Transitions) != 0 {
			transitions[name] = append([]packs.FormulaStorageTransitionKey(nil), comparison.Transitions...)
		}
	}
	if len(transitions) == 0 {
		return nil, nil
	}
	if len(matched) != len(candidate.targets) || len(rows) != len(candidate.targets) {
		return nil, fmt.Errorf("formula storage transition requires an unchanged exact candidate pack set")
	}
	for _, result := range results {
		if result.err != nil {
			return nil, fmt.Errorf("formula storage transition candidate %s: %w", result.name, result.err)
		}
		if !result.comparison.SemanticsEqual {
			return nil, fmt.Errorf(
				"formula storage transition requires a candidate-wide storage-only release; runtime object or formula-function semantics changed in pack %s",
				result.name,
			)
		}
	}
	return transitions, nil
}

func prepareOfflineCandidateMaterializationsUnderApplyLock(
	ctx context.Context,
	r *Runner,
	options *InstallPackOptions,
	selected ...offlineCandidateMaterializationTransition,
) error {
	if options == nil || options.offlineCandidate == nil {
		return nil
	}
	execution := options.offlineCandidate
	preflight := execution.preflight
	if preflight == nil {
		return fmt.Errorf("offline candidate materialization was not prepared")
	}
	applyTarget := func(stage string, target offlineCandidateTarget, transition *offlineCandidateMutableTransition, resumePackInstallID string) error {
		candidateOptions := *options
		candidateOptions.CandidateFinalVerify = nil
		candidateOptions.SourceRevisionAdvance = nil
		candidateOptions.ApplyAllSourceRevisionAdvances = nil
		candidateOptions.offlineCandidate = nil
		candidateOptions.offlineCandidatePreparation = execution
		candidateOptions.offlineCandidateLocalSourceTransition = transition
		candidateOptions.resumeSourceBackedPackInstallID = strings.TrimSpace(resumePackInstallID)
		if _, legacySourceAdoption := execution.plan.legacySourceAdoptions[strings.ToLower(strings.TrimSpace(target.pack.Name()))]; legacySourceAdoption {
			packInstallID := strings.TrimSpace(execution.legacySourceAdoptionPackInstallIDs[strings.ToLower(strings.TrimSpace(target.pack.Name()))])
			if packInstallID == "" {
				return fmt.Errorf("offline candidate %s pack %s legacy source adoption identity is unavailable", stage, target.pack.Name())
			}
			candidateOptions.legacySourceIdentityAdoptionPackInstallID = packInstallID
		}
		if _, releaseSourceAdoption := execution.plan.releaseSourceAdoptions[strings.ToLower(strings.TrimSpace(target.pack.Name()))]; releaseSourceAdoption {
			packInstallID := strings.TrimSpace(execution.releaseSourceAdoptionPackInstallIDs[strings.ToLower(strings.TrimSpace(target.pack.Name()))])
			if packInstallID == "" {
				return fmt.Errorf("offline candidate %s pack %s release source adoption identity is unavailable", stage, target.pack.Name())
			}
			candidateOptions.releaseSourceIdentityAdoptionPackInstallID = packInstallID
		}
		payloadHash, err := packs.PackPayloadHash(target.pack)
		if err != nil || payloadHash != target.payloadHash {
			return fmt.Errorf("offline candidate %s pack %s frozen payload changed", stage, target.pack.Name())
		}
		allowedObjects, err := samePackAllowedOwnerlessObjectsWithOptionsAndComponentKindRegistrationSet(
			ctx,
			runnerSQLDB(r),
			target.pack,
			target.payloadHash,
			candidateOptions,
			r.componentKindRegistrationSet(),
		)
		if err != nil {
			return fmt.Errorf("offline candidate %s pack %s ownerless admission: %w", stage, target.pack.Name(), err)
		}
		pack := target.pack
		if len(allowedObjects) > 0 {
			pack, err = packs.RebuildSourceBackedPackWithAllowedOwnerlessObjectAPIs(target.pack, allowedObjects)
			if err != nil {
				return fmt.Errorf("offline candidate %s pack %s frozen rebuild: %w", stage, target.pack.Name(), err)
			}
		}
		preflight.skipAttributionPreflight = candidateOptions.SkipAttributionPreflight
		apply := func(applyCtx context.Context) error {
			return applyResolvedSourceBackedPackForInstall(
				applyCtx,
				r,
				pack,
				preflight,
				candidateOptions,
			)
		}
		err = runOfflineCandidateFormulaStorageTransition(ctx, execution, pack.Name(), apply)
		if err != nil {
			return fmt.Errorf("offline candidate %s pack %s: %w", stage, target.pack.Name(), err)
		}
		return nil
	}
	transitions := execution.plan.materialize
	if len(selected) != 0 {
		if len(selected) != 1 {
			return fmt.Errorf("offline candidate materialization selection must contain exactly one transition")
		}
		transitions = selected
	}
	for _, transition := range transitions {
		stage := "install"
		var proof *offlineCandidateMutableTransition
		if transition.current != nil {
			stage = "converge mutable"
			proof = &offlineCandidateMutableTransition{
				current: *transition.current,
				target:  transition.target,
			}
		}
		if err := applyTarget(stage, transition.target, proof, transition.resumePackInstallID); err != nil {
			return failOfflineCandidatePreparedMaterializations(ctx, r, execution, err)
		}
	}
	return nil
}

func stageOfflineCandidatePreparedMaterialization(
	execution *offlineCandidateExecution,
	applied appliedAppPack,
	restore func(error) error,
) error {
	if execution == nil || applied.pack == nil || restore == nil {
		return fmt.Errorf("offline candidate prepared materialization is incomplete")
	}
	pack, ok := applied.pack.(*packs.SourceBackedPack)
	if !ok {
		return fmt.Errorf("offline candidate prepared pack %s is not source-backed", applied.pack.Name())
	}
	for _, prepared := range execution.prepared {
		if strings.EqualFold(prepared.applied.pack.Name(), applied.pack.Name()) {
			return fmt.Errorf("offline candidate pack %s was prepared more than once", applied.pack.Name())
		}
	}
	execution.prepared = append(execution.prepared, offlineCandidatePreparedMaterialization{
		applied: applied,
		pack:    pack,
		restore: restore,
	})
	return nil
}

func failOfflineCandidatePreparedMaterializations(
	ctx context.Context,
	r *Runner,
	execution *offlineCandidateExecution,
	applyErr error,
) error {
	prepared := takeOfflineCandidatePreparedMaterializations(execution)
	if len(prepared) == 0 {
		return applyErr
	}
	return failOfflineCandidatePreparedMaterializationSet(ctx, r, prepared, applyErr)
}

func takeOfflineCandidatePreparedMaterializations(
	execution *offlineCandidateExecution,
) []offlineCandidatePreparedMaterialization {
	if execution == nil || len(execution.prepared) == 0 {
		return nil
	}
	prepared := execution.prepared
	execution.prepared = nil
	return prepared
}

func failOfflineCandidatePreparedMaterializationSet(
	ctx context.Context,
	r *Runner,
	preparedSet []offlineCandidatePreparedMaterialization,
	applyErr error,
) error {
	applied := make([]appliedAppPack, 0, len(preparedSet))
	errs := []error{applyErr}
	for _, prepared := range preparedSet {
		applied = append(applied, prepared.applied)
		if err := prepared.restore(applyErr); err != nil {
			errs = append(errs, err)
		}
	}
	if err := markAppliedAppPacksFailed(ctx, runnerSQLDB(r), applied, applyErr); err != nil {
		errs = append(errs, fmt.Errorf("mark offline candidate materializations failed: %w", err))
	}
	if err := restoreInstalledRuntimeMetadataAfterInstallFailure(ctx, r, runnerSQLDB(r)); err != nil {
		errs = append(errs, err)
	}
	return errors.Join(errs...)
}

func finalizeOfflineCandidatePreparedMaterializations(
	ctx context.Context,
	r *Runner,
	options *InstallPackOptions,
	execution *offlineCandidateExecution,
) error {
	return finalizeOfflineCandidatePreparedMaterializationsWithVerification(
		ctx, r, options, execution, true,
	)
}

func finalizeOfflineCandidatePreparedMaterializationsWithVerification(
	ctx context.Context,
	r *Runner,
	options *InstallPackOptions,
	execution *offlineCandidateExecution,
	verifyCandidate bool,
) error {
	if execution == nil || len(execution.prepared) == 0 {
		return nil
	}
	applied := make([]appliedAppPack, 0, len(execution.prepared))
	for _, prepared := range execution.prepared {
		applied = append(applied, prepared.applied)
	}
	verifyCtx := withRuntimeEffectivePeerDeclarationOverrides(ctx, execution.preflight)
	var connectedAppChangedIDs []string
	var candidateVerify func(context.Context, []packs.Pack) error
	if verifyCandidate && execution.verify != nil {
		candidateVerify = func(ctx context.Context, _ []packs.Pack) error {
			if err := execution.verify(ctx, r, execution.candidate.packList()); err != nil {
				return err
			}
			execution.verified = true
			return nil
		}
	}
	err := prepareAppliedAppPacksForActivation(
		verifyCtx,
		runnerSQLDB(r),
		applied,
		func(ctx context.Context, db *sql.DB, installing []packs.Pack) error {
			return packs.RegisterRuntimeMetadataForInstallingPacksWithOverridesAndComponentKindRegistrationSet(
				ctx,
				db,
				installing,
				runtimeDeclarationOverridesFromPreflight(execution.preflight),
				r.componentKindRegistrationSet(),
			)
		},
		runLayoutMaintenance,
		func(ctx context.Context) error {
			for _, prepared := range execution.prepared {
				if err := materializeSourceBackedPackDerived(ctx, r, prepared.pack, options.DeferFormulaViewRebuild); err != nil {
					return err
				}
				if err := handlerconfig.RegisterHandlerConfigsForPack(ctx, runnerSQLDB(r), connectorcapabilities.Default(), prepared.applied.pack.Name()); err != nil {
					return err
				}
			}
			return nil
		},
		candidateVerify,
	)
	if err != nil {
		return failOfflineCandidatePreparedMaterializations(ctx, r, execution, err)
	}
	for index, appPack := range applied {
		var activatedConnectedAppIDs []string
		err := markInstalledPackActiveWithHook(
			ctx,
			runnerSQLDB(r),
			appPack.pack,
			appPack.payloadHash,
			func(ctx context.Context, tx *sql.Tx) error {
				changed, err := packs.PublishSourceBackedConnectedAppsForPackInTx(ctx, tx, appPack.pack.Name())
				if err != nil {
					return err
				}
				activatedConnectedAppIDs = append(activatedConnectedAppIDs, changed...)
				return nil
			},
		)
		if err != nil {
			return failOfflineCandidatePreparedMaterializationActivation(
				ctx, r, execution, index, appPack.pack.Name(), err,
			)
		}
		connectedAppChangedIDs = append(connectedAppChangedIDs, activatedConnectedAppIDs...)
	}
	execution.prepared = nil
	applyOptions := packs.ApplyOptions{
		RepairLegacyUniqueDuplicates: options.RepairLegacyUniqueDuplicates,
		DeferFormulaViewRebuild:      options.DeferFormulaViewRebuild,
	}
	packs.NotifySourceBackedConnectedAppChanges(r.PackDepsWithOptions(applyOptions), connectedAppChangedIDs)
	if err := reconcileConnectorInboundWorkersAfterPackLifecycle(ctx, r); err != nil {
		return fmt.Errorf("reconcile connector inbound workers after offline candidate activation: %w", err)
	}
	return nil
}

func failOfflineCandidatePreparedMaterializationActivation(
	ctx context.Context,
	r *Runner,
	execution *offlineCandidateExecution,
	failedIndex int,
	packName string,
	activationErr error,
) error {
	prepared := takeOfflineCandidatePreparedMaterializations(execution)
	if errors.Is(activationErr, ErrInstalledPackActivationIndeterminate) {
		return fmt.Errorf("activate offline candidate pack %s without compensation: %w", packName, activationErr)
	}
	wrapped := fmt.Errorf("activate offline candidate pack %s: %w", packName, activationErr)
	if failedIndex < 0 || failedIndex >= len(prepared) {
		return errors.Join(wrapped, fmt.Errorf("offline candidate prepared activation index %d is unavailable", failedIndex))
	}
	return failOfflineCandidatePreparedMaterializationSet(
		ctx, r, prepared[failedIndex:], wrapped,
	)
}

// convergeOfflineCandidateMixedTransitionsUnderApplyLock keeps the existing
// per-pack lifecycle and retry contract while executing changed packs in the
// candidate's dependency order. A successful prefix remains active if a later
// transition fails; the next stopped-candidate attempt replans from that exact
// state and resumes the remaining suffix.
func convergeOfflineCandidateMixedTransitionsUnderApplyLock(
	ctx context.Context,
	r *Runner,
	options *InstallPackOptions,
	applyLock controlPlaneApplyLockScope,
) error {
	if r == nil || options == nil || options.offlineCandidate == nil ||
		!options.offlineCandidate.plan.isMixedTransition() {
		return fmt.Errorf("mixed offline candidate convergence is unavailable")
	}
	if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
		return fmt.Errorf("mixed offline candidate convergence requires the active ApplyAll lock: %w", err)
	}
	execution := options.offlineCandidate
	materializations := make(map[string]offlineCandidateMaterializationTransition, len(execution.plan.materialize))
	for _, transition := range execution.plan.materialize {
		if transition.target.pack == nil {
			return fmt.Errorf("mixed offline candidate materialization target is unavailable")
		}
		name := strings.ToLower(strings.TrimSpace(transition.target.pack.Name()))
		if name == "" {
			return fmt.Errorf("mixed offline candidate materialization has an empty pack name")
		}
		if _, duplicate := materializations[name]; duplicate {
			return fmt.Errorf("mixed offline candidate materialization %s is duplicated", transition.target.pack.Name())
		}
		materializations[name] = transition
	}
	advances := make(map[string]SourceRevisionAdvanceAuthorization, len(execution.plan.sourceRevisionAdvances))
	for _, advance := range execution.plan.sourceRevisionAdvances {
		name := strings.ToLower(strings.TrimSpace(advance.Name))
		if name == "" {
			return fmt.Errorf("mixed offline candidate source advance has an empty pack name")
		}
		if _, duplicate := advances[name]; duplicate {
			return fmt.Errorf("mixed offline candidate source advance %s is duplicated", advance.Name)
		}
		advances[name] = advance
	}

	total := len(materializations) + len(advances)
	completed := 0
	for _, target := range execution.candidate.targets {
		if target.pack == nil {
			return fmt.Errorf("mixed offline candidate target is unavailable")
		}
		name := strings.ToLower(strings.TrimSpace(target.pack.Name()))
		materialization, materializes := materializations[name]
		advance, advancesRevision := advances[name]
		if materializes && advancesRevision {
			return fmt.Errorf("mixed offline candidate pack %s has multiple transition kinds", target.pack.Name())
		}
		if !materializes && !advancesRevision {
			continue
		}
		if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
			return fmt.Errorf("mixed offline candidate convergence lost the ApplyAll lock: %w", err)
		}
		verifyCandidate := completed+1 == total
		if materializes {
			if len(execution.prepared) != 0 {
				return fmt.Errorf("mixed offline candidate retained a prepared transition before %s", target.pack.Name())
			}
			if err := prepareOfflineCandidateMaterializationsUnderApplyLock(ctx, r, options, materialization); err != nil {
				return err
			}
			if err := finalizeOfflineCandidatePreparedMaterializationsWithVerification(
				ctx, r, options, execution, verifyCandidate,
			); err != nil {
				return err
			}
			if err := r.RefreshCache(ctx); err != nil {
				return fmt.Errorf("refresh Runtime after offline candidate materialization %s: %w", target.pack.Name(), err)
			}
		} else {
			singleton := *options
			singleton.ApplyAllSourceRevisionAdvances = []SourceRevisionAdvanceAuthorization{advance}
			singleton.offlineCandidateCompleteSourceRevisionReplays =
				offlineCandidateSourceRevisionReplayForName(
					options.offlineCandidateCompleteSourceRevisionReplays, name,
				)
			singleton.CandidateFinalVerify = nil
			if verifyCandidate && execution.verify != nil {
				singleton.CandidateFinalVerify = requireOfflineCandidateFinalPreActivationVerification(execution)
			}
			if err := applyAllSourceRevisionAdvances(ctx, r, singleton, applyLock); err != nil {
				return fmt.Errorf("apply mixed offline candidate source advance %s: %w", advance.Name, err)
			}
		}
		completed++
	}
	if completed != total {
		return fmt.Errorf("mixed offline candidate converged %d transition(s), plan requires %d", completed, total)
	}
	if execution.verify != nil && !execution.verified {
		return fmt.Errorf("mixed offline candidate completed without final pre-activation verification")
	}
	return nil
}

func offlineCandidateSourceRevisionReplayForName(all map[string]struct{}, name string) map[string]struct{} {
	name = strings.ToLower(strings.TrimSpace(name))
	if _, ok := all[name]; !ok {
		return nil
	}
	return map[string]struct{}{name: {}}
}

func requireOfflineCandidateFinalPreActivationVerification(
	execution *offlineCandidateExecution,
) CandidateFinalVerifier {
	return func(ctx context.Context, runner *Runner, _ []packs.Pack) error {
		if execution == nil || execution.verify == nil {
			return fmt.Errorf("offline candidate final verifier is unavailable")
		}
		if execution.verified {
			return nil
		}
		ready, err := offlineCandidateReadyForPreActivationFinalVerify(ctx, runner, execution)
		if err != nil {
			return err
		}
		if !ready {
			return fmt.Errorf("offline candidate is not ready for final pre-activation verification")
		}
		if err := execution.verify(ctx, runner, execution.candidate.packList()); err != nil {
			return err
		}
		execution.verified = true
		return nil
	}
}

func offlineCandidatePreflightOverrides(candidate resolvedOfflineCandidate) (*installPackPreflightOverrides, error) {
	targets := make(map[string]*packs.SourceBackedPack, len(candidate.targets))
	for _, target := range candidate.targets {
		targets[strings.ToLower(strings.TrimSpace(target.pack.Name()))] = target.pack
	}
	desired, err := newInstallPackPreflightPackContext(targets)
	if err != nil {
		return nil, fmt.Errorf("offline candidate desired context: %w", err)
	}
	runtime, err := newInstallPackPreflightPackContext(targets)
	if err != nil {
		return nil, fmt.Errorf("offline candidate Runtime context: %w", err)
	}
	return &installPackPreflightOverrides{
		plannedDesiredPacks:   desired,
		runtimeEffectivePacks: runtime,
	}, nil
}

func (candidate resolvedOfflineCandidate) packList() []packs.Pack {
	out := make([]packs.Pack, 0, len(candidate.targets))
	for _, target := range candidate.targets {
		out = append(out, target.pack)
	}
	return out
}

func verifyOfflineCandidateFinalState(ctx context.Context, r *Runner, execution *offlineCandidateExecution) error {
	if execution == nil {
		return nil
	}
	rows, err := listInstalledSourceBackedPacks(ctx, runnerSQLDB(r))
	if err != nil {
		return fmt.Errorf("read activated source-backed pack set: %w", err)
	}
	if len(rows) != len(execution.candidate.targets) {
		return fmt.Errorf("activated source-backed pack count is %d, candidate requires %d", len(rows), len(execution.candidate.targets))
	}
	for _, row := range rows {
		target, ok := execution.candidate.byName[strings.ToLower(strings.TrimSpace(row.Name))]
		if !ok {
			return fmt.Errorf("activated source-backed pack %s is absent from the candidate", row.Name)
		}
		if err := verifyOfflineCandidateInstalledTarget(row, target); err != nil {
			return err
		}
	}
	if execution.verify != nil && !execution.verified {
		if err := execution.verify(ctx, r, execution.candidate.packList()); err != nil {
			return fmt.Errorf("candidate final verify: %w", err)
		}
		execution.verified = true
	}
	execution.receipt.Verified = true
	return nil
}

// applyOfflineCandidateSourceRevisionDeltaUnderApplyLock executes only the
// already-planned immutable source-revision transitions. The caller owns the
// same global control-plane apply lock used by ApplyAll. The existing completed
// full-bootstrap marker remains untouched: pack generations have their own
// transactional lifecycle and a failed attempt preserves the previous active
// generation instead of making substrate/schema state incomplete.
func applyOfflineCandidateSourceRevisionDeltaUnderApplyLock(
	ctx context.Context,
	db *database.TiDBConnection,
	options InstallPackOptions,
	runnerOptions RunnerOptions,
	applyLock controlPlaneApplyLockScope,
) error {
	execution := options.offlineCandidate
	if execution == nil || !execution.plan.isSourceRevisionDelta() {
		return fmt.Errorf("offline candidate source-revision delta plan is unavailable")
	}
	if options.DeferFormulaViewRebuild {
		return fmt.Errorf("offline candidate source-revision delta does not support deferred formula-view rebuild")
	}
	// This shortcut preserves the completed full-bootstrap marker and bypasses
	// RunRepair. Converge later stamped metadata before refreshing the runner
	// cache or activating the source-revision candidate.
	if _, err := MigrateMasterDetailRequiredBackfill(ctx, db.DB()); err != nil {
		return fmt.Errorf("converge master-detail required metadata before source-revision delta: %w", err)
	}
	r, err := NewRunnerWithOptions(db, runnerOptions)
	if err != nil {
		return fmt.Errorf("construct source-revision delta runner: %w", err)
	}
	if err := r.RefreshCache(ctx); err != nil {
		return fmt.Errorf("refresh source-revision delta runner: %w", err)
	}
	if err := applyAllSourceRevisionAdvances(ctx, r, options, applyLock); err != nil {
		return fmt.Errorf("apply source-revision component deltas: %w", err)
	}
	// Source revision advances may invoke legacy pack writers after the initial
	// convergence/stamp check. Re-run the guarded repair and refresh the runner
	// before any pre-activation or final-state verification.
	if _, err := MigrateMasterDetailRequiredBackfill(ctx, db.DB()); err != nil {
		return fmt.Errorf("converge master-detail required metadata after source-revision delta: %w", err)
	}
	if err := r.RefreshCache(ctx); err != nil {
		return fmt.Errorf("refresh source-revision delta runner after master-detail convergence: %w", err)
	}
	if err := requireOfflineCandidatePreActivationVerification(execution); err != nil {
		return err
	}
	if err := verifyOfflineCandidateFinalState(ctx, r, execution); err != nil {
		return fmt.Errorf("verify source-revision delta candidate: %w", err)
	}
	if options.ApplyAllFinalVerify != nil {
		if err := options.ApplyAllFinalVerify(ctx, db); err != nil {
			return fmt.Errorf("final distribution verify after source-revision delta: %w", err)
		}
	}
	if err := requireApplyAllSourceRevisionAdvancePlanCompleted(ctx, db.DB(), options); err != nil {
		return fmt.Errorf("source-revision delta final gate: %w", err)
	}
	return nil
}

func verifyOfflineCandidateInstalledTarget(row installedPackRow, target offlineCandidateTarget) error {
	if target.pack == nil {
		return fmt.Errorf("activated candidate pack %s target is unavailable", row.Name)
	}
	meta := packs.PackHashMetadataForPack(target.pack)
	currentKind, currentURI, err := canonicalPackSourceIdentity(row.SourceKind.String, row.SourceURI.String)
	if err != nil {
		return fmt.Errorf("activated candidate pack %s source: %w", row.Name, err)
	}
	targetKind, targetURI, err := canonicalPackSourceIdentity(meta.SourceKind, meta.SourceURI)
	if err != nil {
		return fmt.Errorf("target candidate pack %s source: %w", row.Name, err)
	}
	if row.Status != installedPackStatusActive || row.IsDeleted != 0 ||
		strings.TrimSpace(row.Category) != strings.TrimSpace(string(target.pack.Category())) ||
		strings.TrimSpace(row.AppliedPayloadHash.String) != target.payloadHash ||
		strings.TrimSpace(row.DesiredPayloadHash.String) != target.payloadHash ||
		currentKind != targetKind || currentURI != targetURI ||
		strings.TrimSpace(row.SourceRevision.String) != strings.TrimSpace(meta.SourceRevision) {
		return fmt.Errorf("activated candidate pack %s does not match its exact target", row.Name)
	}
	if err := verifyInstalledPackHashMetadata(row, target.pack); err != nil {
		return fmt.Errorf(
			"activated candidate pack %s compatibility metadata does not match its exact target: %w",
			row.Name,
			err,
		)
	}
	if err := verifyInstalledPackHostRequirementsState(row, target.pack); err != nil {
		return fmt.Errorf(
			"activated candidate pack %s host requirements do not match its exact target: %w",
			row.Name,
			err,
		)
	}
	if err := verifyInstalledPersistedRuntimeDeclarationState(row, target.pack, true); err != nil {
		return fmt.Errorf(
			"activated candidate pack %s runtime declarations do not match its exact target: %w",
			row.Name,
			err,
		)
	}
	return nil
}