package controlplane

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"time"
	"unicode"

	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/persistence"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

const (
	installedPackStatusInstalling = "installing"
	installedPackStatusActive     = "active"
	installedPackStatusRemoving   = "removing"
	installedPackStatusFailed     = "failed"

	installedPackActivationCommitReconciliationTimeout = 30 * time.Second
)

// ErrConnectorPackNamespaceConflict is returned when an install would reassign a
// pack name across the connector-pack/non-connector-pack boundary. Connector
// pack namespace reservation is the SSoT for inbound webhook routing, secret
// resolution, and capability dispatch (NCRM-INT-03/-04A/-04); silent re-categorization would
// hand a connector identity to non-connector code or vice versa.
var ErrConnectorPackNamespaceConflict = errors.New("connector-pack namespace conflict")

// ErrInstalledPackActivationIndeterminate means the active transaction
// returned a commit error and its exact durable InstalledPack tuple could not
// be reread. Callers must leave the installation incarnation untouched and let
// the stopped-upgrade retry path reconcile it; writing Failed could overwrite a
// transaction that actually committed.
var ErrInstalledPackActivationIndeterminate = errors.New("installed pack activation commit outcome is indeterminate")

var errInstalledPackActivationNotCommitted = errors.New("installed pack activation was not committed")

// validateNamespaceReservation rejects installs that cross the connector-pack
// boundary against an existing row. Same-category re-installs and soft-deleted
// rebirths are allowed; the connector-pack vs non-connector-pack frontier is
// the only collision that fails closed.
func validateNamespaceReservation(ctx context.Context, db *sql.DB, pack packs.Pack) error {
	name := strings.TrimSpace(pack.Name())
	if name == "" {
		return fmt.Errorf("validate namespace reservation: empty pack name")
	}
	var (
		existingCategory string
		isDeleted        int
	)
	err := db.QueryRowContext(ctx,
		`SELECT Category, COALESCE(IsDeleted, 0) FROM _System_InstalledPack WHERE Name = ? LIMIT 1`,
		name,
	).Scan(&existingCategory, &isDeleted)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return fmt.Errorf("validate namespace reservation for %s: %w", name, err)
	}
	if isDeleted != 0 {
		return nil
	}
	existingCategory = strings.TrimSpace(existingCategory)
	incoming := string(pack.Category())
	if existingCategory == incoming {
		return nil
	}
	if existingCategory == string(packs.ConnectorPack) || incoming == string(packs.ConnectorPack) {
		return fmt.Errorf("%w: pack %q already installed as %q, refusing to install as %q", ErrConnectorPackNamespaceConflict, name, existingCategory, incoming)
	}
	return nil
}

type installedPackRow struct {
	ID                      string
	Name                    string
	PackInstallID           string
	Category                string
	Status                  string
	IsDeleted               int
	PackVersion             sql.NullString
	ManifestFormatVersion   sql.NullString
	PlatformCompatibility   sql.NullString
	DesiredPayloadHash      sql.NullString
	AppliedPayloadHash      sql.NullString
	PayloadHashAlgorithm    sql.NullString
	SourceKind              sql.NullString
	SourceURI               sql.NullString
	SourceRevision          sql.NullString
	HostRequirementsDigest  sql.NullString
	RuntimeObjectSpecDecl   sql.NullString
	RuntimeRelatedLists     sql.NullString
	RuntimeSearchScope      sql.NullString
	RuntimeSearchAllDefault sql.NullString
	RuntimeFormulaBundles   sql.NullString
	RuntimeStartupShallow   sql.NullString
	HasRuntimeFormula       bool
	HasRuntimeStartupDecl   bool
}

type stoppedSourceSetReplacementAdmission struct {
	packName               string
	activePackInstallID    string
	candidatePackInstallID string
}

// prepareStoppedSourceSetReplacementGenerations admits every active target,
// repairs the legacy InstalledPack compatibility projection from its selected
// immutable generation when an older deploy attempt overwrote it, recovers any
// preparing attempt abandoned by an earlier crashed lock holder, and records a
// fresh preparing generation for each candidate. The production caller holds
// the exclusive control-plane apply lock for this whole transition. Preparing
// a replacement never changes the active projection or activation pointer.
func prepareStoppedSourceSetReplacementGenerations(
	ctx context.Context,
	db *sql.DB,
	targetPacks []packs.Pack,
	applyLock controlPlaneApplyLockScope,
) (map[string]stoppedSourceSetReplacementAdmission, error) {
	if ctx == nil || db == nil {
		return nil, fmt.Errorf("%w: stopped source-set replacement requires context and database", ErrPackGenerationPrecondition)
	}
	if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
		return nil, fmt.Errorf("%w: stopped source-set replacement recovery requires the exclusive apply lock: %v", ErrPackGenerationPrecondition, err)
	}
	requiresInboundDrain := false
	for _, target := range targetPacks {
		if target == nil {
			continue
		}
		if target.Category() == packs.ConnectorPack || target.Category() == packs.AppPack {
			requiresInboundDrain = true
			break
		}
	}
	workItemAuthority := false
	if requiresInboundDrain {
		var err error
		workItemAuthority, err = connectorInboundWorkItemSchemaReady(ctx, db)
		if err != nil {
			return nil, fmt.Errorf("resolve stopped source-set inbound drain authority: %w", err)
		}
	}
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return nil, fmt.Errorf("begin stopped source-set replacement lifecycle admission: %w", err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()

	admissions := make(map[string]stoppedSourceSetReplacementAdmission, len(targetPacks))
	for index, target := range targetPacks {
		if target == nil {
			return nil, fmt.Errorf("%w: stopped source-set replacement target %d is empty", ErrPackGenerationPrecondition, index)
		}
		packName := strings.TrimSpace(target.Name())
		if packName == "" || packName != target.Name() {
			return nil, fmt.Errorf("%w: stopped source-set replacement target %d has no canonical name", ErrPackGenerationPrecondition, index)
		}
		key := strings.ToLower(packName)
		if _, duplicate := admissions[key]; duplicate {
			return nil, fmt.Errorf("%w: stopped source-set replacement target %s is duplicated", ErrPackGenerationPrecondition, packName)
		}

		var (
			installedPackInstallID string
			installedCategory      string
			installedStatus        string
			installedIsDeleted     int
		)
		err := tx.QueryRowContext(ctx, `SELECT
COALESCE(PackInstallId, ''), Category, Status, COALESCE(IsDeleted, 0)
FROM _System_InstalledPack
WHERE Name = ?
LIMIT 1
FOR UPDATE`, packName).Scan(
			&installedPackInstallID,
			&installedCategory,
			&installedStatus,
			&installedIsDeleted,
		)
		if errors.Is(err, sql.ErrNoRows) {
			return nil, fmt.Errorf("%w: stopped source-set pack %s has no existing installation to replace", ErrPackGenerationPrecondition, packName)
		}
		if err != nil {
			return nil, fmt.Errorf("lock stopped source-set pack %s installation: %w", packName, err)
		}
		canonicalInstallID := strings.TrimSpace(installedPackInstallID)
		activeInstallID := canonicalInstallID
		activation, activationExists, err := readPackActivationForUpdate(ctx, tx, packName)
		if err != nil {
			return nil, err
		}
		if !activationExists {
			if installedCategory != string(target.Category()) ||
				installedIsDeleted != 0 ||
				installedStatus != installedPackStatusActive ||
				canonicalInstallID == "" ||
				canonicalInstallID != installedPackInstallID {
				return nil, fmt.Errorf(
					"%w: stopped source-set pack %s legacy installation %q/%q is not active with a canonical identity",
					ErrPackGenerationPrecondition,
					packName,
					canonicalInstallID,
					installedStatus,
				)
			}
		} else {
			active, err := verifyActivePackGeneration(ctx, tx, activation)
			if err != nil {
				return nil, fmt.Errorf("%w: stopped source-set pack %s active generation: %v", ErrPackGenerationPrecondition, packName, err)
			}
			activeInstallID = strings.TrimSpace(activation.activePackInstallID)
			if active.packName != packName || active.category != string(target.Category()) {
				return nil, fmt.Errorf(
					"%w: stopped source-set pack %s active generation category is %q, want %q",
					ErrPackGenerationPrecondition,
					packName,
					active.category,
					target.Category(),
				)
			}
			if installedCategory != active.category ||
				installedIsDeleted != 0 ||
				installedStatus != installedPackStatusActive ||
				canonicalInstallID != activeInstallID {
				if err := restoreInstalledPackCompatibilityProjectionFromActiveGeneration(
					ctx, tx, active, activation,
				); err != nil {
					return nil, err
				}
			}
		}
		installIDsToDrain := []string{activeInstallID}
		if canonicalInstallID != "" && canonicalInstallID != activeInstallID {
			installIDsToDrain = append(installIDsToDrain, canonicalInstallID)
		}
		for _, installID := range installIDsToDrain {
			if err := validateStoppedSourceSetInboundDrain(
				ctx, tx, target, installID, workItemAuthority,
			); err != nil {
				return nil, err
			}
		}
		if err := recoverAbandonedStoppedSourceSetPreparingGenerations(
			ctx,
			tx,
			packName,
		); err != nil {
			return nil, err
		}
		candidateInstallID := utils.GenerateID()
		payloadHash, err := packs.PackPayloadHash(target)
		if err != nil {
			return nil, fmt.Errorf("build stopped source-set pack %s candidate payload hash: %w", packName, err)
		}
		candidate, err := preparingStoppedSourceSetPackGeneration(
			target, payloadHash, candidateInstallID,
		)
		if err != nil {
			return nil, err
		}
		if err := insertPackGeneration(ctx, tx, candidate, time.Now().UTC()); err != nil {
			return nil, err
		}
		admissions[key] = stoppedSourceSetReplacementAdmission{
			packName:               packName,
			activePackInstallID:    activeInstallID,
			candidatePackInstallID: candidateInstallID,
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, fmt.Errorf("commit stopped source-set candidate preparation: %w", err)
	}
	committed = true
	return admissions, nil
}

// recoverAbandonedStoppedSourceSetPreparingGenerations is the restart boundary
// for independently committed candidate preparation. The caller already owns
// the exclusive apply lock, so every older preparing row belongs to a process
// that can no longer complete. It is conservatively terminalized as dirty: the
// crash may have happened after owner-scoped metadata mutation even though the
// immutable activation pointer was never switched. Other generation rows are
// retained history and are deliberately not used to reconstruct or validate
// the selected generation; PackActivation is the sole serving selector.
func recoverAbandonedStoppedSourceSetPreparingGenerations(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	packName string,
) error {
	const recoveryMessage = "abandoned stopped source-set candidate recovered by exclusive retry"
	if _, err := tx.ExecContext(ctx, `UPDATE _System_PackGeneration
SET Status = ?, LastError = ?, LastModifiedDate = ?
WHERE PackName = ?
  AND Status = ?
  AND COALESCE(IsDeleted, 0) = 0`,
		packGenerationFailedDirty,
		recoveryMessage,
		time.Now().UTC(),
		packName,
		packGenerationPreparing,
	); err != nil {
		return fmt.Errorf("recover stopped source-set pack %q abandoned candidates: %w", packName, err)
	}

	return nil
}

func restoreInstalledPackCompatibilityProjectionFromActiveGeneration(
	ctx context.Context,
	tx packGenerationActivationTransaction,
	active sourceBackedPackGeneration,
	activation packActivation,
) error {
	result, err := tx.ExecContext(ctx, `UPDATE _System_InstalledPack
SET PackInstallId = ?, Category = ?, Status = ?, PackVersion = ?,
    ManifestFormatVersion = ?, PlatformCompatibility = ?,
    DesiredPayloadHash = ?, AppliedPayloadHash = ?, PayloadHashAlgorithm = ?,
    SourceKind = ?, SourceURI = ?, SourceRevision = ?,
    HostRequirementsDigest = ?, RuntimeObjectSpecDeclarations = CAST(? AS JSON),
    RuntimeRelatedListRules = CAST(? AS JSON),
    RuntimeSearchScopeOrder = CAST(? AS JSON),
    RuntimeSearchAllDefaultFields = CAST(? AS JSON),
    RuntimeFormulaFunctionBundles = CAST(? AS JSON),
    RuntimeStartupShallowDeclarations = CAST(? AS JSON),
    LastError = NULL, LastAppliedDate = ?, LastModifiedDate = ?, IsDeleted = 0
WHERE Name = ?`,
		active.packInstallID,
		active.category,
		installedPackStatusActive,
		nullableValue(active.packVersion),
		nullableValue(active.manifestFormatVersion),
		nullableValue(active.platformCompatibility),
		nullableValue(active.desiredPayloadHash),
		nullableValue(active.appliedPayloadHash),
		nullableValue(active.payloadHashAlgorithm),
		nullableValue(active.sourceKind),
		nullableValue(active.sourceURI),
		nullableValue(active.sourceRevision),
		nullableValue(active.hostRequirementsDigest),
		nullableValue(active.runtimeObjectSpecDecl),
		nullableValue(active.runtimeRelatedLists),
		nullableValue(active.runtimeSearchScope),
		nullableValue(active.runtimeSearchAllDefault),
		nullableValue(active.runtimeFormulaBundles),
		nullableValue(active.runtimeStartupShallow),
		activation.activatedDate.Time,
		time.Now().UTC(),
		active.packName,
	)
	if err != nil {
		return fmt.Errorf("restore pack %q active compatibility projection: %w", active.packName, err)
	}
	affected, err := result.RowsAffected()
	if err != nil || affected != 1 {
		return fmt.Errorf(
			"%w: pack %q active compatibility projection could not be restored",
			ErrPackGenerationConflict,
			active.packName,
		)
	}
	return nil
}

// validateStoppedSourceSetInboundDrain is the read-only, exact-incarnation
// queue barrier for stopped replacement. The later Installing CAS repeats the
// guard atomically; this earlier barrier keeps wrapper invalidation and
// metadata writes from starting when the prior incarnation is already busy.
func validateStoppedSourceSetInboundDrain(
	ctx context.Context,
	queryer packGenerationActivationTransaction,
	pack packs.Pack,
	packInstallID string,
	workItemAuthority bool,
) error {
	if queryer == nil || pack == nil {
		return fmt.Errorf("%w: stopped source-set inbound drain requires pack and transaction", ErrPackGenerationPrecondition)
	}
	scope := connectorPackInboundReapplyScope
	switch pack.Category() {
	case packs.ConnectorPack:
	case packs.AppPack:
		scope = companionAppPackInboundReapplyScope
	default:
		return nil
	}
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		return fmt.Errorf("%w: stopped source-set pack %s has no canonical inbound identity", ErrPackGenerationPrecondition, pack.Name())
	}
	blocker := connectorInboundExactIncarnationBlockerPredicate(scope, workItemAuthority)
	var blocked int
	if err := queryer.QueryRowContext(ctx, `SELECT CASE WHEN (`+blocker+`) THEN 1 ELSE 0 END
  FROM (SELECT ? AS PackInstallId) p`, packInstallID).Scan(&blocked); err != nil {
		return fmt.Errorf("read stopped source-set pack %s inbound drain: %w", pack.Name(), err)
	}
	if blocked == 0 {
		return nil
	}
	return fmt.Errorf(
		"%w: stopped source-set pack %s still has non-terminal connector inbound rows for install %s; drain the queue before replacement",
		ErrPackGenerationPrecondition,
		pack.Name(),
		packInstallID,
	)
}

