package controlplane

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"os"
	"strings"
	"sync/atomic"
	"time"

	"github.com/nexuscrm/analytics"
	"github.com/nexuscrm/automation/autonumber"
	"github.com/nexuscrm/automation/triggersub"
	"github.com/nexuscrm/db"
	fileops "github.com/nexuscrm/files"
	infraPersistence "github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	contractcontrolplane "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

var registerExactCandidateRuntimeMetadata = packs.RegisterRuntimeMetadataForInstallingPacksWithOverridesAndComponentKindRegistrationSet

func logControlPlanePhaseStart(operation string, phase string) time.Time {
	start := time.Now()
	log.Printf("controlplane phase start operation=%s phase=%s", operation, phase)
	return start
}

func logControlPlanePhaseDone(operation string, phase string, start time.Time) {
	log.Printf("controlplane phase done operation=%s phase=%s elapsed=%.1fs", operation, phase, time.Since(start).Seconds())
}

func logControlPlanePhaseFailed(operation string, phase string, start time.Time, err error) {
	log.Printf("controlplane phase failed operation=%s phase=%s elapsed=%.1fs err=%v", operation, phase, time.Since(start).Seconds(), err)
}

func runLoggedControlPlanePhase(operation string, phase string, fn func() error) error {
	start := logControlPlanePhaseStart(operation, phase)
	if err := fn(); err != nil {
		logControlPlanePhaseFailed(operation, phase, start, err)
		return err
	}
	logControlPlanePhaseDone(operation, phase, start)
	return nil
}

