package packs

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

	"github.com/nexuscrm/automation/autonumber"
	"github.com/nexuscrm/db/pkg/schemashape"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-controlplane/specdrift"
)

type packTableRegistryRow struct {
	tableType string
	category  string
	managed   bool
	deleted   bool
}

type packObjectRegistryRow struct {
	tableType                    string
	deleted                      bool
	keyPrefix                    sql.NullString
	defaultChildRows             sql.NullString
	childCloneRules              sql.NullString
	readSource                   sql.NullString
	childCloneRulesPackName      string
	childCloneRulesBusinessKey   string
	childCloneRulesManageability string
	childCloneRulesPackValue     sql.NullString
	isCustomized                 bool
	defaultChildRowsCustomized   bool
	childCloneRulesCustomized    bool
	lockRuleCustomized           bool
	lockRule                     sql.NullString
	displayNameField             sql.NullString
}

type packObjectListFieldsRegistryRow struct {
	listFields           sql.NullString
	listFieldsCustomized bool
	deleted              bool
}

type packObjectExclusiveSiblingRegistryRow struct {
	deleted                             bool
	exclusiveSiblingFields              sql.NullString
	exclusiveSiblingFieldsPackName      string
	exclusiveSiblingFieldsBusinessKey   string
	exclusiveSiblingFieldsManageability string
	exclusiveSiblingFieldsPackValue     sql.NullString
	exclusiveSiblingFieldsCustomized    bool
}

type packFieldRegistryRow struct {
	fieldAPIName string
	ownerKind    string
	ownerPack    string
	isCustomized bool
	isUnique     bool
	maxLength    sql.NullInt64
}

type objectSpecUniqueFieldKey struct {
	tableType     string
	objectAPIName string
	fieldAPIName  string
}

type objectSpecUniqueFieldRegistry map[objectSpecUniqueFieldKey]packFieldRegistryRow

func newObjectSpecUniqueFieldKey(tableType, objectAPIName, fieldAPIName string) objectSpecUniqueFieldKey {
	return objectSpecUniqueFieldKey{
		tableType:     strings.ToLower(strings.TrimSpace(tableType)),
		objectAPIName: strings.ToLower(strings.TrimSpace(objectAPIName)),
		fieldAPIName:  strings.ToLower(strings.TrimSpace(fieldAPIName)),
	}
}

type objectSpecAuthoredFieldSet struct {
	known      bool
	fields     map[objectSpecUniqueFieldKey]bool
	maxLengths map[objectSpecUniqueFieldKey]int
}

func newObjectSpecAuthoredFieldSet(specs []ObjectSpec) objectSpecAuthoredFieldSet {
	set := objectSpecAuthoredFieldSet{
		known:      true,
		fields:     make(map[objectSpecUniqueFieldKey]bool),
		maxLengths: make(map[objectSpecUniqueFieldKey]int),
	}
	for _, spec := range specs {
		for _, field := range spec.Fields {
			key := newObjectSpecUniqueFieldKey(spec.TableType, spec.APIName, field.APIName)
			if key.objectAPIName == "" || key.fieldAPIName == "" {
				continue
			}
			set.fields[key] = true
			set.maxLengths[key] = field.Length
		}
	}
	return set
}

func objectSpecAuthoredFieldSetFromArgs(sets []objectSpecAuthoredFieldSet) objectSpecAuthoredFieldSet {
	if len(sets) == 0 {
		return objectSpecAuthoredFieldSet{}
	}
	return sets[0]
}

func (set objectSpecAuthoredFieldSet) contains(tableType, objectAPIName, fieldAPIName string) bool {
	if !set.known {
		return true
	}
	return set.fields[newObjectSpecUniqueFieldKey(tableType, objectAPIName, fieldAPIName)]
}

func (set objectSpecAuthoredFieldSet) expectedMaxLength(tableType, objectAPIName, fieldAPIName string, fallback int) int {
	key := newObjectSpecUniqueFieldKey(tableType, objectAPIName, fieldAPIName)
	if !set.known || !set.fields[key] {
		return fallback
	}
	return set.maxLengths[key]
}

type packObjectPhysicalSchemaDriftError struct {
	drifts []string
}

type crossPackFieldVerifier struct {
	active   map[string]bool
	declared map[string]bool
}

// Keep object-spec repair classification fail-closed: each caller chooses the
// narrow repair predicate for its lifecycle, and unknown drift bullets remain
// non-repairable.
func (e *packObjectPhysicalSchemaDriftError) Error() string {
	if e == nil || len(e.drifts) == 0 {
		return "pack object physical schema drift"
	}
	return fmt.Errorf("pack object physical schema drift:\n  - %s", strings.Join(e.drifts, "\n  - ")).Error()
}

func (e *packObjectPhysicalSchemaDriftError) HasOnlyKeyPrefixDrift() bool {
	if e == nil || len(e.drifts) == 0 {
		return false
	}
	for _, drift := range e.drifts {
		if !isPackObjectKeyPrefixDriftBullet(drift) {
			return false
		}
	}
	return true
}

func (e *packObjectPhysicalSchemaDriftError) HasAnyMissingPhysicalColumns() bool {
	if e == nil || len(e.drifts) == 0 {
		return false
	}
	for _, drift := range e.drifts {
		if isPackPhysicalColumnMissingDriftBullet(drift) {
			return true
		}
	}
	return false
}

func (e *packObjectPhysicalSchemaDriftError) HasOnlyInstalledPackRepairableDrift() bool {
	if e == nil || len(e.drifts) == 0 {
		return false
	}
	cleanMissingFieldKeys := e.cleanMissingFieldKeys()
	for _, drift := range e.drifts {
		if isPackObjectKeyPrefixDriftBullet(drift) ||
			isPackFieldRowMissingDriftBullet(drift) ||
			isPackPhysicalColumnMissingDriftBullet(drift) {
			continue
		}
		if isPackUniqueIndexMissingDriftBullet(drift) && cleanMissingFieldKeys[fieldKeyFromUniqueIndexMissingBullet(drift)] {
			continue
		}
		return false
	}
	return true
}

// hasColumnMissingWithoutFieldRowMissing returns true when at least one
// physical-column-missing drift bullet has no corresponding _System_Field-row-missing
// drift bullet for the same field. This pattern indicates structural corruption:
// the schema was modified outside pack management while the metadata still expects
// the column. The check is per-field: a mixed error that contains Field1 with only
// column-missing (corruption) alongside Field2 with both column+field-row-missing
// (clean removal) correctly returns true because Field1 is corrupted.
func (e *packObjectPhysicalSchemaDriftError) hasColumnMissingWithoutFieldRowMissing() bool {
	if e == nil || len(e.drifts) == 0 {
		return false
	}
	columnMissingKeys := make(map[string]bool)
	fieldRowMissingKeys := make(map[string]bool)
	for _, drift := range e.drifts {
		if isPackPhysicalColumnMissingDriftBullet(drift) {
			if key := fieldKeyFromColumnMissingBullet(drift); key != "" {
				columnMissingKeys[key] = true
			}
		}
		if isPackFieldRowMissingDriftBullet(drift) {
			if key := fieldKeyFromFieldRowMissingBullet(drift); key != "" {
				fieldRowMissingKeys[key] = true
			}
		}
	}
	for key := range columnMissingKeys {
		if !fieldRowMissingKeys[key] {
			return true
		}
	}
	return false
}

func (e *packObjectPhysicalSchemaDriftError) cleanMissingFieldKeys() map[string]bool {
	columnMissingKeys := make(map[string]bool)
	fieldRowMissingKeys := make(map[string]bool)
	for _, drift := range e.drifts {
		if isPackPhysicalColumnMissingDriftBullet(drift) {
			if key := fieldKeyFromColumnMissingBullet(drift); key != "" {
				columnMissingKeys[key] = true
			}
		}
		if isPackFieldRowMissingDriftBullet(drift) {
			if key := fieldKeyFromFieldRowMissingBullet(drift); key != "" {
				fieldRowMissingKeys[key] = true
			}
		}
	}
	cleanMissingKeys := make(map[string]bool)
	for key := range columnMissingKeys {
		if fieldRowMissingKeys[key] {
			cleanMissingKeys[key] = true
		}
	}
	return cleanMissingKeys
}

// fieldKeyFromColumnMissingBullet extracts the "OBJ.FIELD" identity from a
// physical-column-missing drift bullet of the form "PACK physical column OBJ.FIELD is missing...".
func fieldKeyFromColumnMissingBullet(drift string) string {
	const marker = " physical column "
	start := strings.Index(drift, marker)
	if start < 0 {
		return ""
	}
	rest := drift[start+len(marker):]
	end := strings.Index(rest, " is missing")
	if end < 0 {
		return ""
	}
	return strings.ToLower(rest[:end])
}

// fieldKeyFromFieldRowMissingBullet extracts the "OBJ.FIELD" identity from a
// _System_Field-row-missing drift bullet of the form "PACK _System_Field row for OBJ.FIELD is missing...".
func fieldKeyFromFieldRowMissingBullet(drift string) string {
	const marker = " _System_Field row for "
	start := strings.Index(drift, marker)
	if start < 0 {
		return ""
	}
	rest := drift[start+len(marker):]
	end := strings.Index(rest, " is missing")
	if end < 0 {
		return ""
	}
	return strings.ToLower(rest[:end])
}

// fieldKeyFromUniqueIndexMissingBullet extracts the "OBJ.FIELD" identity from a
// unique-index-missing drift bullet of the form
// "PACK physical unique index on TABLE(COL) is missing for unique field OBJ.FIELD".
func fieldKeyFromUniqueIndexMissingBullet(drift string) string {
	const marker = " is missing for unique field "
	start := strings.Index(drift, marker)
	if start < 0 {
		return ""
	}
	return strings.ToLower(strings.TrimSpace(drift[start+len(marker):]))
}

// fieldKeyFromExtraLivePackOwnedFieldBullet extracts the "OBJ.FIELD" identity
// from a drift bullet of the form "PACK extra live pack-owned field OBJ.FIELD
// remains undeclared by current source".
func fieldKeyFromExtraLivePackOwnedFieldBullet(drift string) string {
	const marker = " extra live pack-owned field "
	start := strings.Index(drift, marker)
	if start < 0 {
		return ""
	}
	rest := drift[start+len(marker):]
	end := strings.Index(rest, " remains undeclared")
	if end < 0 {
		return ""
	}
	return strings.ToLower(rest[:end])
}

func isPackObjectKeyPrefixDriftBullet(drift string) bool {
	return strings.Contains(drift, " _System_Object row for ") && strings.Contains(drift, " KeyPrefix = ")
}

func isPackFieldRowMissingDriftBullet(drift string) bool {
	return strings.Contains(drift, " _System_Field row for ") && strings.Contains(drift, " is missing")
}

func isPackPhysicalColumnMissingDriftBullet(drift string) bool {
	return strings.Contains(drift, " physical column ") && strings.Contains(drift, " is missing")
}

func isPackUniqueIndexMissingDriftBullet(drift string) bool {
	return strings.Contains(drift, " physical unique index on ") && strings.Contains(drift, " is missing for unique field ")
}

func isPackExtraLivePackOwnedFieldDriftBullet(drift string) bool {
	return strings.Contains(drift, " extra live pack-owned field ") && strings.Contains(drift, " remains undeclared by current source")
}

func (r packFieldRegistryRow) isOwnedByPack(packName string) bool {
	return strings.EqualFold(strings.TrimSpace(r.ownerKind), constants.FieldOwnerKindPack) &&
		normalizePackIdentity(r.ownerPack) == normalizePackIdentity(packName)
}

func (r packFieldRegistryRow) isOwnedByRetainedPack(packName string) bool {
	return strings.EqualFold(strings.TrimSpace(r.ownerKind), constants.FieldOwnerKindPackRetained) &&
		normalizePackIdentity(r.ownerPack) == normalizePackIdentity(packName)
}

func (r packFieldRegistryRow) isOwnedByRemovingPack(packName string) bool {
	return strings.EqualFold(strings.TrimSpace(r.ownerKind), constants.FieldOwnerKindPackRemoving) &&
		normalizePackIdentity(r.ownerPack) == normalizePackIdentity(packName)
}

func (r packFieldRegistryRow) isAdminOwned() bool {
	return strings.EqualFold(strings.TrimSpace(r.ownerKind), constants.FieldOwnerKindAdmin)
}

func (r packFieldRegistryRow) isOwnerlessOrRuntime() bool {
	ownerKind := strings.TrimSpace(r.ownerKind)
	return ownerKind == "" || strings.EqualFold(ownerKind, constants.FieldOwnerKindRuntime)
}