func markInstalledPackInstalling(ctx context.Context, db *sql.DB, pack packs.Pack, payloadHash string) error {
	return markInstalledPackInstallingWithPackInstallID(ctx, db, pack, payloadHash, "")
}

// markInstalledPackInstallingWithPackInstallID starts an install incarnation.
// A blank packInstallID mints the normal fresh incarnation. The only non-blank
// caller is an exact repair of an already activated immutable generation or an
// admitted stopped-candidate adoption of the identical active installation:
// both preserve the installation incarnation and must not silently manufacture
// a replacement generation identity.
func markInstalledPackInstallingWithPackInstallID(
	ctx context.Context,
	db *sql.DB,
	pack packs.Pack,
	payloadHash string,
	packInstallID string,
) error {
	return markInstalledPackInstallingWithPolicy(
		ctx, db, pack, payloadHash, packInstallID, false,
	)
}

func markInstalledPackInstallingWithPolicy(
	ctx context.Context,
	db *sql.DB,
	pack packs.Pack,
	payloadHash string,
	packInstallID string,
	allowQueuedInbound bool,
) error {
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		packInstallID = utils.GenerateID()
	}
	if err := validateNamespaceReservation(ctx, db, pack); err != nil {
		return err
	}
	now := time.Now().UTC()
	meta := packs.PackHashMetadataForPack(pack)
	runtimeDecl, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		return fmt.Errorf("build persisted runtime declaration for installing pack %s: %w", pack.Name(), err)
	}
	hostRequirementsDigestColumnPreflighted, hasHostRequirementsDigestColumn, err := preflightInstalledPackHostRequirementsDigestColumn(ctx, db, pack)
	if err != nil {
		return err
	}
	writeHostRequirementsDigest := func() error {
		if hostRequirementsDigestColumnPreflighted {
			return writeInstalledPackHostRequirementsDigest(ctx, db, pack, now, hasHostRequirementsDigestColumn)
		}
		return updateInstalledPackHostRequirementsDigest(ctx, db, pack, now)
	}
	if pack.Category() == packs.ConnectorPack {
		if err := markConnectorPackInstalling(ctx, db, pack, payloadHash, packInstallID, now, meta, allowQueuedInbound); err != nil {
			return err
		}
		return writeHostRequirementsDigest()
	}
	hasRuntimeDeclarationColumns, err := installedPackHasRuntimeDeclarationColumns(ctx, db)
	if err != nil {
		return err
	}
	if !hasRuntimeDeclarationColumns {
		if declaresRuntimeFormulaFunctionBundles(runtimeDecl) {
			return fmt.Errorf(
				"pack %s declares runtime formula function bundles but _System_InstalledPack schema is missing persisted runtime declaration columns; migrate schema before installing this pack",
				pack.Name(),
			)
		}
		updated, err := markExistingNonConnectorPackInstalling(ctx, db, pack, payloadHash, packInstallID, now, meta, nil, allowQueuedInbound)
		if err != nil {
			return err
		}
		if updated {
			return writeHostRequirementsDigest()
		}
		_, err = db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
					(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, InstallSource, LastError, LastAppliedDate, CreatedDate, LastModifiedDate, IsDeleted)
				 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, 'apply', NULL, NULL, ?, ?, 0)
			 `,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusInstalling,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			now, now,
		)
		if err != nil {
			return fmt.Errorf("mark installed pack %s installing: %w", pack.Name(), err)
		}
		return writeHostRequirementsDigest()
	}
	hasRuntimeStartupShallowColumn, err := installedPackHasRuntimeStartupShallowDeclarationColumn(ctx, db)
	if err != nil {
		return err
	}
	hasRuntimeFormulaBundlesColumn, err := installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx, db)
	if err != nil {
		return err
	}
	if !hasRuntimeFormulaBundlesColumn && declaresRuntimeFormulaFunctionBundles(runtimeDecl) {
		return fmt.Errorf(
			"pack %s declares runtime formula function bundles but _System_InstalledPack schema is missing RuntimeFormulaFunctionBundles; migrate schema before installing this pack",
			pack.Name(),
		)
	}
	if !hasRuntimeStartupShallowColumn && !hasRuntimeFormulaBundlesColumn {
		updated, err := markExistingNonConnectorPackInstalling(ctx, db, pack, payloadHash, packInstallID, now, meta, []nonConnectorPackInstallingRuntimeColumn{
			{name: "RuntimeObjectSpecDeclarations", value: runtimeDecl.ObjectPayloadsJSON()},
			{name: "RuntimeRelatedListRules", value: runtimeDecl.RelatedListRulesJSON()},
			{name: "RuntimeSearchScopeOrder", value: runtimeDecl.SearchScopeOrderJSON()},
			{name: "RuntimeSearchAllDefaultFields", value: runtimeDecl.SearchAllDefaultFieldsJSON()},
		}, allowQueuedInbound)
		if err != nil {
			return err
		}
		if updated {
			return writeHostRequirementsDigest()
		}
		_, err = db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
				(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, InstallSource, LastError, LastAppliedDate, CreatedDate, LastModifiedDate, IsDeleted)
			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', NULL, NULL, ?, ?, 0)
			 `,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusInstalling,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(),
			now, now,
		)
		if err != nil {
			return fmt.Errorf("mark installed pack %s installing: %w", pack.Name(), err)
		}
		return writeHostRequirementsDigest()
	}
	if !hasRuntimeStartupShallowColumn {
		updated, err := markExistingNonConnectorPackInstalling(ctx, db, pack, payloadHash, packInstallID, now, meta, []nonConnectorPackInstallingRuntimeColumn{
			{name: "RuntimeObjectSpecDeclarations", value: runtimeDecl.ObjectPayloadsJSON()},
			{name: "RuntimeRelatedListRules", value: runtimeDecl.RelatedListRulesJSON()},
			{name: "RuntimeSearchScopeOrder", value: runtimeDecl.SearchScopeOrderJSON()},
			{name: "RuntimeSearchAllDefaultFields", value: runtimeDecl.SearchAllDefaultFieldsJSON()},
			{name: "RuntimeFormulaFunctionBundles", value: runtimeDecl.FormulaFunctionBundlesJSON()},
		}, allowQueuedInbound)
		if err != nil {
			return err
		}
		if updated {
			return writeHostRequirementsDigest()
		}
		_, err = db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
				(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles, InstallSource, LastError, LastAppliedDate, CreatedDate, LastModifiedDate, IsDeleted)
			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', NULL, NULL, ?, ?, 0)
			 `,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusInstalling,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.FormulaFunctionBundlesJSON(),
			now, now,
		)
		if err != nil {
			return fmt.Errorf("mark installed pack %s installing: %w", pack.Name(), err)
		}
		return writeHostRequirementsDigest()
	}
	if !hasRuntimeFormulaBundlesColumn {
		updated, err := markExistingNonConnectorPackInstalling(ctx, db, pack, payloadHash, packInstallID, now, meta, []nonConnectorPackInstallingRuntimeColumn{
			{name: "RuntimeObjectSpecDeclarations", value: runtimeDecl.ObjectPayloadsJSON()},
			{name: "RuntimeRelatedListRules", value: runtimeDecl.RelatedListRulesJSON()},
			{name: "RuntimeSearchScopeOrder", value: runtimeDecl.SearchScopeOrderJSON()},
			{name: "RuntimeSearchAllDefaultFields", value: runtimeDecl.SearchAllDefaultFieldsJSON()},
			{name: "RuntimeStartupShallowDeclarations", value: runtimeDecl.StartupShallowDeclaredRowsJSON()},
		}, allowQueuedInbound)
		if err != nil {
			return err
		}
		if updated {
			return writeHostRequirementsDigest()
		}
		_, err = db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
			(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations, InstallSource, LastError, LastAppliedDate, CreatedDate, LastModifiedDate, IsDeleted)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', NULL, NULL, ?, ?, 0)
		 `,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusInstalling,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.StartupShallowDeclaredRowsJSON(),
			now, now,
		)
		if err != nil {
			return fmt.Errorf("mark installed pack %s installing: %w", pack.Name(), err)
		}
		return writeHostRequirementsDigest()
	}
	updated, err := markExistingNonConnectorPackInstalling(ctx, db, pack, payloadHash, packInstallID, now, meta, []nonConnectorPackInstallingRuntimeColumn{
		{name: "RuntimeObjectSpecDeclarations", value: runtimeDecl.ObjectPayloadsJSON()},
		{name: "RuntimeRelatedListRules", value: runtimeDecl.RelatedListRulesJSON()},
		{name: "RuntimeSearchScopeOrder", value: runtimeDecl.SearchScopeOrderJSON()},
		{name: "RuntimeSearchAllDefaultFields", value: runtimeDecl.SearchAllDefaultFieldsJSON()},
		{name: "RuntimeFormulaFunctionBundles", value: runtimeDecl.FormulaFunctionBundlesJSON()},
		{name: "RuntimeStartupShallowDeclarations", value: runtimeDecl.StartupShallowDeclaredRowsJSON()},
	}, allowQueuedInbound)
	if err != nil {
		return err
	}
	if updated {
		return writeHostRequirementsDigest()
	}
	_, err = db.ExecContext(ctx,
		`INSERT INTO _System_InstalledPack
			(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles, RuntimeStartupShallowDeclarations, InstallSource, LastError, LastAppliedDate, CreatedDate, LastModifiedDate, IsDeleted)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', NULL, NULL, ?, ?, 0)
		 `,
		installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusInstalling,
		meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		payloadHash, meta.PayloadHashAlgorithm,
		nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
		runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.FormulaFunctionBundlesJSON(), runtimeDecl.StartupShallowDeclaredRowsJSON(),
		now, now,
	)
	if err != nil {
		return fmt.Errorf("mark installed pack %s installing: %w", pack.Name(), err)
	}
	return writeHostRequirementsDigest()
}