// ApplyMetadata runs substrate DDL and managed metadata reconciliation.
// This is NOT a pack — it is the prerequisite for all packs.
// Synchronous and deterministic: returns only when all writes are complete.
// Returns a potentially-refreshed DB connection (TiDB requires reconnect after DDL).
//
// Does NOT include general repair/backfill operations — those belong to
// RunRepair. Fatal metadata-compatibility cutovers that must complete before a
// source-pack attribution preflight may run are part of schema readiness here.
func ApplyMetadata(ctx context.Context, db *database.TiDBConnection) (*database.TiDBConnection, error) {
	job := &JobResult{Operation: "apply-metadata"}
	start := time.Now()
	log.Println("═══ ApplyMetadata: substrate DDL + managed metadata reconciliation ═══")

	// Establish the physical L0 substrate first. The permission-state singleton
	// cannot exist before its table, so the old one-shot InitializeSchema call
	// had no explicit physical/managed phase boundary.
	job.trackStep("EnsureL0SchemaDDL", func() error {
		return schemaProvisioner.EnsureL0SchemaDDL(db)
	})
	if job.Steps[0].Error != nil {
		job.Status = "failed"
		job.Duration = time.Since(start)
		return nil, fmt.Errorf("EnsureL0SchemaDDL: %w", job.Steps[0].Error)
	}

	// Reconnect after the first DDL phase (TiDB schema cache requires a fresh
	// connection) before entering the managed-schema boundary.
	if err := db.Close(); err != nil {
		log.Printf("   ⚠️  Failed to close old DB connection: %v", err)
	}
	newDB, err := database.NewConnection()
	if err != nil {
		return nil, fmt.Errorf("re-create database connection after DDL: %w", err)
	}
	log.Println("   🔄 Database connection refreshed after DDL")
	db = newDB

	// Run the full, compatibility-preserving schema convergence. Its DDL portion
	// is idempotent. InitializeSchema owns the common managed-writer boundary:
	// it initializes Permission State, binds the fail-closed writer coordinator,
	// and only then performs managed metadata writes. Keeping that prerequisite
	// at the public managed-schema boundary also makes direct repair callers safe.
	if err := job.trackStepFatal("InitializeSchema", func() error {
		return schemaProvisioner.InitializeSchema(db)
	}); err != nil {
		job.Duration = time.Since(start)
		return nil, err
	}

	// InitializeSchema may execute idempotent DDL during warm convergence.
	// Refresh again before subsequent direct SQL reconciliation.
	if err := db.Close(); err != nil {
		log.Printf("   ⚠️  Failed to close post-metadata DB connection: %v", err)
	}
	newDB, err = database.NewConnection()
	if err != nil {
		return nil, fmt.Errorf("re-create database connection after managed metadata: %w", err)
	}
	db = newDB

	// NexusCRM #3707 briefly shipped five SCIM system tables before #3882
	// reverted the feature. Warm databases retain their catalog rows because
	// schema bootstrap is additive. Test/staging deploys run targeted source-pack
	// reapply immediately after ApplyMetadata and before RunRepair, so this
	// closure-aware retirement must be fatal here; otherwise attribution rejects
	// the historical rows before the full repair phase can execute. Physical
	// tables and their historical data are deliberately preserved.
	if err := job.trackStepFatal("RetireRevertedSCIMSchema", func() error {
		_, err := RetireRevertedSCIMSchema(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return nil, err
	}

	// Managed metadata reconciliation (persists correct values to DB)
	job.trackStep("ReconcileManagedStandardObjectAccessPolicies", func() error {
		return ReconcileManagedStandardObjectAccessPolicies(db.DB())
	})
	job.trackStep("ReconcileSystemObjectTableTypes", func() error {
		schemaProvisioner.ReconcileSystemObjectTableTypes(ctx, db.DB())
		return nil
	})
	job.trackStep("ReconcileManagedSystemObjectVisibility", func() error {
		return ReconcileManagedSystemObjectVisibility(db.DB(), managedTableDefinitions())
	})
	job.trackStep("EnsureFolderSharingModels", func() error {
		schemaProvisioner.EnsureFolderSharingModels(ctx, db.DB())
		return nil
	})
	// NOTE: ReconcileManagedRelationshipSemantics moved to pack apply
	// (it needs pack object table defs, which only exist after their pack runs).

	job.Status = "completed"
	job.Duration = time.Since(start)
	log.Printf("═══ ApplyMetadata completed (%s) ═══", job.Summary())
	return db, nil
}

// EnsureL0SchemaDDL creates only L0/system physical tables so init-l0 can run
// a contamination precheck before writing managed metadata rows.
func EnsureL0SchemaDDL(ctx context.Context, db *database.TiDBConnection) (*database.TiDBConnection, error) {
	job := &JobResult{Operation: "ensure-l0-schema-ddl"}
	start := time.Now()
	log.Println("═══ EnsureL0SchemaDDL: L0/system table DDL only ═══")

	job.trackStep("EnsureL0SchemaDDL", func() error {
		return schemaProvisioner.EnsureL0SchemaDDL(db)
	})
	if job.Steps[0].Error != nil {
		job.Status = "failed"
		job.Duration = time.Since(start)
		return nil, fmt.Errorf("EnsureL0SchemaDDL: %w", job.Steps[0].Error)
	}

	if err := db.Close(); err != nil {
		log.Printf("   ⚠️  Failed to close old DB connection: %v", err)
	}
	newDB, err := database.NewConnection()
	if err != nil {
		return nil, fmt.Errorf("re-create database connection after L0 DDL: %w", err)
	}
	job.Status = "completed"
	job.Duration = time.Since(start)
	log.Printf("═══ EnsureL0SchemaDDL completed (%s) ═══", job.Summary())
	return newDB, nil
}

// ApplyL0ManagedMetadata writes only L0/system managed metadata. It excludes
// app-shaped bootstrap phases such as task/event activity indexes.
func ApplyL0ManagedMetadata(ctx context.Context, db *database.TiDBConnection) (*database.TiDBConnection, error) {
	job := &JobResult{Operation: "apply-l0-managed-metadata"}
	start := time.Now()
	log.Println("═══ ApplyL0ManagedMetadata: L0/system metadata only ═══")

	job.trackStep("InitializeSchemaL0", func() error {
		return schemaProvisioner.InitializeSchemaL0(db)
	})
	if job.Steps[0].Error != nil {
		job.Status = "failed"
		job.Duration = time.Since(start)
		return nil, fmt.Errorf("InitializeSchemaL0: %w", job.Steps[0].Error)
	}

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

	job.trackStep("ReconcileManagedStandardObjectAccessPolicies", func() error {
		return ReconcileManagedStandardObjectAccessPolicies(db.DB())
	})
	job.trackStep("ReconcileSystemObjectTableTypes", func() error {
		schemaProvisioner.ReconcileSystemObjectTableTypes(ctx, db.DB())
		return nil
	})
	job.trackStep("ReconcileManagedSystemObjectVisibility", func() error {
		return ReconcileManagedSystemObjectVisibility(db.DB(), managedTableDefinitions())
	})
	job.trackStep("EnsureFolderSharingModels", func() error {
		schemaProvisioner.EnsureFolderSharingModels(ctx, db.DB())
		return nil
	})
	for _, step := range job.Steps {
		if step.Error != nil {
			job.Status = "failed"
			job.Duration = time.Since(start)
			return nil, fmt.Errorf("%s: %w", step.Name, step.Error)
		}
	}

	job.Status = "completed"
	job.Duration = time.Since(start)
	log.Printf("═══ ApplyL0ManagedMetadata completed (%s) ═══", job.Summary())
	return db, nil
}

// RunRepair executes historical compatibility fixes, discovery, and backfills.
// These are not normal startup operations — they handle drift, legacy data,
// and one-time migrations. Separately invokable and observable.
func RunRepair(ctx context.Context, db *database.TiDBConnection) error {
	return RunRepairWithOptions(ctx, db, RepairOptions{})
}

func RunRepairWithOptions(ctx context.Context, db *database.TiDBConnection, options RepairOptions) error {
	return RunRepairWithOptionsAndRunnerOptions(ctx, db, options, RunnerOptions{})
}

// RunRepairWithOptionsAndRunnerOptions executes historical repairs with one
// immutable distribution-owned runner option snapshot.
func RunRepairWithOptionsAndRunnerOptions(ctx context.Context, db *database.TiDBConnection, options RepairOptions, runnerOptions RunnerOptions) error {
	job := &JobResult{Operation: "run-repair"}
	start := time.Now()
	log.Println("═══ RunRepair: discovery + backfills + compatibility fixes ═══")
	activePackBackfillOptions := packs.ActivePackBackfillOptions{
		ComponentKindRegistrationSet: runnerOptions.ComponentKindRegistrationSet,
		DeferredPackNames:            options.deferredSourceBackedLegacyBackfills,
	}

	job.trackStep("DiscoverAndRegisterPhysicalTables", func() error {
		return schemaProvisioner.DiscoverAndRegisterPhysicalTables(db)
	})
	if err := job.trackStepFatal("ReconcileManagedTableColumns", func() error {
		return schemaProvisioner.ReconcileTableColumns(db.DB(), managedTableDefinitions())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// NexusCRM #3707 briefly shipped five SCIM system tables before the whole
	// feature was reverted by #3882. Warm-upgraded databases retain the physical
	// tables and their registry metadata because schema bootstrap is additive.
	// Retire only the exact reverted metadata shape before any pack attribution
	// preflight can classify those no-longer-managed objects as contamination.
	// The physical tables and their historical rows are deliberately preserved.
	if err := job.trackStepFatal("RetireRevertedSCIMSchema", func() error {
		_, err := RetireRevertedSCIMSchema(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Assign a fresh PackInstallId UUID to any _System_InstalledPack row that
	// still carries the upgrade-time default (empty string). Source-backed
	// connector packs are not auto-reapplied during ApplyAll/reconcile, so an
	// upgraded org can keep PackInstallId='' on its existing rows indefinitely.
	// Downstream substrate (NCRM-INT-05 connector event queue, PR #1498) keys
	// per-incarnation state on PackInstallId; blank values collapse all
	// incarnations to one scope. Run after ReconcileManagedTableColumns so the
	// column is guaranteed present before the backfill reads it.
	if err := job.trackStepFatal("MigratePackInstallIdBackfill", func() error {
		return MigratePackInstallIdBackfill(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("BackfillListViewLegacyCutover", func() error {
		_, err := BackfillListViewLegacyCutover(ctx, db.DB())
		if err != nil {
			return err
		}
		return bootstrapStateStore.MarkListViewLegacyCutoverCompleted(db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("MigrateConnectorEventEnqueueOrdinalBackfill", func() error {
		_, err := MigrateConnectorEventEnqueueOrdinalBackfill(ctx, db.DB())
		if err != nil {
			return err
		}
		return bootstrapStateStore.MarkConnectorEventEnqueueOrdinalBackfillCompleted(db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("BackfillWebhookSecretKeyReadProtection", func() error {
		_, err := BackfillWebhookSecretKeyReadProtection(ctx, db.DB())
		if err != nil {
			return err
		}
		return bootstrapStateStore.MarkWebhookSecretKeyReadProtectionBackfillCompleted(db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("BackfillScheduledRunReservationCompatibility", func() error {
		_, err := BackfillScheduledRunReservationCompatibility(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("BackfillProcessInstanceHistory", func() error {
		_, err := BackfillProcessInstanceHistory(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("BackfillRecordTypeDeveloperNames", func() error {
		_, err := BackfillRecordTypeDeveloperNames(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Backfill legacy PermissionSet bitmap columns (PermissionsX=1) into the SSoT
	// junction table _System_PermissionSetPermission. This repairs SF-migrated
	// profiles where the boolean columns were loaded but the SSoT rows were never
	// populated, causing /permissions/effective to under-report grants.
	//
	// Run early so operators can restore access quickly even if later repair steps
	// are slow on production-sized snapshots.
	// Convergence-critical for restart (profile system-permission SSoT grants):
	// best-effort here, but a failure is caught by the end-of-repair
	// fail-closed gate under FailClosedOnConvergenceRepairs (set by ApplyAll), so
	// the completed marker is not stamped while this is unconverged.
	job.trackStep("BackfillPermissionSetPermissionSSoT", func() error {
		return packs.BackfillPermissionSetPermissionSSoTFromBooleanColumns(ctx, db.DB())
	})
	// Backfill missing ObjectPermissions / FieldPermissions rows for setup objects
	// (User/Profile/UserRole/Group/etc.) from live persisted write gates and the
	// now-restored system-permission SSoT grants. This repairs migrated profiles
	// that regained ManageUsers/ManageRoles/ManageSharing but still lack the
	// setup-object OP/FP rows required to render/edit those forms.
	job.trackStep("BackfillSetupObjectPermissions", func() error {
		return packs.BackfillSetupObjectPermissionsFromSystemPerms(ctx, db.DB())
	})
	// Re-run profile permission convergence in headless control-plane mode so
	// standalone `setup repair` can also heal partially seeded profile-owned
	// business-object permissions. This mirrors the pack-time InitializePermissions
	// logic instead of forcing operators to run the full `setup apply` sequence
	// just to converge historical permission drift.
	job.trackStep("ReconcileProfileBusinessPermissions", func() error {
		r, err := NewRunnerWithOptions(db, runnerOptions)
		if err != nil {
			return fmt.Errorf("new runner: %w", err)
		}
		if err := r.RefreshCache(ctx); err != nil {
			return fmt.Errorf("refresh metadata cache: %w", err)
		}
		if err := packs.InitializePermissions(r.SvcMgr.Permissions(), r.SvcMgr.Metadata()); err != nil {
			return fmt.Errorf("initialize permissions: %w", err)
		}
		return nil
	})
	// Strip DDL single-quote wrapping on legacy `_System_Field.DefaultValue`
	// rows. Fatal because a successful rewrite without a migration stamp would
	// make future repairs rewrite post-upgrade authored defaults.
	if err := job.trackStepFatal("NormalizeLegacyDDLDefaults", func() error {
		return schemaProvisioner.NormalizeLegacyDDLDefaults(ctx, db)
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Bridge ProcessAction.FieldValue rows from the pre-PR-#271 runtime
	// contract (raw "Literal: <value>" persisted, runtime trimmed at execution
	// time) to the new opaque-FieldValue contract. Without this migration,
	// warm-upgraded orgs corrupt L3 record data on the next approval transition
	// (issue #230, IANTHEREAL P1 PR #271 Round 3).
	//
	// Fatal: the new runtime in approval_service.go no longer trims the
	// "Literal: " prefix, so an unmigrated row would write the raw string
	// straight into the target field. Best-effort logging is not enough for a
	// runtime-contract cutover that otherwise corrupts L3 data — surface the
	// error to ApplyAll so apply aborts before runtime starts.
	if err := job.trackStepFatal("NormalizeProcessActionLiteralFieldValue", func() error {
		_, err := NormalizeProcessActionLiteralFieldValue(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Resolve _System_WorkflowAlert.Template rows that still carry
	// Salesforce-shaped folder paths or DeveloperName references to the
	// matching _System_EmailTemplate.Id. After Flow-native email cutover, this
	// is fatal only when an unresolved row is still reachable from an active
	// legacy ProcessAction.Alert or Flow emailAlert dependency; unresolved
	// historical rows are allowed to remain as read-only inventory.
	if err := job.trackStepFatal("BackfillWorkflowAlertTemplateIds", func() error {
		_, err := BackfillWorkflowAlertTemplateIds(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Cut runtime navigation visibility from legacy _System_App.AssignedProfiles
	// to Salesforce-modeled ProfileAppVisibility/ProfileTabSetting rows. Fatal:
	// /ui-api/apps now fails closed on missing canonical rows and no longer
	// falls back to AssignedProfiles.
	if err := job.trackStepFatal("BackfillNavigationVisibilitySSoT", func() error {
		appRows, tabRows, err := BackfillNavigationVisibilitySSoT(ctx, db.DB())
		if err == nil && (appRows > 0 || tabRows > 0) {
			log.Printf("repair: backfilled navigation visibility rows app=%d tab=%d", appRows, tabRows)
		}
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Hydrate source-backed ChildCloneRules ownership sidecars before the
	// legacy classifier runs. Standalone setup repair must converge warm-upgrade
	// rows without requiring a full setup apply.
	if err := job.trackStepFatal("BackfillObjectSpecChildCloneRulePackOwnership", func() error {
		updated, err := packs.BackfillObjectSpecChildCloneRulePackOwnershipFromActivePacksWithOptions(ctx, db.DB(), activePackBackfillOptions)
		if err == nil && updated > 0 {
			log.Printf("repair: backfilled ChildCloneRules pack ownership rows=%d", updated)
		}
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("BackfillObjectSpecExclusiveSiblingFieldPackOwnership", func() error {
		updated, err := packs.BackfillObjectSpecExclusiveSiblingFieldPackOwnershipFromActivePacksWithOptions(ctx, db.DB(), activePackBackfillOptions)
		if err == nil && updated > 0 {
			log.Printf("repair: backfilled ExclusiveSiblingFields pack ownership rows=%d", updated)
		}
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Rewrite historical source-backed curated default layout IDs back to the
	// canonical curated_<Object>_default identity before steady-state pack
	// apply/verify runs. This is bounded warm-upgrade cleanup, not runtime
	// alternate-identity support.
	if err := job.trackStepFatal("BackfillSourceBackedCuratedDefaultLayoutIDs", func() error {
		updated, err := packs.BackfillLegacySourceBackedCuratedDefaultLayoutIDsFromActivePacksWithOptions(ctx, db.DB(), activePackBackfillOptions)
		if err == nil && updated > 0 {
			log.Printf("repair: backfilled source-backed curated default layout IDs rows=%d", updated)
		}
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Stamp legacy _System_Object.ChildCloneRules rows into the new
	// column-scoped ownership bit after source-backed ownership has been
	// hydrated. The classifier intentionally leaves unresolved sidecar-less rows
	// alone so pack reconciliation can prove ownership first.
	if err := job.trackStepFatal("BackfillChildCloneRulesCustomized", func() error {
		_, err := packs.BackfillChildCloneRulesCustomized(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Repair the historical pre-owner-pack curated User layout state before
	// steady-state platform-ux apply runs. This is a bounded upgrade backfill,
	// not part of the runtime User layout reconcile path.
	if err := job.trackStepFatal("BackfillLegacyCuratedUserLayoutOwnership", func() error {
		return packs.BackfillLegacyOwnerlessCuratedUserLayoutOwnership(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Recompute legacy naive '+s' PluralLabel rows in _System_Object that
	// were written before Pluralize() / ensurePluralLabelsAndNameFields
	// became canonical (e.g. "Opportunitys" → "Opportunities", "Price Book
	// Entrys" → "Price Book Entries"). Best-effort: wrong PluralLabel is a
	// UI-cosmetic / describe.labelPlural wire-shape issue, not L3 data
	// corruption, so a transient failure logs and ApplyAll continues. The
	// migration is stamp-guarded and the WHERE guard (IsCustomized=0 +
	// naive-+s shape + canonical-differs-from-stored) makes a re-run safe
	// after a partial failure and aligns with pack apply's
	// preserveCustomized semantics so admin overrides are never rewritten
	// (issue NX#492).
	job.trackStep("MigrateLegacyNaivePlurals", func() error {
		_, err := MigrateLegacyNaivePlurals(ctx, db.DB())
		return err
	})
	// Remove the hard FK on GroupMember.UserOrGroupId → User(Id) so
	// nested-group membership inserts succeed. Without this migration,
	// warm-upgraded instances hit MySQL error 1452 on any GroupMember
	// insert where UserOrGroupId references a Group record (issue #396).
	if err := job.trackStepFatal("DropGroupMemberUserFK", func() error {
		return schemaProvisioner.DropGroupMemberUserFK(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// The Event-owned execution compatibility path is retired. Removing the
	// catalog index is not enough for warm-upgraded databases, because schema
	// initialization is additive. Converge the physical schema before the
	// WorkItem-only runtime is allowed to serve.
	if err := job.trackStepFatal("DropConnectorEventLegacyRetirementIndex", func() error {
		return DropConnectorEventLegacyRetirementIndex(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if !options.DeferConnectorWorkItemCutover {
		if err := job.trackStepFatal("VerifyConnectorExecutionProtocols", func() error {
			return VerifyConnectorExecutionProtocols(ctx, db.DB())
		}); err != nil {
			job.Duration = time.Since(start)
			return err
		}
	}
	if options.SkipPolymorphicDiscriminatorBackfill {
		job.trackStep("BackfillPolymorphicDiscriminators[operator-skipped]", func() error {
			log.Printf("controlplane: operator override enabled; skipping polymorphic discriminator backfill during RunRepair")
			return nil
		})
	} else {
		if err := job.trackStepFatal("BackfillPolymorphicDiscriminators", func() error {
			return packs.BackfillPolymorphicDiscriminators(ctx, db.DB(), managedTableDefinitions())
		}); err != nil {
			job.Duration = time.Since(start)
			return err
		}
	}
	if err := job.trackStepFatal("RebuildGroupMembershipClosure", func() error {
		return infraPersistence.RebuildGroupMembershipClosureWithLock(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Enforce platform-universal capability bounds without overwriting
	// app-pack or tenant-authored _System_Field.Createable/Updateable values.
	job.trackStep("BackfillFieldCapabilities", func() error {
		return schemaProvisioner.BackfillFieldCapabilities(db)
	})
	// Normalize legacy _System_Object.Id values (disc_<table>, mixed-case obj_<ApiName>_<hash>,
	// and bare UUIDs) to the canonical schema.GenerateObjectID(ApiName) scheme.
	// Must run BEFORE ReconcileOrphanFieldObjectIds so the normalized children
	// remain attached to a valid parent — otherwise the orphan reconciler would
	// see briefly-orphaned rows mid-migration.
	job.trackStep("NormalizeObjectIds", func() error {
		return NormalizeObjectIds(ctx, db.DB())
	})
	job.trackStep("ReconcileOrphanFieldObjectIds", func() error {
		return ReconcileOrphanFieldObjectIds(ctx, db.DB())
	})
	// Fail closed only after object-id normalization and orphan-field repair have
	// had a chance to remove stale metadata rows. Any active lookup still lacking
	// ReferenceTo after those repairs is unresolved live metadata drift.
	// Must run BEFORE MigrateIsPolymorphicMultiTargetBackfill: that migration
	// only scans rows with non-empty ReferenceTo, so a warm-upgraded legacy
	// multi-target lookup with empty ReferenceTo would be invisible to it.
	// Running backfill first ensures ReferenceTo is populated before the
	// polymorphic stamp scan, so the repair flow converges legacy orgs to
	// greenfield polymorphic metadata in a single pass.
	if err := job.trackStepFatal("BackfillLookupReferences", func() error {
		return BackfillLookupReferences(db)
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Stamp IsPolymorphic=true on legacy multi-target reference/lookup rows
	// before runtime consumers rely on persisted-read semantics. Runs after
	// BackfillLookupReferences so rows whose ReferenceTo was empty pre-repair
	// are visible to the candidate scan. Must run before the schema cache
	// refresh (step 3 in ApplyAllWithOptions) so scanField's
	// LifecyclePersistedRead resolver sees the explicit flag on old rows.
	// Idempotent via _System_Migration stamp.
	if err := job.trackStepFatal("MigrateIsPolymorphicMultiTargetBackfill", func() error {
		_, err := MigrateIsPolymorphicMultiTargetBackfill(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Normalize master-detail Required/IsMasterDetail metadata before the schema
	// cache refresh. Older batch/pack writes created NOT NULL columns while
	// persisting the authored false flags; this stamped repair makes describe and
	// create-time required-field validation agree with storage. Idempotent via
	// _System_Migration stamp.
	if err := job.trackStepFatal("MigrateMasterDetailRequiredBackfill", func() error {
		_, err := MigrateMasterDetailRequiredBackfill(ctx, db.DB())
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Physical storage cleanup for legacy multi-select picklist values. Runs
	// after object/field metadata repair so table and column resolution uses the
	// converged metadata read path. Best-effort: runtime comparators dual-read
	// legacy storage shapes, and unresolved wrapper rows remain operator-visible
	// without blocking unrelated repair convergence.
	job.trackStep("MigrateMultiPicklistStorageBackfill", func() error {
		return MigrateMultiPicklistStorageBackfill(ctx, db.DB())
	})
	job.trackStep("MigrateMultiPicklistWrapperRepair", func() error {
		return MigrateMultiPicklistWrapperRepair(ctx, db.DB())
	})
	// Project legacy ContentDocument rows into the generic file substrate when
	// those legacy tables are present. This is a compatibility surface repair, not
	// a new authority over the legacy content rows.
	job.trackStep("BackfillContentCompatSystemFiles", func() error {
		return BackfillContentCompatSystemFiles(ctx, db)
	})
	// Known FlowDefinition.ActiveVersionId drift is already owned by the
	// flow-orphan self-heal. Run that repair before the shadow rebuild so a
	// dangling active-version pointer does not abort apply/repair ahead of the
	// existing compatibility fix.
	if err := job.trackStepFatal("ReconcileFlowOrphans", func() error {
		touched, err := ReconcileFlowOrphans(ctx, db)
		if err == nil && touched > 0 {
			log.Printf("repair: auto-reconciled %d orphan FlowDefinition ActiveVersionId row(s)", touched)
		}
		return err
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Shadow-state substrate only for #291: rebuild current active record-trigger
	// subscriptions for already-installed flows. #285/#288 still own the broader
	// install/apply and verify semantics, but this rebuild itself must fail
	// closed so repair does not continue after deactivating shadow rows.
	if err := job.trackStepFatal("BackfillTriggerSubscriptions", func() error {
		return triggersub.Backfill(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	job.trackStep("CleanupExpiredUploadOperations", func() error {
		deleted, err := fileops.CleanupExpiredUploadOperations(ctx, db.DB(), time.Now().UTC())
		if err == nil && deleted > 0 {
			log.Printf("   🧹 CleanupExpiredUploadOperations: deleted %d expired row(s)", deleted)
		}
		return err
	})

	// Marker-honesty gate (ApplyAll path only). Several repair steps are
	// best-effort (trackStep) so standalone `setup repair` heals what it can
	// without aborting. But when RunRepair feeds the control_plane_apply_all
	// marker (FailClosedOnConvergenceRepairs, set by ApplyAll), a swallowed
	// failure in a CONVERGENCE-CRITICAL best-effort step — one whose output the
	// runtime/describe reads and that pod-readiness verify does not re-check —
	// must NOT flow through to a completed marker, or a later auto restart would
	// skip the repair and serve stale state. This single end-of-repair gate
	// covers the whole class (vs per-step fatality, one block at a time). The
	// allowlisted steps are idempotent convergence operations that no-op or reach
	// the same result on a greenfield/clean DB, so failing closed here never
	// breaks first-boot bootstrap.
	if options.FailClosedOnConvergenceRepairs {
		if failed := job.FailedStepNames(convergenceCriticalRepairSteps); len(failed) > 0 {
			job.Status = "failed"
			job.Duration = time.Since(start)
			err := fmt.Errorf("convergence-critical repair step(s) failed: %s — control plane not converged, not marking bootstrapped", strings.Join(failed, ", "))
			job.Error = err
			log.Printf("═══ RunRepair FAILED (%s) ═══", job.Summary())
			return err
		}
	}

	job.Status = "completed"
	job.Duration = time.Since(start)
	log.Printf("═══ RunRepair completed (%s) ═══", job.Summary())
	return nil
}

// convergenceCriticalRepairSteps are best-effort RunRepair steps whose silent
// failure makes the control_plane_apply_all=completed marker dishonest: their
// output is read by the runtime/describe and is NOT re-checked by pod-readiness
// verify, so a swallowed failure would let auto restart skip the repair and
// serve stale state. Under FailClosedOnConvergenceRepairs (ApplyAll), a failure
// in any of these prevents stamping the completed marker. Each is an idempotent
// convergence step that either no-ops or reaches the same result on a
// greenfield/clean DB, so gating on them is safe for first-boot bootstrap.
var convergenceCriticalRepairSteps = map[string]bool{
	"DiscoverAndRegisterPhysicalTables":   true, // physical-table object-metadata inventory
	"BackfillPermissionSetPermissionSSoT": true, // system-permission SSoT grants
	"BackfillSetupObjectPermissions":      true, // setup-object OP/FP rows
	"BackfillFieldCapabilities":           true, // _System_Field Createable/Updateable bounds
	"MigrateLegacyNaivePlurals":           true, // _System_Object.PluralLabel canonicalization
}

// RunL0Repair executes only repair primitives that are safe for a clean L0
// initialization. Broad compatibility repair must stay out of init-l0 because it
// can discover or mutate app/business metadata left in a reused database.
func RunL0Repair(ctx context.Context, db *database.TiDBConnection) error {
	job := &JobResult{Operation: "run-l0-repair"}
	start := time.Now()
	log.Println("═══ RunL0Repair: L0-safe compatibility fixes only ═══")

	// L0-safe schema repair: GroupMember is part of the platform substrate
	// (auth tables), not an app pack. Warm-upgrade init-l0 must drop the
	// old physical FK so polymorphic CRUD, import, and repair paths can
	// populate Group-target GroupMember rows. The migration is gated by a
	// _System_Migration stamp, so on a greenfield init-l0 (FK already
	// absent) it stamps and exits without touching the schema. Must be
	// fatal: warm-upgrade L0s that fail to drop the FK retain a
	// schema-shape diff vs. greenfield, leaving the substrate inconsistent
	// across deployments.
	if err := job.trackStepFatal("DropGroupMemberUserFK", func() error {
		return schemaProvisioner.DropGroupMemberUserFK(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("DropConnectorEventLegacyRetirementIndex", func() error {
		return DropConnectorEventLegacyRetirementIndex(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	// Repair discriminator data only for the managed L0 definitions. The global
	// IsPolymorphic metadata migration stays in RunRepair after
	// BackfillLookupReferences: running it here could stamp completion while a
	// warm-upgraded customized L0 row still has empty ReferenceTo, permanently
	// hiding that row from the later broad migration.
	if err := job.trackStepFatal("BackfillPolymorphicDiscriminators", func() error {
		return packs.BackfillPolymorphicDiscriminators(ctx, db.DB(), managedTableDefinitions())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}
	if err := job.trackStepFatal("RebuildGroupMembershipClosure", func() error {
		return infraPersistence.RebuildGroupMembershipClosureWithLock(ctx, db.DB())
	}); err != nil {
		job.Duration = time.Since(start)
		return err
	}

	// TODO(#56-backend): Add L0-object-scoped wrappers before promoting broad
	// _System_Object/_System_Field repair steps here. The full RunRepair steps
	// intentionally excluded from init-l0 are:
	// - DiscoverAndRegisterPhysicalTables: would reify app/business tables.
	// - BackfillPermissionSetPermissionSSoT: can materialize non-L0 grants.
	// - BackfillSetupObjectPermissions: seeds object permissions from metadata.
	// - ReconcileProfileBusinessPermissions: converges current object metadata.
	// - NormalizeLegacyDDLDefaults, BackfillFieldCapabilities,
	//   BackfillLookupReferences, NormalizeObjectIds, and
	//   ReconcileOrphanFieldObjectIds until they are L0-object-scoped.
	job.trackStep("CleanupExpiredUploadOperations", func() error {
		deleted, err := fileops.CleanupExpiredUploadOperations(ctx, db.DB(), time.Now().UTC())
		if err == nil && deleted > 0 {
			log.Printf("   🧹 CleanupExpiredUploadOperations: deleted %d expired row(s)", deleted)
		}
		return err
	})

	job.Status = "completed"
	job.Duration = time.Since(start)
	log.Printf("═══ RunL0Repair completed (%s) ═══", job.Summary())
	return nil
}

// EnsureIndexes creates standard indexes on business tables (~900 indexes).
// This is a separate tracked operation — not hidden inside ApplyMetadata.
// Synchronous: returns when all indexes are created. Can be slow.
func EnsureIndexes(_ context.Context, db *database.TiDBConnection) error {
	log.Println("═══ EnsureIndexes: creating standard indexes ═══")
	start := time.Now()
	if err := schemaProvisioner.EnsureStandardIndexes(db); err != nil {
		log.Printf("═══ EnsureIndexes failed: %v (%.1fs) ═══", err, time.Since(start).Seconds())
		return fmt.Errorf("EnsureStandardIndexes: %w", err)
	}
	log.Printf("═══ EnsureIndexes completed (%.1fs) ═══", time.Since(start).Seconds())
	return nil
}

// ApplyPacks applies runtime platform packs in dependency order. DevSample is
// appended only for non-production/dev runs; L1 app packs install through
// InstallPackFromSource.
func ApplyPacks(ctx context.Context, r *Runner, includeDevSample bool) error {
	return ApplyPacksWithOptions(ctx, r, includeDevSample, packs.ApplyOptions{})
}

func ApplyPacksWithOptions(ctx context.Context, r *Runner, includeDevSample bool, options packs.ApplyOptions) error {
	packList := packs.AllPacks(includeDevSample)
	log.Printf("═══ ApplyPacks: %d pack(s) queued ═══", len(packList))
	start := time.Now()
	var appPacks []appliedAppPack

	if err := packs.BackfillLegacyOwnerlessCuratedUserLayoutOwnership(ctx, r.DB.DB()); err != nil {
		return fmt.Errorf("backfill legacy curated User layout ownership: %w", err)
	}

	for _, pack := range packList {
		packStart := time.Now()
		log.Printf("   ▶ pack apply start: %s category=%s", pack.Name(), pack.Category())
		if pack.Category() == packs.PlatformUX {
			if err := reconcileFlowOrphansBeforeFlowRuntimeMaintenance(ctx, r.DB, "apply-packs"); err != nil {
				return err
			}
		}
		var payloadHash string
		if pack.Category() == packs.AppPack {
			var err error
			payloadHash, err = packs.PackPayloadHash(pack)
			if err != nil {
				return fmt.Errorf("pack %s payload hash: %w", pack.Name(), err)
			}
			if err := markInstalledPackInstalling(ctx, r.DB.DB(), pack, payloadHash); err != nil {
				return err
			}
		}
		if err := pack.Apply(ctx, r.PackDepsWithOptions(options)); err != nil {
			if pack.Category() == packs.AppPack {
				if markErr := markInstalledPackFailed(ctx, r.DB.DB(), pack, payloadHash, err); markErr != nil {
					return fmt.Errorf("pack %s: %w", pack.Name(), errors.Join(err, fmt.Errorf("mark installed pack failed: %w", markErr)))
				}
			}
			return fmt.Errorf("pack %s: %w", pack.Name(), err)
		}
		log.Printf("   ✅ pack apply done: %s category=%s (%.1fs)", pack.Name(), pack.Category(), time.Since(packStart).Seconds())
		if pack.Category() == packs.AppPack {
			appPacks = append(appPacks, appliedAppPack{pack: pack, payloadHash: payloadHash})
		}
		// Inter-pack cache refresh at causal boundaries:
		// - After PlatformCore: platform object metadata must be visible downstream.
		// - After PlatformTenant: tenant-overridden picklists/labels must be
		//   visible to downstream packs that read field metadata.
		// - After PlatformConfig: platform config must be visible to UX and app packs.
		// - After app packs: app-authored metadata must be visible to validators.
		if pack.Category() == packs.PlatformCore || pack.Category() == packs.PlatformTenant || pack.Category() == packs.PlatformConfig || pack.Category() == packs.AppPack {
			refreshStart := time.Now()
			log.Printf("   ▶ cache refresh start after pack: %s", pack.Name())
			if err := r.RefreshCache(ctx); err != nil {
				return fmt.Errorf("cache refresh after %s failed (downstream packs depend on it): %w", pack.Name(), err)
			}
			log.Printf("   ✅ cache refresh done after pack: %s (%.1fs)", pack.Name(), time.Since(refreshStart).Seconds())
			// Post-refresh validators for DB-backed metadata surfaces.
			// They no-op when no rows exist and become active when a pack
			// seeds rows on either surface.
			if pack.Category() == packs.PlatformConfig || pack.Category() == packs.AppPack {
				validatorStart := time.Now()
				log.Printf("   ▶ insert-rules validation start after pack: %s", pack.Name())
				if err := runInsertRulesValidators(ctx, r); err != nil {
					return fmt.Errorf("insert-rules validation after %s: %w", pack.Name(), err)
				}
				log.Printf("   ✅ insert-rules validation done after pack: %s (%.1fs)", pack.Name(), time.Since(validatorStart).Seconds())
			}
		}
	}
	if err := finalizeAppliedAppPacks(ctx, r.DB.DB(), appPacks); err != nil {
		return err
	}

	log.Printf("═══ ApplyPacks completed (%.1fs) ═══", time.Since(start).Seconds())
	return nil
}

// ApplyRuntimeRequiredPacks applies only the platform packs required by the L0
// runtime. It deliberately bypasses AllPacks() and DevSamplePack.
func ApplyRuntimeRequiredPacks(ctx context.Context, r *Runner) error {
	log.Println("═══ ApplyRuntimeRequiredPacks ═══")
	start := time.Now()

	if err := packs.BackfillLegacyOwnerlessCuratedUserLayoutOwnership(ctx, r.DB.DB()); err != nil {
		return fmt.Errorf("backfill legacy curated User layout ownership: %w", err)
	}

	for _, pack := range packs.RuntimeRequiredPacks() {
		if err := pack.Apply(ctx, r.PackDeps()); err != nil {
			return fmt.Errorf("pack %s: %w", pack.Name(), err)
		}
		if pack.Category() == packs.PlatformCore || pack.Category() == packs.PlatformTenant || pack.Category() == packs.PlatformConfig {
			if err := r.RefreshCache(ctx); err != nil {
				return fmt.Errorf("cache refresh after %s failed (downstream packs depend on it): %w", pack.Name(), err)
			}
			if pack.Category() == packs.PlatformConfig {
				if err := runInsertRulesValidators(ctx, r); err != nil {
					return fmt.Errorf("insert-rules validation after %s: %w", pack.Name(), err)
				}
			}
		}
	}

	log.Printf("═══ ApplyRuntimeRequiredPacks completed (%.1fs) ═══", time.Since(start).Seconds())
	return nil
}

func runLayoutMaintenance(db *sql.DB) error {
	// Generic layout maintenance is an orchestration concern: run it once after
	// all selected packs have landed so it sees both platform and app metadata.
	// Propagate the error (rather than only logging): the related-list config it
	// writes is read by record-detail UI and is not re-checked by pod-readiness
	// verify, so a swallowed failure here would let ApplyAll stamp the completed
	// marker on layouts that lack their related lists, and a later auto restart
	// would skip the repair (#1590 marker-honesty). It runs only after app packs
	// have applied, so a genuine failure correctly fails closed.
	if err := packs.BackfillLayoutRelatedLists(db); err != nil {
		return fmt.Errorf("related list backfill: %w", err)
	}
	return nil
}

// ApplyPacksWithLock serializes standalone pack application with the same
// advisory lock used by ApplyAll. This covers cmd/setup apply-packs invocations
// that otherwise bypass the full control-plane lock.
func ApplyPacksWithLock(ctx context.Context, r *Runner, includeDevSample bool) error {
	return withControlPlaneMutationLock(ctx, r.DB.DB(), "ApplyPacks", func() error {
		return ApplyPacks(ctx, r, includeDevSample)
	})
}

// runInsertRulesValidators runs the two DB-driven validators for generic
// metadata surfaces:
//
//   - ValidateInsertInheritFromLookupRules inspects every
//     _System_ObjectFieldMapping row with ActionType='insertInheritFromLookup'
//     and rejects rows whose schema references no longer resolve.
//   - ValidateObjectChildCloneRules walks every ObjectMetadata's
//     ChildCloneRules JSON and rejects rules that point at missing
//     metadata.
//   - ValidateObjectDefaultChildRows walks every ObjectMetadata's
//     DefaultChildRows declarations and rejects invalid target metadata
//     before runtime insert reaches them.
//
// Both validators are DB-driven — they read whatever rows currently
// exist and reject the malformed ones. ChildCloneRules may arrive from
// source-backed pack object_spec payload or explicit L2 tenant overlay
// payload; the ChildCloneRulesCustomized discriminator keeps pack reapply
// and tenant/admin ownership from becoming parallel authorities. The
// validators no-op cleanly on a greenfield DB (no rows → no rejections),
// which is why this function is safe to invoke from BOTH the tenant path
// (ApplyTenant) AND the generic platform path (ApplyPacks).
//
// Declared-vs-DB closure (VerifyInsertRulesLanded) is NOT called from
// here. Round-7 reviewer finding #1: that closure reads tenant seed
// files, so invoking it from the shared validator helper leaks tenant-
// specific reads into greenfield `cmd/setup apply`. The platform-boundary
// invariant is that tenant overlay reads live only on the ApplyTenant path.
// The closure call now sits inline in ApplyTenant after this function.
//
// Passes a real engineOwnedByObject function backed by
// sm.AutoNumberMetadata — NOT nil — so the validator's engine-owned check
// at packs/insert_rules_validator.go fires. Shipping nil here would let
// a seed row that writes an engine-owned target field pass validation
// while the runtime applier silently drops it (canInheritTarget rejects
// engine-owned targets) — authoring surface inconsistent with runtime.
//
// Allowed callers are deep verify / apply paths that intentionally validate
// live app or tenant metadata. VerifyL0 is explicitly forbidden: clean-L0
// verification is scoped to generic substrate + RuntimeRequiredPacks().
func runInsertRulesValidators(ctx context.Context, r *Runner) error {
	if r == nil || r.SvcMgr == nil || r.SvcMgr.Metadata() == nil {
		return nil
	}
	meta := r.SvcMgr.Metadata()
	engineOwned := engineOwnedByObjectFromAutoNumber(r.SvcMgr.AutoNumberMetadata())
	if err := packs.ValidateInsertInheritFromLookupRules(ctx, r.DB.DB(), meta, engineOwned); err != nil {
		return err
	}
	if err := packs.ValidateObjectChildCloneRules(ctx, meta, meta.GetSchemas(ctx)); err != nil {
		return err
	}
	if err := packs.ValidateObjectDefaultChildRows(ctx, meta, meta.GetSchemas(ctx), engineOwned); err != nil {
		return err
	}
	return nil
}

// engineOwnedByObjectFromAutoNumber returns a per-object lookup of
// field API names (lower-cased) that the autonumber engine renders —
// exactly the set records.engineOwnedFieldSet produces at insert time.
// Exposed here so the orchestrator's validator call uses the same
// authoritative source as the runtime applier; mismatch would let bad
// metadata pass validation and be silently ignored on insert.
//
// Returns nil when the metadata source is unset (test harnesses) so the
// validator falls back to "no engine-owned fields", which matches the
// runtime applier's nil-guard path.
func engineOwnedByObjectFromAutoNumber(src *autonumber.MetadataSource) func(string) map[string]bool {
	if src == nil {
		return nil
	}
	return func(objectName string) map[string]bool {
		fields := src.LookupByObject(objectName)
		if len(fields) == 0 {
			return nil
		}
		out := make(map[string]bool, len(fields))
		for _, fm := range fields {
			out[strings.ToLower(fm.FieldAPIName)] = true
		}
		return out
	}
}

// RebuildDerived rebuilds derived state from authoritative metadata:
// formula views and the metadata cache.
func RebuildDerived(ctx context.Context, r *Runner) error {
	// The exported entry point does not serialize on the control-plane apply
	// lock, so it must never mutate the rebuild-owed receipt: settlement
	// happens only on lock-holding paths (RebuildDerivedWithLock*, full
	// applies). A full rebuild through this unserialized path leaves the
	// debt outstanding — fail-closed; run `setup rebuild` to settle.
	return rebuildDerived(ctx, r, false, false)
}

// requireDeferFormulaViewPrecondition fail-fasts a DeferFormulaViewRebuild
// apply at ENTRY (read-only, before MarkRunning/ApplyMetadata/any mutation)
// unless every object that requires a formula view has one. Under defer the
// apply never creates or replaces views, so a missing view can never be
// healed by this apply: pre-release drift must be healed by a standard apply
// FIRST (safe here — nothing of the new release has been applied yet), and a
// release that introduces formula objects is not supported under defer (ship
// the objects in a prior regular release; roll-forward-only operations).
//
// Restart safety is by construction: the check is read-only and idempotent,
// so a defer apply that fails anywhere and retries behaves exactly like a
// first attempt — no snapshot or cross-attempt provenance exists to go
// stale (the classification approach this replaces could dead-end retries
// when a failed attempt had already persisted release-introduced formula
// metadata).
// formulaViewRebuildOwedStateKey is the durable write-ahead receipt a
// DeferFormulaViewRebuild apply stamps at ENTRY, after the read-only
// precondition passes and before the running marker or any mutation: from
// the moment this apply may change anything, every formula view is
// intentionally stale until a full rebuild, and neither the running nor the
// completed marker alone can express that debt (a crash mid-apply must leave
// the debt visible, not just a retriable marker). Settlement
// (clearFormulaViewRebuildOwed) happens exclusively inside a settle-enabled
// full view rebuild — the deployment's activation-window `setup rebuild`
// (RebuildDerivedWithLock*) or a full apply. Both receipt mutators are
// package-private and every call site holds the control-plane apply lock,
// so no clear can interleave between a defer apply's stamps and its
// completed marker (the unserialized exported RebuildDerived deliberately
// does not settle — fail-closed). Monitoring and subsequent applies read it
// (FormulaViewRebuildOwed) to detect a cutover that died before its rebuild.
const formulaViewRebuildOwedStateKey = "formula_view_rebuild_owed"

// markFormulaViewRebuildOwed durably records, before any mutation of a defer
// apply, that formula views will remain on the previous compiler generation
// until a full rebuild settles the debt. Package-private: callers must hold
// the control-plane apply lock (receipt mutations are lock-serialized).
// Idempotent: a retry of the same release re-stamps the same pending state.
func markFormulaViewRebuildOwed(ctx context.Context, db *sql.DB) error {
	if _, err := db.ExecContext(ctx,
		"INSERT INTO `_System_BootstrapState` (`StateKey`,`RunID`,`Phase`,`Status`,`LastError`,`CreatedDate`,`LastModifiedDate`) "+
			"VALUES (?,NOW(),'owed','pending','',NOW(),NOW()) "+
			"ON DUPLICATE KEY UPDATE `Phase`=VALUES(`Phase`),`Status`=VALUES(`Status`),`LastModifiedDate`=VALUES(`LastModifiedDate`)",
		formulaViewRebuildOwedStateKey,
	); err != nil {
		return fmt.Errorf("mark formula view rebuild owed: %w", err)
	}
	return nil
}

// clearFormulaViewRebuildOwed settles the receipt. Called only by
// settle-enabled full view rebuilds (rebuildDerived with settleRebuildDebt —
// locked `setup rebuild` and full applies). Package-private: callers must
// hold the control-plane apply lock; there is deliberately no standalone
// clear entry point, so no actor can erase the debt without actually
// rebuilding under serialization. Keeps the row for forensics.
func clearFormulaViewRebuildOwed(ctx context.Context, db *sql.DB) error {
	if _, err := db.ExecContext(ctx,
		"UPDATE `_System_BootstrapState` SET `Status`='cleared', `LastModifiedDate`=NOW() WHERE `StateKey` = ?",
		formulaViewRebuildOwedStateKey,
	); err != nil {
		return fmt.Errorf("clear formula view rebuild owed: %w", err)
	}
	return nil
}

// FormulaViewRebuildOwed reports whether a defer apply's rebuild debt is
// still outstanding.
func FormulaViewRebuildOwed(ctx context.Context, db *sql.DB) (bool, error) {
	var tableExists int
	if err := db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM information_schema.TABLES
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_BootstrapState'`,
	).Scan(&tableExists); err != nil {
		return false, fmt.Errorf("read formula view rebuild owed: %w", err)
	}
	if tableExists == 0 {
		return false, nil
	}
	var status string
	err := db.QueryRowContext(ctx,
		"SELECT `Status` FROM `_System_BootstrapState` WHERE `StateKey` = ?",
		formulaViewRebuildOwedStateKey,
	).Scan(&status)
	switch {
	case err == sql.ErrNoRows:
		return false, nil
	case err != nil:
		return false, fmt.Errorf("read formula view rebuild owed: %w", err)
	}
	return status == "pending", nil
}

func requireDeferFormulaViewPrecondition(ctx context.Context, db *sql.DB) error {
	var systemFieldExists int
	if err := db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM information_schema.TABLES
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_Field'`,
	).Scan(&systemFieldExists); err != nil {
		return fmt.Errorf("defer formula view precondition: %w", err)
	}
	if systemFieldExists == 0 {
		return fmt.Errorf("DeferFormulaViewRebuild requires an already-bootstrapped database; run a standard apply first (greenfield has no old-semantics serving to protect)")
	}
	// The write-ahead rebuild-owed receipt is stamped into
	// `_System_BootstrapState` immediately after this precondition, BEFORE
	// the running marker that would otherwise create the table. A database
	// that ever completed an apply always has it; requiring it here keeps the
	// receipt helper free of duplicated DDL and turns the raw INSERT failure
	// into a guided fail-fast.
	var bootstrapStateExists int
	if err := db.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM information_schema.TABLES
		  WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_BootstrapState'`,
	).Scan(&bootstrapStateExists); err != nil {
		return fmt.Errorf("defer formula view precondition: %w", err)
	}
	if bootstrapStateExists == 0 {
		return fmt.Errorf("DeferFormulaViewRebuild requires a database that completed a prior apply (`_System_BootstrapState` missing — the rebuild-owed receipt has nowhere durable to live); run a standard apply first")
	}
	formulaRows, err := queryFormulaStorageRows(ctx, db, "")
	if err != nil {
		return fmt.Errorf("defer formula view precondition: %w", err)
	}
	seen := make(map[string]bool)
	for _, row := range formulaRows {
		resolved, err := formulastorage.Resolve(formulastorage.Spec{
			FormulaMode:         row.formulaMode,
			FormulaStorageMode:  row.formulaStorageMode,
			Formula:             row.formulaExpr,
			HasProjectionConfig: strings.TrimSpace(row.projectionConfig) != "",
		})
		if err != nil {
			return fmt.Errorf("defer formula view precondition: resolve formula storage for %s: %w", row.objectAPIName, err)
		}
		if !resolved.RequiresFormulaView() || seen[row.objectAPIName] {
			continue
		}
		seen[row.objectAPIName] = true
		viewName := analytics.ViewName(row.objectAPIName)
		exists, existsErr := formulaViewExists(ctx, db, viewName)
		if existsErr != nil {
			return fmt.Errorf("defer formula view precondition: %w", existsErr)
		}
		if !exists {
			return fmt.Errorf(
				"DeferFormulaViewRebuild precondition failed: object %s requires formula view %s but it does not exist; "+
					"either heal the inventory with a standard apply (quiesce serving first if release metadata was already partially applied), "+
					"or this release wrongly introduces an active-formula object under defer — "+
					"cutover releases must not introduce such objects (ship them in a prior regular release)",
				row.objectAPIName, viewName)
		}
		// Existence is not enough: the canonical drift shape is a view that
		// exists in the catalog but is unreadable (dangling SQL SECURITY
		// DEFINER after restore/rotation). Under defer this apply can never
		// heal it, and letting it pass would break the entry-gate guarantee —
		// the failure would surface only after metadata mutation.
		if usableErr := verifyFormulaViewUsable(ctx, db, viewName); usableErr != nil {
			return fmt.Errorf(
				"DeferFormulaViewRebuild precondition failed: formula view %s for object %s exists but is not readable (%v); "+
					"heal the inventory with a standard apply (quiesce serving first if needed) before applying this release",
				viewName, row.objectAPIName, usableErr)
		}
	}
	return nil
}

// rebuildDerived is RebuildDerived with the ApplyAll cutover option applied.
// deferFormulaViews skips ONLY the formula-view CREATE OR REPLACE pass; see
// InstallPackOptions.DeferFormulaViewRebuild for the deployment contract.
// settleRebuildDebt clears the formula_view_rebuild_owed receipt after a
// successful full view rebuild — callers may pass true ONLY while holding
// the control-plane apply lock: every receipt mutation is serialized on that
// lock, which is what makes an interleaved clear between a defer apply's
// stamps and its completed marker impossible.
func rebuildDerived(ctx context.Context, r *Runner, deferFormulaViews bool, settleRebuildDebt bool) error {
	log.Println("═══ RebuildDerived ═══")
	start := time.Now()

	if err := infraPersistence.RebuildGroupMembershipClosureWithLock(ctx, r.DB.DB()); err != nil {
		return fmt.Errorf("group membership closure rebuild: %w", err)
	}

	// Rebuild process-local installed-pack runtime metadata from
	// _System_InstalledPack before compiling formula views. Active-formula SQL
	// lowering reads formula-function bundles from this runtime metadata.
	if err := packs.RegisterInstalledRuntimeMetadataWithComponentKindRegistrationSet(ctx, r.DB.DB(), r.componentKindRegistrationSet()); err != nil {
		return fmt.Errorf("installed runtime metadata rebuild: %w", err)
	}

	if r.SvcMgr.FormulaView() != nil {
		if err := r.SvcMgr.FormulaView().RefreshRuntimeFunctionRegistry(); err != nil {
			return fmt.Errorf("formula view function registry refresh: %w", err)
		}
		if err := r.SvcMgr.FormulaView().RefreshViewCache(ctx); err != nil {
			log.Printf("   ⚠️  Failed to refresh formula view cache: %v", err)
		}
		// Always rebuild (not just "create if missing") so that existing-but-broken
		// views — e.g. views carried over from a TiDB Serverless branch whose DEFINER
		// user no longer exists on the new cluster — are healed on every apply.
		// RebuildView uses CREATE OR REPLACE and is idempotent.
		if deferFormulaViews {
			// Defer mode NEVER creates or replaces a view: replacement would
			// swap shared compiler output under a binary that may still hold
			// old-semantics serving authority, and creation is excluded by
			// the entry precondition (requireDeferFormulaViewPrecondition):
			// every required view existed before this apply, so a missing
			// view here means this release introduced an active-formula
			// object — unsupported under defer. Assert and fail closed; the
			// deployment rebuilds every view through the target runtime
			// before activating traffic.
			log.Println("   ⏭️  formula view rebuild deferred (DeferFormulaViewRebuild); asserting required views exist")
			for _, schema := range r.SvcMgr.Metadata().GetSchemas(ctx) {
				if !formulastorage.HasActiveComputedFields(schema) {
					continue
				}
				viewName := analytics.ViewName(schema.APIName)
				exists, existsErr := formulaViewExists(ctx, r.DB.DB(), viewName)
				if existsErr != nil {
					return existsErr
				}
				if !exists {
					return fmt.Errorf(
						"formula view %s for %s is missing under DeferFormulaViewRebuild; cutover releases must not introduce active-formula objects — ship the object in a prior regular release (roll forward)",
						viewName, schema.APIName)
				}
			}
		} else {
			if err := r.SvcMgr.FormulaView().RebuildAllViews(ctx); err != nil {
				return fmt.Errorf("formula view rebuild: %w", err)
			}
			// A successful FULL view rebuild settles any outstanding
			// deferred-apply debt by definition — every view is now the
			// running binary's compiler output. This is what makes the
			// receipt lifecycle closed: defer apply marks it, the
			// deployment's activation `setup rebuild` (or any later normal
			// apply) clears it, and auto-start refuses to serve while it is
			// outstanding. Settlement only on lock-holding paths: the
			// unserialized exported RebuildDerived keeps the debt.
			if settleRebuildDebt {
				if err := clearFormulaViewRebuildOwed(ctx, r.DB.DB()); err != nil {
					return err
				}
			}
		}
	}

	if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, r.DB, r.SvcMgr.Metadata(), "rebuild-derived"); err != nil {
		return fmt.Errorf("flow runtime maintenance rebuild: %w", err)
	}

	if err := r.RefreshCache(ctx); err != nil {
		return fmt.Errorf("final cache refresh: %w", err)
	}

	// The final RefreshCache above invalidates the process-local validation
	// projection cache (via the metadata cache-invalidation hook). Rehydrate it
	// from the freshly refreshed metadata so greenfield apply ends with the
	// narrowed validation path active instead of falling back to all active
	// computed fields on every save until a later manual view rebuild.
	if r.SvcMgr.FormulaView() != nil {
		if err := r.SvcMgr.FormulaView().RecomputeAllValidationProjections(ctx); err != nil {
			log.Printf("   ⚠️  Failed to recompute validation projections: %v", err)
		}
	}

	log.Printf("═══ RebuildDerived completed (%.1fs) ═══", time.Since(start).Seconds())
	return nil
}

// RebuildDerivedL0 rebuilds derived state only for objects in the L0 oracle.
// It avoids interpreting app metadata even when a reused database contains
// app rows that a future verifier surface might not yet recognize.
func RebuildDerivedL0(ctx context.Context, r *Runner) error {
	log.Println("═══ RebuildDerivedL0 ═══")
	start := time.Now()

	if err := infraPersistence.RebuildGroupMembershipClosureWithLock(ctx, r.DB.DB()); err != nil {
		return fmt.Errorf("group membership closure rebuild: %w", err)
	}

	allowedObjects, err := AllowedL0ObjectAPINames(ctx, r.DB.DB())
	if err != nil {
		return fmt.Errorf("allowed L0 object oracle: %w", err)
	}

	if r.SvcMgr.FormulaView() != nil {
		if err := r.SvcMgr.FormulaView().RefreshViewCache(ctx); err != nil {
			log.Printf("   ⚠️  Failed to refresh formula view cache: %v", err)
		}
		// Collect L0-allowed objects that have active computed fields.
		var l0Candidates []string
		for _, schema := range r.SvcMgr.Metadata().GetSchemas(ctx) {
			if _, ok := allowedObjects[strings.ToLower(schema.APIName)]; !ok {
				continue
			}
			// Only include objects that actually need a formula view.
			if formulastorage.HasActiveComputedFields(schema) {
				l0Candidates = append(l0Candidates, schema.APIName)
			}
		}
		// Order by live-metadata dependency graph. Fails closed on cycles or missing
		// metadata — no lexical / pack-payload fallback per design Forbidden Move #5/#11.
		ordered, err := r.SvcMgr.FormulaView().OrderForBuild(ctx, l0Candidates)
		if err != nil {
			return fmt.Errorf("compute formula view build order for L0: %w", err)
		}
		for _, name := range ordered {
			if err := r.SvcMgr.FormulaView().RebuildView(ctx, name); err != nil {
				return fmt.Errorf("formula view rebuild for %s: %w", name, err)
			}
		}
	}

	if err := r.RefreshCache(ctx); err != nil {
		return fmt.Errorf("final L0 cache refresh: %w", err)
	}

	log.Printf("═══ RebuildDerivedL0 completed (%.1fs) ═══", time.Since(start).Seconds())
	return nil
}

type controlPlaneApplyLockScope struct {
	identity *controlPlaneApplyLockScopeIdentity
	witness  controlPlaneApplyLockScopeWitness
}

type controlPlaneApplyLockScopeIdentity struct {
	active       atomic.Bool
	minted       bool
	leaseContext context.Context
	cancelLease  context.CancelCauseFunc
}

type controlPlaneApplyLockScopeWitness struct {
	identity *controlPlaneApplyLockScopeIdentity
}

const (
	// MySQL advisory locks are scoped to one server session. Long metadata
	// operations use other pool connections, so keep this dedicated lock
	// session active across infrastructure idle timeouts. The ping deadline is
	// a lock-loss boundary, not a query-latency SLO: canceling it makes the MySQL
	// driver close the session and therefore releases the advisory lock. Leave
	// enough room for transient TiDB scheduling stalls during heavy metadata
	// convergence while keeping loss detection bounded.
	controlPlaneApplyLockKeepaliveInterval = 15 * time.Second
	controlPlaneApplyLockKeepaliveTimeout  = 30 * time.Second
	controlPlaneApplyLockReleaseTimeout    = 5 * time.Second
)

func requireControlPlaneApplyLockScope(scope controlPlaneApplyLockScope) error {
	if scope.identity == nil || !scope.identity.minted ||
		scope.witness.identity != scope.identity {
		return fmt.Errorf("control-plane apply lock scope was not minted after lock acquisition")
	}
	if !scope.identity.active.Load() {
		return fmt.Errorf("control-plane apply lock scope is no longer active")
	}
	return nil
}

func withControlPlaneApplyLockScope(
	ctx context.Context,
	rawDB *sql.DB,
	operation string,
	fn func(controlPlaneApplyLockScope) error,
) error {
	return withControlPlaneApplyLockScopeWithLeaseTiming(
		ctx,
		rawDB,
		operation,
		controlPlaneApplyLockKeepaliveInterval,
		controlPlaneApplyLockKeepaliveTimeout,
		fn,
	)
}

func withControlPlaneApplyLockScopeWithLeaseTiming(
	ctx context.Context,
	rawDB *sql.DB,
	operation string,
	keepaliveInterval time.Duration,
	keepaliveTimeout time.Duration,
	fn func(controlPlaneApplyLockScope) error,
) (resultErr error) {
	if rawDB == nil {
		return fmt.Errorf("control-plane apply lock database is unavailable")
	}
	if fn == nil {
		return fmt.Errorf("control-plane apply lock operation is unavailable")
	}
	if keepaliveInterval <= 0 {
		return fmt.Errorf("control-plane apply lock keepalive interval must be positive")
	}
	if keepaliveTimeout <= 0 {
		return fmt.Errorf("control-plane apply lock keepalive timeout must be positive")
	}
	lockConn, err := rawDB.Conn(ctx)
	if err != nil {
		return fmt.Errorf("acquiring connection for apply lock: %w", err)
	}

	var lockResult int
	if err := lockConn.QueryRowContext(ctx,
		"SELECT GET_LOCK('nexuscrm_controlplane_apply', 600)",
	).Scan(&lockResult); err != nil {
		_ = lockConn.Close()
		return fmt.Errorf("acquiring control-plane apply lock: %w", err)
	}
	if lockResult != 1 {
		_ = lockConn.Close()
		return fmt.Errorf("acquiring control-plane apply lock: result=%d", lockResult)
	}
	leaseContext, cancelLease := context.WithCancelCause(ctx)
	identity := &controlPlaneApplyLockScopeIdentity{
		minted:       true,
		leaseContext: leaseContext,
		cancelLease:  cancelLease,
	}
	identity.active.Store(true)
	scope := controlPlaneApplyLockScope{
		identity: identity,
		witness: controlPlaneApplyLockScopeWitness{
			identity: identity,
		},
	}
	keepaliveDone := make(chan struct{})
	keepaliveErr := make(chan error, 1)
	go func() {
		ticker := time.NewTicker(keepaliveInterval)
		defer ticker.Stop()
		for {
			select {
			case <-keepaliveDone:
				keepaliveErr <- nil
				return
			case <-ticker.C:
				pingCtx, cancelPing := context.WithTimeout(context.Background(), keepaliveTimeout)
				var pingResult int
				err := lockConn.QueryRowContext(pingCtx, "SELECT 1").Scan(&pingResult)
				cancelPing()
				if err != nil || pingResult != 1 {
					identity.active.Store(false)
					if err == nil {
						err = fmt.Errorf("unexpected result=%d", pingResult)
					}
					leaseErr := fmt.Errorf("control-plane apply lock keepalive failed during %s: %w", operation, err)
					identity.cancelLease(leaseErr)
					keepaliveErr <- leaseErr
					return
				}
			}
		}
	}()
	defer func() {
		close(keepaliveDone)
		leaseErr := <-keepaliveErr
		identity.cancelLease(leaseErr)
		identity.active.Store(false)
		releaseCtx, cancelRelease := context.WithTimeout(context.Background(), controlPlaneApplyLockReleaseTimeout)
		var releaseResult sql.NullInt64
		releaseErr := lockConn.QueryRowContext(
			releaseCtx,
			"SELECT RELEASE_LOCK('nexuscrm_controlplane_apply')",
		).Scan(&releaseResult)
		cancelRelease()
		if releaseErr != nil {
			releaseErr = fmt.Errorf("releasing control-plane apply lock after %s: %w", operation, releaseErr)
		} else if !releaseResult.Valid || releaseResult.Int64 != 1 {
			releaseErr = fmt.Errorf(
				"releasing control-plane apply lock after %s: result=%v",
				operation,
				releaseResult,
			)
		}
		closeErr := lockConn.Close()
		if closeErr != nil {
			closeErr = fmt.Errorf("closing control-plane apply lock connection after %s: %w", operation, closeErr)
		}
		resultErr = errors.Join(resultErr, leaseErr, releaseErr, closeErr)
	}()
	log.Printf("🔒 Acquired control-plane apply lock for %s", operation)

	return fn(scope)
}

// WithControlPlaneApplyLockContext executes fn while holding the same
// session-scoped advisory lock used by control-plane apply operations. The lock
// is acquired on a dedicated database connection, kept alive for the callback's
// duration, and released before this function returns. If the dedicated lease
// keepalive fails, the callback context is canceled immediately with that exact
// failure as its cause so a long external mutation cannot continue unfenced.
// The operation labels lock diagnostics.
func WithControlPlaneApplyLockContext(
	ctx context.Context,
	rawDB *sql.DB,
	operation string,
	fn func(context.Context) error,
) error {
	if fn == nil {
		return fmt.Errorf("control-plane apply lock operation is unavailable")
	}
	return withControlPlaneApplyLockScope(
		ctx, rawDB, operation,
		func(scope controlPlaneApplyLockScope) error {
			if err := requireControlPlaneApplyLockScope(scope); err != nil {
				return err
			}
			return fn(scope.identity.leaseContext)
		},
	)
}

// WithControlPlaneApplyLock is the context-free compatibility wrapper for
// internal apply operations. External long-running mutations should use
// WithControlPlaneApplyLockContext so lease loss cancels their database work.
func WithControlPlaneApplyLock(ctx context.Context, rawDB *sql.DB, operation string, fn func() error) error {
	if fn == nil {
		return fmt.Errorf("control-plane apply lock operation is unavailable")
	}
	return WithControlPlaneApplyLockContext(
		ctx,
		rawDB,
		operation,
		func(context.Context) error { return fn() },
	)
}

func withControlPlaneMutationLock(ctx context.Context, rawDB *sql.DB, operation string, fn func() error) error {
	return WithControlPlaneApplyLock(ctx, rawDB, operation, func() error {
		if err := bootstrapStateStore.MarkControlPlaneApplyAllInvalidated(rawDB, operation); err != nil {
			return fmt.Errorf("invalidate ApplyAll completion marker before %s: %w", operation, err)
		}
		log.Printf("controlplane: invalidated full ApplyAll marker before %s; next auto restart must run ApplyAll", operation)
		return fn()
	})
}

func ApplyMetadataWithLock(ctx context.Context, db *database.TiDBConnection) (*database.TiDBConnection, error) {
	var newDB *database.TiDBConnection
	err := withControlPlaneMutationLock(ctx, db.DB(), "ApplyMetadata", func() error {
		var err error
		newDB, err = ApplyMetadata(ctx, db)
		return err
	})
	return newDB, err
}

func RunRepairWithLock(ctx context.Context, db *database.TiDBConnection) error {
	return RunRepairWithLockAndRunnerOptions(ctx, db, RunnerOptions{})
}

// RunRepairWithLockAndRunnerOptions executes historical repairs under the
// control-plane mutation lock with one immutable distribution-owned snapshot.
func RunRepairWithLockAndRunnerOptions(ctx context.Context, db *database.TiDBConnection, runnerOptions RunnerOptions) error {
	return withControlPlaneMutationLock(ctx, db.DB(), "RunRepair", func() error {
		return RunRepairWithOptionsAndRunnerOptions(ctx, db, RepairOptions{}, runnerOptions)
	})
}

func EnsureIndexesWithLock(ctx context.Context, db *database.TiDBConnection) error {
	return withControlPlaneMutationLock(ctx, db.DB(), "EnsureIndexes", func() error {
		return EnsureIndexes(ctx, db)
	})
}

func RebuildDerivedWithLock(ctx context.Context, db *database.TiDBConnection) error {
	return RebuildDerivedWithLockAndRunnerOptions(ctx, db, RunnerOptions{})
}

// RebuildDerivedWithLockAndRunnerOptions rebuilds derived state under the
// control-plane mutation lock with one immutable distribution-owned runner
// option snapshot.
func RebuildDerivedWithLockAndRunnerOptions(ctx context.Context, db *database.TiDBConnection, runnerOptions RunnerOptions) error {
	return withControlPlaneMutationLock(ctx, db.DB(), "RebuildDerived", func() error {
		r, err := NewRunnerWithOptions(db, runnerOptions)
		if err != nil {
			return fmt.Errorf("runner init: %w", err)
		}
		if err := r.RefreshCache(ctx); err != nil {
			return fmt.Errorf("initial cache refresh: %w", err)
		}
		// Directly enable settlement: this closure holds the control-plane
		// apply lock, so clearing the rebuild-owed receipt here is
		// serialized with any defer apply's stamp/completed sequence. This
		// is the deployment's designated activation-window settlement path.
		return rebuildDerived(ctx, r, false, true)
	})
}

// ApplyAll runs the complete control-plane sequence:
// metadata → candidate planning → repair → runner init → packs → authorized source advances →
// rebuild → indexes → verify.
// Holds a MySQL advisory lock for the entire duration to prevent concurrent
// multi-pod executions from racing on pack apply.
func ApplyAll(ctx context.Context, db *database.TiDBConnection) error {
	return ApplyAllWithOptions(ctx, db, InstallPackOptions{})
}

func ApplyAllWithOptions(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions) error {
	return ApplyAllWithOptionsAndRunnerOptions(ctx, db, options, RunnerOptions{})
}

// ApplyAllWithOptionsAndRunnerOptions runs the complete control-plane sequence
// with one immutable distribution-owned runner option snapshot.
func ApplyAllWithOptionsAndRunnerOptions(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions, runnerOptions RunnerOptions) error {
	_, err := applyAllWithOptions(ctx, db, options, runnerOptions, applyAllCompletionPolicyAlwaysApply)
	return err
}

// ApplyAllIfNeededWithOptions is the container auto-start variant of ApplyAll:
// it acquires the same control-plane lock, classifies bootstrap state, then
// consults the distribution-owned source-revision plan before trusting a
// completed marker. It either skips, applies, or fails closed for legacy
// pre-marker databases. This is the authoritative startup decision; shell probes
// are advisory and must not be trusted to distinguish "false" from "unknown".
// Explicit operator ApplyAll calls intentionally do not use this path and always
// run the full sequence.
func ApplyAllIfNeededWithOptions(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions) (bool, error) {
	return ApplyAllIfNeededWithOptionsAndRunnerOptions(ctx, db, options, RunnerOptions{})
}

// ApplyAllIfNeededWithOptionsAndRunnerOptions is the container auto-start
// variant with one immutable distribution-owned runner option snapshot.
func ApplyAllIfNeededWithOptionsAndRunnerOptions(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions, runnerOptions RunnerOptions) (bool, error) {
	return applyAllWithOptions(ctx, db, options, runnerOptions, applyAllCompletionPolicyAutoStart)
}

type applyAllCompletionPolicy uint8

const (
	applyAllCompletionPolicyAlwaysApply applyAllCompletionPolicy = iota
	applyAllCompletionPolicyAutoStart
	applyAllCompletionPolicyExactCandidateRetry
)

// recoverExactOfflineCandidateFinalization closes only the narrow crash window
// after a full ApplyAll passed every terminal gate but before its session-scoped
// advisory-lock connection could authorize the completed marker. The durable
// marker is necessary but never sufficient: recovery reacquires the lock, proves
// the exact immutable target, repeats the complete readiness/distribution verify,
// and re-runs every completion gate. Any drift falls back to normal ApplyAll.
func recoverExactOfflineCandidateFinalization(
	ctx context.Context,
	db *database.TiDBConnection,
	options InstallPackOptions,
	runnerOptions RunnerOptions,
	applyLock controlPlaneApplyLockScope,
) (bool, error) {
	execution := options.offlineCandidate
	if execution == nil || execution.verify == nil {
		return false, nil
	}
	// A finalization-ready receipt may predate migrations added after the full
	// ApplyAll that produced it. Converge stamped metadata before any recovery
	// cache refresh or verification can authorize serving that receipt.
	if _, err := MigrateMasterDetailRequiredBackfill(ctx, db.DB()); err != nil {
		return false, fmt.Errorf("converge master-detail required metadata before finalization recovery: %w", err)
	}
	matches, _, err := offlineCandidateMatchesActivatedState(ctx, db.DB(), execution)
	if err != nil {
		return false, fmt.Errorf("classify finalization-recovery candidate: %w", err)
	}
	if !matches {
		return false, nil
	}
	status, err := PendingApplyAllSourceRevisionAdvances(ctx, db.DB(), options)
	if err != nil {
		return false, fmt.Errorf("classify finalization-recovery source revisions: %w", err)
	}
	if status.HasPending() {
		return false, nil
	}
	owed, err := FormulaViewRebuildOwed(ctx, db.DB())
	if err != nil {
		return false, err
	}
	if owed && !options.DeferFormulaViewRebuild {
		return false, nil
	}
	if err := runLoggedControlPlanePhase("ApplyAllRecovery", "demote-stale-connector-route-incarnations", func() error {
		return demoteStaleConnectorInboundRouteIncarnations(ctx, db.DB())
	}); err != nil {
		return false, err
	}
	if err := requireConnectorPackDependencyInstallIDsConverged(ctx, db.DB()); err != nil {
		return false, fmt.Errorf("require connector dependency install identities before finalization recovery: %w", err)
	}
	if err := DropConnectorEventLegacyRetirementIndex(ctx, db.DB()); err != nil {
		return false, fmt.Errorf("converge retired ConnectorEvent index before finalization recovery: %w", err)
	}
	if err := VerifyConnectorExecutionProtocols(ctx, db.DB()); err != nil {
		return false, fmt.Errorf("verify WorkItem-only ConnectorEvent precondition before finalization recovery: %w", err)
	}

	candidatePacks := execution.candidate.packList()
	if err := registerExactCandidateRuntimeMetadata(
		ctx,
		db.DB(),
		candidatePacks,
		nil,
		runnerOptions.ComponentKindRegistrationSet,
	); err != nil {
		return false, fmt.Errorf("prepare finalization-recovery runtime metadata: %w", err)
	}
	runner, err := NewRunnerWithOptions(db, runnerOptions)
	if err != nil {
		return false, fmt.Errorf("construct finalization-recovery runner: %w", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		return false, fmt.Errorf("refresh finalization-recovery runner: %w", err)
	}
	if err := runLoggedControlPlanePhase("ApplyAllRecovery", "verify", func() error {
		if err := VerifyWithRunnerOptions(ctx, db, runnerOptions); err != nil {
			return fmt.Errorf("verify: %w", err)
		}
		if err := verifyOfflineCandidateFinalState(ctx, runner, execution); err != nil {
			return fmt.Errorf("verify exact offline candidate: %w", err)
		}
		if options.ApplyAllFinalVerify != nil {
			if err := options.ApplyAllFinalVerify(ctx, db); err != nil {
				return fmt.Errorf("final distribution verify: %w", err)
			}
		}
		return nil
	}); err != nil {
		return false, err
	}
	if err := runApplyAllCompletionGates(ctx, db.DB(), options); err != nil {
		return false, err
	}
	if options.DeferFormulaViewRebuild {
		if err := markFormulaViewRebuildOwed(ctx, db.DB()); err != nil {
			return false, err
		}
	}
	if err := bootstrapStateStore.MarkControlPlaneApplyAllFinalizationReady(db.DB()); err != nil {
		return false, fmt.Errorf("refresh ApplyAll finalization-ready receipt: %w", err)
	}
	if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
		return false, fmt.Errorf("control-plane apply lock recovery final gate: %w", err)
	}
	if err := bootstrapStateStore.MarkControlPlaneApplyAllCompleted(db.DB()); err != nil {
		return false, fmt.Errorf("mark recovered ApplyAll completed: %w", err)
	}
	log.Printf("Control Plane: exact candidate terminal verification recovered without replaying ApplyAll")
	return true, nil
}

func classifyStandardActionPersistenceForFinalizationRecovery(db *sql.DB, options *InstallPackOptions) (bool, error) {
	current, err := packs.StandardActionsPersistenceSemanticsApplied(db)
	if err != nil {
		return false, fmt.Errorf("classify finalization-recovery standard-action persistence semantics: %w", err)
	}
	options.replayStandardActionPersistenceSemantics = !current
	return current, nil
}

func runApplyAllCompletionGates(ctx context.Context, markerDB *sql.DB, options InstallPackOptions) error {
	if err := runLoggedControlPlanePhase("ApplyAll", "connector-event-enqueue-ordinal-final-gate", func() error {
		completed, err := bootstrapStateStore.IsConnectorEventEnqueueOrdinalBackfillCompleted(markerDB)
		if err != nil {
			return fmt.Errorf("connector event enqueue ordinal final gate: %w", err)
		}
		if !completed {
			return fmt.Errorf("connector event enqueue ordinal backfill is not complete; rerun setup apply after old connector event writers are stopped")
		}
		return nil
	}); err != nil {
		return err
	}
	if err := runLoggedControlPlanePhase("ApplyAll", "connector-event-source-lane-final-gate", func() error {
		completed, err := bootstrapStateStore.IsConnectorEventSourceLaneBackfillCompleted(markerDB)
		if err != nil {
			return fmt.Errorf("connector event source-lane final gate: %w", err)
		}
		if !completed {
			return fmt.Errorf("connector event source-lane backfill is not complete; rerun setup apply after old connector event writers are stopped")
		}
		return nil
	}); err != nil {
		return err
	}
	if err := runLoggedControlPlanePhase("ApplyAll", "listview-legacy-cutover-final-gate", func() error {
		completed, err := bootstrapStateStore.IsListViewLegacyCutoverCompleted(markerDB)
		if err != nil {
			return fmt.Errorf("list-view legacy cutover final gate: %w", err)
		}
		if !completed {
			return fmt.Errorf("list-view legacy cutover is not complete; rerun setup apply after resolving _System_ListView OwnerId drift")
		}
		return nil
	}); err != nil {
		return err
	}
	return runLoggedControlPlanePhase("ApplyAll", "source-revision-advance-final-gate", func() error {
		if err := requireApplyAllSourceRevisionAdvancePlanCompleted(ctx, markerDB, options); err != nil {
			return fmt.Errorf("source revision advance final gate: %w", err)
		}
		return nil
	})
}

func applyAllWithOptions(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions, runnerOptions RunnerOptions, completionPolicy applyAllCompletionPolicy) (bool, error) {
	if completionPolicy == applyAllCompletionPolicyExactCandidateRetry && options.offlineCandidate == nil {
		return false, fmt.Errorf("exact-candidate retry policy requires an offline candidate")
	}
	// A resolved offline candidate owns one immutable source-authority snapshot
	// for the complete operation. Bind it before completion classification so
	// exact-active verification, component-delta retry, and full apply cannot
	// diverge merely because one branch returns before another branch's setup.
	ctx = withOfflineCandidateOperationContext(ctx, options.offlineCandidate)
	log.Println("══════════════════════════════════════════════════════════════")
	log.Println("  Control Plane: ApplyAll")
	log.Println("══════════════════════════════════════════════════════════════")
	start := time.Now()
	applied := false

	err := withControlPlaneApplyLockScope(ctx, db.DB(), "ApplyAll", func(applyLock controlPlaneApplyLockScope) error {
		var (
			newDB                          *database.TiDBConnection
			r                              *Runner
			offlineCandidatePrepared       bool
			offlineCandidateSourceDelta    bool
			standardActionSemanticsChecked bool
		)
		if completionPolicy != applyAllCompletionPolicyAlwaysApply {
			// Completion-aware paths decide under the lock. Runtime auto-start
			// trusts only its completed-marker gates and halts legacy pre-marker
			// databases. An explicit offline-candidate retry may additionally skip
			// when its entire immutable target is already active; otherwise it keeps
			// the explicit operator semantics and runs the full ApplyAll migration.
			decision, classifyErr := bootstrapStateStore.ClassifyControlPlaneAutoStart(db.DB())
			if classifyErr != nil {
				return fmt.Errorf("classify control-plane auto-start state: %w", classifyErr)
			}
			readyMarker := decision.Reason == "full ApplyAll marker is final-verified/ready"
			if decision.Action == contractcontrolplane.ControlPlaneAutoStartApply &&
				(completionPolicy == applyAllCompletionPolicyExactCandidateRetry || readyMarker) {
				recoverableFinalization, recoveryErr := bootstrapStateStore.IsControlPlaneApplyAllFinalizationRecoverable(db.DB())
				if recoveryErr != nil {
					return fmt.Errorf("classify control-plane finalization recovery: %w", recoveryErr)
				}
				if recoverableFinalization {
					if completionPolicy != applyAllCompletionPolicyExactCandidateRetry {
						return fmt.Errorf("control-plane ApplyAll reached terminal verification but completion is not committed; finalize it with the exact immutable offline candidate before serving")
					}
					standardActionSemanticsCurrent, semanticsErr := classifyStandardActionPersistenceForFinalizationRecovery(db.DB(), &options)
					if semanticsErr != nil {
						return semanticsErr
					}
					standardActionSemanticsChecked = true
					if standardActionSemanticsCurrent {
						recovered, err := recoverExactOfflineCandidateFinalization(
							ctx, db, options, runnerOptions, applyLock,
						)
						if err != nil {
							return err
						}
						if recovered {
							return nil
						}
					} else {
						log.Printf("Control Plane: finalization-ready marker predates current standard-action persistence semantics; replaying full ApplyAll")
					}
				}
				if readyMarker && !options.replayStandardActionPersistenceSemantics {
					return fmt.Errorf("control-plane ApplyAll finalization receipt no longer matches the exact immutable candidate; a full stopped-Runtime apply is required")
				}
			}
			// ENTRY gate, before ANY auto-start action: an outstanding
			// rebuild debt blocks every arm — skipping would serve stale
			// views, and a mutating retry would either CREATE OR REPLACE
			// views while the previous binary may still serve (defer
			// disabled) or complete again with the debt still pending
			// (defer enabled). Settling the debt is exclusively the
			// deployment's quiesced `setup rebuild`.
			owedAtEntry, owedAtEntryErr := FormulaViewRebuildOwed(ctx, db.DB())
			if owedAtEntryErr != nil {
				return owedAtEntryErr
			}
			if owedAtEntry && completionPolicy == applyAllCompletionPolicyAutoStart {
				return fmt.Errorf("formula view rebuild owed from a deferred apply; auto-start refuses every action while the debt is outstanding — run `setup rebuild` (inside the deployment's quiesced window) to settle it")
			}
			switch decision.Action {
			case contractcontrolplane.ControlPlaneAutoStartSkip:
				standardActionSemanticsCurrent, semanticsErr := packs.StandardActionsPersistenceSemanticsApplied(db.DB())
				if semanticsErr != nil {
					return fmt.Errorf("classify completed-marker standard-action persistence semantics: %w", semanticsErr)
				}
				standardActionSemanticsChecked = true
				options.replayStandardActionPersistenceSemantics = !standardActionSemanticsCurrent
				status, planErr := PendingApplyAllSourceRevisionAdvances(ctx, db.DB(), options)
				if planErr != nil {
					return fmt.Errorf("classify completed-marker source revision plan: %w", planErr)
				}
				requiresApply := status.HasPending() || owedAtEntry || !standardActionSemanticsCurrent
				if !standardActionSemanticsCurrent {
					log.Printf("Control Plane: completed marker predates current standard-action persistence semantics; promoting startup to ApplyAll")
				}
				if status.HasPending() {
					log.Printf("Control Plane: completed marker has %d pending authorized source-revision operation(s); promoting startup to ApplyAll", len(status.Pending))
				}
				if owedAtEntry {
					log.Printf("Control Plane: completed marker has outstanding formula-view rebuild debt; explicit offline candidate is promoting to full ApplyAll")
				}
				if !requiresApply && options.offlineCandidate != nil {
					matches, installedRows, matchErr := offlineCandidateMatchesActivatedState(ctx, db.DB(), options.offlineCandidate)
					if matchErr != nil {
						return fmt.Errorf("classify completed-marker offline candidate: %w", matchErr)
					}
					// Persisted pack commitments prove candidate identity, not live
					// serving state. Skipping apply is safe only when the distribution
					// supplied its whole-target verifier; otherwise retain the existing
					// full ApplyAll convergence path.
					if matches && options.offlineCandidate.verify == nil {
						matches = false
					}
					if !matches {
						// Classify the exact candidate before choosing a broad bootstrap.
						// A pure immutable source-revision transition owns its own
						// transactional lifecycle and does not need schema, historical
						// repair or unchanged platform-pack replay.
						if completionPolicy == applyAllCompletionPolicyExactCandidateRetry && !owedAtEntry {
							if err := prepareOfflineCandidateFromInstalledRowsUnderApplyLock(ctx, db.DB(), &options, installedRows); err != nil {
								return fmt.Errorf("prepare completed-marker offline candidate: %w", err)
							}
							offlineCandidatePrepared = true
							offlineCandidateSourceDelta = options.offlineCandidate.plan.isSourceRevisionDelta() && !options.DeferFormulaViewRebuild
						}
						requiresApply = true
						if offlineCandidateSourceDelta {
							log.Printf("Control Plane: completed marker differs only by immutable source-backed component revisions; selecting component-delta apply")
						} else {
							log.Printf("Control Plane: completed marker does not match the exact offline candidate; promoting startup to ApplyAll")
						}
					}
				}
				if !requiresApply {
					// A completed ApplyAll marker proves only the migrations that existed
					// when it was stamped. Run later stamped, startup-safe convergence
					// operations under the same lock before declaring startup a no-op;
					// otherwise warm upgrades skip RunRepair and may serve stale metadata
					// or retain retired physical schema indefinitely.
					//
					// Source-backed connector packs are not auto-reapplied during this
					// path. Their live dependency identities must already be converged;
					// otherwise fail closed and require a full setup apply rather than
					// rewriting a historical route before the destructive cutover.
					if err := requireConnectorPackDependencyInstallIDsConverged(ctx, db.DB()); err != nil {
						return fmt.Errorf("require connector dependency install identities before completed-marker cutover: %w", err)
					}
					if _, err := MigrateMasterDetailRequiredBackfill(ctx, db.DB()); err != nil {
						return fmt.Errorf("converge master-detail required metadata before completed-marker skip: %w", err)
					}
					if err := DropConnectorEventLegacyRetirementIndex(ctx, db.DB()); err != nil {
						return fmt.Errorf("converge retired ConnectorEvent index before completed-marker skip: %w", err)
					}
					if err := VerifyConnectorExecutionProtocols(ctx, db.DB()); err != nil {
						return fmt.Errorf("verify WorkItem-only ConnectorEvent precondition before completed-marker skip: %w", err)
					}
					if options.offlineCandidate != nil {
						execution := options.offlineCandidate
						if execution.verify != nil {
							candidatePacks := execution.candidate.packList()
							if err := registerExactCandidateRuntimeMetadata(
								ctx,
								db.DB(),
								candidatePacks,
								nil,
								runnerOptions.ComponentKindRegistrationSet,
							); err != nil {
								return fmt.Errorf("prepare exact-candidate runtime metadata: %w", err)
							}
							candidateRunner, err := NewRunnerWithOptions(db, runnerOptions)
							if err != nil {
								return fmt.Errorf("construct exact-candidate verification runner: %w", err)
							}
							if err := candidateRunner.RefreshCache(ctx); err != nil {
								return fmt.Errorf("refresh exact-candidate verification runner: %w", err)
							}
							if err := execution.verify(ctx, candidateRunner, candidatePacks); err != nil {
								if !errors.Is(err, ErrOfflineCandidateLiveStateDrift) {
									return fmt.Errorf("verify exact already-active candidate: %w", err)
								}
								// Matching immutable pack identities prove which sources were
								// activated, not that every derived Catalog projection still
								// matches those sources. A failed whole-target verifier is
								// evidence that the exact candidate needs a full stopped-Runtime
								// reconcile only when the verifier explicitly classifies it as
								// live-state drift. Operational failures remain fail-closed.
								requiresApply = true
								log.Printf("Control Plane: exact already-active candidate failed whole-target verification; promoting startup to ApplyAll: %v", err)
							} else {
								execution.verified = true
							}
						}
					}
					if requiresApply {
						break
					}
					if options.offlineCandidate != nil {
						options.offlineCandidate.receipt.Verified = true
					}
					log.Printf("Control Plane: ApplyAll already completed; completed-marker startup preflight passed (%s)", decision.Reason)
					if options.ApplyAllFinalVerify != nil {
						return runLoggedControlPlanePhase("ApplyAll", "final-distribution-verify", func() error {
							if err := options.ApplyAllFinalVerify(ctx, db); err != nil {
								return fmt.Errorf("final distribution verify: %w", err)
							}
							return nil
						})
					}
					return nil
				}
			case contractcontrolplane.ControlPlaneAutoStartHaltForMigration:
				if completionPolicy == applyAllCompletionPolicyAutoStart {
					return fmt.Errorf("explicit control-plane migration required before auto start: %s; run a one-time `nexuscrm-setup apply` (or NEXUSCRM_ENTRYPOINT_MODE=apply-only/apply-and-start) before using auto start", decision.Reason)
				}
				log.Printf("Control Plane: explicit offline candidate will perform the required migration (%s)", decision.Reason)
			case contractcontrolplane.ControlPlaneAutoStartApply:
				log.Printf("Control Plane: ApplyAll needed; proceeding with apply (%s)", decision.Reason)
			default:
				return fmt.Errorf("unknown control-plane auto-start action %q (%s)", decision.Action, decision.Reason)
			}
		}
		if offlineCandidateSourceDelta {
			applied = true
			return runLoggedControlPlanePhase("OfflineCandidate", "apply-source-revision-component-deltas", func() error {
				return applyOfflineCandidateSourceRevisionDeltaUnderApplyLock(ctx, db, options, runnerOptions, applyLock)
			})
		}
		// Freeze and classify the exact Catalog candidate before the first
		// mutation. ApplyMetadata may legitimately repair persisted runtime
		// declarations from a mutable source locator; allowing that repair to run
		// first would replace the active-generation semantic baseline with the
		// candidate and make a storage transition impossible to prove.
		prepareFormulaCandidateEarly, err := offlineCandidateDeclaresWriteMaterializedFormulaStorage(options.offlineCandidate)
		if err != nil {
			return fmt.Errorf("classify offline candidate formula storage declarations: %w", err)
		}
		if prepareFormulaCandidateEarly && !offlineCandidatePrepared {
			if err := runLoggedControlPlanePhase("ApplyAll", "prepare-offline-candidate", func() error {
				if err := prepareOfflineCandidateUnderApplyLock(ctx, db.DB(), &options); err != nil {
					return fmt.Errorf("prepare offline candidate: %w", err)
				}
				return nil
			}); err != nil {
				return err
			}
			offlineCandidatePrepared = true
		}
		// Fail-fast BEFORE the first mutation (ApplyMetadata runs next): under
		// defer this apply can never heal a missing view, so the inventory
		// must already be healthy at entry. Read-only and idempotent, so a
		// failed defer apply retries exactly like a first attempt.
		if options.DeferFormulaViewRebuild {
			if err := requireDeferFormulaViewPrecondition(ctx, db.DB()); err != nil {
				return err
			}
			// Write-ahead intent: the rebuild-owed receipt is stamped BEFORE
			// the running marker and before ANY mutation. There is therefore
			// no window in which this apply has mutated state but the receipt
			// is absent — a crash before this write leaves the database
			// untouched (the precondition is read-only), and a crash after it
			// leaves the debt outstanding, which the auto-start entry gate
			// refuses to serve past. Settlement is exclusively a
			// lock-holding full view rebuild (`setup rebuild`, a non-defer
			// full apply, or source convergence).
			if err := markFormulaViewRebuildOwed(ctx, db.DB()); err != nil {
				return fmt.Errorf("stamp formula-view rebuild-owed receipt: %w", err)
			}
			log.Println("Control Plane: DeferFormulaViewRebuild active; views verified at entry and rebuild-owed receipt stamped")
		}
		// Every mutating ApplyAll attempt stamps running before doing work and
		// records failed/completed for THAT attempt. We never preserve a prior
		// completed marker across a mutating reapply: a re-apply that fails after
		// partial state changes must leave a failed marker so the next auto restart
		// re-applies, rather than skipping onto partial/stale state.
		if markerErr := bootstrapStateStore.MarkControlPlaneApplyAllRunning(db.DB()); markerErr != nil {
			return fmt.Errorf("mark ApplyAll running: %w", markerErr)
		}
		if !standardActionSemanticsChecked {
			// A full ApplyAll that did not enter through a trusted completed-marker
			// receipt must converge persistence semantics instead of inferring them
			// from unchanged source payload hashes.
			options.replayStandardActionPersistenceSemantics = true
		}
		applied = true

		failApplyAll := func(err error) error {
			if err == nil {
				return nil
			}
			markerDB := db.DB()
			if newDB != nil {
				markerDB = newDB.DB()
			}
			if markErr := bootstrapStateStore.MarkControlPlaneApplyAllFailed(markerDB, err); markErr != nil {
				log.Printf("⚠️ Failed to persist ApplyAll failure marker: %v (original error: %v)", markErr, err)
			}
			return err
		}

		// 1. Substrate + managed metadata
		if err := runLoggedControlPlanePhase("ApplyAll", "apply-metadata", func() error {
			var phaseErr error
			newDB, phaseErr = ApplyMetadata(ctx, db)
			if phaseErr != nil {
				return fmt.Errorf("apply-metadata: %w", phaseErr)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// Candidates without write-materialized formula declarations do not need
		// the pre-mutation semantic snapshot. Keep their established preparation
		// point after schema readiness and before repair.
		if options.offlineCandidate != nil && !offlineCandidatePrepared {
			if err := runLoggedControlPlanePhase("ApplyAll", "prepare-offline-candidate", func() error {
				if err := prepareOfflineCandidateUnderApplyLock(ctx, newDB.DB(), &options); err != nil {
					return fmt.Errorf("prepare offline candidate: %w", err)
				}
				return nil
			}); err != nil {
				return failApplyAll(err)
			}
		}

		// 2. Repair: discovery + backfills
		if err := runLoggedControlPlanePhase("ApplyAll", "run-repair", func() error {
			if err := RunRepairWithOptionsAndRunnerOptions(ctx, newDB, RepairOptions{
				SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
				deferredSourceBackedLegacyBackfills:  options.offlineCandidate.deferredSourceBackedLegacyBackfillPacks(),
				DeferConnectorWorkItemCutover:        true,
				// Marker path: a convergence-critical best-effort repair failure
				// must prevent stamping control_plane_apply_all=completed.
				FailClosedOnConvergenceRepairs: true,
			}, runnerOptions); err != nil {
				return fmt.Errorf("run-repair: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 3. Build headless service graph
		if err := runLoggedControlPlanePhase("ApplyAll", "runner-init-and-initial-cache-refresh", func() error {
			var phaseErr error
			r, phaseErr = NewRunnerWithOptions(newDB, runnerOptions)
			if phaseErr != nil {
				return fmt.Errorf("runner init: %w", phaseErr)
			}
			if phaseErr := r.RefreshCache(ctx); phaseErr != nil {
				return fmt.Errorf("initial cache refresh: %w", phaseErr)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 4. Apply runtime platform packs
		// (include dev-sample in non-production)
		includeDevSample := !isProduction()
		if err := runLoggedControlPlanePhase("ApplyAll", "apply-packs", func() error {
			if err := ApplyPacksWithOptions(ctx, r, includeDevSample, packs.ApplyOptions{
				SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
				RepairLegacyUniqueDuplicates:         options.RepairLegacyUniqueDuplicates,
				DeferFormulaViewRebuild:              options.DeferFormulaViewRebuild,
			}); err != nil {
				return fmt.Errorf("apply-packs: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// Runtime-required packs now provide the complete WorkItem schema, and
		// deployment-owned ingress fencing is already durable. Prove the prior
		// WorkItem handoff without retiring legacy Event authority, then audit
		// drained projection history before any connector incarnation changes.
		if err := runLoggedControlPlanePhase("ApplyAll", "verify-connector-workitem-only-execution-cutover", func() error {
			if err := VerifyConnectorExecutionProtocols(ctx, newDB.DB()); err != nil {
				return fmt.Errorf("verify WorkItem-only ConnectorEvent cutover: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 5. Adopt legacy installed app-pack rows that predate source-backed
		// identity. This is generic source migration: it resolves
		// official_repo:<installed-name>, applies that pack, and writes source/hash
		// metadata before runtime caches are rebuilt from _System_InstalledPack.
		if err := runLoggedControlPlanePhase("ApplyAll", "adopt-legacy-installed-pack-source-identities", func() error {
			if err := adoptLegacyInstalledPackSourceIdentities(ctx, r, options); err != nil {
				return fmt.Errorf("adopt legacy installed pack source identities: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// A candidate containing only materializations retains the established
		// prepare-all boundary. Mixed transition kinds are converged below in the
		// candidate DAG order so every dependent observes its active target.
		if options.offlineCandidate != nil &&
			!options.offlineCandidate.plan.isMixedTransition() {
			if err := runLoggedControlPlanePhase("ApplyAll", "prepare-offline-candidate-materializations", func() error {
				if err := prepareOfflineCandidateMaterializationsUnderApplyLock(ctx, r, &options); err != nil {
					return fmt.Errorf("prepare offline candidate materializations: %w", err)
				}
				return nil
			}); err != nil {
				return failApplyAll(err)
			}
		}

		// 6. Converge only the exact distribution-authorized transitions. Mixed
		// candidates reuse the existing per-pack lifecycle in dependency order;
		// source-only candidates retain the whole-plan revision runner.
		if err := runLoggedControlPlanePhase("ApplyAll", "apply-authorized-source-revision-advances", func() error {
			var err error
			if options.offlineCandidate != nil && options.offlineCandidate.plan.isMixedTransition() {
				err = convergeOfflineCandidateMixedTransitionsUnderApplyLock(ctx, r, &options, applyLock)
			} else {
				err = applyAllSourceRevisionAdvances(ctx, r, options, applyLock)
			}
			if err != nil {
				return fmt.Errorf("apply authorized source revision advances: %w", err)
			}
			return nil
		}); err != nil {
			if options.offlineCandidate != nil {
				err = failOfflineCandidatePreparedMaterializations(ctx, r, options.offlineCandidate, err)
			}
			return failApplyAll(err)
		}
		if err := requireOfflineCandidatePreActivationVerification(options.offlineCandidate); err != nil {
			return failApplyAll(err)
		}

		if options.offlineCandidate != nil &&
			!options.offlineCandidate.plan.isMixedTransition() {
			if err := runLoggedControlPlanePhase("ApplyAll", "finalize-offline-candidate-materializations", func() error {
				if err := finalizeOfflineCandidatePreparedMaterializations(ctx, r, &options, options.offlineCandidate); err != nil {
					return fmt.Errorf("finalize offline candidate materializations: %w", err)
				}
				return nil
			}); err != nil {
				return failApplyAll(err)
			}
		}

		// 7. Admit every source-backed lifecycle row before mutation, resume
		// only exact revision-bound installing/failed attempts, then reconcile
		// active packs from their stored source identity before the final
		// rebuild/verify gates. This keeps Verify strict while making stopped
		// phase-7 attempts restartable without authorizing arbitrary retries.
		if err := runLoggedControlPlanePhase("ApplyAll", "reconcile-active-installed-pack-sources", func() error {
			reconcileOptions := options
			reconcileOptions.allowInstalledPackSourceStateRepair = true
			if err := reconcileActiveInstalledPackSourceState(ctx, r, reconcileOptions); err != nil {
				return fmt.Errorf("reconcile active installed pack sources: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		if err := runLoggedControlPlanePhase("ApplyAll", "refresh-active-installed-runtime-declarations", func() error {
			if err := refreshPersistedRuntimeDeclarationsForActivePacksWithComponentKindRegistrationSet(ctx, newDB.DB(), r.componentKindRegistrationSet()); err != nil {
				return fmt.Errorf("refresh persisted runtime declarations: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		if err := runLoggedControlPlanePhase("ApplyAll", "backfill-connector-push-legacy-mirrors-after-source-convergence", func() error {
			if _, err := BackfillConnectorPushLegacyMirrorAdoptionsWithComponentKindRegistrationSet(ctx, newDB.DB(), r.componentKindRegistrationSet()); err != nil {
				return fmt.Errorf("connector push legacy mirror backfill after source convergence: %w", err)
			}
			if err := bootstrapStateStore.MarkConnectorPushLegacyMirrorBackfillCompleted(newDB.DB()); err != nil {
				return fmt.Errorf("mark connector push legacy mirror backfill completed: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		if err := runLoggedControlPlanePhase("ApplyAll", "backfill-connector-event-source-lanes", func() error {
			if _, err := BackfillConnectorEventSourceLanes(ctx, newDB.DB()); err != nil {
				return fmt.Errorf("connector event source-lane backfill: %w", err)
			}
			if err := bootstrapStateStore.MarkConnectorEventSourceLaneBackfillCompleted(newDB.DB()); err != nil {
				return fmt.Errorf("mark connector event source-lane backfill completed: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// Warm-upgrade CustomSetting hydration + Name-field metadata repair must
		// run only after source-backed identity adoption/reconcile. Earlier
		// repair phases may still carry legacy installed-pack source identity or
		// pre-normalized object-row state; stamping the migration before source
		// convergence would permanently skip the intended closure path.
		if err := runLoggedControlPlanePhase("ApplyAll", "repair-custom-setting-name-metadata-after-source-convergence", func() error {
			if err := runCustomSettingNameMetadataRepairAfterSourceConvergence(ctx, newDB, r.componentKindRegistrationSet()); err != nil {
				return fmt.Errorf("custom setting/name metadata repair after source convergence: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// Run the convergence rule again after every ApplyAll metadata writer.
		// RunRepair intentionally executes before pack apply/source reconciliation,
		// so its stamp cannot by itself prove that later legacy writers stayed
		// canonical. Refresh the cache before rebuild and terminal verification.
		if err := runLoggedControlPlanePhase("ApplyAll", "converge-master-detail-required-metadata-after-source-writers", func() error {
			if _, err := MigrateMasterDetailRequiredBackfill(ctx, newDB.DB()); err != nil {
				return fmt.Errorf("converge master-detail required metadata after source writers: %w", err)
			}
			if err := r.RefreshCache(ctx); err != nil {
				return fmt.Errorf("refresh metadata cache after master-detail convergence: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 8. Rebuild derived state
		if err := runLoggedControlPlanePhase("ApplyAll", "rebuild-derived", func() error {
			if err := rebuildDerived(ctx, r, options.DeferFormulaViewRebuild, true); err != nil {
				return fmt.Errorf("rebuild-derived: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 9. Standard indexes (synchronous, fatal — consistent with standalone ensure-indexes command)
		if err := runLoggedControlPlanePhase("ApplyAll", "ensure-indexes", func() error {
			if err := EnsureIndexes(ctx, newDB); err != nil {
				return fmt.Errorf("ensure-indexes: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 10. Rebuild flow runtime state immediately before the final verify
		// gate. This is the last-chance self-heal for a Tooling API delete or
		// install/reconcile step that left a dangling ActiveVersionId after
		// the broad RunRepair pass. It must rebuild TriggerSubscription shadow
		// rows in the same pass as orphan repair; clearing ActiveVersionId
		// alone would leave stale active shadow rows until the next apply.
		if err := runLoggedControlPlanePhase("ApplyAll", "flow-runtime-maintenance-before-verify", func() error {
			if err := runFlowRuntimeMaintenanceAfterOrphanRepair(ctx, newDB, r.SvcMgr.Metadata(), "apply final verify"); err != nil {
				return fmt.Errorf("flow runtime maintenance before verify: %w", err)
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// 11. Verify. The completed marker is the restart convergence proof for
		// the online-safe ApplyAll contract, so gate it on the writer-safe
		// readiness Verify rather than VerifyDeep. VerifyDeep includes live-data
		// scans (autonumber V5/V7) that assume record writers are quiesced; the
		// control-plane apply lock serializes control-plane ops only, not the
		// record-write path. Deep live-data audit stays on the explicit
		// operator-only `setup verify` path.
		if err := runLoggedControlPlanePhase("ApplyAll", "verify", func() error {
			if err := VerifyWithRunnerOptions(ctx, newDB, runnerOptions); err != nil {
				return fmt.Errorf("verify: %w", err)
			}
			if options.offlineCandidate != nil {
				if err := verifyOfflineCandidateFinalState(ctx, r, options.offlineCandidate); err != nil {
					return fmt.Errorf("verify exact offline candidate: %w", err)
				}
			}
			if options.ApplyAllFinalVerify != nil {
				if err := options.ApplyAllFinalVerify(ctx, newDB); err != nil {
					return fmt.Errorf("final distribution verify: %w", err)
				}
			}
			return nil
		}); err != nil {
			return failApplyAll(err)
		}

		// Only a FULL apply may stamp the completion marker that restart trusts.
		// SkipPolymorphicDiscriminatorBackfill is an operator partial-recovery
		// escape hatch that omits the fail-closed repair for unresolved polymorphic
		// targets, and neither final Verify nor any later phase re-validates those
		// server-managed <field>_type columns. Stamping completed here would let a
		// deliberately partial control plane be trusted as bootstrapped, and a later
		// auto restart would skip the very repair that was omitted. So leave the
		// marker non-completed (it stays "running"; the next full apply overwrites
		// it). The caller (cmd/setup) exits non-zero on this flag, mirroring
		// apply-source-convergence, so the partial result is not mistaken for a
		// ready control plane.
		if options.SkipPolymorphicDiscriminatorBackfill {
			log.Printf("⚠️ ApplyAll ran with polymorphic discriminator backfill SKIPPED — partial control plane, NOT marked bootstrapped. Rerun a full `setup apply` (without --skip-polymorphic-backfill) before relying on auto start.")
			return nil
		}

		markerDB := db.DB()
		if newDB != nil {
			markerDB = newDB.DB()
		}
		if err := runApplyAllCompletionGates(ctx, markerDB, options); err != nil {
			return failApplyAll(err)
		}
		if options.DeferFormulaViewRebuild {
			// Defense-in-depth re-stamp: the receipt was already written at
			// ENTRY (write-ahead, before the running marker), which closes
			// every crash window after commitment. Receipt mutations are
			// lock-serialized (package-private mutators, all call sites hold
			// the apply lock this function holds), so while this apply's
			// advisory-lock connection is alive no clear can interleave. The
			// advisory lock is however held on a dedicated connection: if
			// that connection dies mid-apply the lock releases while the
			// apply keeps running, and a queued lock-holding rebuild could
			// clear the debt concurrently — this idempotent re-stamp closes
			// the common case of that (rare, platform-wide GET_LOCK) window.
			// Completed must never be stamped with the debt invisible.
			if err := markFormulaViewRebuildOwed(ctx, markerDB); err != nil {
				return failApplyAll(err)
			}
		}
		if err := packs.MarkStandardActionsPersistenceSemanticsApplied(
			markerDB,
			time.Now().UTC().Format("2006-01-02 15:04:05"),
		); err != nil {
			return failApplyAll(fmt.Errorf("mark standard-action persistence semantics applied: %w", err))
		}
		// Persist the terminal-verification boundary before consulting the
		// session-scoped advisory lock one last time. A proxy may retire that
		// dedicated session during a long ApplyAll even though every mutation and
		// readiness gate completed. This receipt is deliberately not serving-ready;
		// an exact-candidate retry must reacquire the lock and repeat all terminal
		// verification before it may stamp completed.
		if err := bootstrapStateStore.MarkControlPlaneApplyAllFinalizationReady(markerDB); err != nil {
			return failApplyAll(fmt.Errorf("mark ApplyAll finalization ready: %w", err))
		}
		if err := requireControlPlaneApplyLockScope(applyLock); err != nil {
			return fmt.Errorf("control-plane apply lock final gate: %w", err)
		}
		if err := bootstrapStateStore.MarkControlPlaneApplyAllCompleted(markerDB); err != nil {
			return fmt.Errorf("mark ApplyAll completed: %w", err)
		}
		if completionPolicy == applyAllCompletionPolicyAutoStart {
			// Auto-start context: the entrypoint starts serving on this
			// function's success, so NO auto-start path — including a
			// completed-marker promotion that happened to run with defer
			// options — may return success while the rebuild debt is
			// outstanding. (The skip branch checks the debt too; this final
			// gate covers every mutating path.) Explicit operator applies
			// are exempt: a deployment's own defer apply legitimately ends
			// with the debt pending and settles it in its activation window.
			owed, owedErr := FormulaViewRebuildOwed(ctx, markerDB)
			if owedErr != nil {
				return owedErr
			}
			if owed {
				return fmt.Errorf("formula view rebuild owed after auto-start apply; run `setup rebuild` to rebuild views and settle the receipt before serving")
			}
		}

		log.Printf("══════════════════════════════════════════════════════════════")
		log.Printf("  Control Plane: ApplyAll completed (%.1fs)", time.Since(start).Seconds())
		log.Printf("══════════════════════════════════════════════════════════════")
		return nil
	})
	return applied, err
}

// ApplySourceConvergence runs only the source/state convergence phases without
// broad repair or final readiness verification. It is an explicit operator
// partial path for environments that need pack/source reconciliation and
// derived-state rebuild before a later full apply/repair.
func ApplySourceConvergence(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions) error {
	return ApplySourceConvergenceWithRunnerOptions(ctx, db, options, RunnerOptions{})
}

// ApplySourceConvergenceWithRunnerOptions runs source/state convergence with
// one immutable distribution-owned runner option snapshot.
func ApplySourceConvergenceWithRunnerOptions(ctx context.Context, db *database.TiDBConnection, options InstallPackOptions, runnerOptions RunnerOptions) error {
	log.Println("══════════════════════════════════════════════════════════════")
	log.Println("  Control Plane: ApplySourceConvergence")
	log.Println("══════════════════════════════════════════════════════════════")
	start := time.Now()

	return withControlPlaneMutationLock(ctx, db.DB(), "ApplySourceConvergence", func() error {
		newDB, err := ApplyMetadata(ctx, db)
		if err != nil {
			return fmt.Errorf("apply-metadata: %w", err)
		}

		if err := RunRepairWithOptionsAndRunnerOptions(ctx, newDB, RepairOptions{
			SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
			DeferConnectorWorkItemCutover:        true,
		}, runnerOptions); err != nil {
			return fmt.Errorf("run-repair: %w", err)
		}

		r, err := NewRunnerWithOptions(newDB, runnerOptions)
		if err != nil {
			return fmt.Errorf("runner init: %w", err)
		}
		if err := r.RefreshCache(ctx); err != nil {
			return fmt.Errorf("initial cache refresh: %w", err)
		}

		includeDevSample := !isProduction()
		if err := ApplyPacksWithOptions(ctx, r, includeDevSample, packs.ApplyOptions{
			SkipPolymorphicDiscriminatorBackfill: options.SkipPolymorphicDiscriminatorBackfill,
			RepairLegacyUniqueDuplicates:         options.RepairLegacyUniqueDuplicates,
			// DeferFormulaViewRebuild deliberately NOT mirrored:
			// ApplySourceConvergence is the operator partial path and keeps
			// always-rebuild semantics (see InstallPackOptions doc).
		}); err != nil {
			return fmt.Errorf("apply-packs: %w", err)
		}

		if err := VerifyConnectorExecutionProtocols(ctx, newDB.DB()); err != nil {
			return fmt.Errorf("verify WorkItem-only ConnectorEvent cutover after runtime pack application: %w", err)
		}

		// ApplySourceConvergence keeps always-rebuild semantics on every leg:
		// the adopt/reconcile paths below reach the pack materializer, so the
		// defer flag must be stripped here or a drift-lost view would
		// hard-fail this operator repair path instead of being healed —
		// and no entry precondition ran on this path.
		convergenceOptions := options
		convergenceOptions.DeferFormulaViewRebuild = false
		if err := adoptLegacyInstalledPackSourceIdentities(ctx, r, convergenceOptions); err != nil {
			return fmt.Errorf("adopt legacy installed pack source identities: %w", err)
		}

		reconcileOptions := convergenceOptions
		reconcileOptions.allowInstalledPackSourceStateRepair = true
		if err := reconcileActiveInstalledPackSourceState(ctx, r, reconcileOptions); err != nil {
			return fmt.Errorf("reconcile active installed pack sources: %w", err)
		}

		if _, err := BackfillConnectorPushLegacyMirrorAdoptionsWithComponentKindRegistrationSet(ctx, newDB.DB(), r.componentKindRegistrationSet()); err != nil {
			return fmt.Errorf("connector push legacy mirror backfill after source convergence: %w", err)
		}

		if err := runCustomSettingNameMetadataRepairAfterSourceConvergence(ctx, newDB, r.componentKindRegistrationSet()); err != nil {
			return fmt.Errorf("custom setting/name metadata repair after source convergence: %w", err)
		}

		// RunRepair precedes every source-backed writer on this partial path.
		// Re-run convergence after apply/adopt/reconcile and refresh the cache so
		// rebuildDerived cannot consume metadata written after the initial stamp.
		if _, err := MigrateMasterDetailRequiredBackfill(ctx, newDB.DB()); err != nil {
			return fmt.Errorf("converge master-detail required metadata after source convergence: %w", err)
		}
		if err := r.RefreshCache(ctx); err != nil {
			return fmt.Errorf("refresh metadata cache after source convergence master-detail repair: %w", err)
		}

		// Direct rebuildDerived with settlement: this closure runs inside
		// withControlPlaneMutationLock (the same control-plane apply lock),
		// and its full RebuildAllViews makes every view the running binary's
		// compiler output — a lock-holding full rebuild settles the
		// rebuild-owed receipt by definition.
		if err := rebuildDerived(ctx, r, false, true); err != nil {
			return fmt.Errorf("rebuild-derived: %w", err)
		}

		if err := EnsureIndexes(ctx, newDB); err != nil {
			return fmt.Errorf("ensure-indexes: %w", err)
		}

		log.Printf("══════════════════════════════════════════════════════════════")
		log.Printf("  Control Plane: ApplySourceConvergence completed (%.1fs)", time.Since(start).Seconds())
		log.Printf("══════════════════════════════════════════════════════════════")
		return nil
	})
}

func runCustomSettingNameMetadataRepairAfterSourceConvergence(ctx context.Context, db *database.TiDBConnection, registrations packs.ComponentKindRegistrationSet) error {
	if db == nil {
		return nil
	}
	updated, err := packs.BackfillObjectMetadataCustomSettingFromActivePacksWithComponentKindRegistrationSet(ctx, db.DB(), registrations)
	if err != nil {
		return err
	}
	if updated > 0 {
		log.Printf("controlplane: post-source-convergence CustomSetting backfill rows=%d", updated)
	}
	_, err = BackfillMissingNameFieldMetadata(ctx, db.DB())
	return err
}

// InitL0 initializes only the application-neutral L0 substrate. It is the
// falsification gate for the strong L0/L1 boundary: no app pack and no
// dev-sample metadata may be installed by this path.
func InitL0(ctx context.Context, db *database.TiDBConnection) error {
	log.Println("══════════════════════════════════════════════════════════════")
	log.Println("  Control Plane: InitL0")
	log.Println("══════════════════════════════════════════════════════════════")
	start := time.Now()

	return withControlPlaneMutationLock(ctx, db.DB(), "InitL0", func() error {
		newDB, err := EnsureL0SchemaDDL(ctx, db)
		if err != nil {
			return fmt.Errorf("ensure-l0-schema-ddl: %w", err)
		}

		if err := verifyNoL0Contamination(ctx, newDB.DB()); err != nil {
			return fmt.Errorf("precheck-l0-contamination: %w", err)
		}

		// ApplyL0ManagedMetadata owns the managed-schema boundary and therefore
		// establishes Permission State before its first protected write.
		newDB, err = ApplyL0ManagedMetadata(ctx, newDB)
		if err != nil {
			return fmt.Errorf("apply-l0-managed-metadata: %w", err)
		}

		if err := RunL0Repair(ctx, newDB); err != nil {
			return fmt.Errorf("run-l0-repair: %w", err)
		}

		r, err := NewRunner(newDB)
		if err != nil {
			return fmt.Errorf("runner init: %w", err)
		}
		if err := r.RefreshCache(ctx); err != nil {
			return fmt.Errorf("initial cache refresh: %w", err)
		}

		if err := ApplyRuntimeRequiredPacks(ctx, r); err != nil {
			return fmt.Errorf("apply-runtime-required-packs: %w", err)
		}

		if err := verifyNoL0Contamination(ctx, r.DB.DB()); err != nil {
			return fmt.Errorf("post-pack-l0-contamination: %w", err)
		}

		if err := packs.RegisterL0RuntimeMetadata(ctx, r.DB.DB()); err != nil {
			return fmt.Errorf("register L0 runtime metadata: %w", err)
		}

		if err := RebuildDerivedL0(ctx, r); err != nil {
			return fmt.Errorf("rebuild-derived-l0: %w", err)
		}

		// Deliberately skip broad EnsureIndexes here. It scans non-system physical
		// tables and can mutate app/business tables left in a reused database.
		// TODO(#56-backend): add an L0-object-scoped index materializer.

		if err := VerifyL0(ctx, newDB); err != nil {
			return fmt.Errorf("verify-l0: %w", err)
		}

		log.Printf("══════════════════════════════════════════════════════════════")
		log.Printf("  Control Plane: InitL0 completed (%.1fs)", time.Since(start).Seconds())
		log.Printf("══════════════════════════════════════════════════════════════")
		return nil
	})
}

func isProduction() bool {
	env := strings.ToLower(strings.TrimSpace(os.Getenv("NEXUS_ENV")))
	return env == "production" || env == "prod" || env == "staging"
}