func (r packFieldRegistryRow) ownerKindNormalized() string {
	return strings.ToLower(strings.TrimSpace(r.ownerKind))
}

func (r packFieldRegistryRow) ownerPackNormalized() string {
	return strings.TrimSpace(r.ownerPack)
}

func (e *packObjectPhysicalSchemaDriftError) HasOnlyInstalledPackRepairableDriftOrAllowedUndeclaredLiveFields(allowedFieldKeys map[string]bool) bool {
	if e == nil || len(e.drifts) == 0 {
		return false
	}
	cleanMissingFieldKeys := e.cleanMissingFieldKeys()
	for _, drift := range e.drifts {
		if isPackObjectKeyPrefixDriftBullet(drift) ||
			isPackFieldRowMissingDriftBullet(drift) ||
			isPackPhysicalColumnMissingDriftBullet(drift) {
			continue
		}
		if isPackUniqueIndexMissingDriftBullet(drift) && cleanMissingFieldKeys[fieldKeyFromUniqueIndexMissingBullet(drift)] {
			continue
		}
		if isPackExtraLivePackOwnedFieldDriftBullet(drift) && allowedFieldKeys[fieldKeyFromExtraLivePackOwnedFieldBullet(drift)] {
			continue
		}
		return false
	}
	return true
}

func normalizePackIdentity(packName string) string {
	return strings.ToLower(strings.TrimSpace(packName))
}

func (r packFieldRegistryRow) isPeerPackField(ownerPack string) bool {
	return (r.ownerKindNormalized() == constants.FieldOwnerKindPack || r.ownerKindNormalized() == constants.FieldOwnerKindPackRetained) &&
		r.ownerPackNormalized() != "" &&
		normalizePackIdentity(r.ownerPackNormalized()) != normalizePackIdentity(ownerPack)
}

func (r packFieldRegistryRow) isOtherPackManagedField(ownerPack string) bool {
	return (r.ownerKindNormalized() == constants.FieldOwnerKindPack ||
		r.ownerKindNormalized() == constants.FieldOwnerKindPackRetained ||
		r.ownerKindNormalized() == constants.FieldOwnerKindPackRemoving) &&
		r.ownerPackNormalized() != "" &&
		normalizePackIdentity(r.ownerPackNormalized()) != normalizePackIdentity(ownerPack)
}

func (v *crossPackFieldVerifier) declarationKey(ownerPack, objectAPIName, fieldAPIName string) string {
	return strings.ToLower(strings.TrimSpace(ownerPack)) + "|" +
		strings.ToLower(strings.TrimSpace(objectAPIName)) + "|" +
		strings.ToLower(strings.TrimSpace(fieldAPIName))
}

func (v *crossPackFieldVerifier) allowsDeclaredPeerPackField(row packFieldRegistryRow, objectAPIName, fieldAPIName, ownerPack string) bool {
	if !row.isPeerPackField(ownerPack) {
		return false
	}
	packName := row.ownerPackNormalized()
	if !v.active[packName] {
		return false
	}
	return v.declared[v.declarationKey(packName, objectAPIName, fieldAPIName)]
}

func buildCrossPackFieldVerifier(ctx context.Context, db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, fieldRegistry map[string]map[string]packFieldRegistryRow) (*crossPackFieldVerifier, error) {
	return buildCrossPackFieldVerifierWithComponentKindRegistrationSet(ctx, db, ownerPack, defs, fieldRegistry, nil)
}

func buildCrossPackFieldVerifierWithComponentKindRegistrationSet(ctx context.Context, db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, fieldRegistry map[string]map[string]packFieldRegistryRow, registrations *ComponentKindRegistrationSet) (*crossPackFieldVerifier, error) {
	verifier := &crossPackFieldVerifier{
		active:   make(map[string]bool),
		declared: make(map[string]bool),
	}
	for _, def := range defs {
		registeredFields := fieldRegistry[strings.ToLower(def.TableName)]
		if len(registeredFields) == 0 {
			continue
		}
		expectedFields := make(map[string]bool, len(def.Columns))
		for _, col := range def.Columns {
			expectedFields[strings.ToLower(col.Name)] = true
		}
		for fieldAPIName, row := range registeredFields {
			if expectedFields[fieldAPIName] {
				continue
			}
			switch {
			case row.isPeerPackField(ownerPack):
				packName := row.ownerPackNormalized()
				if _, ok := verifier.active[packName]; !ok {
					active, err := IsPackActive(ctx, db, packName)
					if err != nil {
						return nil, fmt.Errorf("verify %s.%s peer pack %s active state: %w", def.TableName, row.fieldAPIName, packName, err)
					}
					verifier.active[packName] = active
				}
				if !verifier.active[packName] {
					continue
				}
				key := verifier.declarationKey(packName, def.TableName, row.fieldAPIName)
				if _, ok := verifier.declared[key]; ok {
					continue
				}
				if override, ok := PeerDeclarationOverrideFromContext(ctx, packName); ok {
					declared, err := currentSourceDeclaresField(override, packName, def.TableName, row.fieldAPIName)
					if err != nil {
						return nil, fmt.Errorf("verify %s.%s peer pack %s declaration override: %w", def.TableName, row.fieldAPIName, packName, err)
					}
					verifier.declared[key] = declared
					continue
				}
				declared, err := isFieldDeclaredByPack(ctx, db, packName, def.TableName, row.fieldAPIName, registrations)
				if err != nil {
					return nil, fmt.Errorf("verify %s.%s peer pack %s declaration: %w", def.TableName, row.fieldAPIName, packName, err)
				}
				verifier.declared[key] = declared
			}
		}
	}
	return verifier, nil
}

func verifyPackObjectPhysicalSchema(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, manageability Manageability) error {
	return verifyPackObjectPhysicalSchemaWithContext(context.Background(), db, ownerPack, defs, manageability)
}

func verifyPackObjectPhysicalSchemaWithContext(ctx context.Context, db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, manageability Manageability, requiredFields ...objectSpecRequiredFieldIndex) error {
	return verifyPackObjectPhysicalSchemaWithContextAndAuthoredFields(ctx, db, ownerPack, defs, manageability, objectSpecAuthoredFieldSet{}, requiredFields...)
}

func verifyPackObjectPhysicalSchemaWithContextAndAuthoredFields(ctx context.Context, db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, manageability Manageability, authoredFields objectSpecAuthoredFieldSet, requiredFields ...objectSpecRequiredFieldIndex) error {
	return verifyPackObjectPhysicalSchemaWithContextAndAuthoredFieldsAndComponentKindRegistrationSet(ctx, db, ownerPack, defs, manageability, authoredFields, nil, requiredFields...)
}