func markConnectorPackInstalling(ctx context.Context, db *sql.DB, pack packs.Pack, payloadHash, packInstallID string, now time.Time, meta packs.PackHashMetadata, allowQueuedInbound bool) error {
	workItemAuthority := false
	if !allowQueuedInbound {
		var err error
		workItemAuthority, err = connectorInboundWorkItemSchemaReady(ctx, db)
		if err != nil {
			return fmt.Errorf("resolve connector pack %s inbound reapply guard authority: %w", pack.Name(), err)
		}
	}
	blockerPredicate := "FALSE"
	if !allowQueuedInbound {
		blockerPredicate = connectorInboundReapplyBlockerPredicate(connectorPackInboundReapplyScope, workItemAuthority)
	}
	query := `UPDATE _System_InstalledPack p
		    SET PackInstallId = ?,
		        Category = ?,
		        Status = ?,
		        PackVersion = ?,
		        ManifestFormatVersion = ?,
		        PlatformCompatibility = ?,
		        DesiredPayloadHash = ?,
		        AppliedPayloadHash = NULL,
		        PayloadHashAlgorithm = ?,
		        SourceKind = ?,
		        SourceURI = ?,
		        SourceRevision = ?,
		        InstallSource = 'apply',
		        LastError = NULL,
		        LastAppliedDate = NULL,
		        LastModifiedDate = ?,
		        IsDeleted = 0
		  WHERE p.Name = ?
		    AND p.Category = ?
		    AND NOT ` + blockerPredicate
	args := []any{
		packInstallID, string(pack.Category()), installedPackStatusInstalling,
		meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		payloadHash, meta.PayloadHashAlgorithm,
		nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
		now, pack.Name(), string(packs.ConnectorPack),
	}
	result, err := db.ExecContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("mark connector pack %s installing: %w", pack.Name(), err)
	}
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("mark connector pack %s installing rows affected: %w", pack.Name(), err)
	}
	if rowsAffected > 0 {
		return nil
	}
	if !allowQueuedInbound {
		if err := rejectConnectorPackReapplyWithNonTerminalInboundRows(ctx, db, pack.Name(), workItemAuthority); err != nil {
			return err
		}
	}

	_, err = db.ExecContext(ctx,
		`INSERT INTO _System_InstalledPack
			(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, InstallSource, LastError, LastAppliedDate, CreatedDate, LastModifiedDate, IsDeleted)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, 'apply', NULL, NULL, ?, ?, 0)`,
		installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusInstalling,
		meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		payloadHash, meta.PayloadHashAlgorithm,
		nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
		now, now,
	)
	if err != nil {
		return fmt.Errorf("insert connector pack %s installing: %w", pack.Name(), err)
	}
	return nil
}

type nonConnectorPackInstallingRuntimeColumn struct {
	name  string
	value string
}

func markExistingNonConnectorPackInstalling(ctx context.Context, db *sql.DB, pack packs.Pack, payloadHash, packInstallID string, now time.Time, meta packs.PackHashMetadata, runtimeColumns []nonConnectorPackInstallingRuntimeColumn, allowQueuedInbound bool) (bool, error) {
	workItemAuthority := false
	blockerPredicate := ""
	if pack.Category() == packs.AppPack && !allowQueuedInbound {
		var err error
		workItemAuthority, err = connectorInboundWorkItemSchemaReady(ctx, db)
		if err != nil {
			return false, fmt.Errorf("resolve app pack %s companion inbound reapply guard authority: %w", pack.Name(), err)
		}
		blockerPredicate = connectorInboundReapplyBlockerPredicate(companionAppPackInboundReapplyScope, workItemAuthority)
	}
	assignments := []string{
		"PackInstallId = ?",
		"Category = ?",
		"Status = ?",
		"PackVersion = ?",
		"ManifestFormatVersion = ?",
		"PlatformCompatibility = ?",
		"DesiredPayloadHash = ?",
		"PayloadHashAlgorithm = ?",
		"SourceKind = ?",
		"SourceURI = ?",
		"SourceRevision = ?",
	}
	args := []interface{}{
		packInstallID,
		string(pack.Category()),
		installedPackStatusInstalling,
		meta.PackVersion,
		meta.ManifestFormatVersion,
		nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		payloadHash,
		meta.PayloadHashAlgorithm,
		nullString(meta.SourceKind, meta.SourceKindValid),
		nullString(meta.SourceURI, meta.SourceURIValid),
		nullString(meta.SourceRevision, meta.SourceRevisionValid),
	}
	for _, column := range runtimeColumns {
		assignments = append(assignments, column.name+" = ?")
		args = append(args, column.value)
	}
	assignments = append(assignments,
		"InstallSource = 'apply'",
		"LastError = NULL",
		"LastModifiedDate = ?",
		"IsDeleted = 0",
	)
	args = append(args, now, pack.Name())
	query := `UPDATE _System_InstalledPack p
   SET ` + strings.Join(assignments, ",\n       ") + `
 WHERE p.Name = ?`
	if pack.Category() == packs.AppPack && !allowQueuedInbound {
		query += `
   AND NOT (
       p.Category = ?
       AND COALESCE(p.IsDeleted, 0) = 0
       AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
       AND ` + blockerPredicate + `
   )`
		args = append(args, string(packs.AppPack))
	}
	result, err := db.ExecContext(ctx, query, args...)
	if err != nil {
		return false, fmt.Errorf("mark non-connector pack %s installing: %w", pack.Name(), err)
	}
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return false, fmt.Errorf("mark non-connector pack %s installing rows affected: %w", pack.Name(), err)
	}
	if rowsAffected > 0 {
		return true, nil
	}
	if pack.Category() != packs.AppPack {
		return false, nil
	}
	if !allowQueuedInbound {
		if err := rejectCompanionAppPackReapplyWithNonTerminalInboundRows(ctx, db, pack.Name(), workItemAuthority); err != nil {
			return false, err
		}
	}
	exists, err := installedPackRowExists(ctx, db, pack.Name())
	if err != nil {
		return false, err
	}
	if exists {
		return false, fmt.Errorf("mark non-connector pack %s installing affected 0 rows without a companion inbound drain guard match", pack.Name())
	}
	return false, nil
}

func installedPackRowExists(ctx context.Context, db *sql.DB, packName string) (bool, error) {
	var id string
	err := db.QueryRowContext(ctx, `SELECT Id FROM _System_InstalledPack WHERE Name = ? LIMIT 1`, packName).Scan(&id)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("lookup installed pack %s row existence: %w", packName, err)
	}
	return true, nil
}

func rejectConnectorPackReapplyWithNonTerminalInboundRows(ctx context.Context, db *sql.DB, packName string, workItemAuthority bool) error {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return nil
	}
	var packInstallID string
	blockerPredicate := connectorInboundReapplyBlockerPredicate(connectorPackInboundReapplyScope, workItemAuthority)
	err := db.QueryRowContext(ctx, `
SELECT COALESCE(p.PackInstallId, '')
  FROM _System_InstalledPack p
 WHERE p.Name = ?
   AND p.Category = ?
   AND p.Status IN (?, ?, ?, ?)
   AND COALESCE(p.IsDeleted, 0) = 0
   AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
   AND `+blockerPredicate+`
 LIMIT 1`, packName, string(packs.ConnectorPack), installedPackStatusActive, installedPackStatusInstalling, installedPackStatusRemoving, installedPackStatusFailed).Scan(&packInstallID)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return fmt.Errorf("lookup connector pack %s queued reapply guard: %w", packName, err)
	}
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		return nil
	}
	return fmt.Errorf("connector pack %s still has non-terminal inbound rows for install %s; drain the queue before reapply", packName, packInstallID)
}

func rejectCompanionAppPackReapplyWithNonTerminalInboundRows(ctx context.Context, db *sql.DB, packName string, workItemAuthority bool) error {
	packName = strings.TrimSpace(packName)
	if packName == "" {
		return nil
	}
	var packInstallID string
	blockerPredicate := connectorInboundReapplyBlockerPredicate(companionAppPackInboundReapplyScope, workItemAuthority)
	err := db.QueryRowContext(ctx, `
SELECT COALESCE(p.PackInstallId, '')
  FROM _System_InstalledPack p
 WHERE p.Name = ?
   AND p.Category = ?
   AND p.Status IN (?, ?, ?, ?)
   AND COALESCE(p.IsDeleted, 0) = 0
   AND TRIM(COALESCE(p.PackInstallId, '')) <> ''
   AND `+blockerPredicate+`
 LIMIT 1`, packName, string(packs.AppPack), installedPackStatusActive, installedPackStatusInstalling, installedPackStatusRemoving, installedPackStatusFailed).Scan(&packInstallID)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return fmt.Errorf("lookup app pack %s companion inbound reapply guard: %w", packName, err)
	}
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		return nil
	}
	return fmt.Errorf("app pack %s still has non-terminal connector inbound rows referencing companion install %s; drain the queue before reapply", packName, packInstallID)
}

func markInstalledPackActive(ctx context.Context, db *sql.DB, pack packs.Pack, payloadHash string) error {
	return markInstalledPackActiveWithHook(ctx, db, pack, payloadHash, nil)
}

func markInstalledPackActiveWithHook(ctx context.Context, db *sql.DB, pack packs.Pack, payloadHash string, beforeCommit func(context.Context, *sql.Tx) error) error {
	return markInstalledPackActiveWithGenerationCommitter(
		ctx, db, pack, payloadHash, beforeCommit, nil, nil,
	)
}

type sourceBackedGenerationCommitter func(
	context.Context,
	packGenerationActivationTransaction,
	appliedAppPack,
) error

type stoppedSourceSetProjectionTransition struct {
	previousPackInstallID  string
	candidatePackInstallID string
}

func markInstalledPackActiveForStoppedSourceSet(
	ctx context.Context,
	db *sql.DB,
	pack packs.Pack,
	payloadHash string,
	previousPackInstallID string,
	candidatePackInstallID string,
	beforeCommit func(context.Context, *sql.Tx) error,
) error {
	previousPackInstallID = strings.TrimSpace(previousPackInstallID)
	candidatePackInstallID = strings.TrimSpace(candidatePackInstallID)
	if previousPackInstallID == "" || candidatePackInstallID == "" || previousPackInstallID == candidatePackInstallID {
		return fmt.Errorf("%w: stopped source-set pack %s has invalid previous/candidate installation identity", ErrPackGenerationPrecondition, pack.Name())
	}
	return markInstalledPackActiveWithGenerationCommitter(
		ctx,
		db,
		pack,
		payloadHash,
		beforeCommit,
		&stoppedSourceSetProjectionTransition{
			previousPackInstallID:  previousPackInstallID,
			candidatePackInstallID: candidatePackInstallID,
		},
		func(
			ctx context.Context,
			tx packGenerationActivationTransaction,
			applied appliedAppPack,
		) error {
			return commitStoppedSourceSetPackGeneration(
				ctx,
				tx,
				applied,
				previousPackInstallID,
				candidatePackInstallID,
			)
		},
	)
}