func verifyPackObjectPhysicalSchemaWithContextAndAuthoredFieldsAndComponentKindRegistrationSet(ctx context.Context, db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, manageability Manageability, authoredFields objectSpecAuthoredFieldSet, registrations *ComponentKindRegistrationSet, requiredFields ...objectSpecRequiredFieldIndex) error {
	requiredFieldIndex := objectSpecRequiredFieldIndexFromArgs(requiredFields)
	physicalTables, err := loadPhysicalTableSet(db)
	if err != nil {
		return fmt.Errorf("load physical tables: %w", err)
	}
	physicalColumns, err := schemaProvisioner.LoadPhysicalColumnShapes(db)
	if err != nil {
		return fmt.Errorf("load physical columns: %w", err)
	}
	physicalIndices, err := loadPhysicalIndexSet(db)
	if err != nil {
		return fmt.Errorf("load physical indices: %w", err)
	}
	tableRegistry, err := loadPackTableRegistry(db)
	if err != nil {
		return fmt.Errorf("load table registry: %w", err)
	}
	objectRegistry, err := loadPackObjectRegistry(db)
	if err != nil {
		return fmt.Errorf("load object registry: %w", err)
	}
	fieldRegistry, err := loadPackFieldRegistry(db)
	if err != nil {
		return fmt.Errorf("load field registry: %w", err)
	}
	uniqueConstraintDefs, err := objectSpecUniquePhysicalConstraintDefinitionsForVerify(db, ownerPack, defs, manageability, authoredFields)
	if err != nil {
		return fmt.Errorf("resolve unique physical constraints: %w", err)
	}
	crossPackVerifier, err := buildCrossPackFieldVerifierWithComponentKindRegistrationSet(ctx, db, ownerPack, defs, fieldRegistry, registrations)
	if err != nil {
		return err
	}

	errs := collectObjectSpecUniquePhysicalConstraintDrifts(ownerPack, uniqueConstraintDefs, physicalTables, physicalIndices)
	for _, def := range uniqueConstraintDefs {
		tableKey := strings.ToLower(def.TableName)
		metadataOnly := def.ReadSource != nil
		if metadataOnly {
			_, _, readSourceDrifts := collectReadSourcePhysicalContractDrifts(ownerPack, def, physicalTables, physicalColumns, requiredFieldIndex)
			errs = append(errs, readSourceDrifts...)
		}
		tableRow, hasTableRow := tableRegistry[tableKey]
		if metadataOnly {
			if physicalTables[tableKey] {
				errs = append(errs, fmt.Sprintf("%s stale physical table %s remains after readSource cutover", ownerPack, def.TableName))
			}
			if hasTableRow && !tableRow.deleted {
				errs = append(errs, fmt.Sprintf("%s stale _System_Table row for %s remains after readSource cutover", ownerPack, def.TableName))
			}
		} else {
			if !physicalTables[tableKey] {
				errs = append(errs, fmt.Sprintf("%s physical table %s is missing", ownerPack, def.TableName))
				continue
			}
			if !hasTableRow {
				errs = append(errs, fmt.Sprintf("%s _System_Table row for %s is missing", ownerPack, def.TableName))
			} else {
				if tableRow.deleted {
					errs = append(errs, fmt.Sprintf("%s _System_Table row for %s is deleted", ownerPack, def.TableName))
				}
				if !tableRow.managed {
					errs = append(errs, fmt.Sprintf("%s _System_Table row for %s IsManaged = 0, want 1", ownerPack, def.TableName))
				}
				if tableRow.tableType != def.TableType {
					errs = append(errs, fmt.Sprintf("%s _System_Table row for %s TableType = %q, want %q", ownerPack, def.TableName, tableRow.tableType, def.TableType))
				}
				if tableRow.category != def.Category {
					errs = append(errs, fmt.Sprintf("%s _System_Table row for %s Category = %q, want %q", ownerPack, def.TableName, tableRow.category, def.Category))
				}
			}
		}

		objectRow, ok := objectRegistry[tableKey]
		if !ok {
			errs = append(errs, fmt.Sprintf("%s _System_Object row for %s is missing", ownerPack, def.TableName))
		} else {
			if objectRow.deleted {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s is deleted", ownerPack, def.TableName))
			}
			if objectRow.tableType != def.TableType {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s TableType = %q, want %q", ownerPack, def.TableName, objectRow.tableType, def.TableType))
			}
			// Source-backed object-spec verify reaches this branch because
			// convertSpecToTableDefinitions propagates ObjectSpec.KeyPrefix into
			// TableDefinition.KeyPrefix before verifyPackObjectPhysicalSchema runs.
			if wantKeyPrefix := strings.TrimSpace(def.KeyPrefix); wantKeyPrefix != "" {
				if !objectRow.keyPrefix.Valid || objectRow.keyPrefix.String != wantKeyPrefix {
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s KeyPrefix = %s, want %q", ownerPack, def.TableName, formatPackColumnDefault(objectRow.keyPrefix.Valid, objectRow.keyPrefix.String), wantKeyPrefix))
				}
			}
			wantReadSource, err := canonicalReadSourceJSON(def.ReadSource)
			if err != nil {
				errs = append(errs, fmt.Sprintf("%s object %s ReadSource canonicalization failed: %v", ownerPack, def.TableName, err))
			} else {
				liveReadSource := ""
				var liveReadSourceErr error
				if objectRow.readSource.Valid {
					liveReadSource, liveReadSourceErr = canonicalStoredReadSourceJSON(objectRow.readSource.String)
				}
				switch {
				case liveReadSourceErr != nil:
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ReadSource is invalid: %v", ownerPack, def.TableName, liveReadSourceErr))
				case wantReadSource != "" && liveReadSource != wantReadSource:
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ReadSource = %s, want %q", ownerPack, def.TableName, formatPackColumnDefault(objectRow.readSource.Valid, objectRow.readSource.String), wantReadSource))
				case wantReadSource == "" && liveReadSource != "":
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ReadSource = %q, want empty (spec omits readSource; stale value not cleared)", ownerPack, def.TableName, objectRow.readSource.String))
				}
			}
			if ResolveObjectSpecValueOwnership(manageability, objectRow.isCustomized) == ObjectSpecValueExactRequired {
				wantDisplayName := strings.TrimSpace(def.DisplayNameField)
				liveDisplayName := ""
				if objectRow.displayNameField.Valid {
					liveDisplayName = strings.TrimSpace(objectRow.displayNameField.String)
				}
				switch {
				case wantDisplayName != "" && liveDisplayName != wantDisplayName:
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s DisplayNameField = %s, want %q", ownerPack, def.TableName, formatPackColumnDefault(objectRow.displayNameField.Valid, objectRow.displayNameField.String), wantDisplayName))
				case wantDisplayName == "" && liveDisplayName != "":
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s DisplayNameField = %q, want empty (spec omits displayNameField; stale value not cleared)", ownerPack, def.TableName, objectRow.displayNameField.String))
				}
			}
			// LockRule is pack-applied into _System_Object.LockRule; verify must
			// detect cleared or tampered values so verify-pack and install-pack
			// cover the same payload facts (L0/L1 install lifecycle invariant).
			//
			// Ownership is resolved by ResolveLockRuleValueOwnership, which uses the
			// column-scoped LockRuleCustomized bit (mirrors DefaultChildRowsCustomized):
			//   ManageabilityLocked + any lockRuleCustomized → exact match required
			//     (locked surfaces override admin edits)
			//   ManageabilityDefaultsOnly + lockRuleCustomized=1 → admin owns; skip
			//   ManageabilityDefaultsOnly + lockRuleCustomized=0 → exact match required
			//     (IsCustomized=1 alone must NOT suppress LockRule check; only the
			//     column-scoped bit transfers LockRule ownership to the admin)
			if ResolveLockRuleValueOwnership(manageability, objectRow.lockRuleCustomized) == ObjectSpecValueExactRequired {
				wantLockRule := strings.TrimSpace(def.LockRule)
				liveLockRule := ""
				if objectRow.lockRule.Valid {
					liveLockRule = strings.TrimSpace(objectRow.lockRule.String)
				}
				if wantLockRule != "" {
					if liveLockRule != wantLockRule {
						errs = append(errs, fmt.Sprintf("%s _System_Object row for %s LockRule = %s, want %q", ownerPack, def.TableName, formatPackColumnDefault(objectRow.lockRule.Valid, objectRow.lockRule.String), wantLockRule))
					}
				} else if liveLockRule != "" {
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s LockRule = %q, want empty (spec omits lockRule; stale value not cleared)", ownerPack, def.TableName, objectRow.lockRule.String))
				}
			}
		}

		actualColumns := physicalColumns[tableKey]
		registeredFields := fieldRegistry[tableKey]
		expectedFields := make(map[string]bool, len(def.Columns))
		for _, col := range def.Columns {
			colKey := strings.ToLower(col.Name)
			expectedFields[colKey] = true
			resolvedStorage, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			if err != nil {
				errs = append(errs, fmt.Sprintf("%s field %s.%s has invalid formula storage metadata: %v", ownerPack, def.TableName, col.Name, err))
				continue
			}
			if !metadataOnly && resolvedStorage.RequiresPhysicalColumn() {
				actualColumn, ok := actualColumns[colKey]
				if !ok {
					errs = append(errs, fmt.Sprintf("%s physical column %s.%s is missing", ownerPack, def.TableName, col.Name))
				} else if !packObjectSpecColumnShapeMatches(def.TableName, actualColumn, col, requiredFieldIndex) {
					errs = append(errs, fmt.Sprintf("%s physical column %s.%s shape = %s nullable=%v default=%s, want %s nullable=%v default=%s",
						ownerPack,
						def.TableName,
						col.Name,
						actualColumn.ColumnType,
						actualColumn.Nullable,
						formatPackColumnDefault(actualColumn.DefaultVal.Valid, actualColumn.DefaultVal.String),
						schemaProvisioner.DesiredColumnType(col),
						col.Nullable,
						formatPackColumnDefault(col.Default != "", col.Default),
					))
				}
			}
			if packColumnRequiresFieldMetadata(col.Name) {
				row, ok := registeredFields[colKey]
				if !ok {
					errs = append(errs, fmt.Sprintf("%s _System_Field row for %s.%s is missing", ownerPack, def.TableName, col.Name))
				} else if !row.isOwnedByPack(ownerPack) && !row.isOwnedByRetainedPack(ownerPack) {
					errs = append(errs, fmt.Sprintf("%s _System_Field row for %s.%s owner = %s/%s, want pack/%s", ownerPack, def.TableName, col.Name, row.ownerKind, row.ownerPack, ownerPack))
				} else if fieldMetadataExactRequired(manageability, row) {
					wantMaxLength := authoredFields.expectedMaxLength(def.TableType, def.TableName, col.Name, col.Length)
					if !packMaxLengthMatches(row.maxLength, wantMaxLength) {
						errs = append(errs, fmt.Sprintf("%s _System_Field row for %s.%s MaxLength = %s, want %s", ownerPack, def.TableName, col.Name, formatPackInt64(row.maxLength), formatExpectedPackLength(wantMaxLength)))
					}
				}
			}
		}
		for fieldAPIName, row := range registeredFields {
			if expectedFields[fieldAPIName] {
				continue
			}
			displayFieldAPIName := row.fieldAPIName
			if strings.TrimSpace(displayFieldAPIName) == "" {
				displayFieldAPIName = fieldAPIName
			}
			if row.isOwnedByPack(ownerPack) || row.isOwnedByRetainedPack(ownerPack) || row.isOwnedByRemovingPack(ownerPack) {
				errs = append(errs, fmt.Sprintf("%s extra live pack-owned field %s.%s remains undeclared by current source", ownerPack, def.TableName, displayFieldAPIName))
				continue
			}
			if crossPackVerifier.allowsDeclaredPeerPackField(row, def.TableName, fieldAPIName, ownerPack) {
				continue
			}
			if row.isOtherPackManagedField(ownerPack) {
				errs = append(errs, fmt.Sprintf("%s unexpected peer-pack field %s.%s owner = %s/%s", ownerPack, def.TableName, displayFieldAPIName, row.ownerKind, row.ownerPack))
				continue
			}
			if metadataOnly {
				errs = append(errs, fmt.Sprintf("%s unsupported live field %s.%s owner = %s/%s blocks readSource cutover", ownerPack, def.TableName, displayFieldAPIName, row.ownerKind, row.ownerPack))
			}
		}
		if metadataOnly {
			liveFields, err := loadReadSourceLiveFields(ctx, db, def.TableName)
			if err != nil {
				return fmt.Errorf("load readSource live fields for %s: %w", def.TableName, err)
			}
			for _, row := range liveFields {
				if _, ok := expectedFields[strings.ToLower(strings.TrimSpace(row.fieldAPIName))]; !ok {
					continue
				}
				resolved, err := formulastorage.Resolve(formulastorage.Spec{
					FormulaMode:         row.formulaMode,
					FormulaStorageMode:  row.formulaStorageMode,
					Formula:             row.formula,
					HasProjectionConfig: row.projectionConfig != "",
				})
				_ = err
				if !resolved.RequiresFormulaView() {
					continue
				}
				errs = append(errs, fmt.Sprintf("%s readSource object %s has active computed field %s owner = %s; readSource + formula routing is unsupported", ownerPack, def.TableName, row.fieldAPIName, formatReadSourceFieldOwner(row)))
			}
		}

		if !metadataOnly {
			actualIndices := physicalIndices[tableKey]
			for _, idx := range def.Indices {
				colKey := packIndexColumnKey(idx.Columns)
				actual, ok := actualIndices[colKey]
				if !ok {
					errs = append(errs, fmt.Sprintf("%s physical index on %s(%s) is missing", ownerPack, def.TableName, strings.Join(idx.Columns, ", ")))
					continue
				}
				if actual.unique != idx.Unique {
					errs = append(errs, fmt.Sprintf("%s physical index %s on %s(%s) unique = %v, want %v", ownerPack, actual.name, def.TableName, strings.Join(idx.Columns, ", "), actual.unique, idx.Unique))
				}
			}
		}
	}

	if len(errs) > 0 {
		sort.Strings(errs)
		return &packObjectPhysicalSchemaDriftError{drifts: errs}
	}
	return nil
}

func assertObjectSpecUniquePhysicalConstraints(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition) error {
	physicalTables, err := loadPhysicalTableSet(db)
	if err != nil {
		return fmt.Errorf("load physical tables: %w", err)
	}
	physicalIndices, err := loadPhysicalIndexSet(db)
	if err != nil {
		return fmt.Errorf("load physical indices: %w", err)
	}
	drifts := collectObjectSpecUniquePhysicalConstraintDrifts(ownerPack, defs, physicalTables, physicalIndices)
	if len(drifts) == 0 {
		return nil
	}
	sort.Strings(drifts)
	return &packObjectPhysicalSchemaDriftError{drifts: drifts}
}

func AssertObjectSpecUniquePhysicalConstraints(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition) error {
	return assertObjectSpecUniquePhysicalConstraints(db, ownerPack, defs)
}

func dropOwnedObjectSpecUniqueDowngrades(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition) error {
	physicalIndices, err := loadPhysicalIndexSet(db)
	if err != nil {
		return fmt.Errorf("load physical indices: %w", err)
	}
	dropCount := 0
	for _, def := range defs {
		if def.ReadSource != nil {
			continue
		}
		tableKey := strings.ToLower(def.TableName)
		for _, col := range def.Columns {
			if col.Unique || col.PrimaryKey || tableDefinitionDeclaresSingleColumnUniqueIndex(def, col.Name) {
				continue
			}
			actual, ok := physicalIndices[tableKey][packIndexColumnKey([]string{col.Name})]
			if !ok || !actual.unique {
				continue
			}
			ownedIndexName := domainschema.GenerateUniqueIndexName(def.TableName, col.Name)
			if !strings.EqualFold(actual.name, ownedIndexName) {
				continue
			}
			ddl := fmt.Sprintf("ALTER TABLE %s DROP INDEX %s", quotePackSQLIdentifier(def.TableName), quotePackSQLIdentifier(ownedIndexName))
			if _, err := db.Exec(ddl); err != nil {
				return fmt.Errorf("drop stale owned unique index %s on %s.%s: %w", ownedIndexName, def.TableName, col.Name, err)
			}
			dropCount++
		}
	}
	if dropCount > 0 {
		log.Printf("object_spec unique downgrade: dropped %d stale owned unique index(es) for pack %s", dropCount, ownerPack)
	}
	return nil
}

func DropOwnedObjectSpecUniqueDowngrades(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition) error {
	return dropOwnedObjectSpecUniqueDowngrades(db, ownerPack, defs)
}

func preflightObjectSpecUniqueLegacyDuplicates(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, requiredFields ...objectSpecRequiredFieldIndex) error {
	return preflightOrRepairObjectSpecUniqueLegacyDuplicates(db, ownerPack, defs, false, requiredFields...)
}

func preflightOrRepairObjectSpecUniqueLegacyDuplicates(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, repair bool, requiredFields ...objectSpecRequiredFieldIndex) error {
	requiredFieldIndex := objectSpecRequiredFieldIndexFromArgs(requiredFields)
	requiredFieldIndexKnown := len(requiredFields) > 0
	physicalTables, err := loadPhysicalTableSet(db)
	if err != nil {
		return fmt.Errorf("load physical tables: %w", err)
	}
	physicalColumns, err := schemaProvisioner.LoadPhysicalColumnShapes(db)
	if err != nil {
		return fmt.Errorf("load physical columns: %w", err)
	}
	physicalIndices, err := loadPhysicalIndexSet(db)
	if err != nil {
		return fmt.Errorf("load physical indices: %w", err)
	}
	var liveRequiredFields objectSpecRequiredFieldRegistry
	if repair {
		liveRequiredFields, err = loadObjectSpecRequiredFieldRegistry(db)
		if err != nil {
			return fmt.Errorf("load live required field metadata: %w", err)
		}
	}
	var drifts []string
	for _, def := range defs {
		physicalTableName, err := objectSpecPhysicalTableName(def)
		if err != nil {
			drifts = append(drifts, fmt.Sprintf("%s object %s ReadSource is invalid: %v", ownerPack, def.TableName, err))
			continue
		}
		tableKey := strings.ToLower(physicalTableName)
		if !physicalTables[tableKey] {
			continue
		}
		actualColumns := physicalColumns[tableKey]
		for _, col := range def.Columns {
			if !col.Unique {
				continue
			}
			resolvedStorage, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			if err != nil {
				drifts = append(drifts, fmt.Sprintf("%s field %s.%s has invalid formula storage metadata: %v", ownerPack, def.TableName, col.Name, err))
				continue
			}
			if !resolvedStorage.RequiresPhysicalColumn() {
				continue
			}
			actualColumn, ok := actualColumns[strings.ToLower(col.Name)]
			if !ok {
				continue
			}
			if objectSpecSingleColumnUniqueIndexExists(physicalIndices, tableKey, col.Name) {
				continue
			}
			duplicates, err := loadObjectSpecUniqueDuplicateSamples(db, physicalTableName, col.Name, 5)
			if err != nil {
				return fmt.Errorf("scan duplicate values for %s.%s: %w", physicalTableName, col.Name, err)
			}
			if len(duplicates) > 0 {
				duplicateSummary := formatObjectSpecUniqueDuplicateSamples(duplicates)
				if !repair {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s has duplicate legacy values before unique-index enforcement: %s; rerun with ApplyOptions.RepairLegacyUniqueDuplicates to clear duplicate non-canonical values before DDL", ownerPack, def.TableName, col.Name, duplicateSummary))
					continue
				}
				if def.ReadSource != nil {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s has duplicate legacy values in readSource table %s before unique-index enforcement: %s; repair the source table explicitly before apply", ownerPack, def.TableName, col.Name, physicalTableName, duplicateSummary))
					continue
				}
				required, known := objectSpecUniqueDuplicateRepairRequiredField(def.TableName, col.Name, requiredFieldIndex, requiredFieldIndexKnown, liveRequiredFields)
				if !known {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s has duplicate legacy values before unique-index enforcement but cannot be auto-repaired because logical required metadata is unavailable: %s; repair duplicate values explicitly before apply", ownerPack, def.TableName, col.Name, duplicateSummary))
					continue
				}
				if required {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s has duplicate legacy values before unique-index enforcement but cannot be auto-repaired because the field is logically required: %s; repair duplicate values explicitly before apply", ownerPack, def.TableName, col.Name, duplicateSummary))
					continue
				}
				if !col.Nullable || !actualColumn.Nullable {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s has duplicate legacy values before unique-index enforcement but cannot be auto-repaired because the desired/physical column is not nullable: %s", ownerPack, def.TableName, col.Name, duplicateSummary))
					continue
				}
				if _, ok := actualColumns[strings.ToLower(constants.FieldID)]; !ok {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s has duplicate legacy values before unique-index enforcement but cannot be auto-repaired because physical table %s has no %s column", ownerPack, def.TableName, col.Name, physicalTableName, constants.FieldID))
					continue
				}
				updated, err := repairObjectSpecUniqueDuplicateValues(db, physicalTableName, col.Name)
				if err != nil {
					return fmt.Errorf("repair duplicate values for %s.%s: %w", physicalTableName, col.Name, err)
				}
				log.Printf("object_spec unique repair: cleared %d duplicate non-canonical row value(s) from %s.%s for pack %s", updated, physicalTableName, col.Name, ownerPack)
				remaining, err := loadObjectSpecUniqueDuplicateSamples(db, physicalTableName, col.Name, 5)
				if err != nil {
					return fmt.Errorf("rescan duplicate values for %s.%s: %w", physicalTableName, col.Name, err)
				}
				if len(remaining) > 0 {
					drifts = append(drifts, fmt.Sprintf("%s unique field %s.%s still has duplicate legacy values after repair: %s", ownerPack, def.TableName, col.Name, formatObjectSpecUniqueDuplicateSamples(remaining)))
				}
			}
		}
	}
	if len(drifts) == 0 {
		return nil
	}
	sort.Strings(drifts)
	return &packObjectPhysicalSchemaDriftError{drifts: drifts}
}

// PreflightOrRepairObjectSpecUniqueLegacyDuplicates checks legacy duplicate
// values before object_spec unique-index enforcement. In repair mode, callers
// that cannot provide object-spec requiredness are failed closed before any
// destructive value clearing.
func PreflightOrRepairObjectSpecUniqueLegacyDuplicates(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, repair bool) error {
	return preflightOrRepairObjectSpecUniqueLegacyDuplicates(db, ownerPack, defs, repair)
}

// PreflightOrRepairObjectSpecUniqueLegacyDuplicatesForObjectSpecs is the apply
// path entrypoint when the authored object specs are available. Repair uses
// those specs plus live _System_Field.Required to avoid clearing required-field
// values.
func PreflightOrRepairObjectSpecUniqueLegacyDuplicatesForObjectSpecs(db *sql.DB, ownerPack string, defs []domainschema.TableDefinition, specs []ObjectSpec, repair bool) error {
	return preflightOrRepairObjectSpecUniqueLegacyDuplicates(db, ownerPack, defs, repair, newObjectSpecRequiredFieldIndex(specs))
}

type objectSpecRequiredFieldRegistry map[string]map[string]bool