func markInstalledPackActiveWithGenerationCommitter(
	ctx context.Context,
	db *sql.DB,
	pack packs.Pack,
	payloadHash string,
	beforeCommit func(context.Context, *sql.Tx) error,
	stoppedTransition *stoppedSourceSetProjectionTransition,
	generationCommitter sourceBackedGenerationCommitter,
) error {
	now := time.Now().UTC()
	meta := packs.PackHashMetadataForPack(pack)
	runtimeDecl, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		return fmt.Errorf("build persisted runtime declaration for active pack %s: %w", pack.Name(), err)
	}
	// These two schema-capability checks are read-only and run on db before the
	// transaction begins so they don't hold a tx open during the INFORMATION_SCHEMA
	// queries (which are not transactional anyway on TiDB/MySQL).
	hasRuntimeDeclarationColumns, err := installedPackHasRuntimeDeclarationColumns(ctx, db)
	if err != nil {
		return err
	}
	hasRuntimeStartupShallowColumn := false
	hasRuntimeFormulaBundlesColumn := false
	if hasRuntimeDeclarationColumns {
		hasRuntimeStartupShallowColumn, err = installedPackHasRuntimeStartupShallowDeclarationColumn(ctx, db)
		if err != nil {
			return err
		}
		hasRuntimeFormulaBundlesColumn, err = installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx, db)
		if err != nil {
			return err
		}
		if !hasRuntimeFormulaBundlesColumn && declaresRuntimeFormulaFunctionBundles(runtimeDecl) {
			return fmt.Errorf(
				"pack %s declares runtime formula function bundles but _System_InstalledPack schema is missing RuntimeFormulaFunctionBundles; migrate schema before activating this pack",
				pack.Name(),
			)
		}
	} else if declaresRuntimeFormulaFunctionBundles(runtimeDecl) {
		return fmt.Errorf(
			"pack %s declares runtime formula function bundles but _System_InstalledPack schema is missing persisted runtime declaration columns; migrate schema before activating this pack",
			pack.Name(),
		)
	}
	hostRequirementsDigestColumnPreflighted, hasHostRequirementsDigestColumn, err := preflightInstalledPackHostRequirementsDigestColumn(ctx, db, pack)
	if err != nil {
		return err
	}

	// Open one transaction for every fact that makes the pack consumable. A
	// revision-addressed publication pack must not become active without its
	// immutable generation and activation authority in the same commit.
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("mark installed pack %s active: begin tx: %w", pack.Name(), err)
	}
	committed := false
	defer func() {
		if !committed {
			_ = tx.Rollback()
		}
	}()
	if stoppedTransition != nil {
		result, transitionErr := tx.ExecContext(ctx, `UPDATE _System_InstalledPack
SET PackInstallId = ?
WHERE Name = ?
  AND PackInstallId = ?
  AND Status = ?
  AND COALESCE(IsDeleted, 0) = 0`,
			stoppedTransition.candidatePackInstallID,
			pack.Name(),
			stoppedTransition.previousPackInstallID,
			installedPackStatusActive,
		)
		if transitionErr != nil {
			return fmt.Errorf("select stopped source-set pack %s candidate projection: %w", pack.Name(), transitionErr)
		}
		affected, transitionErr := result.RowsAffected()
		if transitionErr != nil || affected != 1 {
			return fmt.Errorf(
				"%w: stopped source-set pack %s active projection changed before activation",
				ErrPackGenerationConflict,
				pack.Name(),
			)
		}
	}

	if !hasRuntimeDeclarationColumns {
		_, err = tx.ExecContext(ctx,
			`UPDATE _System_InstalledPack
			    SET Status = ?,
			        PackVersion = ?,
			        ManifestFormatVersion = ?,
			        PlatformCompatibility = ?,
			        AppliedPayloadHash = ?,
			        DesiredPayloadHash = ?,
			        PayloadHashAlgorithm = ?,
			        SourceKind = ?,
			        SourceURI = ?,
			        SourceRevision = ?,
			        LastAppliedDate = ?,
			        LastError = NULL,
			        LastModifiedDate = ?,
			        IsDeleted = 0
			  WHERE Name = ?`,
			installedPackStatusActive,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			now, now, pack.Name(),
		)
	} else if !hasRuntimeStartupShallowColumn && !hasRuntimeFormulaBundlesColumn {
		_, err = tx.ExecContext(ctx,
			`UPDATE _System_InstalledPack
			    SET Status = ?,
			        PackVersion = ?,
			        ManifestFormatVersion = ?,
			        PlatformCompatibility = ?,
			        AppliedPayloadHash = ?,
			        DesiredPayloadHash = ?,
			        PayloadHashAlgorithm = ?,
			        SourceKind = ?,
			        SourceURI = ?,
			        SourceRevision = ?,
			        RuntimeObjectSpecDeclarations = ?,
			        RuntimeRelatedListRules = ?,
			        RuntimeSearchScopeOrder = ?,
			        RuntimeSearchAllDefaultFields = ?,
			        LastAppliedDate = ?,
			        LastError = NULL,
			        LastModifiedDate = ?,
			        IsDeleted = 0
			  WHERE Name = ?`,
			installedPackStatusActive,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(),
			now, now, pack.Name(),
		)
	} else if !hasRuntimeStartupShallowColumn {
		_, err = tx.ExecContext(ctx,
			`UPDATE _System_InstalledPack
			    SET Status = ?,
			        PackVersion = ?,
			        ManifestFormatVersion = ?,
			        PlatformCompatibility = ?,
			        AppliedPayloadHash = ?,
			        DesiredPayloadHash = ?,
			        PayloadHashAlgorithm = ?,
			        SourceKind = ?,
			        SourceURI = ?,
			        SourceRevision = ?,
			        RuntimeObjectSpecDeclarations = ?,
			        RuntimeRelatedListRules = ?,
			        RuntimeSearchScopeOrder = ?,
			        RuntimeSearchAllDefaultFields = ?,
			        RuntimeFormulaFunctionBundles = ?,
			        LastAppliedDate = ?,
			        LastError = NULL,
			        LastModifiedDate = ?,
			        IsDeleted = 0
			  WHERE Name = ?`,
			installedPackStatusActive,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.FormulaFunctionBundlesJSON(),
			now, now, pack.Name(),
		)
	} else if !hasRuntimeFormulaBundlesColumn {
		_, err = tx.ExecContext(ctx,
			`UPDATE _System_InstalledPack
			    SET Status = ?,
			        PackVersion = ?,
			        ManifestFormatVersion = ?,
			        PlatformCompatibility = ?,
			        AppliedPayloadHash = ?,
			        DesiredPayloadHash = ?,
			        PayloadHashAlgorithm = ?,
			        SourceKind = ?,
			        SourceURI = ?,
			        SourceRevision = ?,
			        RuntimeObjectSpecDeclarations = ?,
			        RuntimeRelatedListRules = ?,
			        RuntimeSearchScopeOrder = ?,
			        RuntimeSearchAllDefaultFields = ?,
			        RuntimeStartupShallowDeclarations = ?,
			        LastAppliedDate = ?,
			        LastError = NULL,
			        LastModifiedDate = ?,
			        IsDeleted = 0
			  WHERE Name = ?`,
			installedPackStatusActive,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.StartupShallowDeclaredRowsJSON(),
			now, now, pack.Name(),
		)
	} else {
		_, err = tx.ExecContext(ctx,
			`UPDATE _System_InstalledPack
			    SET Status = ?,
			        PackVersion = ?,
			        ManifestFormatVersion = ?,
			        PlatformCompatibility = ?,
			        AppliedPayloadHash = ?,
			        DesiredPayloadHash = ?,
			        PayloadHashAlgorithm = ?,
			        SourceKind = ?,
			        SourceURI = ?,
			        SourceRevision = ?,
			        RuntimeObjectSpecDeclarations = ?,
			        RuntimeRelatedListRules = ?,
			        RuntimeSearchScopeOrder = ?,
			        RuntimeSearchAllDefaultFields = ?,
			        RuntimeFormulaFunctionBundles = ?,
			        RuntimeStartupShallowDeclarations = ?,
			        LastAppliedDate = ?,
			        LastError = NULL,
			        LastModifiedDate = ?,
			        IsDeleted = 0
			  WHERE Name = ?`,
			installedPackStatusActive,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.FormulaFunctionBundlesJSON(), runtimeDecl.StartupShallowDeclaredRowsJSON(),
			now, now, pack.Name(),
		)
	}
	if err != nil {
		return fmt.Errorf("mark installed pack %s active: %w", pack.Name(), err)
	}
	if hostRequirementsDigestColumnPreflighted {
		err = writeInstalledPackHostRequirementsDigest(ctx, tx, pack, now, hasHostRequirementsDigestColumn)
	} else {
		err = updateInstalledPackHostRequirementsDigest(ctx, tx, pack, now)
	}
	if err != nil {
		return err
	}
	if err = upsertPackDependencyRows(ctx, tx, pack, now); err != nil {
		return err
	}
	if err = refreshConnectorInboundProjectionSnapshotsAfterPackActivated(ctx, tx, pack, now); err != nil {
		return err
	}
	if pack.Category() == packs.ConnectorPack {
		if err = ensureEmptyConnectorProjectionProtocolTx(ctx, tx, pack.Name()); err != nil {
			return err
		}
	}
	if _, sourceBacked := pack.(*packs.SourceBackedPack); sourceBacked {
		applied := appliedAppPack{pack: pack, payloadHash: payloadHash}
		if generationCommitter != nil {
			if err := generationCommitter(ctx, tx, applied); err != nil {
				return fmt.Errorf("publish stopped source-set pack generation: %w", err)
			}
		} else {
			publishGeneration, publishErr := packPublishesImmutableGeneration(pack)
			if publishErr != nil {
				return publishErr
			}
			if publishGeneration {
				if err := commitFirstSourceBackedPackGeneration(ctx, tx, applied); err != nil {
					return fmt.Errorf("publish source-backed pack generation: %w", err)
				}
			} else if err := verifyMutableSourceHasNoGenerationAuthority(
				ctx,
				tx,
				pack.Name(),
			); err != nil {
				return err
			}
		}
	}
	if beforeCommit != nil {
		if err := beforeCommit(ctx, tx); err != nil {
			return err
		}
	}
	if err = tx.Commit(); err != nil {
		commitErr := err
		if generationCommitter != nil {
			return fmt.Errorf(
				"%w for %s: stopped source-set activation commit returned an unknown outcome: %v",
				ErrInstalledPackActivationIndeterminate,
				pack.Name(),
				commitErr,
			)
		}
		reconcileCtx, cancel := context.WithTimeout(
			context.WithoutCancel(ctx),
			installedPackActivationCommitReconciliationTimeout,
		)
		reconcileErr := reconcileInstalledPackActivationCommit(
			reconcileCtx,
			db,
			pack,
			payloadHash,
		)
		cancel()
		if reconcileErr == nil {
			committed = true
			return nil
		}
		if errors.Is(reconcileErr, errInstalledPackActivationNotCommitted) {
			return fmt.Errorf("mark installed pack %s active: commit: %w", pack.Name(), commitErr)
		}
		return fmt.Errorf(
			"%w for %s: %w",
			ErrInstalledPackActivationIndeterminate,
			pack.Name(),
			errors.Join(commitErr, reconcileErr),
		)
	}
	committed = true
	return nil
}

type installedPackActivationCommitReader func(
	context.Context,
	*sql.DB,
	string,
) (installedPackRow, bool, error)

type installedPackActivationCommitReconciliationOperations struct {
	readInstalledRow             installedPackActivationCommitReader
	verifySourceBackedGeneration func(context.Context, *sql.DB, appliedAppPack) error
	verifyNoMutableGeneration    func(context.Context, *sql.DB, string) error
}

func reconcileInstalledPackActivationCommit(
	ctx context.Context,
	db *sql.DB,
	pack packs.Pack,
	payloadHash string,
) error {
	return reconcileInstalledPackActivationCommitWith(
		ctx,
		db,
		pack,
		payloadHash,
		installedPackActivationCommitReconciliationOperations{
			readInstalledRow:             getInstalledPackRowByNameForSourceRevisionAdvance,
			verifySourceBackedGeneration: verifyPublishedSourceBackedPackGeneration,
			verifyNoMutableGeneration:    verifyMutableSourceHasNoPublishedGenerationAuthority,
		},
	)
}