func loadObjectSpecRequiredFieldRegistry(db *sql.DB) (objectSpecRequiredFieldRegistry, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT o.`%s`, f.`%s`, COALESCE(f.`%s`, 0) FROM `%s` f JOIN `%s` o ON o.`%s` = f.`%s` WHERE COALESCE(f.`%s`, 0) = 0 AND COALESCE(o.`%s`, 0) = 0",
		constants.FieldSysObject_APIName,
		constants.FieldSysField_APIName,
		constants.FieldSysField_Required,
		constants.TableField,
		constants.TableObject,
		constants.FieldSysObject_ID,
		constants.FieldSysField_ObjectID,
		constants.FieldSysField_IsDeleted,
		constants.FieldSysObject_IsDeleted,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(objectSpecRequiredFieldRegistry)
	for rows.Next() {
		var objectAPIName, fieldAPIName string
		var required int
		if err := rows.Scan(&objectAPIName, &fieldAPIName, &required); err != nil {
			return nil, err
		}
		objectKey := strings.ToLower(strings.TrimSpace(objectAPIName))
		fieldKey := strings.ToLower(strings.TrimSpace(fieldAPIName))
		if objectKey == "" || fieldKey == "" {
			continue
		}
		if registry[objectKey] == nil {
			registry[objectKey] = map[string]bool{}
		}
		registry[objectKey][fieldKey] = required != 0
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func (registry objectSpecRequiredFieldRegistry) required(objectAPIName string, fieldAPIName string) (bool, bool) {
	if len(registry) == 0 {
		return false, false
	}
	fields := registry[strings.ToLower(strings.TrimSpace(objectAPIName))]
	if len(fields) == 0 {
		return false, false
	}
	required, ok := fields[strings.ToLower(strings.TrimSpace(fieldAPIName))]
	return required, ok
}

func objectSpecUniqueDuplicateRepairRequiredField(
	objectAPIName string,
	fieldAPIName string,
	requiredFields objectSpecRequiredFieldIndex,
	requiredFieldsKnown bool,
	liveRequiredFields objectSpecRequiredFieldRegistry,
) (bool, bool) {
	if requiredFieldsKnown && requiredFields.required(objectAPIName, fieldAPIName) {
		return true, true
	}
	liveRequired, liveRequiredKnown := liveRequiredFields.required(objectAPIName, fieldAPIName)
	if liveRequiredKnown && liveRequired {
		return true, true
	}
	if !requiredFieldsKnown {
		return false, false
	}
	if liveRequiredKnown {
		return liveRequired, true
	}
	return false, true
}

func repairObjectSpecUniqueDuplicateValues(db *sql.DB, tableName, columnName string) (int64, error) {
	quotedTable := quotePackSQLIdentifier(tableName)
	quotedColumn := quotePackSQLIdentifier(columnName)
	quotedID := quotePackSQLIdentifier(constants.FieldID)
	query := fmt.Sprintf(
		`UPDATE %s AS target
JOIN (
	SELECT ranked.%s
	FROM (
		SELECT %s,
			ROW_NUMBER() OVER (PARTITION BY %s ORDER BY %s) AS duplicate_rank
		FROM %s
		WHERE %s IS NOT NULL
	) AS ranked
	WHERE ranked.duplicate_rank > 1
) AS duplicate_rows ON duplicate_rows.%s = target.%s
SET target.%s = NULL`,
		quotedTable,
		quotedID,
		quotedID,
		quotedColumn,
		quotedID,
		quotedTable,
		quotedColumn,
		quotedID,
		quotedID,
		quotedColumn,
	)
	result, err := db.Exec(query)
	if err != nil {
		return 0, err
	}
	updated, err := result.RowsAffected()
	if err != nil {
		return 0, err
	}
	return updated, nil
}

func collectObjectSpecUniquePhysicalConstraintDrifts(ownerPack string, defs []domainschema.TableDefinition, physicalTables map[string]bool, physicalIndices map[string]map[string]packIndexInfo) []string {
	var drifts []string
	for _, def := range defs {
		physicalTableName, err := objectSpecPhysicalTableName(def)
		if err != nil {
			drifts = append(drifts, fmt.Sprintf("%s object %s ReadSource is invalid: %v", ownerPack, def.TableName, err))
			continue
		}
		tableKey := strings.ToLower(physicalTableName)
		for _, col := range def.Columns {
			if !col.Unique {
				if col.PrimaryKey || tableDefinitionDeclaresSingleColumnUniqueIndex(def, col.Name) {
					continue
				}
				actual, ok := physicalIndices[tableKey][packIndexColumnKey([]string{col.Name})]
				if ok && actual.unique && !strings.EqualFold(actual.name, "PRIMARY") {
					drifts = append(drifts, fmt.Sprintf("%s physical unique index %s on %s(%s) remains for non-unique field %s.%s", ownerPack, actual.name, physicalTableName, col.Name, def.TableName, col.Name))
				}
				continue
			}
			resolvedStorage, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			if err != nil {
				drifts = append(drifts, fmt.Sprintf("%s field %s.%s has invalid formula storage metadata: %v", ownerPack, def.TableName, col.Name, err))
				continue
			}
			if !resolvedStorage.RequiresPhysicalColumn() {
				drifts = append(drifts, fmt.Sprintf("%s field %s.%s declares unique but has no physical storage column", ownerPack, def.TableName, col.Name))
				continue
			}
			if !physicalTables[tableKey] {
				drifts = append(drifts, fmt.Sprintf("%s physical table %s for unique field %s.%s is missing", ownerPack, physicalTableName, def.TableName, col.Name))
				continue
			}
			actual, ok := physicalIndices[tableKey][packIndexColumnKey([]string{col.Name})]
			if !ok {
				drifts = append(drifts, fmt.Sprintf("%s physical unique index on %s(%s) is missing for unique field %s.%s", ownerPack, physicalTableName, col.Name, def.TableName, col.Name))
				continue
			}
			if !actual.unique {
				drifts = append(drifts, fmt.Sprintf("%s physical index %s on %s(%s) unique = false, want true for unique field %s.%s", ownerPack, actual.name, physicalTableName, col.Name, def.TableName, col.Name))
			}
		}
	}
	return drifts
}

func tableDefinitionDeclaresSingleColumnUniqueIndex(def domainschema.TableDefinition, columnName string) bool {
	for _, idx := range def.Indices {
		if !idx.Unique || len(idx.Columns) != 1 {
			continue
		}
		if strings.EqualFold(strings.TrimSpace(idx.Columns[0]), strings.TrimSpace(columnName)) {
			return true
		}
	}
	return false
}

type objectSpecUniqueDuplicateSample struct {
	value string
	count int64
}

func loadObjectSpecUniqueDuplicateSamples(db *sql.DB, tableName, columnName string, limit int) ([]objectSpecUniqueDuplicateSample, error) {
	query := fmt.Sprintf(
		"SELECT CAST(%s AS CHAR) AS duplicate_value, COUNT(*) FROM %s WHERE %s IS NOT NULL GROUP BY %s HAVING COUNT(*) > 1 ORDER BY COUNT(*) DESC LIMIT %d",
		quotePackSQLIdentifier(columnName),
		quotePackSQLIdentifier(tableName),
		quotePackSQLIdentifier(columnName),
		quotePackSQLIdentifier(columnName),
		limit,
	)
	rows, err := db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []objectSpecUniqueDuplicateSample
	for rows.Next() {
		var value sql.NullString
		var count int64
		if err := rows.Scan(&value, &count); err != nil {
			return nil, err
		}
		display := "<null>"
		if value.Valid {
			display = value.String
		}
		out = append(out, objectSpecUniqueDuplicateSample{value: display, count: count})
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return out, nil
}

func formatObjectSpecUniqueDuplicateSamples(samples []objectSpecUniqueDuplicateSample) string {
	parts := make([]string, 0, len(samples))
	for _, sample := range samples {
		parts = append(parts, fmt.Sprintf("%q x%d", sample.value, sample.count))
	}
	return strings.Join(parts, ", ")
}

func objectSpecSingleColumnUniqueIndexExists(physicalIndices map[string]map[string]packIndexInfo, tableKey, columnName string) bool {
	actual, ok := physicalIndices[tableKey][packIndexColumnKey([]string{columnName})]
	return ok && actual.unique
}

func objectSpecPhysicalTableName(def domainschema.TableDefinition) (string, error) {
	if def.ReadSource == nil {
		return def.TableName, nil
	}
	normalized, err := models.NormalizeObjectReadSourceConfig(def.ReadSource)
	if err != nil {
		return "", err
	}
	return normalized.PhysicalTable, nil
}

func collectReadSourcePhysicalContractDrifts(ownerPack string, def domainschema.TableDefinition, physicalTables map[string]bool, physicalColumns map[string]map[string]schemashape.PhysicalColumnShape, requiredFields objectSpecRequiredFieldIndex) (string, string, []string) {
	if def.ReadSource == nil {
		tableKey := strings.ToLower(def.TableName)
		return def.TableName, tableKey, nil
	}

	normalizedReadSource, err := models.NormalizeObjectReadSourceConfig(def.ReadSource)
	if err != nil {
		return def.TableName, strings.ToLower(def.TableName), []string{
			fmt.Sprintf("%s object %s ReadSource is invalid: %v", ownerPack, def.TableName, err),
		}
	}

	readSourceTableName := normalizedReadSource.PhysicalTable
	readSourceTableKey := strings.ToLower(readSourceTableName)
	var drifts []string
	if strings.EqualFold(strings.TrimSpace(def.TableName), constants.ObjectRecordType) && readSourceTableName != constants.TableRecordType {
		drifts = append(drifts, fmt.Sprintf("%s object %s ReadSource physical table = %q, want %q", ownerPack, def.TableName, readSourceTableName, constants.TableRecordType))
		return readSourceTableName, readSourceTableKey, drifts
	}
	if !physicalTables[readSourceTableKey] {
		drifts = append(drifts, fmt.Sprintf("%s readSource physical table %s for %s is missing", ownerPack, readSourceTableName, def.TableName))
		return readSourceTableName, readSourceTableKey, drifts
	}

	actualColumns := physicalColumns[readSourceTableKey]
	for _, col := range def.Columns {
		resolvedStorage, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
		if err != nil {
			drifts = append(drifts, fmt.Sprintf("%s field %s.%s has invalid formula storage metadata: %v", ownerPack, def.TableName, col.Name, err))
			continue
		}
		if !resolvedStorage.RequiresPhysicalColumn() {
			continue
		}
		colKey := strings.ToLower(col.Name)
		actualColumn, ok := actualColumns[colKey]
		if !ok {
			drifts = append(drifts, fmt.Sprintf("%s readSource column %s.%s is missing from %s", ownerPack, def.TableName, col.Name, readSourceTableName))
			continue
		}
		if !readSourceColumnShapeMatches(def, readSourceTableName, actualColumn, col, requiredFields) {
			drifts = append(drifts, fmt.Sprintf("%s readSource column %s.%s shape in %s = %s nullable=%v default=%s, want %s nullable=%v default=%s",
				ownerPack,
				def.TableName,
				col.Name,
				readSourceTableName,
				actualColumn.ColumnType,
				actualColumn.Nullable,
				formatPackColumnDefault(actualColumn.DefaultVal.Valid, actualColumn.DefaultVal.String),
				schemaProvisioner.DesiredColumnType(col),
				col.Nullable,
				formatPackColumnDefault(col.Default != "", col.Default),
			))
		}
	}
	return readSourceTableName, readSourceTableKey, drifts
}

func readSourceColumnShapeMatches(def domainschema.TableDefinition, readSourceTableName string, actual schemashape.PhysicalColumnShape, desired domainschema.ColumnDefinition, requiredFields objectSpecRequiredFieldIndex) bool {
	if isRecordTypeReadSourceContract(def, readSourceTableName) {
		typeOnlyDesired := desired
		typeOnlyDesired.Nullable = actual.Nullable
		typeOnlyDesired.Default = ""
		if actual.DefaultVal.Valid {
			typeOnlyDesired.Default = actual.DefaultVal.String
			if typeOnlyDesired.Default == "" {
				typeOnlyDesired.Default = "''"
			}
		}
		typeOnlyDesired.CollationName = ""
		if actual.CollationName.Valid {
			typeOnlyDesired.CollationName = actual.CollationName.String
		}
		return schemaProvisioner.ColumnShapeMatches(actual, typeOnlyDesired)
	}
	return packObjectSpecColumnShapeMatches(def.TableName, actual, desired, requiredFields)
}

func isRecordTypeReadSourceContract(def domainschema.TableDefinition, readSourceTableName string) bool {
	return strings.EqualFold(strings.TrimSpace(def.TableName), constants.ObjectRecordType) &&
		strings.EqualFold(strings.TrimSpace(readSourceTableName), constants.TableRecordType)
}

func IsPackObjectKeyPrefixDrift(err error) bool {
	if err == nil {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			var driftErr *packObjectPhysicalSchemaDriftError
			if !errors.As(childErr, &driftErr) || !driftErr.HasOnlyKeyPrefixDrift() {
				return false
			}
		}
		return true
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.HasOnlyKeyPrefixDrift()
	}
	return false
}

// IsPackObjectReconcileVerifyRepairableDrift is the strict classifier for
// background-only reconcile callers. It accepts only KeyPrefix drift and
// MissingLiveFieldRowsError as repairable; it rejects pack-owned _System_Field
// row loss and missing physical columns, which require the full apply-path
// (reconcileActiveInstalledPackSourceState) to recover from stored source identity.
//
// Callers in the ApplyAll / setup-apply / boot-time path must use
// IsPackObjectInstalledPackRepairableDrift, which is intentionally broader.
func IsPackObjectReconcileVerifyRepairableDrift(err error) bool {
	if err == nil {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(childErr, &specdriftErr) {
				continue
			}
			var driftErr *packObjectPhysicalSchemaDriftError
			if !errors.As(childErr, &driftErr) || !driftErr.HasOnlyKeyPrefixDrift() {
				return false
			}
		}
		return true
	}
	if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(err, &specdriftErr) {
		return true
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.HasOnlyKeyPrefixDrift()
	}
	return false
}

func IsPackObjectInstalledPackRepairableDrift(err error) bool {
	if err == nil {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(childErr, &specdriftErr) {
				continue
			}
			if specdrift.IsUniqueOrDeleteConstraintOnlyFieldDrift(childErr) {
				continue
			}
			var driftErr *packObjectPhysicalSchemaDriftError
			if !errors.As(childErr, &driftErr) || !driftErr.HasOnlyInstalledPackRepairableDrift() {
				return false
			}
		}
		return true
	}
	if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(err, &specdriftErr) {
		return true
	}
	if specdrift.IsUniqueOrDeleteConstraintOnlyFieldDrift(err) {
		return true
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.HasOnlyInstalledPackRepairableDrift()
	}
	return false
}

func IsPackObjectInstalledPackRepairableLiveFieldDrift(err error, allowedFieldKeys map[string]bool) bool {
	if err == nil {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(childErr, &specdriftErr) {
				continue
			}
			var driftErr *packObjectPhysicalSchemaDriftError
			if !errors.As(childErr, &driftErr) || !driftErr.HasOnlyInstalledPackRepairableDriftOrAllowedUndeclaredLiveFields(allowedFieldKeys) {
				return false
			}
		}
		return true
	}
	if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(err, &specdriftErr) {
		return true
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.HasOnlyInstalledPackRepairableDriftOrAllowedUndeclaredLiveFields(allowedFieldKeys)
	}
	return false
}

// IsPackObjectPhysicalColumnMissingDrift returns true iff err is a pure
// physical-column-missing drift with no other drift types and no aggregate.
// Diagnostic use only — does not match objectSpecVerifyError aggregates.
// Uses direct type assertion (not errors.As) to avoid traversing aggregate trees.
func IsPackObjectPhysicalColumnMissingDrift(err error) bool {
	if err == nil {
		return false
	}
	driftErr, ok := err.(*packObjectPhysicalSchemaDriftError)
	if !ok || driftErr == nil || len(driftErr.drifts) == 0 {
		return false
	}
	for _, drift := range driftErr.drifts {
		if !isPackPhysicalColumnMissingDriftBullet(drift) {
			return false
		}
	}
	return true
}

// IsPackObjectAnyPhysicalColumnMissingDrift returns true if err contains any
// missing-physical-column drift bullet, including within objectSpecVerifyError
// aggregates that mix column-missing with other drift types.
func IsPackObjectAnyPhysicalColumnMissingDrift(err error) bool {
	if err == nil {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			var driftErr *packObjectPhysicalSchemaDriftError
			if errors.As(childErr, &driftErr) && driftErr.HasAnyMissingPhysicalColumns() {
				return true
			}
		}
		return false
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.HasAnyMissingPhysicalColumns()
	}
	return false
}

// IsPackObjectColumnMissingAloneWithoutFieldRowMissing returns true if err
// contains physical-column-missing drift but no _System_Field-row-missing drift.
// This indicates structural corruption (schema mutated outside pack management
// while metadata still expects the column) rather than a clean field removal
// (where both the physical column and the _System_Field row are absent together).
// Callers use this to fail-closed in the background reconcile verify path for
// the structural-corruption case while still allowing auto-repair when a full
// field removal is detected (column and metadata row both gone).
func IsPackObjectColumnMissingAloneWithoutFieldRowMissing(err error) bool {
	if err == nil {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			var driftErr *packObjectPhysicalSchemaDriftError
			if errors.As(childErr, &driftErr) && driftErr.hasColumnMissingWithoutFieldRowMissing() {
				return true
			}
		}
		return false
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.hasColumnMissingWithoutFieldRowMissing()
	}
	return false
}

// IsPackObjectBackgroundReconcileRepairableDrift returns true for errors that
// the background reconcile verify path may auto-repair: KeyPrefix drift and
// MissingLiveFieldRowsError. Fails closed for any drift containing a missing
// physical column — those require explicit apply-all (split-1090b tracks the
// hash-drift reapply path).
func IsPackObjectBackgroundReconcileRepairableDrift(err error) bool {
	if err == nil {
		return false
	}
	if IsPackObjectAnyPhysicalColumnMissingDrift(err) {
		return false
	}
	var aggregate *objectSpecVerifyError
	if errors.As(err, &aggregate) {
		if aggregate == nil || len(aggregate.errs) == 0 {
			return false
		}
		for _, childErr := range aggregate.errs {
			if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(childErr, &specdriftErr) {
				continue
			}
			var driftErr *packObjectPhysicalSchemaDriftError
			if !errors.As(childErr, &driftErr) || !driftErr.HasOnlyKeyPrefixDrift() {
				return false
			}
		}
		return true
	}
	if specdriftErr := (*specdrift.MissingLiveFieldRowsError)(nil); errors.As(err, &specdriftErr) {
		return true
	}
	var driftErr *packObjectPhysicalSchemaDriftError
	if errors.As(err, &driftErr) {
		return driftErr.HasOnlyKeyPrefixDrift()
	}
	return false
}