func reconcileInstalledPackActivationCommitWith(
	ctx context.Context,
	db *sql.DB,
	pack packs.Pack,
	payloadHash string,
	operations installedPackActivationCommitReconciliationOperations,
) error {
	if ctx == nil || db == nil || pack == nil || operations.readInstalledRow == nil {
		return fmt.Errorf("installed pack activation reconciliation dependencies are incomplete")
	}
	row, exists, err := operations.readInstalledRow(ctx, db, pack.Name())
	if err != nil {
		return fmt.Errorf("reread installed pack activation: %w", err)
	}
	if !exists {
		return fmt.Errorf("%w: pack %q is absent", errInstalledPackActivationNotCommitted, pack.Name())
	}
	if strings.TrimSpace(row.Name) != strings.TrimSpace(pack.Name()) ||
		strings.TrimSpace(row.PackInstallID) == "" ||
		strings.TrimSpace(row.Category) != strings.TrimSpace(string(pack.Category())) ||
		strings.TrimSpace(row.Status) != installedPackStatusActive ||
		row.IsDeleted != 0 {
		return fmt.Errorf(
			"%w: pack %q durable lifecycle tuple is status=%q category=%q install=%q deleted=%d",
			errInstalledPackActivationNotCommitted,
			pack.Name(),
			row.Status,
			row.Category,
			row.PackInstallID,
			row.IsDeleted,
		)
	}
	if err := verifyInstalledPackHashState(row, pack); err != nil {
		return fmt.Errorf("%w: pack %q durable hash tuple: %v", errInstalledPackActivationNotCommitted, pack.Name(), err)
	}
	if strings.TrimSpace(row.AppliedPayloadHash.String) != strings.TrimSpace(payloadHash) ||
		strings.TrimSpace(row.DesiredPayloadHash.String) != strings.TrimSpace(payloadHash) {
		return fmt.Errorf(
			"%w: pack %q durable payload hash does not equal the activating payload",
			errInstalledPackActivationNotCommitted,
			pack.Name(),
		)
	}
	meta := packs.PackHashMetadataForPack(pack)
	if !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: pack %q durable source identity differs from the activating source",
			errInstalledPackActivationNotCommitted,
			pack.Name(),
		)
	}
	if err := verifyInstalledPackHostRequirementsState(row, pack); err != nil {
		return fmt.Errorf("%w: pack %q durable host requirements: %v", errInstalledPackActivationNotCommitted, pack.Name(), err)
	}
	if _, sourceBacked := pack.(*packs.SourceBackedPack); sourceBacked {
		publishGeneration, publishErr := packPublishesImmutableGeneration(pack)
		if publishErr != nil {
			return publishErr
		}
		if publishGeneration {
			if operations.verifySourceBackedGeneration == nil {
				return fmt.Errorf("source-backed generation reconciliation verifier is required")
			}
			if err := operations.verifySourceBackedGeneration(
				ctx,
				db,
				appliedAppPack{pack: pack, payloadHash: payloadHash},
			); err != nil {
				return fmt.Errorf(
					"reread source-backed generation activation for pack %q: %w",
					pack.Name(),
					err,
				)
			}
		} else if operations.verifyNoMutableGeneration == nil {
			return fmt.Errorf("mutable source generation absence reconciliation verifier is required")
		} else if err := operations.verifyNoMutableGeneration(ctx, db, pack.Name()); err != nil {
			return fmt.Errorf(
				"reread mutable source generation absence for pack %q: %w",
				pack.Name(),
				err,
			)
		}
	}
	return nil
}

func installedPackCommitNullStringEquals(got sql.NullString, want string, wantValid bool) bool {
	gotValue := strings.TrimSpace(got.String)
	want = strings.TrimSpace(want)
	if wantValid {
		return got.Valid && gotValue == want
	}
	return !got.Valid || gotValue == ""
}

// execQuerier is satisfied by both *sql.DB and *sql.Tx so installed-pack
// lifecycle helpers can run inside caller-owned transactions without requiring
// an explicit tx parameter type.
type execQuerier interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
	QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
}

// upsertPackDependencyRows reconciles _System_InstalledPackDependency rows for
// pack on every (re)apply:
//   - UPSERTs every currently-declared dependency (ON DUPLICATE KEY UPDATE, IsDeleted=0).
//   - Soft-deletes (IsDeleted=1) every previously-declared dependency that is no longer
//     present in the current manifest so that enqueue consent does not keep authorizing
//     stale cross-pack routing after a warm reapply that dropped a dep.
//   - When the manifest declares zero deps, soft-deletes all existing non-deleted rows for
//     this DependentPackName (pack dropped all deps).
//
// Not a PackDependencyProvider → no-op (connector-packs and legacy packs that never
// declare dependencies).
func upsertPackDependencyRows(ctx context.Context, db execQuerier, pack packs.Pack, now time.Time) error {
	dp, ok := pack.(packs.PackDependencyProvider)
	if !ok {
		// Not a source-backed pack or no dependency accessor — nothing to write.
		return nil
	}
	deps := dp.PackDependencies()

	// Read the PackInstallId that was just set by markInstalledPackActive.
	// Needed even when len(deps)==0 because we still need to update LastModifiedDate
	// correctly; but we skip the read when there is no pack row yet (early-return above
	// already handles non-PackDependencyProvider packs — here we always proceed).
	var packInstallID string
	if err := db.QueryRowContext(ctx,
		`SELECT COALESCE(PackInstallId, '') FROM _System_InstalledPack WHERE Name = ? LIMIT 1`,
		pack.Name(),
	).Scan(&packInstallID); err != nil {
		return fmt.Errorf("upsert pack dependency rows for %s: read PackInstallId: %w", pack.Name(), err)
	}
	packInstallID = strings.TrimSpace(packInstallID)
	if packInstallID == "" {
		return fmt.Errorf("upsert pack dependency rows for %s: PackInstallId is blank after markInstalledPackActive", pack.Name())
	}

	// Build the set of currently-declared dep names for the soft-delete reconcile step.
	currentDepNames := make(map[string]struct{}, len(deps))

	// UPSERT all currently-declared deps.
	for _, dep := range deps {
		depName := strings.TrimSpace(dep.Name)
		if depName == "" {
			continue
		}
		currentDepNames[depName] = struct{}{}
		rowID := installedPackDepID(pack.Name(), depName)
		vc := strings.TrimSpace(dep.VersionConstraint)
		var vcArg interface{}
		if vc == "" {
			vcArg = nil
		} else {
			vcArg = vc
		}
		if _, err := db.ExecContext(ctx, `
INSERT INTO _System_InstalledPackDependency
  (Id, DependentPackName, DependentPackInstallId, DependsOnPackName, VersionConstraint, IsDeleted, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, 0, ?, ?)
ON DUPLICATE KEY UPDATE
  DependentPackInstallId = VALUES(DependentPackInstallId),
  VersionConstraint      = VALUES(VersionConstraint),
  IsDeleted              = 0,
  LastModifiedDate       = VALUES(LastModifiedDate)`,
			rowID, pack.Name(), packInstallID, depName, vcArg, now, now,
		); err != nil {
			return fmt.Errorf("upsert pack dependency row %s→%s: %w", pack.Name(), depName, err)
		}
	}

	// Soft-delete rows for deps that were previously declared but are no longer in the manifest.
	// When len(deps)==0 this soft-deletes ALL existing non-deleted rows for this DependentPackName.
	if len(currentDepNames) == 0 {
		// No current deps — soft-delete everything for this pack.
		if _, err := db.ExecContext(ctx, `
UPDATE _System_InstalledPackDependency
   SET IsDeleted = 1, LastModifiedDate = ?
 WHERE DependentPackName = ?
   AND COALESCE(IsDeleted, 0) = 0`,
			now, pack.Name(),
		); err != nil {
			return fmt.Errorf("soft-delete all dep rows for %s: %w", pack.Name(), err)
		}
	} else {
		// Soft-delete rows whose DependsOnPackName is NOT in currentDepNames.
		// Build parameterized NOT IN placeholders.
		notInNames := make([]interface{}, 0, len(currentDepNames)+1)
		notInNames = append(notInNames, now, pack.Name())
		placeholders := make([]string, 0, len(currentDepNames))
		for name := range currentDepNames {
			placeholders = append(placeholders, "?")
			notInNames = append(notInNames, name)
		}
		query := `UPDATE _System_InstalledPackDependency
   SET IsDeleted = 1, LastModifiedDate = ?
 WHERE DependentPackName = ?
   AND COALESCE(IsDeleted, 0) = 0
   AND DependsOnPackName NOT IN (` + strings.Join(placeholders, ", ") + `)`
		if _, err := db.ExecContext(ctx, query, notInNames...); err != nil {
			return fmt.Errorf("soft-delete stale dep rows for %s: %w", pack.Name(), err)
		}
	}
	return nil
}

// installedPackDepID returns a deterministic row Id for a (dependentPackName, dependsOnPackName)
// pair so that reapply is an idempotent UPSERT.
func installedPackDepID(dependentPackName, dependsOnPackName string) string {
	// Collision-resistant, width-bounded surrogate PK over the EXACT raw
	// (dependent, dependsOn) pair. A lossy lowercase/alphanumeric encoding would
	// alias distinct valid pack pairs (e.g. "a-b"->"dep" and "a_b"->"dep" both
	// collapse to the same Id); because the UPSERT refreshes only install/version
	// fields and never rewrites the name columns, a later apply would then refresh
	// the wrong dependency row and authorize/deny the wrong cross-pack routes. A
	// hash over the raw names with a NUL separator (which cannot appear in a pack
	// name) removes both the char-collapse and name-boundary ambiguity. The Id is
	// derived from the same trimmed values written to the natural unique key
	// (DependentPackName, DependsOnPackName) so it stays consistent on reapply.
	sum := sha256.Sum256([]byte(strings.TrimSpace(dependentPackName) + "\x00" + strings.TrimSpace(dependsOnPackName)))
	return "ipd_" + hex.EncodeToString(sum[:])
}