func loadPhysicalTableSet(db *sql.DB) (map[string]bool, error) {
	rows, err := db.Query(`SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE()`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	tables := make(map[string]bool)
	for rows.Next() {
		var tableName string
		if err := rows.Scan(&tableName); err != nil {
			return nil, err
		}
		tables[strings.ToLower(tableName)] = true
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return tables, nil
}

func formatPackColumnDefault(valid bool, value string) string {
	if !valid {
		return "<null>"
	}
	return value
}

func formatPackInt64(value sql.NullInt64) string {
	if !value.Valid {
		return "<null>"
	}
	return fmt.Sprintf("%d", value.Int64)
}

func formatExpectedPackLength(length int) string {
	if length <= 0 {
		return "<null>"
	}
	return fmt.Sprintf("%d", length)
}

func packMaxLengthMatches(live sql.NullInt64, wantLength int) bool {
	if wantLength <= 0 {
		return !live.Valid
	}
	return live.Valid && live.Int64 == int64(wantLength)
}

func fieldMetadataExactRequired(manageability Manageability, row packFieldRegistryRow) bool {
	return manageability != ManageabilityDefaultsOnly || !row.isCustomized
}

func fieldMetadataExactRequiredForApply(preserveCustomized bool, row packFieldRegistryRow) bool {
	return !preserveCustomized || !row.isCustomized
}

func objectSpecUniquePhysicalConstraintDefinitionsForApply(db *sql.DB, _ string, defs []domainschema.TableDefinition, preserveCustomized bool, authoredFieldSets ...objectSpecAuthoredFieldSet) ([]domainschema.TableDefinition, error) {
	if !tableDefinitionsDeclareObjectSpecFields(defs) {
		return defs, nil
	}
	if !preserveCustomized {
		return objectSpecUniquePhysicalConstraintDefinitions(defs, nil, nil), nil
	}
	authoredFields := objectSpecAuthoredFieldSetFromArgs(authoredFieldSets)
	fieldRegistry, err := loadObjectSpecUniqueFieldRegistry(db)
	if err != nil {
		return nil, err
	}
	return objectSpecUniquePhysicalConstraintDefinitionsForAuthoredFields(defs, fieldRegistry, authoredFields, func(row packFieldRegistryRow) bool {
		return fieldMetadataExactRequiredForApply(preserveCustomized, row)
	}), nil
}

func objectSpecUniquePhysicalConstraintDefinitionsForVerify(db *sql.DB, _ string, defs []domainschema.TableDefinition, manageability Manageability, authoredFieldSets ...objectSpecAuthoredFieldSet) ([]domainschema.TableDefinition, error) {
	if !tableDefinitionsDeclareObjectSpecFields(defs) {
		return defs, nil
	}
	if manageability != ManageabilityDefaultsOnly {
		return objectSpecUniquePhysicalConstraintDefinitions(defs, nil, nil), nil
	}
	authoredFields := objectSpecAuthoredFieldSetFromArgs(authoredFieldSets)
	fieldRegistry, err := loadObjectSpecUniqueFieldRegistry(db)
	if err != nil {
		return nil, err
	}
	return objectSpecUniquePhysicalConstraintDefinitionsForAuthoredFields(defs, fieldRegistry, authoredFields, func(row packFieldRegistryRow) bool {
		return fieldMetadataExactRequired(manageability, row)
	}), nil
}

func objectSpecUniquePhysicalConstraintDefinitions(defs []domainschema.TableDefinition, fieldRegistry objectSpecUniqueFieldRegistry, exactRequired func(packFieldRegistryRow) bool) []domainschema.TableDefinition {
	return objectSpecUniquePhysicalConstraintDefinitionsForAuthoredFields(defs, fieldRegistry, objectSpecAuthoredFieldSet{}, exactRequired)
}

func objectSpecUniquePhysicalConstraintDefinitionsForAuthoredFields(defs []domainschema.TableDefinition, fieldRegistry objectSpecUniqueFieldRegistry, authoredFields objectSpecAuthoredFieldSet, exactRequired func(packFieldRegistryRow) bool) []domainschema.TableDefinition {
	effective := make([]domainschema.TableDefinition, len(defs))
	for i, def := range defs {
		effective[i] = def
		if len(def.Columns) == 0 {
			continue
		}
		cols := make([]domainschema.ColumnDefinition, len(def.Columns))
		copy(cols, def.Columns)
		indices := def.Indices
		for j, col := range cols {
			if col.Unique {
				indices = objectSpecEffectiveSingleColumnIndexDefinitions(indices, col.Name, true)
			}
			if fieldRegistry == nil || exactRequired == nil {
				continue
			}
			if !authoredFields.contains(def.TableType, def.TableName, col.Name) {
				continue
			}
			row, ok := fieldRegistry[newObjectSpecUniqueFieldKey(def.TableType, def.TableName, col.Name)]
			if !ok || exactRequired(row) {
				continue
			}
			cols[j].Unique = row.isUnique
			indices = objectSpecEffectiveSingleColumnIndexDefinitions(indices, col.Name, row.isUnique)
		}
		effective[i].Columns = cols
		effective[i].Indices = indices
	}
	return effective
}

func objectSpecEffectiveSingleColumnIndexDefinitions(indices []domainschema.IndexDefinition, columnName string, unique bool) []domainschema.IndexDefinition {
	var effective []domainschema.IndexDefinition
	for i, idx := range indices {
		if len(idx.Columns) != 1 || !strings.EqualFold(strings.TrimSpace(idx.Columns[0]), strings.TrimSpace(columnName)) || idx.Unique == unique {
			continue
		}
		if effective == nil {
			effective = make([]domainschema.IndexDefinition, len(indices))
			copy(effective, indices)
		}
		effective[i].Unique = unique
	}
	if effective != nil {
		return effective
	}
	return indices
}

func tableDefinitionsDeclareObjectSpecFields(defs []domainschema.TableDefinition) bool {
	for _, def := range defs {
		if len(def.Columns) > 0 {
			return true
		}
	}
	return false
}

func loadPackTableRegistry(db *sql.DB) (map[string]packTableRegistryRow, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, `%s`, COALESCE(`%s`, ''), COALESCE(`%s`, 0), COALESCE(`%s`, 0) FROM `%s`",
		constants.FieldSysTable_TableName,
		constants.FieldSysTable_TableType,
		constants.FieldSysTable_Category,
		constants.FieldSysTable_IsManaged,
		constants.FieldSysTable_IsDeleted,
		constants.TableTable,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(map[string]packTableRegistryRow)
	for rows.Next() {
		var tableName, tableType, category string
		var managed, deleted int
		if err := rows.Scan(&tableName, &tableType, &category, &managed, &deleted); err != nil {
			return nil, err
		}
		registry[strings.ToLower(tableName)] = packTableRegistryRow{
			tableType: tableType,
			category:  category,
			managed:   managed != 0,
			deleted:   deleted != 0,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func loadPackObjectRegistry(db *sql.DB) (map[string]packObjectRegistryRow, error) {
	return loadPackObjectRegistryWithExecutor(db)
}

func loadPackObjectRegistryWithExecutor(exec persistence.Executor) (map[string]packObjectRegistryRow, error) {
	rows, err := exec.Query(fmt.Sprintf(
		"SELECT `%s`, `%s`, `%s`, `%s`, `%s`, `%s`, COALESCE(`%s`, ''), COALESCE(`%s`, ''), COALESCE(`%s`, ''), `%s`, COALESCE(`%s`, 0), COALESCE(`%s`, 0), COALESCE(`%s`, 0), COALESCE(`%s`, 0), COALESCE(`%s`, 0), `%s`, `%s` FROM `%s`",
		constants.FieldSysObject_APIName,
		constants.FieldSysObject_TableType,
		constants.FieldSysObject_KeyPrefix,
		constants.FieldSysObject_DefaultChildRows,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ReadSource,
		constants.FieldSysObject_ChildCloneRulesPackName,
		constants.FieldSysObject_ChildCloneRulesPackBusinessKey,
		constants.FieldSysObject_ChildCloneRulesPackManageability,
		constants.FieldSysObject_ChildCloneRulesPackValue,
		constants.FieldSysObject_IsCustomized,
		constants.FieldSysObject_DefaultChildRowsCustomized,
		constants.FieldSysObject_ChildCloneRulesCustomized,
		constants.FieldSysObject_LockRuleCustomized,
		constants.FieldSysObject_IsDeleted,
		constants.FieldSysObject_LockRule,
		constants.FieldSysObject_DisplayNameField,
		constants.TableObject,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(map[string]packObjectRegistryRow)
	for rows.Next() {
		var apiName, tableType, childCloneRulesPackName, childCloneRulesBusinessKey, childCloneRulesManageability string
		var keyPrefix, defaultChildRows, childCloneRules, readSource, lockRule, displayNameField sql.NullString
		var childCloneRulesPackValue sql.NullString
		var isCustomized, defaultChildRowsCustomized, childCloneRulesCustomized, lockRuleCustomized, deleted int
		if err := rows.Scan(&apiName, &tableType, &keyPrefix, &defaultChildRows, &childCloneRules, &readSource, &childCloneRulesPackName, &childCloneRulesBusinessKey, &childCloneRulesManageability, &childCloneRulesPackValue, &isCustomized, &defaultChildRowsCustomized, &childCloneRulesCustomized, &lockRuleCustomized, &deleted, &lockRule, &displayNameField); err != nil {
			return nil, err
		}
		registry[strings.ToLower(apiName)] = packObjectRegistryRow{
			tableType:                    tableType,
			deleted:                      deleted != 0,
			keyPrefix:                    keyPrefix,
			defaultChildRows:             defaultChildRows,
			childCloneRules:              childCloneRules,
			readSource:                   readSource,
			childCloneRulesPackName:      strings.TrimSpace(childCloneRulesPackName),
			childCloneRulesBusinessKey:   strings.TrimSpace(childCloneRulesBusinessKey),
			childCloneRulesManageability: strings.TrimSpace(childCloneRulesManageability),
			childCloneRulesPackValue:     childCloneRulesPackValue,
			isCustomized:                 isCustomized != 0,
			defaultChildRowsCustomized:   defaultChildRowsCustomized != 0,
			childCloneRulesCustomized:    childCloneRulesCustomized != 0,
			lockRuleCustomized:           lockRuleCustomized != 0,
			lockRule:                     lockRule,
			displayNameField:             displayNameField,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func loadPackObjectExclusiveSiblingRegistry(db *sql.DB) (map[string]packObjectExclusiveSiblingRegistryRow, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, `%s`, COALESCE(`%s`, ''), COALESCE(`%s`, ''), COALESCE(`%s`, ''), `%s`, COALESCE(`%s`, 0), COALESCE(`%s`, 0) FROM `%s`",
		constants.FieldSysObject_APIName,
		constants.FieldSysObject_ExclusiveSiblingFields,
		constants.FieldSysObject_ExclusiveSiblingFieldsPackName,
		constants.FieldSysObject_ExclusiveSiblingFieldsPackBusinessKey,
		constants.FieldSysObject_ExclusiveSiblingFieldsPackManageability,
		constants.FieldSysObject_ExclusiveSiblingFieldsPackValue,
		constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
		constants.FieldSysObject_IsDeleted,
		constants.TableObject,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(map[string]packObjectExclusiveSiblingRegistryRow)
	for rows.Next() {
		var apiName, packName, businessKey, manageability string
		var exclusiveSiblingFields, packValue sql.NullString
		var customized, deleted int
		if err := rows.Scan(&apiName, &exclusiveSiblingFields, &packName, &businessKey, &manageability, &packValue, &customized, &deleted); err != nil {
			return nil, err
		}
		registry[strings.ToLower(apiName)] = packObjectExclusiveSiblingRegistryRow{
			deleted:                             deleted != 0,
			exclusiveSiblingFields:              exclusiveSiblingFields,
			exclusiveSiblingFieldsPackName:      strings.TrimSpace(packName),
			exclusiveSiblingFieldsBusinessKey:   strings.TrimSpace(businessKey),
			exclusiveSiblingFieldsManageability: strings.TrimSpace(manageability),
			exclusiveSiblingFieldsPackValue:     packValue,
			exclusiveSiblingFieldsCustomized:    customized != 0,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func canonicalReadSourceJSON(cfg *models.ObjectReadSourceConfig) (string, error) {
	if cfg == nil {
		return "", nil
	}
	return models.CanonicalObjectReadSourceJSON(cfg)
}

func canonicalStoredReadSourceJSON(raw string) (string, error) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return "", nil
	}
	var document any
	if err := json.Unmarshal([]byte(raw), &document); err != nil {
		return "", err
	}
	canonical, err := json.Marshal(document)
	if err != nil {
		return "", err
	}
	return string(canonical), nil
}

func VerifyPackObjectDefaultChildRows(db *sql.DB, ownerPack string, specs []ObjectSpec, manageability Manageability) error {
	objectRegistry, err := loadPackObjectRegistry(db)
	if err != nil {
		return err
	}
	exactRequired := make([]ObjectSpec, 0, len(specs))
	for _, spec := range specs {
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		if ResolveDefaultChildRowsValueOwnership(manageability, row.defaultChildRowsCustomized) == ObjectSpecValueLiveOverrideAllowed {
			continue
		}
		exactRequired = append(exactRequired, spec)
	}
	if err := verifyLivePackDefaultChildRows(db, exactRequired); err != nil {
		return fmt.Errorf("%s live DefaultChildRows contract verify: %w", ownerPack, err)
	}
	return verifyPackObjectDefaultChildRowsAgainstRegistry(ownerPack, exactRequired, manageability, objectRegistry)
}

func VerifyPackObjectListFields(db *sql.DB, ownerPack string, specs []ObjectSpec, manageability Manageability) error {
	objectRegistry, err := loadPackObjectListFieldsRegistry(db)
	if err != nil {
		return err
	}
	return verifyPackObjectListFieldsAgainstRegistry(ownerPack, specs, manageability, objectRegistry)
}

func loadPackObjectListFieldsRegistry(db *sql.DB) (map[string]packObjectListFieldsRegistryRow, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT `%s`, `%s`, COALESCE(`%s`, 0), COALESCE(`%s`, 0) FROM `%s`",
		constants.FieldSysObject_APIName,
		constants.FieldSysObject_ListFields,
		constants.FieldSysObject_ListFieldsCustomized,
		constants.FieldSysObject_IsDeleted,
		constants.TableObject,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(map[string]packObjectListFieldsRegistryRow)
	for rows.Next() {
		var apiName string
		var listFields sql.NullString
		var customized, deleted int
		if err := rows.Scan(&apiName, &listFields, &customized, &deleted); err != nil {
			return nil, err
		}
		registry[strings.ToLower(apiName)] = packObjectListFieldsRegistryRow{
			listFields:           listFields,
			listFieldsCustomized: customized != 0,
			deleted:              deleted != 0,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func verifyPackObjectListFieldsAgainstRegistry(ownerPack string, specs []ObjectSpec, manageability Manageability, objectRegistry map[string]packObjectListFieldsRegistryRow) error {
	var errs []string
	for _, spec := range specs {
		if spec.ListFields == nil {
			continue
		}
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		if ResolveListFieldsValueOwnership(manageability, row.listFieldsCustomized) == ObjectSpecValueLiveOverrideAllowed {
			continue
		}
		want, err := canonicalListFields(spec.ListFields)
		if err != nil {
			return fmt.Errorf("%s %s listFields canonicalize spec: %w", ownerPack, spec.APIName, err)
		}
		got, err := canonicalListFieldsJSON(row.listFields)
		if err != nil {
			return fmt.Errorf("%s %s listFields canonicalize live row: %w", ownerPack, spec.APIName, err)
		}
		if string(want) != string(got) {
			errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ListFields = %s, want %s",
				ownerPack, spec.APIName, formatPackColumnDefault(row.listFields.Valid, row.listFields.String), string(want)))
		}
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s", strings.Join(errs, "\n  - "))
}

func verifyPackObjectDefaultChildRowsAgainstRegistry(ownerPack string, specs []ObjectSpec, manageability Manageability, objectRegistry map[string]packObjectRegistryRow) error {
	var errs []string
	for _, spec := range specs {
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		if ResolveDefaultChildRowsValueOwnership(manageability, row.defaultChildRowsCustomized) == ObjectSpecValueLiveOverrideAllowed {
			continue
		}
		want, err := canonicalDefaultChildRows(spec.DefaultChildRows)
		if err != nil {
			return fmt.Errorf("%s %s defaultChildRows canonicalize spec: %w", ownerPack, spec.APIName, err)
		}
		got, err := canonicalDefaultChildRowsJSON(row.defaultChildRows)
		if err != nil {
			return fmt.Errorf("%s %s defaultChildRows canonicalize live row: %w", ownerPack, spec.APIName, err)
		}
		if string(want) != string(got) {
			errs = append(errs, fmt.Sprintf("%s _System_Object row for %s DefaultChildRows = %s, want %s",
				ownerPack, spec.APIName, formatPackColumnDefault(row.defaultChildRows.Valid, row.defaultChildRows.String), string(want)))
		}
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s", strings.Join(errs, "\n  - "))
}

func VerifyPackObjectChildCloneRules(db *sql.DB, ownerPack, businessKey string, specs []ObjectSpec, manageability Manageability) error {
	objectRegistry, err := loadPackObjectRegistry(db)
	if err != nil {
		return err
	}
	return verifyPackObjectChildCloneRulesAgainstRegistry(ownerPack, businessKey, specs, manageability, objectRegistry)
}

func verifyPackObjectChildCloneRulesAgainstRegistry(ownerPack, businessKey string, specs []ObjectSpec, manageability Manageability, objectRegistry map[string]packObjectRegistryRow) error {
	var errs []string
	enforceSidecars := childCloneRulesComponentSidecarsHydrated(specs, objectRegistry)
	for _, spec := range specs {
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		want, err := canonicalChildCloneRules(spec.ChildCloneRules)
		if err != nil {
			return fmt.Errorf("%s %s childCloneRules canonicalize spec: %w", ownerPack, spec.APIName, err)
		}
		if enforceSidecars {
			if row.childCloneRulesPackName != strings.TrimSpace(ownerPack) {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRulesPackName = %q, want %q",
					ownerPack, spec.APIName, row.childCloneRulesPackName, strings.TrimSpace(ownerPack)))
			}
			if row.childCloneRulesBusinessKey != strings.TrimSpace(businessKey) {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRulesPackBusinessKey = %q, want %q",
					ownerPack, spec.APIName, row.childCloneRulesBusinessKey, strings.TrimSpace(businessKey)))
			}
			if row.childCloneRulesManageability != string(manageability) {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRulesPackManageability = %q, want %q",
					ownerPack, spec.APIName, row.childCloneRulesManageability, string(manageability)))
			}
			if !row.childCloneRulesPackValue.Valid || strings.TrimSpace(row.childCloneRulesPackValue.String) == "" {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRulesPackValue is empty, want %s",
					ownerPack, spec.APIName, string(want)))
			} else {
				gotPack, err := canonicalChildCloneRulesJSON(row.childCloneRulesPackValue)
				if err != nil {
					return fmt.Errorf("%s %s childCloneRules canonicalize pack ownership value: %w", ownerPack, spec.APIName, err)
				}
				if string(want) != string(gotPack) {
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRulesPackValue = %s, want %s",
						ownerPack, spec.APIName, row.childCloneRulesPackValue.String, string(want)))
				}
			}
		}
		if ResolveChildCloneRulesValueOwnership(manageability, row.childCloneRulesCustomized) == ObjectSpecValueLiveOverrideAllowed {
			continue
		}
		if shouldDeferLegacyDefaultsOnlyChildCloneRulesVerify(manageability, enforceSidecars, row) {
			if string(want) == "[]" {
				got, err := canonicalChildCloneRulesJSON(row.childCloneRules)
				if err != nil {
					return fmt.Errorf("%s %s childCloneRules canonicalize live row: %w", ownerPack, spec.APIName, err)
				}
				if string(got) != "[]" {
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRules = %s, want [] after pack removed defaults_only rules; run pack apply to converge the live value",
						ownerPack, spec.APIName, formatPackColumnDefault(row.childCloneRules.Valid, row.childCloneRules.String)))
				}
			}
			continue
		}
		got, err := canonicalChildCloneRulesJSON(row.childCloneRules)
		if err != nil {
			return fmt.Errorf("%s %s childCloneRules canonicalize live row: %w", ownerPack, spec.APIName, err)
		}
		if string(want) != string(got) {
			errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ChildCloneRules = %s, want %s",
				ownerPack, spec.APIName, formatPackColumnDefault(row.childCloneRules.Valid, row.childCloneRules.String), string(want)))
		}
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s", strings.Join(errs, "\n  - "))
}

func childCloneRulesComponentSidecarsHydrated(specs []ObjectSpec, objectRegistry map[string]packObjectRegistryRow) bool {
	for _, spec := range specs {
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		if childCloneRulesSidecarHydrated(row) {
			return true
		}
	}
	return false
}

func shouldDeferLegacyDefaultsOnlyChildCloneRulesVerify(manageability Manageability, componentSidecarsHydrated bool, row packObjectRegistryRow) bool {
	// Warm-upgraded defaults_only rows need ownership hydration before a live
	// mismatch can be classified as pack drift instead of a retained L2 override.
	return manageability == ManageabilityDefaultsOnly &&
		!componentSidecarsHydrated &&
		!childCloneRulesSidecarHydrated(row)
}

func childCloneRulesSidecarHydrated(row packObjectRegistryRow) bool {
	return strings.TrimSpace(row.childCloneRulesPackName) != "" ||
		strings.TrimSpace(row.childCloneRulesBusinessKey) != "" ||
		strings.TrimSpace(row.childCloneRulesManageability) != "" ||
		(row.childCloneRulesPackValue.Valid && strings.TrimSpace(row.childCloneRulesPackValue.String) != "")
}

func VerifyPackObjectExclusiveSiblingFields(db *sql.DB, ownerPack, businessKey string, specs []ObjectSpec, manageability Manageability) error {
	objectRegistry, err := loadPackObjectExclusiveSiblingRegistry(db)
	if err != nil {
		return err
	}
	return verifyPackObjectExclusiveSiblingFieldsAgainstRegistry(ownerPack, businessKey, specs, manageability, objectRegistry)
}

func verifyPackObjectExclusiveSiblingFieldsAgainstRegistry(ownerPack, businessKey string, specs []ObjectSpec, manageability Manageability, objectRegistry map[string]packObjectExclusiveSiblingRegistryRow) error {
	var errs []string
	enforceSidecars := exclusiveSiblingFieldsComponentSidecarsHydrated(ownerPack, businessKey, specs, objectRegistry)
	for _, spec := range specs {
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		if !exclusiveSiblingFieldsSpecNeedsVerifyOwnership(ownerPack, businessKey, spec, row) {
			continue
		}
		want, err := canonicalExclusiveSiblingFields(spec.ExclusiveSiblingFields)
		if err != nil {
			return fmt.Errorf("%s %s exclusiveSiblingFields canonicalize spec: %w", ownerPack, spec.APIName, err)
		}
		if enforceSidecars {
			if row.exclusiveSiblingFieldsPackName != strings.TrimSpace(ownerPack) {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFieldsPackName = %q, want %q",
					ownerPack, spec.APIName, row.exclusiveSiblingFieldsPackName, strings.TrimSpace(ownerPack)))
			}
			if row.exclusiveSiblingFieldsBusinessKey != strings.TrimSpace(businessKey) {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFieldsPackBusinessKey = %q, want %q",
					ownerPack, spec.APIName, row.exclusiveSiblingFieldsBusinessKey, strings.TrimSpace(businessKey)))
			}
			if row.exclusiveSiblingFieldsManageability != string(manageability) {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFieldsPackManageability = %q, want %q",
					ownerPack, spec.APIName, row.exclusiveSiblingFieldsManageability, string(manageability)))
			}
			if !row.exclusiveSiblingFieldsPackValue.Valid || strings.TrimSpace(row.exclusiveSiblingFieldsPackValue.String) == "" {
				errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFieldsPackValue is empty, want %s",
					ownerPack, spec.APIName, string(want)))
			} else {
				gotPack, err := canonicalExclusiveSiblingFieldsJSON(row.exclusiveSiblingFieldsPackValue)
				if err != nil {
					return fmt.Errorf("%s %s exclusiveSiblingFields canonicalize pack ownership value: %w", ownerPack, spec.APIName, err)
				}
				if string(want) != string(gotPack) {
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFieldsPackValue = %s, want %s",
						ownerPack, spec.APIName, row.exclusiveSiblingFieldsPackValue.String, string(want)))
				}
			}
		}
		if ResolveExclusiveSiblingFieldsValueOwnership(manageability, row.exclusiveSiblingFieldsCustomized) == ObjectSpecValueLiveOverrideAllowed {
			continue
		}
		if shouldDeferLegacyDefaultsOnlyExclusiveSiblingFieldsVerify(manageability, enforceSidecars, row) {
			if string(want) == "[]" {
				got, err := canonicalExclusiveSiblingFieldsJSON(row.exclusiveSiblingFields)
				if err != nil {
					return fmt.Errorf("%s %s exclusiveSiblingFields canonicalize live row: %w", ownerPack, spec.APIName, err)
				}
				if string(got) != "[]" {
					errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFields = %s, want [] after pack removed defaults_only fields; run pack apply to converge the live value",
						ownerPack, spec.APIName, formatPackColumnDefault(row.exclusiveSiblingFields.Valid, row.exclusiveSiblingFields.String)))
				}
			}
			continue
		}
		got, err := canonicalExclusiveSiblingFieldsJSON(row.exclusiveSiblingFields)
		if err != nil {
			return fmt.Errorf("%s %s exclusiveSiblingFields canonicalize live row: %w", ownerPack, spec.APIName, err)
		}
		if string(want) != string(got) {
			errs = append(errs, fmt.Sprintf("%s _System_Object row for %s ExclusiveSiblingFields = %s, want %s",
				ownerPack, spec.APIName, formatPackColumnDefault(row.exclusiveSiblingFields.Valid, row.exclusiveSiblingFields.String), string(want)))
		}
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s", strings.Join(errs, "\n  - "))
}

func exclusiveSiblingFieldsComponentSidecarsHydrated(ownerPack, businessKey string, specs []ObjectSpec, objectRegistry map[string]packObjectExclusiveSiblingRegistryRow) bool {
	for _, spec := range specs {
		row, ok := objectRegistry[strings.ToLower(spec.APIName)]
		if !ok || row.deleted {
			continue
		}
		if exclusiveSiblingFieldsSpecNeedsVerifyOwnership(ownerPack, businessKey, spec, row) && exclusiveSiblingFieldsSidecarHydrated(row) {
			return true
		}
	}
	return false
}

func exclusiveSiblingFieldsSpecNeedsVerifyOwnership(ownerPack, businessKey string, spec ObjectSpec, row packObjectExclusiveSiblingRegistryRow) bool {
	return objectSpecDeclaresExclusiveSiblingFields(spec) || exclusiveSiblingFieldsRegistryRowOwnedBy(row, ownerPack, businessKey)
}

func exclusiveSiblingFieldsRegistryRowOwnedBy(row packObjectExclusiveSiblingRegistryRow, ownerPack, businessKey string) bool {
	return row.exclusiveSiblingFieldsPackName == strings.TrimSpace(ownerPack) &&
		row.exclusiveSiblingFieldsBusinessKey == strings.TrimSpace(businessKey) &&
		(row.exclusiveSiblingFieldsPackName != "" || row.exclusiveSiblingFieldsBusinessKey != "")
}