func markInstalledPackFailed(ctx context.Context, db *sql.DB, pack packs.Pack, payloadHash string, applyErr error) error {
	if applyErr == nil {
		return nil
	}
	if err := validateNamespaceReservation(ctx, db, pack); err != nil {
		// Matches this function's existing semantic of swallowing the INSERT
		// error: namespace-conflict skips overwriting a colliding row with a
		// Failed marker; transient SELECT errors are silently dropped here just
		// like a transient INSERT error below.
		return nil
	}

	// Read the existing PackInstallId so we preserve the incarnation identity
	// that was minted by markInstalledPackInstalling at the start of this
	// install attempt. Overwriting it here would orphan any per-incarnation
	// state (e.g. connector event queue rows from NCRM-INT-05 / PR #1498)
	// that were keyed on the installing-phase UUID.
	//
	// If no prior row exists (first write for this pack ever), mint fresh so
	// the NOT NULL constraint is satisfied.
	var packInstallID string
	scanErr := db.QueryRowContext(ctx,
		`SELECT COALESCE(PackInstallId, '') FROM _System_InstalledPack WHERE Id = ? LIMIT 1`,
		installedPackID(pack.Name())).Scan(&packInstallID)
	if scanErr != nil && scanErr != sql.ErrNoRows {
		// Transient read error — fall through with a fresh UUID rather than
		// silently dropping the failed marker entirely.
		packInstallID = ""
	}
	if strings.TrimSpace(packInstallID) == "" {
		// No prior row or blank value: mint fresh for this write.
		packInstallID = utils.GenerateID()
	}
	if pack.Category() == packs.ConnectorPack {
		if err := softDeleteFailedConnectorPackRuntimeRows(ctx, db, pack.Name()); err != nil {
			return err
		}
	}

	msg := applyErr.Error()
	if len(msg) > 4000 {
		msg = msg[:4000]
	}
	now := time.Now().UTC()
	meta := packs.PackHashMetadataForPack(pack)
	runtimeDecl, declErr := packs.BuildPersistedRuntimeDeclaration(pack)
	if declErr != nil {
		return nil
	}
	hasRuntimeDeclarationColumns, err := installedPackHasRuntimeDeclarationColumns(ctx, db)
	if err != nil {
		return err
	}
	if !hasRuntimeDeclarationColumns {
		if _, err := db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
					(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, InstallSource, LastError, CreatedDate, LastModifiedDate, IsDeleted)
				 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, 'apply', ?, ?, ?, 0)
			 ON DUPLICATE KEY UPDATE
				Category = VALUES(Category),
				Status = VALUES(Status),
				PackVersion = VALUES(PackVersion),
				ManifestFormatVersion = VALUES(ManifestFormatVersion),
				PlatformCompatibility = VALUES(PlatformCompatibility),
				DesiredPayloadHash = VALUES(DesiredPayloadHash),
				PayloadHashAlgorithm = VALUES(PayloadHashAlgorithm),
				SourceKind = VALUES(SourceKind),
				SourceURI = VALUES(SourceURI),
				SourceRevision = VALUES(SourceRevision),
				LastError = VALUES(LastError),
				LastModifiedDate = VALUES(LastModifiedDate),
				IsDeleted = 0`,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusFailed,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			msg, now, now,
		); err != nil {
			return fmt.Errorf("mark installed pack %s failed: %w", pack.Name(), err)
		}
		if pack.Category() == packs.ConnectorPack {
			if err := reconcileRegisteredConnectorInboundWorkers(ctx, db); err != nil {
				return fmt.Errorf("reconcile connector inbound workers after failed pack %s: %w", pack.Name(), err)
			}
		}
		return nil
	}
	hasRuntimeStartupShallowColumn, err := installedPackHasRuntimeStartupShallowDeclarationColumn(ctx, db)
	if err != nil {
		return err
	}
	hasRuntimeFormulaBundlesColumn, err := installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx, db)
	if err != nil {
		return err
	}
	if !hasRuntimeStartupShallowColumn && !hasRuntimeFormulaBundlesColumn {
		if _, err := db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
				(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, InstallSource, LastError, CreatedDate, LastModifiedDate, IsDeleted)
			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', ?, ?, ?, 0)
			 ON DUPLICATE KEY UPDATE
				Category = VALUES(Category),
				Status = VALUES(Status),
				PackVersion = VALUES(PackVersion),
				ManifestFormatVersion = VALUES(ManifestFormatVersion),
				PlatformCompatibility = VALUES(PlatformCompatibility),
				DesiredPayloadHash = VALUES(DesiredPayloadHash),
				PayloadHashAlgorithm = VALUES(PayloadHashAlgorithm),
				SourceKind = VALUES(SourceKind),
				SourceURI = VALUES(SourceURI),
				SourceRevision = VALUES(SourceRevision),
				RuntimeObjectSpecDeclarations = VALUES(RuntimeObjectSpecDeclarations),
				RuntimeRelatedListRules = VALUES(RuntimeRelatedListRules),
				RuntimeSearchScopeOrder = VALUES(RuntimeSearchScopeOrder),
				RuntimeSearchAllDefaultFields = VALUES(RuntimeSearchAllDefaultFields),
				LastError = VALUES(LastError),
				LastModifiedDate = VALUES(LastModifiedDate),
				IsDeleted = 0`,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusFailed,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(),
			msg, now, now,
		); err != nil {
			return fmt.Errorf("mark installed pack %s failed: %w", pack.Name(), err)
		}
		if pack.Category() == packs.ConnectorPack {
			if err := reconcileRegisteredConnectorInboundWorkers(ctx, db); err != nil {
				return fmt.Errorf("reconcile connector inbound workers after failed pack %s: %w", pack.Name(), err)
			}
		}
		return nil
	}
	if !hasRuntimeStartupShallowColumn {
		if _, err := db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
				(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles, InstallSource, LastError, CreatedDate, LastModifiedDate, IsDeleted)
			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', ?, ?, ?, 0)
			 ON DUPLICATE KEY UPDATE
				Category = VALUES(Category),
				Status = VALUES(Status),
				PackVersion = VALUES(PackVersion),
				ManifestFormatVersion = VALUES(ManifestFormatVersion),
				PlatformCompatibility = VALUES(PlatformCompatibility),
				DesiredPayloadHash = VALUES(DesiredPayloadHash),
				PayloadHashAlgorithm = VALUES(PayloadHashAlgorithm),
				SourceKind = VALUES(SourceKind),
				SourceURI = VALUES(SourceURI),
				SourceRevision = VALUES(SourceRevision),
				RuntimeObjectSpecDeclarations = VALUES(RuntimeObjectSpecDeclarations),
				RuntimeRelatedListRules = VALUES(RuntimeRelatedListRules),
				RuntimeSearchScopeOrder = VALUES(RuntimeSearchScopeOrder),
				RuntimeSearchAllDefaultFields = VALUES(RuntimeSearchAllDefaultFields),
				RuntimeFormulaFunctionBundles = VALUES(RuntimeFormulaFunctionBundles),
				LastError = VALUES(LastError),
				LastModifiedDate = VALUES(LastModifiedDate),
				IsDeleted = 0`,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusFailed,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.FormulaFunctionBundlesJSON(),
			msg, now, now,
		); err != nil {
			return fmt.Errorf("mark installed pack %s failed: %w", pack.Name(), err)
		}
		if pack.Category() == packs.ConnectorPack {
			if err := reconcileRegisteredConnectorInboundWorkers(ctx, db); err != nil {
				return fmt.Errorf("reconcile connector inbound workers after failed pack %s: %w", pack.Name(), err)
			}
		}
		return nil
	}
	if !hasRuntimeFormulaBundlesColumn {
		if _, err := db.ExecContext(ctx,
			`INSERT INTO _System_InstalledPack
			(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations, InstallSource, LastError, CreatedDate, LastModifiedDate, IsDeleted)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', ?, ?, ?, 0)
		 ON DUPLICATE KEY UPDATE
			Category = VALUES(Category),
			Status = VALUES(Status),
			PackVersion = VALUES(PackVersion),
			ManifestFormatVersion = VALUES(ManifestFormatVersion),
			PlatformCompatibility = VALUES(PlatformCompatibility),
			DesiredPayloadHash = VALUES(DesiredPayloadHash),
			PayloadHashAlgorithm = VALUES(PayloadHashAlgorithm),
			SourceKind = VALUES(SourceKind),
			SourceURI = VALUES(SourceURI),
			SourceRevision = VALUES(SourceRevision),
			RuntimeObjectSpecDeclarations = VALUES(RuntimeObjectSpecDeclarations),
			RuntimeRelatedListRules = VALUES(RuntimeRelatedListRules),
			RuntimeSearchScopeOrder = VALUES(RuntimeSearchScopeOrder),
			RuntimeSearchAllDefaultFields = VALUES(RuntimeSearchAllDefaultFields),
			RuntimeStartupShallowDeclarations = VALUES(RuntimeStartupShallowDeclarations),
			LastError = VALUES(LastError),
			LastModifiedDate = VALUES(LastModifiedDate),
			IsDeleted = 0`,
			installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusFailed,
			meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
			payloadHash, meta.PayloadHashAlgorithm,
			nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
			runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.StartupShallowDeclaredRowsJSON(),
			msg, now, now,
		); err != nil {
			return fmt.Errorf("mark installed pack %s failed: %w", pack.Name(), err)
		}
		if pack.Category() == packs.ConnectorPack {
			if err := reconcileRegisteredConnectorInboundWorkers(ctx, db); err != nil {
				return fmt.Errorf("reconcile connector inbound workers after failed pack %s: %w", pack.Name(), err)
			}
		}
		return nil
	}
	if _, err := db.ExecContext(ctx,
		`INSERT INTO _System_InstalledPack
			(Id, Name, PackInstallId, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles, RuntimeStartupShallowDeclarations, InstallSource, LastError, CreatedDate, LastModifiedDate, IsDeleted)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'apply', ?, ?, ?, 0)
		 ON DUPLICATE KEY UPDATE
			Category = VALUES(Category),
			Status = VALUES(Status),
			PackVersion = VALUES(PackVersion),
			ManifestFormatVersion = VALUES(ManifestFormatVersion),
			PlatformCompatibility = VALUES(PlatformCompatibility),
			DesiredPayloadHash = VALUES(DesiredPayloadHash),
			PayloadHashAlgorithm = VALUES(PayloadHashAlgorithm),
			SourceKind = VALUES(SourceKind),
			SourceURI = VALUES(SourceURI),
			SourceRevision = VALUES(SourceRevision),
			RuntimeObjectSpecDeclarations = VALUES(RuntimeObjectSpecDeclarations),
			RuntimeRelatedListRules = VALUES(RuntimeRelatedListRules),
			RuntimeSearchScopeOrder = VALUES(RuntimeSearchScopeOrder),
			RuntimeSearchAllDefaultFields = VALUES(RuntimeSearchAllDefaultFields),
			RuntimeFormulaFunctionBundles = VALUES(RuntimeFormulaFunctionBundles),
			RuntimeStartupShallowDeclarations = VALUES(RuntimeStartupShallowDeclarations),
			LastError = VALUES(LastError),
			LastModifiedDate = VALUES(LastModifiedDate),
			IsDeleted = 0`,
		installedPackID(pack.Name()), pack.Name(), packInstallID, string(pack.Category()), installedPackStatusFailed,
		meta.PackVersion, meta.ManifestFormatVersion, nullString(meta.PlatformCompatibility, meta.PlatformCompatibilityValid),
		payloadHash, meta.PayloadHashAlgorithm,
		nullString(meta.SourceKind, meta.SourceKindValid), nullString(meta.SourceURI, meta.SourceURIValid), nullString(meta.SourceRevision, meta.SourceRevisionValid),
		runtimeDecl.ObjectPayloadsJSON(), runtimeDecl.RelatedListRulesJSON(), runtimeDecl.SearchScopeOrderJSON(), runtimeDecl.SearchAllDefaultFieldsJSON(), runtimeDecl.FormulaFunctionBundlesJSON(), runtimeDecl.StartupShallowDeclaredRowsJSON(),
		msg, now, now,
	); err != nil {
		return fmt.Errorf("mark installed pack %s failed: %w", pack.Name(), err)
	}
	if pack.Category() == packs.ConnectorPack {
		if err := reconcileRegisteredConnectorInboundWorkers(ctx, db); err != nil {
			return fmt.Errorf("reconcile connector inbound workers after failed pack %s: %w", pack.Name(), err)
		}
	}
	return nil
}

func softDeleteFailedConnectorPackRuntimeRows(ctx context.Context, db *sql.DB, packName string) error {
	return softDeleteFailedConnectorPackRuntimeRowsWithExec(ctx, db.ExecContext, packName)
}

func softDeleteFailedConnectorPackRuntimeRowsWithExec(
	ctx context.Context,
	exec func(context.Context, string, ...any) (sql.Result, error),
	packName string,
) error {
	trimmedName := strings.TrimSpace(packName)
	if trimmedName == "" {
		return nil
	}
	lowerName := strings.ToLower(trimmedName)
	var errs []error
	if _, err := exec(ctx,
		`UPDATE _System_ConnectorCapabilityDeclaration
		    SET IsDeleted = 1, LastModifiedDate = CURRENT_TIMESTAMP
		  WHERE InstalledPackName = ? AND COALESCE(IsDeleted, 0) = 0`,
		trimmedName,
	); err != nil {
		errs = append(errs, fmt.Errorf("soft-delete connector capability declarations for failed pack %s: %w", trimmedName, err))
	}
	if _, err := exec(ctx,
		`UPDATE _System_ConnectorRateLimit
		    SET IsDeleted = 1, LastModifiedDate = CURRENT_TIMESTAMP
		  WHERE LOWER(TRIM(OwnerPack)) = ? AND COALESCE(IsDeleted, 0) = 0`,
		lowerName,
	); err != nil {
		errs = append(errs, fmt.Errorf("soft-delete connector rate limits for failed pack %s: %w", trimmedName, err))
	}
	if _, err := exec(ctx,
		`UPDATE _System_ConnectorInboundHandlerConfig
		    SET IsDeleted = 1, LastModifiedDate = CURRENT_TIMESTAMP
		  WHERE Namespace = ? AND COALESCE(IsDeleted, 0) = 0`,
		trimmedName,
	); err != nil {
		errs = append(errs, fmt.Errorf("soft-delete connector inbound handler configs for failed pack %s: %w", trimmedName, err))
	}
	if _, err := exec(ctx,
		`UPDATE _System_ConnectorCalloutProfile
		    SET IsDeleted = 1, LastModifiedDate = CURRENT_TIMESTAMP
		  WHERE OwnerPack = ? AND COALESCE(IsDeleted, 0) = 0`,
		trimmedName,
	); err != nil {
		errs = append(errs, fmt.Errorf("soft-delete connector callout profiles for failed pack %s: %w", trimmedName, err))
	}
	return errors.Join(errs...)
}

func refreshPersistedRuntimeDeclarationsForActivePacks(ctx context.Context, db *sql.DB) error {
	return refreshPersistedRuntimeDeclarationsForActivePacksWithComponentKindRegistrationSet(ctx, db, packs.ComponentKindRegistrationSet{})
}

func refreshPersistedRuntimeDeclarationsForActivePacksWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, registrations packs.ComponentKindRegistrationSet) error {
	hasRuntimeDeclarationColumns, err := installedPackHasRuntimeDeclarationColumns(ctx, db)
	if err != nil {
		return err
	}
	if !hasRuntimeDeclarationColumns {
		return nil
	}
	hasRuntimeStartupShallowColumn, err := installedPackHasRuntimeStartupShallowDeclarationColumn(ctx, db)
	if err != nil {
		return err
	}
	hasRuntimeFormulaBundlesColumn, err := installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx, db)
	if err != nil {
		return err
	}
	rows, err := listInstalledSourceBackedPacks(ctx, db)
	if err != nil {
		return err
	}
	formulaFunctionBundlesForValidation, hasFormulaFunctionBundlesForValidation, err := runtimeFormulaFunctionBundlesFromInstalledPackRows(rows)
	if err != nil {
		return fmt.Errorf("refresh active installed runtime declaration formula function bundles: %w", err)
	}
	for _, row := range rows {
		if row.Status != installedPackStatusActive {
			continue
		}
		if !row.SourceKind.Valid || strings.TrimSpace(row.SourceKind.String) == "" || !row.SourceURI.Valid || strings.TrimSpace(row.SourceURI.String) == "" {
			continue
		}
		pack, err := installedAppPackForRowWithComponentKindRegistrationSet(ctx, db, row, formulaFunctionBundlesForValidation, hasFormulaFunctionBundlesForValidation, registrations)
		if err != nil {
			return err
		}
		if err := verifyInstalledPackHashState(row, pack); err != nil {
			return err
		}
		if err := verifyInstalledPackHostRequirementsState(row, pack); err != nil {
			return err
		}
		decl, err := packs.BuildPersistedRuntimeDeclaration(pack)
		if err != nil {
			return fmt.Errorf("build persisted runtime declaration for active installed pack %s: %w", row.Name, err)
		}
		if installedPackRuntimeDeclarationsCurrent(row, decl, hasRuntimeStartupShallowColumn, hasRuntimeFormulaBundlesColumn) {
			continue
		}
		switch {
		case !hasRuntimeStartupShallowColumn && !hasRuntimeFormulaBundlesColumn:
			result, err := db.ExecContext(ctx, `
				UPDATE _System_InstalledPack
				   SET RuntimeObjectSpecDeclarations = ?,
				       RuntimeRelatedListRules = ?,
				       RuntimeSearchScopeOrder = ?,
				       RuntimeSearchAllDefaultFields = ?
				 WHERE Name = ?
				   AND Category = ?
				   AND Status = ?
				   AND COALESCE(IsDeleted, 0) = 0
				   AND SourceKind = ?
				   AND SourceURI = ?
				   AND SourceRevision <=> ?
				   AND AppliedPayloadHash = ?
				   AND DesiredPayloadHash = ?
				   AND AppliedPayloadHash = DesiredPayloadHash
				   AND RuntimeObjectSpecDeclarations <=> ?
				   AND RuntimeRelatedListRules <=> ?
				   AND RuntimeSearchScopeOrder <=> ?
				   AND RuntimeSearchAllDefaultFields <=> ?`,
				decl.ObjectPayloadsJSON(),
				decl.RelatedListRulesJSON(),
				decl.SearchScopeOrderJSON(),
				decl.SearchAllDefaultFieldsJSON(),
				row.Name,
				row.Category,
				installedPackStatusActive,
				row.SourceKind.String,
				row.SourceURI.String,
				nullStringArg(row.SourceRevision),
				row.AppliedPayloadHash.String,
				row.DesiredPayloadHash.String,
				nullStringArg(row.RuntimeObjectSpecDecl),
				nullStringArg(row.RuntimeRelatedLists),
				nullStringArg(row.RuntimeSearchScope),
				nullStringArg(row.RuntimeSearchAllDefault),
			)
			if err != nil {
				return fmt.Errorf("write persisted runtime declaration for active installed pack %s: %w", row.Name, err)
			}
			if err := requireRuntimeDeclarationRefreshAffected(result, row.Name); err != nil {
				return err
			}
		case !hasRuntimeStartupShallowColumn:
			result, err := db.ExecContext(ctx, `
				UPDATE _System_InstalledPack
				   SET RuntimeObjectSpecDeclarations = ?,
				       RuntimeRelatedListRules = ?,
				       RuntimeSearchScopeOrder = ?,
				       RuntimeSearchAllDefaultFields = ?,
				       RuntimeFormulaFunctionBundles = ?
				 WHERE Name = ?
				   AND Category = ?
				   AND Status = ?
				   AND COALESCE(IsDeleted, 0) = 0
				   AND SourceKind = ?
				   AND SourceURI = ?
				   AND SourceRevision <=> ?
				   AND AppliedPayloadHash = ?
				   AND DesiredPayloadHash = ?
				   AND AppliedPayloadHash = DesiredPayloadHash
				   AND RuntimeObjectSpecDeclarations <=> ?
				   AND RuntimeRelatedListRules <=> ?
				   AND RuntimeSearchScopeOrder <=> ?
				   AND RuntimeSearchAllDefaultFields <=> ?
				   AND RuntimeFormulaFunctionBundles <=> ?`,
				decl.ObjectPayloadsJSON(),
				decl.RelatedListRulesJSON(),
				decl.SearchScopeOrderJSON(),
				decl.SearchAllDefaultFieldsJSON(),
				decl.FormulaFunctionBundlesJSON(),
				row.Name,
				row.Category,
				installedPackStatusActive,
				row.SourceKind.String,
				row.SourceURI.String,
				nullStringArg(row.SourceRevision),
				row.AppliedPayloadHash.String,
				row.DesiredPayloadHash.String,
				nullStringArg(row.RuntimeObjectSpecDecl),
				nullStringArg(row.RuntimeRelatedLists),
				nullStringArg(row.RuntimeSearchScope),
				nullStringArg(row.RuntimeSearchAllDefault),
				nullStringArg(row.RuntimeFormulaBundles),
			)
			if err != nil {
				return fmt.Errorf("write persisted runtime declaration for active installed pack %s: %w", row.Name, err)
			}
			if err := requireRuntimeDeclarationRefreshAffected(result, row.Name); err != nil {
				return err
			}
		case !hasRuntimeFormulaBundlesColumn:
			result, err := db.ExecContext(ctx, `
			UPDATE _System_InstalledPack
			   SET RuntimeObjectSpecDeclarations = ?,
			       RuntimeRelatedListRules = ?,
			       RuntimeSearchScopeOrder = ?,
			       RuntimeSearchAllDefaultFields = ?,
			       RuntimeStartupShallowDeclarations = ?
			 WHERE Name = ?
			   AND Category = ?
			   AND Status = ?
			   AND COALESCE(IsDeleted, 0) = 0
			   AND SourceKind = ?
			   AND SourceURI = ?
			   AND SourceRevision <=> ?
			   AND AppliedPayloadHash = ?
			   AND DesiredPayloadHash = ?
			   AND AppliedPayloadHash = DesiredPayloadHash
			   AND RuntimeObjectSpecDeclarations <=> ?
			   AND RuntimeRelatedListRules <=> ?
			   AND RuntimeSearchScopeOrder <=> ?
			   AND RuntimeSearchAllDefaultFields <=> ?
			   AND RuntimeStartupShallowDeclarations <=> ?`,
				decl.ObjectPayloadsJSON(),
				decl.RelatedListRulesJSON(),
				decl.SearchScopeOrderJSON(),
				decl.SearchAllDefaultFieldsJSON(),
				decl.StartupShallowDeclaredRowsJSON(),
				row.Name,
				row.Category,
				installedPackStatusActive,
				row.SourceKind.String,
				row.SourceURI.String,
				nullStringArg(row.SourceRevision),
				row.AppliedPayloadHash.String,
				row.DesiredPayloadHash.String,
				nullStringArg(row.RuntimeObjectSpecDecl),
				nullStringArg(row.RuntimeRelatedLists),
				nullStringArg(row.RuntimeSearchScope),
				nullStringArg(row.RuntimeSearchAllDefault),
				nullStringArg(row.RuntimeStartupShallow),
			)
			if err != nil {
				return fmt.Errorf("write persisted runtime declaration for active installed pack %s: %w", row.Name, err)
			}
			if err := requireRuntimeDeclarationRefreshAffected(result, row.Name); err != nil {
				return err
			}
		default:
			result, err := db.ExecContext(ctx, `
			UPDATE _System_InstalledPack
			   SET RuntimeObjectSpecDeclarations = ?,
			       RuntimeRelatedListRules = ?,
			       RuntimeSearchScopeOrder = ?,
			       RuntimeSearchAllDefaultFields = ?,
			       RuntimeFormulaFunctionBundles = ?,
			       RuntimeStartupShallowDeclarations = ?
			 WHERE Name = ?
			   AND Category = ?
			   AND Status = ?
			   AND COALESCE(IsDeleted, 0) = 0
			   AND SourceKind = ?
			   AND SourceURI = ?
			   AND SourceRevision <=> ?
			   AND AppliedPayloadHash = ?
			   AND DesiredPayloadHash = ?
			   AND AppliedPayloadHash = DesiredPayloadHash
			   AND RuntimeObjectSpecDeclarations <=> ?
			   AND RuntimeRelatedListRules <=> ?
			   AND RuntimeSearchScopeOrder <=> ?
			   AND RuntimeSearchAllDefaultFields <=> ?
			   AND RuntimeFormulaFunctionBundles <=> ?
			   AND RuntimeStartupShallowDeclarations <=> ?`,
				decl.ObjectPayloadsJSON(),
				decl.RelatedListRulesJSON(),
				decl.SearchScopeOrderJSON(),
				decl.SearchAllDefaultFieldsJSON(),
				decl.FormulaFunctionBundlesJSON(),
				decl.StartupShallowDeclaredRowsJSON(),
				row.Name,
				row.Category,
				installedPackStatusActive,
				row.SourceKind.String,
				row.SourceURI.String,
				nullStringArg(row.SourceRevision),
				row.AppliedPayloadHash.String,
				row.DesiredPayloadHash.String,
				nullStringArg(row.RuntimeObjectSpecDecl),
				nullStringArg(row.RuntimeRelatedLists),
				nullStringArg(row.RuntimeSearchScope),
				nullStringArg(row.RuntimeSearchAllDefault),
				nullStringArg(row.RuntimeFormulaBundles),
				nullStringArg(row.RuntimeStartupShallow),
			)
			if err != nil {
				return fmt.Errorf("write persisted runtime declaration for active installed pack %s: %w", row.Name, err)
			}
			if err := requireRuntimeDeclarationRefreshAffected(result, row.Name); err != nil {
				return err
			}
		}
	}
	return nil
}

func runtimeFormulaFunctionBundlesFromInstalledPackRows(rows []installedPackRow) ([]formula.FunctionBundle, bool, error) {
	var out []formula.FunctionBundle
	hasContext := false
	for _, row := range rows {
		if row.Status != installedPackStatusActive {
			continue
		}
		if row.HasRuntimeFormula || row.RuntimeFormulaBundles.Valid {
			hasContext = true
		}
		if !row.RuntimeFormulaBundles.Valid {
			continue
		}
		var bundles []formula.FunctionBundle
		if err := unmarshalInstalledRuntimeFormulaFunctionBundles(row.RuntimeFormulaBundles.String, &bundles); err != nil {
			return nil, false, fmt.Errorf("parse persisted runtime formula function bundles for installed pack %s: %w", strings.TrimSpace(row.Name), err)
		}
		if err := formula.ValidateFunctionBundles(bundles); err != nil {
			return nil, false, fmt.Errorf("validate persisted runtime formula function bundles for installed pack %s: %w", strings.TrimSpace(row.Name), err)
		}
		out = append(out, bundles...)
	}
	return formula.CloneFunctionBundles(out), hasContext, nil
}

func unmarshalInstalledRuntimeFormulaFunctionBundles(raw string, dst *[]formula.FunctionBundle) error {
	trimmed := strings.TrimSpace(raw)
	if trimmed == "" {
		return fmt.Errorf("missing persisted declaration payload")
	}
	if !strings.HasPrefix(trimmed, "[") {
		return fmt.Errorf("persisted declaration payload must be a JSON array")
	}
	return json.Unmarshal([]byte(trimmed), dst)
}

type persistedRuntimeDeclarationJSON interface {
	ObjectPayloadsJSON() string
	RelatedListRulesJSON() string
	SearchScopeOrderJSON() string
	SearchAllDefaultFieldsJSON() string
	FormulaFunctionBundlesJSON() string
	StartupShallowDeclaredRowsJSON() string
}

func installedPackRuntimeDeclarationsCurrent(row installedPackRow, decl persistedRuntimeDeclarationJSON, includeStartupShallow bool, includeFormulaFunctionBundles bool) bool {
	if !nullStringJSONMatches(row.RuntimeObjectSpecDecl, decl.ObjectPayloadsJSON()) ||
		!nullStringJSONMatches(row.RuntimeRelatedLists, decl.RelatedListRulesJSON()) ||
		!nullStringJSONMatches(row.RuntimeSearchScope, decl.SearchScopeOrderJSON()) ||
		!nullStringJSONMatches(row.RuntimeSearchAllDefault, decl.SearchAllDefaultFieldsJSON()) {
		return false
	}
	if includeFormulaFunctionBundles && !nullStringJSONMatches(row.RuntimeFormulaBundles, decl.FormulaFunctionBundlesJSON()) {
		return false
	}
	if includeStartupShallow {
		return nullStringJSONMatches(row.RuntimeStartupShallow, decl.StartupShallowDeclaredRowsJSON())
	}
	return true
}

func nullStringJSONMatches(value sql.NullString, expected string) bool {
	return value.Valid && runtimeDeclarationJSONEqual(value.String, expected)
}

func runtimeDeclarationJSONEqual(stored, expected string) bool {
	var storedValue any
	var expectedValue any
	if err := json.Unmarshal([]byte(strings.TrimSpace(stored)), &storedValue); err != nil {
		return strings.TrimSpace(stored) == strings.TrimSpace(expected)
	}
	if err := json.Unmarshal([]byte(strings.TrimSpace(expected)), &expectedValue); err != nil {
		return strings.TrimSpace(stored) == strings.TrimSpace(expected)
	}
	return reflect.DeepEqual(storedValue, expectedValue)
}

func nullStringArg(value sql.NullString) any {
	if !value.Valid {
		return nil
	}
	return value.String
}

func requireRuntimeDeclarationRefreshAffected(result sql.Result, packName string) error {
	affected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("confirm persisted runtime declaration refresh for active installed pack %s: %w", packName, err)
	}
	if affected == 0 {
		return fmt.Errorf("refresh persisted runtime declaration for active installed pack %s affected 0 rows; installed-pack row changed concurrently or is no longer active/source-backed/hash-stable", packName)
	}
	return nil
}

// listInstalledSourceBackedPacks returns all non-deleted installed packs whose
// Category is either app-pack or connector-pack. Used by install preflights,
// verify, reconcile, and the clean-L0 contamination check to enumerate all
// source-backed packs regardless of category.
func listInstalledSourceBackedPacks(ctx context.Context, db *sql.DB) ([]installedPackRow, error) {
	hasRuntimeDeclarationColumns, err := installedPackHasRuntimeDeclarationColumns(ctx, db)
	if err != nil {
		return nil, err
	}
	hasRuntimeStartupShallowColumn := false
	hasRuntimeFormulaBundlesColumn := false
	if hasRuntimeDeclarationColumns {
		hasRuntimeStartupShallowColumn, err = installedPackHasRuntimeStartupShallowDeclarationColumn(ctx, db)
		if err != nil {
			return nil, err
		}
		hasRuntimeFormulaBundlesColumn, err = installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx, db)
		if err != nil {
			return nil, err
		}
	}
	rows, includeHostRequirementsDigest, err := queryInstalledSourceBackedPacks(
		ctx,
		db,
		hasRuntimeDeclarationColumns,
		hasRuntimeStartupShallowColumn,
		hasRuntimeFormulaBundlesColumn,
		true,
	)
	if err != nil && persistence.IsUnknownColumnError(err) {
		rows, includeHostRequirementsDigest, err = queryInstalledSourceBackedPacks(
			ctx,
			db,
			hasRuntimeDeclarationColumns,
			hasRuntimeStartupShallowColumn,
			hasRuntimeFormulaBundlesColumn,
			false,
		)
	}
	if err != nil {
		return nil, fmt.Errorf("query installed source-backed packs: %w", err)
	}
	defer rows.Close()

	var out []installedPackRow
	for rows.Next() {
		var row installedPackRow
		dest := []any{
			&row.Name,
			&row.Category,
			&row.Status,
			&row.PackVersion,
			&row.ManifestFormatVersion,
			&row.PlatformCompatibility,
			&row.DesiredPayloadHash,
			&row.AppliedPayloadHash,
			&row.PayloadHashAlgorithm,
			&row.SourceKind,
			&row.SourceURI,
			&row.SourceRevision,
		}
		if includeHostRequirementsDigest {
			dest = append(dest, &row.HostRequirementsDigest)
		}
		if hasRuntimeDeclarationColumns {
			dest = append(dest,
				&row.RuntimeObjectSpecDecl,
				&row.RuntimeRelatedLists,
				&row.RuntimeSearchScope,
				&row.RuntimeSearchAllDefault,
			)
			if hasRuntimeFormulaBundlesColumn {
				row.HasRuntimeFormula = true
				dest = append(dest, &row.RuntimeFormulaBundles)
			}
			if hasRuntimeStartupShallowColumn {
				row.HasRuntimeStartupDecl = true
				dest = append(dest, &row.RuntimeStartupShallow)
			}
		}
		err = rows.Scan(dest...)
		if err != nil {
			return nil, fmt.Errorf("scan installed source-backed pack: %w", err)
		}
		out = append(out, row)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate installed source-backed packs: %w", err)
	}
	return out, nil
}

func queryInstalledSourceBackedPacks(
	ctx context.Context,
	db *sql.DB,
	hasRuntimeDeclarationColumns bool,
	hasRuntimeStartupShallowColumn bool,
	hasRuntimeFormulaBundlesColumn bool,
	includeHostRequirementsDigest bool,
) (*sql.Rows, bool, error) {
	columns := []string{
		"Name",
		"Category",
		"Status",
		"PackVersion",
		"ManifestFormatVersion",
		"PlatformCompatibility",
		"DesiredPayloadHash",
		"AppliedPayloadHash",
		"PayloadHashAlgorithm",
		"SourceKind",
		"SourceURI",
		"SourceRevision",
	}
	if includeHostRequirementsDigest {
		columns = append(columns, "HostRequirementsDigest")
	}
	if hasRuntimeDeclarationColumns {
		columns = append(columns,
			"RuntimeObjectSpecDeclarations",
			"RuntimeRelatedListRules",
			"RuntimeSearchScopeOrder",
			"RuntimeSearchAllDefaultFields",
		)
		if hasRuntimeFormulaBundlesColumn {
			columns = append(columns, "RuntimeFormulaFunctionBundles")
		}
		if hasRuntimeStartupShallowColumn {
			columns = append(columns, "RuntimeStartupShallowDeclarations")
		}
	}
	query := `SELECT ` + strings.Join(columns, ", ") + `
		   FROM _System_InstalledPack
		  WHERE COALESCE(IsDeleted, 0) = 0 AND Category IN (?, ?)
		  ORDER BY Name`
	rows, err := db.QueryContext(ctx,
		query,
		string(packs.AppPack), string(packs.ConnectorPack),
	)
	return rows, includeHostRequirementsDigest, err
}

func installedPackHasRuntimeDeclarationColumns(ctx context.Context, db *sql.DB) (bool, error) {
	var count int
	if err := db.QueryRowContext(ctx, `
		SELECT COUNT(*)
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME IN (?, ?, ?, ?)`,
		"_System_InstalledPack",
		"RuntimeObjectSpecDeclarations",
		"RuntimeRelatedListRules",
		"RuntimeSearchScopeOrder",
		"RuntimeSearchAllDefaultFields",
	).Scan(&count); err != nil {
		return false, fmt.Errorf("query installed-pack runtime declaration columns: %w", err)
	}
	return count == 4, nil
}

func installedPackHasRuntimeStartupShallowDeclarationColumn(ctx context.Context, db *sql.DB) (bool, error) {
	var count int
	if err := db.QueryRowContext(ctx, `
		SELECT COUNT(*)
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME = ?`,
		"_System_InstalledPack",
		"RuntimeStartupShallowDeclarations",
	).Scan(&count); err != nil {
		return false, fmt.Errorf("query installed-pack runtime startup shallow declaration column: %w", err)
	}
	return count == 1, nil
}

func installedPackHasRuntimeFormulaFunctionBundlesColumn(ctx context.Context, db *sql.DB) (bool, error) {
	var count int
	if err := db.QueryRowContext(ctx, `
		SELECT COUNT(*)
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME = ?`,
		"_System_InstalledPack",
		"RuntimeFormulaFunctionBundles",
	).Scan(&count); err != nil {
		return false, fmt.Errorf("query installed-pack runtime formula function bundles column: %w", err)
	}
	return count == 1, nil
}

func installedPackHasHostRequirementsDigestColumn(ctx context.Context, db execQuerier) (bool, error) {
	var count int
	if err := db.QueryRowContext(ctx, `
		SELECT COUNT(*)
		FROM INFORMATION_SCHEMA.COLUMNS
		WHERE TABLE_SCHEMA = DATABASE()
		  AND TABLE_NAME = ?
		  AND COLUMN_NAME = ?`,
		"_System_InstalledPack",
		"HostRequirementsDigest",
	).Scan(&count); err != nil {
		return false, fmt.Errorf("query installed-pack host requirements digest column: %w", err)
	}
	return count == 1, nil
}

func ensureInstalledPackHostRequirementsDigestColumn(ctx context.Context, db execQuerier, pack packs.Pack) (bool, error) {
	hasColumn, err := installedPackHasHostRequirementsDigestColumn(ctx, db)
	if err != nil {
		return false, err
	}
	if !hasColumn && len(packs.PackHostRequirements(pack)) > 0 {
		return false, fmt.Errorf("pack %s declares hostRequirements but _System_InstalledPack schema is missing HostRequirementsDigest; migrate schema before installing this pack", pack.Name())
	}
	return hasColumn, nil
}

func preflightInstalledPackHostRequirementsDigestColumn(ctx context.Context, db execQuerier, pack packs.Pack) (bool, bool, error) {
	if len(packs.PackHostRequirements(pack)) == 0 {
		return false, false, nil
	}
	hasColumn, err := ensureInstalledPackHostRequirementsDigestColumn(ctx, db, pack)
	if err != nil {
		return false, false, err
	}
	return true, hasColumn, nil
}

func updateInstalledPackHostRequirementsDigest(ctx context.Context, db execQuerier, pack packs.Pack, now time.Time) error {
	hasColumn, err := ensureInstalledPackHostRequirementsDigestColumn(ctx, db, pack)
	if err != nil {
		return err
	}
	return writeInstalledPackHostRequirementsDigest(ctx, db, pack, now, hasColumn)
}

func writeInstalledPackHostRequirementsDigest(ctx context.Context, db execQuerier, pack packs.Pack, now time.Time, hasColumn bool) error {
	if !hasColumn {
		return nil
	}
	digest, err := packs.PackHostRequirementsDigest(pack)
	if err != nil {
		return fmt.Errorf("pack %s host requirements digest: %w", pack.Name(), err)
	}
	if _, err := db.ExecContext(ctx, `
UPDATE _System_InstalledPack
   SET HostRequirementsDigest = ?,
       LastModifiedDate = ?
 WHERE Name = ?`,
		digest, now, pack.Name(),
	); err != nil {
		return fmt.Errorf("update installed pack %s host requirements digest: %w", pack.Name(), err)
	}
	return nil
}

func installedPackID(name string) string {
	var b strings.Builder
	b.WriteString("installed_pack_")
	for _, r := range strings.ToLower(name) {
		switch {
		case unicode.IsLetter(r) || unicode.IsDigit(r):
			b.WriteRune(r)
		default:
			b.WriteByte('_')
		}
	}
	return b.String()
}

func declaresRuntimeFormulaFunctionBundles(decl interface{ FormulaFunctionBundlesJSON() string }) bool {
	if decl == nil {
		return false
	}
	return strings.TrimSpace(decl.FormulaFunctionBundlesJSON()) != "[]"
}

func nullString(value string, valid bool) sql.NullString {
	return sql.NullString{String: value, Valid: valid}
}