func shouldDeferLegacyDefaultsOnlyExclusiveSiblingFieldsVerify(manageability Manageability, componentSidecarsHydrated bool, row packObjectExclusiveSiblingRegistryRow) bool {
	return manageability == ManageabilityDefaultsOnly &&
		!componentSidecarsHydrated &&
		!exclusiveSiblingFieldsSidecarHydrated(row)
}

func exclusiveSiblingFieldsSidecarHydrated(row packObjectExclusiveSiblingRegistryRow) bool {
	return strings.TrimSpace(row.exclusiveSiblingFieldsPackName) != "" ||
		strings.TrimSpace(row.exclusiveSiblingFieldsBusinessKey) != "" ||
		strings.TrimSpace(row.exclusiveSiblingFieldsManageability) != "" ||
		(row.exclusiveSiblingFieldsPackValue.Valid && strings.TrimSpace(row.exclusiveSiblingFieldsPackValue.String) != "")
}

func canonicalChildCloneRulesJSON(raw sql.NullString) ([]byte, error) {
	if !raw.Valid || strings.TrimSpace(raw.String) == "" {
		return canonicalChildCloneRules(nil)
	}
	var rules []models.ChildCloneRule
	if err := json.Unmarshal([]byte(raw.String), &rules); err != nil {
		return nil, err
	}
	return canonicalChildCloneRules(rules)
}

func canonicalChildCloneRules(rules []models.ChildCloneRule) ([]byte, error) {
	if len(rules) == 0 {
		return []byte("[]"), nil
	}
	type normalizedRule struct {
		ParentLookupField string            `json:"parentLookupField"`
		SourceObject      string            `json:"sourceObject"`
		SourceLookupField string            `json:"sourceLookupField"`
		SourceIdField     string            `json:"sourceIdField,omitempty"`
		HydrateSource     bool              `json:"hydrateSource,omitempty"`
		TargetObject      string            `json:"targetObject"`
		TargetLookupField string            `json:"targetLookupField"`
		FieldMappings     map[string]string `json:"fieldMappings"`
	}
	normalized := make([]normalizedRule, 0, len(rules))
	for _, rule := range rules {
		fieldMappings := map[string]string{}
		for key, value := range rule.FieldMappings {
			fieldMappings[key] = value
		}
		normalized = append(normalized, normalizedRule{
			ParentLookupField: rule.ParentLookupField,
			SourceObject:      rule.SourceObject,
			SourceLookupField: rule.SourceLookupField,
			SourceIdField:     rule.SourceIdField,
			HydrateSource:     rule.HydrateSource,
			TargetObject:      rule.TargetObject,
			TargetLookupField: rule.TargetLookupField,
			FieldMappings:     fieldMappings,
		})
	}
	sort.Slice(normalized, func(i, j int) bool {
		left, _ := json.Marshal(normalized[i])
		right, _ := json.Marshal(normalized[j])
		return bytes.Compare(left, right) < 0
	})
	return json.Marshal(normalized)
}

func verifyLivePackDefaultChildRows(db *sql.DB, specs []ObjectSpec) error {
	if len(specs) == 0 {
		return nil
	}
	metadataRepo := persistence.NewMetadataRepository(db)
	schemaRepo := persistence.NewSchemaRepository(db, nil)
	metadataSvc := metadata.NewMetadataService(metadataRepo, metadata.NewSchemaManager(schemaRepo), nil)
	autoNumberSrc := autonumber.NewMetadataSource(db, nil)
	if err := autoNumberSrc.Refresh(context.Background()); err != nil {
		return fmt.Errorf("refresh live autonumber metadata: %w", err)
	}
	fallbackEngineOwned := func(objectName string) map[string]bool {
		fields := autoNumberSrc.LookupByObject(objectName)
		if len(fields) == 0 {
			return nil
		}
		out := make(map[string]bool, len(fields))
		for _, field := range fields {
			out[strings.ToLower(field.FieldAPIName)] = true
		}
		return out
	}
	for _, obj := range buildObjectMetadata(specs) {
		if obj == nil || len(obj.DefaultChildRows) == 0 {
			continue
		}
		if _, err := records.ResolveDefaultChildRowsContracts(context.Background(), metadataSvc, obj, obj.DefaultChildRows, fallbackEngineOwned); err != nil {
			return err
		}
	}
	return nil
}

func canonicalDefaultChildRowsJSON(raw sql.NullString) ([]byte, error) {
	if !raw.Valid || strings.TrimSpace(raw.String) == "" {
		return canonicalDefaultChildRows(nil)
	}
	var rows []models.DefaultChildRow
	if err := json.Unmarshal([]byte(raw.String), &rows); err != nil {
		return nil, err
	}
	return canonicalDefaultChildRows(rows)
}

func canonicalListFieldsJSON(raw sql.NullString) ([]byte, error) {
	var fields []string
	if raw.Valid && strings.TrimSpace(raw.String) != "" {
		if err := json.Unmarshal([]byte(raw.String), &fields); err != nil {
			return nil, err
		}
	}
	return canonicalListFields(fields)
}

func canonicalListFields(fields []string) ([]byte, error) {
	normalized := sourceBackedNormalizeStringSlice(fields)
	if normalized == nil {
		normalized = []string{}
	}
	return json.Marshal(normalized)
}

func canonicalDefaultChildRows(rows []models.DefaultChildRow) ([]byte, error) {
	if len(rows) == 0 {
		return []byte("[]"), nil
	}
	type normalizedRow struct {
		TargetObject      string                 `json:"targetObject"`
		TargetLookupField string                 `json:"targetLookupField"`
		FieldValues       map[string]interface{} `json:"fieldValues"`
	}
	normalized := make([]normalizedRow, 0, len(rows))
	for _, row := range rows {
		fieldValues := map[string]interface{}{}
		for key, value := range row.FieldValues {
			fieldValues[key] = value
		}
		normalized = append(normalized, normalizedRow{
			TargetObject:      row.TargetObject,
			TargetLookupField: row.TargetLookupField,
			FieldValues:       fieldValues,
		})
	}
	sort.Slice(normalized, func(i, j int) bool {
		leftFields, _ := json.Marshal(normalized[i].FieldValues)
		rightFields, _ := json.Marshal(normalized[j].FieldValues)
		left := normalized[i].TargetObject + "\x00" + normalized[i].TargetLookupField + "\x00" + string(leftFields)
		right := normalized[j].TargetObject + "\x00" + normalized[j].TargetLookupField + "\x00" + string(rightFields)
		return left < right
	})
	return json.Marshal(normalized)
}

func loadPackFieldRegistry(db *sql.DB) (map[string]map[string]packFieldRegistryRow, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT o.`%s`, f.`%s`, COALESCE(f.`%s`, ''), COALESCE(f.`%s`, ''), COALESCE(f.`%s`, 0), f.`%s` FROM `%s` f JOIN `%s` o ON o.`%s` = f.`%s` WHERE COALESCE(f.`%s`, 0) = 0 AND COALESCE(o.`%s`, 0) = 0",
		constants.FieldSysObject_APIName,
		constants.FieldSysField_APIName,
		constants.FieldSysField_OwnerKind,
		constants.FieldSysField_OwnerPack,
		constants.FieldSysField_IsCustomized,
		constants.FieldSysField_MaxLength,
		constants.TableField,
		constants.TableObject,
		constants.FieldSysObject_ID,
		constants.FieldSysField_ObjectID,
		constants.FieldSysField_IsDeleted,
		constants.FieldSysObject_IsDeleted,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(map[string]map[string]packFieldRegistryRow)
	for rows.Next() {
		var objectAPIName, fieldAPIName, ownerKind, ownerPack string
		var isCustomized int
		var maxLength sql.NullInt64
		if err := rows.Scan(&objectAPIName, &fieldAPIName, &ownerKind, &ownerPack, &isCustomized, &maxLength); err != nil {
			return nil, err
		}
		objectKey := strings.ToLower(objectAPIName)
		if registry[objectKey] == nil {
			registry[objectKey] = make(map[string]packFieldRegistryRow)
		}
		registry[objectKey][strings.ToLower(fieldAPIName)] = packFieldRegistryRow{
			fieldAPIName: strings.TrimSpace(fieldAPIName),
			ownerKind:    strings.TrimSpace(strings.ToLower(ownerKind)),
			ownerPack:    strings.TrimSpace(ownerPack),
			isCustomized: isCustomized != 0,
			maxLength:    maxLength,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func loadObjectSpecUniqueFieldRegistry(db *sql.DB) (objectSpecUniqueFieldRegistry, error) {
	rows, err := db.Query(fmt.Sprintf(
		"SELECT o.`%s`, o.`%s`, f.`%s`, COALESCE(f.`%s`, ''), COALESCE(f.`%s`, ''), COALESCE(f.`%s`, 0), COALESCE(f.`%s`, 0) FROM `%s` f JOIN `%s` o ON o.`%s` = f.`%s` WHERE COALESCE(f.`%s`, 0) = 0 AND COALESCE(o.`%s`, 0) = 0",
		constants.FieldSysObject_APIName,
		constants.FieldSysObject_TableType,
		constants.FieldSysField_APIName,
		constants.FieldSysField_OwnerKind,
		constants.FieldSysField_OwnerPack,
		constants.FieldSysField_IsCustomized,
		constants.FieldSysField_IsUnique,
		constants.TableField,
		constants.TableObject,
		constants.FieldSysObject_ID,
		constants.FieldSysField_ObjectID,
		constants.FieldSysField_IsDeleted,
		constants.FieldSysObject_IsDeleted,
	))
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	registry := make(objectSpecUniqueFieldRegistry)
	for rows.Next() {
		var objectAPIName, tableType, fieldAPIName, ownerKind, ownerPack string
		var isCustomized, isUnique int
		if err := rows.Scan(&objectAPIName, &tableType, &fieldAPIName, &ownerKind, &ownerPack, &isCustomized, &isUnique); err != nil {
			return nil, err
		}
		registry[newObjectSpecUniqueFieldKey(tableType, objectAPIName, fieldAPIName)] = packFieldRegistryRow{
			fieldAPIName: strings.TrimSpace(fieldAPIName),
			ownerKind:    strings.TrimSpace(strings.ToLower(ownerKind)),
			ownerPack:    strings.TrimSpace(ownerPack),
			isCustomized: isCustomized != 0,
			isUnique:     isUnique != 0,
		}
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return registry, nil
}

func packColumnRequiresFieldMetadata(columnName string) bool {
	return strings.TrimSpace(columnName) != ""
}